summaryrefslogtreecommitdiff
path: root/src/object.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/object.cpp')
-rw-r--r--src/object.cpp122
1 files changed, 78 insertions, 44 deletions
diff --git a/src/object.cpp b/src/object.cpp
index 324450f..90c015a 100644
--- a/src/object.cpp
+++ b/src/object.cpp
@@ -24,19 +24,18 @@
#include "err.hpp"
#include "pipe.hpp"
#include "io_thread.hpp"
-#include "owned.hpp"
#include "session.hpp"
#include "socket_base.hpp"
-zmq::object_t::object_t (ctx_t *ctx_, uint32_t thread_slot_) :
+zmq::object_t::object_t (ctx_t *ctx_, uint32_t slot_) :
ctx (ctx_),
- thread_slot (thread_slot_)
+ slot (slot_)
{
}
zmq::object_t::object_t (object_t *parent_) :
ctx (parent_->ctx),
- thread_slot (parent_->thread_slot)
+ slot (parent_->slot)
{
}
@@ -44,9 +43,9 @@ zmq::object_t::~object_t ()
{
}
-uint32_t zmq::object_t::get_thread_slot ()
+uint32_t zmq::object_t::get_slot ()
{
- return thread_slot;
+ return slot;
}
zmq::ctx_t *zmq::object_t::get_ctx ()
@@ -58,8 +57,12 @@ void zmq::object_t::process_command (command_t &cmd_)
{
switch (cmd_.type) {
- case command_t::revive:
- process_revive ();
+ case command_t::activate_reader:
+ process_activate_reader ();
+ break;
+
+ case command_t::activate_writer:
+ process_activate_writer (cmd_.args.activate_writer.msgs_read);
break;
case command_t::stop:
@@ -90,10 +93,6 @@ void zmq::object_t::process_command (command_t &cmd_)
process_seqnum ();
break;
- case command_t::reader_info:
- process_reader_info (cmd_.args.reader_info.msgs_read);
- break;
-
case command_t::pipe_term:
process_pipe_term ();
return;
@@ -123,16 +122,6 @@ void zmq::object_t::process_command (command_t &cmd_)
deallocate_command (&cmd_);
}
-void zmq::object_t::register_pipe (class pipe_t *pipe_)
-{
- ctx->register_pipe (pipe_);
-}
-
-void zmq::object_t::unregister_pipe (class pipe_t *pipe_)
-{
- ctx->unregister_pipe (pipe_);
-}
-
int zmq::object_t::register_endpoint (const char *addr_, socket_base_t *socket_)
{
return ctx->register_endpoint (addr_, socket_);
@@ -148,9 +137,19 @@ zmq::socket_base_t *zmq::object_t::find_endpoint (const char *addr_)
return ctx->find_endpoint (addr_);
}
-zmq::io_thread_t *zmq::object_t::choose_io_thread (uint64_t taskset_)
+void zmq::object_t::log (zmq_msg_t *msg_)
+{
+ ctx->log (msg_);
+}
+
+zmq::io_thread_t *zmq::object_t::choose_io_thread (uint64_t affinity_)
+{
+ return ctx->choose_io_thread (affinity_);
+}
+
+void zmq::object_t::zombify_socket (socket_base_t *socket_)
{
- return ctx->choose_io_thread (taskset_);
+ ctx->zombify_socket (socket_);
}
void zmq::object_t::send_stop ()
@@ -158,26 +157,35 @@ void zmq::object_t::send_stop ()
// 'stop' command goes always from administrative thread to
// the current object.
command_t cmd;
+#if defined ZMQ_MAKE_VALGRIND_HAPPY
+ memset (&cmd, 0, sizeof (cmd));
+#endif
cmd.destination = this;
cmd.type = command_t::stop;
- ctx->send_command (thread_slot, cmd);
+ ctx->send_command (slot, cmd);
}
-void zmq::object_t::send_plug (owned_t *destination_, bool inc_seqnum_)
+void zmq::object_t::send_plug (own_t *destination_, bool inc_seqnum_)
{
if (inc_seqnum_)
destination_->inc_seqnum ();
command_t cmd;
+#if defined ZMQ_MAKE_VALGRIND_HAPPY
+ memset (&cmd, 0, sizeof (cmd));
+#endif
cmd.destination = destination_;
cmd.type = command_t::plug;
send_command (cmd);
}
-void zmq::object_t::send_own (socket_base_t *destination_, owned_t *object_)
+void zmq::object_t::send_own (own_t *destination_, own_t *object_)
{
destination_->inc_seqnum ();
command_t cmd;
+#if defined ZMQ_MAKE_VALGRIND_HAPPY
+ memset (&cmd, 0, sizeof (cmd));
+#endif
cmd.destination = destination_;
cmd.type = command_t::own;
cmd.args.own.object = object_;
@@ -191,6 +199,9 @@ void zmq::object_t::send_attach (session_t *destination_, i_engine *engine_,
destination_->inc_seqnum ();
command_t cmd;
+#if defined ZMQ_MAKE_VALGRIND_HAPPY
+ memset (&cmd, 0, sizeof (cmd));
+#endif
cmd.destination = destination_;
cmd.type = command_t::attach;
cmd.args.attach.engine = engine_;
@@ -211,14 +222,16 @@ void zmq::object_t::send_attach (session_t *destination_, i_engine *engine_,
send_command (cmd);
}
-void zmq::object_t::send_bind (socket_base_t *destination_,
- reader_t *in_pipe_, writer_t *out_pipe_, const blob_t &peer_identity_,
- bool inc_seqnum_)
+void zmq::object_t::send_bind (own_t *destination_, reader_t *in_pipe_,
+ writer_t *out_pipe_, const blob_t &peer_identity_, bool inc_seqnum_)
{
if (inc_seqnum_)
destination_->inc_seqnum ();
command_t cmd;
+#if defined ZMQ_MAKE_VALGRIND_HAPPY
+ memset (&cmd, 0, sizeof (cmd));
+#endif
cmd.destination = destination_;
cmd.type = command_t::bind;
cmd.args.bind.in_pipe = in_pipe_;
@@ -240,27 +253,36 @@ void zmq::object_t::send_bind (socket_base_t *destination_,
send_command (cmd);
}
-void zmq::object_t::send_revive (object_t *destination_)
+void zmq::object_t::send_activate_reader (reader_t *destination_)
{
command_t cmd;
+#if defined ZMQ_MAKE_VALGRIND_HAPPY
+ memset (&cmd, 0, sizeof (cmd));
+#endif
cmd.destination = destination_;
- cmd.type = command_t::revive;
+ cmd.type = command_t::activate_reader;
send_command (cmd);
}
-void zmq::object_t::send_reader_info (writer_t *destination_,
+void zmq::object_t::send_activate_writer (writer_t *destination_,
uint64_t msgs_read_)
{
command_t cmd;
+#if defined ZMQ_MAKE_VALGRIND_HAPPY
+ memset (&cmd, 0, sizeof (cmd));
+#endif
cmd.destination = destination_;
- cmd.type = command_t::reader_info;
- cmd.args.reader_info.msgs_read = msgs_read_;
+ cmd.type = command_t::activate_writer;
+ cmd.args.activate_writer.msgs_read = msgs_read_;
send_command (cmd);
}
void zmq::object_t::send_pipe_term (writer_t *destination_)
{
command_t cmd;
+#if defined ZMQ_MAKE_VALGRIND_HAPPY
+ memset (&cmd, 0, sizeof (cmd));
+#endif
cmd.destination = destination_;
cmd.type = command_t::pipe_term;
send_command (cmd);
@@ -269,32 +291,44 @@ void zmq::object_t::send_pipe_term (writer_t *destination_)
void zmq::object_t::send_pipe_term_ack (reader_t *destination_)
{
command_t cmd;
+#if defined ZMQ_MAKE_VALGRIND_HAPPY
+ memset (&cmd, 0, sizeof (cmd));
+#endif
cmd.destination = destination_;
cmd.type = command_t::pipe_term_ack;
send_command (cmd);
}
-void zmq::object_t::send_term_req (socket_base_t *destination_,
- owned_t *object_)
+void zmq::object_t::send_term_req (own_t *destination_,
+ own_t *object_)
{
command_t cmd;
+#if defined ZMQ_MAKE_VALGRIND_HAPPY
+ memset (&cmd, 0, sizeof (cmd));
+#endif
cmd.destination = destination_;
cmd.type = command_t::term_req;
cmd.args.term_req.object = object_;
send_command (cmd);
}
-void zmq::object_t::send_term (owned_t *destination_)
+void zmq::object_t::send_term (own_t *destination_)
{
command_t cmd;
+#if defined ZMQ_MAKE_VALGRIND_HAPPY
+ memset (&cmd, 0, sizeof (cmd));
+#endif
cmd.destination = destination_;
cmd.type = command_t::term;
send_command (cmd);
}
-void zmq::object_t::send_term_ack (socket_base_t *destination_)
+void zmq::object_t::send_term_ack (own_t *destination_)
{
command_t cmd;
+#if defined ZMQ_MAKE_VALGRIND_HAPPY
+ memset (&cmd, 0, sizeof (cmd));
+#endif
cmd.destination = destination_;
cmd.type = command_t::term_ack;
send_command (cmd);
@@ -310,7 +344,7 @@ void zmq::object_t::process_plug ()
zmq_assert (false);
}
-void zmq::object_t::process_own (owned_t *object_)
+void zmq::object_t::process_own (own_t *object_)
{
zmq_assert (false);
}
@@ -327,12 +361,12 @@ void zmq::object_t::process_bind (reader_t *in_pipe_, writer_t *out_pipe_,
zmq_assert (false);
}
-void zmq::object_t::process_revive ()
+void zmq::object_t::process_activate_reader ()
{
zmq_assert (false);
}
-void zmq::object_t::process_reader_info (uint64_t msgs_read_)
+void zmq::object_t::process_activate_writer (uint64_t msgs_read_)
{
zmq_assert (false);
}
@@ -347,7 +381,7 @@ void zmq::object_t::process_pipe_term_ack ()
zmq_assert (false);
}
-void zmq::object_t::process_term_req (owned_t *object_)
+void zmq::object_t::process_term_req (own_t *object_)
{
zmq_assert (false);
}
@@ -369,6 +403,6 @@ void zmq::object_t::process_seqnum ()
void zmq::object_t::send_command (command_t &cmd_)
{
- ctx->send_command (cmd_.destination->get_thread_slot (), cmd_);
+ ctx->send_command (cmd_.destination->get_slot (), cmd_);
}