Loopback 妙用

楔子

multi-threading programming 免不了需要實作 Inter-thread communication(ITC)，最常見的大概就是 consumer/producer：

大部分人第一個想法就是用 queue：

於是類似下方的 code 大概一百間公司會出現一百次：

//safequeue.h
#ifndef SAFEQUEUE_H
#define SAFEQUEUE_H

#include <QQueue>
#include <QMutex>
#include <QString>

class SafeQueue
{    
    QQueue<QString> m_queue;
    mutable QMutex m_queue_mtx;

public:
    SafeQueue();
    int count();
    QString pop();
    void push(const QString& s);

};

#endif // SAFEQUEUE_H

#include "safequeue.h"

SafeQueue::SafeQueue()
{
}

int SafeQueue::count()
{
    QMutexLocker locker(&m_queue_mtx);
    return m_queue.size();
}

QString SafeQueue::pop()
{
    QMutexLocker locker(&m_queue_mtx);
    if(m_queue.size() > 0)
        return m_queue.dequeue();
    else
        return "";
}

void SafeQueue::push(const QString &s)
{
    QMutexLocker locker(&m_queue_mtx);
    m_queue.enqueue(s);
}

Consumer Thread 則是長得像下面這樣：

void ConsumerThread(void *p)
{
    SafeQueue *queue = (SafeQueue*)p;
    //...
    while(1)
    {
        if(queue->count() > 0)
        {
            QString s = queue->pop();
            //do something
        }
        else
        {
            Sleep(1); //sleep 1ms
        }

    }
    //...
}

這樣寫不能說完全不對，但是有三個缺點：

1. 以現代 CPU 來講，1ms 可以做很多事(就算 ARM 現在 1 GHz CPU 也很常見)，而這裡是完全浪費掉了。
2. 以 Windows 來說，若沒有特別調整，Sleep() 的精度只有 30ms，也就是你浪費了 30ms。
3. ConsumerThread 反應不夠靈敏：理想上應該是一有資料就要消化。這邊 1/30ms = 33.33，一秒鐘只能處理 33 次反應也太慢了吧？

方法2: 條件變數(condition variable)

站長最早從 PThread 習得這個方法，Windows 要到 8.0 之後才有相關 API，不過有些 framework(e.g. Qt) 有模擬出來。這個方法的主要精神就是 Thread(Consumer) 發現條件沒有成立就進入睡眠狀態，一旦由其他 Thread(Producer) 促成條件成立就喚醒睡眠中的 Thread(Consumer)，實作如下：

//safequeue2.h
#ifndef SAFEQUEUE2_H
#define SAFEQUEUE2_H

#include <QQueue>
#include <QWaitCondition>
#include <QMutex>
#include <QString>

class SafeQueue
{        
    QQueue<QString> m_queue;
    mutable QMutex m_queue_mtx;
    mutable QWaitCondition m_queue_wc; //<--

public:
    SafeQueue();
    int count();
    QString pop();
    void push(const QString& s);

};

#endif // SAFEQUEUE_H

//safequeue2.cpp
#include "safequeue.h"

SafeQueue::SafeQueue()
{
}

int SafeQueue::count()
{
    QMutexLocker locker(&m_queue_mtx);
    return m_queue.size();
}

QString SafeQueue::pop()
{
    m_queue_mtx.lock();
    if(m_queue.size() == 0)
    {
        m_queue_wc.wait(&m_queue_mtx);
        QString s = m_queue.dequeue();
        m_queue_mtx.unlock();
        return s;
    }
    else
    {
        QString s = m_queue.dequeue();
        m_queue_mtx.unlock();
        return s;
    }
}

void SafeQueue::push(const QString &s)
{
    QMutexLocker locker(&m_queue_mtx);
    int old = m_queue.size();
    m_queue.enqueue(s);
    if(old == 0)
        m_queue_wc.wakeOne();
}

當 Consumer Thread 呼叫 m_queue.pop() 時若是沒有資料，則會呼叫 m_queue_wc.wait() 進入睡眠，並且 unlock mutex(m_queue_mtx)，等到 Producer Thread 呼叫 m_queue.push() 後，發現之前 queue 是空的(表示可能有 Consumer Thread 在等待)，就呼叫 m_queue_wc.wakeOne() 叫醒 Consumer Thread，Consumer Thread 會重新取得 lock mutex(m_queue_mtx)，取得資料然後返回。

