ZeroMQ 3 Advanced Request-Reply Patterns

基於 request-reply pattern 的 advanced request-reply pattern

• request-reply 機制如何運作
• 如何組合 REQ, REP, DEALER, ROUTER sockets
• ROUTER sockets 運作方式
• load balancing pattern
• 如何製作 simple load balancing message broker
• 以 ØMQ 設計高階 API
• 如何製作非同步 request-reply server
• 詳細的 inter-broker routing 範例

Request-Reply 機制

reply message envelopes

envelope 是包裝 data 增加 address 的方法，藉由分離 address 到 envelope 的方法，可以製作一般化的 intermediaries，例如 APIs, proxies，用來 create, read, remove addresses。

在 request-reply pattern 中，envelope 會儲存 return address(用在 reply)，這也是無狀態的 ØMQ network 可以建立 round-trip request-reply dialog 的方法。

如果是使用 REQ, REP sockets，也不會看到 envelopes，sockets 會自動處理。

The Simple Reply Envelope

request-reply 訊息交換包含了 request 以及 reply message，最簡單的 pattern 針對每一個 request 只有一個 reply，在 advanced pattern 裡面，requests 及 replies 可以用非同步方式運作，不過 reply envelope 都是一樣的。

ØMQ reply enveope 包含了 0 到多個 reply addresses，後面是 empty frame (envelope delimiter)，接下來是 message body (0 到多個 frames)，envelope 是由多個 socket 合作產生出來的。

由 REQ socket 發送最間單的 "Hello" 開始，REQ socket 會產生最簡單的 reply envelope，沒有 addresses，只有一個 empty delimiter frame 及 message frame。

REP socket 會 strips off the envelope，儲存整個 envelope，將 "Hello" 傳給 application，因此 application 並不會看到這個 envelope。

如果監聽 hwclient 及 hwserver 之間的封包，會發現每一個 request/reply message 都有兩個 frames。

Extended Reply Envelope

Extended Request-Reply Pattern: 透過 ROUTER-DEALER proxy 延伸 REQ-REP pair。

proxy 的 pseudo-code

prepare context, frontend and backend sockets
while true:
    poll on both sockets
    if frontend had input:
        read all frames from frontend
        send to backend
    if backend had input:
        read all frames from backend
        send to frontend

ROUTER socket 不同於其他 sockets，會追蹤 connection，並通知 caller，方法是在每一個收到的訊息欠面加上 connection identity: address，address 只是一個 binary string，用意是：connection 唯一的 handle。再發送 message 到 ROUTER socket 時，會先發送 identity frame。

收到 ZMQ_ROUTER socket 的訊息時，要先 prepend a message part，描述訊息的來源 id，訊息在所有 connected peers 間是 fair-queued 接收。發送訊息到ZMQ_ROUTER socket 時，要先移除第一個message part，用來判斷要將訊息 route 到哪一個 node。

ØMQ v2.2 以前是使用 UUDIs 為 identities，3.0 以後，會產生 5 bytes identity ( 0 + random 32bits integer )。

如果有三個 REQ socket 連接到 ROUTER，會對每一個 REQ socket 都產生一個 random identity。

如果 REQ 的 identity 為 "ABC"，在 ROUTER socket 會收到以下 3 frames 的 message

proxy 核心功能是 "由一個 socket 讀資料，寫入另一個 socket"，由 DEALER socket 發送給 REP socket 的 message 為以下格式。REP socket 一次只能處理一個 request-reply 交換機制。

當 DEALER 讀取 3 frames，透過 ROUTER socket 發送三個 message。ROUTER 取得第一個 frame (ABC)，並尋找該 connection，找到時，就發送後面 2 個 frames。

REQ socket 取得訊息，檢查一個 frame 為 empty delimiter。然後會丟棄該 frmame，並發送 "World" 到 applicaiton 呼叫端。

優點

雖然 (extended) request-reply pattern 沒有在 server 因為 application bug 而發生問題的 recover 機制，但還是很好用。

• 每次 ROUTER 提供訊息時，都會告訴發送端的 identity
• 可使用 hash table (以identity 為 key) 追蹤是否有增加新的 peer
• 如果在第一個 frame 增加 identity，ROUTER 會以非同步方式轉發訊息到任一個 node。

ROUTER 不知道 envelope，也不知道 empty delimiter，只知道第一個是 identity frame。

recap

• REQ socket 發送 message 是同步的，在訊息前面會加上 empty delimiter frame，永遠是 send 以後，就等待 reply。REQ socket 一次只會跟一個 peer 溝通。
• REP socket 會讀並儲存所有 identities，包含 empty delimiter，並傳送給 caller。如果將 REP socket 連上多個 peers，由 peer 發送的 request 會以 fair 方式平均讀取。
• DEALER socket 容易忽略 reply envelope，DEALER socket 為非同步的，會以 fair-queue 方式由所有 connection 讀取訊息。
• ROUTER socket 容易忽略 reply envelope。他會產生 connection 的 identity，identity 以第一個 frame 方式傳送。ROUTER 是非同步的。

Request-Reply Combinations

有四種 request-reply sockets，各種組合有不同的用途。

以下是合法的組合:

• REQ to REP
• DEALER to REP
• REQ to ROUTER
• DEALER to ROUTER
• DEALER to DEALER
• ROUTER to ROUTER

以下是非法的組合

• REQ to REQ
• REQ to DEALER
• REP to REP
• REP to ROUTER

注意，因為 DEALER 是 asynchronous REQ socket，ROUTER 是 asynchronous REP socket，當使用 REQ socket，可以用 DEALER 替代 (需要自己讀寫 envelope)。也可以將 REP socket 換成 ROUTER (需要自己管理 identities)。

把 REQ, DEALER 視為 "client"，要用 connect，連接到 REQ, ROUTER，而 REP, ROUTER 視為 "server"，要用 bind。

REQ to REP

必須要由 REQ client 先發送訊息，然後由 REP server 發送回應。發生問題會有 EFSM error。

DEALER to REP

把 REQ 替換成 DEALER (asynchronous)，DEALER 可跟多個 REP server 溝通。

使用 DEALER 必須要正確 emulate envelope，否則 REP socket 會直接丟棄 message，發送時必須要：

• 發送 empty message frame with MORE flag
• 再發送 message body

接收時

• 接收第一個 frame，如果不是 empty，就丟棄 message
• 接收第二個 frame，傳送給 application

REQ to ROUTER

將 REP 換成 ROUTER (asynchronous server)，ROUTER 可跟多個 REQ clients 溝通。

可用兩種方法使用 ROUTER:

1. 在 fronend, backend socket 之間，當作 proxy ROUTER 會讀取所有 frames (包含 identity frame) 傳送出去
2. 作為 application，讀取並處理 message ROUTER 必須知道 reply envelope 的格式。ROUTER 會收到 identity frame, empty frame, data frame。

DEALER to ROUTER

asynchronous client 跟 asynchronous server 溝通

可以處理任意格式的 message format，必須要自己設計 protocol。可決定要不要模仿 REQ/REP reply envelope，也就是自己決定要不要發送 replies

DEALER to DEALER

非同步，可發送任意數量的 replies，要自行管理 reply envelopes。

這是特殊的 pattern，很少用到。

ROUTER to ROUTER

N-to-N connection，最難使用的組合，在 chap4 有 example

Invalid Combinations

以下是錯誤的原因

• REQ to REQ 兩邊都想先開始發送訊息

• REQ to DEALER DEALER 無法發送 reply 給原本的 peer

• REP to REP 兩邊都在等待第一個 request message

• REP to ROUTER 理論上可用 ROUTER initiate dialog，發送第一個 request，但要預先知道 REP socket 已經連線，且知道該 connection 的 identity

binds 那一端是 server, broker, publisher, collector connects 那端是 client, worker

Exploring ROUTER sockets

解釋如何識別connection，ROUTER socket 在無法發送訊息時會怎麼處理。

Identities and Addresses

identity 是 ØMQ 讓 ROUTER 識別 connection 的方法，用在 reply envelope 的 addresses。通常 identity 是亂數，在 ROUTER socket 裡面有個 hash table。另外 peer 會有實體的 address ( network endpoint ex:"tcp://192.168.55.115:5670" ) 或是 logical address (UUID or email address or other unique key)

application 可使用 ROUTER socket 跟特定 peers 溝通，會轉換 logical address 為 identity。因為 ROUTER socket 只會在 peer 發送訊息時，才會使用 identity of a connection，可以只回應該訊息，不是直接跟 peer 通訊。

可強制 ROUTER socket 改用 logical address 為 identity (利用 zmq_setsockopt)

• peer application 在 peer socket (DEALER or REQ) binding 或 connection 前，設定 ZMQ_IDENTITY option
• 通常 peer 會連接到已綁定的 ROUTER socket，但 ROUTER 會連接到 peer
• 在 connection time，peer socket 會告訴 router socket "請在此 connection 使用這個 identity"
• 如果 peer socket 沒有提供 identity，router 會自己產生 random identity
• ROUTER socket 提供 logical address 給 application 作為自該 peer 接收的訊息的 identity frame
• ROUTER 也會使用 logical address 為未來發送訊息的 identity frame

identity.py，產生兩個 REQ socket，一個用預設的 random identity，一個用自訂的 identity

# encoding: utf-8
#
#   Demonstrate identities as used by the request-reply pattern.  Run this
#   program by itself.

import zmq
import zhelpers

context = zmq.Context()

sink = context.socket(zmq.ROUTER)
sink.bind("inproc://example")

# First allow 0MQ to set the identity
anonymous = context.socket(zmq.REQ)
anonymous.connect("inproc://example")
anonymous.send(b"ROUTER uses a generated 5 byte identity")
zhelpers.dump(sink)

# Then set the identity ourselves
identified = context.socket(zmq.REQ)
identified.setsockopt(zmq.IDENTITY, b"PEER2")
identified.connect("inproc://example")
identified.send(b"ROUTER socket uses REQ's socket identity")
zhelpers.dump(sink)

執行結果

$ python identity.py
----------------------------------------
[005] 0x00800041a7
[000]
[039] ROUTER uses a generated 5 byte identity
----------------------------------------
[005] PEER2
[000]
[040] ROUTER socket uses REQ's socket identity

Router Error Handling

ROUTER socket 針對無法發送的訊息有暴力的解決方法：就是直接丟棄。雖然有效，但很難 debug。

ØMQ v3.2 以後，增加了 socket option 可以 catch error: ZMQ_ROUTER_MANDATORY，如果有設定，在 socket 發送 unroutable identity message 時，socket 會發出 EHOSTUNREACH error。

Load Balancing Pattern

瞭解如何連接 ROUTER 到 REQ 然後到 DEALER。load balancing pattern 就是利用 ROUTER socket 處理 routing。

load balancing pattern 解決了 (PUSH, DEALER) 提供的 simple round robin routing 的問題：因為 round robin 無法處理 worker 處理時間不同的問題，會變得沒有效率。

如果每一個郵局櫃檯人員有一條 queue，有些人買郵票 (simple transaction)，有些人開戶頭 (slow transaction)，round robin 處理時，會發現買郵票的人，會卡在 queue 裡面。

郵局的問題，解決方式是，建立一條單一 queue，讓幾個櫃檯處理 slow transaction，其他櫃檯以 first-come first-serve 方式服務客戶。

PUSH, DEALER 使用 simple round robin 的原因是迴避 sheer performance。如果在機場入境處，會發現大排長龍，不是用單一 queue，而是讓使用者預先往前走。因為入境檢查花費的時間大部分都差不多，造成 more or less fair 的結果。這也是 PUSH, DEALER 的策略，先將工單發送出去，最後移動的距離會比較短。

不同情境要用不同的策略。

回到 worker (DEALER or REQ) 連接到 broker (ROUTER) 的問題上，broker 需要知道哪一個 worker 有空，並維持一份 least recently used worker 列表。

解決方式為：worker 在啟動及完成一項工作時，發送 "ready" message。broker 會依順序讀取訊息，每次收到訊息，就是 last used worker 發送的。因為是用 ROUTER，可以取得 identity，就可以繼續往該 worker 發送工作訊息。

每個 task 會跟著 reply 一起發送，該 task 的 response 會以新的 request 發送。

ROUTER Broker and REQ Workers

使用 ROUTER broker 實作 load balancing pattern，跟一組 REQ workers 溝通。

rtreq.py

# encoding: utf-8
#
#   Custom routing Router to Mama (ROUTER to REQ)

import time
import random
from threading import Thread

import zmq

import zhelpers

NBR_WORKERS = 10


def worker_thread(context=None):
    context = context or zmq.Context.instance()
    worker = context.socket(zmq.REQ)

    # We use a string identity for ease here
    zhelpers.set_id(worker)
    worker.connect("tcp://localhost:5671")

    total = 0
    while True:
        # Tell the router we're ready for work
        worker.send(b"ready")

        # Get workload from router, until finished
        # 發送 b"ready" 後，自 router 持續接收工作的訊息
        workload = worker.recv()

        # 判斷是否收到 b"END" 的訊息，就是要結束工作
        finished = workload == b"END"
        if finished:
            print("Processed: %d tasks" % total)
            break
        total += 1

        # Do some random work
        time.sleep(0.1 * random.random())


