added stuff

git-svn-id: svn://elaine/murooma/trunk@119 d8a302eb-03bc-478d-80e4-98257eca68ef

rtr.cpp

@@ -0,0 +1,84 @@
+//
+//  Custom routing Router to Mama (ROUTER to REQ)
+//
+// Olivier Chamoux <olivier.chamoux@fr.thalesgroup.com>
+
+#include "zhelpers.hpp"
+
+#define NBR_WORKERS 10
+
+static void *
+worker_thread (void *arg) {
+
+    zmq::context_t * context = (zmq::context_t *)arg;
+    zmq::socket_t worker (*context, ZMQ_REQ);
+    
+    //  We use a string identity for ease here
+    s_set_id (worker);
+    worker.connect("ipc://routing.ipc");
+
+    int total = 0;
+    while (1) {
+        //  Tell the router we're ready for work
+        s_send (worker, "ready");
+
+        //  Get workload from router, until finished
+        std::string workload = s_recv (worker);
+        int finished = (workload.compare("END") == 0);
+        
+        if (finished) {
+            std::cout << "Processed: " << total << " tasks" << std::endl;
+            break;
+        }
+        total++;
+
+        //  Do some random work
+        s_sleep(within (100) + 1);
+    }
+    return (NULL);
+}
+
+int main () {
+    zmq::context_t context(1);
+    zmq::socket_t client (context, ZMQ_ROUTER);
+    client.bind("ipc://routing.ipc");
+
+    int worker_nbr;
+    for (worker_nbr = 0; worker_nbr < NBR_WORKERS; worker_nbr++) {
+        pthread_t worker;
+        pthread_create (&worker, NULL, worker_thread, &context);
+    }
+    int task_nbr;
+    for (task_nbr = 0; task_nbr < NBR_WORKERS * 10; task_nbr++) {
+        //  LRU worker is next waiting in queue
+        std::string address = s_recv (client);
+        {
+            // receiving and discarding'empty' message
+            s_recv (client);
+            // receiving and discarding 'ready' message
+            s_recv (client);
+        }
+
+        s_sendmore (client, address);
+        s_sendmore (client, "");
+        s_send (client, "This is the workload");
+    }
+    //  Now ask mamas to shut down and report their results
+    for (worker_nbr = 0; worker_nbr < NBR_WORKERS; worker_nbr++) {
+        std::string address = s_recv (client);
+        {
+            // receiving and discarding'empty' message
+            s_recv (client);
+            // receiving and discarding 'ready' message
+            s_recv (client);
+        }
+
+        s_sendmore (client, address);
+        s_sendmore (client, "");
+        s_send (client, "END");
+    }
+    sleep (1);              //  Give 0MQ/2.0.x time to flush output
+    return 0;
+}
+
+

rtrClient.cpp

@@ -0,0 +1,41 @@
+//
+//  Custom routing Router to Mama (ROUTER to REQ)
+//
+// Olivier Chamoux <olivier.chamoux@fr.thalesgroup.com>
+
+#include "zhelpers.hpp"
+
+#define NBR_WORKERS 10
+
+
+int main () {
+    zmq::context_t context(1);
+    zmq::socket_t worker (context, ZMQ_REQ);
+    srand (time(NULL));
+    //  We use a string identity for ease here
+    s_set_id (worker);
+    worker.connect("tcp://127.0.0.1:10000");
+
+    int total = 0;
+    while (1) {
+        //  Tell the router we're ready for work
+        s_send (worker, "ready");
+
+        //  Get workload from router, until finished
+        std::string workload = s_recv (worker);
+std::cout << "workload: " << workload << "\n";
+        int finished = (workload.compare("END") == 0);
+        
+        if (finished) {
+            std::cout << "Processed: " << total << " tasks" << std::endl;
+            break;
+        }
+        total++;
+
+        //  Do some random work
+        s_sleep(within (100) + 1);
+    }
+    return 0;
+}
+
+

rtrServer.cpp

