Imported Debian patch 2.1.3-1debian/2.1.3-1

1 files changed, 115 insertions, 91 deletions

diff --git a/src/socket_base.hpp b/src/socket_base.hpp
index 3d95cec..15ac83c 100644
--- a/src/socket_base.hpp
+++ b/src/socket_base.hpp
@@ -1,50 +1,65 @@
 /*
-    Copyright (c) 2007-2010 iMatix Corporation
+    Copyright (c) 2007-2011 iMatix Corporation
+    Copyright (c) 2007-2011 Other contributors as noted in the AUTHORS file
 
     This file is part of 0MQ.
 
     0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
+    the terms of the GNU Lesser General Public License as published by
     the Free Software Foundation; either version 3 of the License, or
     (at your option) any later version.
 
     0MQ is distributed in the hope that it will be useful,
     but WITHOUT ANY WARRANTY; without even the implied warranty of
     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-    Lesser GNU General Public License for more details.
+    GNU Lesser General Public License for more details.
 
-    You should have received a copy of the Lesser GNU General Public License
+    You should have received a copy of the GNU Lesser General Public License
     along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
 
 #ifndef __ZMQ_SOCKET_BASE_HPP_INCLUDED__
 #define __ZMQ_SOCKET_BASE_HPP_INCLUDED__
 
-#include <set>
 #include <map>
 #include <vector>
 
 #include "../include/zmq.h"
 
-#include "i_endpoint.hpp"
-#include "object.hpp"
-#include "yarray_item.hpp"
+#include "own.hpp"
+#include "array.hpp"
 #include "mutex.hpp"
-#include "options.hpp"
 #include "stdint.hpp"
+#include "poller.hpp"
 #include "atomic_counter.hpp"
+#include "i_poll_events.hpp"
+#include "mailbox.hpp"
 #include "stdint.hpp"
 #include "blob.hpp"
+#include "own.hpp"
 
 namespace zmq
 {
 
     class socket_base_t :
-        public object_t, public i_endpoint, public yarray_item_t
+        public own_t,
+        public array_item_t,
+        public i_poll_events
     {
+        friend class reaper_t;
+
     public:
 
-        socket_base_t (class app_thread_t *parent_);
+        //  Create a socket of a specified type.
+        static socket_base_t *create (int type_, class ctx_t *parent_,
+            uint32_t tid_);
+
+        //  Returns the mailbox associated with this socket.
+        mailbox_t *get_mailbox ();
+
+        //  Interrupt blocking call if the socket is stuck in one.
+        //  This function can be called from a different thread!
+        void stop ();
 
         //  Interface for communication with the API layer.
         int setsockopt (int option_, const void *optval_, size_t optvallen_);
@@ -55,88 +70,113 @@ namespace zmq
         int recv (zmq_msg_t *msg_, int flags_);
         int close ();
 
-        //  When another owned object wants to send command to this object
-        //  it calls this function to let it know it should not shut down
-        //  before the command is delivered.
-        void inc_seqnum ();
-
-        //  This function is used by the polling mechanism to determine
-        //  whether the socket belongs to the application thread the poll
-        //  is called from.
-        class app_thread_t *get_thread ();
-
         //  These functions are used by the polling mechanism to determine
         //  which events are to be reported from this socket.
         bool has_in ();
         bool has_out ();
 
-        //  The list of sessions cannot be accessed via inter-thread
-        //  commands as it is unacceptable to wait for the completion of the
-        //  action till user application yields control of the application
-        //  thread to 0MQ. Locking is used instead.
-        //  There are two distinct types of sessions: those identified by name
-        //  and those identified by ordinal number. Thus two sets of session
-        //  management functions.
-        bool register_session (const blob_t &peer_identity_,
-            class session_t *session_);
-        void unregister_session (const blob_t &peer_identity_);
-        class session_t *find_session (const blob_t &peer_identity_);
-        uint64_t register_session (class session_t *session_);
-        void unregister_session (uint64_t ordinal_);
-        class session_t *find_session (uint64_t ordinal_);
-
-        //  i_endpoint interface implementation.
-        void attach_pipes (class reader_t *inpipe_, class writer_t *outpipe_,
-            const blob_t &peer_identity_);
-        void detach_inpipe (class reader_t *pipe_);
-        void detach_outpipe (class writer_t *pipe_);
-        void kill (class reader_t *pipe_);
-        void revive (class reader_t *pipe_);
-        void revive (class writer_t *pipe_);
+        //  Registry of named sessions.
+        bool register_session (const blob_t &name_, class session_t *session_);
+        void unregister_session (const blob_t &name_);
+        class session_t *find_session (const blob_t &name_);
+
+        //  i_reader_events interface implementation.
+        void activated (class reader_t *pipe_);
+        void terminated (class reader_t *pipe_);
+
+        //  i_writer_events interface implementation.
+        void activated (class writer_t *pipe_);
+        void terminated (class writer_t *pipe_);
+
+        //  Using this function reaper thread ask the socket to regiter with
+        //  its poller.
+        void start_reaping (poller_t *poller_);
+
+        //  i_poll_events implementation. This interface is used when socket
+        //  is handled by the poller in the reaper thread.
+        void in_event ();
+        void out_event ();
+        void timer_event (int id_);
+
+        //  To be called after processing commands or invoking any command
+        //  handlers explicitly. If required, it will deallocate the socket.
+        void check_destroy ();
 
     protected:
 
-        //  Destructor is protected. Socket is closed using 'close' function.
+        socket_base_t (class ctx_t *parent_, uint32_t tid_);
         virtual ~socket_base_t ();
 
-        //  Pipe management is done by individual socket types.
+        //  Concrete algorithms for the x- methods are to be defined by
+        //  individual socket types.
         virtual void xattach_pipes (class reader_t *inpipe_,
             class writer_t *outpipe_, const blob_t &peer_identity_) = 0;
-        virtual void xdetach_inpipe (class reader_t *pipe_) = 0;
-        virtual void xdetach_outpipe (class writer_t *pipe_) = 0;
-        virtual void xkill (class reader_t *pipe_) = 0;
-        virtual void xrevive (class reader_t *pipe_) = 0;
-        virtual void xrevive (class writer_t *pipe_) = 0;
 
-        //  Actual algorithms are to be defined by individual socket types.
+        //  The default implementation assumes there are no specific socket
+        //  options for the particular socket type. If not so, overload this
+        //  method.
         virtual int xsetsockopt (int option_, const void *optval_,
-            size_t optvallen_) = 0;
-        virtual int xsend (zmq_msg_t *msg_, int options_) = 0;
-        virtual int xrecv (zmq_msg_t *msg_, int options_) = 0;
-        virtual bool xhas_in () = 0;
-        virtual bool xhas_out () = 0;
+            size_t optvallen_);
+
+        //  The default implementation assumes that send is not supported.
+        virtual bool xhas_out ();
+        virtual int xsend (zmq_msg_t *msg_, int options_);
+
+        //  The default implementation assumes that recv in not supported.
+        virtual bool xhas_in ();
+        virtual int xrecv (zmq_msg_t *msg_, int options_);
 
-        //  Socket options.
-        options_t options;
+        //  We are declaring termination handler as protected so that
+        //  individual socket types can hook into the termination process
+        //  by overloading it.
+        void process_term (int linger_);
+
+        //  Delay actual destruction of the socket.
+        void process_destroy ();
 
     private:
 
+        //  If true, associated context was already terminated.
+        bool ctx_terminated;
+
+        //  If true, object should have been already destroyed. However,
+        //  destruction is delayed while we unwind the stack to the point
+        //  where it doesn't intersect the object being destroyed.
+        bool destroyed;
+
+        //  Parse URI string.
+        int parse_uri (const char *uri_, std::string &protocol_,
+            std::string &address_);
+
+        //  Check whether transport protocol, as specified in connect or
+        //  bind, is available and compatible with the socket type.
+        int check_protocol (const std::string &protocol_);
+
+        //  If no identity set generate one and call xattach_pipes ().
+        void attach_pipes (class reader_t *inpipe_, class writer_t *outpipe_,
+            const blob_t &peer_identity_);
+
+        //  Processes commands sent to this socket (if any). If 'block' is
+        //  set to true, returns only after at least one command was processed.
+        //  If throttle argument is true, commands are processed at most once
+        //  in a predefined time period.
+        int process_commands (bool block_, bool throttle_);
+
         //  Handlers for incoming commands.
-        void process_own (class owned_t *object_);
+        void process_stop ();
         void process_bind (class reader_t *in_pipe_, class writer_t *out_pipe_,
             const blob_t &peer_identity_);
-        void process_term_req (class owned_t *object_);
-        void process_term_ack ();
-        void process_seqnum ();
+        void process_unplug ();
 
-        //  List of all I/O objects owned by this socket. The socket is
-        //  responsible for deallocating them before it quits.
-        typedef std::set <class owned_t*> io_objects_t;
-        io_objects_t io_objects;
+        //  Socket's mailbox object.
+        mailbox_t mailbox;
 
-        //  Number of I/O objects that were already asked to terminate
-        //  but haven't acknowledged it yet.
-        int pending_term_acks;
+        //  Reaper's poller and handle of this socket within it.
+        poller_t *poller;
+        poller_t::handle_t handle;
+
+        //  Timestamp of when commands were processed the last time.
+        uint64_t last_tsc;
 
         //  Number of messages received since last command processing.
         int ticks;
@@ -144,32 +184,16 @@ namespace zmq
         //  If true there's a half-read message in the socket.
         bool rcvmore;
 
-        //  Application thread the socket lives in.
-        class app_thread_t *app_thread;
-
-        //  If true, socket is already shutting down. No new work should be
-        //  started.
-        bool shutting_down;
-
-        //  Sequence number of the last command sent to this object.
-        atomic_counter_t sent_seqnum;
-
-        //  Sequence number of the last command processed by this object.
-        uint64_t processed_seqnum;
-
-        //  Lists of existing sessions. This lists are never referenced from
-        //  within the socket, instead they are used by I/O objects owned by
+        //  Lists of existing sessions. This list is never referenced from
+        //  within the socket, instead it is used by objects owned by
         //  the socket. As those objects can live in different threads,
         //  the access is synchronised by mutex.
-        typedef std::map <blob_t, session_t*> named_sessions_t;
-        named_sessions_t named_sessions;
-        typedef std::map <uint64_t, session_t*> unnamed_sessions_t;
-        unnamed_sessions_t unnamed_sessions;
-        uint64_t next_ordinal;
+        typedef std::map <blob_t, session_t*> sessions_t;
+        sessions_t sessions;
         mutex_t sessions_sync;
 
         socket_base_t (const socket_base_t&);
-        void operator = (const socket_base_t&);
+        const socket_base_t &operator = (const socket_base_t&);
     };
 
 }