summaryrefslogtreecommitdiff
path: root/src/protocol
diff options
context:
space:
mode:
Diffstat (limited to 'src/protocol')
-rw-r--r--src/protocol/bus/bus.c28
-rw-r--r--src/protocol/pair/pair_v0.c20
-rw-r--r--src/protocol/pair/pair_v1.c20
-rw-r--r--src/protocol/pipeline/pull.c27
-rw-r--r--src/protocol/pipeline/push.c18
-rw-r--r--src/protocol/pubsub/pub.c18
-rw-r--r--src/protocol/pubsub/sub.c40
-rw-r--r--src/protocol/reqrep/rep.c109
-rw-r--r--src/protocol/reqrep/req.c91
-rw-r--r--src/protocol/survey/respond.c66
-rw-r--r--src/protocol/survey/survey.c64
11 files changed, 371 insertions, 130 deletions
diff --git a/src/protocol/bus/bus.c b/src/protocol/bus/bus.c
index cad21989..046aa19a 100644
--- a/src/protocol/bus/bus.c
+++ b/src/protocol/bus/bus.c
@@ -22,6 +22,9 @@ typedef struct bus_pipe bus_pipe;
typedef struct bus_sock bus_sock;
static void bus_sock_getq(bus_sock *);
+static void bus_sock_send(void *, nni_aio *);
+static void bus_sock_recv(void *, nni_aio *);
+
static void bus_pipe_getq(bus_pipe *);
static void bus_pipe_send(bus_pipe *);
static void bus_pipe_recv(bus_pipe *);
@@ -39,6 +42,8 @@ struct bus_sock {
nni_aio * aio_getq;
nni_list pipes;
nni_mtx mtx;
+ nni_msgq *uwq;
+ nni_msgq *urq;
};
// A bus_pipe is our per-pipe protocol private structure.
@@ -82,6 +87,8 @@ bus_sock_init(void **sp, nni_sock *nsock)
}
s->nsock = nsock;
s->raw = 0;
+ s->uwq = nni_sock_sendq(nsock);
+ s->urq = nni_sock_recvq(nsock);
*sp = s;
return (0);
@@ -259,7 +266,6 @@ bus_sock_getq_cb(void *arg)
bus_sock *s = arg;
bus_pipe *p;
bus_pipe *lastp;
- nni_msgq *uwq = nni_sock_sendq(s->nsock);
nni_msg * msg;
nni_msg * dup;
uint32_t sender;
@@ -338,6 +344,24 @@ bus_sock_getopt_raw(void *arg, void *buf, size_t *szp)
return (nni_getopt_int(s->raw, buf, szp));
}
+static void
+bus_sock_send(void *arg, nni_aio *aio)
+{
+ bus_sock *s = arg;
+
+ nni_sock_send_pending(s->nsock);
+ nni_msgq_aio_put(s->uwq, aio);
+}
+
+static void
+bus_sock_recv(void *arg, nni_aio *aio)
+{
+ bus_sock *s = arg;
+
+ nni_sock_recv_pending(s->nsock);
+ nni_msgq_aio_get(s->urq, aio);
+}
+
static nni_proto_pipe_ops bus_pipe_ops = {
.pipe_init = bus_pipe_init,
.pipe_fini = bus_pipe_fini,
@@ -360,6 +384,8 @@ static nni_proto_sock_ops bus_sock_ops = {
.sock_fini = bus_sock_fini,
.sock_open = bus_sock_open,
.sock_close = bus_sock_close,
+ .sock_send = bus_sock_send,
+ .sock_recv = bus_sock_recv,
.sock_options = bus_sock_options,
};
diff --git a/src/protocol/pair/pair_v0.c b/src/protocol/pair/pair_v0.c
index a87af593..cca03cc8 100644
--- a/src/protocol/pair/pair_v0.c
+++ b/src/protocol/pair/pair_v0.c
@@ -243,6 +243,24 @@ pair0_sock_getopt_raw(void *arg, void *buf, size_t *szp)
return (nni_getopt_int(s->raw, buf, szp));
}
+static void
+pair0_sock_send(void *arg, nni_aio *aio)
+{
+ pair0_sock *s = arg;
+
+ nni_sock_send_pending(s->nsock);
+ nni_msgq_aio_put(s->uwq, aio);
+}
+
+static void
+pair0_sock_recv(void *arg, nni_aio *aio)
+{
+ pair0_sock *s = arg;
+
+ nni_sock_recv_pending(s->nsock);
+ nni_msgq_aio_get(s->urq, aio);
+}
+
static nni_proto_pipe_ops pair0_pipe_ops = {
.pipe_init = pair0_pipe_init,
.pipe_fini = pair0_pipe_fini,
@@ -265,6 +283,8 @@ static nni_proto_sock_ops pair0_sock_ops = {
.sock_fini = pair0_sock_fini,
.sock_open = pair0_sock_open,
.sock_close = pair0_sock_close,
+ .sock_send = pair0_sock_send,
+ .sock_recv = pair0_sock_recv,
.sock_options = pair0_sock_options,
};
diff --git a/src/protocol/pair/pair_v1.c b/src/protocol/pair/pair_v1.c
index 6c9d4bd3..f43a4785 100644
--- a/src/protocol/pair/pair_v1.c
+++ b/src/protocol/pair/pair_v1.c
@@ -446,6 +446,24 @@ pair1_sock_getopt_poly(void *arg, void *buf, size_t *szp)
return (nni_getopt_int(s->poly, buf, szp));
}
+static void
+pair1_sock_send(void *arg, nni_aio *aio)
+{
+ pair1_sock *s = arg;
+
+ nni_sock_send_pending(s->nsock);
+ nni_msgq_aio_put(s->uwq, aio);
+}
+
+static void
+pair1_sock_recv(void *arg, nni_aio *aio)
+{
+ pair1_sock *s = arg;
+
+ nni_sock_recv_pending(s->nsock);
+ nni_msgq_aio_get(s->urq, aio);
+}
+
static nni_proto_pipe_ops pair1_pipe_ops = {
.pipe_init = pair1_pipe_init,
.pipe_fini = pair1_pipe_fini,
@@ -478,6 +496,8 @@ static nni_proto_sock_ops pair1_sock_ops = {
.sock_fini = pair1_sock_fini,
.sock_open = pair1_sock_open,
.sock_close = pair1_sock_close,
+ .sock_recv = pair1_sock_recv,
+ .sock_send = pair1_sock_send,
.sock_options = pair1_sock_options,
};
diff --git a/src/protocol/pipeline/pull.c b/src/protocol/pipeline/pull.c
index 80db2425..7dd0c8ed 100644
--- a/src/protocol/pipeline/pull.c
+++ b/src/protocol/pipeline/pull.c
@@ -26,6 +26,7 @@ static void pull_putq(pull_pipe *, nni_msg *);
struct pull_sock {
nni_msgq *urq;
int raw;
+ nni_sock *sock;
};
// A pull_pipe is our per-pipe protocol private structure.
@@ -44,10 +45,11 @@ pull_sock_init(void **sp, nni_sock *sock)
if ((s = NNI_ALLOC_STRUCT(s)) == NULL) {
return (NNG_ENOMEM);
}
- s->raw = 0;
- s->urq = nni_sock_recvq(sock);
- *sp = s;
- nni_sock_senderr(sock, NNG_ENOTSUP);
+ s->raw = 0;
+ s->urq = nni_sock_recvq(sock);
+ s->sock = sock;
+
+ *sp = s;
return (0);
}
@@ -185,6 +187,21 @@ pull_sock_getopt_raw(void *arg, void *buf, size_t *szp)
return (nni_getopt_int(s->raw, buf, szp));
}
+static void
+pull_sock_send(void *arg, nni_aio *aio)
+{
+ nni_aio_finish_error(aio, NNG_ENOTSUP);
+}
+
+static void
+pull_sock_recv(void *arg, nni_aio *aio)
+{
+ pull_sock *s = arg;
+
+ nni_sock_recv_pending(s->sock);
+ nni_msgq_aio_get(s->urq, aio);
+}
+
static nni_proto_pipe_ops pull_pipe_ops = {
.pipe_init = pull_pipe_init,
.pipe_fini = pull_pipe_fini,
@@ -207,6 +224,8 @@ static nni_proto_sock_ops pull_sock_ops = {
.sock_fini = pull_sock_fini,
.sock_open = pull_sock_open,
.sock_close = pull_sock_close,
+ .sock_send = pull_sock_send,
+ .sock_recv = pull_sock_recv,
.sock_options = pull_sock_options,
};
diff --git a/src/protocol/pipeline/push.c b/src/protocol/pipeline/push.c
index 77db7fcb..af7b80ca 100644
--- a/src/protocol/pipeline/push.c
+++ b/src/protocol/pipeline/push.c
@@ -54,7 +54,6 @@ push_sock_init(void **sp, nni_sock *sock)
s->sock = sock;
s->uwq = nni_sock_sendq(sock);
*sp = s;
- nni_sock_recverr(sock, NNG_ENOTSUP);
return (0);
}
@@ -205,6 +204,21 @@ push_sock_getopt_raw(void *arg, void *buf, size_t *szp)
return (nni_getopt_int(s->raw, buf, szp));
}
+static void
+push_sock_send(void *arg, nni_aio *aio)
+{
+ push_sock *s = arg;
+
+ nni_sock_send_pending(s->sock);
+ nni_msgq_aio_put(s->uwq, aio);
+}
+
+static void
+push_sock_recv(void *arg, nni_aio *aio)
+{
+ nni_aio_finish_error(aio, NNG_ENOTSUP);
+}
+
static nni_proto_pipe_ops push_pipe_ops = {
.pipe_init = push_pipe_init,
.pipe_fini = push_pipe_fini,
@@ -228,6 +242,8 @@ static nni_proto_sock_ops push_sock_ops = {
.sock_open = push_sock_open,
.sock_close = push_sock_close,
.sock_options = push_sock_options,
+ .sock_send = push_sock_send,
+ .sock_recv = push_sock_recv,
};
static nni_proto push_proto = {
diff --git a/src/protocol/pubsub/pub.c b/src/protocol/pubsub/pub.c
index 03f4603a..4604d0ff 100644
--- a/src/protocol/pubsub/pub.c
+++ b/src/protocol/pubsub/pub.c
@@ -82,7 +82,6 @@ pub_sock_init(void **sp, nni_sock *sock)
s->uwq = nni_sock_sendq(sock);
*sp = s;
- nni_sock_recverr(sock, NNG_ENOTSUP);
return (0);
}
@@ -281,6 +280,21 @@ pub_sock_getopt_raw(void *arg, void *buf, size_t *szp)
return (nni_getopt_int(s->raw, buf, szp));
}
+static void
+pub_sock_recv(void *arg, nni_aio *aio)
+{
+ nni_aio_finish_error(aio, NNG_ENOTSUP);
+}
+
+static void
+pub_sock_send(void *arg, nni_aio *aio)
+{
+ pub_sock *s = arg;
+
+ nni_sock_send_pending(s->sock);
+ nni_msgq_aio_put(s->uwq, aio);
+}
+
static nni_proto_pipe_ops pub_pipe_ops = {
.pipe_init = pub_pipe_init,
.pipe_fini = pub_pipe_fini,
@@ -303,6 +317,8 @@ static nni_proto_sock_ops pub_sock_ops = {
.sock_fini = pub_sock_fini,
.sock_open = pub_sock_open,
.sock_close = pub_sock_close,
+ .sock_send = pub_sock_send,
+ .sock_recv = pub_sock_recv,
.sock_options = pub_sock_options,
};
diff --git a/src/protocol/pubsub/sub.c b/src/protocol/pubsub/sub.c
index 8b2ed209..323bbb2f 100644
--- a/src/protocol/pubsub/sub.c
+++ b/src/protocol/pubsub/sub.c
@@ -40,6 +40,7 @@ struct sub_sock {
nni_list topics;
nni_msgq *urq;
int raw;
+ nni_mtx lk;
};
// An nni_rep_pipe is our per-pipe protocol private structure.
@@ -58,13 +59,13 @@ sub_sock_init(void **sp, nni_sock *sock)
if ((s = NNI_ALLOC_STRUCT(s)) == NULL) {
return (NNG_ENOMEM);
}
+ nni_mtx_init(&s->lk);
NNI_LIST_INIT(&s->topics, sub_topic, node);
s->sock = sock;
s->raw = 0;
s->urq = nni_sock_recvq(sock);
- nni_sock_senderr(sock, NNG_ENOTSUP);
- *sp = s;
+ *sp = s;
return (0);
}
@@ -79,6 +80,7 @@ sub_sock_fini(void *arg)
nni_free(topic->buf, topic->len);
NNI_FREE_STRUCT(topic);
}
+ nni_mtx_fini(&s->lk);
NNI_FREE_STRUCT(s);
}
@@ -190,6 +192,7 @@ sub_subscribe(void *arg, const void *buf, size_t sz)
sub_topic *topic;
sub_topic *newtopic;
+ nni_mtx_lock(&s->lk);
NNI_LIST_FOREACH (&s->topics, topic) {
int rv;
@@ -201,6 +204,7 @@ sub_subscribe(void *arg, const void *buf, size_t sz)
if (rv == 0) {
if (topic->len == sz) {
// Already inserted.
+ nni_mtx_unlock(&s->lk);
return (0);
}
if (topic->len > sz) {
@@ -212,9 +216,11 @@ sub_subscribe(void *arg, const void *buf, size_t sz)
}
if ((newtopic = NNI_ALLOC_STRUCT(newtopic)) == NULL) {
+ nni_mtx_unlock(&s->lk);
return (NNG_ENOMEM);
}
if ((newtopic->buf = nni_alloc(sz)) == NULL) {
+ nni_mtx_unlock(&s->lk);
return (NNG_ENOMEM);
}
NNI_LIST_NODE_INIT(&newtopic->node);
@@ -225,6 +231,7 @@ sub_subscribe(void *arg, const void *buf, size_t sz)
} else {
nni_list_append(&s->topics, newtopic);
}
+ nni_mtx_unlock(&s->lk);
return (0);
}
@@ -235,6 +242,7 @@ sub_unsubscribe(void *arg, const void *buf, size_t sz)
sub_topic *topic;
int rv;
+ nni_mtx_lock(&s->lk);
NNI_LIST_FOREACH (&s->topics, topic) {
if (topic->len >= sz) {
rv = memcmp(topic->buf, buf, sz);
@@ -244,18 +252,22 @@ sub_unsubscribe(void *arg, const void *buf, size_t sz)
if (rv == 0) {
if (topic->len == sz) {
nni_list_remove(&s->topics, topic);
+ nni_mtx_unlock(&s->lk);
nni_free(topic->buf, topic->len);
NNI_FREE_STRUCT(topic);
return (0);
}
if (topic->len > sz) {
+ nni_mtx_unlock(&s->lk);
return (NNG_ENOENT);
}
}
if (rv > 0) {
+ nni_mtx_unlock(&s->lk);
return (NNG_ENOENT);
}
}
+ nni_mtx_unlock(&s->lk);
return (NNG_ENOENT);
}
@@ -273,8 +285,23 @@ sub_sock_getopt_raw(void *arg, void *buf, size_t *szp)
return (nni_getopt_int(s->raw, buf, szp));
}
+static void
+sub_sock_send(void *arg, nni_aio *aio)
+{
+ nni_aio_finish_error(aio, NNG_ENOTSUP);
+}
+
+static void
+sub_sock_recv(void *arg, nni_aio *aio)
+{
+ sub_sock *s = arg;
+
+ nni_sock_recv_pending(s->sock);
+ nni_msgq_aio_get(s->urq, aio);
+}
+
static nni_msg *
-sub_sock_rfilter(void *arg, nni_msg *msg)
+sub_sock_filter(void *arg, nni_msg *msg)
{
sub_sock * s = arg;
sub_topic *topic;
@@ -282,7 +309,9 @@ sub_sock_rfilter(void *arg, nni_msg *msg)
size_t len;
int match;
+ nni_mtx_lock(&s->lk);
if (s->raw) {
+ nni_mtx_unlock(&s->lk);
return (msg);
}
@@ -308,6 +337,7 @@ sub_sock_rfilter(void *arg, nni_msg *msg)
break;
}
}
+ nni_mtx_unlock(&s->lk);
if (!match) {
nni_msg_free(msg);
return (NULL);
@@ -349,8 +379,10 @@ static nni_proto_sock_ops sub_sock_ops = {
.sock_fini = sub_sock_fini,
.sock_open = sub_sock_open,
.sock_close = sub_sock_close,
+ .sock_send = sub_sock_send,
+ .sock_recv = sub_sock_recv,
+ .sock_filter = sub_sock_filter,
.sock_options = sub_sock_options,
- .sock_rfilter = sub_sock_rfilter,
};
static nni_proto sub_proto = {
diff --git a/src/protocol/reqrep/rep.c b/src/protocol/reqrep/rep.c
index cd9411d9..510beab3 100644
--- a/src/protocol/reqrep/rep.c
+++ b/src/protocol/reqrep/rep.c
@@ -32,6 +32,7 @@ struct rep_sock {
nni_sock * sock;
nni_msgq * uwq;
nni_msgq * urq;
+ nni_mtx lk;
int raw;
int ttl;
nni_idhash *pipes;
@@ -62,6 +63,7 @@ rep_sock_fini(void *arg)
if (s->btrace != NULL) {
nni_free(s->btrace, s->btrace_len);
}
+ nni_mtx_fini(&s->lk);
NNI_FREE_STRUCT(s);
}
@@ -74,6 +76,7 @@ rep_sock_init(void **sp, nni_sock *sock)
if ((s = NNI_ALLOC_STRUCT(s)) == NULL) {
return (NNG_ENOMEM);
}
+ nni_mtx_init(&s->lk);
if (((rv = nni_idhash_init(&s->pipes)) != 0) ||
((rv = nni_aio_init(&s->aio_getq, rep_sock_getq_cb, s)) != 0)) {
rep_sock_fini(s);
@@ -89,7 +92,6 @@ rep_sock_init(void **sp, nni_sock *sock)
s->urq = nni_sock_recvq(sock);
*sp = s;
- nni_sock_senderr(sock, NNG_ESTATE);
return (0);
}
@@ -215,6 +217,7 @@ rep_sock_getq_cb(void *arg)
// Look for the pipe, and attempt to put the message there
// (nonblocking) if we can. If we can't for any reason, then we
// free the message.
+ // XXX: LOCKING?!?!
if ((rv = nni_idhash_find(s->pipes, id, (void **) &p)) == 0) {
rv = nni_msgq_tryput(p->sendq, msg);
}
@@ -347,10 +350,10 @@ rep_sock_setopt_raw(void *arg, const void *buf, size_t sz)
{
rep_sock *s = arg;
int rv;
+
+ nni_mtx_lock(&s->lk);
rv = nni_setopt_int(&s->raw, buf, sz, 0, 1);
- if (rv == 0) {
- nni_sock_senderr(s->sock, s->raw ? 0 : NNG_ESTATE);
- }
+ nni_mtx_unlock(&s->lk);
return (rv);
}
@@ -376,53 +379,18 @@ rep_sock_getopt_maxttl(void *arg, void *buf, size_t *szp)
}
static nni_msg *
-rep_sock_sfilter(void *arg, nni_msg *msg)
-{
- rep_sock *s = arg;
-
- if (s->raw) {
- return (msg);
- }
-
- // Cannot send again until a receive is done...
- nni_sock_senderr(s->sock, NNG_ESTATE);
-
- // If we have a stored backtrace, append it to the header...
- // if we don't have a backtrace, discard the message.
- if (s->btrace == NULL) {
- nni_msg_free(msg);
- return (NULL);
- }
-
- // drop anything else in the header...
- nni_msg_header_clear(msg);
-
- if (nni_msg_header_append(msg, s->btrace, s->btrace_len) != 0) {
- nni_free(s->btrace, s->btrace_len);
- s->btrace = NULL;
- s->btrace_len = 0;
- nni_msg_free(msg);
- return (NULL);
- }
-
- nni_free(s->btrace, s->btrace_len);
- s->btrace = NULL;
- s->btrace_len = 0;
- return (msg);
-}
-
-static nni_msg *
-rep_sock_rfilter(void *arg, nni_msg *msg)
+rep_sock_filter(void *arg, nni_msg *msg)
{
rep_sock *s = arg;
char * header;
size_t len;
+ nni_mtx_lock(&s->lk);
if (s->raw) {
+ nni_mtx_unlock(&s->lk);
return (msg);
}
- nni_sock_senderr(s->sock, 0);
len = nni_msg_header_len(msg);
header = nni_msg_header(msg);
if (s->btrace != NULL) {
@@ -437,9 +405,61 @@ rep_sock_rfilter(void *arg, nni_msg *msg)
s->btrace_len = len;
memcpy(s->btrace, header, len);
nni_msg_header_clear(msg);
+ nni_mtx_unlock(&s->lk);
return (msg);
}
+static void
+rep_sock_send(void *arg, nni_aio *aio)
+{
+ rep_sock *s = arg;
+ int rv;
+ nni_msg * msg;
+
+ nni_mtx_lock(&s->lk);
+ if (s->raw) {
+ // Pass thru
+ nni_mtx_unlock(&s->lk);
+ nni_sock_send_pending(s->sock);
+ nni_msgq_aio_put(s->uwq, aio);
+ return;
+ }
+ if (s->btrace == NULL) {
+ nni_mtx_unlock(&s->lk);
+ nni_aio_finish_error(aio, NNG_ESTATE);
+ return;
+ }
+
+ msg = nni_aio_get_msg(aio);
+
+ // drop anything else in the header... (it should already be
+ // empty, but there can be stale backtrace info there.)
+ nni_msg_header_clear(msg);
+
+ if ((rv = nni_msg_header_append(msg, s->btrace, s->btrace_len)) != 0) {
+ nni_mtx_unlock(&s->lk);
+ nni_aio_finish_error(aio, rv);
+ return;
+ }
+
+ nni_free(s->btrace, s->btrace_len);
+ s->btrace = NULL;
+ s->btrace_len = 0;
+
+ nni_mtx_unlock(&s->lk);
+ nni_sock_send_pending(s->sock);
+ nni_msgq_aio_put(s->uwq, aio);
+}
+
+static void
+rep_sock_recv(void *arg, nni_aio *aio)
+{
+ rep_sock *s = arg;
+
+ nni_sock_recv_pending(s->sock);
+ nni_msgq_aio_get(s->urq, aio);
+}
+
// This is the global protocol structure -- our linkage to the core.
// This should be the only global non-static symbol in this file.
static nni_proto_pipe_ops rep_pipe_ops = {
@@ -470,8 +490,9 @@ static nni_proto_sock_ops rep_sock_ops = {
.sock_open = rep_sock_open,
.sock_close = rep_sock_close,
.sock_options = rep_sock_options,
- .sock_rfilter = rep_sock_rfilter,
- .sock_sfilter = rep_sock_sfilter,
+ .sock_filter = rep_sock_filter,
+ .sock_send = rep_sock_send,
+ .sock_recv = rep_sock_recv,
};
static nni_proto nni_rep_proto = {
diff --git a/src/protocol/reqrep/req.c b/src/protocol/reqrep/req.c
index 24d01df2..81abd306 100644
--- a/src/protocol/reqrep/req.c
+++ b/src/protocol/reqrep/req.c
@@ -99,7 +99,7 @@ req_sock_init(void **sp, nni_sock *sock)
s->uwq = nni_sock_sendq(sock);
s->urq = nni_sock_recvq(sock);
*sp = s;
- nni_sock_recverr(sock, NNG_ESTATE);
+
return (0);
}
@@ -249,12 +249,7 @@ static int
req_sock_setopt_raw(void *arg, const void *buf, size_t sz)
{
req_sock *s = arg;
- int rv;
- rv = nni_setopt_int(&s->raw, buf, sz, 0, 1);
- if (rv == 0) {
- nni_sock_recverr(s->sock, s->raw ? 0 : NNG_ESTATE);
- }
- return (rv);
+ return (nni_setopt_int(&s->raw, buf, sz, 0, 1));
}
static int
@@ -505,35 +500,46 @@ req_resend(req_sock *s)
}
}
-static nni_msg *
-req_sock_sfilter(void *arg, nni_msg *msg)
+static void
+req_sock_send(void *arg, nni_aio *aio)
{
req_sock *s = arg;
uint32_t id;
+ size_t len;
+ nni_msg * msg;
+ int rv;
+ nni_mtx_lock(&s->mtx);
if (s->raw) {
- // No automatic retry, and the request ID must
- // be in the header coming down.
- return (msg);
+ nni_mtx_unlock(&s->mtx);
+ nni_sock_send_pending(s->sock);
+ nni_msgq_aio_put(s->uwq, aio);
+ return;
}
+ msg = nni_aio_get_msg(aio);
+ len = nni_msg_len(msg);
+
+ // In cooked mode, because we need to manage our own resend logic,
+ // we bypass the upper writeq entirely.
+
// Generate a new request ID. We always set the high
// order bit so that the peer can locate the end of the
// backtrace. (Pipe IDs have the high order bit clear.)
id = (s->nextid++) | 0x80000000u;
-
// Request ID is in big endian format.
NNI_PUT32(s->reqid, id);
- if (nni_msg_header_append(msg, s->reqid, 4) != 0) {
- // Should be ENOMEM.
- nni_msg_free(msg);
- return (NULL);
+ if ((rv = nni_msg_header_append(msg, s->reqid, 4)) != 0) {
+ nni_mtx_unlock(&s->mtx);
+ nni_aio_finish_error(aio, rv);
+ return;
}
- // NB: The socket lock is also held, so this is always self-serialized.
- // But we have to serialize against other async callbacks.
- nni_mtx_lock(&s->mtx);
+ // XXX: I think we should just not do this... and leave the
+ // socket "permanently writeable". This does screw up all the
+ // backpressure.
+ // nni_sock_send_pending(s->sock);
// If another message is there, this cancels it.
if (s->reqmsg != NULL) {
@@ -541,6 +547,8 @@ req_sock_sfilter(void *arg, nni_msg *msg)
s->reqmsg = NULL;
}
+ nni_aio_set_msg(aio, NULL);
+
// Make a duplicate message... for retries.
s->reqmsg = msg;
// Schedule for immediate send
@@ -548,40 +556,41 @@ req_sock_sfilter(void *arg, nni_msg *msg)
s->wantw = 1;
req_resend(s);
- nni_mtx_unlock(&s->mtx);
- // Clear the error condition.
- nni_sock_recverr(s->sock, 0);
+ nni_mtx_unlock(&s->mtx);
- return (NULL);
+ nni_aio_finish(aio, 0, len);
}
static nni_msg *
-req_sock_rfilter(void *arg, nni_msg *msg)
+req_sock_filter(void *arg, nni_msg *msg)
{
req_sock *s = arg;
nni_msg * rmsg;
+ nni_mtx_lock(&s->mtx);
if (s->raw) {
// Pass it unmolested
+ nni_mtx_unlock(&s->mtx);
return (msg);
}
if (nni_msg_header_len(msg) < 4) {
+ nni_mtx_unlock(&s->mtx);
nni_msg_free(msg);
return (NULL);
}
- nni_mtx_lock(&s->mtx);
-
if ((rmsg = s->reqmsg) == NULL) {
- // We had no outstanding request.
+ // We had no outstanding request. (Perhaps canceled,
+ // or duplicate response.)
nni_mtx_unlock(&s->mtx);
nni_msg_free(msg);
return (NULL);
}
+
if (memcmp(nni_msg_header(msg), s->reqid, 4) != 0) {
- // Wrong request id
+ // Wrong request id.
nni_mtx_unlock(&s->mtx);
nni_msg_free(msg);
return (NULL);
@@ -591,12 +600,29 @@ req_sock_rfilter(void *arg, nni_msg *msg)
s->pendpipe = NULL;
nni_mtx_unlock(&s->mtx);
- nni_sock_recverr(s->sock, NNG_ESTATE);
nni_msg_free(rmsg);
return (msg);
}
+static void
+req_sock_recv(void *arg, nni_aio *aio)
+{
+ req_sock *s = arg;
+
+ nni_mtx_lock(&s->mtx);
+ if (!s->raw) {
+ if (s->reqmsg == NULL) {
+ nni_mtx_unlock(&s->mtx);
+ nni_aio_finish_error(aio, NNG_ESTATE);
+ return;
+ }
+ }
+ nni_mtx_unlock(&s->mtx);
+ nni_sock_recv_pending(s->sock);
+ nni_msgq_aio_get(s->urq, aio);
+}
+
static nni_proto_pipe_ops req_pipe_ops = {
.pipe_init = req_pipe_init,
.pipe_fini = req_pipe_fini,
@@ -630,8 +656,9 @@ static nni_proto_sock_ops req_sock_ops = {
.sock_open = req_sock_open,
.sock_close = req_sock_close,
.sock_options = req_sock_options,
- .sock_rfilter = req_sock_rfilter,
- .sock_sfilter = req_sock_sfilter,
+ .sock_filter = req_sock_filter,
+ .sock_send = req_sock_send,
+ .sock_recv = req_sock_recv,
};
static nni_proto req_proto = {
diff --git a/src/protocol/survey/respond.c b/src/protocol/survey/respond.c
index fcb067b0..4c3ea8e3 100644
--- a/src/protocol/survey/respond.c
+++ b/src/protocol/survey/respond.c
@@ -77,6 +77,7 @@ resp_sock_init(void **sp, nni_sock *nsock)
if ((s = NNI_ALLOC_STRUCT(s)) == NULL) {
return (NNG_ENOMEM);
}
+ nni_mtx_init(&s->mtx);
if (((rv = nni_idhash_init(&s->pipes)) != 0) ||
((rv = nni_aio_init(&s->aio_getq, resp_sock_getq_cb, s)) != 0)) {
resp_sock_fini(s);
@@ -91,10 +92,7 @@ resp_sock_init(void **sp, nni_sock *nsock)
s->urq = nni_sock_recvq(nsock);
s->uwq = nni_sock_sendq(nsock);
- nni_mtx_init(&s->mtx);
-
*sp = s;
- nni_sock_senderr(nsock, NNG_ESTATE);
return (0);
}
@@ -349,9 +347,9 @@ resp_sock_setopt_raw(void *arg, const void *buf, size_t sz)
resp_sock *s = arg;
int rv;
- if ((rv = nni_setopt_int(&s->raw, buf, sz, 0, 1)) == 0) {
- nni_sock_senderr(s->nsock, s->raw ? 0 : NNG_ESTATE);
- }
+ nni_mtx_lock(&s->mtx);
+ rv = nni_setopt_int(&s->raw, buf, sz, 0, 1);
+ nni_mtx_unlock(&s->mtx);
return (rv);
}
@@ -376,54 +374,62 @@ resp_sock_getopt_maxttl(void *arg, void *buf, size_t *szp)
return (nni_getopt_int(s->ttl, buf, szp));
}
-static nni_msg *
-resp_sock_sfilter(void *arg, nni_msg *msg)
+static void
+resp_sock_send(void *arg, nni_aio *aio)
{
resp_sock *s = arg;
+ nni_msg * msg;
+ int rv;
+ nni_mtx_lock(&s->mtx);
if (s->raw) {
- return (msg);
+ nni_mtx_unlock(&s->mtx);
+ nni_sock_send_pending(s->nsock);
+ nni_msgq_aio_put(s->uwq, aio);
+ return;
}
- // Cannot send again until a receive is done...
- nni_sock_senderr(s->nsock, NNG_ESTATE);
+ msg = nni_aio_get_msg(aio);
// If we have a stored backtrace, append it to the header...
// if we don't have a backtrace, discard the message.
if (s->btrace == NULL) {
- nni_msg_free(msg);
- return (NULL);
+ nni_mtx_unlock(&s->mtx);
+ nni_aio_finish_error(aio, NNG_ESTATE);
+ return;
}
// drop anything else in the header...
nni_msg_header_clear(msg);
- if (nni_msg_header_append(msg, s->btrace, s->btrace_len) != 0) {
- nni_free(s->btrace, s->btrace_len);
- s->btrace = NULL;
- s->btrace_len = 0;
- nni_msg_free(msg);
- return (NULL);
+ if ((rv = nni_msg_header_append(msg, s->btrace, s->btrace_len)) != 0) {
+ nni_mtx_unlock(&s->mtx);
+ nni_aio_finish_error(aio, rv);
+ return;
}
nni_free(s->btrace, s->btrace_len);
s->btrace = NULL;
s->btrace_len = 0;
- return (msg);
+
+ nni_mtx_unlock(&s->mtx);
+ nni_sock_send_pending(s->nsock);
+ nni_msgq_aio_put(s->uwq, aio);
}
static nni_msg *
-resp_sock_rfilter(void *arg, nni_msg *msg)
+resp_sock_filter(void *arg, nni_msg *msg)
{
resp_sock *s = arg;
char * header;
size_t len;
+ nni_mtx_lock(&s->mtx);
if (s->raw) {
+ nni_mtx_unlock(&s->mtx);
return (msg);
}
- nni_sock_senderr(s->nsock, 0);
len = nni_msg_header_len(msg);
header = nni_msg_header(msg);
if (s->btrace != NULL) {
@@ -432,15 +438,26 @@ resp_sock_rfilter(void *arg, nni_msg *msg)
s->btrace_len = 0;
}
if ((s->btrace = nni_alloc(len)) == NULL) {
+ nni_mtx_unlock(&s->mtx);
nni_msg_free(msg);
return (NULL);
}
s->btrace_len = len;
memcpy(s->btrace, header, len);
nni_msg_header_clear(msg);
+ nni_mtx_unlock(&s->mtx);
return (msg);
}
+static void
+resp_sock_recv(void *arg, nni_aio *aio)
+{
+ resp_sock *s = arg;
+
+ nni_sock_recv_pending(s->nsock);
+ nni_msgq_aio_get(s->urq, aio);
+}
+
static nni_proto_pipe_ops resp_pipe_ops = {
.pipe_init = resp_pipe_init,
.pipe_fini = resp_pipe_fini,
@@ -468,9 +485,10 @@ static nni_proto_sock_ops resp_sock_ops = {
.sock_fini = resp_sock_fini,
.sock_open = resp_sock_open,
.sock_close = resp_sock_close,
+ .sock_filter = resp_sock_filter,
+ .sock_send = resp_sock_send,
+ .sock_recv = resp_sock_recv,
.sock_options = resp_sock_options,
- .sock_rfilter = resp_sock_rfilter,
- .sock_sfilter = resp_sock_sfilter,
};
static nni_proto resp_proto = {
diff --git a/src/protocol/survey/survey.c b/src/protocol/survey/survey.c
index 1205402e..1c15054f 100644
--- a/src/protocol/survey/survey.c
+++ b/src/protocol/survey/survey.c
@@ -92,7 +92,6 @@ surv_sock_init(void **sp, nni_sock *nsock)
s->urq = nni_sock_recvq(nsock);
*sp = s;
- nni_sock_recverr(nsock, NNG_ESTATE);
return (0);
}
@@ -273,11 +272,12 @@ surv_sock_setopt_raw(void *arg, const void *buf, size_t sz)
surv_sock *s = arg;
int rv;
+ nni_mtx_lock(&s->mtx);
if ((rv = nni_setopt_int(&s->raw, buf, sz, 0, 1)) == 0) {
- nni_sock_recverr(s->nsock, s->raw ? 0 : NNG_ESTATE);
s->survid = 0;
nni_timer_cancel(&s->timer);
}
+ nni_mtx_unlock(&s->mtx);
return (rv);
}
@@ -344,22 +344,43 @@ surv_timeout(void *arg)
{
surv_sock *s = arg;
- nni_sock_lock(s->nsock);
+ nni_mtx_lock(&s->mtx);
s->survid = 0;
- nni_sock_recverr(s->nsock, NNG_ESTATE);
+ nni_mtx_unlock(&s->mtx);
nni_msgq_set_get_error(s->urq, NNG_ETIMEDOUT);
- nni_sock_unlock(s->nsock);
}
-static nni_msg *
-surv_sock_sfilter(void *arg, nni_msg *msg)
+static void
+surv_sock_recv(void *arg, nni_aio *aio)
+{
+ surv_sock *s = arg;
+
+ nni_mtx_lock(&s->mtx);
+ if (s->survid == 0) {
+ nni_mtx_unlock(&s->mtx);
+ nni_aio_finish_error(aio, NNG_ESTATE);
+ return;
+ }
+ nni_mtx_unlock(&s->mtx);
+ nni_sock_recv_pending(s->nsock);
+ nni_msgq_aio_get(s->urq, aio);
+}
+
+static void
+surv_sock_send(void *arg, nni_aio *aio)
{
surv_sock *s = arg;
+ nni_msg * msg;
+ int rv;
+ nni_mtx_lock(&s->mtx);
if (s->raw) {
// No automatic retry, and the request ID must
// be in the header coming down.
- return (msg);
+ nni_mtx_unlock(&s->mtx);
+ nni_sock_send_pending(s->nsock);
+ nni_msgq_aio_put(s->uwq, aio);
+ return;
}
// Generate a new request ID. We always set the high
@@ -367,10 +388,12 @@ surv_sock_sfilter(void *arg, nni_msg *msg)
// backtrace. (Pipe IDs have the high order bit clear.)
s->survid = (s->nextid++) | 0x80000000u;
- if (nni_msg_header_append_u32(msg, s->survid) != 0) {
- // Should be ENOMEM.
- nni_msg_free(msg);
- return (NULL);
+ msg = nni_aio_get_msg(aio);
+ nni_msg_header_clear(msg);
+ if ((rv = nni_msg_header_append_u32(msg, s->survid)) != 0) {
+ nni_mtx_unlock(&s->mtx);
+ nni_aio_finish_error(aio, rv);
+ return;
}
// If another message is there, this cancels it. We move the
@@ -379,20 +402,21 @@ surv_sock_sfilter(void *arg, nni_msg *msg)
s->expire = nni_clock() + s->survtime;
nni_timer_schedule(&s->timer, s->expire);
- // Clear the error condition.
- nni_sock_recverr(s->nsock, 0);
- // nni_msgq_set_get_error(nni_sock_recvq(psock->nsock), 0);
+ nni_mtx_unlock(&s->mtx);
- return (msg);
+ nni_sock_send_pending(s->nsock);
+ nni_msgq_aio_put(s->uwq, aio);
}
static nni_msg *
-surv_sock_rfilter(void *arg, nni_msg *msg)
+surv_sock_filter(void *arg, nni_msg *msg)
{
surv_sock *s = arg;
+ nni_mtx_lock(&s->mtx);
if (s->raw) {
// Pass it unmolested
+ nni_mtx_unlock(&s->mtx);
return (msg);
}
@@ -402,6 +426,7 @@ surv_sock_rfilter(void *arg, nni_msg *msg)
nni_msg_free(msg);
return (NULL);
}
+ nni_mtx_unlock(&s->mtx);
return (msg);
}
@@ -433,9 +458,10 @@ static nni_proto_sock_ops surv_sock_ops = {
.sock_fini = surv_sock_fini,
.sock_open = surv_sock_open,
.sock_close = surv_sock_close,
+ .sock_send = surv_sock_send,
+ .sock_recv = surv_sock_recv,
+ .sock_filter = surv_sock_filter,
.sock_options = surv_sock_options,
- .sock_rfilter = surv_sock_rfilter,
- .sock_sfilter = surv_sock_sfilter,
};
static nni_proto surv_proto = {
generated by cgit v1.2.3-70-g09d2 (git 2.46.0) at 2026-01-18 16:52:16 +0000