簡(jiǎn)介： Boost C++ 庫(kù)讓并發(fā)編程變得既簡(jiǎn)單又有趣。學(xué)習(xí)如何使用兩個(gè) Boost 庫(kù) —— Interprocess (IPC) 庫(kù)和 Message Passing Interface (MPI) 實(shí)現(xiàn)共享內(nèi)存對(duì)象、同步文件鎖和分布式通信等功能。
使用非常流行的 Boost 庫(kù)進(jìn)行并發(fā)編程非常有意思。Boost 有幾個(gè)用于并發(fā)編程領(lǐng)域的庫(kù)：Interprocess (IPC) 庫(kù)用于實(shí)現(xiàn)共享內(nèi)存、內(nèi)存映射的 I/O 和消息隊(duì)列；Thread 庫(kù)用于實(shí)現(xiàn)可移植的多線程；Message Passing Interface (MPI) 庫(kù)用于分布式計(jì)算中的消息傳遞；Asio 庫(kù)用于使用套接字和其他低層功能實(shí)現(xiàn)可移植的連網(wǎng)功能。本文介紹 IPC 和 MPI 庫(kù)以及它們提供的一些功能。
本文中將學(xué)習(xí)如何使用 Boost IPC 庫(kù)實(shí)現(xiàn)共享內(nèi)存對(duì)象、消息隊(duì)列和同步文件鎖。通過(guò)使用 Boost MPI 庫(kù)，了解 environment 和communicator 類(lèi)，以及如何實(shí)現(xiàn)分布式通信。
注意：本文中的代碼已經(jīng)用 gcc-4.3.4 和 boost-1.45 包測(cè)試過(guò)了。
#include <boost/interprocess/shared_memory_object.hpp>using namespace boost::interprocess; //… your sources follow …
在把信息傳遞給編譯器時(shí)，您要求進(jìn)程根據(jù)安裝相應(yīng)地修改 include 路徑。然后，編譯代碼：
bash-4.1$ g++ ipc1.cpp –I../boost_1_45_0
我們先從傳統(tǒng)的 "Hello World!" 程序開(kāi)始。有兩個(gè)進(jìn)程：第一個(gè)進(jìn)程把字符串 "Hello World!" 寫(xiě)入內(nèi)存，另一個(gè)進(jìn)程讀取并顯示此字符串。像
清單 2 這樣創(chuàng)建共享內(nèi)存對(duì)象。
#include <boost/interprocess/shared_memory_object.hpp>int main(int argc, char* argv[ ]) { using namespace using boost::interprocess; try { // creating our first shared memory object. shared_memory_object sharedmem1 (create_only, "Hello", read_write); // setting the size of the shared memory sharedmem1.truncate (256); // … more code follows } catch (interprocess_exception& e) { // .. . clean up } }
sharedmem1 對(duì)象的類(lèi)型是 shared_memory_object（在 Boost 頭文件中聲明并定義），它的構(gòu)造函數(shù)有三個(gè)參數(shù)：
第一個(gè)參數(shù) — create_only — 表示要?jiǎng)?chuàng)建這個(gè)共享內(nèi)存對(duì)象而且還沒(méi)有創(chuàng)建它。如果已經(jīng)存在同名的共享對(duì)象，就會(huì)拋出異常。對(duì)于希望訪問(wèn)已經(jīng)創(chuàng)建的共享內(nèi)存的進(jìn)程，第一個(gè)參數(shù)應(yīng)該是 open_only。
第二個(gè)參數(shù) — Hello — 是共享內(nèi)存區(qū)域的名稱(chēng)。另一個(gè)進(jìn)程將使用這個(gè)名稱(chēng)訪問(wèn)這個(gè)共享內(nèi)存。
第三個(gè)參數(shù) — read_write — 是共享內(nèi)存對(duì)象的訪問(wèn)指示符。因?yàn)檫@個(gè)進(jìn)程要修改共享內(nèi)存對(duì)象的內(nèi)容，所以使用read_write。只從共享內(nèi)存讀取數(shù)據(jù)的進(jìn)程使用 read_only 指示符。
truncate 方法以字節(jié)為單位設(shè)置共享內(nèi)存的大小。最好把代碼放在 try-catch 塊中。例如，如果無(wú)法創(chuàng)建共享內(nèi)存對(duì)象，就拋出類(lèi)型為 boost::interprocess_exception 的異常。
使用共享內(nèi)存對(duì)象的進(jìn)程必須在自己的地址空間中映射對(duì)象。使用在頭文件 mapped_region.hpp 中聲明并定義的 mapped_region 類(lèi)執(zhí)行映射。使用 mapped_region 的另一個(gè)好處是可以對(duì)共享內(nèi)存對(duì)象進(jìn)行完全和部分訪問(wèn)。
清單 3 演示如何使用 mapped_region。
#include <boost/interprocess/shared_memory_object.hpp>#include <boost/interprocess/mapped_region.hpp>int main(int argc, char* argv[ ]) { using namespace boost::interprocess; try { // creating our first shared memory object. shared_memory_object sharedmem1 (create_only, "Hello", read_write); // setting the size of the shared memory sharedmem1.truncate (256); // map the shared memory to current process mapped_region mmap (sharedmem1, 256); // access the mapped region using get_address std::strcpy(static_cast<char* >(region.get_address()), "Hello World!\n"); } catch (interprocess_exception& e) { // .. . clean up } }
就這么簡(jiǎn)單?，F(xiàn)在已經(jīng)創(chuàng)建了您自己的 mapped_region 對(duì)象并使用 get_address 方法訪問(wèn)了它。執(zhí)行了 static_cast，因?yàn)間et_address 返回一個(gè) void*。
當(dāng)主進(jìn)程退出時(shí)，并不刪除共享內(nèi)存。要想刪除共享內(nèi)存，需要調(diào)用 shared_memory_object::remove。第二個(gè)進(jìn)程的訪問(wèn)機(jī)制也很簡(jiǎn)單：
清單 4 證明了這一點(diǎn)。
#include <boost/interprocess/shared_memory_object.hpp>#include <boost/interprocess/mapped_region.hpp>#include <cstring>#include <cstdlib>#include <iostream>int main(int argc, char *argv[ ]){ using namespace boost::interprocess; try { // opening an existing shared memory object shared_memory_object sharedmem2 (open_only, "Hello", read_only); // map shared memory object in current address space mapped_region mmap (sharedmem2, read_only); // need to type-cast since get_address returns void* char *str1 = static_cast<char*> (mmap.get_address()); std::cout << str1 << std::endl; } catch (interprocess_exception& e) { std::cout << e.what( ) << std::endl; } return 0;}
在清單 4 中，使用 open_only 和 read_only 屬性創(chuàng)建共享內(nèi)存對(duì)象。如果無(wú)法找到這個(gè)共享內(nèi)存對(duì)象，就會(huì)拋出異?！，F(xiàn)在，構(gòu)建并運(yùn)行
清單 3 和
清單 4 中的代碼。應(yīng)該會(huì)在終端上看到 "Hello World!"。
接下來(lái)，在第二個(gè)進(jìn)程的代碼（
清單 4）中 std::cout 后面添加以下代碼并重新構(gòu)建代碼：
// std::cout code hereshared_memory_object::remove("Hello");// } catch(interprocess_exception& e) {
連續(xù)執(zhí)行代碼兩次，第二次執(zhí)行會(huì)顯示 "No such file or directory"，這證明共享內(nèi)存已經(jīng)被刪除了。
回頁(yè)首現(xiàn)在，研究另一種流行的進(jìn)程間通信機(jī)制：消息隊(duì)列。每個(gè)參與通信的進(jìn)程都可以在隊(duì)列中添加消息和從隊(duì)列讀取消息。消息隊(duì)列具有以下性質(zhì)：
它有名稱(chēng)，進(jìn)程使用名稱(chēng)訪問(wèn)它。
在創(chuàng)建隊(duì)列時(shí)，用戶(hù)必須指定隊(duì)列的最大長(zhǎng)度和一個(gè)消息的最大大小。
隊(duì)列是持久的，這意味著當(dāng)創(chuàng)建它的進(jìn)程死亡之后它仍然留在內(nèi)存中?？梢酝ㄟ^(guò)顯式地調(diào)用boost::interprocess::message_queue::remove 刪除隊(duì)列。
在
清單 5 所示的代碼片段中，進(jìn)程創(chuàng)建了一個(gè)可包含 20 個(gè)整數(shù)的消息隊(duì)列。
#include <boost/interprocess/ipc/message_queue.hpp>#include <iostream> int main(int argc, char* argv[ ]) { using namespace boost::interprocess; try { // creating a message queue message_queue mq (create_only, // only create "mq", // name 20, //max message count sizeof(int) //max message size ); // … more code follows } catch (interprocess_exception& e) { std::cout << e.what( ) << std::endl; } }
注意傳遞給 message_queue 的構(gòu)造函數(shù)的 create_only 屬性。與共享內(nèi)存對(duì)象相似，對(duì)于以只讀方式打開(kāi)消息隊(duì)列，應(yīng)該把open_only 屬性傳遞給構(gòu)造函數(shù)。
在發(fā)送方，使用隊(duì)列的 send 方法添加數(shù)據(jù)。send 方法有三個(gè)輸入?yún)?shù)：原始數(shù)據(jù)的指針 (void*)、數(shù)據(jù)的大小和優(yōu)先級(jí)。目前，以相同的優(yōu)先級(jí)發(fā)送所有數(shù)據(jù)。
清單 6 給出代碼。
#include <boost/interprocess/ipc/message_queue.hpp>#include <iostream> int main(int argc, char* argv[ ]) { using namespace boost::interprocess; try { // creating a message queue message_queue mq (create_only, // only create "mq", // name 20, //max message count sizeof(int) //max message size ); // now send the messages to the queue for (int i=0; i<20; ++i) mq.send(&i, sizeof(int), 0); // the 3rd argument is the priority } catch (interprocess_exception& e) { std::cout << e.what( ) << std::endl; } }
在接收方，使用 open_only 屬性創(chuàng)建隊(duì)列。通過(guò)調(diào)用 message_queue 類(lèi)的 receive 方法從隊(duì)列獲取消息。
清單 7 給出 receive 的方法簽名。
void receive (void *buffer, std::size_t buffer_size, std::size_t &recvd_size, unsigned int &priority );
我們來(lái)仔細(xì)看一下。第一個(gè)參數(shù)是從隊(duì)列接收的數(shù)據(jù)將被存儲(chǔ)到的位置。第二個(gè)參數(shù)是接收的數(shù)據(jù)的預(yù)期大小。第三個(gè)參數(shù)是接收的數(shù)據(jù)的實(shí)際大小。第四個(gè)參數(shù)是接收的消息的優(yōu)先級(jí)。顯然，如果在執(zhí)行程序期間第二個(gè)和第三個(gè)參數(shù)不相等，就是出現(xiàn)錯(cuò)誤了。
清單 8 給出接收者進(jìn)程的代碼。
#include <boost/interprocess/ipc/message_queue.hpp>#include <iostream> int main(int argc, char* argv[ ]) { using namespace boost::interprocess; try { // opening the message queue whose name is mq message_queue mq (open_only, // only open "mq" // name ); size_t recvd_size; unsigned int priority; // now send the messages to the queue for (int i=0; i<20; ++i) { int buffer; mq.receive ((void*) &buffer, sizeof(int), recvd_size, priority); if (recvd_size != sizeof(int)) ; // do the error handling std::cout << buffer << " " << recvd_size << " " << priority; } } catch (interprocess_exception& e) { std::cout << e.what( ) << std::endl; } }
這相當(dāng)簡(jiǎn)單。注意，仍然沒(méi)有從內(nèi)存中刪除消息隊(duì)列；與共享內(nèi)存對(duì)象一樣，這個(gè)隊(duì)列是持久的。要想刪除隊(duì)列，應(yīng)該在使用完隊(duì)列之后添加以下行：
message_queue::remove("mq"); // remove the queue using its name
在發(fā)送方，做
清單 9 所示的修改。接收方代碼不需要修改。
message_queue::remove("mq"); // remove the old queue message_queue mq (…); // create as before for (int i=0; i<20; ++i) mq.send(&i, sizeof(int), i%2); // 第 3 個(gè)參數(shù)為消息的優(yōu)先級(jí) // … rest as usual
再次運(yùn)行代碼時(shí)，應(yīng)該會(huì)看到
清單 10 所示的輸出。
1 4 13 4 15 4 17 4 19 4 111 4 113 4 115 4 117 4 119 4 10 4 02 4 04 4 06 4 08 4 010 4 012 4 014 4 016 4 018 4 0
清單 10 證實(shí)，第二個(gè)進(jìn)程優(yōu)先接收優(yōu)先級(jí)高的消息。
回頁(yè)首共享內(nèi)存和消息隊(duì)列很不錯(cuò)，但是文件 I/O 也是重要的進(jìn)程間通信工具。對(duì)并發(fā)進(jìn)程用于通信的文件訪問(wèn)進(jìn)行同步并非易事，但是 Boost IPC 庫(kù)提供的文件鎖功能讓同步變得簡(jiǎn)單了。在進(jìn)一步解釋之前，來(lái)看一下
清單 11，了解 file_lock 對(duì)象是如何工作的。
#include <fstream> #include <iostream> #include <boost/interprocess/sync/file_lock.hpp> #include <cstdlib>int main() { using namespace boost::interprocess; std::string fileName("test"); std::fstream file; file.open(fileName.c_str(), std::ios::out | std::ios::binary | std::ios::trunc); if (!file.is_open() || file.bad()) { std::cout << "Open failed" << std::endl; exit(-1); } try { file_lock f_lock(fileName.c_str()); f_lock.lock(); std::cout << "Locked in Process 1" << std::endl; file.write("Process 1", 9); file.flush(); f_lock.unlock(); std::cout << "Unlocked from Process 1" << std::endl; } catch (interprocess_exception& e) { std::cout << e.what( ) << std::endl; } file.close(); return 0; }
代碼首先打開(kāi)一個(gè)文件，然后使用 file_lock 鎖定它。寫(xiě)操作完成之后，它刷新文件緩沖區(qū)并解除文件鎖。使用 lock 方法獲得對(duì)文件的獨(dú)占訪問(wèn)。如果另一個(gè)進(jìn)程也試圖對(duì)此文件進(jìn)行寫(xiě)操作并已經(jīng)請(qǐng)求了鎖，那么它會(huì)等待，直到第一個(gè)進(jìn)程使用 unlock 自愿地放棄鎖。file_lock 類(lèi)的構(gòu)造函數(shù)接受要鎖定的文件的名稱(chēng)，一定要在調(diào)用 lock 之前打開(kāi)文件；否則會(huì)拋出異常。
現(xiàn)在，復(fù)制
清單 11 中的代碼并做一些修改。具體地說(shuō)，讓第二個(gè)進(jìn)程請(qǐng)求這個(gè)鎖。
清單 12 給出相關(guān)修改。
// .. as in Listing 11 file_lock f_lock(fileName.c_str()); f_lock.lock(); std::cout << "Locked in Process 2" << std::endl; system("sleep 4"); file.write("Process 2", 9); file.flush(); f_lock.unlock(); std::cout << "Unlocked from Process 2" << std::endl; // file.close();
現(xiàn)在，如果這兩個(gè)進(jìn)程同時(shí)運(yùn)行，有 50% 的機(jī)會(huì)看到第一個(gè)進(jìn)程等待 4 秒后才獲得 file_lock，其他情況都不變。
在使用 file_lock 時(shí)，必須記住幾點(diǎn)。這里討論的主題是進(jìn)程間通信，重點(diǎn)在進(jìn)程 上。這意味著，不是使用 file_lock 來(lái)同步同一進(jìn)程中各個(gè)線程的數(shù)據(jù)訪問(wèn)。在與 POSIX 兼容的系統(tǒng)上，文件句柄是進(jìn)程屬性，而不是 線程屬性。下面是使用文件鎖的幾條規(guī)則：
對(duì)于每個(gè)進(jìn)程，每個(gè)文件使用一個(gè) file_lock 對(duì)象。
使用相同的線程來(lái)鎖定和解鎖文件。
在解鎖文件之前，通過(guò)調(diào)用 C 的 flush 庫(kù)例程或 flush 方法（如果喜歡使用 C++ fstream 的話），刷新寫(xiě)入者進(jìn)程中的數(shù)據(jù)。
在執(zhí)行程序時(shí)，可能會(huì)出現(xiàn)拋出異常而文件沒(méi)有解鎖的情況。這種情況可能會(huì)導(dǎo)致意外的程序行為。為了避免這種情況，可以考慮把file_lock 對(duì)象放在（boost/interprocess/sync/scoped_lock.hpp 中定義的）scoped_lock 中。如果使用 scoped_lock，就不需要顯式地鎖定或解鎖文件；鎖定發(fā)生在構(gòu)造器內(nèi)，每當(dāng)您退出該范圍，就會(huì)自動(dòng)發(fā)生解鎖。
清單 13 給出對(duì)
清單 11 的修改，使之使用有范圍的鎖。
#include <boost/interprocess/sync/scoped_lock.hpp>#include <boost/interprocess/sync/file_lock.hpp>//… code as in Listing 11file_lock f_lock(fileName.c_str());scoped_lock<file_lock> s_lock(f_lock); // internally calls f_lock.lock( ); // No need to call explicit lock anymorestd::cout << "Locked in Process 1" << std::endl;file.write("Process 1", 9);// … code as in Listing 11
注意：關(guān)于 Resource Acquisition Is Initialization (RAII) 編程習(xí)慣法的更多信息，參見(jiàn)
參考資料 中的鏈接。
回頁(yè)首如果您不熟悉 Message Passing Interface，那么在討論 Boost MPI 之前，應(yīng)該先瀏覽
參考資料 中提供的 MPI 參考資料。MPI 是一個(gè)容易使用的標(biāo)準(zhǔn)，它采用通過(guò)傳遞消息實(shí)現(xiàn)進(jìn)程間通信的模型。不需要使用套接字或其他通信原語(yǔ)；MPI 后端管理所有底層處理。那么，使用 Boost MPI 有什么好處？Boost MPI 的創(chuàng)建者提供了更高層的抽象，并在 MPI 提供的 API 之上構(gòu)建了一套簡(jiǎn)單的例程，比如 MPI_Init 和 MPI_Bcast。
Boost MPI 不是一個(gè)單獨(dú)的庫(kù)，不能在下載和構(gòu)建之后直接使用。相反，必須安裝任意 MPI 實(shí)現(xiàn)（比如 MPICH 或 Open MPI）并構(gòu)建 Boost Serialization 庫(kù)。關(guān)于如何構(gòu)建 Boost MPI 的詳細(xì)信息參見(jiàn)
參考資料。通常，使用以下命令構(gòu)建 Boost MPI：
bash-4.1$ bjam –with-mpi
Windows? 用戶(hù)可以從 BoostPro 下載預(yù)先構(gòu)建的 MPI 庫(kù)（見(jiàn)
參考資料）。這些庫(kù)與 Microsoft? HPC Pack 2008 和 2008 R2 兼容（見(jiàn)
參考資料），適用于帶 Service Pack 3 的 Windows XP 或更高版本的客戶(hù)機(jī)操作環(huán)境。
回頁(yè)首您必須了解 Boost MPI 庫(kù)中的兩個(gè)主要類(lèi)：environment 類(lèi)和 communicator 類(lèi)。前者負(fù)責(zé)分布式環(huán)境的初始化；后者用于進(jìn)程之間的通信。因?yàn)檫@里討論的是分布式計(jì)算，我們有四個(gè)進(jìn)程，它們都在終端上輸出 "Hello World"。
清單 14 給出代碼。
#include <boost/mpi.hpp>#include <iostream>int main(int argc, char* argv[]){ boost::mpi::environment env(argc, argv); boost::mpi::communicator world; std::cout << argc << std::endl; std::cout << argv[0] << std::endl; std::cout << "Hello World! from process " << world.rank() << std::endl; return 0;}
現(xiàn)在，構(gòu)建
清單 14 中的代碼并鏈接 Boost MPI 和 Serialization 庫(kù)。在 shell 提示上運(yùn)行可執(zhí)行程序。應(yīng)該會(huì)看到 "Hello World! from process 0"。接下來(lái)，使用 MPI 分派器工具（例如，對(duì)于 Open MPI 用戶(hù)，使用 mpirun；對(duì)于 Microsoft HPC Pack 2008，使用mpiexec）并運(yùn)行可執(zhí)行程序：
mpirun –np 4 <executable name> ORmpiexec –n 4 <executable name>
現(xiàn)在應(yīng)該會(huì)看到與
清單 15 相似的輸出，其中的 mympi1 是可執(zhí)行程序名稱(chēng)。
1mympi1Hello, World! from process 31mympi11mympi1Hello, World! from process 1Hello, World! from process 21mympi1Hello, World! from process 0
在 MPI 框架中，已經(jīng)創(chuàng)建了相同進(jìn)程的四個(gè)拷貝。在 MPI 環(huán)境中，每個(gè)進(jìn)程有惟一的 ID（由 communicator 對(duì)象決定）。現(xiàn)在，試試進(jìn)程間通信。使用 send 和 receive 函數(shù)調(diào)用讓一個(gè)進(jìn)程與另一個(gè)進(jìn)程通信。發(fā)送消息的進(jìn)程稱(chēng)為主進(jìn)程，接收消息的進(jìn)程稱(chēng)為工作者進(jìn)程。主進(jìn)程和接收者進(jìn)程的源代碼是相同的，使用 world 對(duì)象提供的等級(jí)決定功能（見(jiàn)
清單 16）。
#include <boost/mpi.hpp>#include <iostream>int main(int argc, char* argv[]) { boost::mpi::environment env(argc, argv); boost::mpi::communicator world; if (world.rank() == 0) { world.send(1, 9, 32); world.send(2, 9, 33); } else { int data; world.recv(0, 9, data); std::cout << "In process " << world.rank( ) << "with data " << data << std::endl; } return 0;}
先看一下 send 函數(shù)。第一個(gè)參數(shù)是接收者進(jìn)程的 ID；第二個(gè)是消息數(shù)據(jù)的 ID；第三個(gè)是實(shí)際數(shù)據(jù)。為什么需要消息標(biāo)簽？接收者進(jìn)程在執(zhí)行期間的特定點(diǎn)上可能希望處理具有特定標(biāo)簽的消息，所以這個(gè)方案會(huì)有幫助。對(duì)于進(jìn)程 1 和 2，recv 函數(shù)被阻塞，這意味著程序會(huì)等待，直到從進(jìn)程 0 收到標(biāo)簽 ID 為 9 的消息。當(dāng)收到這個(gè)消息時(shí)，把信息存儲(chǔ)在 data 中。下面是運(yùn)行代碼的輸出：
In process 1 with data 32In process 2 with data 33
如果在接收方有 world.recv(0, 1, data); 這樣的代碼，會(huì)發(fā)生什么？代碼阻塞，但實(shí)際上是，接收者進(jìn)程在等待一個(gè)永遠(yuǎn)不會(huì)到達(dá)的消息。
回頁(yè)首本文只討論了這兩個(gè)庫(kù)提供的功能的很小一部分。這些庫(kù)提供的其他功能包括 IPC 的內(nèi)存映射 I/O 和 MPI 的廣播功能。從易用性的角度來(lái)說(shuō)，IPC 更好。MPI 庫(kù)依賴(lài)于原生的 MPI 實(shí)現(xiàn)，而原生 MPI 庫(kù)以及預(yù)先構(gòu)建的 Boost MPI 和 Serialization 庫(kù)的現(xiàn)成可用性仍然是個(gè)問(wèn)題。但是，花點(diǎn)兒精力構(gòu)建 MPI 實(shí)現(xiàn)和 Boost 的源代碼是值得的。
