2010-04-23 191 views
16

我正在尋找一種方法來實現無鎖隊列數據結構,該數據結構支持單生產者和多個消費者。我看過Maged Michael和Michael Scott(1996)的經典方法,但他們的版本使用鏈接列表。我想要一個使用有界的循環緩衝區的實現。使用原子變量的東西?鎖定免費隊列 - 單個生產者,多個消費者

在附註上,我不確定爲什麼這些經典方法是爲需要大量動態內存管理的鏈接列表設計的。在多線程程序中,所有的內存管理例程都被序列化。我們不是通過將它們與動態數據結構結合使用來打破無鎖方法的好處嗎?

我想在英特爾64位架構上使用pthread庫在C/C++中編寫此代碼。

謝謝 希裏什

+1

有限公司大小的緩衝區意味着,如果有一個在它沒有空的空間製片人可能會失敗。你能接受嗎? – doublep 2010-04-23 22:38:09

+1

另請注意,在C++中,您可以將自己的分配器提供給'std :: list'。由於你只有一個生產者,所以這個分配器不需要同步。例如,它可以從預先分配的緩衝區中「分配」列表節點,並且當空間用完時,使用全局同步的'malloc()'''''''''''''''''''''''''''''''''''這意味着它只會在1%的通話中使用同步。 – doublep 2010-04-23 22:41:52

+0

如果你想優化線程的內存使用率,tcmalloc是一個很好的庫。由於它爲每個線程維護內存池,因此可能避免內存例程序列化問題。 – 2010-04-23 22:49:58

回答

1

對於傳統的單塊循環緩衝區我覺得這根本無法與原子操作安全進行。你需要在一次閱讀中這麼做。假設你有一個具有這樣的結構:

uint8_t* buf; 
unsigned int size; // Actual max. buffer size 
unsigned int length; // Actual stored data length (suppose in write prohibited from being > size) 
unsigned int offset; // Start of current stored data 

在你需要做以下的讀取(這是我如何實現它無論如何,你可以交換一些步驟就像我以後會討論):

  1. 檢查讀取長度不超過存儲長度
  2. 檢查偏移+閱讀長度不超過緩衝區邊界
  3. 讀取數據輸出
  4. 增加所抵消,減少長

你應該怎麼做同步(如此原子)才能使這項工作?事實上,合併步驟1和4在一個原子一步,或澄清:這樣做同步:

  1. 檢查read_length,這可能是某事像read_length=min(read_length,length);
  2. 與read_length下降長度:length-=read_length
  3. 得到一個本地副本從偏移unsigned int local_offset = offset
  4. 增加與read_length偏移:offset+=read_length

之後你可以做一個的memcpy(或W從你的local_offset開始,檢查你的讀是否超過循環緩衝區大小(分爲2個memcpy's),...。這是'相當'的線程安全,您的寫入方法仍然可以覆蓋您正在閱讀的內存,所以請確保您的緩衝區足夠大,以最大限度地減少這種可能性。儘管我可以想象你可以在原子操作中組合3和4(我猜這就是他們在鏈表列表中所做的事情),甚至1和2,但是我看不到你在一個原子操作:)。

然而,如果您的消費者非常聰明,並且總是知道要閱讀什麼,那麼您可以嘗試刪除「長度」。您還需要一個新的woffset變量,因爲用於確定寫入偏移量的(offset + length)%大小的舊方法不再適用。請注意,這與鏈接列表的情況很接近,實際上您總是從列表中讀取一個元素(=固定的,已知大小)。同樣在這裏,如果你把它作爲一個循環鏈表,你可以閱讀到很多,或者寫下你當時正在閱讀的位置!

最後:我的建議,只是去鎖,我使用CircularBuffer類,完全安全的閱讀&寫)實時720p60視頻流媒體,我沒有速度問題的鎖定。

8

使用循環緩衝區使得鎖定成爲必需,因爲需要阻塞來防止頭部通過尾部。但是否則頭部和尾部指針可以很容易地自動更新。或者在某些情況下,緩衝區可能非常大,覆蓋不成問題。 (在現實生活中,您會在自動交易系統中看到這一點,通過循環緩衝區大小可以容納X分鐘的市場數據。如果您落後X分鐘,與覆蓋緩衝區相比,您會遇到更嚴重的問題)。

當我在C++中實現了MS隊列時,我使用堆棧構建了一個無鎖分配器,這非常容易實現。如果我有MSQueue,那麼在編譯時我知道sizeof(MSQueue :: node)。然後我製作一堆所需大小的N個緩衝區。 N可以增長,即如果pop()返回null,很容易去向堆請求更多的塊,並將它們推入堆棧。除了可能阻塞的更多內存之外,這是一個無鎖操作。

請注意,T不能有一個不平凡的dtor。我研究了一個允許使用非平凡dtors的版本,它確實有效。但是我發現,讓T成爲我想要的T的指針就更容易了,生產者發佈所有權,消費者獲得所有權。這當然要求T本身使用無鎖方法進行分配,但是我也使用了與堆棧相同的分配器。

在任何情況下,無鎖編程的重點不在於數據結構本身較慢。要點是這樣的:

  1. lock free使我獨立於調度器。基於鎖的編程取決於調度器,以確保鎖的持有者正在運行,以便他們可以釋放鎖。這是導致「優先級反轉」的原因在Linux上有一些鎖定屬性以確保發生這種情況
  2. 如果我獨立於調度程序,操作系統管理時間片的時間更容易,而且我獲得的上下文切換要少得多
  3. 使用無鎖定方法編寫正確的多線程程序比較容易,因爲我不必擔心死鎖,活鎖,調度,同步等。這尤其適用於共享內存實現,其中一個進程在共享內存中持有鎖時可能死亡,並且無法釋放鎖定鎖定自由方法更容易擴展。實際上,我已經通過網絡實現了無鎖定方法。像這樣的分佈式鎖是一場噩夢

這就是說,有很多情況下,基於鎖的方法是優選和/或更新的東西都是昂貴或不可能複製時,需要

  1. 。大多數無鎖定方法使用某種形式的版本控制,即複製對象,更新它,然後檢查共享版本是否與複製版本相同,然後使用當前版本更新版本。 Els再次拷貝它,應用更新並再次檢查。繼續這樣做直到它工作。當物體很小但是它很大或者包含文件句柄等時,這很好,那麼不建議使用
  2. 大多數類型都不可能以無鎖的方式訪問,例如,任何STL容器。這些不變式需要非原子訪問,例如assert(vector.size()== vector。端() - vector.begin())。因此,如果您正在更新/閱讀共享的矢量,則必須將其鎖定。