context = zmq.Context.instance()
client = context.socket(zmq.ROUTER)
client.bind("tcp://*:5671")

# 產生 10 個 worker
for _ in range(NBR_WORKERS):
    Thread(target=worker_thread).start()

# 產生 100 個 task，接收到 worker 發送的 b"ready" 後(夾帶了 worker address)，就回送 task 訊息
for _ in range(NBR_WORKERS * 10):
    # LRU worker is next waiting in the queue
    address, empty, ready = client.recv_multipart()

    client.send_multipart([
        address,
        b'',
        b'This is the workload',
    ])

# Now ask mama to shut down and report their results
# 發送 b"END" 訊息給 worker 要結束工作
for _ in range(NBR_WORKERS):
    address, empty, ready = client.recv_multipart()
    client.send_multipart([
        address,
        b'',
        b'END',
    ])

執行結果

$ python rtreq.py
Processed: 8 tasks
Processed: 10 tasks
Processed: 10 tasks
Processed: 9 tasks
Processed: 9 tasks
Processed: 12 tasks
Processed: 8 tasks
Processed: 11 tasks
Processed: 10 tasks
Processed: 13 tasks

REQ 發送的訊息格式如下

ROUTER Broker and DEALER Workers

在能使用 REQ 的地方，都可以改用 DEALER，但有兩個差異

1. REQ socket 會在 data frame 前面，先送 empty delimiter frame，但是 DEALER 不會
2. REQ socket 在收到 reply 前只會發送一個 message，而 DEALER 是 asynchronous 的

synchronous/asynchronous 的差異看起來不明顯，差異比較大的在 failure recovering，在 chap4 討論。

rtdealer.py

# encoding: utf-8
#
#   Custom routing Router to Dealer

import time
import random
from threading import Thread

import zmq


# We have two workers, here we copy the code, normally these would
# run on different boxes...
# 兩個 worker 分別有不同的 identity
def worker_a(context=None):
    context = context or zmq.Context.instance()
    worker = context.socket(zmq.DEALER)
    worker.setsockopt(zmq.IDENTITY, b'A')
    worker.connect("ipc://routing.ipc")

    total = 0
    while True:
        # We receive one part, with the workload
        request = worker.recv()
        # 接收到 b"END" 代表要結束 worker
        finished = request == b"END"
        if finished:
            print("A received: %s" % total)
            break
        total += 1


def worker_b(context=None):
    context = context or zmq.Context.instance()
    worker = context.socket(zmq.DEALER)
    worker.setsockopt(zmq.IDENTITY, b'B')
    worker.connect("ipc://routing.ipc")

    total = 0
    while True:
        # We receive one part, with the workload
        request = worker.recv()
        finished = request == b"END"
        if finished:
            print("B received: %s" % total)
            break
        total += 1


context = zmq.Context.instance()
client = context.socket(zmq.ROUTER)
client.bind("ipc://routing.ipc")

Thread(target=worker_a).start()
Thread(target=worker_b).start()

# Wait for threads to stabilize
time.sleep(1)

# Send 10 tasks scattered to A twice as often as B
for _ in range(10):
    # Send two message parts, first the address...
    # 以 2:1 的比例，發送 task 給 worker A, B
    ident = random.choice([b'A', b'A', b'B'])
    # And then the workload
    work = b"This is the workload"
    client.send_multipart([ident, work])

client.send_multipart([b'A', b'END'])
client.send_multipart([b'B', b'END'])

執行結果

$ python rtdealer.py
B received: 3
A received: 7

程式跟上一個差不多，只是改成 DEALER。

A Load Balancing Message Broker

前面的例子中，沒有跟 clint 溝通的方法。將中間改為 proxy，就可以在 frontend backend 間轉發訊息

broker 有以下工作

• 接收來自多個 clients 的連線
• 接收來自多個 workers 的連線
• 接收 client 的 request，存放在單一 queue
• 以 load balancing pattern 發送 requests
• 接收來自 workers 的 replies
• 回送 replies 給 client

lbbroker.py

測試後發現，會有部分 client 訊息遺失，但將 ipc 改為 tcp 後，程式就是正常的。

"""
Load-balancing broker

Clients and workers are shown here in-process.
"""

from __future__ import print_function

import multiprocessing

import zmq


NBR_CLIENTS = 10
NBR_WORKERS = 3


def client_task(ident):
    """Basic request-reply client using REQ socket."""
    socket = zmq.Context().socket(zmq.REQ)
    socket.identity = u"Client-{}".format(ident).encode("ascii")
    # socket.connect("ipc://frontend.ipc")
    socket.connect("tcp://localhost:20000")

    # Send request, get reply
    socket.send(b"HELLO")
    reply = socket.recv()
    print("{}: {}".format(socket.identity.decode("ascii"),
                          reply.decode("ascii")))


def worker_task(ident):
    """Worker task, using a REQ socket to do load-balancing."""
    socket = zmq.Context().socket(zmq.REQ)
    socket.identity = u"Worker-{}".format(ident).encode("ascii")
    # socket.connect("ipc://backend.ipc")
    socket.connect("tcp://localhost:20001")

    # Tell broker we're ready for work
    socket.send(b"READY")

    while True:
        address, empty, request = socket.recv_multipart()
        print("{}: {}".format(socket.identity.decode("ascii"),
                              request.decode("ascii")))
        socket.send_multipart([address, b"", b"OK"])


def main():
    """Load balancer main loop."""
    # Prepare context and sockets
    context = zmq.Context.instance()
    frontend = context.socket(zmq.ROUTER)
    # frontend.bind("ipc://frontend.ipc")
    frontend.bind("tcp://*:20000")
    backend = context.socket(zmq.ROUTER)
    # backend.bind("ipc://backend.ipc")
    backend.bind("tcp://*:20001")

    # Start background tasks
    def start(task, *args):
        process = multiprocessing.Process(target=task, args=args)
        process.daemon = True
        process.start()
    for i in range(NBR_CLIENTS):
        start(client_task, i)
    for i in range(NBR_WORKERS):
        start(worker_task, i)

    # Initialize main loop state
    count = NBR_CLIENTS
    workers = []
    poller = zmq.Poller()
    # Only poll for requests from backend until workers are available
    poller.register(backend, zmq.POLLIN)

    while True:
        sockets = dict(poller.poll())

        if backend in sockets:
            # Handle worker activity on the backend
            request = backend.recv_multipart()
            worker, empty, client = request[:3]
            if not workers:
                # Poll for clients now that a worker is available
                poller.register(frontend, zmq.POLLIN)
            workers.append(worker)
            if client != b"READY" and len(request) > 3:
                # If client reply, send rest back to frontend
                empty, reply = request[3:]
                frontend.send_multipart([client, b"", reply])
                count -= 1
                if not count:
                    break

        if frontend in sockets:
            # Get next client request, route to last-used worker
            client, empty, request = frontend.recv_multipart()
            worker = workers.pop(0)
            backend.send_multipart([worker, b"", client, b"", request])
            if not workers:
                # Don't poll clients if no workers are available
                poller.unregister(frontend)

    # Clean up
    backend.close()
    frontend.close()
    context.term()


if __name__ == "__main__":
    main()

程式要注意的部分

1. 每個 socket 讀寫的訊息 envelope
2. load balancing algorithm

Client 會發送 "Hello" 到 broker

REQ socket 會增加一個 empty delimiter frame ROUTER socket 會增加 conneciton identity，因此 broker 的 frontend 接收訊息後，會收到 client address, empty delimiter frame, data frame

broker 會再增加 empty delimiter，然後再加上 address of worker，透過 backend 發送出去

worker 發送的訊息

worker 需要儲存 envelope，包含 empty delimiter 的部分。REP socket 會自動處理，但使用 ROUTER 必須自己處理 message frames。

回傳訊息的部分，backend 會給 broker 5 frames 的訊息，broker 發送 3 frames 訊息給 client，而 client 只會取得結果的訊息

---

load balancing 的部分

1. 建立 pollset，固定會 pools backend，但只在有 workers 的時候，會 polls frontend
2. poll for activity 有無限的 timeout 時間
3. 如果 backend 有活動，不是取得 "ready" 訊息，就是取得處理結果 (要回傳給 client)。兩種狀況下都儲存 worker addres 到一個 worker list queue
4. 如果 frontend 有活動，取得 client request -> 由 list 取出一個 worker，並發送 request 到 backend。也就是 worker address, empty delimiter, 及 client request 的三個 frames

A High-Level API for ZeroMQ

因為需要撰寫更複雜的範例，所以要先更深入了解高階的 API

這是 load balancing broker 的 worker thread

while (true) {
    //  Get one address frame and empty delimiter
    char *address = s_recv (worker);
    char *empty = s_recv (worker);
    assert (*empty == 0);
    free (empty);

    //  Get request, send reply
    char *request = s_recv (worker);
    printf ("Worker: %s\n", request);
    free (request);

    s_sendmore (worker, address);
    s_sendmore (worker, "");
    s_send     (worker, "OK");
    free (address);
}

程式不夠彈性，因為只能處理一個 replay address 的 envelope。如果直接用 libzmq API 需要改寫為

while (true) {
    //  Get one address frame and empty delimiter
    char address [255];
    int address_size = zmq_recv (worker, address, 255, 0);
    if (address_size == -1)
        break;

    char empty [1];
    int empty_size = zmq_recv (worker, empty, 1, 0);
    zmq_recv (worker, &empty, 0);
    assert (empty_size <= 0);
    if (empty_size == -1)
        break;

    //  Get request, send reply
    char request [256];
    int request_size = zmq_recv (worker, request, 255, 0);
    if (request_size == -1)
        return NULL;
    request [request_size] = 0;
    printf ("Worker: %s\n", request);
    
    zmq_send (worker, address, address_size, ZMQ_SNDMORE);
    zmq_send (worker, empty, 0, ZMQ_SNDMORE);
    zmq_send (worker, "OK", 2, 0);
}

高階 API 的功能

• string helper (s_send, s_recv)
• frame: message frame
• message: list of frames

剛剛的 worker thread 可改寫簡化為

while (true) {
    zmsg_t *msg = zmsg_recv (worker);
    zframe_reset (zmsg_last (msg), "OK", 2);
    zmsg_send (&msg, worker);
}

• automatic handling of sockets: 要手動關閉 socket 很麻煩，某些狀況需要指定 linger timeout。要有個好方法，可以在關閉 context 時，自動關閉 sockets
• portable thread management: 通常 ZeroMQ application 會使用 threads，但是 POSIX threads 並不 portable，因此 API 應該要封裝這個處理方法

• piping from parent to child threads: 這是 recurrent problem，如何在 parent 及 child threds 之間發送 signal，API 要提供 ZeroMQ pipe (broker 的 frontend 接收訊息後，自動使用 PAIR socket 及 inproc)

• portable clocks: 取得 ms 時間，或是 sleep ms 時間都不 portable，application 需要 API 提供正確的 clock 時間資訊

• a reactor to replace zmq_poll()。pool loop 雖簡單，但不夠好。程式常會重複: calculating itmers, 在 socket ready 時 calling code。需要簡單餓 reactor，且有 socket reader 及 timer 功能。

• proper handling of Ctrl-C

CZMQ High-Level API

C 語言，可改使用 CZMQ: ZeroMQ binding for C

lbbroker2.py

跟上一版的 lbbroker.py 的差異為

• 不會一直呼叫 poller.register(frontend, zmq.POLLIN) 或是 poller.unregister(frontend)，而改用一個 available_workers 變數，判斷是否要 pool frontend 的 message
• inproc 可以運作
• 增加 assert，檢查 empty delimiter
• 在 worker 檢查 zmq.ContextTerminated 的錯誤訊號

"""
   Least-recently used (LRU) queue device
   Clients and workers are shown here in-process
"""

from __future__ import print_function
import threading
import time
import zmq

NBR_CLIENTS = 10
NBR_WORKERS = 3


def worker_thread(worker_url, context, i):
    """ Worker using REQ socket to do LRU routing """

    socket = context.socket(zmq.REQ)

    # set worker identity
    socket.identity = (u"Worker-%d" % (i)).encode('ascii')

    socket.connect(worker_url)

    # Tell the broker we are ready for work
    socket.send(b"READY")

    try:
        while True:

            address, empty, request = socket.recv_multipart()

            print("%s: %s\n" % (socket.identity.decode('ascii'),
                                request.decode('ascii')), end='')

            socket.send_multipart([address, b'', b'OK'])

    except zmq.ContextTerminated:
        # context terminated so quit silently
        return


def client_thread(client_url, context, i):
    """ Basic request-reply client using REQ socket """

    socket = context.socket(zmq.REQ)

    # Set client identity. Makes tracing easier
    socket.identity = (u"Client-%d" % (i)).encode('ascii')

    socket.connect(client_url)

    #  Send request, get reply
    socket.send(b"HELLO")
    reply = socket.recv()

    print("%s: %s\n" % (socket.identity.decode('ascii'),
                        reply.decode('ascii')), end='')


