Missing files for GENERIC socket implementation added

Signed-off-by: Martin Sustrik <sustrik@250bpm.com>

src/generic.cpp

+/*
+    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 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
+    GNU Lesser General Public License for more details.
+
+    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/>.
+*/
+
+#include "generic.hpp"
+#include "pipe.hpp"
+#include "wire.hpp"
+#include "random.hpp"
+#include "likely.hpp"
+#include "wire.hpp"
+#include "err.hpp"
+
+zmq::generic_t::generic_t (class ctx_t *parent_, uint32_t tid_) :
+    socket_base_t (parent_, tid_),
+    prefetched (false),
+    more_in (false),
+    current_out (NULL),
+    more_out (false),
+    next_peer_id (generate_random ())
+{
+    options.type = ZMQ_GENERIC;
+
+    prefetched_msg.init ();
+}
+
+zmq::generic_t::~generic_t ()
+{
+    zmq_assert (outpipes.empty ());
+    prefetched_msg.close ();
+}
+
+void zmq::generic_t::xattach_pipe (pipe_t *pipe_)
+{
+    zmq_assert (pipe_);
+
+    //  Generate a new peer ID. Take care to avoid duplicates.
+    outpipes_t::iterator it = outpipes.lower_bound (next_peer_id);
+    if (!outpipes.empty ()) {
+        while (true) {
+            if (it == outpipes.end ())
+                it = outpipes.begin ();
+            if (it->first != next_peer_id)
+                break;
+            ++next_peer_id;
+            ++it;
+        }
+    }
+
+    //  Add the pipe to the map out outbound pipes.
+    outpipe_t outpipe = {pipe_, true};
+    bool ok = outpipes.insert (outpipes_t::value_type (
+        next_peer_id, outpipe)).second;
+    zmq_assert (ok);
+
+    //  Add the pipe to the list of inbound pipes.
+    pipe_->set_pipe_id (next_peer_id);
+    fq.attach (pipe_);
+
+    //  Queue the connection command.
+    pending_command_t cmd = {1, next_peer_id};
+    pending_commands.push_back (cmd);
+
+    //  Advance next peer ID so that if new connection is dropped shortly after
+    //  its creation we don't accidentally get two subsequent peers with
+    //  the same ID.
+    ++next_peer_id;
+}
+
+void zmq::generic_t::xterminated (pipe_t *pipe_)
+{
+    fq.terminated (pipe_);
+
+    for (outpipes_t::iterator it = outpipes.begin ();
+          it != outpipes.end (); ++it) {
+        if (it->second.pipe == pipe_) {
+
+            //  Queue the disconnection command.
+            pending_command_t cmd = {2, it->first};
+            pending_commands.push_back (cmd);
+
+            //  Remove the pipe.
+            outpipes.erase (it);
+            if (pipe_ == current_out)
+                current_out = NULL;
+            return;
+        }
+    }
+    zmq_assert (false);
+}
+
+void zmq::generic_t::xread_activated (pipe_t *pipe_)
+{
+    fq.activated (pipe_);
+}
+
+void zmq::generic_t::xwrite_activated (pipe_t *pipe_)
+{
+    for (outpipes_t::iterator it = outpipes.begin ();
+          it != outpipes.end (); ++it) {
+        if (it->second.pipe == pipe_) {
+            zmq_assert (!it->second.active);
+            it->second.active = true;
+            return;
+        }
+    }
+    zmq_assert (false);
+}
+
+int zmq::generic_t::xsend (msg_t *msg_, int flags_)
+{
+    //  If this is the first part of the message it's the ID of the
+    //  peer to send the message to.
+    if (!more_out) {
+        zmq_assert (!current_out);
+
+        //  If we have malformed message (prefix with no subsequent message)
+        //  then just silently ignore it.
+        //  TODO: The connections should be killed instead.
+        if (msg_->flags () & msg_t::label) {
+
+            more_out = true;
+
+            //  Find the pipe associated with the peer ID stored in the prefix.
+            //  If there's no such pipe just silently ignore the message.
+            zmq_assert (msg_->size () == 4);
+            uint32_t peer_id = get_uint32 ((unsigned char*) msg_->data ());
+            outpipes_t::iterator it = outpipes.find (peer_id);
+
+            if (it != outpipes.end ()) {
+                current_out = it->second.pipe;
+                msg_t empty;
+                int rc = empty.init ();
+                errno_assert (rc == 0);
+                if (!current_out->check_write (&empty)) {
+                    it->second.active = false;
+                    more_out = false;
+                    current_out = NULL;
+                }
+                rc = empty.close ();
+                errno_assert (rc == 0);
+            }
+        }
+
+        int rc = msg_->close ();
+        errno_assert (rc == 0);
+        rc = msg_->init ();
+        errno_assert (rc == 0);
+        return 0;
+    }
+
+    //  Check whether this is the last part of the message.
+    more_out = msg_->flags () & (msg_t::more | msg_t::label) ? true : false;
+
+    //  Push the message into the pipe. If there's no out pipe, just drop it.
+    if (current_out) {
+        bool ok = current_out->write (msg_);
+        zmq_assert (ok);
+        if (!more_out) {
+            current_out->flush ();
+            current_out = NULL;
+        }
+    }
+    else {
+        int rc = msg_->close ();
+        errno_assert (rc == 0);
+    }
+
+    //  Detach the message from the data buffer.
+    int rc = msg_->init ();
+    errno_assert (rc == 0);
+
+    return 0;
+}
+
+int zmq::generic_t::xrecv (msg_t *msg_, int flags_)
+{
+    //  If there's a queued command, pass it to the caller.
+    if (unlikely (!more_in && !pending_commands.empty ())) {
+        msg_->init_size (5);
+        unsigned char *data = (unsigned char*) msg_->data ();
+        put_uint8 (data, pending_commands.front ().cmd);
+        put_uint32 (data + 1, pending_commands.front ().peer);
+        msg_->set_flags (msg_t::command);
+        pending_commands.pop_front ();
+        return 0;
+    }
+
+    //  If there is a prefetched message, return it.
+    if (prefetched) {
+        int rc = msg_->move (prefetched_msg);
+        errno_assert (rc == 0);
+        more_in = msg_->flags () & (msg_t::more | msg_t::label) ? true : false;
+        prefetched = false;
+        return 0;
+    }
+
+    //  Get next message part.
+    pipe_t *pipe;
+    int rc = fq.recvpipe (msg_, flags_, &pipe);
+    if (rc != 0)
+        return -1;
+
+    //  If we are in the middle of reading a message, just return the next part.
+    if (more_in) {
+        more_in = msg_->flags () & (msg_t::more | msg_t::label) ? true : false;
+        return 0;
+    }
+ 
+    //  We are at the beginning of a new message. Move the message part we
+    //  have to the prefetched and return the ID of the peer instead.
+    rc = prefetched_msg.move (*msg_);
+    errno_assert (rc == 0);
+    prefetched = true;
+    rc = msg_->close ();
+    errno_assert (rc == 0);
+    rc = msg_->init_size (4);
+    errno_assert (rc == 0);
+    put_uint32 ((unsigned char*) msg_->data (), pipe->get_pipe_id ());
+    msg_->set_flags (msg_t::label);
+    return 0;
+}
+
+int zmq::generic_t::rollback (void)
+{
+    if (current_out) {
+        current_out->rollback ();
+        current_out = NULL;
+        more_out = false;
+    }
+    return 0;
+}
+
+bool zmq::generic_t::xhas_in ()
+{
+    if (prefetched)
+        return true;
+    return fq.has_in ();
+}
+
+bool zmq::generic_t::xhas_out ()
+{
+    //  In theory, GENERIC socket is always ready for writing. Whether actual
+    //  attempt to write succeeds depends on whitch pipe the message is going
+    //  to be routed to.
+    return true;
+}
+
+
+

