aboutsummaryrefslogtreecommitdiff
path: root/src/core
diff options
context:
space:
mode:
Diffstat (limited to 'src/core')
-rw-r--r--src/core/msgqueue.c113
-rw-r--r--src/core/msgqueue.h36
-rw-r--r--src/core/protocol.h16
-rw-r--r--src/core/socket.c142
-rw-r--r--src/core/socket.h29
5 files changed, 165 insertions, 171 deletions
diff --git a/src/core/msgqueue.c b/src/core/msgqueue.c
index 6fd95f98..c1f3701c 100644
--- a/src/core/msgqueue.c
+++ b/src/core/msgqueue.c
@@ -27,12 +27,17 @@ struct nni_msgq {
int mq_puterr;
int mq_geterr;
int mq_draining;
+ int mq_besteffort;
nni_msg **mq_msgs;
nni_list mq_aio_putq;
nni_list mq_aio_getq;
nni_list mq_aio_notify_get;
nni_list mq_aio_notify_put;
+
+ // Filters.
+ nni_msgq_filter mq_filter_fn;
+ void * mq_filter_arg;
};
int
@@ -157,6 +162,13 @@ nni_msgq_set_error(nni_msgq *mq, int error)
nni_mtx_unlock(&mq->mq_lock);
}
+void
+nni_msgq_set_filter(nni_msgq *mq, nni_msgq_filter filter, void *arg)
+{
+ mq->mq_filter_fn = filter;
+ mq->mq_filter_arg = arg;
+}
+
static void
nni_msgq_run_putq(nni_msgq *mq)
{
@@ -173,12 +185,19 @@ 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) {
- nni_aio_set_msg(waio, NULL);
- nni_aio_list_remove(raio);
+ nni_aio_set_msg(waio, NULL);
nni_aio_list_remove(waio);
+
+ if (mq->mq_filter_fn != NULL) {
+ msg = mq->mq_filter_fn(mq->mq_filter_arg, msg);
+ }
+ if (msg != NULL) {
+ nni_aio_list_remove(raio);
+ nni_aio_finish_msg(raio, msg);
+ }
+
nni_aio_finish(waio, 0, len);
- nni_aio_finish_msg(raio, msg);
continue;
}
@@ -195,40 +214,76 @@ nni_msgq_run_putq(nni_msgq *mq)
continue;
}
+ // If we are in best effort mode, just drop the message
+ // as if we delivered.
+ if (mq->mq_besteffort) {
+ nni_list_remove(&mq->mq_aio_putq, waio);
+ nni_aio_set_msg(waio, NULL);
+ nni_msg_free(msg);
+ nni_aio_finish(waio, 0, len);
+ continue;
+ }
+
// Unable to make progress, leave the aio where it is.
break;
}
}
+void
+nni_msgq_set_best_effort(nni_msgq *mq, int on)
+{
+ nni_mtx_lock(&mq->mq_lock);
+ mq->mq_besteffort = on;
+ if (on) {
+ nni_msgq_run_putq(mq);
+ }
+ nni_mtx_unlock(&mq->mq_lock);
+}
+
static void
nni_msgq_run_getq(nni_msgq *mq)
{
nni_aio *raio;
nni_aio *waio;
- nni_msg *msg;
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) {
- msg = mq->mq_msgs[mq->mq_get++];
+ nni_msg *msg = mq->mq_msgs[mq->mq_get++];
if (mq->mq_get == mq->mq_alloc) {
mq->mq_get = 0;
}
mq->mq_len--;
- nni_aio_list_remove(raio);
- nni_aio_finish_msg(raio, msg);
+
+ if (mq->mq_filter_fn != NULL) {
+ msg = mq->mq_filter_fn(mq->mq_filter_arg, msg);
+ }
+ if (msg != NULL) {
+ nni_aio_list_remove(raio);
+ nni_aio_finish_msg(raio, msg);
+ }
continue;
}
// Nothing queued (unbuffered?), maybe a writer is waiting.
if ((waio = nni_list_first(&mq->mq_aio_putq)) != NULL) {
+ nni_msg *msg;
+ size_t len;
msg = nni_aio_get_msg(waio);
+ len = nni_msg_len(msg);
+
nni_aio_set_msg(waio, NULL);
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);
+ if (mq->mq_filter_fn != NULL) {
+ msg = mq->mq_filter_fn(mq->mq_filter_arg, msg);
+ }
+ if (msg != NULL) {
+ nni_aio_list_remove(raio);
+ nni_aio_finish_msg(raio, msg);
+ }
+
+ nni_aio_finish(waio, 0, len);
continue;
}
@@ -393,44 +448,6 @@ nni_msgq_tryput(nni_msgq *mq, nni_msg *msg)
return (NNG_EAGAIN);
}
-int
-nni_msgq_get_until(nni_msgq *mq, nni_msg **msgp, nni_time expire)
-{
- nni_aio *aio;
- int rv;
-
- if ((rv = nni_aio_init(&aio, NULL, NULL)) != 0) {
- return (rv);
- }
- nni_aio_set_timeout(aio, expire);
- nni_msgq_aio_get(mq, aio);
- nni_aio_wait(aio);
- if ((rv = nni_aio_result(aio)) == 0) {
- *msgp = nni_aio_get_msg(aio);
- nni_aio_set_msg(aio, NULL);
- }
- nni_aio_fini(aio);
- return (rv);
-}
-
-int
-nni_msgq_put_until(nni_msgq *mq, nni_msg *msg, nni_time expire)
-{
- nni_aio *aio;
- int rv;
-
- if ((rv = nni_aio_init(&aio, NULL, NULL)) != 0) {
- return (rv);
- }
- nni_aio_set_timeout(aio, expire);
- nni_aio_set_msg(aio, msg);
- nni_msgq_aio_put(mq, aio);
- nni_aio_wait(aio);
- rv = nni_aio_result(aio);
- nni_aio_fini(aio);
- return (rv);
-}
-
void
nni_msgq_drain(nni_msgq *mq, nni_time expire)
{
diff --git a/src/core/msgqueue.h b/src/core/msgqueue.h
index fa6b845a..bbac505d 100644
--- a/src/core/msgqueue.h
+++ b/src/core/msgqueue.h
@@ -41,28 +41,11 @@ extern void nni_msgq_aio_get(nni_msgq *, nni_aio *);
extern void nni_msgq_aio_notify_get(nni_msgq *, nni_aio *);
extern void nni_msgq_aio_notify_put(nni_msgq *, nni_aio *);
-// nni_msgq_put puts the message to the queue. It blocks until it
-// was able to do so, or the queue is closed, or a timeout is reached.
-// It returns 0 on success, NNG_ECLOSED if the queue was closed, or
-// NNG_ETIMEDOUT if the timeout is reached. If an error is returned,
-// the caller is responsible for freeing the message with nni_msg_free(),
-// otherwise the message is "owned" by the queue, and the caller is not
-// permitted to access it further.
-extern int nni_msgq_put_until(nni_msgq *, nni_msg *, nni_time);
-
// nni_msgq_tryput performs a non-blocking attempt to put a message on
// the message queue. It is the same as calling nng_msgq_put_until with
// a zero time.
extern int nni_msgq_tryput(nni_msgq *, nni_msg *);
-// nni_msgq_get_until gets a message from the queue. It blocks until a
-// message is available, the queue is closed, or time out is reached.
-// It returns 0 on success, NNG_ECLOSED if the queue was closed, or
-// NNG_ETIMEDOUT if the timeout is reached. On successful return,
-// the caller assumes ownership of the message and must call
-// nni_msg_free() when it is finished with it.
-extern int nni_msgq_get_until(nni_msgq *, nni_msg **, nni_time);
-
// nni_msgq_set_error sets an error condition on the message queue,
// which causes all current and future readers/writes to return the
// given error condition (if non-zero). Threads waiting to put or get
@@ -81,6 +64,25 @@ extern void nni_msgq_set_put_error(nni_msgq *, int);
// Readers (nni_msgq_put*) are unaffected.
extern void nni_msgq_set_get_error(nni_msgq *, int);
+// nni_msgq_set_best_effort marks the message queue best effort on send.
+// What this does is treat the message queue condition as if it were
+// successful, returning 0, and discarding the message. If zero is
+// passed then this mode is reset to normal.
+extern void nni_msgq_set_best_effort(nni_msgq *, int);
+
+// nni_msgq_filter is a callback function used to filter messages.
+// The function is called on entry (put) or exit (get). The void
+// argument is an opaque pointer supplied with the function at registration
+// time. The primary use for these functions is to support the protocol
+// socket needs.
+typedef nni_msg *(*nni_msgq_filter)(void *, nni_msg *);
+
+// nni_msgq_set_filter sets the filter on the queue. Messages
+// are filtered through this just before they are returned via the get
+// functions. If the filter returns NULL, then the message is silently
+// discarded instead, and any get waiters remain waiting.
+extern void nni_msgq_set_filter(nni_msgq *, nni_msgq_filter, void *);
+
// nni_msgq_close closes the queue. After this all operates on the
// message queue will return NNG_ECLOSED. Messages inside the queue
// are freed. Unlike closing a go channel, this operation is idempotent.
diff --git a/src/core/protocol.h b/src/core/protocol.h
index 0c0d93ce..253452d8 100644
--- a/src/core/protocol.h
+++ b/src/core/protocol.h
@@ -74,15 +74,17 @@ struct nni_proto_sock_ops {
// it can signal the socket worker threads to exit.
void (*sock_close)(void *);
- // Receive filter. This may be NULL, but if it isn't, then
+ // Send a message.
+ void (*sock_send)(void *, nni_aio *);
+
+ // Receive a message.
+ void (*sock_recv)(void *, nni_aio *);
+
+ // Message filter. This may be NULL, but if it isn't, then
// messages coming into the system are routed here just before being
- // delivered to the application. To drop the message, the prtocol
+ // delivered to the application. To drop the message, the protocol
// should return NULL, otherwise the message (possibly modified).
- nni_msg *(*sock_rfilter)(void *, nni_msg *);
-
- // Send filter. This may be NULL, but if it isn't, then messages
- // here are filtered just after they come from the application.
- nni_msg *(*sock_sfilter)(void *, nni_msg *);
+ nni_msg *(*sock_filter)(void *, nni_msg *);
// Options. Must not be NULL. Final entry should have NULL name.
nni_proto_sock_option *sock_options;
diff --git a/src/core/socket.c b/src/core/socket.c
index 8895f7a7..bc1f446d 100644
--- a/src/core/socket.c
+++ b/src/core/socket.c
@@ -66,12 +66,9 @@ struct nni_socket {
nni_list s_eps; // active endpoints
nni_list s_pipes; // active pipes
- int s_ep_pend; // EP dial/listen in progress
- int s_closing; // Socket is closing
- int s_closed; // Socket closed, protected by global lock
- int s_besteffort; // Best effort mode delivery
- int s_senderr; // Protocol state machine use
- int s_recverr; // Protocol state machine use
+ int s_ep_pend; // EP dial/listen in progress
+ int s_closing; // Socket is closing
+ int s_closed; // Socket closed, protected by global lock
nni_event s_recv_ev; // Event for readability
nni_event s_send_ev; // Event for sendability
@@ -374,15 +371,19 @@ nni_sock_pipe_remove(nni_sock *sock, nni_pipe *pipe)
}
void
-nni_sock_lock(nni_sock *sock)
+nni_sock_send_pending(nni_sock *sock)
{
- nni_mtx_lock(&sock->s_mx);
+ if (sock->s_send_fd.sn_init) {
+ nni_plat_pipe_clear(sock->s_send_fd.sn_rfd);
+ }
}
void
-nni_sock_unlock(nni_sock *sock)
+nni_sock_recv_pending(nni_sock *sock)
{
- nni_mtx_unlock(&sock->s_mx);
+ if (sock->s_recv_fd.sn_init) {
+ nni_plat_pipe_clear(sock->s_recv_fd.sn_rfd);
+ }
}
static void
@@ -551,6 +552,11 @@ nni_sock_create(nni_sock **sp, const nni_proto *proto)
return (rv);
}
+ if (s->s_sock_ops.sock_filter != NULL) {
+ nni_msgq_set_filter(
+ s->s_urq, s->s_sock_ops.sock_filter, s->s_data);
+ }
+
*sp = s;
return (rv);
}
@@ -779,13 +785,23 @@ nni_sock_closeall(void)
}
}
+void
+nni_sock_send(nni_sock *sock, nni_aio *aio)
+{
+ sock->s_sock_ops.sock_send(sock->s_data, aio);
+}
+
int
nni_sock_sendmsg(nni_sock *sock, nni_msg *msg, int flags)
{
int rv;
- int besteffort;
nni_time expire;
nni_time timeo = sock->s_sndtimeo;
+ nni_aio *aio;
+
+ if ((rv = nni_aio_init(&aio, NULL, NULL)) != 0) {
+ return (rv);
+ }
if ((flags == NNG_FLAG_NONBLOCK) || (timeo == 0)) {
expire = NNI_TIME_ZERO;
@@ -795,47 +811,24 @@ nni_sock_sendmsg(nni_sock *sock, nni_msg *msg, int flags)
expire = nni_clock();
expire += timeo;
}
+ nni_aio_set_timeout(aio, expire);
+ nni_aio_set_msg(aio, msg);
- // Senderr is typically set by protocols when the state machine
- // indicates that it is no longer valid to send a message. E.g.
- // a REP socket with no REQ pending.
- nni_mtx_lock(&sock->s_mx);
- if (sock->s_closing) {
- nni_mtx_unlock(&sock->s_mx);
- return (NNG_ECLOSED);
- }
- if ((rv = sock->s_senderr) != 0) {
- nni_mtx_unlock(&sock->s_mx);
- return (rv);
- }
- besteffort = sock->s_besteffort;
-
- if (sock->s_sock_ops.sock_sfilter != NULL) {
- msg = sock->s_sock_ops.sock_sfilter(sock->s_data, msg);
- }
- nni_mtx_unlock(&sock->s_mx);
+ nni_sock_send(sock, aio);
+ nni_aio_wait(aio);
- if (msg == NULL) {
- return (0);
- }
+ rv = nni_aio_result(aio);
+ nni_aio_fini(aio);
- if (besteffort) {
- // BestEffort mode -- if we cannot handle the message due to
- // backpressure, we just throw it away, and don't complain.
- expire = NNI_TIME_ZERO;
- }
- if (sock->s_send_fd.sn_init) {
- nni_plat_pipe_clear(sock->s_send_fd.sn_rfd);
- }
- rv = nni_msgq_put_until(sock->s_uwq, msg, expire);
- if (besteffort && (rv == NNG_ETIMEDOUT)) {
- // Pretend this worked... it didn't, but pretend.
- nni_msg_free(msg);
- return (0);
- }
return (rv);
}
+void
+nni_sock_recv(nni_sock *sock, nni_aio *aio)
+{
+ sock->s_sock_ops.sock_recv(sock->s_data, aio);
+}
+
int
nni_sock_recvmsg(nni_sock *sock, nni_msg **msgp, int flags)
{
@@ -843,6 +836,11 @@ nni_sock_recvmsg(nni_sock *sock, nni_msg **msgp, int flags)
nni_msg *msg;
nni_time expire;
nni_time timeo = sock->s_rcvtimeo;
+ nni_aio *aio;
+
+ if ((rv = nni_aio_init(&aio, NULL, NULL)) != 0) {
+ return (rv);
+ }
if ((flags == NNG_FLAG_NONBLOCK) || (timeo == 0)) {
expire = NNI_TIME_ZERO;
@@ -853,39 +851,23 @@ nni_sock_recvmsg(nni_sock *sock, nni_msg **msgp, int flags)
expire += timeo;
}
- nni_mtx_lock(&sock->s_mx);
- if (sock->s_closing) {
- nni_mtx_unlock(&sock->s_mx);
- return (NNG_ECLOSED);
- }
- if ((rv = sock->s_recverr) != 0) {
- nni_mtx_unlock(&sock->s_mx);
- return (rv);
- }
- nni_mtx_unlock(&sock->s_mx);
-
- if (sock->s_recv_fd.sn_init) {
- nni_plat_pipe_clear(sock->s_recv_fd.sn_rfd);
- }
-
for (;;) {
- rv = nni_msgq_get_until(sock->s_urq, &msg, expire);
+ nni_aio_set_timeout(aio, expire);
+ nni_sock_recv(sock, aio);
+ nni_aio_wait(aio);
+
+ rv = nni_aio_result(aio);
if (rv != 0) {
- return (rv);
- }
- if (sock->s_sock_ops.sock_rfilter != NULL) {
- nni_mtx_lock(&sock->s_mx);
- msg = sock->s_sock_ops.sock_rfilter(sock->s_data, msg);
- nni_mtx_unlock(&sock->s_mx);
- }
- if (msg != NULL) {
break;
}
- // Protocol dropped the message; try again.
- }
+ msg = nni_aio_get_msg(aio);
+ nni_aio_set_msg(aio, NULL);
- *msgp = msg;
- return (0);
+ *msgp = msg;
+ break;
+ }
+ nni_aio_fini(aio);
+ return (rv);
}
// nni_sock_protocol returns the socket's 16-bit protocol number.
@@ -949,18 +931,6 @@ nni_sock_ep_remove(nni_sock *sock, nni_ep *ep)
nni_mtx_unlock(&sock->s_mx);
}
-void
-nni_sock_recverr(nni_sock *sock, int err)
-{
- sock->s_recverr = err;
-}
-
-void
-nni_sock_senderr(nni_sock *sock, int err)
-{
- sock->s_senderr = err;
-}
-
int
nni_sock_setopt(nni_sock *s, const char *name, const void *val, size_t size)
{
diff --git a/src/core/socket.h b/src/core/socket.h
index 850c4641..edfe6447 100644
--- a/src/core/socket.h
+++ b/src/core/socket.h
@@ -22,15 +22,14 @@ extern void nni_sock_closeall(void);
extern int nni_sock_shutdown(nni_sock *);
extern uint16_t nni_sock_proto(nni_sock *);
extern uint16_t nni_sock_peer(nni_sock *);
-extern int nni_sock_setopt(nni_sock *, const char *, const void *, size_t);
-extern int nni_sock_getopt(nni_sock *, const char *, void *, size_t *);
-extern int nni_sock_recvmsg(nni_sock *, nni_msg **, int);
-extern int nni_sock_sendmsg(nni_sock *, nni_msg *, int);
+extern int nni_sock_setopt(nni_sock *, const char *, const void *, size_t);
+extern int nni_sock_getopt(nni_sock *, const char *, void *, size_t *);
+extern int nni_sock_recvmsg(nni_sock *, nni_msg **, int);
+extern int nni_sock_sendmsg(nni_sock *, nni_msg *, int);
+extern void nni_sock_send(nni_sock *, nni_aio *);
+extern void nni_sock_recv(nni_sock *, nni_aio *);
extern uint32_t nni_sock_id(nni_sock *);
-extern void nni_sock_lock(nni_sock *);
-extern void nni_sock_unlock(nni_sock *);
-
extern nni_notify *nni_sock_notify(nni_sock *, int, nng_notify_func, void *);
extern void nni_sock_unnotify(nni_sock *, nni_notify *);
@@ -46,16 +45,20 @@ extern int nni_sock_pipe_start(nni_sock *, nni_pipe *p);
extern int nni_sock_ep_add(nni_sock *, nni_ep *);
extern void nni_sock_ep_remove(nni_sock *, nni_ep *);
-// Set error codes for applications. These are only ever
-// called from the filter functions in protocols, and thus
-// already have the socket lock held.
-extern void nni_sock_recverr(nni_sock *, int);
-extern void nni_sock_senderr(nni_sock *, int);
-
// These are socket methods that protocol operations can expect to call.
// Note that each of these should be called without any locks held, since
// the socket can reenter the protocol.
+// nni_sock_send_pending is called by the protocol when it enqueues
+// a send operation. The main purpose of this is to clear the raised
+// signal raised on the event descriptor.
+extern void nni_sock_send_pending(nni_sock *);
+
+// nni_sock_recv_pending is called by the protocl when it enqueues
+// a receive operation. The main purpose of this is to clear the raised
+// signal raised on the event descriptor.
+extern void nni_sock_recv_pending(nni_sock *);
+
// nni_socket_sendq obtains the upper writeq. The protocol should
// recieve messages from this, and place them on the appropriate pipe.
extern nni_msgq *nni_sock_sendq(nni_sock *);
generated by cgit v1.2.3-70-g09d2 (git 2.46.0) at 2026-01-18 11:27:38 +0000