def main():
    """ main method """

    url_worker = "inproc://workers"
    url_client = "inproc://clients"
    client_nbr = NBR_CLIENTS

    # Prepare our context and sockets
    context = zmq.Context()
    frontend = context.socket(zmq.ROUTER)
    frontend.bind(url_client)
    backend = context.socket(zmq.ROUTER)
    backend.bind(url_worker)

    # create workers and clients threads
    for i in range(NBR_WORKERS):
        thread = threading.Thread(target=worker_thread,
                                  args=(url_worker, context, i, ))
        thread.start()

    for i in range(NBR_CLIENTS):
        thread_c = threading.Thread(target=client_thread,
                                    args=(url_client, context, i, ))
        thread_c.start()

    # Logic of LRU loop
    # - Poll backend always, frontend only if 1+ worker ready
    # - If worker replies, queue worker as ready and forward reply
    # to client if necessary
    # - If client requests, pop next worker and send request to it

    # Queue of available workers
    available_workers = 0
    workers_list = []

    # init poller
    poller = zmq.Poller()

    # Always poll for worker activity on backend
    poller.register(backend, zmq.POLLIN)

    # Poll front-end only if we have available workers
    poller.register(frontend, zmq.POLLIN)

    while True:

        socks = dict(poller.poll())

        # Handle worker activity on backend
        if (backend in socks and socks[backend] == zmq.POLLIN):

            # Queue worker address for LRU routing
            message = backend.recv_multipart()
            assert available_workers < NBR_WORKERS

            worker_addr = message[0]

            # add worker back to the list of workers
            available_workers += 1
            workers_list.append(worker_addr)

            #   Second frame is empty
            empty = message[1]
            assert empty == b""

            # Third frame is READY or else a client reply address
            client_addr = message[2]

            # If client reply, send rest back to frontend
            if client_addr != b'READY':

                # Following frame is empty
                empty = message[3]
                assert empty == b""

                reply = message[4]

                frontend.send_multipart([client_addr, b"", reply])

                client_nbr -= 1

                if client_nbr == 0:
                    break  # Exit after N messages

        # poll on frontend only if workers are available
        if available_workers > 0:

            if (frontend in socks and socks[frontend] == zmq.POLLIN):
                # Now get next client request, route to LRU worker
                # Client request is [address][empty][request]

                [client_addr, empty, request] = frontend.recv_multipart()

                assert empty == b""

                #  Dequeue and drop the next worker address
                available_workers += -1
                worker_id = workers_list.pop()

                backend.send_multipart([worker_id, b"",
                                        client_addr, b"", request])

    #out of infinite loop: do some housekeeping
    time.sleep(1)

    frontend.close()
    backend.close()
    context.term()


if __name__ == "__main__":
    main()

---

這是最好的版本的 lbbroker，使用 IOLoop

lbbroker3.py

"""

   Least-recently used (LRU) queue device
   Demonstrates use of pyzmq IOLoop reactor

   While this example runs in a single process, that is just to make
   it easier to start and stop the example. Each thread has its own
   context and conceptually acts as a separate process.
"""

from __future__ import print_function
import threading
import time
import zmq

from zmq.eventloop.ioloop import IOLoop
from zmq.eventloop.zmqstream import ZMQStream

NBR_CLIENTS = 10
NBR_WORKERS = 3


def worker_thread(worker_url, i):
    """ Worker using REQ socket to do LRU routing """
    context = zmq.Context.instance()

    socket = context.socket(zmq.REQ)

    # set worker identity
    socket.identity = (u"Worker-%d" % (i)).encode('ascii')

    socket.connect(worker_url)

    # Tell the broker we are ready for work
    socket.send(b"READY")

    try:
        while True:

            address, empty, request = socket.recv_multipart()

            print("%s: %s\n" % (socket.identity.decode('ascii'),
                                request.decode('ascii')), end='')

            socket.send_multipart([address, b'', b'OK'])

    except zmq.ContextTerminated:
        # context terminated so quit silently
        return


def client_thread(client_url, i):
    """ Basic request-reply client using REQ socket """
    context = zmq.Context.instance()

    socket = context.socket(zmq.REQ)

    # Set client identity. Makes tracing easier
    socket.identity = (u"Client-%d" % (i)).encode('ascii')

    socket.connect(client_url)

    #  Send request, get reply
    socket.send(b"HELLO")
    reply = socket.recv()

    print("%s: %s\n" % (socket.identity.decode('ascii'),
                        reply.decode('ascii')), end='')


class LRUQueue(object):
    """LRUQueue class using ZMQStream/IOLoop for event dispatching"""

    def __init__(self, backend_socket, frontend_socket):
        self.available_workers = 0
        self.workers = []
        self.client_nbr = NBR_CLIENTS

        self.backend = ZMQStream(backend_socket)
        self.frontend = ZMQStream(frontend_socket)
        self.backend.on_recv(self.handle_backend)

        self.loop = IOLoop.instance()

    def handle_backend(self, msg):
        # 處理 backend 接收 worker 發送的訊息
        # address, empty, request
        # Queue worker address for LRU routing
        worker_addr, empty, client_addr = msg[:3]

        assert self.available_workers < NBR_WORKERS

        # add worker back to the list of workers
        # 收到訊息後，增加 workers
        self.available_workers += 1
        self.workers.append(worker_addr)

        #   Second frame is empty
        assert empty == b""

        # Third frame is READY or else a client reply address
        # If client reply, send rest back to frontend
        # 因為 worker 啟動時，只會發送 b"READY"，其他狀況都是 address, empty, request
        if client_addr != b"READY":
            empty, reply = msg[3:]

            # Following frame is empty
            assert empty == b""

            self.frontend.send_multipart([client_addr, b'', reply])

            # 以 client request 數量，判斷是否要停止 IOLoop
            # 0 表示已經接收到 10 個 worker 傳來的，已經轉發給 client 的處理結果
            self.client_nbr -= 1
            if self.client_nbr == 0:
                # Exit after N messages
                self.loop.add_timeout(time.time() + 1, self.loop.stop)

        if self.available_workers == 1:
            # on first recv, start accepting frontend messages
            # 啟動接收 frontend 訊息
            self.frontend.on_recv(self.handle_frontend)

    def handle_frontend(self, msg):
        # Now get next client request, route to LRU worker
        # Client request is [address][empty][request]
        client_addr, empty, request = msg

        assert empty == b""

        #  Dequeue and drop the next worker address
        self.available_workers -= 1
        worker_id = self.workers.pop()

        self.backend.send_multipart([worker_id, b'', client_addr, b'', request])
        if self.available_workers == 0:
            # stop receiving until workers become available again
            # 如果沒有 workers 就暫停接收 frontend 訊息
            self.frontend.stop_on_recv()


def main():
    """main method"""

    url_worker = "ipc://backend.ipc"
    url_client = "ipc://frontend.ipc"

    # Prepare our context and sockets
    context = zmq.Context()
    frontend = context.socket(zmq.ROUTER)
    frontend.bind(url_client)
    backend = context.socket(zmq.ROUTER)
    backend.bind(url_worker)

    # create workers and clients threads
    for i in range(NBR_WORKERS):
        thread = threading.Thread(target=worker_thread, args=(url_worker, i, ))
        thread.daemon = True
        thread.start()

    for i in range(NBR_CLIENTS):
        thread_c = threading.Thread(target=client_thread,
                                    args=(url_client, i, ))
        thread_c.daemon = True
        thread_c.start()

    # create queue with the sockets
    queue = LRUQueue(backend, frontend)

    # start reactor
    IOLoop.instance().start()

if __name__ == "__main__":
    main()

執行時，會出現 IOLoop warning

lbbroker3.py:21: VisibleDeprecationWarning: zmq.eventloop.minitornado is deprecated in pyzmq 14.0 and will be removed.
    Install tornado itself to use zmq with the tornado IOLoop.

  from zmq.eventloop.ioloop import IOLoop

這邊注意 pyzmq 14.0 以後，需要使用者改直接安裝 tornado，使用 tornado IOLoop，在不修改程式的狀況下，直接安裝 tornado 後，就不會再出現 warning

sudo pip3 install tornado

Asynchronous Client/Server Pattern

在 ROUTER/DEALER 範例中，已經看過 1-N 的使用情境，一個 server 可非同步跟多個 worker 溝通，現在要改為 N-to-1 的架構，多個 clients 跟 一個 server 以非同步方式溝通。

• clients 連到 server，發送 requests
• 針對每個 request，server 會送 0 到多個 replies
• clients 不需要等待 replies，可發送多個 requests
• server 不需要等待 requests，可發送多個 replies

asyncsrv.py

import zmq
import sys
import threading
import time
from random import randint, random

__author__ = "Felipe Cruz <felipecruz@loogica.net>"
__license__ = "MIT/X11"

def tprint(msg):
    """like print, but won't get newlines confused with multiple threads"""
    sys.stdout.write(msg + '\n')
    sys.stdout.flush()

class ClientTask(threading.Thread):
    """ClientTask"""
    def __init__(self, id):
        self.id = id
        threading.Thread.__init__ (self)

    def run(self):
        context = zmq.Context()
        socket = context.socket(zmq.DEALER)
        identity = u'worker-%d' % self.id
        socket.identity = identity.encode('ascii')
        socket.connect('tcp://localhost:5570')
        print('Client %s started' % (identity))
        poll = zmq.Poller()
        poll.register(socket, zmq.POLLIN)
        reqs = 0
        while True:
            # 發送 request
            reqs = reqs + 1
            print('Req #%d sent..' % (reqs))
            socket.send_string(u'request #%d' % (reqs))

            # 取出 replies 執行 5 次
            for i in range(5):
                sockets = dict(poll.poll(1000))
                if socket in sockets:
                    msg = socket.recv()
                    tprint('Client %s received: %s' % (identity, msg))

        socket.close()
        context.term()

class ServerTask(threading.Thread):
    """ServerTask"""
    def __init__(self):
        threading.Thread.__init__ (self)

    def run(self):
        context = zmq.Context()
        frontend = context.socket(zmq.ROUTER)
        frontend.bind('tcp://*:5570')

        backend = context.socket(zmq.DEALER)
        backend.bind('inproc://backend')

        # 啟動 5 個 server workers
        workers = []
        for i in range(5):
            worker = ServerWorker(context)
            worker.start()
            workers.append(worker)

        # 以 proxy 轉接 frontend, backend
        zmq.proxy(frontend, backend)

        frontend.close()
        backend.close()
        context.term()

class ServerWorker(threading.Thread):
    """ServerWorker"""
    def __init__(self, context):
        threading.Thread.__init__ (self)
        self.context = context

    def run(self):
        worker = self.context.socket(zmq.DEALER)
        worker.connect('inproc://backend')
        tprint('Worker started')
        while True:
            # 收到 task，處理後，發送 reply
            ident, msg = worker.recv_multipart()
            tprint('Worker received %s from %s' % (msg, ident))

            # 發送 0~5 個replies
            replies = randint(0,4)
            for i in range(replies):
                time.sleep(1. / (randint(1,10)))
                worker.send_multipart([ident, msg])

        worker.close()


def main():
    """main function"""
    server = ServerTask()
    server.start()
    for i in range(3):
        client = ClientTask(i)
        client.start()

    server.join()


if __name__ == "__main__":
    main()

以 multithread 方式模擬多個 process 的狀況。執行後會有 3 個 clients。

• clients 每秒發送一個 request，取得 0~多個 replies。為了讓 zmq_poll() 運作，不能 pool with 1 sec timeout，或是改成收到 reply 後再發送新的 request，因此程式是以 100 times at 1/100th of a second per pool 方式，高速進行 poll，讓程式接近正確
• server 會使用 pool of worker threads，每一個 server 一次處理一個 request，透過 internal queue 連接到 frontend socket，透過 zmq_proxy() 連接 frontend, backend sockets

clinet-server 是使用 DEALER-ROUTER dialog，server 內部的 server-worker 是使用 DEALER-DEALER，如果 worker 需要同步機制，就改用 REP。因為要發送多個 replies，所以需要使用非同部的 DELAER。

client 發送單一 frame 的 request，server 收到 2 frames(前面增加 client address)，server 發送 2 frames 給 worker，收到 2 frames 結果，再用第一個 frame 將結果傳回原本的 client。

     client          server       frontend       worker
   [ DEALER ]<---->[ ROUTER <----> DEALER <----> DEALER ]
             1 part         2 parts       2 parts

在 server-worker 之間，可以改用 ROUTER-DEALER 增加 load balancing 的機制。

如果要實作有保存 client 狀態的 server，會遇到過一個問題，如果 server 儲存了每一個 client 的狀態，因為 client 來來去去，最終會耗盡資源。如果使用 default identity，也會讓每一個 connection 都像是新的。

我們可以實作 server，並只在一小段時間內(足夠讓 worker 處理 request)保存狀態，然後就丟棄狀態，但這樣的機制並不實用。為了實作 stateful asychronous server，儲存 client state，必須：

1. 在 client-server 間處理 heartbeating，例如每秒發送一個 request
2. 使用 client identity 為 key，用來儲存 state
3. 偵測 heartbeat 停止機制，如果 client 沒有發送新的 request (ex: 2 seconds)，server 可偵測到，並在 2s 後刪除 state

Example: Inter-Broker Routing

實作一個 real application

實作細節

