zmq筆記三:socket和mailbox
int major, minor, patch;
zmq_version(&major, &minor, &patch); //4.2.0
本文主要是分析代碼,方便自己日後查閱.
=========================================
1. socket類型
每個socket類型有一個類與之對應. 所有的這些類都繼承於socket_base_t.各子類的繼承關系圖請查看筆記一.
class socket_base_t :
public own_t,
public array_item_t <>,
public i_poll_events,
public i_pipe_events
{
friend class reaper_t;
public:
......
int send (zmq::msg_t *msg_, int flags_);
int recv (zmq::msg_t *msg_, int flags_);
int add_signaler (signaler_t *s);
int remove_signaler (signaler_t *s);
int close ();
// 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 ();
......
// 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_);
......
protected:
socket_base_t (zmq::ctx_t *parent_, uint32_t tid_, int sid_, bool thread_safe_ = false);
virtual ~socket_base_t ();
.....
// The default implementation assumes that send is not supported.
virtual bool xhas_out ();
virtual int xsend (zmq::msg_t *msg_);
// The default implementation assumes that recv in not supported.
virtual bool xhas_in ();
virtual int xrecv (zmq::msg_t *msg_);
......
private:
// Creates new endpoint ID and adds the endpoint to the map.
void add_endpoint (const char *addr_, own_t *endpoint_, pipe_t *pipe);
// Map of open endpoints.
typedef std::pair <own_t *, pipe_t*> endpoint_pipe_t;
typedef std::multimap <std::string, endpoint_pipe_t> endpoints_t;
endpoints_t endpoints;
// Map of open inproc endpoints.
typedef std::multimap <std::string, pipe_t *> inprocs_t;
inprocs_t inprocs;
// Moves the flags from the message to local variables,
// to be later retrieved by getsockopt.
void extract_flags (msg_t *msg_);
......
int process_commands (int timeout_, bool throttle_);
// Socket‘s mailbox object.
i_mailbox *mailbox;
// List of attached pipes.
typedef array_t <pipe_t, 3> pipes_t;
pipes_t pipes;
// Reaper‘s poller and handle of this socket within it.
poller_t *poller;
poller_t::handle_t handle;
......
};
socket_base_t這個父類做了大部分邏輯,子類再按需實現函數重載. 拿req_t為例, req_t繼承dealer_t,dealer_t繼承socket_base_t. 子類以實現xsend/xrecv等帶x前綴的重載函數為主,而父類socket_base_t對外暴露的是不帶前綴x的函數.
class req_t : public dealer_t
{
public:
req_t (zmq::ctx_t *parent_, uint32_t tid_, int sid_);
~req_t ();
// Overrides of functions from socket_base_t.
int xsend (zmq::msg_t *msg_);
int xrecv (zmq::msg_t *msg_);
bool xhas_in ();
bool xhas_out ();
int xsetsockopt (int option_, const void *optval_, size_t optvallen_);
void xpipe_terminated (zmq::pipe_t *pipe_);
protected:
......
private:
......
// The pipe the request was sent to and where the reply is expected.
zmq::pipe_t *reply_pipe;
......
req_t (const req_t&);
const req_t &operator = (const req_t&);
};
2.mailbox
基類socket_base_t有一個成員變量 i_mailbox *mailbox. 這就是socket的郵箱了,所有投遞給socket的命令消息command_t都會放到這個郵箱的隊列裏.
zmq::socket_base_t::socket_base_t (ctx_t *parent_, uint32_t tid_, int sid_, bool thread_safe_) :
own_t (parent_, tid_),
......
thread_safe (thread_safe_),
reaper_signaler (NULL)
{
options.socket_id = sid_;
options.ipv6 = (parent_->get (ZMQ_IPV6) != 0);
options.linger = parent_->get (ZMQ_BLOCKY)? -1: 0;
if (thread_safe)
mailbox = new mailbox_safe_t(&sync);
else {
mailbox_t *m = new mailbox_t();
if (m->get_fd () != retired_fd)
mailbox = m;
else {
LIBZMQ_DELETE (m);
mailbox = NULL;
}
}
}由構造函數可知,mailbox是有線程安全的分別的, mailbox_safe_t和mailbox_t都是mutex_t sync作為訪問互斥. 這是因為mailbox的消息隊列 ypipe_t是無鎖鏈表,讀寫需要同步,ypipe_t更詳細的實現和分析可參考這篇博客.
// The pipe to store actual commands.
typedef ypipe_t <command_t, command_pipe_granularity> cpipe_t;
cpipe_t cpipe;
郵箱的sync在mailbox_safe_t是以socket_base_t的sync指針來初始化的,而mailbox_t則是獨立於socket本身的.
對於mailbox_t來說,任意時刻只能有一個線程去讀它的命令消息隊列,讀消息不用加鎖,並只需要一個signaler去通知讀線程; 而寫入消息隊列時,卻可能有多個線程寫,所以需要在寫入隊列時加鎖互斥.
// Signaler to pass signals from writer thread to reader thread.
signaler_t signaler;
對於mailbox_safe_t則是根據對socket本身的互斥訪問來讀寫它的命令消息隊列,並且有多個signaler來通知可讀狀態.
std::vector <zmq::signaler_t* > signalers;
實際上讀的時候它使用的是pthread_cond_wait和pthread_cond_broadcast的組合來獲得鎖.
void zmq::mailbox_safe_t::send (const command_t &cmd_)
{
sync->lock ();
cpipe.write (cmd_, false);
const bool ok = cpipe.flush ();
if (!ok) {
cond_var.broadcast (); //調用pthread_cond_broadcast喚醒正在等待pthread_cond_wait返回的讀線程
for (std::vector<signaler_t*>::iterator it = signalers.begin(); it != signalers.end(); ++it){
(*it)->send(); //喚醒各個reader有消息可讀
}
}
sync->unlock ();
}
int zmq::mailbox_safe_t::recv (command_t *cmd_, int timeout_)
{
// Try to get the command straight away.
if (cpipe.read (cmd_)) //無鎖隊列,能獲取消息則必定由一個線程取出,compare_and_swap原子操作
return 0;
// Wait for signal from the command sender.
int rc = cond_var.wait (sync, timeout_); //獲取sync的鎖,並休眠等待pthread_cond_broadcast信號喚醒; 註意,pthread_cond_wait返回後,其實同時也獲得了sync的鎖
if (rc == -1) {
errno_assert (errno == EAGAIN || errno == EINTR);
return -1;
}
// Another thread may already fetch the command
const bool ok = cpipe.read (cmd_);
if (!ok) {
errno = EAGAIN;
return -1;
}
return 0;
}
筆者的分析是基於mailbox_t而不是mailbox_safe_t,所以對mailbox_safe_t的使用場合並沒有經驗研究.
3.signaler
郵箱是否有可待讀取的命令消息,依靠signaler來通知.先來看一下這個類結構:
class signaler_t
{
public:
signaler_t ();
~signaler_t ();
fd_t get_fd () const;
void send ();
int wait (int timeout_);
void recv ();
int recv_failable ();
......
private:
// Creates a pair of file descriptors that will be used
// to pass the signals.
static int make_fdpair (fd_t *r_, fd_t *w_);
// Underlying write & read file descriptor
// Will be -1 if we exceeded number of available handles
fd_t w;
fd_t r;
......
};
signaler類主要是提供一對socket句柄(w/r).在支持socketpair的平臺下(*nix),可直接調用返回;而在windows平臺下,是通過打通w/r兩個socket句柄的通信.當有寫線程給mailbox發送命令消息時,判斷如果持有mailbox的讀線程掛起了,就調用mailbox的signaler->send():
void zmq::signaler_t::send ()
{
#if defined HAVE_FORK
if (unlikely (pid != getpid ())) {
//printf("Child process %d signaler_t::send returning without sending #1\n", getpid());
return; // do not send anything in forked child context
}
#endif
#if defined ZMQ_HAVE_EVENTFD
......
#elif defined ZMQ_HAVE_WINDOWS
unsigned char dummy = 0;
int nbytes = ::send (w, (char *) &dummy, sizeof (dummy), 0);
wsa_assert (nbytes != SOCKET_ERROR);
zmq_assert (nbytes == sizeof (dummy));
#else
unsigned char dummy = 0;
while (true) {
ssize_t nbytes = ::send (w, &dummy, sizeof (dummy), 0);
if (unlikely (nbytes == -1 && errno == EINTR))
continue;
#if defined(HAVE_FORK)
if (unlikely (pid != getpid ())) {
//printf("Child process %d signaler_t::send returning without sending #2\n", getpid());
errno = EINTR;
break;
}
#endif
zmq_assert (nbytes == sizeof dummy);
break;
}
#endif
}
給w發送消息,這樣r變成可讀狀態,掛起的select阻塞調用立即返回. mailbox.get_fd()返回的其實就是mailbox.signaler.r. *請註意*, signaler的w/r套接字句柄是可阻塞的.
對於非線程安全的mailbox_t,對於socket類對象,它們本身並沒有I/O線程的loop()輪詢函數,那麽它的mailbox的可讀消息狀態是由signaler的r句柄通知,由signaler.wait()函數對r進行select調用,而signaler.wait()是一般是通過socket_base_t:process_commands() -> mailbox_t:recv () -> signaler:wait () 調用鏈.當mailbox的命令隊列為空,r也沒可讀狀態時,signaler:wait (int timeout) ,傳入的timeout=-1,由於signaler的w/r是可阻塞的,這時調用process_commands()的線程將會阻塞在wait()的select調用.當然,context的I/O線程依然會繼續loop()輪詢.
那麽阻塞了socket的線程如何被喚醒? 答案是通過給socket的mailbox發送消息.
zmq::socket_base_t *zmq::ctx_t::create_socket (int type_){
......
// Create the socket and register its mailbox.
socket_base_t *s = socket_base_t::create (type_, this, slot, sid);
if (!s) {
empty_slots.push_back (slot);
slot_sync.unlock ();
return NULL;
}
sockets.push_back (s);
slots [slot] = s->get_mailbox ();
......
}
在create_socket這個函數裏,為context新增一個socket時,socket的mailbox就加入了slots的數組管理器裏. 當I/O線程(或其他知道該scoket的mailbox對應的slot id的實例)給對應的mailbox發送消息,就會喚醒正在阻塞的socket了.
zmq筆記三:socket和mailbox