diff options
Diffstat (limited to 'src/socket_base.hpp')
| -rw-r--r-- | src/socket_base.hpp | 170 |
1 files changed, 87 insertions, 83 deletions
diff --git a/src/socket_base.hpp b/src/socket_base.hpp index 3d95cec..4a72a1f 100644 --- a/src/socket_base.hpp +++ b/src/socket_base.hpp @@ -20,31 +20,41 @@ #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 "atomic_counter.hpp" +#include "signaler.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: - 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 slot_); + + // Returns the signaler associated with this socket. + signaler_t *get_signaler (); + + // 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 +65,98 @@ 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_); + + // This function should be called only on zombie sockets. It tries + // to deallocate the zombie. Returns true is object is destroyed. + bool dezombify (); protected: - // Destructor is protected. Socket is closed using 'close' function. + socket_base_t (class ctx_t *parent_, uint32_t slot_); 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 (); + + // 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; + + // 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; + // App thread's signaler object. + signaler_t signaler; - // Number of I/O objects that were already asked to terminate - // but haven't acknowledged it yet. - int pending_term_acks; + // Timestamp of when commands were processed the last time. + uint64_t last_tsc; // Number of messages received since last command processing. int ticks; @@ -144,28 +164,12 @@ 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&); |