1. worker 運作在多種 hardware 上，每個 cluster 有上百個 workers，全部有 12 個 cluster
2. clients 會產生 task 給 workers，每一個 task 是獨立的工作單位，所有 client 都需要盡快找到一個 worker 並分配 task。有很多 clients，且 client 可隨時連線或斷線。
3. 在任意時間點，可以增加或減少 clusters，這部分很難做。cluster 要能夠即時 joing/leave cloud，並帶來 workers 與 clients
4. 如果在某個 cluster 沒有 workers，client 的 task 會去 cloud 尋找其他 workers
5. clients 一次發送一個 task，如果在 X seconds 內沒有收到結果，會再重新發送一次，這部分由 client 自行決定要怎麼做。

6. workers 一次處理一個 task，如果 crash 就直接用 script 重新啟動。

• 這是 clusters 之間的 super-duper network interconnect 機制
• 每個 cluster 有上千個 clients，每一個 client 每秒有 10 requests，request 跟 reply 都很小，不會超過 1K bytes

簡單計算一下 2500 clients * 10/second * 1000 bytes * 2 directions = 50 MB/sec or 400 Mb/sec，這對 Gb ethernet 來說沒有問題，可以運作。

單一 cluster 的架構

workers, clients 是 synchronous，要用 load balancing pattern 將 task 轉發到 workers。所有 workers 都是一樣的，不需要識別不同 services 的機制，workers are anonymous。不提供保證送達的機制，發生問題就 retry。

client 及 worker 不會直接連線，因為這樣做會很難動態新增或移除 nodes，因此架構上是使用先前看過的 request-reply load balacing arachitecture。

提升到 multiple clusters

要增加 cluster，每一個 cluster 處理一組 clients 及 workers

討論如何讓不同 cluster 之間的 clients 及 workers 都可以互相溝通，有以下幾種方式，各有優缺點

1. client 直接連到所有 brokers，優點是不需要修改 brokers/workers，但 client 會變得比較複雜，要知道 overall topology，如果新增 3rd or 4th cluster，會影響到所有 clients，也就是要在 clients 實作 routing 及 failover logic

2. workers 直接連到所有 brokers，但REQ worker 不能這樣做，他們只能連接到單一個 broker，我們可使用 REPs，但 REP 不能客製 broker-to-worker routing (load balancing)，只能使用內建的 load balancing。這樣會讓我們無法將 task 分配給空閒地 workers，有一個解決方案是在 worker node 改用 ROUTER，這個標記為 "Idea #1"

   

   看起來可以運作，但還是有問題，因為 clients 應該盡可能先使用 local workers，另外 workers 需要發送 "ready" signal 給兩個 brokers，可能會一次取得兩個 tasks。

3. brokers 互相連線，這種方式最簡單，可產生最短路徑。但無法動態新增 clusters。clients 及 workers 一樣不需要知道 network topology，brokers 之間會分享空閒 workers 的資訊，這個標記為 "Idea #2"

   

   這個方案比較好，brokers 之間互相連線交換資訊。brokers 之間需要複雜一點的 routing algorithm

4. 把同一個 cluster 的 clients, workers 視為預設的選項，另外增加例外狀況，在 clusters 之間交換 tasks

5. 可針對不同類型的 task 採用不同的 message flows，例如使用不同類型的 network connection

6. 要改成 3 個 brokers 也不會太難，如果有問題，也可以增加一個 super-broker

以下就開發一個範例，先將整個範例放到 1 process 裡面，ZeroMQ 可以由 micro-level scale 很快擴增到 macro-level。將 thread 改成 processes，每一個 cluster process 都包含 clients threads, worker threads, a broker thread

• REQ client threads 產生 workloads，傳送給 broker (ROUTER)
• REQ worker threads 接收 workloads，回傳結果給 broker(ROUTER)
• broker 使用 load balancing pattern，進行 queues 並分配 workloads

Federation Versus Peering

brokers 之間需要互相通知 "we have capacity" 的資訊，並接收更多 tasks。也需要互相通知 "stop, we're full" 的資訊。這些資訊不需要非常即時且正確，有時候多收一些無法立刻處理的 task 也沒關係。

最簡單的方式是 "federation"，brokers 之間互相模擬成 clients, workers。也就是把 frontend 街道另一個 broker 的 backend。

這個方案可解決問題，但會讓 broker 一次只能接受一個來自另一個 broker 的 task。

federation model 可處理 service-oriented architecture (SOAs)，就是用 service name 做 routing 而不是 load balancing or round robin。

另一種方案是 "peering"，brokers 以特殊的 channel 互相溝通，N brokers 就是有 N-1 peers

• broker 要告訴 peers，目前有幾個 workers，可以只簡單通知數量就好了，正確的方式是要用 pub-sub socket pattern。每個 broker 使用 PUB 發布 state information，並使用 SUB socket 接收來自其他 peers 的 state information
• broker 需要非同步發送 task 給其他 peers 並取得 replies 的方法。可使用 ROUTER sockets。broker 需要兩個 sockets，一個用來接收 replies，一個用來發送 tasks，用量個 sockets 可簡化 frames 內容的設計，不需要判斷是 request 或是 reply

Naming Ceremony

broker 需要用到 3 flows * 2 sockets/flow = 6 sockets，需要對 socket 有個好的命名機制。

3 flows 分別為

1. local request-reply flow: 用在 broker 跟 clients, workers 之間
2. cloud request-reply flow: 用在 broker 跟 peer brokers 之間
3. state flow: 用在 broker 跟 peer broker 之間

已有意義且使用相同的長度的的名稱，會讓程式運作更好，我們用以下名稱

1. localfe, localbe for local flow
2. cloudfw, cloudbe for cloud flow
3. statefe, statebe for state flow

範例先使用 ipc 模擬，不使用 inproc，ipc 跟 tcp 是 disconnected transport。ipc endpoint 名稱訂為 something-local, something-cloud, something-state

Prototyping the State Flow

peering1.py

#
#   Broker peering simulation (part 1) in Python
#   Prototypes the state flow
#
#   Author : Piero Cornice
#   Contact: root(at)pieroland(dot)net
#
import sys
import time
import random

import zmq


def main(myself, others):
    print("Hello, I am %s" % myself)

    context = zmq.Context()

    # State Back-End
    statebe = context.socket(zmq.PUB)

    # State Front-End
    statefe = context.socket(zmq.SUB)
    statefe.setsockopt(zmq.SUBSCRIBE, b'')

    bind_address = u"ipc://%s-state.ipc" % myself
    statebe.bind(bind_address)

    for other in others:
        statefe.connect(u"ipc://%s-state.ipc" % other)
        time.sleep(1.0)

    poller = zmq.Poller()
    poller.register(statefe, zmq.POLLIN)

    while True:

########## Solution with poll() ##########
        socks = dict(poller.poll(1000))

        # Handle incoming status message
        if socks.get(statefe) == zmq.POLLIN:
            msg = statefe.recv_multipart()
            print('%s Received: %s' % (myself, msg))

        else:
            # Send our address and a random value
            # for worker availability
            msg = [bind_address, (u'%i' % random.randrange(1, 10))]
            msg = [ m.encode('ascii') for m in msg]
            statebe.send_multipart(msg)
##################################

######### Solution with select() #########
#        pollin, pollout, pollerr = zmq.select([statefe], [], [], 1)
#
#        if pollin and pollin[0] == statefe:
#            # Handle incoming status message
#            msg = statefe.recv_multipart()
#            print 'Received:', msg
#
#        else:
#            # Send our address and a random value
#            # for worker availability
#            msg = [bind_address, str(random.randrange(1, 10))]
#            statebe.send_multipart(msg)
##################################


if __name__ == '__main__':
    if len(sys.argv) >= 2:
        main(myself=sys.argv[1], others=sys.argv[2:])
    else:
        print("Usage: peering.py <myself> <peer_1> ... <peer_N>")
        sys.exit(1)

執行結果

$ python peering1.py p1 p2
Hello, I am p1
p1 Received: [b'ipc://p2-state.ipc', b'5']
p1 Received: [b'ipc://p2-state.ipc', b'6']
p1 Received: [b'ipc://p2-state.ipc', b'7']

$ python peering1.py p2 p1
Hello, I am p2
p2 Received: [b'ipc://p1-state.ipc', b'7']
p2 Received: [b'ipc://p1-state.ipc', b'7']
p2 Received: [b'ipc://p1-state.ipc', b'4']
p2 Received: [b'ipc://p1-state.ipc', b'5']

• broker 使用 identity 產生 ipc endpoint names，實際上正式環境的 broker 需要用 TCP transport 才對。
• 使用 zmq_poll() loop 作為程式的核心，只會在沒有收到任何 incoming messages 時，會等 1s 然後發送 state message，如果在每次收到訊息後，就發送一個 state message，會造成 message storms
• 使用 2-parts pub-sub message 包含 sender address 及 data。需要知道 publisher 的 address，因為要發送 task，唯一的方式，是放在訊息的一部分。
• 不需要設定 subscribers 的 identities，因為不需要取得已經在運作中的 brokers 的 outdated state information。
• 不需要設定 publisher 的 HWM

實際上並不需要定時發送 state messages，只要在 state change (a worker become available or unavailable) 時發送即可。

如果要定時發送 state message，可建立 child thread 並在該 thread 使用 statebe socket，可在 state change 時，發送給 child thread，再由該 thread，定時發送 state message。

Prototyping the Local and Cloud Flows

由 client 取得 requests，並轉發給 local workers，或是 cloud peers

需要有兩條 queues，一個是 local clients 的 requests，一個是 cloud clients的 requests。因為 ZeroMQ 已經有內建 queue，所以要使用 ZeroMQ 的 socket buffers。

這是在 load balancing broker 使用的技術，由 2 個 frontends 讀入 requests，由 backend 讀入，取得 replies 使用的 route back address，如果沒有 backend workers，也不需要讀取 frontends。

main loop 為

• pool backends for activity 如取得 message，可能是 worker 傳送的 "ready"，或是 reply，如果是 reply 就 route back 回 local 或 cloud frontend
• 如果 worker 發送 reply，就表示該worker 可接收新的 task
• 如果還有 available workers，由 local frontend 取出一個 request，轉發給 local worker，或是 randomly 轉發給 cloud peer (目前暫時先用 random 方式傳給 cloud peer)

在 brokers 之間使用 broker identity 去 route messages，每個 broker 都有自己的名字

peering2.py

#
#   Broker peering simulation (part 2) in Python
#   Prototypes the request-reply flow
#
#   While this example runs in a single process, that is just to make
#   it easier to start and stop the example. Each thread has its own
#   context and conceptually acts as a separate process.
#
#   Author : Min RK
#   Contact: benjaminrk(at)gmail(dot)com
#
import random
import sys
import threading
import time

import zmq

try:
    raw_input
except NameError:
    # Python 3
    raw_input = input

NBR_CLIENTS = 10
NBR_WORKERS = 3

def tprint(msg):
    sys.stdout.write(msg + '\n')
    sys.stdout.flush()

def client_task(name, i):
    """Request-reply client using REQ socket"""
    ctx = zmq.Context()
    client = ctx.socket(zmq.REQ)
    client.identity = (u"Client-%s-%s" % (name, i)).encode('ascii')
    client.connect("ipc://%s-localfe.ipc" % name)
    while True:
        client.send(b"HELLO")
        try:
            reply = client.recv()
        except zmq.ZMQError:
            # interrupted
            return
        tprint("Client-%s: %s" % (i, reply))
        time.sleep(1)


def worker_task(name, i):
    """Worker using REQ socket to do LRU routing"""
    ctx = zmq.Context()
    worker = ctx.socket(zmq.REQ)
    worker.identity = (u"Worker-%s-%s" % (name, i)).encode('ascii')
    worker.connect("ipc://%s-localbe.ipc" % name)

    # Tell broker we're ready for work
    worker.send(b"READY")

    # Process messages as they arrive
    while True:
        try:
            msg = worker.recv_multipart()
        except zmq.ZMQError:
            # interrupted
            return
        tprint("Worker-%s: %s\n" % (i, msg))
        msg[-1] = b"OK"
        worker.send_multipart(msg)

