From a8b410e66c3c75809c8e9c01dd3e35c579f02347 Mon Sep 17 00:00:00 2001
From: Martin Sustrik <sustrik@fastmq.commkdir>
Date: Sat, 8 Aug 2009 16:01:58 +0200
Subject: lockfree interaction patter for 3 theads implemented

---
 src/socket_base.cpp | 129 ++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 129 insertions(+)
 create mode 100644 src/socket_base.cpp

(limited to 'src/socket_base.cpp')

diff --git a/src/socket_base.cpp b/src/socket_base.cpp
new file mode 100644
index 0000000..3737410
--- /dev/null
+++ b/src/socket_base.cpp
@@ -0,0 +1,129 @@
+/*
+    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 <http://www.gnu.org/licenses/>.
+*/
+
+#include <algorithm>
+
+#include "../include/zmq.h"
+
+#include "socket_base.hpp"
+#include "app_thread.hpp"
+#include "err.hpp"
+#include "zmq_listener.hpp"
+#include "io_thread.hpp"
+
+zmq::socket_base_t::socket_base_t (app_thread_t *parent_) :
+    object_t (parent_),
+    pending_term_acks (0),
+    app_thread (parent_)
+{
+}
+
+zmq::socket_base_t::~socket_base_t ()
+{
+    while (true) {
+
+        //  On third pass of the loop there should be no more I/O objects
+        //  because all connecters and listerners were destroyed during
+        //  the first pass and all engines delivered by delayed 'own' commands
+        //  are destroyed during the second pass.
+        if (io_objects.empty () && !pending_term_acks)
+            break;
+
+        //  Send termination request to all associated I/O objects.
+        for (io_objects_t::size_type i = 0; i != io_objects.size (); i++)
+            send_term (io_objects [i]);
+
+        //  Move the objects to the list of pending term acks.
+        pending_term_acks += io_objects.size ();
+        io_objects.clear ();
+
+        //  Process commands till we get all the termination acknowledgements.
+        while (pending_term_acks)
+            app_thread->process_commands (true);
+    }
+}
+
+int zmq::socket_base_t::bind (const char *addr_, struct zmq_opts *opts_)
+{
+    uint64_t taskset = opts_ ? opts_->taskset : 0;
+    object_t *listener = new zmq_listener_t (choose_io_thread (taskset), this);
+    send_plug (listener);
+    send_own (this, listener);
+    return 0;
+}
+
+int zmq::socket_base_t::connect (const char *addr_, struct zmq_opts *opts_)
+{
+    zmq_assert (false);
+}
+
+int zmq::socket_base_t::subscribe (const char *criteria_)
+{
+    zmq_assert (false);
+}
+
+int zmq::socket_base_t::send (struct zmq_msg *msg_, int flags_)
+{
+    zmq_assert (false);
+}
+
+int zmq::socket_base_t::flush ()
+{
+    zmq_assert (false);
+}
+
+int zmq::socket_base_t::recv (struct zmq_msg *msg_, int flags_)
+{
+    zmq_assert (false);
+}
+
+int zmq::socket_base_t::close ()
+{
+    app_thread->remove_socket (this);
+    delete this;
+    return 0;
+}
+
+void zmq::socket_base_t::process_own (object_t *object_)
+{
+    io_objects.push_back (object_);
+}
+
+void zmq::socket_base_t::process_term_req (object_t *object_)
+{
+    //  If I/O object is well and alive ask it to terminate.
+    //  TODO: Following find may produce an unacceptable jitter in
+    //        C10K-style applications. If so, use set instead of vector.
+    io_objects_t::iterator it = std::find (io_objects.begin (),
+        io_objects.end (), object_);
+    if (it != io_objects.end ()) {
+        pending_term_acks++;
+        io_objects.erase (it);
+        send_term (object_);
+    }
+
+    //  If not found, we assume that termination request was already sent to
+    //  the object so we can sagely ignore the request.
+}
+
+void zmq::socket_base_t::process_term_ack ()
+{
+    zmq_assert (pending_term_acks);
+    pending_term_acks--;
+}
-- 
cgit v1.2.3