void
ioset_init(void)
{
- assert(engine == NULL);
+ if (engine) /* someone beat us to it */
+ return;
#if WITH_IOSET_KQUEUE
if (!engine && io_engine_kqueue.init())
- engine = &io_engine_kqueue;
+ engine = &io_engine_kqueue;
#endif
#if WITH_IOSET_EPOLL
if (!engine && io_engine_epoll.init())
- engine = &io_engine_epoll;
+ engine = &io_engine_epoll;
#endif
#if WITH_IOSET_WIN32
if (engine) {
/* we found one that works */
} else if (io_engine_select.init())
- engine = &io_engine_select;
+ engine = &io_engine_select;
else
- log_module(MAIN_LOG, LOG_FATAL, "No usable I/O engine found.");
+ log_module(MAIN_LOG, LOG_FATAL, "No usable I/O engine found.");
log_module(MAIN_LOG, LOG_DEBUG, "Using %s I/O engine.", engine->name);
}
log_module(MAIN_LOG, LOG_ERROR, "Somebody called ioset_add(%d) on a negative fd!", fd);
return 0;
}
+ if (!engine)
+ ioset_init();
res = calloc(1, sizeof(*res));
if (!res)
return 0;
res->fd = fd;
ioq_init(&res->send, 1024);
ioq_init(&res->recv, 1024);
- engine->add(res);
+#if defined(F_GETFL)
flags = fcntl(fd, F_GETFL);
fcntl(fd, F_SETFL, flags|O_NONBLOCK);
+ flags = fcntl(fd, F_GETFD);
+ fcntl(fd, F_SETFD, flags|FD_CLOEXEC);
+#else
+ /* I hope you're using the Win32 backend or something else that
+ * automatically marks the file descriptor non-blocking...
+ */
+ (void)flags;
+#endif
+ engine->add(res);
return res;
}
fd = socket(local ? local->sa_family : PF_INET, SOCK_STREAM, 0);
if (fd < 0) {
- log_module(MAIN_LOG, LOG_ERROR, "Unable to create listening socket: %s", strerror(errno));
- return NULL;
+ log_module(MAIN_LOG, LOG_ERROR, "Unable to create listening socket: %s", strerror(errno));
+ return NULL;
}
if (local && sa_size) {
- res = bind(fd, local, sa_size);
- if (res < 0) {
- log_module(MAIN_LOG, LOG_ERROR, "Unable to bind listening socket %d: %s", fd, strerror(errno));
- close(fd);
- return NULL;
- }
+ res = bind(fd, local, sa_size);
+ if (res < 0) {
+ log_module(MAIN_LOG, LOG_ERROR, "Unable to bind listening socket %d: %s", fd, strerror(errno));
+ close(fd);
+ return NULL;
+ }
opt = 1;
res = setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (const char*)&opt, sizeof(opt));
res = listen(fd, 1);
if (res < 0) {
- log_module(MAIN_LOG, LOG_ERROR, "Unable to listen on socket %d: %s", fd, strerror(errno));
- close(fd);
- return NULL;
+ log_module(MAIN_LOG, LOG_ERROR, "Unable to listen on socket %d: %s", fd, strerror(errno));
+ close(fd);
+ return NULL;
}
io_fd = ioset_add(fd);
if (!io_fd) {
- close(fd);
- return NULL;
+ close(fd);
+ return NULL;
}
io_fd->state = IO_LISTENING;
io_fd->data = data;
struct io_fd *
ioset_connect(struct sockaddr *local, unsigned int sa_size, const char *peer, unsigned int port, int blocking, void *data, void (*connect_cb)(struct io_fd *fd, int error)) {
- int fd, res;
+ struct addrinfo hints;
+ struct addrinfo *ai;
struct io_fd *io_fd;
- struct addrinfo hints, *ai;
+ struct io_fd *old_active;
+ int res;
+ int fd;
char portnum[10];
memset(&hints, 0, sizeof(hints));
case EHOSTUNREACH:
case ECONNREFUSED:
ioset_close(io_fd, 1);
- engine->update(io_fd);
return NULL;
}
}
io_fd->state = IO_CONNECTED;
+ old_active = active_fd;
if (connect_cb)
connect_cb(io_fd, ((res < 0) ? errno : 0));
- engine->update(io_fd);
+ if (active_fd)
+ engine->update(io_fd);
+ if (old_active != io_fd)
+ active_fd = old_active;
return io_fd;
}
+void ioset_update(struct io_fd *fd) {
+ engine->update(fd);
+}
+
static void
ioset_try_write(struct io_fd *fd) {
int res;
- unsigned int req = ioq_get_avail(&fd->send);
+ unsigned int req;
+
+ req = ioq_get_avail(&fd->send);
res = write(fd->fd, fd->send.buf+fd->send.get, req);
if (res < 0) {
switch (errno) {
void
ioset_close(struct io_fd *fdp, int os_close) {
if (!fdp)
- return;
+ return;
if (active_fd == fdp)
active_fd = NULL;
if (fdp->destroy_cb)
fdp->destroy_cb(fdp);
- if (fdp->send.get != fdp->send.put) {
+#if defined(HAVE_WSAEVENTSELECT)
+ /* This is one huge kludge. Sorry! */
+ if (fdp->send.get != fdp->send.put && (os_close & 2)) {
+ engine->remove(fdp, 0);
+ ioset_try_write(fdp);
+ /* it may need to send the beginning of the buffer now.. */
+ if (fdp->send.get != fdp->send.put)
+ ioset_try_write(fdp);
+ }
+ free(fdp->send.buf);
+ free(fdp->recv.buf);
+ if (os_close & 1)
+ closesocket(fdp->fd);
+#else
+ if (fdp->send.get != fdp->send.put && (os_close & 2)) {
int flags;
- flags = fcntl(fdp->fd, F_GETFL);
+ flags = fcntl(fdp->fd, F_GETFL);
fcntl(fdp->fd, F_SETFL, flags&~O_NONBLOCK);
ioset_try_write(fdp);
/* it may need to send the beginning of the buffer now.. */
}
free(fdp->send.buf);
free(fdp->recv.buf);
- if (os_close)
+ if (os_close & 1)
close(fdp->fd);
- engine->remove(fdp);
+ engine->remove(fdp, os_close & 1);
+#endif
free(fdp);
}
static void
ioset_accept(struct io_fd *listener)
{
- struct io_fd *old_active_fd;
+ struct io_fd *old_active;
struct io_fd *new_fd;
int fd;
fd = accept(listener->fd, NULL, 0);
if (fd < 0) {
- log_module(MAIN_LOG, LOG_ERROR, "Unable to accept new connection on listener %d: %s", listener->fd, strerror(errno));
- return;
+ log_module(MAIN_LOG, LOG_ERROR, "Unable to accept new connection on listener %d: %s", listener->fd, strerror(errno));
+ return;
}
new_fd = ioset_add(fd);
new_fd->state = IO_CONNECTED;
- old_active_fd = active_fd;
+ old_active = active_fd;
active_fd = new_fd;
listener->accept_cb(listener, new_fd);
assert(active_fd == NULL || active_fd == new_fd);
else
engine->update(new_fd);
}
- active_fd = old_active_fd;
+ active_fd = old_active;
}
static int
if (fd->recv.put == fd->recv.size)
fd->recv.put = 0;
fdnum = fd->fd;
- while (fd->wants_reads && (fd->line_len > 0)) {
+ while (fd->line_len > 0) {
struct io_fd *old_active;
+ int died = 0;
old_active = active_fd;
active_fd = fd;
fd->readable_cb(fd);
if (active_fd)
ioset_find_line_length(fd);
+ else
+ died = 1;
if (old_active != fd)
active_fd = old_active;
+ if (died)
+ break;
}
}
}
int
ioset_line_read(struct io_fd *fd, char *dest, int max) {
- int avail, done;
- if ((fd->state == IO_CLOSED) && (!ioq_get_avail(&fd->recv) || (fd->line_len < 0)))
+ int line_len;
+ int avail;
+ int done;
+
+ line_len = fd->line_len;
+ if ((fd->state == IO_CLOSED) && (!ioq_get_avail(&fd->recv) || (line_len < 0)))
return 0;
- if (fd->line_len < 0)
+ if (line_len < 0)
return -1;
- if (fd->line_len < max)
- max = fd->line_len;
+ if (line_len < max)
+ max = line_len;
avail = ioq_get_avail(&fd->recv);
if (max > avail) {
memcpy(dest, fd->recv.buf + fd->recv.get, avail);
- fd->recv.get += avail;
- assert(fd->recv.get == fd->recv.size);
+ assert(fd->recv.get + avail == fd->recv.size);
fd->recv.get = 0;
done = avail;
} else {
fd->recv.get += max - done;
if (fd->recv.get == fd->recv.size)
fd->recv.get = 0;
- dest[max] = 0;
+ dest[max - 1] = 0;
ioset_find_line_length(fd);
- return max;
+ return line_len;
}
void
ioset_events(struct io_fd *fd, int readable, int writable)
{
if (!fd || (!readable && !writable))
- return;
+ return;
active_fd = fd;
switch (fd->state) {
case IO_CLOSED:
break;
case IO_CONNECTING:
assert(active_fd == NULL || active_fd == fd);
- if (active_fd && writable) {
+ if (active_fd && readable) {
socklen_t arglen;
int rc;
arglen = sizeof(rc);
if (getsockopt(fd->fd, SOL_SOCKET, SO_ERROR, &rc, &arglen) < 0)
rc = errno;
- fd->state = IO_CONNECTED;
+ fd->state = IO_CLOSED;
if (fd->connect_cb)
fd->connect_cb(fd, rc);
- if (active_fd == fd)
- engine->update(fd);
+ } else if (active_fd && writable) {
+ fd->state = IO_CONNECTED;
+ if (fd->connect_cb)
+ fd->connect_cb(fd, 0);
}
+ if (active_fd != fd)
+ break;
+ engine->update(fd);
/* and fall through */
case IO_CONNECTED:
assert(active_fd == NULL || active_fd == fd);
timeout.tv_sec = wakey - now;
timeout.tv_usec = 0;
- if (engine->loop(&timeout))
- continue;
+ if (engine->loop(&timeout))
+ continue;
/* Call any timeq events we need to call. */
timeq_run();
projects / srvx.git / blobdiff