def main(myself, peers):
    print("I: preparing broker at %s..." % myself)

    # Prepare our context and sockets
    ctx = zmq.Context()

    # Bind cloud frontend to endpoint
    cloudfe = ctx.socket(zmq.ROUTER)
    if not isinstance(myself, bytes):
        ident = myself.encode('ascii')
    else:
        ident = myself
    cloudfe.identity = ident
    cloudfe.bind("ipc://%s-cloud.ipc" % myself)

    # Connect cloud backend to all peers
    cloudbe = ctx.socket(zmq.ROUTER)
    cloudbe.identity = ident
    for peer in peers:
        tprint("I: connecting to cloud frontend at %s" % peer)
        cloudbe.connect("ipc://%s-cloud.ipc" % peer)


    if not isinstance(peers[0], bytes):
        peers = [peer.encode('ascii') for peer in peers]

    # Prepare local frontend and backend
    localfe = ctx.socket(zmq.ROUTER)
    localfe.bind("ipc://%s-localfe.ipc" % myself)
    localbe = ctx.socket(zmq.ROUTER)
    localbe.bind("ipc://%s-localbe.ipc" % myself)

    # Get user to tell us when we can start...
    raw_input("Press Enter when all brokers are started: ")

    # create workers and clients threads
    for i in range(NBR_WORKERS):
        thread = threading.Thread(target=worker_task, args=(myself, i))
        thread.daemon = True
        thread.start()

    for i in range(NBR_CLIENTS):
        thread_c = threading.Thread(target=client_task, args=(myself, i))
        thread_c.daemon = True
        thread_c.start()

    # Interesting part
    # -------------------------------------------------------------
    # Request-reply flow
    # - Poll backends and process local/cloud replies
    # - While worker available, route localfe to local or cloud

    workers = []

    # setup pollers
    pollerbe = zmq.Poller()
    pollerbe.register(localbe, zmq.POLLIN)
    pollerbe.register(cloudbe, zmq.POLLIN)

    pollerfe = zmq.Poller()
    pollerfe.register(localfe, zmq.POLLIN)
    pollerfe.register(cloudfe, zmq.POLLIN)

    while True:
        # If we have no workers anyhow, wait indefinitely
        try:
            events = dict(pollerbe.poll(1000 if workers else None))
        except zmq.ZMQError:
            break  # interrupted

        # Handle reply from local worker
        msg = None
        if localbe in events:
            msg = localbe.recv_multipart()
            (address, empty), msg = msg[:2], msg[2:]
            workers.append(address)

            # If it's READY, don't route the message any further
            if msg[-1] == b'READY':
                msg = None
        elif cloudbe in events:
            msg = cloudbe.recv_multipart()
            (address, empty), msg = msg[:2], msg[2:]

            # We don't use peer broker address for anything

        if msg is not None:
            address = msg[0]
            if address in peers:
                # Route reply to cloud if it's addressed to a broker
                cloudfe.send_multipart(msg)
            else:
                # Route reply to client if we still need to
                localfe.send_multipart(msg)

        # Now route as many clients requests as we can handle
        while workers:
            events = dict(pollerfe.poll(0))
            reroutable = False
            # We'll do peer brokers first, to prevent starvation
            # 先處理 cloudfe, cloudfe 的 msg 不能再繞回原本的 peer
            if cloudfe in events:
                msg = cloudfe.recv_multipart()
                reroutable = False
            elif localfe in events:
                msg = localfe.recv_multipart()
                reroutable = True
            else:
                break  # No work, go back to backends

            # If reroutable, send to cloud 20% of the time
            # Here we'd normally use cloud status information
            # 以 20% 比例送給 cloudbe
            if reroutable and peers and random.randint(0, 4) == 0:
                # Route to random broker peer
                msg = [random.choice(peers), b''] + msg
                cloudbe.send_multipart(msg)
            else:
                msg = [workers.pop(0), b''] + msg
                localbe.send_multipart(msg)

if __name__ == '__main__':
    if len(sys.argv) >= 2:
        main(myself=sys.argv[1], peers=sys.argv[2:])
    else:
        print("Usage: peering2.py <me> [<peer_1> [... <peer_N>]]")
        sys.exit(1)

執行結果，peer task 會增加 b'p2', b'' 兩個 frame

$ python peering2.py p2 p1
I: preparing broker at p2...
I: connecting to cloud frontend at p1
Press Enter when all brokers are started:
Worker-0: [b'Client-p2-0', b'', b'HELLO']


$ python peering2.py p1 p2
I: preparing broker at p1...
I: connecting to cloud frontend at p2
Press Enter when all brokers are started:
Worker-0: [b'p2', b'', b'Client-p2-1', b'', b'HELLO']

Worker-1: [b'p2', b'', b'Client-p2-5', b'', b'HELLO']

• C 語言如果使用 CZMQ library 會比較容易開發
• 因為沒有從 peer 取得 state information，所以是假設遠端 workers 都有啟動。

Putting it all together

將 peering1 及 peering2 合併在一起

peering3.py

#
#   Broker peering simulation (part 3) in Python
#   Prototypes the full flow of status and tasks
#
#   While this example runs in a single process, that is just to make
#   it easier to start and stop the example. Each thread has its own
#   context and conceptually acts as a separate process.

import random
import sys
import threading
import time

import zmq


NBR_CLIENTS = 10
NBR_WORKERS = 5

def asbytes(obj):
    s = str(obj)
    if str is not bytes:
        # Python 3
        s = s.encode('ascii')
    return s

def client_task(name, i):
    """Request-reply client using REQ socket"""
    ctx = zmq.Context()
    client = ctx.socket(zmq.REQ)
    client.identity = (u"Client-%s-%s" % (name, i)).encode('ascii')

    client.connect("ipc://%s-localfe.ipc" % name)
    monitor = ctx.socket(zmq.PUSH)
    monitor.connect("ipc://%s-monitor.ipc" % name)

    poller = zmq.Poller()
    poller.register(client, zmq.POLLIN)
    while True:
        time.sleep(random.randint(0, 5))
        for _ in range(random.randint(0, 15)):
            # send request with random hex ID
            task_id = u"%04X" % random.randint(0, 10000)
            client.send_string(task_id)

            # wait max 10 seconds for a reply, then complain
            # 等 10s，看看有沒有收到 message，沒有訊息就是發生 error
            try:
                events = dict(poller.poll(10000))
            except zmq.ZMQError:
                return # interrupted

            if events:
                reply = client.recv_string()
                assert reply == task_id, "expected %s, got %s" % (task_id, reply)
                monitor.send_string(reply)
            else:
                monitor.send_string(u"E: CLIENT EXIT - lost task %s" % task_id)
                return

def worker_task(name, i):
    """Worker using REQ socket to do LRU routing"""
    ctx = zmq.Context()
    worker = ctx.socket(zmq.REQ)
    worker.identity = ("Worker-%s-%s" % (name, i)).encode('ascii')
    worker.connect("ipc://%s-localbe.ipc" % name)

    # Tell broker we're ready for work
    worker.send(b"READY")

    # Process messages as they arrive
    while True:
        try:
            msg = worker.recv_multipart()
        except zmq.ZMQError:
            # interrupted
            return
        # Workers are busy for 0/1 seconds
        time.sleep(random.randint(0, 1))
        worker.send_multipart(msg)

def main(myself, peers):
    print("I: preparing broker at %s..." % myself)

    # Prepare our context and sockets
    ctx = zmq.Context()

    # Bind cloud frontend to endpoint
    cloudfe = ctx.socket(zmq.ROUTER)
    cloudfe.setsockopt(zmq.IDENTITY, myself)
    cloudfe.bind("ipc://%s-cloud.ipc" % myself)

    # Bind state backend / publisher to endpoint
    statebe = ctx.socket(zmq.PUB)
    statebe.bind("ipc://%s-state.ipc" % myself)

    # Connect cloud and state backends to all peers
    cloudbe = ctx.socket(zmq.ROUTER)
    statefe = ctx.socket(zmq.SUB)

    # statefe 註冊取得所有 messages
    statefe.setsockopt(zmq.SUBSCRIBE, b"")
    cloudbe.setsockopt(zmq.IDENTITY, myself)

    for peer in peers:
        print("I: connecting to cloud frontend at %s" % peer)
        cloudbe.connect("ipc://%s-cloud.ipc" % peer)
        print("I: connecting to state backend at %s" % peer)
        statefe.connect("ipc://%s-state.ipc" % peer)

    # Prepare local frontend and backend
    localfe = ctx.socket(zmq.ROUTER)
    localfe.bind("ipc://%s-localfe.ipc" % myself)
    localbe = ctx.socket(zmq.ROUTER)
    localbe.bind("ipc://%s-localbe.ipc" % myself)

    # Prepare monitor socket
    monitor = ctx.socket(zmq.PULL)
    monitor.bind("ipc://%s-monitor.ipc" % myself)

    # Get user to tell us when we can start...
    # raw_input("Press Enter when all brokers are started: ")

    # create workers and clients threads
    for i in range(NBR_WORKERS):
        thread = threading.Thread(target=worker_task, args=(myself, i))
        thread.daemon = True
        thread.start()

    for i in range(NBR_CLIENTS):
        thread_c = threading.Thread(target=client_task, args=(myself, i))
        thread_c.daemon = True
        thread_c.start()

    # Interesting part
    # -------------------------------------------------------------
    # Publish-subscribe flow
    # - Poll statefe and process capacity updates
    # - Each time capacity changes, broadcast new value
    # Request-reply flow
    # - Poll primary and process local/cloud replies
    # - While worker available, route localfe to local or cloud

    local_capacity = 0
    cloud_capacity = 0
    workers = []

    # setup backend poller
    pollerbe = zmq.Poller()
    pollerbe.register(localbe, zmq.POLLIN)
    pollerbe.register(cloudbe, zmq.POLLIN)
    pollerbe.register(statefe, zmq.POLLIN)
    pollerbe.register(monitor, zmq.POLLIN)

    while True:
        # If we have no workers anyhow, wait indefinitely
        try:
            events = dict(pollerbe.poll(1000 if local_capacity else None))
        except zmq.ZMQError:
            break  # interrupted

        previous = local_capacity
        # Handle reply from local worker
        msg = None
        if localbe in events:
            msg = localbe.recv_multipart()
            (address, empty), msg = msg[:2], msg[2:]
            workers.append(address)
            local_capacity += 1

            # If it's READY, don't route the message any further
            if msg[-1] == b'READY':
                msg = None
        elif cloudbe in events:
            msg = cloudbe.recv_multipart()
            (address, empty), msg = msg[:2], msg[2:]

            # We don't use peer broker address for anything

        if msg is not None:
            address = msg[0]
            if address in peers:
                # Route reply to cloud if it's addressed to a broker
                cloudfe.send_multipart(msg)
            else:
                # Route reply to client if we still need to
                localfe.send_multipart(msg)

        # Handle capacity updates
        if statefe in events:
            peer, s = statefe.recv_multipart()
            cloud_capacity = int(s)

        # handle monitor message
        if monitor in events:
            print(monitor.recv_string())


        # Now route as many clients requests as we can handle
        # - If we have local capacity we poll both localfe and cloudfe
        # - If we have cloud capacity only, we poll just localfe
        # - Route any request locally if we can, else to cloud
        while local_capacity + cloud_capacity:
            secondary = zmq.Poller()
            secondary.register(localfe, zmq.POLLIN)
            if local_capacity:
                secondary.register(cloudfe, zmq.POLLIN)
            events = dict(secondary.poll(0))

            # We'll do peer brokers first, to prevent starvation
            # 先處理 cloudfe, task from peers
            if cloudfe in events:
                msg = cloudfe.recv_multipart()
            elif localfe in events:
                msg = localfe.recv_multipart()
            else:
                break  # No work, go back to backends

            if local_capacity:
                msg = [workers.pop(0), b''] + msg
                localbe.send_multipart(msg)
                local_capacity -= 1
            else:
                # Route to random broker peer
                msg = [random.choice(peers), b''] + msg
                cloudbe.send_multipart(msg)

        # local_capacity 異動時，發送 state information 到 statebe
        if local_capacity != previous:
            statebe.send_multipart([myself, asbytes(local_capacity)])

if __name__ == '__main__':
    if len(sys.argv) >= 2:
        myself = asbytes(sys.argv[1])
        main(myself, peers=[ asbytes(a) for a in sys.argv[2:] ])
    else:
        print("Usage: peering3.py <me> [<peer_1> [... <peer_N>]]")
        sys.exit(1)

• client thread 會偵測並回報失敗的 request。是以 10s 內沒有取得 response 的方式偵測。
• client thread 不會直接 print，而是發送給 monitor socket (PUSH)，然後另一端 POLL 進行 print。這是一種用 ZMQ 處理 logging 的方式。
• client 會模擬不同的 loading，偶爾讓 cluster 達到 100%，這時候所有 task 會轉發到 cloud
• main loop 有兩個 poolsets，事實上是三個: information, backend, frontend。

開發時要注意的問題

1. client 可能會停住，因為 request/replies 遺失造成的。ROUTER socket 會丟棄無法 route 的訊息，對應方法是修改 client thread，並偵測遺失的問題。另外加上在 main loop 發送前及接收後，都呼叫 zmsg_dump()，直到問題消失為止。
2. main loop 可能會讀到超過 1 個 ready socket，這是因為第一個訊息遺失了，以只讀取第一個 ready socket 解決

3. zmsg class 不能將 UUIDs 正確編碼為 C strings，因為 UUIDs 失敗時，會變成 0 bytes，將 UUIDs 編碼為 printable hex strings 修正這個問題。

4. 程式沒有偵測 cloud peer 消失的問題，如果啟動多個 peers後，停止一個，因為他已經有廣播 capacity，其他 peers 會發送 task 給他，這會造成 client 會發出 lost message report。解決方式為：只保存 capacity information 一小段時間。或是增加 request-reply chain 的 reliability。

References

ØMQ - The Guide

ZeroMQ 2 Sockets and Patterns

說明 ØMQ patterns: request-reply, pub-sub, pipeline，包含以下項目

