From 82e058a2abae40760f0f9956cdb9d21f8a512624 Mon Sep 17 00:00:00 2001 From: Garrett D'Amore Date: Fri, 27 Jan 2017 01:15:46 -0800 Subject: Add device support & testing. Bus semantic fix. This adds nn_device and nng_device. There were some internal changes required to fix shutdown / close issues. Note that we shut down the sockets when exiting from device -- this is required to make both threads see the failure and bail, since we are not using a single event loop. I also noticed that the bus protocol had a bug where it would send messages back to the originator. This was specifically tested for in the compat_device test, and we have fixed it. --- src/CMakeLists.txt | 2 + src/core/device.c | 137 +++++++++++++++++++++++++++++++++++++++++++++++++ src/core/device.h | 19 +++++++ src/core/nng_impl.h | 1 + src/core/socket.c | 36 +++++++++++-- src/core/socket.h | 1 + src/nng.c | 34 +++++++++--- src/nng_compat.c | 24 +++++++++ src/nng_compat.h | 1 + src/protocol/bus/bus.c | 21 ++++++++ 10 files changed, 264 insertions(+), 12 deletions(-) create mode 100644 src/core/device.c create mode 100644 src/core/device.h (limited to 'src') diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index e9e8a31b..c81286c1 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -34,6 +34,8 @@ set (NNG_SOURCES core/clock.c core/clock.h + core/device.c + core/device.h core/endpt.c core/endpt.h core/event.c diff --git a/src/core/device.c b/src/core/device.c new file mode 100644 index 00000000..7b2b5b87 --- /dev/null +++ b/src/core/device.c @@ -0,0 +1,137 @@ +// +// Copyright 2017 Garrett D'Amore +// +// This software is supplied under the terms of the MIT License, a +// copy of which should be located in the distribution where this +// file was obtained (LICENSE.txt). A copy of the license may also be +// found online at https://opensource.org/licenses/MIT. +// + +#include "core/nng_impl.h" + +#include + +struct nni_device_pair { + nni_thr thrs[2]; + nni_sock * socks[2]; + int err[2]; +}; + +typedef struct nni_device_pair nni_device_pair; + +static int +nni_device_loop(nni_sock *from, nni_sock *to) +{ + nni_msg *msg; + int rv = 0; + + for (;;) { + // Take messages sock[0], and send to sock[1]. + // If an error occurs, we close both sockets. + if ((rv = nni_sock_recvmsg(from, &msg, NNI_TIME_NEVER)) != 0) { + break; + } + if ((rv = nni_sock_sendmsg(to, msg, NNI_TIME_NEVER)) != 0) { + nni_msg_free(msg); + break; + } + } + + return (rv); +} + + +static void +nni_device_fwd(void *p) +{ + nni_device_pair *pair = p; + + pair->err[0] = nni_device_loop(pair->socks[0], pair->socks[1]); + nni_sock_shutdown(pair->socks[0]); + nni_sock_shutdown(pair->socks[1]); +} + + +static void +nni_device_rev(void *p) +{ + nni_device_pair *pair = p; + int rv; + + pair->err[1] = nni_device_loop(pair->socks[1], pair->socks[0]); + nni_sock_shutdown(pair->socks[0]); + nni_sock_shutdown(pair->socks[1]); +} + + +int +nni_device(nni_sock *sock1, nni_sock *sock2) +{ + nni_device_pair pair; + int rv; + + memset(&pair, 0, sizeof (pair)); + pair.socks[0] = sock1; + pair.socks[1] = sock2; + + if (sock1 == NULL) { + sock1 = sock2; + } + if (sock2 == NULL) { + sock2 = sock1; + } + if ((sock1 == NULL) || (sock2 == NULL)) { + rv = NNG_EINVAL; + goto out; + } + if ((sock1->s_peer != sock2->s_protocol) || + (sock2->s_peer != sock1->s_protocol)) { + rv = NNG_EINVAL; + goto out; + } + + pair.socks[0] = sock1; + pair.socks[1] = sock2; + + if ((rv = nni_thr_init(&pair.thrs[0], nni_device_fwd, &pair)) != 0) { + goto out; + } + if ((rv = nni_thr_init(&pair.thrs[1], nni_device_rev, &pair)) != 0) { + nni_thr_fini(&pair.thrs[0]); + goto out; + } + if (((sock1->s_flags & NNI_PROTO_FLAG_RCV) != 0) && + ((sock2->s_flags & NNI_PROTO_FLAG_SND) != 0)) { + nni_thr_run(&pair.thrs[0]); + } + // If the sockets are the same, then its a simple one way forwarder, + // and we don't need two workers (but would be harmless if we did it). + if ((sock1 != sock2) && + ((sock2->s_flags & NNI_PROTO_FLAG_RCV) != 0) && + ((sock1->s_flags & NNI_PROTO_FLAG_SND) != 0)) { + nni_thr_run(&pair.thrs[1]); + } + + // This blocks on both threads (though if we didn't start one, that + // will return immediately.) + nni_thr_fini(&pair.thrs[0]); + nni_thr_fini(&pair.thrs[1]); + + nni_sock_rele(sock1); + if (sock1 != sock2) { + nni_sock_rele(sock2); + } + + rv = pair.err[0]; + if (rv == 0) { + rv = pair.err[1]; + } + if (rv == 0) { + // This can happen if neither thread ran. Shouldn't happen + // really. + rv = NNG_EINVAL; + } + +out: + return (rv); +} diff --git a/src/core/device.h b/src/core/device.h new file mode 100644 index 00000000..973a4cb7 --- /dev/null +++ b/src/core/device.h @@ -0,0 +1,19 @@ +// +// Copyright 2017 Garrett D'Amore +// +// This software is supplied under the terms of the MIT License, a +// copy of which should be located in the distribution where this +// file was obtained (LICENSE.txt). A copy of the license may also be +// found online at https://opensource.org/licenses/MIT. +// + +#ifndef CORE_DEVICE_H +#define CORE_DEVICE_H + +// Device takes messages from one side, and forwards them to the other. +// It works in both directions. Arguably we should build versions of this +// that are unidirectional, and we could extend this API with user-defined +// filtering functions. +extern int nni_device(nni_sock *, nni_sock *); + +#endif // CORE_DEVICE_H diff --git a/src/core/nng_impl.h b/src/core/nng_impl.h index 007e20a6..871fcf67 100644 --- a/src/core/nng_impl.h +++ b/src/core/nng_impl.h @@ -24,6 +24,7 @@ #include "core/defs.h" #include "core/clock.h" +#include "core/device.h" #include "core/idhash.h" #include "core/init.h" #include "core/list.h" diff --git a/src/core/socket.c b/src/core/socket.c index 6ccf3025..52a4d6d9 100644 --- a/src/core/socket.c +++ b/src/core/socket.c @@ -64,7 +64,7 @@ nni_sock_rele(nni_sock *sock) { nni_mtx_lock(nni_idlock); sock->s_refcnt--; - if ((sock->s_closed) && (sock->s_refcnt == 1)) { + if ((sock->s_closed) && (sock->s_refcnt == 0)) { nni_cv_wake(&sock->s_refcv); } nni_mtx_unlock(nni_idlock); @@ -90,9 +90,13 @@ nni_sock_hold_close(nni_sock **sockp, uint32_t id) nni_mtx_unlock(nni_idlock); return (NNG_ECLOSED); } + nni_idhash_remove(nni_sockets, id); + sock->s_id = 0; sock->s_closed = 1; - sock->s_refcnt++; - while (sock->s_refcnt != 1) { + nni_mtx_unlock(nni_idlock); + nni_sock_shutdown(sock); + nni_mtx_lock(nni_idlock); + while (sock->s_refcnt != 0) { nni_cv_wait(&sock->s_refcv); } nni_mtx_unlock(nni_idlock); @@ -102,6 +106,28 @@ nni_sock_hold_close(nni_sock **sockp, uint32_t id) } +// nni_sock_held_close uses an existing hold on the socket, but is +// otherwise pretty much the same as nni_sock_hold_close. When this +// returns there will be no other user-land references to the socket. +void +nni_sock_held_close(nni_sock *sock) +{ + nni_mtx_lock(nni_idlock); + sock->s_closed = 1; + if (sock->s_id != 0) { + nni_idhash_remove(nni_sockets, sock->s_id); + sock->s_id = 0; + } + nni_mtx_unlock(nni_idlock); + nni_sock_shutdown(sock); + nni_mtx_lock(nni_idlock); + while (sock->s_refcnt != 0) { + nni_cv_wait(&sock->s_refcv); + } + nni_mtx_unlock(nni_idlock); +} + + // Because we have to call back into the socket, and possibly also the proto, // and wait for threads to terminate, we do this in a special thread. The // assumption is that closing is always a "fast" operation. @@ -537,7 +563,9 @@ nni_sock_close(nni_sock *sock) // the results may be tragic. nni_mtx_lock(nni_idlock); - nni_idhash_remove(nni_sockets, sock->s_id); + if (sock->s_id != 0) { + nni_idhash_remove(nni_sockets, sock->s_id); + } if (nni_idhash_count(nni_sockets) == 0) { nni_idhash_reclaim(nni_pipes); nni_idhash_reclaim(nni_endpoints); diff --git a/src/core/socket.h b/src/core/socket.h index df11ded6..d7a7eb5e 100644 --- a/src/core/socket.h +++ b/src/core/socket.h @@ -74,6 +74,7 @@ struct nni_socket { extern int nni_sock_hold(nni_sock **, uint32_t); extern int nni_sock_hold_close(nni_sock **, uint32_t); +extern void nni_sock_held_close(nni_sock *); extern void nni_sock_rele(nni_sock *); extern int nni_sock_open(nni_sock **, uint16_t); extern void nni_sock_close(nni_sock *); diff --git a/src/nng.c b/src/nng.c index d77a05c7..38d259fd 100644 --- a/src/nng.c +++ b/src/nng.c @@ -348,6 +348,30 @@ nng_event_type(nng_event *ev) } +int +nng_device(nng_socket s1, nng_socket s2) +{ + int rv; + nni_sock *sock1 = NULL; + nni_sock *sock2 = NULL; + + if ((s1 > 0) && (s1 != (nng_socket)-1)) { + if ((rv = nni_sock_hold(&sock1, s1)) != 0) { + return (rv); + } + } + if (((s2 > 0) && (s2 != (nng_socket)-1)) && (s2 != s1)) { + if ((rv = nni_sock_hold(&sock2, s2)) != 0) { + nni_sock_rele(sock1); + return (rv); + } + } + + rv = nni_device(sock1, sock2); + return (rv); +} + + // Misc. const char * nng_strerror(int num) @@ -617,14 +641,6 @@ nng_stat_value(nng_stat *stat) } -int -nng_device(nng_socket sock1, nng_socket sock2) -{ - // Device TBD. - return (NNG_ENOTSUP); -} - - // These routines exist as utility functions, exposing some of our "guts" // to the external world for the purposes of test code and bundled utilities. // They should not be considered part of our public API, and applications @@ -653,6 +669,8 @@ nng_thread_create(void **thrp, void (*func)(void *), void *arg) nni_thr *thr; int rv; + nni_init(); + if ((thr = NNI_ALLOC_STRUCT(thr)) == NULL) { return (NNG_ENOMEM); } diff --git a/src/nng_compat.c b/src/nng_compat.c index 24085f30..ee52a074 100644 --- a/src/nng_compat.c +++ b/src/nng_compat.c @@ -729,6 +729,30 @@ nn_cmsg_next(struct nn_msghdr *mh, struct nn_cmsghdr *first) } +int +nn_device(int s1, int s2) +{ + int rv; + + rv = nng_device((nng_socket) s1, (nng_socket) s2); + // rv must always be nonzero + nn_seterror(rv); + return (-1); +} + + +// You should not use this function if you can help it. +void +nn_term(void) +{ + // XXX: Implement something to do something. Probably we + // should go through the nni_sockets idhash and clobber all + // of the sockets. This function is relatively toxic, since + // it can affect all sockets in the process, including those + // in use by libraries, etc. +} + + // Internal test support routines. void diff --git a/src/nng_compat.h b/src/nng_compat.h index edf4bd6a..1896f6f7 100644 --- a/src/nng_compat.h +++ b/src/nng_compat.h @@ -298,6 +298,7 @@ NN_DECL void *nn_reallocmsg(void *, size_t); NN_DECL int nn_freemsg(void *); NN_DECL int nn_errno(void); NN_DECL const char *nn_strerror(int); +NN_DECL void nn_term(void); // This stuff is intended to be exposed only for test programs and our diff --git a/src/protocol/bus/bus.c b/src/protocol/bus/bus.c index c9a1b42b..c6ecd42a 100644 --- a/src/protocol/bus/bus.c +++ b/src/protocol/bus/bus.c @@ -155,6 +155,7 @@ nni_bus_pipe_receiver(void *arg) nni_msgq *urq = nni_sock_recvq(psock->nsock); nni_msgq *uwq = nni_sock_sendq(psock->nsock); nni_pipe *npipe = ppipe->npipe; + uint32_t id = nni_pipe_id(npipe); nni_msg *msg; int rv; @@ -163,6 +164,11 @@ nni_bus_pipe_receiver(void *arg) if (rv != 0) { break; } + if ((rv = nni_msg_prepend_header(msg, &id, 4)) != 0) { + // XXX: bump a nomemory stat + nni_msg_free(msg); + continue; + } rv = nni_msgq_put_sig(urq, msg, &ppipe->sigclose); if (rv != 0) { nni_msg_free(msg); @@ -216,6 +222,7 @@ nni_bus_sock_sender(void *arg) nni_msgq *uwq = nni_sock_sendq(psock->nsock); nni_mtx *mx = nni_sock_mtx(psock->nsock); nni_msg *msg, *dup; + uint32_t sender; for (;;) { nni_bus_pipe *ppipe; @@ -226,9 +233,23 @@ nni_bus_sock_sender(void *arg) break; } + // The header being present indicates that the message + // was received locally and we are rebroadcasting. (Device + // is doing this probably.) In this case grab the pipe + // ID from the header, so we can exclude it. + if (nni_msg_header_len(msg) >= 4) { + memcpy(&sender, nni_msg_header(msg), 4); + nni_msg_trim_header(msg, 4); + } else { + sender = 0; + } + nni_mtx_lock(mx); last = nni_list_last(&psock->pipes); NNI_LIST_FOREACH (&psock->pipes, ppipe) { + if (nni_pipe_id(ppipe->npipe) == sender) { + continue; + } if (ppipe != last) { rv = nni_msg_dup(&dup, msg); if (rv != 0) { -- cgit v1.2.3-70-g09d2