diff options
Diffstat (limited to 'src/protocol/pair')
| -rw-r--r-- | src/protocol/pair/pair.c | 265 |
1 files changed, 265 insertions, 0 deletions
diff --git a/src/protocol/pair/pair.c b/src/protocol/pair/pair.c new file mode 100644 index 00000000..4b00bf87 --- /dev/null +++ b/src/protocol/pair/pair.c @@ -0,0 +1,265 @@ +/* + * 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 <stdlib.h> +#include <string.h> + +#include "core/nng_impl.h" + +/* + * Pair protocol. The PAIR protocol is a simple 1:1 messaging pattern. + */ + +typedef struct pair *pair_t; +typedef struct pairpipe *pairpipe_t; + +/* + * Note that pair can only have a single pipe, so we don't need + * to create separate data structures for diferent pipe instances. + */ +struct pair { + nng_socket_t sock; + nni_mutex_t mx; + nng_pipe_t pipe; + nni_msgqueue_t uwq; + nni_msgqueue_t urq; +}; + +struct pairpipe { + nng_pipe_t pipe; + pair_t pair; + int good; + nni_thread_t sthr; + nni_thread_t rthr; + int sigclose; +}; + +static void pair_receiver(void *); +static void pair_sender(void *); + +static int +pair_create(void **pairp, nng_socket_t sock) +{ + pair_t pair; + int rv; + + if ((pair = nni_alloc(sizeof (*pair))) == NULL) { + return (NNG_ENOMEM); + } + if ((rv = nni_mutex_create(&pair->mx)) != 0) { + nni_free(pair, sizeof (*pair)); + return (rv); + } + pair->sock = sock; + pair->uwq = nni_socket_sendq(sock); + pair->urq = nni_socket_recvq(sock); + *pairp = pair; + return (0); +} + +static void +pair_destroy(void *arg) +{ + pair_t pair = arg; + nni_mutex_destroy(pair->mx); + nni_free(pair, sizeof (*pair)); +} + +static void +pair_shutdown(void *arg, uint64_t usec) +{ + pair_t pair = arg; + nng_pipe_t pipe; + + NNI_ARG_UNUSED(usec); + + /* + * XXX: correct implementation here is to set a draining flag, + * and wait a bit for the sender to finish draining (linger), + * then reap the pipe. For now we just act a little more harshly. + */ + nni_mutex_enter(pair->mx); + pipe = pair->pipe; + pair->pipe = NULL; + nni_mutex_exit(pair->mx); + + nni_pipe_close(pipe); +} + +static int +pair_add_pipe(void *arg, nng_pipe_t pipe) +{ + pair_t pair = arg; + pairpipe_t pp; + int rv; + + pp = nni_alloc(sizeof (*pp)); + pp->pipe = pipe; + pp->good = 0; + pp->sigclose = 0; + pp->sthr = NULL; + pp->rthr = NULL; + + nni_mutex_enter(pair->mx); + if (pair->pipe != NULL) { + /* Already have a peer, denied. */ + nni_mutex_exit(pair->mx); + nni_free(pp, sizeof (*pp)); + return (NNG_EBUSY); + } + if ((rv = nni_thread_create(&pp->rthr, pair_receiver, pp)) != 0) { + nni_mutex_exit(pair->mx); + return (rv); + } + if ((rv = nni_thread_create(&pp->sthr, pair_sender, pp)) != 0) { + nni_mutex_exit(pair->mx); + return (rv); + } + pp->good = 1; + pair->pipe = pipe; + nni_mutex_exit(pair->mx); + return (NNG_EINVAL); +} + +static int +pair_remove_pipe(void *arg, nng_pipe_t pipe) +{ + pairpipe_t pp = arg; + pair_t pair = pp->pair; + + if (pp->sthr) { + (void) nni_thread_reap(pp->sthr); + } + if (pp->rthr) { + (void) nni_thread_reap(pp->rthr); + } + nni_mutex_enter(pair->mx); + if (pair->pipe != pipe) { + nni_mutex_exit(pair->mx); + return (NNG_EINVAL); + } + nni_mutex_exit(pair->mx); + return (NNG_EINVAL); +} + +static void +pair_sender(void *arg) +{ + pairpipe_t pp = arg; + pair_t pair = pp->pair; + nni_msgqueue_t uwq = pair->uwq; + nni_msgqueue_t urq = pair->urq; + nng_pipe_t pipe = pp->pipe; + nng_msg_t msg; + int rv; + + nni_mutex_enter(pair->mx); + if (!pp->good) { + nni_mutex_exit(pair->mx); + return; + } + nni_mutex_exit(pair->mx); + + + for (;;) { + rv = nni_msgqueue_get_sig(uwq, &msg, -1, &pp->sigclose); + if (rv != 0) { + (void) nni_pipe_close(pipe); + return; + } + rv = nni_pipe_send(pipe, msg); + if (rv != 0) { + nng_msg_free(msg); + (void) nni_pipe_close(pipe); + /* signal the other side */ + nni_msgqueue_signal(urq, &pp->sigclose); + return; + } + } +} + +static void +pair_receiver(void *arg) +{ + pairpipe_t pp = arg; + pair_t pair = pp->pair; + nni_msgqueue_t urq = pair->urq; + nni_msgqueue_t uwq = pair->uwq; + nng_pipe_t pipe = pp->pipe; + nng_msg_t msg; + int rv; + + nni_mutex_enter(pair->mx); + if (!pp->good) { + nni_mutex_exit(pair->mx); + return; + } + nni_mutex_exit(pair->mx); + + for (;;) { + rv = nni_pipe_recv(pipe, &msg); + if (rv != 0) { + nng_msg_free(msg); + (void) nni_pipe_close(pipe); + /* signal the other side */ + nni_msgqueue_signal(uwq, &pp->sigclose); + return; + } + rv = nni_msgqueue_put_sig(urq, msg, -1, &pp->sigclose); + if (rv != 0) { + (void) nni_pipe_close(pipe); + return; + } + } +} + +static int +pair_setopt(void *arg, int opt, const void *buf, size_t sz) +{ + return (NNG_ENOTSUP); +} + +static int +pair_getopt(void *arg, int opt, void *buf, size_t *szp) +{ + return (NNG_ENOTSUP); +} + +/* + * Global inproc state - this contains the list of active endpoints + * which we use for coordinating rendezvous. + */ + +struct nni_protocol nni_pair_protocol = { + NNG_PROTO_PAIR, /* proto_self */ + NNG_PROTO_PAIR, /* proto_peer */ + pair_create, + pair_destroy, + pair_shutdown, + pair_add_pipe, + pair_remove_pipe, + pair_setopt, + pair_getopt, + NULL, /* proto_recvfilter */ + NULL, /* proto_sendfilter */ +}; + |