Merge pull request #858 from lalebarde/master

add proxy_chain, a multi proxies chaining in the same thread feature

doc/zmq_proxy_chain.txt

+zmq_proxy_chain(3)
+==================
+
+NAME
+----
+zmq_proxy_chain - start built-in 0MQ a proxy chain in the same thread
+control flow
+
+
+SYNOPSIS
+--------
+*int zmq_proxy_chain (const void '**frontends', const void '**backends', 
+     const void '*capture', const void **hooks_, const void '*control');*
+
+
+DESCRIPTION
+-----------
+The _zmq_proxy_chain()_ function starts the built-in 0MQ proxy in the 
+current application thread, as _zmq_proxy()_, _zmq_proxy_steerable()_, or 
+_zmq_proxy_hook()_ do. Please, refer to these functions for their general 
+description and usage. We describe here only the additional proxy chaining 
+capability.
+
+Note that compared to the other proxy functions, the arguments _frontends_, 
+_backends_ and _hooks_ receive arrays instead of single values. Say one need 
+to implement the following architecture:
+
+*Process      client              proxy1                  proxy2              worker*
+               |             |-----------|             |----------|            |
+*socket*         cl           f1          b1            f2          b2           wk
+*endpoint*       |c----e1-----b|           |c----e2-----b|          |c----e3----b|
+
+Note: "c" is for connect, "b" for bind.
+
+With the other proxy functions, one needs typically one thread for each proxy:
+----
+thread 1: zmq_proxy(f1, b1);
+thread 2: zmq_proxy(f2, b2);
+----
+
+With _zmq_proxy_chain_, it can be performed with only one thread:
+----
+void** f = {f1, f2, NULL);
+void** b = {b1, b2, NULL);
+single thread: zmq_proxy_chain(f, b, NULL, NULL, NULL);
+----
+
+Note: the three NULL arguments are for capture, hooks, and control, since 
+_zmq_proxy_chain_ is built on top of _zmq_proxy_hook_, itself built on top 
+of _zmq_proxy_steerable_, itself built on top of _zmq_proxy_. Of course, hook and  
+steering features can be used along with chaining.
+
+We have limited the number of sockets that can be chained in a single command to 10, what 
+should be largely sufficient. The reason is to avoid dynamic memory allocation.
+
+Arguments frontends and backends shall be arrays of sockets of type void*, terminated 
+by NULL. Both arrays shall terminate by NULL at the same indice, otherwise, an error is 
+returned. Argument hooks shall be NULL or of the same length than the socket arrays. No 
+NULL is required at the end of hooks. Any number of elements may be NULL where no hook 
+is implemented in some proxies.
+
+
+Refer to linkzmq:zmq_socket[3] for a description of the available socket types.
+Refer to linkzmq:zmq_proxy[3] for a description of the zmq_proxy.
+Refer to linkzmq:zmq_proxy_steerable[3] for a description of the zmq_steerable.
+Refer to linkzmq:zmq_proxy_hook[3] for a description of the zmq_hook.
+
+EXAMPLE USAGE
+-------------
+_zmq_proxy_chain_ aims at building protocol layers by easing the chaining of some 
+proxies typically by chaining:
+DEALER | ROUTER <---> STREAM <---> DEALER
+in the same thread. Any kind of protocol feature can be added via hooks.
+
+cf also zmq_proxy, zmq_proxy_steerable, zmq_proxy_hook.
+
+RETURN VALUE
+------------
+The _zmq_proxy_chain()_ function returns 0 if TERMINATE is sent to its
+control socket. Otherwise, it returns `-1` and 'errno' set to *ETERM* (the
+0MQ 'context' associated with either of the specified sockets was terminated).
+
+
+EXAMPLE
+-------
+cf test_proxy_chain.cpp
+An example capable of proxying CURVE will be added soon.
+
+SEE ALSO
+--------
+linkzmq:zmq_proxy[3]
+linkzmq:zmq_proxy_steerable[3]
+linkzmq:zmq_proxy_hook[3]
+linkzmq:zmq_bind[3]
+linkzmq:zmq_connect[3]
+linkzmq:zmq_socket[3]
+linkzmq:zmq[7]
+
+
+AUTHORS
+-------
+This page was written by the 0MQ community. To make a change please
+read the 0MQ Contribution Policy at <http://www.zeromq.org/docs:contributing>.

include/zmq.h

@@ -398,9 +398,11 @@ ZMQ_EXPORT int zmq_poll (zmq_pollitem_t *items, int nitems, long timeout);
 
 /*  Built-in message proxy (3-way) */
 
+#define ZMQ_PROXY_CHAIN_MAX_LENGTH 10
 ZMQ_EXPORT int zmq_proxy (void *frontend, void *backend, void *capture);
 ZMQ_EXPORT int zmq_proxy_steerable (void *frontend, void *backend, void *capture, void *control);
 ZMQ_EXPORT int zmq_proxy_hook (void *frontend, void *backend, void *capture, void *hook, void *control);