@@ -0,0 +1,36 @@
+//
+//  Custom routing Router to Mama (ROUTER to REQ)
+//
+// Olivier Chamoux <olivier.chamoux@fr.thalesgroup.com>
+
+#include "zhelpers.hpp"
+
+#define NBR_WORKERS 10
+
+
+int main () {
+    zmq::context_t context(2);
+    zmq::socket_t client (context, ZMQ_ROUTER);
+    client.bind("tcp://127.0.0.1:10000");
+
+    int task_nbr;
+	while (true)
+	{
+        //  LRU worker is next waiting in queue
+        std::string address = s_recv (client);
+		std::cout << "Address: " << address << "\n";
+		
+        // receiving and discarding'empty' message
+        s_recv (client);
+        // receiving and discarding 'ready' message
+        std::string msg = s_recv (client);
+		std::cout << "msg: " << msg << "\n";
+
+        s_sendmore (client, address);
+        s_sendmore (client, "");
+        s_send (client, std::string("hello: ") + address);
+    }
+    return 0;
+}
+
+

zhelpers.hpp

@@ -0,0 +1,245 @@
+/*  =========================================================================
+    zhelpers.h - ZeroMQ helpers for example applications
+
+    Copyright (c) 1991-2010 iMatix Corporation and contributors
+
+    This 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.
+
+    This software 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/>.
+    =========================================================================
+*/
+
+// Olivier Chamoux <olivier.chamoux@fr.thalesgroup.com>
+
+
+#ifndef __ZHELPERS_HPP_INCLUDED__
+#define __ZHELPERS_HPP_INCLUDED__
+
+//  Include a bunch of headers that we will need in the examples
+
+#include <zmq.hpp>
+
+#include <iostream>
+#include <iomanip>
+#include <string>
+#include <sstream>
+
+#if (!defined(__WINDOWS__))
+#include <sys/time.h>
+#include <unistd.h>
+#include <pthread.h>
+#endif
+#include <time.h>
+#include <assert.h>
+#include <stdlib.h>        // random()  RAND_MAX
+#include <stdio.h>
+#include <stdarg.h>
+#include <signal.h>
+
+//  Bring Windows MSVC up to C99 scratch
+#if (defined (__WINDOWS__))
+    typedef unsigned long ulong;
+    typedef unsigned int  uint;
+    typedef __int64 int64_t;
+#endif
+
+//  Provide random number from 0..(num-1)
+#if (!defined(__WINDOWS__))
+#define within(num) (int) ((float) (num) * random () / (RAND_MAX + 1.0))
+#else
+#define within(num) (int) ((float) (num) * rand () / (RAND_MAX + 1.0))
+#endif
+
+//  Receive 0MQ string from socket and convert into string
+static std::string
+s_recv (zmq::socket_t & socket) {
+
+    zmq::message_t message;
+    socket.recv(&message);
+
+    return std::string(static_cast<char*>(message.data()), message.size());
+}
+
+//  Convert string to 0MQ string and send to socket
+static bool
+s_send (zmq::socket_t & socket, const std::string & string) {
+
+    zmq::message_t message(string.size());
+    memcpy (message.data(), string.data(), string.size());
+
+    bool rc = socket.send (message);
+    return (rc);
+}
+
+//  Sends string as 0MQ string, as multipart non-terminal
+static bool
+s_sendmore (zmq::socket_t & socket, const std::string & string) {
+
+    zmq::message_t message(string.size());
+    memcpy (message.data(), string.data(), string.size());
+
+    bool rc = socket.send (message, ZMQ_SNDMORE);
+    return (rc);
+}
+
+//  Receives all message parts from socket, prints neatly
+//
+static void
+s_dump (zmq::socket_t & socket)
+{
+    std::cout << "----------------------------------------" << std::endl;
+
+    while (1) {
+        //  Process all parts of the message
+        zmq::message_t message;
+        socket.recv(&message);
+
+        //  Dump the message as text or binary
+        int size = message.size();
+        std::string data(static_cast<char*>(message.data()), size);
+
+        bool is_text = true;
+
+        int char_nbr;
+        unsigned char byte;
+        for (char_nbr = 0; char_nbr < size; char_nbr++) {
+            byte = data [char_nbr];
+            if (byte < 32 || byte > 127)
+                is_text = false;
+        }
+        std::cout << "[" << std::setfill('0') << std::setw(3) << size << "]";
+        for (char_nbr = 0; char_nbr < size; char_nbr++) {
+            if (is_text)
+                std::cout << (char)data [char_nbr];
+            else
+                std::cout << std::setfill('0') << std::setw(2)
+                   << std::hex << (unsigned int) data [char_nbr];
+        }
+        std::cout << std::endl;
+
+        int more = 0;           //  Multipart detection
+        size_t more_size = sizeof (more);
+        socket.getsockopt (ZMQ_RCVMORE, &more, &more_size);
+        if (!more)
+            break;              //  Last message part
+    }
+}
+
+//  Set simple random printable identity on socket
+//
+inline std::string
+s_set_id (zmq::socket_t & socket)
+{
+    std::stringstream ss;
+    ss << std::hex << std::uppercase
+       << std::setw(4) << std::setfill('0') << within (0x10000) << "-"
+       << std::setw(4) << std::setfill('0') << within (0x10000);
+    socket.setsockopt(ZMQ_IDENTITY, ss.str().c_str(), ss.str().length());
+    return ss.str();
+}
+
+//  Report 0MQ version number
+//
+static void
+s_version (void)
+{
+    int major, minor, patch;
+    zmq_version (&major, &minor, &patch);
+    std::cout << "Current 0MQ version is " << major << "." << minor << "." << patch << std::endl;
+}
+
+static void
+s_version_assert (int want_major, int want_minor)
+{
+    int major, minor, patch;
+    zmq_version (&major, &minor, &patch);
+    if (major < want_major
+    || (major == want_major && minor < want_minor)) {
+        std::cout << "Current 0MQ version is " << major << "." << minor << std::endl;
+        std::cout << "Application needs at least " << want_major << "." << want_minor
+              << " - cannot continue" << std::endl;
+        exit (EXIT_FAILURE);
+    }
+}
+
+//  Return current system clock as milliseconds
+static int64_t
+s_clock (void)
+{
+#if (defined (__WINDOWS__))
+    SYSTEMTIME st;
+    GetSystemTime (&st);
+    return (int64_t) st.wSecond * 1000 + st.wMilliseconds;
+#else
+    struct timeval tv;
+    gettimeofday (&tv, NULL);
+    return (int64_t) (tv.tv_sec * 1000 + tv.tv_usec / 1000);
+#endif
+}
+
+//  Sleep for a number of milliseconds
+static void
+s_sleep (int msecs)
+{
+#if (defined (__WINDOWS__))
+    Sleep (msecs);
+#else
+    struct timespec t;
+    t.tv_sec = msecs / 1000;
+    t.tv_nsec = (msecs % 1000) * 1000000;
+    nanosleep (&t, NULL);
+#endif
+}
+
+static void
+s_console (const char *format, ...)
+{
+    time_t curtime = time (NULL);
+    struct tm *loctime = localtime (&curtime);
+    char *formatted = new char[20];
+    strftime (formatted, 20, "%y-%m-%d %H:%M:%S ", loctime);
+    printf ("%s", formatted);
+    delete[] formatted;
+
+    va_list argptr;
+    va_start (argptr, format);
+    vprintf (format, argptr);
+    va_end (argptr);
+    printf ("\n");
+}
+
+//  ---------------------------------------------------------------------
+//  Signal handling
+//
+//  Call s_catch_signals() in your application at startup, and then exit
+//  your main loop if s_interrupted is ever 1. Works especially well with
+//  zmq_poll.
+
+static int s_interrupted = 0;
+static void s_signal_handler (int signal_value)
+{
+    s_interrupted = 1;
+}
+
+static void s_catch_signals ()
+{
+#if (!defined(__WINDOWS__))
+    struct sigaction action;
+    action.sa_handler = s_signal_handler;
+    action.sa_flags = 0;
+    sigemptyset (&action.sa_mask);
+    sigaction (SIGINT, &action, NULL);
+    sigaction (SIGTERM, &action, NULL);
+#endif
+}
+
+#endif