This is an automated email from the git hooks/post-receive script. It was generated because a ref change was pushed to the repository containing the project "aesop Repository". The branch, master has been updated via 3ec79f55ec2327ca8cd591398277aa4e369efdc0 (commit) via 0c3892fa5fe69291ca98d83fbdd6067ca158be95 (commit) via 16a8718a577148847ecd5c0af10003c3ba642c31 (commit) from 6334b6b00d0a665a09b44a770e44450506b9f9df (commit) Those revisions listed above that are new to this repository have not appeared on any other notification email; so we list those revisions in full, below. - Log ----------------------------------------------------------------- commit 3ec79f55ec2327ca8cd591398277aa4e369efdc0 Author: Dries Kimpe <[email protected]> Date: Tue Feb 7 13:36:21 2012 -0600 aesop socket module using resourcebuilder(untested) Code is not production quality. Use with caution. commit 0c3892fa5fe69291ca98d83fbdd6067ca158be95 Author: Dries Kimpe <[email protected]> Date: Mon Feb 6 14:32:06 2012 -0600 Add some more ae error codes AESOP_ERR_CANCELLED AESOP_ERR_OTHER commit 16a8718a577148847ecd5c0af10003c3ba642c31 Author: Dries Kimpe <[email protected]> Date: Mon Feb 6 14:31:54 2012 -0600 Add AE_MUTEX_INITIALIZER to ae-thread.h ----------------------------------------------------------------------- Summary of changes: ae-error.h | 2 + ae-thread.h | 2 + configure.ac | 2 + src/socket/aesocket.ae | 395 ++++++++++++++++++++++++++++++++++++++++++ src/socket/aesocket.hae | 80 +++++++++ src/socket/module.mk.in | 5 + src/socket/test/module.mk.in | 4 + src/socket/test/simple.ae | 169 ++++++++++++++++++ 8 files changed, 659 insertions(+), 0 deletions(-) create mode 100644 src/socket/aesocket.ae create mode 100644 src/socket/aesocket.hae create mode 100644 src/socket/module.mk.in create mode 100644 src/socket/test/module.mk.in create mode 100644 src/socket/test/simple.ae Diff of changes: diff --git a/ae-error.h b/ae-error.h index 0b5b97e..255f2ce 100644 --- a/ae-error.h +++ b/ae-error.h @@ -17,5 +17,7 @@ #define AE_ERR_EXIST (-4) /* object or entity already exists */ #define AE_ERR_TIMEDOUT (-5) /* timed out */ #define AE_ERR_OVERFLOW (-6) /* overflow of some resource limitation */ +#define AE_ERR_CANCELLED (-7) /* Call was cancelled. */ +#define AE_ERR_OTHER (-8) /* Other error */ #endif diff --git a/ae-thread.h b/ae-thread.h index eeb133b..fce17ed 100644 --- a/ae-thread.h +++ b/ae-thread.h @@ -8,6 +8,8 @@ */ #include "triton-thread.h" +#define AE_MUTEX_INITIALIZER TRITON_MUTEX_INITIALIZER + typedef triton_mutex_t ae_mutex_t; #define ae_mutex_init triton_mutex_init #define ae_mutex_lock triton_mutex_lock diff --git a/configure.ac b/configure.ac index edc99ff..ccd61a0 100644 --- a/configure.ac +++ b/configure.ac @@ -292,5 +292,7 @@ module.mk tests/module.mk parser/module.mk parser/tests/blocking/module.mk +src/socket/module.mk +src/socket/test/module.mk ]) diff --git a/src/socket/aesocket.ae b/src/socket/aesocket.ae new file mode 100644 index 0000000..2b98b98 --- /dev/null +++ b/src/socket/aesocket.ae @@ -0,0 +1,395 @@ +#include "aesocket.hae" +#include "resources/resourcebuilder/resourcebuilder.hae" + +#include <unistd.h> +#include <assert.h> +#include <fcntl.h> +#include <signal.h> +#include <errno.h> +#include "libev/ev.h" + +/** + * If true, before testing if the socket is ready, try to read/write under the + * assumption that it will be ready. When it is not ready, wait until ready + * and try again. + */ +static int immediate_completion = 1; + + +/* Number of zero byte reads in a row that we will observe before assuming + * that the peer has closed the socket; this is a hack to weed out + * spurious zero byte reads that can show up on some systems. + */ +#define ZERO_BYTE_READ_THRESHOLD 0 + + +static ae_mutex_t mutex = AE_MUTEX_INITIALIZER; +static pthread_cond_t cond = PTHREAD_COND_INITIALIZER; +static unsigned int refcount = 0; + +static pthread_t socket_thread; +static volatile sig_atomic_t socket_thread_done; +static struct ev_async socket_thread_wake; + +static struct ev_loop * socket_ev_loop = 0; + +enum { MODE_WAITING, MODE_COMPLETED, MODE_REMOVED }; +struct aesocket_entry_t +{ + ev_io watcher; + rb_slot_t slot; + struct aesocket_entry_t * next; + int flags; + int mode; +}; + +typedef struct aesocket_entry_t aesocket_entry_t; + +static aesocket_entry_t * socket_thread_add = 0; +static aesocket_entry_t * socket_thread_remove = 0; + +static void socket_thread_wake_cb (struct ev_loop * loop, ev_async * w, int revents) +{ +} + +static void aesocket_check_queue (void) +{ + ae_mutex_lock (&mutex); + + while (socket_thread_add != 0) + { + aesocket_entry_t * e = socket_thread_add; + ev_io_start (socket_ev_loop, &e->watcher); + socket_thread_add = socket_thread_add->next; + } + + while (socket_thread_remove != 0) + { + aesocket_entry_t * e = socket_thread_remove; + ev_io_stop (socket_ev_loop, &e->watcher); + e->mode = MODE_COMPLETED; + socket_thread_remove = socket_thread_remove->next; + } + + pthread_cond_broadcast (&cond); + ae_mutex_unlock (&mutex); +} + +static void aesocket_thread_wakeup (void) +{ + ev_async_send (socket_ev_loop, &socket_thread_wake); +} + +static void * aesocket_thread_main (void * arg) +{ + while (!socket_thread_done) + { + aesocket_check_queue (); + ev_loop (socket_ev_loop, EVLOOP_ONESHOT); + } +} + +int aesocket_init (void) +{ + ae_mutex_lock (&mutex); + if (!refcount++) + { + socket_ev_loop = ev_loop_new (EVFLAG_AUTO); + socket_thread_done = 0; + ev_async_init (&socket_thread_wake, &socket_thread_wake_cb); + + pthread_create (&socket_thread, 0, aesocket_thread_main, 0); + } + ae_mutex_unlock (&mutex); + return AE_SUCCESS; +} + +int aesocket_done (void) +{ + ae_mutex_lock (&mutex); + if (0 == --refcount) + { + socket_thread_done = 1; + aesocket_thread_wakeup (); + pthread_join (socket_thread, 0); + + ev_loop_destroy (socket_ev_loop); + } + ae_mutex_unlock (&mutex); + return AE_SUCCESS; +} + +/** + * libev callback for when fd is ready. + */ +static void aesocket_fd_ready (struct ev_loop * loop, struct ev_io * io, + int revents) +{ + aesocket_entry_t * e = io->data; + e->mode = MODE_COMPLETED; + rb_slot_complete (&e->slot); +} + +__blocking int aesocket_ready (int fd, int flags, int * err) +{ + int ret; + aesocket_entry_t entry; + + entry.flags = flags; + ev_io_init (&entry.watcher, &aesocket_fd_ready, fd, + (flags & AESOCKET_READ ? EV_READ : 0) + | (flags & AESOCKET_WRITE ? EV_WRITE : 0)); + + entry.watcher.data = &entry; + entry.mode = MODE_WAITING; + + rb_slot_initialize (&entry.slot); + + ae_mutex_lock (&mutex); + entry.next = socket_thread_add->next; + socket_thread_add = &entry; + ae_mutex_unlock (&mutex); + + aesocket_thread_wakeup (); + + ret = rb_slot_capture (&entry.slot); + if (ret != AE_SUCCESS) + { + /* note: could save time here by checking for completion */ + + int added = 1; + /* we're cancelled */ + + ae_mutex_lock (&mutex); + + /* check if the libev thread already added the fd */ + if (socket_thread_add == &entry) + { + socket_thread_add = entry.next; + added = 0; + } + else + { + aesocket_entry_t * cur = socket_thread_add; + while (cur) + { + if (cur->next == &entry) + { + cur->next = entry.next; + added = 0; + break; + } + cur = cur->next; + } + } + + if (added) + { + /* we need to add it to the remove list and wait + * until it is removed */ + entry.next = socket_thread_remove; + socket_thread_remove = &entry; + + while (entry.mode != MODE_REMOVED) + { + pthread_cond_wait (&cond, &mutex); + } + } + + ae_mutex_unlock (&mutex); + + ret = AE_ERR_CANCELLED; + } + else + { + /* completed OK */ + ret = AE_SUCCESS; + } + + rb_slot_destroy (&entry.slot); + *err = 0; + + return ret; +} + + +int aesocket_prepare (int fd) +{ + /* mark socket as nonblocking */ + return fcntl(fd, F_SETFL, O_NONBLOCK); +} + +__blocking int aesocket_accept( + int sockfd, + struct sockaddr *addr, + socklen_t * addrlen, + int *newfd, + int * err) +{ + int ret; + int r; + + *newfd = -1; + + *err = 0; + + while(1) + { + if(!immediate_completion) + { + ret = aesocket_ready(sockfd, AESOCKET_READ, err); + if (ret != AE_SUCCESS) + return ret; + } + + r = accept(sockfd, addr, addrlen); + if (r >= 0) + { + *newfd = r; + return(AE_SUCCESS); + } + else if (errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR) + { + *err = errno; + return AE_ERR_OTHER; + } + + if(immediate_completion) + { + ret = aesocket_ready(sockfd, AESOCKET_READ, err); + if (ret != AE_SUCCESS) + return ret; + } + } + + return AE_SUCCESS; +} + + +__blocking int aesocket_read( + int fd, + void *buf, + size_t count, + int *len, + int * err) +{ + int ret; + int r; + char* tmp_buf = (char*)buf; + int tmp_count = count; + int zero_reads = 0; + + assert(buf && count); + *len = 0; + + *err = 0; + + do + { + if(!immediate_completion) + { + ret = aesocket_ready(fd, AESOCKET_READ, err); + + if (ret != AE_SUCCESS) + return ret; + } + r = read(fd, tmp_buf, tmp_count); + + if(r > 0) + { + tmp_buf += r; + tmp_count -= r; + zero_reads = 0; + } + else if(r == 0) + { + zero_reads++; + if(zero_reads >= ZERO_BYTE_READ_THRESHOLD) + { + /* assume that the peer has closed the socket */ + *err = errno; + return(AE_ERR_OTHER); + } + } + else if(errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR) + { + *err = errno; + return AE_ERR_OTHER; + } + + if(immediate_completion && tmp_count > 0) + { + ret = aesocket_ready(fd, AESOCKET_READ, err); + + if (ret != AE_SUCCESS) + return ret; + } + + } while (tmp_count > 0); + + *len = count; + + return AE_SUCCESS; +} + +__blocking int aesocket_write( + int fd, + const void *buf, + size_t count, + int *len, + int * err) +{ + int ret; + int r; + const char* tmp_buf = (const char*)buf; + int tmp_count = count; + + assert(buf && count); + *len = 0; + + *err = 0; + + do + { + if(!immediate_completion) + { + ret = aesocket_ready(fd, AESOCKET_WRITE, err); + + if (ret != AE_SUCCESS) + return ret; + } + + r = write(fd, tmp_buf, tmp_count); + if(r >= 0) + { + tmp_buf += r; + tmp_count -= r; + } + else if(errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR) + { + *err = errno; + return AE_ERR_OTHER; + } + + if(immediate_completion && tmp_count > 0) + { + ret = aesocket_ready(fd, AESOCKET_WRITE, err); + + if (ret != AE_ERR_OTHER) + return ret; + } + + } while (tmp_count > 0); + *len = count; + + return AE_SUCCESS; +} + +/* + * Local variables: + * c-indent-level: 4 + * c-basic-offset: 4 + * End: + * + * vim: ts=8 sts=4 sw=4 expandtab + */ diff --git a/src/socket/aesocket.hae b/src/socket/aesocket.hae new file mode 100644 index 0000000..503b1a8 --- /dev/null +++ b/src/socket/aesocket.hae @@ -0,0 +1,80 @@ +#ifndef SOCKET_TRITON_SOCKET_HAE +#define SOCKET_TRITON_SOCKET_HAE + +#include <sys/types.h> +#include <sys/socket.h> +#include "aesop.h" + + +/** + * Prepare fd for use with the functions in the triton-socket module. + * This function might change the blocking/non-blocking nature of the fd. + **/ +int aesocket_prepare (int fd); + +/** initialize the socket module. + * Can be called multiple times + */ +int aesocket_init (void); + +/** + * Inform the socket module it is no longer needed. + */ +int aesocket_done (void); + + +/** + * Wait for a successful accept on the fd. + * Returns AE_ERR_CANCELLED if cancelled, + * AE_ERR_OTHER if the accept call returned an error + * (storing the error in *error), + * and AE_SUCCESS otherwise. + */ +__blocking int aesocket_accept( + int sockfd, + struct sockaddr *addr, + socklen_t * addrlen, + int *newfd, + int * err); + +/** + * Returns the number of bytes read in *ret. + * Returns errno on error, 0 otherwise. + */ +__blocking int aesocket_read( + int fd, + void *buf, + size_t count, + int *ret, + int * err); + +/** + * Returns the number of bytes written in *ret. + * Returns errno on error, otherwise 0. + */ +__blocking int aesocket_write( + int fd, + const void *buf, + size_t count, + int *ret, + int * err); + + +enum +{ + AESOCKET_READ = 0x01, + AESOCKET_WRITE = 0x02 +}; + +__blocking int aesocket_ready (int fd, int flags, int * err); + +#endif + +/* + * Local variables: + * c-indent-level: 4 + * c-basic-offset: 4 + * End: + * + * vim: ts=8 sts=4 sw=4 expandtab + */ diff --git a/src/socket/module.mk.in b/src/socket/module.mk.in new file mode 100644 index 0000000..1b86cbb --- /dev/null +++ b/src/socket/module.mk.in @@ -0,0 +1,5 @@ +DIR := src/socket + +AELIBSRC += $(DIR)/aesocket.ae +AESOP_HDR += $(DIR)/aesocket.hae + diff --git a/src/socket/test/module.mk.in b/src/socket/test/module.mk.in new file mode 100644 index 0000000..f7476ce --- /dev/null +++ b/src/socket/test/module.mk.in @@ -0,0 +1,4 @@ +DIR := src/socket/test + +AETESTSRC += $(DIR)/simple.ae + diff --git a/src/socket/test/simple.ae b/src/socket/test/simple.ae new file mode 100644 index 0000000..7ee9be2 --- /dev/null +++ b/src/socket/test/simple.ae @@ -0,0 +1,169 @@ +#include "aesop.h" +#include "aesop-support.hae" +#include "resources/timer/timer.hae" +#include "src/socket/aesocket.hae" + +#include <unistd.h> +#include <fcntl.h> +#include <stdio.h> +#include <assert.h> + +static __blocking int test1 () +{ + int fd; + char filename[128]; + strcpy (filename, "/tmp/aesocketXXXXXX"); + + fd = mkstemp (filename); + + pwait + { + pbranch + { + int err; + aesocket_ready (fd, AESOCKET_WRITE, &err); + aesop_cancel_branches_wait (); + } + pbranch + { + int tret; + tret = triton_timer (10000); + assert (tret != AE_SUCCESS && "File did not become read??"); + } + } + + close (fd); + + return 0; +} + +static __blocking int doSendRecv (int * fd) +{ + size_t todo = 1024*1024*64; + int ret = AE_SUCCESS; + pwait + { + pbranch + { + int i; + i=0; + char buf[4096]; + int iter2 = 0; + while (i < todo) + { + int r; + int err; + int tret; + ++iter2; + tret = aesocket_ready (fd[0], AESOCKET_READ, &err); + if (tret != AE_SUCCESS) + { + ret = tret; + printf ("Cancelled read!"); + pbreak; + } + printf ("R"); + r= read (fd[0], &buf[0], sizeof(buf)); + if (r< 0) + perror ("read"); + i += r; + } + } + pbranch + { + int i; + int err; + i=0; + char buf[1024]; + int w; + int iter = 0; + while (i< todo) + { + int tret; + ++iter; + tret = aesocket_ready (fd[1], AESOCKET_WRITE, &err); + if (tret != AE_SUCCESS) + { + ret = tret; + printf ("Cancelled write!"); + pbreak; + } + printf ("W"); + w = write (fd[1], &buf[0], sizeof(buf)); + if (w < 0) + perror ("write"); + i += w; + } + } + } + return ret; +} + +static __blocking int test2 () +{ + int fd[2]; + if (pipe (&fd[0]) < 0) + { + perror ("pipe"); + assert (0); + } + fcntl (fd[0], F_SETFL, O_NONBLOCK); + fcntl (fd[1], F_SETFL, O_NONBLOCK); + + pwait + { + pbranch + { + int ret; + ret = doSendRecv (fd); + if (ret == AE_SUCCESS) + { + printf ("Test done... Cancelling timeout!\n"); + aesop_cancel_branches_wait (); + } + else + { + printf ("Test cancelled!\n"); + } + } + pbranch + { + int ret; + ret = triton_timer (10000); + if (ret == AE_SUCCESS) + { + printf ("Cancelling!"); + aesop_cancel_branches_wait (); + } + else + { + printf ("Cancelling timer\n"); + } + } + } + + close (fd[0]); + close (fd[1]); + + return 0; +} + +static __blocking int dotest (int argc, char ** args) +{ + aesocket_init (); + test1 (); + test2 (); + aesocket_done (); + return 0; +} + +aesop_main_set(dotest); + +/* + * Local variables: + * c-indent-level: 4 + * c-basic-offset: 4 + * End: + * + * vim: ft=c ts=8 sts=4 sw=4 expandtab + */ hooks/post-receive -- aesop Repository