+ZMQ_EXPORT int zmq_proxy_chain (void **frontends_, void **backends_, void *capture_, void **hooks_, void *control_);
 
 typedef int (*zmq_hook_f)(void *frontend, void *backend, void *capture, zmq_msg_t* msg_, size_t n_, void *data_);
 typedef struct zmq_proxy_hook_t {

src/proxy.cpp

@@ -119,14 +119,12 @@ forward(
 
 int
 zmq::proxy (
-    class socket_base_t *frontend_,
-    class socket_base_t *backend_,
-    class socket_base_t *capture_,
-    class socket_base_t *control_,
-    zmq::proxy_hook_t *hook_)
+        class socket_base_t **frontend_,
+        class socket_base_t **backend_,
+        class socket_base_t *capture_,
+        class socket_base_t *control_,
+        zmq::proxy_hook_t **hook_)
 {
-    static zmq::proxy_hook_t dummy_hook = {NULL, NULL, NULL};
-
     msg_t msg;
     int rc = msg.init ();
     if (rc != 0)
@@ -137,12 +135,41 @@ zmq::proxy (
 
     int more;
     size_t moresz;
-    zmq_pollitem_t items [] = {
-        { frontend_, 0, ZMQ_POLLIN, 0 },
-        { backend_, 0, ZMQ_POLLIN, 0 },
-        { control_, 0, ZMQ_POLLIN, 0 }
-    };
-    int qt_poll_items = (control_ ? 3 : 2);
+    size_t n = 0; // number of pair of sockets: the array ends with NULL
+    for (;; n++) { // counts the number of pair of sockets
+        if (!frontend_[n] && !backend_[n])
+            break;
+        if (!frontend_[n] || !backend_[n]) {
+            errno = EFAULT;
+            return -1;
+        }
+    }
+    if (!n) {
+        errno = EFAULT;
+        return -1;
+    }
+    // avoid dynamic allocation as we have no guarranty to reach the deallocator => limit the chain length
+    zmq_assert(n <= ZMQ_PROXY_CHAIN_MAX_LENGTH);
+    zmq_pollitem_t items [2 * ZMQ_PROXY_CHAIN_MAX_LENGTH + 1]; // +1 for the control socket
+    static zmq_pollitem_t null_item = { NULL, 0, ZMQ_POLLIN, 0 };
+    static zmq::proxy_hook_t dummy_hook = {NULL, NULL, NULL};
+    static zmq::proxy_hook_t* no_hooks[ZMQ_PROXY_CHAIN_MAX_LENGTH];
+    if (!hook_)
+        hook_ = no_hooks;
+    else
+        for (size_t i = 0; i < n; i++)
+            if (!hook_[i]) // Check if a hook is used
+                hook_[i] = &dummy_hook;
+    for (size_t i = 0; i < n; i++) {
+        memcpy(&items[2 * i], &null_item, sizeof(null_item));
+        items[2 * i].socket =     frontend_[i];
+        memcpy(&items[2 * i + 1], &null_item, sizeof(null_item));
+        items[2 * i + 1].socket = backend_[i];
+        no_hooks[i] = &dummy_hook;
+    }
+    memcpy(&items[2 * n], &null_item, sizeof(null_item));
+    items[2 * n].socket =     control_;
+    int qt_poll_items = (control_ ? 2 * n + 1 : 2 * n);
 
     //  Proxy can be in these three states
     enum {
@@ -158,7 +185,7 @@ zmq::proxy (
             return -1;
 
         //  Process a control command if any
-        if (control_ && items [2].revents & ZMQ_POLLIN) {
+        if (control_ && items [2 * n].revents & ZMQ_POLLIN) {
             rc = control_->recv (&msg, 0);
             if (unlikely (rc < 0))
                 return -1;
@@ -187,22 +214,23 @@ zmq::proxy (
                 zmq_assert (false);
             }
         }
-        // Check if a hook is used
-        if (!hook_)
-            hook_ = &dummy_hook;
-        //  Process a request
-        if (state == active
-        &&  items [0].revents & ZMQ_POLLIN) {
-            rc = forward(frontend_, backend_, capture_, msg, hook_->front2back_hook, hook_->data);
-            if (unlikely (rc < 0))
-                return -1;
-        }
-        //  Process a reply
-        if (state == active
-        &&  items [1].revents & ZMQ_POLLIN) {
-            rc = forward(backend_, frontend_, capture_, msg, hook_->back2front_hook, hook_->data);
-            if (unlikely (rc < 0))
-                return -1;
+
+        // process each pair of sockets
+        for (size_t i = 0; i < n; i++) {
+            //  Process a request
+            if (state == active
+            &&  items [2 * i].revents & ZMQ_POLLIN) {
+                rc = forward(frontend_[i], backend_[i], capture_, msg, hook_[i]->front2back_hook, hook_[i]->data);
+                if (unlikely (rc < 0))
+                    return -1;
+            }
+            //  Process a reply
+            if (state == active
+            &&  items [2 * i + 1].revents & ZMQ_POLLIN) {
+                rc = forward(backend_[i], frontend_[i], capture_, msg, hook_[i]->back2front_hook, hook_[i]->data);
+                if (unlikely (rc < 0))
+                    return -1;
+            }
         }
     }
     return 0;

src/proxy.hpp

@@ -32,11 +32,11 @@ namespace zmq
     };
 
     int proxy (
-        class socket_base_t *frontend_,
-        class socket_base_t *backend_,
-        class socket_base_t *capture_ = NULL,
-        class socket_base_t *control_ = NULL, // backward compatibility without this argument
-        proxy_hook_t *hook_ = NULL // backward compatibility without this argument
+            class socket_base_t **frontend_,
+            class socket_base_t **backend_,
+            class socket_base_t *capture_ = NULL,
+            class socket_base_t *control_ = NULL, // backward compatibility without this argument
+            proxy_hook_t **hook_ = NULL // backward compatibility without this argument
         );
 }
 

src/zmq.cpp

@@ -1025,48 +1025,55 @@ typedef char check_proxy_hook_t_size
 
 int zmq_proxy (void *frontend_, void *backend_, void *capture_)
 {
-    if (!frontend_ || !backend_) {
-        errno = EFAULT;
-        return -1;
-    }
+    zmq::socket_base_t* frontends_[] = {(zmq::socket_base_t*) frontend_, NULL};
+    zmq::socket_base_t* backends_[] = {(zmq::socket_base_t*) backend_, NULL};
     return zmq::proxy (
-        (zmq::socket_base_t*) frontend_,
-        (zmq::socket_base_t*) backend_,
+        (zmq::socket_base_t**) frontends_,
+        (zmq::socket_base_t**) backends_,
         (zmq::socket_base_t*) capture_);
 }
 
 int zmq_proxy_steerable (void *frontend_, void *backend_, void *capture_, void *control_)
 {
-    if (!frontend_ || !backend_) {
-        errno = EFAULT;
-        return -1;
-    }
+    zmq::socket_base_t* frontends_[] = {(zmq::socket_base_t*) frontend_, NULL};
+    zmq::socket_base_t* backends_[] = {(zmq::socket_base_t*) backend_, NULL};
     return zmq::proxy (
-        (zmq::socket_base_t*) frontend_,
-        (zmq::socket_base_t*) backend_,
+        (zmq::socket_base_t**) frontends_,
+        (zmq::socket_base_t**) backends_,
         (zmq::socket_base_t*) capture_,
         (zmq::socket_base_t*) control_);
 }
 
 int zmq_proxy_hook (void *frontend_, void *backend_, void *capture_, void *hook_, void *control_)
 {
-    if (!frontend_ || !backend_) {
-        errno = EFAULT;
-        return -1;
-    }
+    zmq::socket_base_t* frontends_[] = {(zmq::socket_base_t*) frontend_, NULL};
+    zmq::socket_base_t* backends_[] = {(zmq::socket_base_t*) backend_, NULL};
+    zmq::proxy_hook_t* hooks_[] = {(zmq::proxy_hook_t*) hook_};
+    return zmq::proxy (
+        (zmq::socket_base_t**) frontends_,
+        (zmq::socket_base_t**) backends_,
+        (zmq::socket_base_t*) capture_,
+        (zmq::socket_base_t*) control_,
+        (zmq::proxy_hook_t**)  hooks_);
+}
+
+int zmq_proxy_chain (void **frontends_, void **backends_, void *capture_, void **hooks_, void *control_)
+{
     return zmq::proxy (
-        (zmq::socket_base_t*) frontend_,
-        (zmq::socket_base_t*) backend_,
+        (zmq::socket_base_t**) frontends_,
+        (zmq::socket_base_t**) backends_,
         (zmq::socket_base_t*) capture_,
         (zmq::socket_base_t*) control_,
-        (zmq::proxy_hook_t*)  hook_);
+        (zmq::proxy_hook_t**)  hooks_);
 }
 
 //  The deprecated device functionality
 
 int zmq_device (int /* type */, void *frontend_, void *backend_)
 {
+    zmq::socket_base_t* frontends_[] = {(zmq::socket_base_t*) frontend_, NULL};
+    zmq::socket_base_t* backends_[] = {(zmq::socket_base_t*) backend_, NULL};
     return zmq::proxy (
-        (zmq::socket_base_t*) frontend_,
-        (zmq::socket_base_t*) backend_);
+        (zmq::socket_base_t**) frontends_,
+        (zmq::socket_base_t**) backends_);
 }

tests/Makefile.am

@@ -44,6 +44,7 @@ noinst_PROGRAMS = test_system \
                   test_inproc_connect \
                   test_issue_566 \
                   test_proxy \
+                  test_proxy_chain \
                   test_abstract_ipc \
                   test_many_sockets \
                   test_ipc_wildcard \
@@ -110,6 +111,7 @@ test_conflate_SOURCES = test_conflate.cpp
 test_inproc_connect_SOURCES = test_inproc_connect.cpp
 test_issue_566_SOURCES = test_issue_566.cpp
 test_proxy_SOURCES = test_proxy.cpp
+test_proxy_chain_SOURCES = test_proxy_chain.cpp
 test_abstract_ipc_SOURCES = test_abstract_ipc.cpp
 test_many_sockets_SOURCES = test_many_sockets.cpp
 test_ipc_wildcard_SOURCES = test_ipc_wildcard.cpp