src/generic.hpp

+/*
+    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 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
+    GNU Lesser General Public License for more details.
+
+    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_GENERIC_HPP_INCLUDED__
+#define __ZMQ_GENERIC_HPP_INCLUDED__
+
+#include <map>
+#include <deque>
+
+#include "socket_base.hpp"
+#include "stdint.hpp"
+#include "msg.hpp"
+#include "fq.hpp"
+
+namespace zmq
+{
+
+    //  TODO: This class uses O(n) scheduling. Rewrite it to use O(1) algorithm.
+    class generic_t :
+        public socket_base_t
+    {
+    public:
+
+        generic_t (class ctx_t *parent_, uint32_t tid_);
+        ~generic_t ();
+
+        //  Overloads of functions from socket_base_t.
+        void xattach_pipe (class pipe_t *pipe_);
+        int xsend (class msg_t *msg_, int flags_);
+        int xrecv (class msg_t *msg_, int flags_);
+        bool xhas_in ();
+        bool xhas_out ();
+        void xread_activated (class pipe_t *pipe_);
+        void xwrite_activated (class pipe_t *pipe_);
+        void xterminated (class pipe_t *pipe_);
+
+    protected:
+
+        //  Rollback any message parts that were sent but not yet flushed.
+        int rollback ();
+
+    private:
+
+        //  Fair queueing object for inbound pipes.
+        fq_t fq;
+
+        //  Have we prefetched a message.
+        bool prefetched;
+
+        //  Holds the prefetched message.
+        msg_t prefetched_msg;
+
+        //  If true, more incoming message parts are expected.
+        bool more_in;
+
+        struct outpipe_t
+        {
+            class pipe_t *pipe;
+            bool active;
+        };
+
+        //  Outbound pipes indexed by the peer IDs.
+        typedef std::map <uint32_t, outpipe_t> outpipes_t;
+        outpipes_t outpipes;
+
+        //  The pipe we are currently writing to.
+        class pipe_t *current_out;
+
+        //  If true, more outgoing message parts are expected.
+        bool more_out;
+
+        //  Peer ID are generated. It's a simple increment and wrap-over
+        //  algorithm. This value is the next ID to use (if not used already).
+        uint32_t next_peer_id;
+
+        //  Commands to be delivered to the user.
+        struct pending_command_t
+        {
+            uint8_t cmd;
+            uint32_t peer;
+        };
+        typedef std::deque <pending_command_t> pending_commands_t;
+        pending_commands_t pending_commands;
+
+        generic_t (const generic_t&);
+        const generic_t &operator = (const generic_t&);
+    };
+
+}
+
+#endif