diff options
| -rw-r--r-- | src/core/msqueue.c | 111 | ||||
| -rw-r--r-- | src/core/nng_impl.h | 41 | ||||
| -rw-r--r-- | src/core/panic.c | 2 | ||||
| -rw-r--r-- | src/core/socket.c | 40 | ||||
| -rw-r--r-- | src/platform/platform.h | 31 | ||||
| -rw-r--r-- | src/platform/posix/posix_clock.h | 149 | ||||
| -rw-r--r-- | src/platform/posix/posix_config.h | 60 | ||||
| -rw-r--r-- | src/platform/posix/posix_impl.h | 4 | ||||
| -rw-r--r-- | src/platform/posix/posix_synch.h | 82 |
9 files changed, 489 insertions, 31 deletions
diff --git a/src/core/msqueue.c b/src/core/msqueue.c new file mode 100644 index 00000000..3e4ade79 --- /dev/null +++ b/src/core/msqueue.c @@ -0,0 +1,111 @@ +/* + * Copyright 2016 Garrett D'Amore <garrett@damore.org> + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), + * to deal in the Software without restriction, including without limitation + * the rights to use, copy, modify, merge, publish, distribute, sublicense, + * and/or sell copies of the Software, and to permit persons to whom + * the Software is furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included + * in all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL + * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS + * IN THE SOFTWARE. + */ + +#include "../nng.h" + +#include "nng_impl.h" + +/* + * Message queue. These operate in some respects like Go channels, + * but as we have access to the internals, we have made some fundamental + * differences and improvements. For example, these can grow, and either + * side can close, and they may be closed more than once. + */ + +struct nni_msgqueue { + nni_mutex_t mq_lock; + nni_cond_t mq_readable; + nni_cond_t mq_writeable; + int mq_cap; + int mq_len; + int mq_get; + int mq_put; + int mq_closed; + nng_msg_t *mq_msgs; +}; + +int +nni_msgqueue_create(nni_msgqueue_t *mqp, int cap) +{ + struct nni_msgqueue *mq; + int rv; + + if (cap < 1) { + return (NNG_EINVAL); + } + if ((mq = nni_alloc(sizeof (*mq))) == NULL) { + return (NNG_ENOMEM); + } + if ((rv = nni_mutex_create(&mq->mq_lock)) != 0) { + nni_free(mq, sizeof (*mq)); + return (rv); + } + if ((rv = nni_cond_create(&mq->mq_readable, mq->mq_lock)) != 0) { + nni_mutex_destroy(mq->mq_lock); + nni_free(mq, sizeof (*mq)); + return (NNG_ENOMEM); + } + if ((rv = nni_cond_create(&mq->mq_writeable, mq->mq_lock)) != 0) { + nni_cond_destroy(mq->mq_readable); + nni_mutex_destroy(mq->mq_lock); + return (NNG_ENOMEM); + } + if ((mq->mq_msgs = nni_alloc(sizeof (nng_msg_t) * cap)) == NULL) { + nni_cond_destroy(mq->mq_writeable); + nni_cond_destroy(mq->mq_readable); + nni_mutex_destroy(mq->mq_lock); + return (NNG_ENOMEM); + } + + mq->mq_cap = cap; + mq->mq_len = 0; + mq->mq_get = 0; + mq->mq_put = 0; + mq->mq_closed = 0; + *mqp = mq; + + return (0); +} + +void +nni_msgqueue_destroy(nni_msgqueue_t mq) +{ + nng_msg_t msg; + + nni_cond_destroy(mq->mq_writeable); + nni_cond_destroy(mq->mq_readable); + nni_mutex_destroy(mq->mq_lock); + + /* Free any orphaned messages. */ + while (mq->mq_len > 0) { + msg = mq->mq_msgs[mq->mq_get]; + mq->mq_get++; + if (mq->mq_get > mq->mq_cap) { + mq->mq_get = 0; + } + mq->mq_len--; + nng_msg_free(msg); + } + + nni_free(mq->mq_msgs, mq->mq_cap * sizeof (nng_msg_t)); + nni_free(mq, sizeof (*mq)); +} diff --git a/src/core/nng_impl.h b/src/core/nng_impl.h index a25b87a4..a62ac101 100644 --- a/src/core/nng_impl.h +++ b/src/core/nng_impl.h @@ -61,4 +61,45 @@ extern void nni_snprintf(char *, size_t, const char *, ...); */ extern void nni_panic(const char *, ...); +/* + * Message queues. Message queues work in some ways like Go channels; + * they are a thread-safe way to pass messages between subsystems. + */ +typedef struct nni_msgqueue *nni_msgqueue_t; + +/* + * nni_msgqueue_create creates a message queue with the given capacity, + * which must be a positive number. It returns NNG_EINVAL if the capacity + * is invalid, or NNG_ENOMEM if resources cannot be allocated. + */ +extern int nni_msgqueue_create(nni_msgqueue_t *, int); + +/* + * nni_msgqueue_destroy destroys a message queue. It will also free any + * messages that may be in the queue. + */ +extern void nni_msgqueue_destroy(nni_msgqueue_t); + +extern int nni_msgqueue_len(nni_msgqueue_t); +extern int nni_msgqueue_cap(nni_msgqueue_t); + +/* + * nni_msgqueue_put attempts to put a message to the queue. It will wait + * for the timeout (us), if the value is positive. If the value is negative + * then it will wait forever. If the value is zero, it will just check, and + * return immediately whether a message can be put or not. Valid returns are + * NNG_ECLOSED if the queue is closed or NNG_ETIMEDOUT if the message cannot + * be placed after a time, or NNG_EAGAIN if the operation cannot succeed + * immediately and a zero timeout is specified. Note that timeout granularity + * may be limited -- for example Windows systems have a millisecond resolution + * timeout capability. + */ +extern int nni_msgqueue_put(nni_msgqueue_t, nng_msg_t, int); + +/* + * nni_msgqueue_get gets the message from the queue, using a timeout just + * like nni_msgqueue_put. + */ +extern int nni_msgqueue_get(nni_msgqueue_t, nng_msg_t *, int); + #endif /* NNG_IMPL_H */ diff --git a/src/core/panic.c b/src/core/panic.c index e993ddb9..60c790ac 100644 --- a/src/core/panic.c +++ b/src/core/panic.c @@ -28,8 +28,6 @@ #include <execinfo.h> #endif -#include "nng.h" - #include "nng_impl.h" /* diff --git a/src/core/socket.c b/src/core/socket.c new file mode 100644 index 00000000..247c5640 --- /dev/null +++ b/src/core/socket.c @@ -0,0 +1,40 @@ +/* + * Copyright 2016 Garrett D'Amore <garrett@damore.org> + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), + * to deal in the Software without restriction, including without limitation + * the rights to use, copy, modify, merge, publish, distribute, sublicense, + * and/or sell copies of the Software, and to permit persons to whom + * the Software is furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included + * in all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL + * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS + * IN THE SOFTWARE. + */ + +#include "../nng.h" + +#include "nng_impl.h" + +/* + * Socket implementation. + */ + +struct nng_socket { + int s_proto; + nni_mutex_t s_mx; + + /* uwq */ + /* urq */ + /* options */ + /* pipes */ + /* endpoints */ +}; diff --git a/src/platform/platform.h b/src/platform/platform.h index 36ec207c..0d958c65 100644 --- a/src/platform/platform.h +++ b/src/platform/platform.h @@ -23,10 +23,16 @@ #ifndef PLATFORM_H #define PLATFORM_H -#include <stdlib.h> -#include <stdint.h> +/* + * We require some standard C header files. The only one of these that might + * be problematic is <stdint.h>, which is required for C99. Older versions + * of the Windows compilers might not have this. However, latest versions of + * MS Studio have a functional <stdint.h>. If this impacts you, just upgrade + * your tool chain. + */ #include <stdarg.h> -#include <string.h> +#include <stddef.h> +#include <stdint.h> /* * These are the APIs that a platform must implement to support nng. @@ -113,11 +119,24 @@ void nni_cond_wait(nni_cond_t); /* * nni_cond_timedwait waits for a wakeup on the condition variable, just * as with nni_condwait, but it will also wake after the given number of - * milliseconds has passed. (This is a relative timed wait.) Early + * microseconds has passed. (This is a relative timed wait.) Early * wakeups are permitted, and the caller must take care to double check any * conditions. The return value is 0 on success, or an error code, which - * can be NNG_ETIMEDOUT. + * can be NNG_ETIMEDOUT. Note that it is permissible to wait for longer + * than the timeout based on the resolution of your system clock. + */ +int nni_cond_timedwait(nni_cond_t, uint64_t); + +/* + * nn_clock returns a number of microseconds since some arbitrary time + * in the past. The values returned by nni_clock may be used with + * nni_cond_timedwait. + */ +uint64_t nni_clock(void); + +/* + * nni_usleep sleeps for the specified number of microseconds (at least). */ -int nnp_cond_timedwait(nni_cond_t, int); +void nni_usleep(uint64_t); #endif /* PLATFORM_H */ diff --git a/src/platform/posix/posix_clock.h b/src/platform/posix/posix_clock.h new file mode 100644 index 00000000..07fc603d --- /dev/null +++ b/src/platform/posix/posix_clock.h @@ -0,0 +1,149 @@ +/* + * Copyright 2016 Garrett D'Amore <garrett@damore.org> + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), + * to deal in the Software without restriction, including without limitation + * the rights to use, copy, modify, merge, publish, distribute, sublicense, + * and/or sell copies of the Software, and to permit persons to whom + * the Software is furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included + * in all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL + * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS + * IN THE SOFTWARE. + */ + +/* + * This is more of a direct #include of a .c rather than .h file. + * But having it be a .h makes compiler rules work out properly. Do + * not include this more than once into your program, or you will + * get multiple symbols defined. + */ + +/* + * POSIX clock stuff. + */ + +#include <time.h> +#include <errno.h> +#include <string.h> + +#ifndef NNG_USE_GETTIMEOFDAY + +/* + * Use POSIX realtime stuff. + */ + +uint64_t +nni_clock(void) +{ + struct timespec ts; + uint64_t usec; + + if (clock_gettime(NNG_USE_CLOCKID, &ts) != 0) { + /* This should never ever occur. */ + nni_panic("clock_gettime failed: %s", strerror(errno)); + } + + usec = ts.tv_sec; + usec *= 1000000; + usec += (ts.tv_nsec / 1000); + return (usec); +} + + +void +nni_usleep(uint64_t usec) +{ + struct timespec ts; + + ts.tv_sec = usec / 1000000; + ts.tv_nsec = (usec % 1000000) * 1000; + + /* Do this in a loop, so that interrupts don't actually wake us. */ + while (ts.tv_sec || ts.tv_nsec) { + (void) nanosleep(&ts, &ts); + } +} + +#else /* NNG_USE_GETTIMEOFDAY */ + +/* + * If you're here, its because you don't have a modern clock_gettime with + * monotonic clocks, or the necessary pthread_condattr_settclock(). In + * this case, you should be advised that *bad* things can happen if your + * system clock changes time while programs using this library are running. + * (Basically, timeouts can take longer or shorter, leading to either hangs + * or apparent spurious errors. Eventually it should all sort itself out, + * but if you change the clock by a large amount you might wonder what the + * heck is happening until it does.) + */ + +#include <pthread.h> +#include <sys/time.h> + +uint64_t +nni_clock(void) +{ + uint64_t usec; + + struct timeval tv; + if (gettimeofday(&tv, NULL) != 0) { + nni_panic("gettimeofday failed: %s", strerror(errno)); + } + + usec = tv.tv_sec; + usec *= 1000000; + usec += tv.tv_usec; + return (usec); +} + +void +nni_usleep(uint64_t usec) +{ + /* + * So probably there is no nanosleep. We could in theory use + * pthread condition variables, but that means doing memory + * allocation, or forcing the use of pthreads where the platform + * might be preferring the use of another threading package. + * Additionally, use of pthreads means that we cannot use + * relative times in a clock_settime safe manner. + * So we can use poll() instead, which is rather coarse, but + * pretty much guaranteed to work. + */ + struct pollfd pfd; + uint64_t now; + uint64_t expire; + + /* + * Possibly we could pass NULL instead of pfd, but passing a valid + * pointer ensures that if the system dereferences the pointer it + * won't come back with EFAULT. + */ + pfd.fd = -1; + pfd.events = 0; + + now = nni_clock(); + expire = now + usec; + + while (now < expire) { + /* + * In theory we could round up to a whole number of msec, + * but under the covers poll already does some rounding up, + * and the loop above guarantees that we will not bail out + * early. So this gives us a better chance to avoid adding + * nearly an extra unneeded millisecond to the wait. + */ + (void) poll(&pfd, 0, (int)((expire - now) / 1000)); + now = nni_clock(); + } +} + +#endif /* NNG_USE_GETTIMEOFDAY */ diff --git a/src/platform/posix/posix_config.h b/src/platform/posix/posix_config.h new file mode 100644 index 00000000..1e83a948 --- /dev/null +++ b/src/platform/posix/posix_config.h @@ -0,0 +1,60 @@ +/* + * Copyright 2016 Garrett D'Amore <garrett@damore.org> + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), + * to deal in the Software without restriction, including without limitation + * the rights to use, copy, modify, merge, publish, distribute, sublicense, + * and/or sell copies of the Software, and to permit persons to whom + * the Software is furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included + * in all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL + * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS + * IN THE SOFTWARE. + */ + +/* + * The following adjustments to the platform may be defined. These can + * be defined in either platform/config.h or loaded in via external + * defines using cmake. + * + * #define NNG_USE_GETTIMEOFDAY + * This macro is defined if you lack a working clock_gettime, + * nanosleep, or pthread_condattr_setclock. In this case the + * library uses the system clock for relative sleeps, timers, etc. + * This can be dangerous if the system clock is changed, so only + * use this if you have no other choice. If it appears that + * the system lacks clock_gettime, then it will choose this automatically. + * This value may be ignored on platforms that don't use POSIX clocks. + * + * #define NNG_USE_CLOCKID + * This macro may be defined to a different clock id (see + * clock_gettime()). By default we use CLOCK_MONOTONIC if it exists, + * or CLOCK_REALTIME otherwise. This is ignored if NNG_USE_GETTIMEOFDAY + * is defined. Platforms that don't use POSIX clocks will probably + * ignore any setting here. + * + * #define NNG_HAVE_BACKTRACE + * If your system has a working backtrace(), and backtrace_symbols(), + * along with <execinfo.h>, you can define this to get richer backtrace + * information for debugging. + */ + +#include <time.h> + +#ifndef CLOCK_REALTIME +#define NNG_USE_GETTIMEOFDAY +#elif !defined(NNG_USE_CLOCKID) +#ifdef CLOCK_MONOTONIC +#define NNG_USE_CLOCKID CLOCK_MONOTONIC +#else +#define NNG_USE_CLOCKID CLOCK_REALTIME +#endif +#endif /* CLOCK_REALTIME */ diff --git a/src/platform/posix/posix_impl.h b/src/platform/posix/posix_impl.h index 291e3f2e..89022e60 100644 --- a/src/platform/posix/posix_impl.h +++ b/src/platform/posix/posix_impl.h @@ -31,11 +31,11 @@ */ #ifdef PLATFORM_POSIX - +#include "platform/posix/posix_config.h" #include "platform/posix/posix_debug.h" #include "platform/posix/posix_alloc.h" +#include "platform/posix/posix_clock.h" #include "platform/posix/posix_synch.h" /* #include "platform/posix/posix_thread.h" */ #include "platform/posix/posix_vsnprintf.h" - #endif diff --git a/src/platform/posix/posix_synch.h b/src/platform/posix/posix_synch.h index 0eae133e..b3b15fee 100644 --- a/src/platform/posix/posix_synch.h +++ b/src/platform/posix/posix_synch.h @@ -28,7 +28,8 @@ */ /* - * POSIX synchronization (mutexes and condition variables). + * POSIX synchronization (mutexes and condition variables). This uses + * pthreads. */ #include <pthread.h> @@ -117,19 +118,73 @@ nni_mutex_tryenter(nni_mutex_t m) return (0); } + +int +cond_attr(pthread_condattr_t **attrpp) +{ +#if defined(NNG_USE_GETTIMEOFDAY) || NNG_USE_CLOCKID == CLOCK_REALTIME + *attrpp = NULL; + return (0); +#else + /* In order to make this fast, avoid reinitializing attrs. */ + static pthread_condattr_t attr; + static pthread_mutex_t mx = PTHREAD_MUTEX_INITIALIZER; + static int init = 0; + int rv; + + /* + * For efficiency's sake, we try to reuse the same attr for the + * life of the library. This avoids many reallocations. Technically + * this means that we will leak the attr on exit(), but this is + * preferable to constantly allocating and reallocating it. + */ + if (init) { + *attrpp = &attr; + return (0); + } + + (void) pthread_mutex_lock(&mx); + while (!init) { + if ((rv = pthread_condattr_init(&attr)) != 0) { + (void) pthread_mutex_unlock(&mx); + return (NNG_ENOMEM); + } + rv = pthread_condattr_setclock(&attr, NNG_USE_CLOCKID); + if (rv != 0) { + nni_panic("condattr_setclock: %s", strerror(rv)); + } + init = 1; + } + (void) pthread_mutex_unlock(&mx); + *attrpp = &attr; + return (0); +#endif +} + int nni_cond_create(nni_cond_t *cvp, nni_mutex_t mx) { + /* + * By preference, we use a CLOCK_MONOTONIC version of condition + * variables, which insulates us from changes to the system time. + */ struct nni_cond *c; + pthread_condattr_t *attrp; + int rv; + + if ((rv = cond_attr(&attrp)) != 0) { + return (rv); + } if ((c = nni_alloc(sizeof (*c))) == NULL) { return (NNG_ENOMEM); } c->mx = &mx->mx; - if (pthread_cond_init(&c->cv, NULL) != 0) { + if (pthread_cond_init(&c->cv, attrp) != 0) { /* In theory could be EAGAIN, but handle like ENOMEM */ nni_free(c, sizeof (*c)); return (NNG_ENOMEM); } + *cvp = c; return (0); } @@ -168,30 +223,15 @@ nni_cond_wait(nni_cond_t c) } int -nni_cond_timedwait(nni_cond_t c, int msec) +nni_cond_timedwait(nni_cond_t c, uint64_t usec) { struct timespec ts; int rv; - /* POSIX says clock_gettime exists since SUSv2 at least. */ - - rv = clock_gettime(CLOCK_REALTIME, &ts); - if (rv != 0) { - /* - * If the clock_gettime() is not working, its a problem with - * the platform. Arguably we could use gettimeofday instead, - * but for now we just panic(). We can fix this when someone - * finds a platform that returns ENOSYS here. - */ - nni_panic("clock_gettime failed: %s", strerror(errno)); - } - ts.tv_nsec += (msec * 1000000); + usec += nni_clock(); - /* Normalize -- its not clear if this is strictly necessary. */ - while (ts.tv_nsec > 1000000000) { - ts.tv_nsec -= 1000000000; - ts.tv_sec++; - } + ts.tv_sec = usec / 1000000; + ts.tv_nsec = (usec % 10000) * 1000; if ((rv = pthread_cond_timedwait(&c->cv, c->mx, &ts)) != 0) { if (rv == ETIMEDOUT) { |