1. create and work with ØMQ sockets
2. send/receive messages
3. 如何使用非同步 I/O model 實作 application
4. 如何在一個 thread 處理多個 sockets
5. 如何處理 fatal and nonfatal errors
6. 如何關閉 ØMQ application
7. 如何 send/receive multipart messages
8. 如何 forward message
9. 如何建立一個簡單的 message queueing broker
10. 如何實作 multithread application

Socket API

1. creating and destroying sockets: zmq_socket(), zmq_close()
2. socket 的 options 參數: zmq_setsockopt(), zmq_getsockopt()
3. writing/receiving messages: zmq_send(), zmq_recv()

注意 socket 永遠是 void pointer，message 是 structure。

注意 ØMQ 是非同步的

---

要將兩個 node 連接起來，一個用 zmq_bind()，另一個用 zmq_connect()，通常呼叫 zmq_bind() 的是 "server"，有固定的 network address，另一個是 "client"

ØMQ 跟舊的 TCP connection 的差異在

1. 可使用 inproc, ipc, tcp, pgm, epgm 這些 transport layer
2. 一個 socket 可有多個 outgoing 及 incoming connections
3. 沒有 zmq_accept() method，當 socket 連接到一個 endpoint，會自動開始接受 connection
4. network 連線在背景處理，當網路異常斷線後，ØMQ會自動 reconnect
5. application 不能直接處理 network connecton

最常見的是 client/server model，對所有 clients 來說，server 要保持 visible，因此 server 要使用 zmq_bind()，而 client 要用 zmq_connect()。ØMQ 可接受先啟動 client 再啟動 server。也就是先啟動呼叫 zmq_connect() 的 node。

server node 可使用 socket bind to 很多種 protocol 的 endpoints。但不能 bind 到相同的 endpoint 兩次以上。不過 ipc transport 可以讓一個 process bind to 已經被某個 process 使用的 endpoint，也就是可在 crash 以後，recover socket。

zmq_bind (socket, "tcp://*:5555");
zmq_bind (socket, "tcp://*:9999");
zmq_bind (socket, "inproc://somename");

server 就是架構中的 static component，而 client 是 dynamic parts。

socket 有特殊的類型分別，遵循 messaging pattern 的限制，例如：不能將 publisher 連接到 subscriber socket。

---

使用 zmq_msg_send() zmq_msg_recv() 收送訊息。 ØMQ 的 I/O model 跟傳統的 TCP 不同，TCP socket 是 1-to-1 的。

• ØMQ socket 傳遞訊息類似 UDP，是帶有長度的 binary data，不像是 TCP 是 stream of bytes。
• ØMQ socket 會在背景 thread 處理 I/O，訊息會送到 local input queue，且由 local output queue 送出，不管 application 在處理什麼東西。
• ØMQ socket 根據類型不同，會有內建 one-to-N routing 行為

zmq_send 並不會真的將訊息送到 socket connection，而是先放到 I/O thread 的 queue，並以非同步方式發送。

---

Unicast Transports

ØMQ 提供 inproc, ipc, tcp 這些 unicast teansports，還有 epgm, pgm 這些 multicast transports。

最常見的是使用 tcp: disconnected TCP transport。disconnected 是因為在 endpoints 存在前，就可以連接。

windows platform 不支援 ipc，通常會用 ".ipc" extension，在 UNIX 要注意權限，因為不同 user id 都要有存取權限。

inproc: inter-thread transport，是 connected signaling transport，比 tcp, ipc 快，在 client 執行 connect 以前，server 必須要先執行 bind。

---

ØMQ is not a neutral carrier

有時會有 "如何用 ØMQ 實作 http server" 的問題。但結論是不行，因為 ØMQ 並不是一般化的 socket library，ØMQ 的 framing 機制無法跟其他 protocol 相容。

ØMQ v3.3 以後提供了 ZMQ_ROUTER_RAW 選項，可讓我讀寫沒有 ØMQ framing 機制的 message，就可以利用這個方式，讀寫正常的 http request 與 response，但目前還是實驗性的功能。

---

I/O Threads

ØMQ 是在 background thread 處理 I/O，通常一個 thread 就夠了，當產生一個 context，就會啟動一個 I/O thread，每秒可處理約 1 GB 的資料，但也可以在產生 socket 前，呼叫 zmq_ctx_set() 增加 I/O thread 的數量。

int io_threads = 4;
void *context = zmq_ctx_new ();
zmq_ctx_set (context, ZMQ_IO_THREADS, io_threads);
assert (zmq_ctx_get (context, ZMQ_IO_THREADS) == io_threads);

以往已有案例，一個 socket 一次可以處理上千個 connections，傳統做法是一個 process/thread 處理一個 connection，該 process/thread 處理一個 socket，ØMQ 將整個架構壓縮到單一 process。

如果把 ØMQ 用在 inter-thread communication (multithreaded application)，沒有 external socket I/O，可以將 I/O thread 設定為 0。

Messaging Patterns

ØMQ pattern 是由 pairs of sockets with matching type 實作的，在了解 pattern 前要先了解 socket type。

ØMQ 內建 pattern 為

• Request-Reply: 連接 client 到 services，這是 remote procedure call 及 task distribution pattern

• Pub-Sub: 連接一組 publishers 到 subscribers，這是 data distribution pattern

• Pipeline: 用 fan-out/fan-in pattern 連接 nodes，可有多個 steps 及 loops，這是 parallel task distribution 與 collection pattern

• Exclusive Pair: exclusivly 連接兩個 sockets，這是在連接一個 process 兩個 threads。

前三種在 chap1 已經說明了，現在要討論第四種。

以下是不同的 socket 組合的有效的 connect-bind pair，任意一端都可執行 bind

• PUB and SUB
• REQ and REP
• REQ and ROUTER (take care, REQ inserts an extra null frame)
• DEALER and REP (take care, REP assumes a null frame)
• DEALER and ROUTER
• DEALER and DEALER
• ROUTER and ROUTER
• PUSH and PULL
• PAIR and PAIR

另外還有 XPUB, XSUB，這是 raw versions of PUB 及 SUB

---

以上是四種 ØMQ core patterns，以 C++ 實作。

根據這四個 pattern 還可組成 high-level messaging patterns，這些不在 core library 裡面，可在 ØMQ community 中找到，例如 Majordomo pattern: Reliable Request-Reply Pattern (chap4)，可在 ØMQ 的 github repo 裡面找到。

Working with Messages

ØMQ 有兩組收送訊息的 API，有簡單的 one-liner 的zmq_send() 及 zmq_recv()，但是 zmq_recv() 並不能處理任意長度的 message size，如果超過 buffer 長度會 truncates messages，有另一組 API 會使用 zmq_msg_t structure

• 訊息初始化: zmq_msg_init(), zmq_msg_init_size(), zmq_msg_init_data()
• 收送訊息: zmq_msg_send(), zmq_msg_recv()
• 釋放訊息: zmq_msg_close()
• 存取訊息內容: zmq_msg_data(), zmq_msg_size(), zmq_msg_more()
• 處理訊息屬性: zmq_msg_get(), zmq_msg_set()
• 操作訊息: zmq_msg_copy(), zmq_msg_move()

ØMQ message 是 blobs of any size，可傳送任意一種資料格式，例如 protocol buffers, msgpack, JSON...

在記憶體中，使用 zmq_msg_t structure，在 C 語言的使用規則如下

1. 產生 zmq_msg_t 物件，不是 blocks of data
2. 用 zmq_msg_init() 產生空白的訊息，並傳給 zmq_msg_recv()
3. 使用 zmq_msg_init_size() 產生訊息，然後 allocate a block of data，利用 memcpy 複製資料，傳送訊息給 zmq_msg_send()
4. 呼叫 zmq_msg_close() 釋放訊息，這會釋放 reference，最終會丟棄訊息
5. 使用 zmq_msg_data() 存取訊息資料，使用 zmq_msg_size() 取得訊息長度
6. 使用 zmq_msg_move(), zmq_msg_copy(), zmq_msg_init_data() 前要詳讀手冊
7. 以 zmq_msg_send() 傳送訊息後，ØMQ 會清除 message，也就是設定 size 為 0，不能傳送相同的訊息 2次，發送訊息後，就不能存取該訊息
8. 這些規則不適用於 zmq_send(), zmq_recv()，那兩個 method 是傳送 byte arrays 而不是 message structure

如果想發送相同訊息2次，可用 zmq_msg_init() 並用 zmq_msg_copy() 複製，複製時只會複製 data reference。

ØMQ 支援 multipart messages，可在單一 message 傳送 list of frames，frames (也稱為 message parts) 是 ØMQ message 的 basic wire format，frame 是有指定長度的 block of data。

另外有 wire-level protocol 稱為 ZMTP，定義了 ØMQ 如何在 TCP connection 讀寫 frames。

一開始 ØMQ 只能送單一 frame，後來 API 增加了 "more" flag，可傳送 multipart messages。

• message 可以是 one or more parts
• message parts 稱為 "frames"
• 每一個 frame 都是 zmq_msg_t object
• 在 API 底層，收送訊息都是分開獨立的
• higher-level APIs 提供發送整個 multipart messages 的 wrappers

另外要注意

• 可發送長度為 0 的訊息，例如當作 signal
• ØMQ 保證發送 all parts for a message，或是 none of them
• ØMQ 不會立即發送訊息，但要確認記憶體可放入 multipart message
• message (single or multipart) 必須放入 memory，如果要發送 file，應該要拆成多個 pieces，分別用 single-part message 傳送，因為 multipart data 會耗用較多 memory
• 完成接收訊息後，要呼叫 zmq_msg_close()，某些 programming language api 並不會自動 destroy objects

先不要使用 zmq_msg_init_data()，這是 zero-copy method，可能會造成一些問題

Handling Multiple Sockets

目前 main loop 都是用以下程序

1. wait for message on socket
2. process message
3. repeat

但如果要同時在多個 endpoints 讀取資料要怎麼辦？最簡單的方式，是將所有 endpoints 都連接到一個 socket，然後讓 ØMQ 處理 fan-in，這種方式適用於有相同 pattern 的 endpoints

如果要一次讀取多個 sockets，可使用 zmq_poll()

msreader.py，使用一個簡單的 loop 讀取多個 sockets，後面的 sleep 可能會造成一些 delay 的問題。

另外因為依序在兩個 socket 讀取資料，也就是用 fair read 的形式。

# encoding: utf-8
#
#   Reading from multiple sockets
#   This version uses a simple recv loop

import zmq
import time

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

# Connect to task ventilator
receiver = context.socket(zmq.PULL)
receiver.connect("tcp://localhost:5557")

# Connect to weather server
subscriber = context.socket(zmq.SUB)
subscriber.connect("tcp://localhost:5556")
subscriber.setsockopt(zmq.SUBSCRIBE, b"10001")

# Process messages from both sockets
# We prioritize traffic from the task ventilator
while True:

    # Process any waiting tasks
    while True:
        try:
            msg = receiver.recv(zmq.DONTWAIT)
        except zmq.Again:
            break
        # process task

    # Process any waiting weather updates
    while True:
        try:
            msg = subscriber.recv(zmq.DONTWAIT)
        except zmq.Again:
            break
        # process weather update

    # No activity, so sleep for 1 msec
    time.sleep(0.001)

mspoller.py，這是比較正確，使用 zmq_poll() 讀取多個 socket 的程式

# encoding: utf-8
#
#   Reading from multiple sockets
#   This version uses zmq.Poller()

import zmq

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

# Connect to task ventilator
receiver = context.socket(zmq.PULL)
receiver.connect("tcp://localhost:5557")

# Connect to weather server
subscriber = context.socket(zmq.SUB)
subscriber.connect("tcp://localhost:5556")
subscriber.setsockopt(zmq.SUBSCRIBE, b"10001")

# Initialize poll set
poller = zmq.Poller()
poller.register(receiver, zmq.POLLIN)
poller.register(subscriber, zmq.POLLIN)

# Process messages from both sockets
while True:
    try:
        socks = dict(poller.poll())
    except KeyboardInterrupt:
        break

    if receiver in socks:
        message = receiver.recv()
        # process task

    if subscriber in socks:
        message = subscriber.recv()
        # process weather update

items structure 有以下四個 members:

typedef struct {
    void *socket;       //  ZeroMQ socket to poll on
    int fd;             //  OR, native file handle to poll on
    short events;       //  Events to poll on
    short revents;      //  Events returned after poll
} zmq_pollitem_t;

Multipart Message

ØMQ 可用多個 frames 組成一個 message，也就是 multipart message。

現在要了解如何實作 proxy，讀寫 forward 資料，但不修改資料。

multipart messages 裡面每一個 part 都是 zmq_msg item，如果發送了 5 parts message，就必須產生、發送並銷毀 5 個 zmq_msg items。

發送 multipart message 的方式 (接收每一個 frame 到一個 message object)

zmq_msg_send (&message, socket, ZMQ_SNDMORE);
…
zmq_msg_send (&message, socket, ZMQ_SNDMORE);
…
zmq_msg_send (&message, socket, 0);

這是接收並處理單一訊息多個 parts 的方式

while (1) {
    zmq_msg_t message;
    zmq_msg_init (&message);
    zmq_msg_recv (&message, socket, 0);
    //  Process the message frame
    …
    zmq_msg_close (&message);
    if (!zmq_msg_more (&message))
        break;      //  Last message frame
}

