X-Git-Url: http://dolda2000.com/gitweb/?a=blobdiff_plain;f=lib%2Fmtio.c;h=183f9f279532c54bffba137eb34a4cb70eb8cc81;hb=122c246222afdefdc2d50fdbac6f867849a33d5f;hp=3bb7b860a59535d008fd50bfe2dadd14e68a86fe;hpb=3dc69b452f437790c0a2f8b7211f542fc67310b4;p=ashd.git diff --git a/lib/mtio.c b/lib/mtio.c index 3bb7b86..183f9f2 100644 --- a/lib/mtio.c +++ b/lib/mtio.c @@ -16,109 +16,241 @@ along with this program. If not, see . */ +#ifdef HAVE_CONFIG_H +#include +#endif #include +#include #include +#include #include -#include #include -#include +#include -#ifdef HAVE_CONFIG_H -#include -#endif #include #include #include #include -struct blocker { - struct blocker *n, *p; - int fd; +static ssize_t mtread(void *cookie, void *buf, size_t len) +{ + struct stdiofd *d = cookie; int ev; - time_t to; - struct muth *th; -}; + ssize_t ret; + + while(1) { + ret = read(d->fd, buf, len); + if((ret < 0) && (errno == EAGAIN)) { + ev = block(d->fd, EV_READ, d->timeout); + if(ev < 0) { + /* If we just go on, we should get the real error. */ + continue; + } else if(ev == 0) { + errno = ETIMEDOUT; + return(-1); + } else { + continue; + } + } else { + return(ret); + } + } +} + +static ssize_t mtwrite(void *cookie, const void *buf, size_t len) +{ + struct stdiofd *d = cookie; + int ev; + ssize_t ret; + + while(1) { + if(d->sock) + ret = send(d->fd, buf, len, MSG_DONTWAIT | MSG_NOSIGNAL); + else + ret = write(d->fd, buf, len); + if((ret < 0) && (errno == EAGAIN)) { + ev = block(d->fd, EV_WRITE, d->timeout); + if(ev < 0) { + /* If we just go on, we should get the real error. */ + continue; + } else if(ev == 0) { + errno = ETIMEDOUT; + return(-1); + } + } else { + return(ret); + } + } +} -static struct blocker *blockers; +static int mtclose(void *cookie) +{ + struct stdiofd *d = cookie; + + close(d->fd); + free(d); + return(0); +} -int block(int fd, int ev, time_t to) +FILE *mtstdopen(int fd, int issock, int timeout, char *mode, struct stdiofd **infop) { - struct blocker *bl; - int rv; + struct stdiofd *d; + FILE *ret; + int r, w; - omalloc(bl); - bl->fd = fd; - bl->ev = ev; - if(to > 0) - bl->to = time(NULL) + to; - bl->th = current; - bl->n = blockers; - if(blockers) - blockers->p = bl; - blockers = bl; - rv = yield(); - if(bl->n) - bl->n->p = bl->p; - if(bl->p) - bl->p->n = bl->n; - if(bl == blockers) - blockers = bl->n; - return(rv); + if(!strcmp(mode, "r")) { + r = 1; w = 0; + } else if(!strcmp(mode, "w")) { + r = 0; w = 1; + } else if(!strcmp(mode, "r+")) { + r = w = 1; + } else { + return(NULL); + } + omalloc(d); + d->fd = fd; + d->sock = issock; + d->timeout = timeout; + if(!(ret = funstdio(d, r?mtread:NULL, w?mtwrite:NULL, NULL, mtclose))) { + free(d); + return(NULL); + } + fcntl(fd, F_SETFL, fcntl(fd, F_GETFL) | O_NONBLOCK); + if(infop) + *infop = d; + return(ret); } -void ioloop(void) +struct bufio *mtbioopen(int fd, int issock, int timeout, char *mode, struct stdiofd **infop) { - int ret; - fd_set rfds, wfds, efds; - struct blocker *bl, *nbl; - struct timeval toval; - time_t now, timeout; - int maxfd; - int ev; + static struct bufioops ops = { + .read = mtread, .write = mtwrite, .close = mtclose, + }; + struct stdiofd *d; + struct bufio *ret; - while(blockers != NULL) { - FD_ZERO(&rfds); - FD_ZERO(&wfds); - FD_ZERO(&efds); - maxfd = 0; - now = time(NULL); - timeout = 0; - for(bl = blockers; bl; bl = bl->n) { - if(bl->ev & EV_READ) - FD_SET(bl->fd, &rfds); - if(bl->ev & EV_WRITE) - FD_SET(bl->fd, &wfds); - FD_SET(bl->fd, &efds); - if(bl->fd > maxfd) - maxfd = bl->fd; - if((bl->to != 0) && ((timeout == 0) || (timeout > bl->to))) - timeout = bl->to; + if(!strcmp(mode, "r")) { + } else if(!strcmp(mode, "w")) { + } else if(!strcmp(mode, "r+")) { + } else { + return(NULL); + } + omalloc(d); + d->fd = fd; + d->sock = issock; + d->timeout = timeout; + if(!(ret = bioopen(d, &ops))) { + free(d); + return(NULL); + } + fcntl(fd, F_SETFL, fcntl(fd, F_GETFL) | O_NONBLOCK); + if(infop) + *infop = d; + return(ret); +} + +struct pipe { + struct charbuf data; + size_t bufmax; + int closed; + struct muth *r, *w; +}; + +static void freepipe(struct pipe *p) +{ + buffree(p->data); + free(p); +} + +static ssize_t piperead(void *pdata, void *buf, size_t len) +{ + struct pipe *p = pdata; + ssize_t ret; + + while(p->data.d == 0) { + if(p->closed & 2) + return(0); + if(p->r) { + errno = EBUSY; + return(-1); } - toval.tv_sec = timeout - now; - toval.tv_usec = 0; - ret = select(maxfd + 1, &rfds, &wfds, &efds, timeout?(&toval):NULL); - if(ret < 0) { - if(errno != EINTR) { - flog(LOG_CRIT, "ioloop: select errored out: %s", strerror(errno)); - /* To avoid CPU hogging in case it's bad, which it - * probably is. */ - sleep(1); - } + p->r = current; + yield(); + p->r = NULL; + } + ret = min(len, p->data.d); + memcpy(buf, p->data.b, ret); + memmove(p->data.b, p->data.b + ret, p->data.d -= ret); + if(p->w) + resume(p->w, 0); + return(ret); +} + +static int piperclose(void *pdata) +{ + struct pipe *p = pdata; + + if(p->closed & 2) { + freepipe(p); + } else { + p->closed |= 1; + if(p->w) + resume(p->w, 0); + } + return(0); +} + +static ssize_t pipewrite(void *pdata, const void *buf, size_t len) +{ + struct pipe *p = pdata; + ssize_t ret; + + if(p->closed & 1) { + errno = EPIPE; + return(-1); + } + while(p->data.d >= p->bufmax) { + if(p->w) { + errno = EBUSY; + return(-1); } - now = time(NULL); - for(bl = blockers; bl; bl = nbl) { - nbl = bl->n; - ev = 0; - if(FD_ISSET(bl->fd, &rfds)) - ev |= EV_READ; - if(FD_ISSET(bl->fd, &wfds)) - ev |= EV_WRITE; - if(FD_ISSET(bl->fd, &efds)) - ev = -1; - if((ev < 0) || (ev & bl->ev)) - resume(bl->th, ev); - else if((bl->to != 0) && (bl->to <= now)) - resume(bl->th, 0); + p->w = current; + yield(); + p->w = NULL; + if(p->closed & 1) { + errno = EPIPE; + return(-1); } } + ret = min(len, p->bufmax - p->data.d); + sizebuf(p->data, p->data.d + ret); + memcpy(p->data.b + p->data.d, buf, ret); + p->data.d += ret; + if(p->r) + resume(p->r, 0); + return(ret); +} + +static int pipewclose(void *pdata) +{ + struct pipe *p = pdata; + + if(p->closed & 1) { + freepipe(p); + } else { + p->closed |= 2; + if(p->r) + resume(p->r, 0); + } + return(0); +} + +void mtiopipe(FILE **read, FILE **write) +{ + struct pipe *p; + + omalloc(p); + p->bufmax = 4096; + *read = funstdio(p, piperead, NULL, NULL, piperclose); + *write = funstdio(p, NULL, pipewrite, NULL, pipewclose); }