aboutsummaryrefslogtreecommitdiff
path: root/src/core/pipe.c
diff options
context:
space:
mode:
authorGarrett D'Amore <garrett@damore.org>2016-12-25 18:08:44 -0800
committerGarrett D'Amore <garrett@damore.org>2016-12-25 18:08:44 -0800
commit0cd2fa7310f1fdf45443a8a9e3335658b1c3c64c (patch)
tree1098c7f4976033bb311b45c378079700c9330b62 /src/core/pipe.c
parent64de60d98e8e4a896f9d13e4aa70343f329d88b4 (diff)
downloadnng-0cd2fa7310f1fdf45443a8a9e3335658b1c3c64c.tar.gz
nng-0cd2fa7310f1fdf45443a8a9e3335658b1c3c64c.tar.bz2
nng-0cd2fa7310f1fdf45443a8a9e3335658b1c3c64c.zip
Substantial fixes for listen & dialers.
At this point listening and dialing operations appear to function properly. As part of this I had to break the close logic up since otherwise we had a loop trying to reap a thread from itself. So there is now a separate reaper thread for pipes per-socket. I also changed lists to be a bit more rigid, and allocations now zero memory initially. (We had bugs due to uninitialized memory, and rather than hunt them all down, lets just init them to sane zero values.)
Diffstat (limited to 'src/core/pipe.c')
-rw-r--r--src/core/pipe.c42
1 files changed, 32 insertions, 10 deletions
diff --git a/src/core/pipe.c b/src/core/pipe.c
index cc92e6fe..0a7bbed1 100644
--- a/src/core/pipe.c
+++ b/src/core/pipe.c
@@ -25,14 +25,14 @@ nni_pipe_id(nni_pipe *p)
int
nni_pipe_send(nni_pipe *p, nng_msg *msg)
{
- return (p->p_ops.p_send(p->p_data, msg));
+ return (p->p_ops.p_send(p->p_trandata, msg));
}
int
nni_pipe_recv(nni_pipe *p, nng_msg **msgp)
{
- return (p->p_ops.p_recv(p->p_data, msgp));
+ return (p->p_ops.p_recv(p->p_trandata, msgp));
}
@@ -42,38 +42,60 @@ nni_pipe_recv(nni_pipe *p, nng_msg **msgp)
void
nni_pipe_close(nni_pipe *p)
{
- p->p_ops.p_close(p->p_data);
+ nni_socket *sock = p->p_sock;
+
+ p->p_ops.p_close(p->p_trandata);
+
+ nni_mutex_enter(&sock->s_mx);
+ if (!p->p_reap) {
+ // schedule deferred reap/close
+ p->p_reap = 1;
+ if (p->p_active) {
+ nni_list_remove(&sock->s_pipes, p);
+ p->p_active = 0;
+ }
+ nni_list_append(&sock->s_reaps, p);
+ nni_cond_broadcast(&sock->s_cv);
+ }
+ nni_mutex_exit(&sock->s_mx);
}
uint16_t
nni_pipe_peer(nni_pipe *p)
{
- return (p->p_ops.p_peer(p->p_data));
+ return (p->p_ops.p_peer(p->p_trandata));
}
void
nni_pipe_destroy(nni_pipe *p)
{
- if (p->p_data != NULL) {
- p->p_ops.p_destroy(p->p_data);
+ if (p->p_trandata != NULL) {
+ p->p_ops.p_destroy(p->p_trandata);
}
nni_free(p, sizeof (*p));
}
int
-nni_pipe_create(nni_pipe **pp, const nni_pipe_ops *ops)
+nni_pipe_create(nni_pipe **pp, nni_endpt *ep)
{
nni_pipe *p;
if ((p = nni_alloc(sizeof (*p))) == NULL) {
return (NNG_ENOMEM);
}
- p->p_data = NULL;
- p->p_ops = *ops;
+ p->p_trandata = NULL;
+ p->p_protdata = NULL;
+ p->p_ops = *ep->ep_ops.ep_pipe_ops;
p->p_id = nni_plat_nextid();
+ p->p_ep = ep;
+ p->p_sock = ep->ep_sock;
+ if (ep->ep_dialer != NULL) {
+ ep->ep_pipe = p;
+ }
+ NNI_LIST_NODE_INIT(&p->p_node);
*pp = p;
return (0);
}
@@ -86,5 +108,5 @@ nni_pipe_getopt(nni_pipe *p, int opt, void *val, size_t *szp)
if (p->p_ops.p_getopt == NULL) {
return (NNG_ENOTSUP);
}
- return (p->p_ops.p_getopt(p->p_data, opt, val, szp));
+ return (p->p_ops.p_getopt(p->p_trandata, opt, val, szp));
}