aboutsummaryrefslogtreecommitdiff
path: root/src/core
diff options
context:
space:
mode:
Diffstat (limited to 'src/core')
-rw-r--r--src/core/aio.c320
-rw-r--r--src/core/aio.h32
-rw-r--r--src/core/endpt.c11
-rw-r--r--src/core/msgqueue.c103
-rw-r--r--src/core/taskq.c1
5 files changed, 209 insertions, 258 deletions
diff --git a/src/core/aio.c b/src/core/aio.c
index e6157786..792b63f2 100644
--- a/src/core/aio.c
+++ b/src/core/aio.c
@@ -11,22 +11,50 @@
#include "core/nng_impl.h"
#include <string.h>
-enum nni_aio_flags {
- NNI_AIO_INIT = 0x1,
- NNI_AIO_DONE = 0x2,
- NNI_AIO_FINI = 0x4,
-};
-
+static nni_mtx nni_aio_lk;
// These are used for expiration.
-static nni_mtx nni_aio_expire_mtx;
static nni_cv nni_aio_expire_cv;
static int nni_aio_expire_exit;
-static nni_list nni_aio_expire_aios;
static nni_thr nni_aio_expire_thr;
-static nni_aio *nni_aio_expire_current;
+static nni_list nni_aio_expire_aios;
+
+// Design notes.
+//
+// AIOs are only ever "completed" by the provider, which must call
+// one of the nni_aio_finish variants. Until this occurs, the provider
+// guarantees that the AIO is valid. The provider must guarantee that
+// an AIO will be "completed" (with a call to nni_aio_finish & friends)
+// exactly once.
+//
+// Note that the cancellation routine may be called by the framework
+// several times. The framework (or the consumer) guarantees that the
+// AIO will remain valid across these calls, so that the provider is
+// free to examine the aio for list membership, etc. The provider must
+// not call finish more than once though.
+//
+// A single lock, nni_aio_lk, is used to protect the flags on the AIO,
+// as well as the expire list on the AIOs. We will not permit an AIO
+// to be marked done if an expiration is outstanding.
+//
+// In order to synchronize with the expiration, we set a flag when we
+// are going to cancel due to expiration, and then let the expiration
+// thread dispatch the notification to the user (after ensuring that
+// the provider is done with the aio.) This ensures that the completion
+// task will be dispatch *exactly* once, and only after nothing in
+// the provider or the framework is using it further. (The consumer
+// will probably still be using, but if the consumer calls nni_aio_wait
+// or nni_aio_stop, then the consumer will have exclusive access to it.
+// Provided, of course, that the consumer does not reuse the aio for
+// another operation in the callback.)
+//
+// In order to guard against aio reuse during teardown, we set a fini
+// flag. Any attempt to initialize for a new operation after that point
+// will fail and the caller will get NNG_ESTATE indicating this. The
+// provider that calls nni_aio_start() MUST check the return value, and
+// if it comes back nonzero (NNG_ESTATE) then it must simply discard the
+// request and return.
static void nni_aio_expire_add(nni_aio *);
-static void nni_aio_expire_remove(nni_aio *);
int
nni_aio_init(nni_aio *aio, nni_cb cb, void *arg)
@@ -34,15 +62,11 @@ nni_aio_init(nni_aio *aio, nni_cb cb, void *arg)
int rv;
memset(aio, 0, sizeof(*aio));
- if ((rv = nni_mtx_init(&aio->a_lk)) != 0) {
- return (rv);
- }
- if ((rv = nni_cv_init(&aio->a_cv, &aio->a_lk)) != 0) {
- nni_mtx_fini(&aio->a_lk);
+ if ((rv = nni_cv_init(&aio->a_cv, &nni_aio_lk)) != 0) {
return (rv);
}
aio->a_expire = NNI_TIME_NEVER;
- aio->a_flags = NNI_AIO_INIT;
+ aio->a_init = 1;
nni_task_init(NULL, &aio->a_task, cb, arg);
return (0);
@@ -55,7 +79,6 @@ nni_aio_fini(nni_aio *aio)
// At this point the AIO is done.
nni_cv_fini(&aio->a_cv);
- nni_mtx_fini(&aio->a_lk);
if ((aio->a_naddrs != 0) && (aio->a_addrs != NULL)) {
NNI_FREE_STRUCTS(aio->a_addrs, aio->a_naddrs);
@@ -71,30 +94,23 @@ nni_aio_fini(nni_aio *aio)
void
nni_aio_stop(nni_aio *aio)
{
- if ((aio->a_flags & NNI_AIO_INIT) == 0) {
+ if (!aio->a_init) {
// Never initialized, so nothing should have happened.
return;
}
- nni_mtx_lock(&aio->a_lk);
- aio->a_flags |= NNI_AIO_FINI; // this prevents us from being scheduled
- nni_mtx_unlock(&aio->a_lk);
+ nni_mtx_lock(&nni_aio_lk);
+ aio->a_fini = 1;
+ nni_mtx_unlock(&nni_aio_lk);
nni_aio_cancel(aio, NNG_ECANCELED);
- // Wait for any outstanding task to complete. We won't schedule
- // new stuff because nni_aio_start will fail (due to AIO_FINI).
- nni_task_wait(&aio->a_task);
+ nni_aio_wait(aio);
}
int
nni_aio_result(nni_aio *aio)
{
- int rv;
-
- nni_mtx_lock(&aio->a_lk);
- rv = aio->a_result;
- nni_mtx_unlock(&aio->a_lk);
- return (rv);
+ return (aio->a_result);
}
size_t
@@ -106,131 +122,116 @@ nni_aio_count(nni_aio *aio)
void
nni_aio_wait(nni_aio *aio)
{
- nni_mtx_lock(&aio->a_lk);
- while ((aio->a_flags & (NNI_AIO_DONE | NNI_AIO_FINI)) == 0) {
+ nni_mtx_lock(&nni_aio_lk);
+ while ((aio->a_active) && (!aio->a_done)) {
nni_cv_wait(&aio->a_cv);
}
- nni_mtx_unlock(&aio->a_lk);
+ nni_mtx_unlock(&nni_aio_lk);
nni_task_wait(&aio->a_task);
}
int
-nni_aio_start(nni_aio *aio, void (*cancel)(nni_aio *), void *data)
+nni_aio_start(nni_aio *aio, nni_aio_cancelfn cancelfn, void *data)
{
- nni_mtx_lock(&aio->a_lk);
- aio->a_flags &= ~NNI_AIO_DONE;
- if (aio->a_flags & NNI_AIO_FINI) {
+ nni_mtx_lock(&nni_aio_lk);
+ if (aio->a_fini) {
// We should not reschedule anything at this point.
- nni_mtx_unlock(&aio->a_lk);
+ aio->a_active = 0;
+ aio->a_result = NNG_ECANCELED;
+ nni_mtx_unlock(&nni_aio_lk);
return (NNG_ECANCELED);
}
+ aio->a_done = 0;
+ aio->a_pend = 0;
aio->a_result = 0;
aio->a_count = 0;
- aio->a_prov_cancel = cancel;
+ aio->a_prov_cancel = cancelfn;
aio->a_prov_data = data;
+ aio->a_active = 1;
if (aio->a_expire != NNI_TIME_NEVER) {
nni_aio_expire_add(aio);
}
- nni_mtx_unlock(&aio->a_lk);
+ nni_mtx_unlock(&nni_aio_lk);
return (0);
}
+// nni_aio_cancel is called by a consumer which guarantees that the aio
+// is still valid.
void
nni_aio_cancel(nni_aio *aio, int rv)
{
- void (*cancelfn)(nni_aio *);
-
- nni_mtx_lock(&aio->a_lk);
- if (aio->a_flags & NNI_AIO_DONE) {
- // The operation already completed - so there's nothing
- // left for us to do.
- nni_mtx_unlock(&aio->a_lk);
- return;
- }
- aio->a_flags |= NNI_AIO_DONE;
- aio->a_result = rv;
- cancelfn = aio->a_prov_cancel;
- aio->a_prov_cancel = NULL;
-
- aio->a_refcnt++;
- nni_mtx_unlock(&aio->a_lk);
+ nni_aio_cancelfn cancelfn;
- // Guaraneed to just be a list operation.
- nni_aio_expire_remove(aio);
+ nni_mtx_lock(&nni_aio_lk);
+ cancelfn = aio->a_prov_cancel;
+ nni_mtx_unlock(&nni_aio_lk);
// Stop any I/O at the provider level.
if (cancelfn != NULL) {
- cancelfn(aio);
+ cancelfn(aio, rv);
}
-
- nni_mtx_lock(&aio->a_lk);
-
- aio->a_refcnt--;
- nni_cv_wake(&aio->a_cv);
-
- // These should have already been cleared by the cancel function.
- aio->a_prov_data = NULL;
- aio->a_prov_cancel = NULL;
-
- nni_task_dispatch(&aio->a_task);
- nni_mtx_unlock(&aio->a_lk);
}
// I/O provider related functions.
-int
-nni_aio_finish(nni_aio *aio, int result, size_t count)
+static void
+nni_aio_finish_impl(
+ nni_aio *aio, int result, size_t count, void *pipe, nni_msg *msg)
{
- nni_mtx_lock(&aio->a_lk);
- if (aio->a_flags & NNI_AIO_DONE) {
- // Operation already done (canceled or timed out?)
- nni_mtx_unlock(&aio->a_lk);
- return (NNG_ESTATE);
- }
- aio->a_flags |= NNI_AIO_DONE;
+ nni_mtx_lock(&nni_aio_lk);
+ NNI_ASSERT(aio->a_pend == 0); // provider only calls us *once*
+
+ nni_list_node_remove(&aio->a_expire_node);
+ aio->a_pend = 1;
aio->a_result = result;
aio->a_count = count;
aio->a_prov_cancel = NULL;
aio->a_prov_data = NULL;
+ if (pipe) {
+ aio->a_pipe = pipe;
+ }
+ if (msg) {
+ aio->a_msg = msg;
+ }
- // This is guaranteed to just be a list operation at this point,
- // because done wasn't set.
- nni_aio_expire_remove(aio);
aio->a_expire = NNI_TIME_NEVER;
- nni_cv_wake(&aio->a_cv);
- nni_task_dispatch(&aio->a_task);
- nni_mtx_unlock(&aio->a_lk);
- return (0);
+ // If we are expiring, then we rely on the expiration thread to
+ // complete this; we must not because the expiration thread is
+ // still holding the reference.
+ if (!aio->a_expiring) {
+ aio->a_done = 1;
+ nni_cv_wake(&aio->a_cv);
+ nni_task_dispatch(&aio->a_task);
+ }
+ nni_mtx_unlock(&nni_aio_lk);
}
-int
-nni_aio_finish_pipe(nni_aio *aio, int result, void *pipe)
+void
+nni_aio_finish(nni_aio *aio, int result, size_t count)
{
- nni_mtx_lock(&aio->a_lk);
- if (aio->a_flags & NNI_AIO_DONE) {
- // Operation already done (canceled or timed out?)
- nni_mtx_unlock(&aio->a_lk);
- return (NNG_ESTATE);
- }
- aio->a_flags |= NNI_AIO_DONE;
+ nni_aio_finish_impl(aio, result, count, NULL, NULL);
+}
- aio->a_result = result;
- aio->a_count = 0;
- aio->a_prov_cancel = NULL;
- aio->a_prov_data = NULL;
- aio->a_pipe = pipe;
+void
+nni_aio_finish_error(nni_aio *aio, int result)
+{
+ nni_aio_finish_impl(aio, result, 0, NULL, NULL);
+}
- // This is guaranteed to just be a list operation at this point,
- // because done wasn't set.
- nni_aio_expire_remove(aio);
- aio->a_expire = NNI_TIME_NEVER;
- nni_cv_wake(&aio->a_cv);
+void
+nni_aio_finish_pipe(nni_aio *aio, void *pipe)
+{
+ NNI_ASSERT(pipe != NULL);
+ nni_aio_finish_impl(aio, 0, 0, pipe, NULL);
+}
- nni_task_dispatch(&aio->a_task);
- nni_mtx_unlock(&aio->a_lk);
- return (0);
+void
+nni_aio_finish_msg(nni_aio *aio, nni_msg *msg)
+{
+ NNI_ASSERT(msg != NULL);
+ nni_aio_finish_impl(aio, 0, nni_msg_len(msg), NULL, msg);
}
void
@@ -261,12 +262,9 @@ nni_aio_list_active(nni_aio *aio)
static void
nni_aio_expire_add(nni_aio *aio)
{
- nni_mtx * mtx = &nni_aio_expire_mtx;
- nni_cv * cv = &nni_aio_expire_cv;
nni_list *list = &nni_aio_expire_aios;
nni_aio * naio;
- nni_mtx_lock(mtx);
// This is a reverse walk of the list. We're more likely to find
// a match at the end of the list.
for (naio = nni_list_last(list); naio != NULL;
@@ -280,105 +278,69 @@ nni_aio_expire_add(nni_aio *aio)
// This has the shortest time, so insert at the start.
nni_list_prepend(list, aio);
// And, as we are the latest, kick the thing.
- nni_cv_wake(cv);
+ nni_cv_wake(&nni_aio_expire_cv);
}
- nni_mtx_unlock(mtx);
-}
-
-static void
-nni_aio_expire_remove(nni_aio *aio)
-{
- nni_mtx * mtx = &nni_aio_expire_mtx;
- nni_cv * cv = &nni_aio_expire_cv;
- nni_list *list = &nni_aio_expire_aios;
-
- nni_mtx_lock(mtx);
- if (nni_list_active(list, aio)) {
- nni_list_remove(list, aio);
- }
- while (aio == nni_aio_expire_current) {
- nni_cv_wait(cv);
- }
- nni_mtx_unlock(mtx);
}
static void
nni_aio_expire_loop(void *arg)
{
- nni_mtx * mtx = &nni_aio_expire_mtx;
- nni_cv * cv = &nni_aio_expire_cv;
- nni_list *aios = &nni_aio_expire_aios;
- nni_aio * aio;
- nni_time now;
-
- void (*cancelfn)(nni_aio *);
+ nni_list * aios = &nni_aio_expire_aios;
+ nni_aio * aio;
+ nni_time now;
+ nni_aio_cancelfn cancelfn;
NNI_ARG_UNUSED(arg);
for (;;) {
- nni_mtx_lock(mtx);
-
- // If we are resuming this loop after processing an AIO,
- // note that we are done with it, and wake anyone waiting
- // for that to clear up.
- if ((aio = nni_aio_expire_current) != NULL) {
- nni_aio_expire_current = NULL;
- nni_cv_wake(cv);
- }
+ nni_mtx_lock(&nni_aio_lk);
if (nni_aio_expire_exit) {
- nni_mtx_unlock(mtx);
+ nni_mtx_unlock(&nni_aio_lk);
return;
}
if ((aio = nni_list_first(aios)) == NULL) {
- nni_cv_wait(cv);
- nni_mtx_unlock(mtx);
+ nni_cv_wait(&nni_aio_expire_cv);
+ nni_mtx_unlock(&nni_aio_lk);
continue;
}
now = nni_clock();
if (now < aio->a_expire) {
// Unexpired; the list is ordered, so we just wait.
- nni_cv_until(cv, aio->a_expire);
- nni_mtx_unlock(mtx);
+ nni_cv_until(&nni_aio_expire_cv, aio->a_expire);
+ nni_mtx_unlock(&nni_aio_lk);
continue;
}
// This aio's time has come. Expire it, canceling any
// outstanding I/O.
-
nni_list_remove(aios, aio);
- nni_aio_expire_current = aio;
- nni_mtx_unlock(mtx);
-
- cancelfn = NULL;
- nni_mtx_lock(&aio->a_lk);
- if ((aio->a_flags & (NNI_AIO_DONE | NNI_AIO_FINI)) != 0) {
- nni_mtx_unlock(&aio->a_lk);
- continue;
- }
+ // Mark it as expiring. This acts as a hold on
+ // the aio, similar to the consumers. The actual taskq
+ // dispatch on completion won't occur until this is cleared,
+ // and the done flag won't be set either.
+ aio->a_expiring = 1;
+ cancelfn = aio->a_prov_cancel;
- aio->a_flags |= NNI_AIO_DONE;
-
- aio->a_result = NNG_ETIMEDOUT;
- cancelfn = aio->a_prov_cancel;
- aio->a_prov_cancel = NULL;
- nni_cv_wake(&aio->a_cv);
- nni_mtx_unlock(&aio->a_lk);
-
- // Cancel any outstanding activity.
+ // Cancel any outstanding activity. This is always non-NULL
+ // for a valid aio, and becomes NULL only when an AIO is
+ // already being canceled or finished.
if (cancelfn != NULL) {
- cancelfn(aio);
+ nni_mtx_unlock(&nni_aio_lk);
+ cancelfn(aio, NNG_ETIMEDOUT);
+ nni_mtx_lock(&nni_aio_lk);
}
- // Arguably we could avoid dispatching, and execute the
- // callback inline here as we are already on a separate
- // thread. But keeping it separate is clearer, and more
- // consistent with other uses. And this should not be a
- // hot code path.
+ NNI_ASSERT(aio->a_pend); // nni_aio_finish was run
+ NNI_ASSERT(aio->a_prov_cancel == NULL);
+ aio->a_expiring = 0;
+ aio->a_done = 1;
+ nni_cv_wake(&aio->a_cv);
nni_task_dispatch(&aio->a_task);
+ nni_mtx_unlock(&nni_aio_lk);
}
}
@@ -386,7 +348,7 @@ int
nni_aio_sys_init(void)
{
int rv;
- nni_mtx *mtx = &nni_aio_expire_mtx;
+ nni_mtx *mtx = &nni_aio_lk;
nni_cv * cv = &nni_aio_expire_cv;
nni_thr *thr = &nni_aio_expire_thr;
@@ -409,7 +371,7 @@ fail:
void
nni_aio_sys_fini(void)
{
- nni_mtx *mtx = &nni_aio_expire_mtx;
+ nni_mtx *mtx = &nni_aio_lk;
nni_cv * cv = &nni_aio_expire_cv;
nni_thr *thr = &nni_aio_expire_thr;
diff --git a/src/core/aio.h b/src/core/aio.h
index 31a54f12..0f41c01f 100644
--- a/src/core/aio.h
+++ b/src/core/aio.h
@@ -18,6 +18,8 @@
typedef struct nni_aio_ops nni_aio_ops;
+typedef void (*nni_aio_cancelfn)(nni_aio *, int);
+
// An nni_aio is an async I/O handle.
struct nni_aio {
int a_result; // Result code (nng_errno)
@@ -25,10 +27,14 @@ struct nni_aio {
nni_time a_expire;
// These fields are private to the aio framework.
- nni_mtx a_lk;
nni_cv a_cv;
- unsigned a_flags;
- int a_refcnt; // prevent use-after-free
+ unsigned a_init : 1; // initialized flag
+ unsigned a_fini : 1; // shutting down (no new operations)
+ unsigned a_done : 1; // operation has completed
+ unsigned a_pend : 1; // completion routine pending
+ unsigned a_active : 1; // aio was started
+ unsigned a_expiring : 1; // expiration callback in progress
+ unsigned a_pad : 27; // ensure 32-bit alignment
nni_task a_task;
// Read/write operations.
@@ -47,9 +53,9 @@ struct nni_aio {
int a_naddrs;
// Provider-use fields.
- void (*a_prov_cancel)(nni_aio *);
- void * a_prov_data;
- nni_list_node a_prov_node;
+ nni_aio_cancelfn a_prov_cancel;
+ void * a_prov_data;
+ nni_list_node a_prov_node;
// Expire node.
nni_list_node a_expire_node;
@@ -106,21 +112,17 @@ extern void nni_aio_list_remove(nni_aio *);
extern int nni_aio_list_active(nni_aio *);
// nni_aio_finish is called by the provider when an operation is complete.
-// The provider gives the result code (0 for success, an NNG errno otherwise),
-// and the amount of data transferred (if any). If the return code is
-// non-zero, it indicates that the operation failed (usually because the aio
-// was already canceled.) This is important for providers that need to
-// prevent resources (new pipes for example) from accidentally leaking
-// during close operations.
-extern int nni_aio_finish(nni_aio *, int, size_t);
-extern int nni_aio_finish_pipe(nni_aio *, int, void *);
+extern void nni_aio_finish(nni_aio *, int, size_t);
+extern void nni_aio_finish_error(nni_aio *, int);
+extern void nni_aio_finish_pipe(nni_aio *, void *);
+extern void nni_aio_finish_msg(nni_aio *, nni_msg *);
// nni_aio_cancel is used to cancel an operation. Any pending I/O or
// timeouts are canceled if possible, and the callback will be returned
// with the indicated result (NNG_ECLOSED or NNG_ECANCELED is recommended.)
extern void nni_aio_cancel(nni_aio *, int rv);
-extern int nni_aio_start(nni_aio *, void (*)(nni_aio *), void *);
+extern int nni_aio_start(nni_aio *, nni_aio_cancelfn, void *);
// nni_aio_stop is used to abort all further operations on the AIO.
// When this is executed, no further operations or callbacks will be
diff --git a/src/core/endpt.c b/src/core/endpt.c
index c0f9d399..aa4c5a31 100644
--- a/src/core/endpt.c
+++ b/src/core/endpt.c
@@ -175,6 +175,7 @@ nni_ep_close(nni_ep *ep)
nni_aio_cancel(&ep->ep_acc_aio, NNG_ECLOSED);
nni_aio_cancel(&ep->ep_con_aio, NNG_ECLOSED);
nni_aio_cancel(&ep->ep_con_syn, NNG_ECLOSED);
+ nni_aio_cancel(&ep->ep_backoff, NNG_ECLOSED);
// Stop the underlying transport.
ep->ep_ops.ep_close(ep->ep_data);
@@ -188,6 +189,7 @@ nni_ep_reap(nni_ep *ep)
nni_aio_stop(&ep->ep_acc_aio);
nni_aio_stop(&ep->ep_con_aio);
nni_aio_stop(&ep->ep_con_syn);
+ nni_aio_stop(&ep->ep_backoff);
// Take us off the sock list.
nni_sock_ep_remove(ep->ep_sock, ep);
@@ -233,6 +235,13 @@ nni_ep_stop(nni_ep *ep)
}
static void
+nni_ep_backoff_cancel(nni_aio *aio, int rv)
+{
+ // The only way this ever gets "finished", is via cancellation.
+ nni_aio_finish_error(aio, rv);
+}
+
+static void
nni_ep_backoff_start(nni_ep *ep)
{
nni_duration backoff;
@@ -255,7 +264,7 @@ nni_ep_backoff_start(nni_ep *ep)
// random number, but this really doesn't matter.
ep->ep_backoff.a_expire = nni_clock() + (nni_random() % backoff);
- nni_aio_start(&ep->ep_backoff, NULL, ep);
+ nni_aio_start(&ep->ep_backoff, nni_ep_backoff_cancel, ep);
}
static void
diff --git a/src/core/msgqueue.c b/src/core/msgqueue.c
index d98c68be..2ebc4927 100644
--- a/src/core/msgqueue.c
+++ b/src/core/msgqueue.c
@@ -121,26 +121,17 @@ nni_msgq_fini(nni_msgq *mq)
NNI_FREE_STRUCT(mq);
}
-static void
-nni_msgq_finish(nni_aio *aio, int rv)
-{
- nni_aio_list_remove(aio);
- nni_aio_finish(aio, rv, 0);
-}
-
void
nni_msgq_set_get_error(nni_msgq *mq, int error)
{
- nni_aio *naio;
nni_aio *aio;
// Let all pending blockers know we are closing the queue.
nni_mtx_lock(&mq->mq_lock);
if (error != 0) {
- naio = nni_list_first(&mq->mq_aio_getq);
- while ((aio = naio) != NULL) {
- naio = nni_list_next(&mq->mq_aio_getq, aio);
- nni_msgq_finish(aio, error);
+ while ((aio = nni_list_first(&mq->mq_aio_getq)) != NULL) {
+ nni_aio_list_remove(aio);
+ nni_aio_finish_error(aio, error);
}
}
mq->mq_geterr = error;
@@ -150,16 +141,14 @@ nni_msgq_set_get_error(nni_msgq *mq, int error)
void
nni_msgq_set_put_error(nni_msgq *mq, int error)
{
- nni_aio *naio;
nni_aio *aio;
// Let all pending blockers know we are closing the queue.
nni_mtx_lock(&mq->mq_lock);
if (error != 0) {
- naio = nni_list_first(&mq->mq_aio_putq);
- while ((aio = naio) != NULL) {
- naio = nni_list_next(&mq->mq_aio_putq, aio);
- nni_msgq_finish(aio, error);
+ while ((aio = nni_list_first(&mq->mq_aio_putq)) != NULL) {
+ nni_aio_list_remove(aio);
+ nni_aio_finish_error(aio, error);
}
}
mq->mq_puterr = error;
@@ -169,21 +158,15 @@ nni_msgq_set_put_error(nni_msgq *mq, int error)
void
nni_msgq_set_error(nni_msgq *mq, int error)
{
- nni_aio *naio;
nni_aio *aio;
// Let all pending blockers know we are closing the queue.
nni_mtx_lock(&mq->mq_lock);
if (error != 0) {
- naio = nni_list_first(&mq->mq_aio_getq);
- while ((aio = naio) != NULL) {
- naio = nni_list_next(&mq->mq_aio_getq, aio);
- nni_msgq_finish(aio, error);
- }
- naio = nni_list_first(&mq->mq_aio_putq);
- while ((aio = naio) != NULL) {
- naio = nni_list_next(&mq->mq_aio_putq, aio);
- nni_msgq_finish(aio, error);
+ while (((aio = nni_list_first(&mq->mq_aio_getq)) != NULL) ||
+ ((aio = nni_list_first(&mq->mq_aio_putq)) != NULL)) {
+ nni_aio_list_remove(aio);
+ nni_aio_finish_error(aio, error);
}
}
mq->mq_puterr = error;
@@ -207,11 +190,12 @@ nni_msgq_run_putq(nni_msgq *mq)
// the queue is empty, otherwise it would have just taken
// data from the queue.
if ((raio = nni_list_first(&mq->mq_aio_getq)) != NULL) {
- raio->a_msg = msg;
waio->a_msg = NULL;
- nni_msgq_finish(raio, 0);
- nni_msgq_finish(waio, 0);
+ nni_aio_list_remove(raio);
+ nni_aio_list_remove(waio);
+ nni_aio_finish(waio, 0, len);
+ nni_aio_finish_msg(raio, msg);
continue;
}
@@ -224,7 +208,7 @@ nni_msgq_run_putq(nni_msgq *mq)
}
mq->mq_len++;
waio->a_msg = NULL;
- nni_msgq_finish(waio, 0);
+ nni_aio_finish(waio, 0, len);
continue;
}
@@ -243,14 +227,14 @@ nni_msgq_run_getq(nni_msgq *mq)
while ((raio = nni_list_first(&mq->mq_aio_getq)) != NULL) {
// If anything is waiting in the queue, get it first.
if (mq->mq_len != 0) {
- nni_list_remove(&mq->mq_aio_getq, raio);
msg = mq->mq_msgs[mq->mq_get++];
if (mq->mq_get == mq->mq_alloc) {
mq->mq_get = 0;
}
mq->mq_len--;
raio->a_msg = msg;
- nni_msgq_finish(raio, 0);
+ nni_aio_list_remove(raio);
+ nni_aio_finish_msg(raio, msg);
continue;
}
@@ -258,9 +242,11 @@ nni_msgq_run_getq(nni_msgq *mq)
if ((waio = nni_list_first(&mq->mq_aio_putq)) != NULL) {
msg = waio->a_msg;
waio->a_msg = NULL;
- raio->a_msg = msg;
- nni_msgq_finish(raio, 0);
- nni_msgq_finish(waio, 0);
+ nni_aio_list_remove(waio);
+ nni_aio_list_remove(raio);
+
+ nni_aio_finish(waio, 0, nni_msg_len(msg));
+ nni_aio_finish_msg(raio, msg);
continue;
}
@@ -300,16 +286,15 @@ nni_msgq_run_notify(nni_msgq *mq)
}
static void
-nni_msgq_cancel(nni_aio *aio)
+nni_msgq_cancel(nni_aio *aio, int rv)
{
nni_msgq *mq = aio->a_prov_data;
- if (mq == NULL) {
- return;
- }
-
nni_mtx_lock(&mq->mq_lock);
- nni_aio_list_remove(aio);
+ if (nni_aio_list_active(aio)) {
+ nni_aio_list_remove(aio);
+ nni_aio_finish_error(aio, rv);
+ }
nni_mtx_unlock(&mq->mq_lock);
}
@@ -346,12 +331,12 @@ nni_msgq_aio_put(nni_msgq *mq, nni_aio *aio)
return;
}
if (mq->mq_closed) {
- nni_aio_finish(aio, NNG_ECLOSED, 0);
+ nni_aio_finish_error(aio, NNG_ECLOSED);
nni_mtx_unlock(&mq->mq_lock);
return;
}
if (mq->mq_puterr) {
- nni_aio_finish(aio, mq->mq_puterr, 0);
+ nni_aio_finish_error(aio, mq->mq_puterr);
nni_mtx_unlock(&mq->mq_lock);
return;
}
@@ -372,12 +357,12 @@ nni_msgq_aio_get(nni_msgq *mq, nni_aio *aio)
return;
}
if (mq->mq_closed) {
- nni_aio_finish(aio, NNG_ECLOSED, 0);
+ nni_aio_finish_error(aio, NNG_ECLOSED);
nni_mtx_unlock(&mq->mq_lock);
return;
}
if (mq->mq_geterr) {
- nni_aio_finish(aio, mq->mq_geterr, 0);
+ nni_aio_finish_error(aio, mq->mq_geterr);
nni_mtx_unlock(&mq->mq_lock);
return;
}
@@ -439,9 +424,7 @@ nni_msgq_tryput(nni_msgq *mq, nni_msg *msg)
if ((raio = nni_list_first(&mq->mq_aio_getq)) != NULL) {
nni_list_remove(&mq->mq_aio_getq, raio);
- raio->a_msg = msg;
-
- nni_aio_finish(raio, 0, len);
+ nni_aio_finish_msg(raio, msg);
nni_mtx_unlock(&mq->mq_lock);
return (0);
}
@@ -512,13 +495,16 @@ nni_msgq_drain(nni_msgq *mq, nni_time expire)
break;
}
}
- // If we timedout, free any remaining messages in the queue.
- // Also complete the putq as NNG_ECLOSED.
+ // Timed out or writers drained.
+
+ // Complete the putq as NNG_ECLOSED.
while ((aio = nni_list_first(&mq->mq_aio_putq)) != NULL) {
nni_aio_list_remove(aio);
- nni_aio_finish(aio, NNG_ECLOSED, 0);
+ nni_aio_finish_error(aio, NNG_ECLOSED);
}
+
+ // Free any remaining messages in the queue.
while (mq->mq_len > 0) {
nni_msg *msg = mq->mq_msgs[mq->mq_get++];
if (mq->mq_get > mq->mq_alloc) {
@@ -551,17 +537,10 @@ nni_msgq_close(nni_msgq *mq)
// Let all pending blockers know we are closing the queue.
naio = nni_list_first(&mq->mq_aio_getq);
- while ((aio = naio) != NULL) {
- naio = nni_list_next(&mq->mq_aio_getq, aio);
- nni_aio_list_remove(aio);
- nni_aio_finish(aio, NNG_ECLOSED, 0);
- }
-
- naio = nni_list_first(&mq->mq_aio_putq);
- while ((aio = naio) != NULL) {
- naio = nni_list_next(&mq->mq_aio_putq, aio);
+ while (((aio = nni_list_first(&mq->mq_aio_getq)) != NULL) ||
+ ((aio = nni_list_first(&mq->mq_aio_putq)) != NULL)) {
nni_aio_list_remove(aio);
- nni_aio_finish(aio, NNG_ECLOSED, 0);
+ nni_aio_finish_error(aio, NNG_ECLOSED);
}
nni_mtx_unlock(&mq->mq_lock);
diff --git a/src/core/taskq.c b/src/core/taskq.c
index 5fbcdb33..14b04085 100644
--- a/src/core/taskq.c
+++ b/src/core/taskq.c
@@ -158,7 +158,6 @@ void
nni_taskq_fini(nni_taskq *tq)
{
int i;
- int busy;
// First drain the taskq completely. This is necessary since some
// tasks that are presently running may need to schedule additional
generated by cgit v1.2.3-70-g09d2 (git 2.46.0) at 2026-01-18 19:02:27 +0000