處理 multiparts message 要注意

1. 發送 multipart message 時，在發送最後一個 part 完成後，才會真的將訊息發送出去
2. 如果使用 zmq_poll()，接收了訊息的第一個 part 後，其他部分也都會收到
3. 接收 multipart 訊息，會收到全部，要不然就是全部都沒收到
4. 每一個 part 都是獨立的 zmq_msg item
5. 不管有沒有檢查 more property，都會收到所有 parts of a message
6. ØMQ 會在收到所有 parts of a message 後，才會發送出去
7. 除非 close socket，否則沒有 cancel 部分 parts 的方法

Intermediaries and Proxies

ØMQ 重點是 decentralized intelligence，這不代表不需要 middle nodes，ØMQ 可實作出 pipes 或是 service oriented brokers，messaging 領域將這種產品稱為 intermidiation。ØMQ 則稱為 proxies, queues, forwarders, device 或 brokers。

在現實世界，稱為 wholesalers, distributors, managers...

Dynamic Discovery Problem

在大型分散式架構中，最大的問題是 discovery，各 components 之間如何互相知道對方的位址，也就是 "dynamic discovery problem"

最簡單的解決方式就是，hard-coding (configuring) network architecture，問題在於增加新的 piece 需要 reconfigure network。

這是簡單的 Pub-Sub 架構，如果有 1 publisher，上百個 subscriber，需要在每一個 subscriber 都設定 publisher endpoint。subscriber 是動態增加的，publisher 是固定的，但如果要臨時增加 publisher，不容易調整架構。

另一種簡單的解決方案，就是增加一個固定的 network point，讓其他 node 連接，傳統 messaging 就是用這種做法，稱為 message broker。雖然 ØMQ 沒有內建 broker，但可以簡單完成一個 intermediary。

最好是將 intermediaries 視為簡單的 stateless message switches，類似 http proxy，增加 pub-sub proxy 可解決 dynamic discovery problem，proxy 會 open 1 XSUB socket，1 XPUB socket，並以 well-known IP addresses, ports 連接在一起，其他 processes 會連到這個 proxy 而不是互相連線。

Shared Queue (DEALER and ROUTER sockets)

基本的 client/server application如下，但通常會有多個 services 及 clients，這種方式的限制是 services 必須要是 stateless，request 的 state 可存放在 database。

如果要連接多個 clients 到多個 servers，可直接將每一個 client 連到多個 service endpoints。以下是將 client 連到 service A, B, C，發送四個 request，其中 service A 收到 R1, R4, service B 收到 R2, service C 收到 R3。

問題在於如果要臨時增加 3 個 services，就需要重新設定舊的 clients，增加 services，並 restart clients。

可改用以下 broker 解決這個問題。會使用 nonblocking zmq_poll() 監控兩個 sockets，然後在兩個 socket 之間傳遞訊息。

chap3 另外有 asynchronous request-reply flow 的方法

rrclient.py，連接到 broker，發送 10 個 requests，並等待 response

#
#   Request-reply client in Python
#   Connects REQ socket to tcp://localhost:5559
#   Sends "Hello" to server, expects "World" back
#
import zmq

#  Prepare our context and sockets
context = zmq.Context()
socket = context.socket(zmq.REQ)
socket.connect("tcp://localhost:5559")

#  Do 10 requests, waiting each time for a response
for request in range(1,11):
    socket.send(b"Hello")
    message = socket.recv()
    print("Received reply %s [%s]" % (request, message))

rrbroker.py，可正確處理 multipart messages

# Simple request-reply broker
import zmq

# Prepare our context and sockets
context = zmq.Context()
frontend = context.socket(zmq.ROUTER)
backend = context.socket(zmq.DEALER)
frontend.bind("tcp://*:5559")
backend.bind("tcp://*:5560")

# Initialize poll set
poller = zmq.Poller()
poller.register(frontend, zmq.POLLIN)
poller.register(backend, zmq.POLLIN)

# Switch messages between sockets
while True:
    socks = dict(poller.poll())

    if socks.get(frontend) == zmq.POLLIN:
        message = frontend.recv_multipart()
        backend.send_multipart(message)

    if socks.get(backend) == zmq.POLLIN:
        message = backend.recv_multipart()
        frontend.send_multipart(message)

rrworker.py

#
#   Request-reply service in Python
#   Connects REP socket to tcp://localhost:5560
#   Expects "Hello" from client, replies with "World"
#
import zmq

context = zmq.Context()
socket = context.socket(zmq.REP)
socket.connect("tcp://localhost:5560")

while True:
    message = socket.recv()
    print("Received request: %s" % message)
    socket.send(b"World")

以下是 Request-Reply Broker 的示意圖，只有中間的 broker 是固定的 node

ØMQ 內建的 proxy function

剛剛的 rrbroker 是很有用的元件，可用來建立 pub-sub forwarders，shared queues，ØMQ 包裝成一個 method zmq_proxy(frontend, backend, capture)

msgqueue.py，可直接替換掉 rrbroker.py

"""

   Simple message queuing broker
   Same as request-reply broker but using ``zmq.proxy``
"""

import zmq


def main():
    """ main method """

    context = zmq.Context()

    # Socket facing clients
    frontend = context.socket(zmq.ROUTER)
    frontend.bind("tcp://*:5559")

    # Socket facing services
    backend  = context.socket(zmq.DEALER)
    backend.bind("tcp://*:5560")

    zmq.proxy(frontend, backend)

    # We never get here...
    frontend.close()
    backend.close()
    context.term()


if __name__ == "__main__":
    main()

實際上可以將 ROUTER/DEALER, XSUB/XPUB, PULL/PUSH 這三組 socket type 連接到 zmq.proxy()

Transport Bridging

ØMQ 常見問題是，如何將 ØMQ network 連接到某一種技術的網路中？ 可能是 networking 或 messaging technology

解決方式是使用 bridge，bridge 就是在某個 socket 使用某一種 protocol，在另一個 socket 轉換到另一種 protocol，也就是 protocol interpreter

wuproxy.py，會使用 SUB 連接 weather sever，然後轉換以 PUB 發送到 external network

import zmq

context = zmq.Context()

# This is where the weather server sits
frontend = context.socket(zmq.SUB)
frontend.connect("tcp://192.168.55.210:5556")

# This is our public endpoint for subscribers
backend = context.socket(zmq.PUB)
backend.bind("tcp://10.1.1.0:8100")

# Subscribe on everything
frontend.setsockopt(zmq.SUBSCRIBE, b'')

# Shunt messages out to our own subscribers
while True:
    # Process all parts of the message
    message = frontend.recv_multipart()
    backend.send_multipart(message)

Handling Errors and ETERM

ØMQ 以 fail-fast 及 resilience 方式處理 error。當 process 發生 internal error，就容易受到外界的攻擊，因此如果發生 internal error，就會 self-destruct。

當 ØMQ 偵測到 external fault，會對 calling node 發出 error，某些少數特殊狀況下，會直接丟棄訊息，沒有從這種 error 復原的方法。

在 C 的程式裡，需要對每一個 ØMQ call 都進行 error handling。

• create object 的 method 回傳 null 就是 failed
• 處理資料的 method，回傳了已處理的 bytes 數量，如果是 -1 就是發生 error
• 其他 method 如果回傳 0 就是成功，-1 就是 error
• error code 是在 errno 或是 zmq_errno() 提供
• 可用 zmq_strerror() 轉換為 error 文字說明

void *context = zmq_ctx_new ();
assert (context);
void *socket = zmq_socket (context, ZMQ_REP);
assert (socket);
int rc = zmq_bind (socket, "tcp://*:5555");
if (rc == -1) {
    printf ("E: bind failed: %s\n", strerror (errno));
    return -1;
}

有兩種例外狀況，應該視為 nonfatal

1. 收到訊息，夾帶 ZMQ_DONTWAIT option，沒有 waiting data，ØMQ 會 return -1 並設定 errno 為 EAGAIN
2. 某個 thread 呼叫 zmq_ctx_destory()，其他thread 會停止運作，zmq_ctx_destory() 會把 context 關掉，並 block 所有 function call，exit with -1，errno 設定為 ETERM

---

以下是如何正確停止 process 的範例。

因為 PUSH/PULL socket 是單向的，要增加 PUB/SUB socket，用來發送 KILL signal 給 worker

• sink 產生新的 PUB socket 到新的 endpoint
• worker bind input socket 到 endpoint
• 當 sink 偵測到工作完成，就發送 kill 到 PUB socket
• 當 worker 偵測 kill message，就 exit

taskworker2.py

# encoding: utf-8
#
#   Task worker - design 2
#   Adds pub-sub flow to receive and respond to kill signal

import sys
import time
import zmq

context = zmq.Context()

# Socket to receive messages on
receiver = context.socket(zmq.PULL)
receiver.connect("tcp://localhost:5557")

# Socket to send messages to
sender = context.socket(zmq.PUSH)
sender.connect("tcp://localhost:5558")

# Socket for control input
controller = context.socket(zmq.SUB)
controller.connect("tcp://localhost:5559")
controller.setsockopt(zmq.SUBSCRIBE, b"")

# Process messages from receiver and controller
poller = zmq.Poller()
poller.register(receiver, zmq.POLLIN)
poller.register(controller, zmq.POLLIN)
# Process messages from both sockets
while True:
    socks = dict(poller.poll())

    if socks.get(receiver) == zmq.POLLIN:
        message = receiver.recv_string()

        # Process task
        workload = int(message)  # Workload in msecs

        # Do the work
        time.sleep(workload / 1000.0)

        # Send results to sink
        sender.send_string(message)

        # Simple progress indicator for the viewer
        sys.stdout.write(".")
        sys.stdout.flush()

    # Any waiting controller command acts as 'KILL'
    if socks.get(controller) == zmq.POLLIN:
        break

# Finished
receiver.close()
sender.close()
controller.close()
context.term()

tasksink2.py，收到 100 個 task 結果後，就發送 controller.send(b"KILL")

# encoding: utf-8
#
#   Task sink - design 2
#   Adds pub-sub flow to send kill signal to workers

import sys
import time
import zmq

context = zmq.Context()

# Socket to receive messages on
receiver = context.socket(zmq.PULL)
receiver.bind("tcp://*:5558")

# Socket for worker control
controller = context.socket(zmq.PUB)
controller.bind("tcp://*:5559")

# Wait for start of batch
receiver.recv()

# Start our clock now
tstart = time.time()

# Process 100 confirmiations
for task_nbr in range(100):
    receiver.recv()
    if task_nbr % 10 == 0:
        sys.stdout.write(":")
    else:
        sys.stdout.write(".")
    sys.stdout.flush()

# Calculate and report duration of batch
tend = time.time()
tdiff = tend - tstart
total_msec = tdiff * 1000
print("Total elapsed time: %d msec" % total_msec)

# Send kill signal to workers
controller.send(b"KILL")

# Finished
receiver.close()
controller.close()
context.term()

Handling Interrupt Signals

正式程式需要針對中斷信號 ex: Ctrl-C 或 SIGTERM 停止程式。預設狀況下，只會 kill process，這表示 messages 不會被 flushed，檔案不會正常關閉。

interrupt.py，這是處理 Ctrl-C 的方式

#
#   Shows how to handle Ctrl-C
#
import signal
import time
import zmq

context = zmq.Context()
socket = context.socket(zmq.REP)
socket.bind("tcp://*:5558")

# SIGINT will normally raise a KeyboardInterrupt, just like any other Python call
try:
    socket.recv()
except KeyboardInterrupt:
    print("W: interrupt received, stopping...")
finally:
    # clean up
    socket.close()
    context.term()

程式提供 s_catch_signals() method，會抓到 Ctrl-C(SIGINT) 及 SIGTERM 信號，並設定 global variable: s_interrupted。

在程式一開始就要呼叫 s_catch_signals()，處理 signaling handling

s_catch_signals ();
client = zmq_socket (...);
while (!s_interrupted) {
    char *message = s_recv (client);
    if (!message)
        break;          //  Ctrl-C used
}
zmq_close (client);

偵測 Memory Leak

如果是使用自動管理記憶體的程式語言，就不用處理這個問題。如果是使用 C/C++，就要管理記憶體。

首先要安裝 valgrind

sudo apt-get install valgrind

預設狀況下，ØMQ 會讓 valgrind 產生很多資料，可建立一個vg.supp 檔案，填入以下內容，移除 warnings

{
    <socketcall_sendto>
    Memcheck:Param
    socketcall.sendto(msg)
    fun:send
    ...
}
{
    <socketcall_sendto>
    Memcheck:Param
    socketcall.send(msg)
    fun:send
    ...
}

以 -DDEBUG 選項，編譯程式

利用 valgrind 執行程式

valgrind --tool=memcheck --leak-check=full --suppressions=vg.supp someprog

Multithreading with ØMQ

ØMQ 應該是寫 multithread 程式做好的工具，且不需要 mutexes, locks, 或其他 inter-thread communication 機制

concurrent programming 要注意: "don't share state"

利用 ØMQ 撰寫 multithread code 要注意的事項

