add experimental iom API for multithreaded I/O multiplexing (in io.h)
parent
b3bccb9d88
commit
ce595ae0cc
@ -0,0 +1,19 @@
|
||||
.TH iom_abort 3
|
||||
.SH NAME
|
||||
iom_abort \- abort all pending iom_wait calls
|
||||
.SH SYNTAX
|
||||
.B #include <libowfat/io.h>
|
||||
|
||||
int \fBiom_abort\fP(iomux_t* c);
|
||||
.SH DESCRIPTION
|
||||
\fIiom_abort\fR will cause all currently running instances of
|
||||
\fIiom_wait\fR to return immediately with return value -2.
|
||||
|
||||
.SH "LINKING"
|
||||
You may have to add \fI-lpthread\fR to the command line in the linking
|
||||
step.
|
||||
|
||||
.SH "RETURN VALUE"
|
||||
iom_abort returns 0 on success and -1 on error, setting errno.
|
||||
.SH "SEE ALSO"
|
||||
iom_init, iom_add, iom_wait
|
@ -0,0 +1,10 @@
|
||||
#include "io_internal.h"
|
||||
|
||||
int iom_abort(iomux_t* c) {
|
||||
c->working=-2;
|
||||
#ifdef __dietlibc__
|
||||
return cnd_broadcast(&c->sem);
|
||||
#else
|
||||
return sem_post(&c->sem);
|
||||
#endif
|
||||
}
|
@ -0,0 +1,33 @@
|
||||
.TH iom_add 3
|
||||
.SH NAME
|
||||
iom_add \- add event to I/O multiplexer
|
||||
.SH SYNTAX
|
||||
.B #include <libowfat/io.h>
|
||||
|
||||
int \fBiom_add\fP(iomux_t* c, int64 fd, unsigned int events);
|
||||
.SH DESCRIPTION
|
||||
iom_add adds an event you are interested in to an I/O multiplexer.
|
||||
|
||||
\fIfd\fR is the file descriptor (usually a socket) you are interested
|
||||
in, and \fIevents\fR is the operation you want to do. It can be IOM_READ
|
||||
or IOM_WRITE.
|
||||
|
||||
If that operation becomes possible on that descriptor, and some thread
|
||||
is calling \fIiom_wait\fR at the time, it will return and tell you the
|
||||
fd and the event.
|
||||
|
||||
Note that the event registration is removed from the iomux_t context if
|
||||
it occurs. You will have to call \fIiom_wait\fR again after you handled
|
||||
the event, if you are still interested in it.
|
||||
|
||||
Closing a file descriptor with registered events will discard the event
|
||||
registration.
|
||||
|
||||
.SH "LINKING"
|
||||
You may have to add \fI-lpthread\fR to the command line in the linking
|
||||
step.
|
||||
|
||||
.SH "RETURN VALUE"
|
||||
iom_add returns 0 on success and -1 on error, setting errno.
|
||||
.SH "SEE ALSO"
|
||||
iom_init, iom_wait, iom_abort
|
@ -0,0 +1,28 @@
|
||||
#include "io_internal.h"
|
||||
#ifdef HAVE_EPOLL
|
||||
#include <sys/epoll.h>
|
||||
#endif
|
||||
#ifdef HAVE_KQUEUE
|
||||
#include <sys/types.h>
|
||||
#include <sys/event.h>
|
||||
#include <sys/time.h>
|
||||
#endif
|
||||
|
||||
int iom_add(iomux_t* c,int64 s,unsigned int events) {
|
||||
#ifdef HAVE_EPOLL
|
||||
struct epoll_event e = { .events=EPOLLONESHOT, .data.fd=s };
|
||||
if (events & IOM_READ) e.events|=EPOLLIN;
|
||||
if (events & IOM_WRITE) e.events|=EPOLLOUT;
|
||||
return epoll_ctl(c->ctx, EPOLL_CTL_ADD, s, &e);
|
||||
#elif defined(HAVE_KQUEUE)
|
||||
struct kevent kev;
|
||||
struct timespec ts = { 0 };
|
||||
EV_SET(&kev, s,
|
||||
(events & IOM_READ ? EVFILT_READ : 0) +
|
||||
(events & IOM_WRITE ? EVFILT_WRITE : 0),
|
||||
EV_ADD | EV_ENABLE | EV_ONESHOT, 0, 0, (void*)s);
|
||||
return kevent(c->ctx, &kev, 1, 0, 0, &ts);
|
||||
#else
|
||||
#warning "only epoll and kqueue supported for now"
|
||||
#endif
|
||||
}
|
@ -0,0 +1,30 @@
|
||||
.TH iom_init 3
|
||||
.SH NAME
|
||||
iom_init \- create new I/O multiplexer
|
||||
.SH SYNTAX
|
||||
.B #include <libowfat/io.h>
|
||||
|
||||
int \fBiom_init\fP(iomux_t* c);
|
||||
.SH DESCRIPTION
|
||||
iom_init initializes an I/O multiplexer.
|
||||
|
||||
An I/O multiplexer is a context that can be used to do I/O multiplexing
|
||||
with support for multiple threads. Add events to a multiplexer using
|
||||
\fIiom_add\fR, and then get the next available event with
|
||||
\fIiom_wait\fR. If you are done and want to signal all the threads
|
||||
something, set a volatile global variable to tell the threads to stop
|
||||
and then fall \fIiom_abort\fR to tell all pending iom_wait operations in
|
||||
all threads to return immediately.
|
||||
|
||||
After \fIiom_init\fR is done, \fIiom_add\fR and \fIiom_wait\fR can be
|
||||
called from different threads on the same context, and they will
|
||||
synchronize internally.
|
||||
|
||||
.SH "LINKING"
|
||||
You may have to add \fI-lpthread\fR to the command line in the linking
|
||||
step.
|
||||
|
||||
.SH "RETURN VALUE"
|
||||
iom_init returns 0 on success and -1 on error, setting errno.
|
||||
.SH "SEE ALSO"
|
||||
iom_add, iom_wait, iom_abort
|
@ -0,0 +1,38 @@
|
||||
#include "io_internal.h"
|
||||
#ifdef HAVE_EPOLL
|
||||
#include <sys/epoll.h>
|
||||
#endif
|
||||
#ifdef HAVE_KQUEUE
|
||||
#include <sys/types.h>
|
||||
#include <sys/event.h>
|
||||
#include <sys/time.h>
|
||||
#endif
|
||||
|
||||
int iom_init(iomux_t* c) {
|
||||
#ifdef HAVE_EPOLL
|
||||
c->ctx = epoll_create1(EPOLL_CLOEXEC);
|
||||
#elif defined(HAVE_KQUEUE)
|
||||
if ((c->ctx = kqueue()) != -1) {
|
||||
if (fcntl(c->ctx,F_SETFD,FD_CLOEXEC) == -1) {
|
||||
close(c->ctx);
|
||||
c->ctx=-1;
|
||||
}
|
||||
}
|
||||
#else
|
||||
#warning "only epoll and kqueue supported for now"
|
||||
#endif
|
||||
unsigned int i;
|
||||
c->working=0;
|
||||
c->h=c->l=0; /* no elements in queue */
|
||||
for (i=0; i<SLOTS; ++i) {
|
||||
c->q[i].fd=-1;
|
||||
c->q[i].events=0;
|
||||
}
|
||||
#ifdef __dietlibc__
|
||||
mtx_init(&c->mtx, mtx_timed);
|
||||
cnd_init(&c->sem);
|
||||
#else
|
||||
sem_init(&c->sem, 0, 1);
|
||||
#endif
|
||||
return (c->ctx!=-1);
|
||||
}
|
@ -0,0 +1,38 @@
|
||||
.TH iom_wait 3
|
||||
.SH NAME
|
||||
iom_wait \- wait for event from I/O multiplexer
|
||||
.SH SYNTAX
|
||||
.B #include <libowfat/io.h>
|
||||
|
||||
int \fBiom_wait\fP(iomux_t* c,
|
||||
int64* fd, unsigned int* events,
|
||||
unsigned long timeout);
|
||||
.SH DESCRIPTION
|
||||
iom_wait will wait for events registered to the I/O multiplexer with
|
||||
\fIiom_add\fR. It will wait \fItimeout\fR milliseconds.
|
||||
|
||||
If during that time any of the registered events occur, \fIiom_wait\fR
|
||||
will set \fIfd\fR to the file descriptor the event happened on, and
|
||||
\fIevents\fR to the sum of IOM_READ, IOM_WRITE and IOM_ERROR, depending
|
||||
on what event actually happened, and return 1.
|
||||
|
||||
If nothing happens during that time, it will return 0 and leave \fIfd\fR
|
||||
and \fIevents\fR alone.
|
||||
|
||||
Note that the event registration is removed from the iomux_t context if
|
||||
it occurs. You will have to call \fIiom_wait\fR again after you handled
|
||||
the event, if you are still interested in it.
|
||||
|
||||
Closing a file descriptor with registered events will discard the event
|
||||
registration.
|
||||
|
||||
.SH "LINKING"
|
||||
You may have to add \fI-lpthread\fR to the command line in the linking
|
||||
step.
|
||||
|
||||
.SH "RETURN VALUE"
|
||||
iom_wait returns 1 on success, 0 if there was a timeout, and -1 on
|
||||
error, setting errno. If \fIiom_abort\fR was called on the I/O
|
||||
multiplexer context, it will return -2.
|
||||
.SH "SEE ALSO"
|
||||
iom_init, iom_add, iom_abort
|
@ -0,0 +1,124 @@
|
||||
#include "io_internal.h"
|
||||
#ifdef HAVE_EPOLL
|
||||
#include <sys/epoll.h>
|
||||
#endif
|
||||
#ifdef HAVE_KQUEUE
|
||||
#include <sys/types.h>
|
||||
#include <sys/event.h>
|
||||
#include <sys/time.h>
|
||||
#endif
|
||||
#include <errno.h>
|
||||
|
||||
int iom_wait(iomux_t* c,int64* s,unsigned int* revents,unsigned long timeout) {
|
||||
for (;;) {
|
||||
/* If we have an event in the queue, use that one */
|
||||
int r;
|
||||
if (c->working==-2) return -2; /* iomux was aborted */
|
||||
for (;;) {
|
||||
unsigned int f=c->l;
|
||||
if (f == c->h)
|
||||
break; /* no elements in queue */
|
||||
int n=(f+1)%SLOTS;
|
||||
if (__sync_bool_compare_and_swap(&c->l,f,n)) {
|
||||
/* we got one, and its index is in f */
|
||||
*s=c->q[f].fd;
|
||||
*revents=c->q[f].events;
|
||||
}
|
||||
/* collided with another thread, try again */
|
||||
}
|
||||
/* The queue was empty. If someone else is already calling
|
||||
* epoll_wait/kevent, then use the semaphore */
|
||||
if (__sync_bool_compare_and_swap(&c->working,0,1)) {
|
||||
/* we have the job to fill the struct. */
|
||||
int freeslots = (c->h - c->l);
|
||||
if (!freeslots) freeslots=SLOTS;
|
||||
|
||||
#ifdef HAVE_EPOLL
|
||||
struct epoll_event ee[SLOTS];
|
||||
int i;
|
||||
r=epoll_wait(c->ctx, ee, freeslots, timeout);
|
||||
if (r<=0) {
|
||||
/* we ran into a timeout, so let someone else take over */
|
||||
c->working=0;
|
||||
#ifdef __dietlibc__
|
||||
cnd_broadcast(&c->sem);
|
||||
#else
|
||||
sem_post(&c->sem);
|
||||
#endif
|
||||
return r;
|
||||
}
|
||||
for (i=0; i<r; ++i) {
|
||||
/* convert events */
|
||||
int e = ((ee[i].events & (EPOLLIN|EPOLLHUP|EPOLLERR)) ? IOM_READ : 0) |
|
||||
((ee[i].events & (EPOLLOUT|EPOLLHUP|EPOLLERR)) ? IOM_WRITE : 0) |
|
||||
((ee[i].events & EPOLLERR) ? IOM_ERROR : 0);
|
||||
if (i+1==r) {
|
||||
/* return last event instead of enqueueing it */
|
||||
*s=ee[i].data.fd;
|
||||
*revents=e;
|
||||
} else {
|
||||
c->q[c->h].fd=ee[i].data.fd;
|
||||
c->q[c->h].events=e;
|
||||
c->h = (c->h + 1) % SLOTS;
|
||||
}
|
||||
}
|
||||
#elif defined(HAVE_KQUEUE)
|
||||
struct kevent kev[SLOTS];
|
||||
struct timespec ts = { .tv_sec=timeout/1000, .tv_nsec=(timeout%1000)*1000000 };
|
||||
int r=kevent(c->ctx, 0, 0, &kev, freeslots, &ts);
|
||||
if (r<=0) {
|
||||
/* we ran into a timeout, so let someone else take over */
|
||||
c->working=0;
|
||||
#ifdef __dietlibc__
|
||||
cnd_broadcast(&c->sem);
|
||||
#else
|
||||
sem_post(&c->sem);
|
||||
#endif
|
||||
return r;
|
||||
}
|
||||
for (i=0; i<r; ++i) {
|
||||
/* convert events */
|
||||
int e = (kev[i].filter == EVFILT_READ ? IOM_READ : 0) |
|
||||
(kev[i].filter == EVFILT_WRITE ? IOM_WRITE : 0);
|
||||
if (i+1==r) {
|
||||
/* return last event instead of enqueueing it */
|
||||
*s=kev.ident;
|
||||
*revents=e;
|
||||
} else {
|
||||
c->q[c->h].fd=kev[i].ident;
|
||||
c->q[c->h].events=e;
|
||||
c->h = (c->h + 1) % SLOTS;
|
||||
}
|
||||
}
|
||||
#else
|
||||
#warning "only epoll and kqueue supported for now"
|
||||
#endif
|
||||
/* We need to signal the other threads.
|
||||
Either there are other events left, or we need one of them to
|
||||
wake up and call epoll_wait/kevent next, because we aren't
|
||||
doing it anymore */
|
||||
c->working=0;
|
||||
#ifdef __dietlibc__
|
||||
cnd_signal(&c->sem);
|
||||
#else
|
||||
sem_post(&c->sem);
|
||||
#endif
|
||||
return 1;
|
||||
} else {
|
||||
/* somebody else has the job to fill the queue */
|
||||
struct timespec ts;
|
||||
ts.tv_sec = timeout / 1000;
|
||||
ts.tv_nsec = (timeout % 1000) * 1000000;
|
||||
#ifdef __dietlibc__
|
||||
r=cnd_timedwait(&c->sem,&c->mtx,&ts);
|
||||
#else
|
||||
r=sem_timedwait(&c->sem,&ts);
|
||||
#endif
|
||||
if (r==-1) {
|
||||
if (errno==ETIMEDOUT) return 0;
|
||||
return -1;
|
||||
}
|
||||
/* fall through into next loop iteration */
|
||||
}
|
||||
}
|
||||
}
|
Loading…
Reference in New Issue