mirror of
https://github.com/zeromq/libzmq.git
synced 2025-03-14 10:02:04 +01:00
da31917f4f
Relicense permission collected from all relevant authors as tallied at: https://github.com/rlenferink/libzmq-relicense/blob/master/checklist.md The relicense grants are collected under RELICENSE/ and will be moved to the above repository in a later commit. Fixes https://github.com/zeromq/libzmq/issues/2376
123 lines
2.5 KiB
C++
123 lines
2.5 KiB
C++
/* SPDX-License-Identifier: MPL-2.0 */
|
|
|
|
#include "precompiled.hpp"
|
|
#include "macros.hpp"
|
|
#include "reaper.hpp"
|
|
#include "socket_base.hpp"
|
|
#include "err.hpp"
|
|
|
|
zmq::reaper_t::reaper_t (class ctx_t *ctx_, uint32_t tid_) :
|
|
object_t (ctx_, tid_),
|
|
_mailbox_handle (static_cast<poller_t::handle_t> (NULL)),
|
|
_poller (NULL),
|
|
_sockets (0),
|
|
_terminating (false)
|
|
{
|
|
if (!_mailbox.valid ())
|
|
return;
|
|
|
|
_poller = new (std::nothrow) poller_t (*ctx_);
|
|
alloc_assert (_poller);
|
|
|
|
if (_mailbox.get_fd () != retired_fd) {
|
|
_mailbox_handle = _poller->add_fd (_mailbox.get_fd (), this);
|
|
_poller->set_pollin (_mailbox_handle);
|
|
}
|
|
|
|
#ifdef HAVE_FORK
|
|
_pid = getpid ();
|
|
#endif
|
|
}
|
|
|
|
zmq::reaper_t::~reaper_t ()
|
|
{
|
|
LIBZMQ_DELETE (_poller);
|
|
}
|
|
|
|
zmq::mailbox_t *zmq::reaper_t::get_mailbox ()
|
|
{
|
|
return &_mailbox;
|
|
}
|
|
|
|
void zmq::reaper_t::start ()
|
|
{
|
|
zmq_assert (_mailbox.valid ());
|
|
|
|
// Start the thread.
|
|
_poller->start ("Reaper");
|
|
}
|
|
|
|
void zmq::reaper_t::stop ()
|
|
{
|
|
if (get_mailbox ()->valid ()) {
|
|
send_stop ();
|
|
}
|
|
}
|
|
|
|
void zmq::reaper_t::in_event ()
|
|
{
|
|
while (true) {
|
|
#ifdef HAVE_FORK
|
|
if (unlikely (_pid != getpid ())) {
|
|
//printf("zmq::reaper_t::in_event return in child process %d\n", (int)getpid());
|
|
return;
|
|
}
|
|
#endif
|
|
|
|
// Get the next command. If there is none, exit.
|
|
command_t cmd;
|
|
const int rc = _mailbox.recv (&cmd, 0);
|
|
if (rc != 0 && errno == EINTR)
|
|
continue;
|
|
if (rc != 0 && errno == EAGAIN)
|
|
break;
|
|
errno_assert (rc == 0);
|
|
|
|
// Process the command.
|
|
cmd.destination->process_command (cmd);
|
|
}
|
|
}
|
|
|
|
void zmq::reaper_t::out_event ()
|
|
{
|
|
zmq_assert (false);
|
|
}
|
|
|
|
void zmq::reaper_t::timer_event (int)
|
|
{
|
|
zmq_assert (false);
|
|
}
|
|
|
|
void zmq::reaper_t::process_stop ()
|
|
{
|
|
_terminating = true;
|
|
|
|
// If there are no sockets being reaped finish immediately.
|
|
if (!_sockets) {
|
|
send_done ();
|
|
_poller->rm_fd (_mailbox_handle);
|
|
_poller->stop ();
|
|
}
|
|
}
|
|
|
|
void zmq::reaper_t::process_reap (socket_base_t *socket_)
|
|
{
|
|
// Add the socket to the poller.
|
|
socket_->start_reaping (_poller);
|
|
|
|
++_sockets;
|
|
}
|
|
|
|
void zmq::reaper_t::process_reaped ()
|
|
{
|
|
--_sockets;
|
|
|
|
// If reaped was already asked to terminate and there are no more sockets,
|
|
// finish immediately.
|
|
if (!_sockets && _terminating) {
|
|
send_done ();
|
|
_poller->rm_fd (_mailbox_handle);
|
|
_poller->stop ();
|
|
}
|
|
}
|