1. 分離 data，不要在多個 thread 之間分享資料，只能用 ØMQ context (threadsafe) 傳遞資料。
2. 不要使用傳統的 concurrency mechanisms，例如: mutexes, criticl sections, semaphores...
3. 在 process 啟動時，產生一個 ØMQ context，然後傳給其他需要使用 inproc socket 的 threads
4. 利用 attached thread 產生 structure，並以 inproc 的 PAIR socket 連接到 parent threads，pattern 為 bind parent socket，然後產生連接該 socket 的 child thread
5. 使用 attached threads 以他們自己的 contexts 模擬獨立 tasks，以 tcp 方式連線。然後再將程式移到 stand-alone processes。
6. 只能用 ØMQ message 在 thread 之間傳遞資料
7. 不要在 thread 之間分享 ØMQ socket，因為 ØMQ socket 不是 threadsafe

只能在產生 socket 的 thread 裡面使用或關閉 sockets

ØMQ 是使用 native OS threads，而不是 "green" threads，可利用 Intel 的 ThreadChecker 工具查看程式在做的工作。缺點是 native threading API 並不 portable

單一個 ØMQ thread 可以在單 CPU core 全速運作，不需要 wait。

mtserver.py

"""
   Multithreaded Hello World server
"""
import time
import threading
import zmq


def worker_routine(worker_url, context=None):
    """Worker routine"""
    context = context or zmq.Context.instance()
    # Socket to talk to dispatcher
    socket = context.socket(zmq.REP)

    socket.connect(worker_url)

    while True:

        string  = socket.recv()

        print("Received request: [ %s ]" % (string))

        # do some 'work'
        time.sleep(1)

        #send reply back to client
        socket.send(b"World")


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

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

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

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

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

    # Launch pool of worker threads
    for i in range(5):
        thread = threading.Thread(target=worker_routine, args=(url_worker,))
        thread.daemon = True
        thread.start()

    zmq.proxy(clients, workers)

    # We never get here but clean up anyhow
    clients.close()
    workers.close()
    context.term()


if __name__ == "__main__":
    main()

• server 啟動多個 worker threads，每一個 worker thread 都產生 REP socket，處理 request。worker thread 是單一 threaded 的 server，差異只是由 tcp 換成 inproc
• server 產生 ROUTER socket，連接 clients (tcp)
• server 產生 DEALER socket，連接 worker (inproc)
• server 啟動 proxy，連接 ROUTER/DEALER sockets

create thread 的 function 在每一種 programming language 都不同，POSIX library 是 pthreads，但 windows 是不同的 API

Signaling Between Threads (PAIR Sockets)

如何在多個 thread 之間協作：使用 ØMQ messages

mtrelay.py

"""
   Multithreaded relay
"""

import threading
import zmq

def step1(context=None):
    """Step 1"""
    context = context or zmq.Context.instance()
    # Signal downstream to step 2
    sender = context.socket(zmq.PAIR)
    sender.connect("inproc://step2")

    sender.send(b"")


def step2(context=None):
    """Step 2"""
    context = context or zmq.Context.instance()
    # Bind to inproc: endpoint, then start upstream thread
    receiver = context.socket(zmq.PAIR)
    receiver.bind("inproc://step2")

    thread = threading.Thread(target=step1)
    thread.start()

    # Wait for signal
    msg = receiver.recv()

    # Signal downstream to step 3
    sender = context.socket(zmq.PAIR)
    sender.connect("inproc://step3")
    sender.send(b"")


def main():
    """ server routine """
    # Prepare our context and sockets
    context = zmq.Context.instance()

    # Bind to inproc: endpoint, then start upstream thread
    receiver = context.socket(zmq.PAIR)
    receiver.bind("inproc://step3")

    thread = threading.Thread(target=step2)
    thread.start()

    # Wait for signal
    string = receiver.recv()

    print("Test successful!")

    receiver.close()
    context.term()


if __name__ == "__main__":
    main()

• thread 透過 shared context 以 inproc socket 通訊
• parent thread 產生 socket，連接到 inproc:// endpoint，將 context 傳給 child thread

使用 PAIR socket 的原因：

1. 可使用 PUSH/PULL，但 PUSH 會將 message 傳給所有 receivers，如果不小心產生了兩個 receivers，可能會遺失一半的 signals，PAIR 的優點是，只能有一個 connection，it's "exclusive"
2. 可使用 DEALER/ROUTER，但 ROUTER 會將訊息包裝起來，將size 為 0 的 signal 轉換為 multipart message，如果不在意，就可以使用 DEALER/ROUTER。另外 DEALER 跟 PUSH 一樣，會平均分配訊息給接收端
3. 可使用 PUB/SUB，且 PUB 沒有 PUSH, DEALER 的問題。但需要以 empty subscription 設定 subscriber，用起來麻煩。

Node Coordination

當我們需要在網路多個 nodes 之間協作，PAIR socket 不適用，在 threads 跟 nodes 協作是不同的，因為 nodes 可能會隨時上線或斷線，PAIR sockets 在遠端 node 斷線後，不會自動 reconnect。

node與 thread 另一個差異是 thread 數量通常是固定的，但 nodes 數量會異動。

• publisher 已經先知道會有幾個 subscribers
• publisher 啟動並等待所有 subscribers 連線，subscriber 透過 REQ/REP 通知 publisher 已經上線
• 當所有 subscribers 都連線後，publisher 就開始發送資料

synpub.py

#
#  Synchronized publisher
#
import zmq

#  We wait for 10 subscribers
SUBSCRIBERS_EXPECTED = 10

def main():
    context = zmq.Context()

    # Socket to talk to clients
    publisher = context.socket(zmq.PUB)
    # set SNDHWM, so we don't drop messages for slow subscribers
    publisher.sndhwm = 1100000
    publisher.bind('tcp://*:5561')

    # Socket to receive signals
    syncservice = context.socket(zmq.REP)
    syncservice.bind('tcp://*:5562')

    # Get synchronization from subscribers
    subscribers = 0
    while subscribers < SUBSCRIBERS_EXPECTED:
        # wait for synchronization request
        msg = syncservice.recv()
        # send synchronization reply
        syncservice.send(b'')
        subscribers += 1
        print("+1 subscriber (%i/%i)" % (subscribers, SUBSCRIBERS_EXPECTED))

    # Now broadcast exactly 1M updates followed by END
    for i in range(1000000):
        publisher.send(b'Rhubarb')

    publisher.send(b'END')

if __name__ == '__main__':
    main()

synpub.py

#
#  Synchronized subscriber
#
import time

import zmq

def main():
    context = zmq.Context()

    # First, connect our subscriber socket
    subscriber = context.socket(zmq.SUB)
    subscriber.connect('tcp://localhost:5561')
    subscriber.setsockopt(zmq.SUBSCRIBE, b'')

    time.sleep(1)

    # Second, synchronize with publisher
    syncclient = context.socket(zmq.REQ)
    syncclient.connect('tcp://localhost:5562')

    # send a synchronization request
    syncclient.send(b'')

    # wait for synchronization reply
    syncclient.recv()

    # Third, get our updates and report how many we got
    nbr = 0
    while True:
        msg = subscriber.recv()
        if msg == b'END':
            break
        nbr += 1

    print ('Received %d updates' % nbr)

if __name__ == '__main__':
    main()

以 bash 啟動

echo "Starting subscribers..."

for ((a=0; a<10; a++)); do
        python syncsub.py &
done

echo "Starting publisher..."
python syncpub.py

Zero-Copy

ØMQ message API 可由 application buffer 直接傳遞訊息，不需要 copying data，這可改善效能。

先以 zmq_msg_init_data() 產生訊息，並設定 reference of data block (利用 malloc() 取得的資料區塊)，傳遞給 zmq_msg_send()

void my_free (void *data, void *hint) {
    free (data);
}
//  Send message from buffer, which we allocate and ZeroMQ will free for us
zmq_msg_t message;
zmq_msg_init_data (&message, buffer, 1000, my_free, NULL);
zmq_msg_send (&message, socket, 0);

發送 message 後不需要呼叫 zmq_msg_close()，libzmq 會自動呼叫

無法在 receive 端提供 Zero-Copy 功能。

multipart message 也可使用 zero-copy

Pub-Sub Message Envelopes

在 pub-sub pattern 可將 key/data 分到不同的 message frame 裡面，稱為 envelope

psenvpub.py

"""
   Pubsub envelope publisher
"""
import time
import zmq

def main():
    """main method"""

    # Prepare our context and publisher
    context   = zmq.Context()
    publisher = context.socket(zmq.PUB)
    publisher.bind("tcp://*:5563")

    while True:
        # Write two messages, each with an envelope and content
        publisher.send_multipart([b"A", b"We don't want to see this"])
        publisher.send_multipart([b"B", b"We would like to see this"])
        time.sleep(1)

    # We never get here but clean up anyhow
    publisher.close()
    context.term()


if __name__ == "__main__":
    main()

psenvsub.py

"""
   Pubsub envelope subscriber
"""
import zmq

def main():
    """ main method """

    # Prepare our context and publisher
    context    = zmq.Context()
    subscriber = context.socket(zmq.SUB)
    subscriber.connect("tcp://localhost:5563")
    subscriber.setsockopt(zmq.SUBSCRIBE, b"B")

    while True:
        # Read envelope with address
        [address, contents] = subscriber.recv_multipart()
        print("[%s] %s" % (address, contents))

    # We never get here but clean up anyhow
    subscriber.close()
    context.term()


if __name__ == "__main__":
    main()

如果有多個 publisher，需要知道 address，可以在發送時，增加 address 的資料。

High-Water Mark

如果在 process 之間發送很多訊息，但記憶體不足，或是一些delay 可能會造成 server crash。

想像 process A 快速發送很多訊息給 process B，但 B 因為 busy 處理速度變慢，有些訊息會停在 B 的 network buffer，有些在網路上，有些在 A 的 network buffer，其他的會累積在 A 的記憶體裡面，很快地會造成 A crash。

這在 message brokers 常常發生，B 造成這樣的問題。

一種解決方式是 flow control，通知 A 停止發送訊息。

但某些情境無法使用這種方法，例如會持續發生的 twitter feed。

另一種方式，是由 transport layer 提供機制，將訊息的 buffer 設定容量上限，而在 buffer 到達上限時，採用某種處理方案，例如 wait。

ØMQ 採用 HWM (high-water mark) 的概念，定義 internal pipes 的容量，每一個連出或連入的 socket 都有獨立的 pipe，設定發送或接收的 HWM，例如 PUB, SUB 只有 send buffer，而 SUB, PULL, REQ, REP 只有 receive buffer， 而 DEALER, ROUTER, PAIR 有 send 及 receive buffers。

在 ØMQ v2.x，HWM 預設為無限大。在 ØMQ v3.x 預設為 1000。

當 socket buffer 到達 HWM 容量，可以根據 socket type 決定 block 或是 drop data。PUB, ROUTER sockets 會 drop data，其他的 sockets 會 block data。

在 inproc 傳輸裡面，sender 與 receiver 會共用 buffers，因此實際上 HWM 是雙向設定的 HWM 的總和。

HWM 並不一定是正確值，預設是最多可使用 1000 messages，實際上會因為 libmq 實作 queue 的方式，造成 buffer size 比 1000 小。

Missing Message Problem Solver

在實作 ØMQ application 時，常會遇到遺失訊息的問題，以下是遺失訊息可能的發生原因：

• SUB socket 必須要用 zmq_setsockopt() 設定 ZMQ_SUBSCRIBE subscription 條件，否則可能會收到到訊息，因為是以 prefix 方式做為條件，如果設定為 "" 會收到所有訊息
• 在啟動 PUB socket 並發送 data 後，才啟動 SUB socket，可能會遺失前面的訊息，必須調整架構讓 SUB socket 先啟動，然後 PUB socket 再發送訊息。
• 如果同步了 SUB, PUB socket，還是有可能會遺失訊息，這是因為 internal queue 會在連線完整建立後，才會產生出來，如果遇到這樣的問題，就將 bind/connect 的方向交換，讓 SUB socket binds，而 PUB socket connects
• 如果使用了 REP, REQ sockets，但沒有遵循 send/recv/send/recv 的順序，ØMQ 會發生 error。就好像遺失訊息一樣，使用 REQ, REQ 必須遵循 sned/recv 的順序，並檢查 ØMQ calls 的 error code
• 使用 PUSH socket，第一個 PULL socket 會取得較多訊息，因為只會在所有 PULL sockets 都連上的時候(需要幾 ms)，才能公平分配訊息。對於 low data rate 的狀況，可以用 ROUTER/DEALER 以及 load balancing pattern 取代 PUSH/PULL
• 如果透過 threads 分享 socket，可能會造成 crash

• 如果使用 inproc，確保 sockets 都是使用相同的 context，否則 connecting side 會發生問題。另外要先 bind 再 connect，因為 inproc 不像是 tcp，不是 disconnected transport protocol。

• 如果使用 ROUTER，意外狀況會發生遺失訊息的狀況，例如發送了 malformed identify frames，或是忘了發送 identity frame。通常要在 ROUTER 設定 ZMQ_ROUTER_MANDATORY option，且要在 send call 都要檢查 return code

References

ØMQ - The Guide