summaryrefslogtreecommitdiff
path: root/src/socket_base.hpp
diff options
context:
space:
mode:
authorMartin Lucina <mato@kotelna.sk>2011-03-28 10:39:51 +0200
committerMartin Lucina <martin@lucina.net>2012-01-23 08:53:37 +0100
commit3e20cb1b8a2b1ca222011df37334e5f4f88dd565 (patch)
tree4a753775186bc7f583f1ceb3f9aa675b6f110596 /src/socket_base.hpp
parent3f0085ddbef1a44b6bb7a0b23af497d56e0025fa (diff)
parente645fc2693acc796304498909786b7b47005b429 (diff)
Imported Debian patch 2.1.3-1debian/2.1.3-1
Diffstat (limited to 'src/socket_base.hpp')
-rw-r--r--src/socket_base.hpp206
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&);
};
}