aboutsummaryrefslogtreecommitdiff
path: root/src/protocol/survey/respond.c
diff options
context:
space:
mode:
Diffstat (limited to 'src/protocol/survey/respond.c')
-rw-r--r--src/protocol/survey/respond.c161
1 files changed, 72 insertions, 89 deletions
diff --git a/src/protocol/survey/respond.c b/src/protocol/survey/respond.c
index 73cc4792..6a784738 100644
--- a/src/protocol/survey/respond.c
+++ b/src/protocol/survey/respond.c
@@ -16,8 +16,8 @@
// the surveyor pattern. This is useful for building service discovery, or
// voting algorithsm, for example.
-typedef struct nni_resp_pipe nni_resp_pipe;
-typedef struct nni_resp_sock nni_resp_sock;
+typedef struct nni_resp_pipe nni_resp_pipe;
+typedef struct nni_resp_sock nni_resp_sock;
static void nni_resp_recv_cb(void *);
static void nni_resp_putq_cb(void *);
@@ -28,31 +28,30 @@ static void nni_resp_pipe_fini(void *);
// An nni_resp_sock is our per-socket protocol private structure.
struct nni_resp_sock {
- nni_sock * nsock;
- nni_msgq * urq;
- nni_msgq * uwq;
- int raw;
- int ttl;
- nni_idhash pipes;
- char * btrace;
- size_t btrace_len;
- nni_aio aio_getq;
- nni_mtx mtx;
+ nni_sock * nsock;
+ nni_msgq * urq;
+ nni_msgq * uwq;
+ int raw;
+ int ttl;
+ nni_idhash pipes;
+ char * btrace;
+ size_t btrace_len;
+ nni_aio aio_getq;
+ nni_mtx mtx;
};
// An nni_resp_pipe is our per-pipe protocol private structure.
struct nni_resp_pipe {
- nni_pipe * npipe;
- nni_resp_sock * psock;
- uint32_t id;
- nni_msgq * sendq;
- nni_aio aio_getq;
- nni_aio aio_putq;
- nni_aio aio_send;
- nni_aio aio_recv;
+ nni_pipe * npipe;
+ nni_resp_sock *psock;
+ uint32_t id;
+ nni_msgq * sendq;
+ nni_aio aio_getq;
+ nni_aio aio_putq;
+ nni_aio aio_send;
+ nni_aio aio_recv;
};
-
static void
nni_resp_sock_fini(void *arg)
{
@@ -69,23 +68,22 @@ nni_resp_sock_fini(void *arg)
}
}
-
static int
nni_resp_sock_init(void **pp, nni_sock *nsock)
{
nni_resp_sock *psock;
- int rv;
+ int rv;
if ((psock = NNI_ALLOC_STRUCT(psock)) == NULL) {
return (NNG_ENOMEM);
}
- psock->ttl = 8; // Per RFC
- psock->nsock = nsock;
- psock->raw = 0;
- psock->btrace = NULL;
+ psock->ttl = 8; // Per RFC
+ psock->nsock = nsock;
+ psock->raw = 0;
+ psock->btrace = NULL;
psock->btrace_len = 0;
- psock->urq = nni_sock_recvq(nsock);
- psock->uwq = nni_sock_sendq(nsock);
+ psock->urq = nni_sock_recvq(nsock);
+ psock->uwq = nni_sock_sendq(nsock);
if (((rv = nni_idhash_init(&psock->pipes)) != 0) ||
((rv = nni_mtx_init(&psock->mtx)) != 0)) {
goto fail;
@@ -104,7 +102,6 @@ fail:
return (rv);
}
-
static void
nni_resp_sock_open(void *arg)
{
@@ -113,7 +110,6 @@ nni_resp_sock_open(void *arg)
nni_msgq_aio_get(psock->uwq, &psock->aio_getq);
}
-
static void
nni_resp_sock_close(void *arg)
{
@@ -122,12 +118,11 @@ nni_resp_sock_close(void *arg)
nni_aio_stop(&psock->aio_getq);
}
-
static int
nni_resp_pipe_init(void **pp, nni_pipe *npipe, void *psock)
{
nni_resp_pipe *ppipe;
- int rv;
+ int rv;
if ((ppipe = NNI_ALLOC_STRUCT(ppipe)) == NULL) {
return (NNG_ENOMEM);
@@ -154,7 +149,7 @@ nni_resp_pipe_init(void **pp, nni_pipe *npipe, void *psock)
ppipe->npipe = npipe;
ppipe->psock = psock;
- *pp = ppipe;
+ *pp = ppipe;
return (0);
fail:
@@ -162,7 +157,6 @@ fail:
return (rv);
}
-
static void
nni_resp_pipe_fini(void *arg)
{
@@ -176,13 +170,12 @@ nni_resp_pipe_fini(void *arg)
NNI_FREE_STRUCT(ppipe);
}
-
static int
nni_resp_pipe_start(void *arg)
{
nni_resp_pipe *ppipe = arg;
nni_resp_sock *psock = ppipe->psock;
- int rv;
+ int rv;
ppipe->id = nni_pipe_id(ppipe->npipe);
@@ -199,7 +192,6 @@ nni_resp_pipe_start(void *arg)
return (rv);
}
-
static void
nni_resp_pipe_stop(void *arg)
{
@@ -220,7 +212,6 @@ nni_resp_pipe_stop(void *arg)
nni_mtx_unlock(&psock->mtx);
}
-
// nni_resp_sock_send watches for messages from the upper write queue,
// extracts the destination pipe, and forwards it to the appropriate
// destination pipe via a separate queue. This prevents a single bad
@@ -230,16 +221,16 @@ void
nni_resp_sock_getq_cb(void *arg)
{
nni_resp_sock *psock = arg;
- nni_msg *msg;
- uint8_t *header;
- uint32_t id;
+ nni_msg * msg;
+ uint8_t * header;
+ uint32_t id;
nni_resp_pipe *ppipe;
- int rv;
+ int rv;
if (nni_aio_result(&psock->aio_getq) != 0) {
return;
}
- msg = psock->aio_getq.a_msg;
+ msg = psock->aio_getq.a_msg;
psock->aio_getq.a_msg = NULL;
// We yank the outgoing pipe id from the header
@@ -269,7 +260,6 @@ nni_resp_sock_getq_cb(void *arg)
nni_mtx_unlock(&psock->mtx);
}
-
void
nni_resp_getq_cb(void *arg)
{
@@ -286,7 +276,6 @@ nni_resp_getq_cb(void *arg)
nni_pipe_send(ppipe->npipe, &ppipe->aio_send);
}
-
void
nni_resp_send_cb(void *arg)
{
@@ -302,17 +291,16 @@ nni_resp_send_cb(void *arg)
nni_msgq_aio_get(ppipe->sendq, &ppipe->aio_getq);
}
-
static void
nni_resp_recv_cb(void *arg)
{
nni_resp_pipe *ppipe = arg;
nni_resp_sock *psock = ppipe->psock;
- nni_msgq *urq;
- nni_msg *msg;
- uint8_t idbuf[4];
- int hops;
- int rv;
+ nni_msgq * urq;
+ nni_msg * msg;
+ uint8_t idbuf[4];
+ int hops;
+ int rv;
if (nni_aio_result(&ppipe->aio_recv) != 0) {
goto error;
@@ -322,7 +310,7 @@ nni_resp_recv_cb(void *arg)
NNI_PUT32(idbuf, ppipe->id);
- msg = ppipe->aio_recv.a_msg;
+ msg = ppipe->aio_recv.a_msg;
ppipe->aio_recv.a_msg = NULL;
// Store the pipe id in the header, first thing.
@@ -334,7 +322,7 @@ nni_resp_recv_cb(void *arg)
// Move backtrace from body to header
hops = 0;
for (;;) {
- int end = 0;
+ int end = 0;
uint8_t *body;
if (hops >= psock->ttl) {
@@ -346,8 +334,8 @@ nni_resp_recv_cb(void *arg)
goto error;
}
body = nni_msg_body(msg);
- end = (body[0] & 0x80) ? 1 : 0;
- rv = nni_msg_append_header(msg, body, 4);
+ end = (body[0] & 0x80) ? 1 : 0;
+ rv = nni_msg_append_header(msg, body, 4);
if (rv != 0) {
nni_msg_free(msg);
goto error;
@@ -367,7 +355,6 @@ error:
nni_pipe_stop(ppipe->npipe);
}
-
static void
nni_resp_putq_cb(void *arg)
{
@@ -382,13 +369,12 @@ nni_resp_putq_cb(void *arg)
nni_pipe_recv(ppipe->npipe, &ppipe->aio_recv);
}
-
static int
nni_resp_sock_setopt(void *arg, int opt, const void *buf, size_t sz)
{
nni_resp_sock *psock = arg;
- int rv;
- int oldraw;
+ int rv;
+ int oldraw;
switch (opt) {
case NNG_OPT_MAXTTL:
@@ -396,7 +382,7 @@ nni_resp_sock_setopt(void *arg, int opt, const void *buf, size_t sz)
break;
case NNG_OPT_RAW:
oldraw = psock->raw;
- rv = nni_setopt_int(&psock->raw, buf, sz, 0, 1);
+ rv = nni_setopt_int(&psock->raw, buf, sz, 0, 1);
if (oldraw != psock->raw) {
if (!psock->raw) {
nni_sock_senderr(psock->nsock, 0);
@@ -411,12 +397,11 @@ nni_resp_sock_setopt(void *arg, int opt, const void *buf, size_t sz)
return (rv);
}
-
static int
nni_resp_sock_getopt(void *arg, int opt, void *buf, size_t *szp)
{
nni_resp_sock *psock = arg;
- int rv;
+ int rv;
switch (opt) {
case NNG_OPT_MAXTTL:
@@ -431,7 +416,6 @@ nni_resp_sock_getopt(void *arg, int opt, void *buf, size_t *szp)
return (rv);
}
-
static nni_msg *
nni_resp_sock_sfilter(void *arg, nni_msg *msg)
{
@@ -454,38 +438,38 @@ nni_resp_sock_sfilter(void *arg, nni_msg *msg)
// drop anything else in the header...
nni_msg_trunc_header(msg, nni_msg_header_len(msg));
- if (nni_msg_append_header(msg, psock->btrace, psock->btrace_len) != 0) {
+ if (nni_msg_append_header(msg, psock->btrace, psock->btrace_len) !=
+ 0) {
nni_free(psock->btrace, psock->btrace_len);
- psock->btrace = NULL;
+ psock->btrace = NULL;
psock->btrace_len = 0;
nni_msg_free(msg);
return (NULL);
}
nni_free(psock->btrace, psock->btrace_len);
- psock->btrace = NULL;
+ psock->btrace = NULL;
psock->btrace_len = 0;
return (msg);
}
-
static nni_msg *
nni_resp_sock_rfilter(void *arg, nni_msg *msg)
{
nni_resp_sock *psock = arg;
- char *header;
- size_t len;
+ char * header;
+ size_t len;
if (psock->raw) {
return (msg);
}
nni_sock_senderr(psock->nsock, 0);
- len = nni_msg_header_len(msg);
+ len = nni_msg_header_len(msg);
header = nni_msg_header(msg);
if (psock->btrace != NULL) {
nni_free(psock->btrace, psock->btrace_len);
- psock->btrace = NULL;
+ psock->btrace = NULL;
psock->btrace_len = 0;
}
if ((psock->btrace = nni_alloc(len)) == NULL) {
@@ -498,30 +482,29 @@ nni_resp_sock_rfilter(void *arg, nni_msg *msg)
return (msg);
}
-
static nni_proto_pipe_ops nni_resp_pipe_ops = {
- .pipe_init = nni_resp_pipe_init,
- .pipe_fini = nni_resp_pipe_fini,
- .pipe_start = nni_resp_pipe_start,
- .pipe_stop = nni_resp_pipe_stop,
+ .pipe_init = nni_resp_pipe_init,
+ .pipe_fini = nni_resp_pipe_fini,
+ .pipe_start = nni_resp_pipe_start,
+ .pipe_stop = nni_resp_pipe_stop,
};
static nni_proto_sock_ops nni_resp_sock_ops = {
- .sock_init = nni_resp_sock_init,
- .sock_fini = nni_resp_sock_fini,
- .sock_open = nni_resp_sock_open,
- .sock_close = nni_resp_sock_close,
- .sock_setopt = nni_resp_sock_setopt,
- .sock_getopt = nni_resp_sock_getopt,
- .sock_rfilter = nni_resp_sock_rfilter,
- .sock_sfilter = nni_resp_sock_sfilter,
+ .sock_init = nni_resp_sock_init,
+ .sock_fini = nni_resp_sock_fini,
+ .sock_open = nni_resp_sock_open,
+ .sock_close = nni_resp_sock_close,
+ .sock_setopt = nni_resp_sock_setopt,
+ .sock_getopt = nni_resp_sock_getopt,
+ .sock_rfilter = nni_resp_sock_rfilter,
+ .sock_sfilter = nni_resp_sock_sfilter,
};
nni_proto nni_respondent_proto = {
- .proto_self = NNG_PROTO_RESPONDENT,
- .proto_peer = NNG_PROTO_SURVEYOR,
- .proto_name = "respondent",
- .proto_flags = NNI_PROTO_FLAG_SNDRCV,
+ .proto_self = NNG_PROTO_RESPONDENT,
+ .proto_peer = NNG_PROTO_SURVEYOR,
+ .proto_name = "respondent",
+ .proto_flags = NNI_PROTO_FLAG_SNDRCV,
.proto_sock_ops = &nni_resp_sock_ops,
.proto_pipe_ops = &nni_resp_pipe_ops,
};
generated by cgit v1.2.3-70-g09d2 (git 2.46.0) at 2026-01-18 12:50:49 +0000