我們解決了 Consumer Thread 傻等的問題，但碰到下面的情形呢？

當然，SafeQueue::push() 已經加上 mutex 保護了，所以可以放心給 A/B/C 三個 threads 同時呼叫。不過若是深度思考，Thread A 在 push 時 Thread B/C 被阻擋在門外，在 CPU 核心越來越多的今天似乎效能被浪費掉了，有沒有讓 A/B/C 同時可以 push 的方法？

直覺的想法就是把 queue 一分為三：

接著可以借助 semaphore(e.g. win32: CreateSemaphore)，與 WaitForMultipleObjects 這兩樣武器，不過 WaitForMultipleObjects 很容易誤用，這個方案也會走到越來越複雜的境界。

沒別的方法了嗎？

方法3: Loopback

這個方法的核心就是用 TCP loopback 通訊 (UDP 應該也行，站長沒試過)替代 DIY queue，對 Winsock/BSD socket 有粗淺認識的人都知道一旦 connect()/accept() 成功，就會產生一組全雙工通訊管道：

我們只要把這個 client/server 連結的行為放在本地端，server binding local address (IPv4: 127.0.0.1、IPv6: ::1)，就可以讓兩個 thread 互相通訊：

這樣做有幾個優點：

• Thread A/B 各自使用獨立的 socket handle 進行通訊，不需要 mutex，也就是 lock-free。
• 通訊通道記憶體管理由作業系統負責，程序員無須關心。
• Thread(e.g. Consumer Thread) 可以使用 select() 掃瞄多個 socket handle，select() 是跨平台函式，同樣的作法可以移植到 Linux，反之亦然。

示意圖：

如果你覺得這樣是旁門左道，微軟在 Windows Server 2012 與 Windows 8 添加了 "TCP Loopback Fast Path"，而 "Fast TCP Loopback Performance and Low Latency with Windows Server 2012 TCP Loopback Fast Path" 這篇文章明白指出這個新功能就是替使用 TCP loopback 做 IPC 的軟體加速。

事實上在 UNIX 裡早有個使用多年的 system call: socketpair() 用來產生上述用來通訊用的成對 socket handle，其第一個參數 domain 在大多數的實作裡(e.g. Linux)，只能填入 AF_UNIX(UNIX domain socket)，表示 socketpair 的用途就是 IPC/ITC。

在 Linux 世界裡頗流行的 D-Bus，也是建築在 UNIX domain socket 之上。

腦筋動的快的人可以發現，這樣的架構不只可以用在 IPC/ITC 上，既然原來就是使用 TCP，那表示也有機會把 Thread 打散到不同的機器上進行分散式運算。如果平台限定在 Linux 上，則可以使用效率更高的 epoll。

如果您讀到這裡躍躍欲試，但不知道怎麼開始，不用辛苦的 k winsock API(當然 Linux user 輕舟已過萬重山)，可以參考 libevent 的 evutil_ersatz_socketpair()，他已經在那邊等您了 :)

缺點:

1. 跟 DIY queue 相比，socket 是有限的資源。
2. 傳遞資料從 C++ object 變為 byte stream。

關於 1.，Windows 限制單一 process 能使用的 socket 數上限為 32767，這對大部分應用來說都非常足夠，而 select() 在 Windows 上預設也只能輪詢 64 個 socket handles，如果您要使用很多 socket handles + threads，恐怕要先討論的是是不是要把雞蛋全放在同一個籃子，而且如同站長之前的文章解釋過，同時啟用太多 threads 並不會有什麼好下場。甚至於如果您有大量輪詢 socket handles 的需求，可以考慮移師 Linux，因為 Linux epoll() 輪詢上千上萬個 handles 並不罕見。

2.  的問題變成要設計通訊協議，在過去也是可能一百家公司有一百種作法，好在 Google 很大方的開放了自家設計的 ProtoBuf 供大眾使用，只要先定義好 schema 就會自動產生相關讀寫程式碼，免除各家自造輪子不相容的問題。

參考資料

Python Cookbook, 3rd Edition >12.13. Polling Multiple Thread Queues

希望這篇文章對您有幫助 ^^