From 4ed70a930202b103e7e80b8dc925e0aaa4622595 Mon Sep 17 00:00:00 2001 From: Martin Sustrik Date: Wed, 29 Jul 2009 12:07:54 +0200 Subject: initial commit --- src/data_distributor.cpp | 155 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 155 insertions(+) create mode 100644 src/data_distributor.cpp (limited to 'src/data_distributor.cpp') diff --git a/src/data_distributor.cpp b/src/data_distributor.cpp new file mode 100644 index 0000000..8f89c46 --- /dev/null +++ b/src/data_distributor.cpp @@ -0,0 +1,155 @@ +/* + Copyright (c) 2007-2009 FastMQ Inc. + + 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 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. + + You should have received a copy of the Lesser GNU General Public License + along with this program. If not, see . +*/ + +#include "../include/zs.h" + +#include "data_distributor.hpp" +#include "pipe_writer.hpp" +#include "err.hpp" +#include "session.hpp" +#include "msg.hpp" + +zs::data_distributor_t::data_distributor_t () : + session (NULL) +{ +} + +void zs::data_distributor_t::set_session (session_t *session_) +{ + zs_assert (!session); + session = session_; +} + +void zs::data_distributor_t::shutdown () +{ + // No need to deallocate pipes here. They'll be deallocated during the + // shutdown of the dispatcher. + delete this; +} + +void zs::data_distributor_t::terminate () +{ + // Pipe unregisters itself during the call to terminate, so the pipes + // list shinks by one in each iteration. + while (!pipes.empty ()) + pipes [0]->terminate (); + + delete this; +} + +zs::data_distributor_t::~data_distributor_t () +{ +} + +void zs::data_distributor_t::attach_pipe (pipe_writer_t *pipe_) +{ + // Associate demux with a new pipe. + pipe_->set_demux (this); + pipe_->set_index (pipes.size ()); + pipes.push_back (pipe_); +} + +void zs::data_distributor_t::detach_pipe (pipe_writer_t *pipe_) +{ + // Release the reference to the pipe. + int index = pipe_->get_index (); + pipe_->set_index (-1); + pipes [index] = pipes.back (); + pipes [index]->set_index (index); + pipes.pop_back (); +} + +bool zs::data_distributor_t::empty () +{ + return pipes.empty (); +} + +bool zs::data_distributor_t::send (zs_msg *msg_) +{ + int pipes_count = pipes.size (); + + // If there are no pipes available, simply drop the message. + if (pipes_count == 0) { + zs_msg_close (msg_); + zs_msg_init (msg_); + return true; + } + + // TODO: ??? + // First check whether all pipes are available for writing. +// for (pipes_t::iterator it = pipes.begin (); it != pipes.end (); it ++) +// if (!(*it)->check_write (msg_)) +// return false; + + // For VSMs the copying is straighforward. + if (msg_->content == (zs_msg_content*) ZS_VSM) { + for (pipes_t::iterator it = pipes.begin (); it != pipes.end (); it ++) + write_to_pipe (*it, msg_); + zs_msg_init (msg_); + return true; + } + + // Optimisation for the case when there's only a single pipe + // to send the message to - no refcount adjustment (i.e. atomic + // operations) needed. + if (pipes_count == 1) { + write_to_pipe (*pipes.begin (), msg_); + zs_msg_init (msg_); + return true; + } + + // There are at least 2 destinations for the message. That means we have + // to deal with reference counting. First add N-1 references to + // the content (we are holding one reference anyway, that's why the -1). + if (msg_->shared) + msg_->content->refcnt.add (pipes_count - 1); + else { + msg_->shared = true; + // TODO: Add memory barrier here. + msg_->content->refcnt.set (pipes_count); + } + + // Push the message to all destinations. + for (pipes_t::iterator it = pipes.begin (); it != pipes.end (); it ++) + write_to_pipe (*it, msg_); + + // Detach the original message from the data buffer. + zs_msg_init (msg_); + + return true; +} + +void zs::data_distributor_t::flush () +{ + // Flush all pipes. If there's large number of pipes, it can be pretty + // inefficient (especially if there's new message only in a single pipe). + // Can it be improved? + for (pipes_t::iterator it = pipes.begin (); it != pipes.end (); it ++) + (*it)->flush (); +} + +void zs::data_distributor_t::write_to_pipe (class pipe_writer_t *pipe_, + struct zs_msg *msg_) +{ + if (!pipe_->write (msg_)) { + // TODO: Push gap notification to the pipe. + zs_assert (false); + } +} + -- cgit v1.2.3