diff options
Diffstat (limited to 'src/socket_base.hpp')
| -rw-r--r-- | src/socket_base.hpp | 206 |
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&); }; } |