3

I am looking to implement a REQ-REP pattern with Python and ZeroMQ using multithreading.

With Python, I can create a new thread when a new client connects to the server. This thread will handle all communications with that particular client, until the socket is closed:

# Thread that will handle client's requests
class ClientThread(threading.Thread):
    # Implementation...
    def __init__(self, socket):
        threading.Thread.__init__(self)
        self.socket = socket
    def run(self):
        while keep_alive:
            # Thread can receive from client
            data = self.socket.recv(1024)
            # Processing...
            # And send back a reply
            self.socket.send(reply)

while True:
    # The server accepts an incoming connection
    conn, addr = sock.accept()
    # And creates a new thread to handle the client's requests
    newthread = ClientThread(conn)
    # Starting the thread
    newthread.start()

Is it possible to do the same[*] using ZeroMQ? I have seen some examples of multithreading with ZeroMQ and Python, but in all of them a pool of threads is created with a fixed number of threads at the beginning and it seems to be more oriented to load balancing.

[*] Notice what I want is to keep the connection between a client and its thread alive, as the thread is expecting multiple REQ messages from the client and it will store information that must be kept between messages (i.e.: a variable counter that increments its value on a new REQ message; so each thread has its own variable and no other client should ever be able to access that thread). New client = new thread.

1 Answer 1

5

Yes, ZeroMQ is a powerful can-do toolbox

However, the major surprise will be, that ZeroMQ <socket>-s are by far more structured than their plain counterparts, you use in the sample.

{ aZmqContext -> aZmqSocket -> aBehavioralPrimitive }

ZeroMQ builds a remarkable, abstraction-rich framework, under a hood of a "singleton" ZMQ-Context, which is (and shall remain) the only thing used as "shared".

Threads shall not "share" any other "derived" objects, the less any their state, as there is a strong distributed-responsibility framework architecture implemented, both in the sake of clean-design and a high performance & low-latency.

For all ZMQ-Socket-s one shall rather imagine a much smarter, layered sub-structure, where one receives off-loaded worries about I/O-activities ( managed inside ZMQ-Context responsibility -- thus keep-alive issues, timing issues and fair-queue buffering / select-polling issues simply cease to be visible for you ... ), with one sort of a formal communication pattern behaviour ( given by a chosen ZMQ-Socket-type archetype ).

Finally

ZeroMQ and similarly nanomsg libraries are rather LEGO-like projects, that empower you, as an architect & designer, more than one typically realises at the very beginning.

enter image description here

One thus can focus on distributed-system behaviour, as opposed to lose time and energy on solving just-another-socket-messaging-[nightmare].

( Definitely worth a look into both books from Pieter Hintjens, co-father of the ZeroMQ. There you find plenty Aha!-moments on this great subject. )

... and as a cherry on a cake -- you get all of this as a Transport-agnostic, universal environment, whether passing some messages on inproc://, other over ipc:// and also in parallel listening / speaking over tcp:// layers.

EDIT#12014-08-19 17:00 [UTC+0000]

Kindly check comments below and further review your -- both elementary and advanced -- design-options for a <trivial-failure-prone>-spin-off processing, for a <load-balanced>-REP-worker queueing, for a <scale-able>-distributed processing and a <fault-resilient_mode>-REP-worker binary-start shaded processing.

No heap of mock-up SLOC(s), no single code-sample will do a One-Size-Fits-All.

This is exponentially valid in designing distributed messaging systems.

"""REQ/REP modified with QUEUE/ROUTER/DEALER add-on ---------------------------

   Multithreaded Hello World server

   Author: Guillaume Aubert (gaubert) <guillaume(dot)aubert(at)gmail(dot)com>

"""
import time
import threading
import zmq

print "ZeroMQ version sanity-check: ", zmq.__version__

def aWorker_asRoutine( aWorker_URL, aContext = None ):
    """Worker routine"""
    #Context to get inherited or create a new one trick------------------------------
    aContext = aContext or zmq.Context.instance()
    
    # Socket to talk to dispatcher --------------------------------------------------
    socket = aContext.socket( zmq.REP )
    
    socket.connect( aWorker_URL )
    
    while True:

        string  = socket.recv()

        print( "Received request: [ %s ]" % ( string ) )
        
        # do some 'work' -----------------------------------------------------------
        time.sleep(1)

        #send reply back to client, who asked --------------------------------------
        socket.send( b"World" )

def main():
    """Server routine"""

    url_worker = "inproc://workers"
    url_client = "tcp://*:5555"

    # Prepare our context and sockets ------------------------------------------------
    aLocalhostCentralContext = zmq.Context.instance()

    # Socket to talk to clients ------------------------------------------------------
    clients = aLocalhostCentralContext.socket( zmq.ROUTER )
    clients.bind( url_client )

    # Socket to talk to workers ------------------------------------------------------
    workers = aLocalhostCentralContext.socket( zmq.DEALER )
    workers.bind( url_worker )

    # --------------------------------------------------------------------||||||||||||--
    # Launch pool of worker threads --------------< or spin-off by one in OnDemandMODE >
    for i in range(5):
        thread = threading.Thread( target = aWorker_asRoutine, args = ( url_worker, ) )
        thread.start()

    zmq.device( zmq.QUEUE, clients, workers )

    # ----------------------|||||||||||||||------------------------< a fair practice >--
    # We never get here but clean up anyhow
    clients.close()
    workers.close()
    aLocalhostCentralContext.term()

if __name__ == "__main__":
    main()
5
  • Thanks for your answer. I still don't get how to do it (which parameters should I pass to the thread?). Could you provide an example?
    – Peque
    Aug 19, 2014 at 13:25
  • @Peque Well, pitty you have found zero time to open the great Pieter book. There you have countless explanations, incl. code samples. IMHO SLOC-s never get anyone that far, compared to the proper understanding of this such complex subject. Better then to start with ZeroMQ [E=mc^2] principle >>> slideshare.net/pieterh/fosdem-2011-0mq, then with [multithreading] >>> zeromq.org/whitepapers:multithreading-magic and finally indeed check the [source-code] in Multi-threading chapter >>> zguide.zeromq.org/page:all#Multithreading-with-MQ to start realise the power Aug 19, 2014 at 16:05
  • @Peque a commented example provided for your convenience. If these recommendations help you "read-into" the new world of distributed messaging, worth your time to get the whole view before coding (mis-understood) individual parameters Aug 19, 2014 at 16:39
  • Thanks for the links and for your effort. I have already started with Code Connected, BTW. Anyway, your sample seems to create a pool of workers which is what I wanted to avoid, as specified in my question. And it does not seem to ensure that a client will always access the same thread in the pool (and no others will access that given thread).
    – Peque
    Aug 20, 2014 at 13:20
  • @Peque Good news, sir. Sure, it was commented in the illustrative sample. There are much more design-side-decisions to address and to take into account, on scheduler / dispatcher side, to meet your OP idea. [Code Connected, Vol.1] will IMHO give you a lot on this. Aug 20, 2014 at 14:10

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.