From 0e9ab2e8a3f5bc22f2c331c14236a2918a5512a8 Mon Sep 17 00:00:00 2001 From: Martin Sustrik Date: Sat, 21 Nov 2009 20:59:55 +0100 Subject: inproc transport - initial commit --- src/socket_base.cpp | 62 +++++++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 60 insertions(+), 2 deletions(-) (limited to 'src/socket_base.cpp') diff --git a/src/socket_base.cpp b/src/socket_base.cpp index 6583608..86e1205 100644 --- a/src/socket_base.cpp +++ b/src/socket_base.cpp @@ -43,7 +43,9 @@ zmq::socket_base_t::socket_base_t (app_thread_t *parent_) : pending_term_acks (0), ticks (0), app_thread (parent_), - shutting_down (false) + shutting_down (false), + sent_seqnum (0), + processed_seqnum (0) { } @@ -81,6 +83,9 @@ int zmq::socket_base_t::bind (const char *addr_) addr_type = addr.substr (0, pos); addr_args = addr.substr (pos + 3); + if (addr_type == "inproc") + return register_endpoint (addr_args.c_str (), this); + if (addr_type == "tcp") { zmq_listener_t *listener = new zmq_listener_t ( choose_io_thread (options.affinity), this, options); @@ -126,6 +131,42 @@ int zmq::socket_base_t::connect (const char *addr_) addr_type = addr.substr (0, pos); addr_args = addr.substr (pos + 3); + if (addr_type == "inproc") { + + // Find the peer socket. + socket_base_t *peer = find_endpoint (addr_args.c_str ()); + if (!peer) + return -1; + + pipe_t *in_pipe = NULL; + pipe_t *out_pipe = NULL; + + // Create inbound pipe, if required. + if (options.requires_in) { + in_pipe = new pipe_t (this, peer, options.hwm, options.lwm); + zmq_assert (in_pipe); + } + + // Create outbound pipe, if required. + if (options.requires_out) { + out_pipe = new pipe_t (peer, this, options.hwm, options.lwm); + zmq_assert (out_pipe); + } + + // Attach the pipes to this socket object. + attach_pipes (in_pipe ? &in_pipe->reader : NULL, + out_pipe ? &out_pipe->writer : NULL); + + // Attach the pipes to the peer socket. Note that peer's seqnum + // was incremented in find_endpoint function. When this command + // is delivered, peer will consider the seqnum to be processed. + // TODO: Seems that 'session' parameter is unused... + send_bind (peer, NULL, out_pipe ? &out_pipe->reader : NULL, + in_pipe ? &in_pipe->writer : NULL); + + return 0; + } + // Create the session. io_thread_t *io_thread = choose_io_thread (options.affinity); session_t *session = new session_t (io_thread, this, session_name.c_str (), @@ -319,13 +360,24 @@ int zmq::socket_base_t::recv (::zmq_msg_t *msg_, int flags_) int zmq::socket_base_t::close () { + shutting_down = true; + + // Let the thread know that the socket is no longer available. app_thread->remove_socket (this); // Pointer to the dispatcher must be retrieved before the socket is // deallocated. Afterwards it is not available. dispatcher_t *dispatcher = get_dispatcher (); - shutting_down = true; + // Unregister all inproc endpoints associated with this socket. + // From this point we are sure that inc_seqnum won't be called again + // on this object. + dispatcher->unregister_endpoints (this); + + // Wait till all undelivered commands are delivered. This should happen + // very quickly. There's no way to wait here for extensive period of time. + while (processed_seqnum != sent_seqnum.get ()) + app_thread->process_commands (true, false); while (true) { @@ -364,6 +416,12 @@ int zmq::socket_base_t::close () return 0; } +void zmq::socket_base_t::inc_seqnum () +{ + // NB: This function may be called from a different thread! + sent_seqnum.add (1); +} + zmq::app_thread_t *zmq::socket_base_t::get_thread () { return app_thread; -- cgit v1.2.3 From 64634605b3ccb90d582cfdf380535c89bf900a0e Mon Sep 17 00:00:00 2001 From: Martin Sustrik Date: Sat, 21 Nov 2009 21:13:29 +0100 Subject: obsolete parameter removed from 'bind' command --- src/socket_base.cpp | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) (limited to 'src/socket_base.cpp') diff --git a/src/socket_base.cpp b/src/socket_base.cpp index 86e1205..e242e05 100644 --- a/src/socket_base.cpp +++ b/src/socket_base.cpp @@ -160,8 +160,7 @@ int zmq::socket_base_t::connect (const char *addr_) // Attach the pipes to the peer socket. Note that peer's seqnum // was incremented in find_endpoint function. When this command // is delivered, peer will consider the seqnum to be processed. - // TODO: Seems that 'session' parameter is unused... - send_bind (peer, NULL, out_pipe ? &out_pipe->reader : NULL, + send_bind (peer, out_pipe ? &out_pipe->reader : NULL, in_pipe ? &in_pipe->writer : NULL); return 0; @@ -510,8 +509,7 @@ void zmq::socket_base_t::process_own (owned_t *object_) io_objects.insert (object_); } -void zmq::socket_base_t::process_bind (owned_t *session_, - reader_t *in_pipe_, writer_t *out_pipe_) +void zmq::socket_base_t::process_bind (reader_t *in_pipe_, writer_t *out_pipe_) { attach_pipes (in_pipe_, out_pipe_); } -- cgit v1.2.3 From c41daca3da6ffd033f93c3e24898414567f71eb3 Mon Sep 17 00:00:00 2001 From: Martin Sustrik Date: Sat, 21 Nov 2009 21:30:09 +0100 Subject: race condition in inproc transport shutdown fixed --- src/socket_base.cpp | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) (limited to 'src/socket_base.cpp') diff --git a/src/socket_base.cpp b/src/socket_base.cpp index e242e05..a614759 100644 --- a/src/socket_base.cpp +++ b/src/socket_base.cpp @@ -158,10 +158,10 @@ int zmq::socket_base_t::connect (const char *addr_) out_pipe ? &out_pipe->writer : NULL); // Attach the pipes to the peer socket. Note that peer's seqnum - // was incremented in find_endpoint function. When this command - // is delivered, peer will consider the seqnum to be processed. + // was incremented in find_endpoint function. The callee is notified + // about the fact via the last parameter. send_bind (peer, out_pipe ? &out_pipe->reader : NULL, - in_pipe ? &in_pipe->writer : NULL); + in_pipe ? &in_pipe->writer : NULL, true); return 0; } @@ -509,8 +509,16 @@ void zmq::socket_base_t::process_own (owned_t *object_) io_objects.insert (object_); } -void zmq::socket_base_t::process_bind (reader_t *in_pipe_, writer_t *out_pipe_) +void zmq::socket_base_t::process_bind (reader_t *in_pipe_, writer_t *out_pipe_, + bool adjust_seqnum_) { + // In case of inproc transport, the seqnum should catch up here. + // For other transports the seqnum modification can be optimised out + // because final handshaking between the socket and the session ensures + // that no 'bind' command will be left unprocessed. + if (adjust_seqnum_) + processed_seqnum++; + attach_pipes (in_pipe_, out_pipe_); } -- cgit v1.2.3