構(gòu)造函數(shù)
構(gòu)造函數(shù)的主要動作就是調(diào)用CreateIoCompletionPort創(chuàng)建了一個初始iocp。
Dispatch和post的區(qū)別
Post一定是PostQueuedCompletionStatus并且在GetQueuedCompletionStatus 之后執(zhí)行。
Dispatch會首先檢查當前thread是不是io_service.run/runonce/poll/poll_once線程,如果是,則直接運行。
poll和run的區(qū)別
兩 者代碼幾乎一樣,都是首先檢查是否有outstanding的消息,如果沒有直接返回,否則調(diào)用do_one()。唯一的不同是在調(diào)用size_t do_one(bool block, boost::system::error_code& ec)時前者block = false,后者block = true。
該參數(shù)的作用體現(xiàn)在:
BOOL ok = ::GetQueuedCompletionStatus(iocp_.handle, &bytes_transferred,
&completion_key, &overlapped, block ? timeout : 0);
因此可以看出,poll處理的是已經(jīng)完成了的消息,也即GetQueuedCompletionStatus立刻能返回的。而run則會導(dǎo)致等待。
poll 的作用是依次處理當前已經(jīng)完成了的消息,直到所有已經(jīng)完成的消息處理完成為止。如果沒有已經(jīng)完成了得消息,函數(shù)將退出。poll不會等待。這個函數(shù)有點類 似于PeekMessage。鑒于PeekMessage很少用到,poll的使用場景我也有點疑惑。poll的一個應(yīng)用場景是如果希望handler的 處理有優(yōu)先級,也即,如果消息完成速度很快,同時可能完成多個消息,而消息的處理過程可能比較耗時,那么可以在完成之后的消息處理函數(shù)中不真正處理數(shù)據(jù), 而是把handler保存在隊列中,然后按優(yōu)先級統(tǒng)一處理。代碼如下:
while (io_service.run_one()) {
// The custom invocation hook adds the handlers to the priority queue
// rather than executing them from within the poll_one() call.
while (io_service.poll_one()) ; pri_queue.execute_all(); }
循環(huán)執(zhí)行poll_one讓已經(jīng)完成的消息的wrap_handler處理完畢,也即插入一個隊列中,然后再統(tǒng)一處理之。這里的wrap_handler是一個class,在post的時候,用如下代碼:
io_service.post(pri_queue.wrap(0, low_priority_handler));或者 acceptor.async_accept(server_socket, pri_queue.wrap(100, high_priority_handler));
template <typename Handler> wrapped_handler<Handler> handler_priority_queue::wrap(int priority, Handler handler)
{ return wrapped_handler<Handler>(*this, priority, handler); }
參見boost_asio/example/invocation/prioritised_handlers.cpp
這個sample也同時表現(xiàn)了wrap的使用場景。
也即把handler以及參數(shù)都wrap成一個object,然后把object插入一個隊列,在pri_queue.execute_all中按優(yōu)先級統(tǒng)一處理。
run的作用是處理消息,如果有消息未完成將一直等待到所有消息完成并處理之后才退出。
reset和stop
文檔中reset的解釋是重置io_service以便下一次調(diào)用。
當 run,run_one,poll,poll_one是被stop掉導(dǎo)致退出,或者由于完成了所有任務(wù)(正常退出)導(dǎo)致退出時,在調(diào)用下一次 run,run_one,poll,poll_one之前,必須調(diào)用此函數(shù)。reset不能在run,run_one,poll,poll_one正在運 行時調(diào)用。如果是消息處理handler(用戶代碼)拋出異常,則可以在處理之后直接繼續(xù)調(diào)用 io.run,run_one,poll,poll_one。 例如:
boost::asio::io_service io_service;
...
for (;;)
{
try
{
io_service.run();
break; // run() exited normally
}
catch (my_exception& e)
{
// Deal with exception as appropriate.
}
}
在拋出了異常的情況下,stopped_還沒來得及被asio設(shè)置為1,所以無需調(diào)用reset。
reset函數(shù)的代碼僅有一行:
void reset()
{
::InterlockedExchange(&stopped_, 0);
}
也即,當io.stop時,會設(shè)置stopped_=1。當完成所有任務(wù)時,也會設(shè)置。
總的來說,單線程情況下,不管io.run是如何退出的,在下一次調(diào)用io.run之前調(diào)用一次reset沒有什么壞處。例如:
for(;;)
{
try
{
io.run();
}
catch(…)
{
}
io.reset();
}
如果是多線程在運行io.run,則應(yīng)該小心,因為reset必須是所有的run,run_one,poll,poll_one退出后才能調(diào)用。
文檔中的stop的解釋是停止io_service的處理循環(huán)。
此函數(shù)不是阻塞函數(shù),也即,它僅僅只是給iocp發(fā)送一個退出消息而并不是等待其真正退出。因為poll和poll_one本來就不等待 (GetQueuedCompletionStatus時timeout = 0),所以此函數(shù)對poll和poll_one無意義。對于run_one來說,如果該事件還未完成,則run_one會立刻返回。如果該事件已經(jīng)完成, 并且還在處理中,則stop并無特殊意義(會等待handler完成后自然退出)。對于run來說,stop的調(diào)用會導(dǎo)致run中的 GetQueuedCompletionStatus立刻返回。并且由于設(shè)置了stopped = 1,此前完成的消息的handlers也不會被調(diào)用。考慮一下這種情況:在io.stop之前,有1k個消息已經(jīng)完成但尚未處理,io.run正在依次從 GetQueuedCompletionStatus中獲得信息并且調(diào)用handlers,調(diào)用io.stop設(shè)置stopped=1將導(dǎo)致后許 GetQueuedCompletionStatus返回的消息直接被丟棄,直到收到退出消息并退出io.run為止。
void stop()
{
if (::InterlockedExchange(&stopped_, 1) == 0)
{
if (!::PostQueuedCompletionStatus(iocp_.handle, 0, 0, 0))
{
DWORD last_error = ::GetLastError();
boost::system::system_error e(
boost::system::error_code(last_error,
boost::asio::error::get_system_category()),
"pqcs");
boost::throw_exception(e);
}
}
}
注意除了讓當前代碼退出之外還有一個副作用就是設(shè)置了stopped_=1。這個副作用導(dǎo)致在stop之后如果不調(diào)用reset,所有run,run_one,poll,poll_one都將直接退出。
另一個需要注意的是,stop會導(dǎo)致所有未完成的消息以及完成了但尚未處理得消息都直接被丟棄,不會導(dǎo)致handlers倍調(diào)用。
注意這兩個函數(shù)都不會CloseHandle(iocp.handle_),那是析構(gòu)函數(shù)干的事情。
注意此處有個細節(jié):一次PostQueuedCompletionStatus僅導(dǎo)致一次 GetQueuedCompletionStatus返回,那么如果有多個thread此時都在io.run,并且block在 GetQueuedCompletionStatus時,調(diào)用io.stop將PostQueuedCompletionStatus并且導(dǎo)致一個 thread的GetQueuedCompletionStatus返回。那么其他的thread呢?進入io_service的do_one(由run 函數(shù)調(diào)用)代碼可以看到,當GetQueuedCompletionStatus返回并且發(fā)現(xiàn)是退出消息時,會再發(fā)送一次 PostQueuedCompletionStatus。代碼如下:
else
{
// Relinquish responsibility for dispatching timers. If the io_service
// is not being stopped then the thread will get an opportunity to
// reacquire timer responsibility on the next loop iteration.
if (dispatching_timers)
{
::InterlockedCompareExchange(&timer_thread_, 0, this_thread_id);
}
// The stopped_ flag is always checked to ensure that any leftover
// interrupts from a previous run invocation are ignored.
if (::InterlockedExchangeAdd(&stopped_, 0) != 0)
{
// Wake up next thread that is blocked on GetQueuedCompletionStatus.
if (!::PostQueuedCompletionStatus(iocp_.handle, 0, 0, 0))
{
last_error = ::GetLastError();
ec = boost::system::error_code(last_error,
boost::asio::error::get_system_category());
return 0;
}
ec = boost::system::error_code();
return 0;
}
}
}
Wrap
這個函數(shù)是一個語法糖。
Void func(int a);
io_service.wrap(func)(a);
相當于io_service.dispatch(bind(func,a));
可以保存io_service.wrap(func)到g,以便在稍后某些時候調(diào)用g(a);
例如:
socket_.async_read_some(boost::asio::buffer(buffer_), strand_.wrap(
boost::bind(&connection::handle_read, shared_from_this(),
boost::asio::placeholders::error,
boost::asio::placeholders::bytes_transferred)));
這是一個典型的wrap用法。注意async_read_some要求的參數(shù)是一個handler,在read_some結(jié)束后被調(diào)用。由于希望真正被調(diào)用的handle_read是串行化的,在這里再post一個消息給io_service。以上代碼類似于:
void A::func(error,bytes_transferred)
{
strand_.dispatch(boost::bind(handle_read,shared_from_this(),error,bytes_transferred);
}
socket_.async_read_some(boost::asio::buffer(buffer_), func);
注意1點:
io_service.dispatch(bind(func,a1,…an)),這里面都是傳值,無法指定bind(func,ref(a1)…an)); 所以如果要用ref語義,則應(yīng)該在傳入wrap時顯式指出。例如:
void func(int& i){i+=1;}
void main()
{
int i = 0;
boost::asio::io_service io;
io.wrap(func)(boost::ref(i));
io.run();
printf("i=%d\n");
}
當然在某些場合下,傳遞shared_ptr也是可以的(也許更好)。
從handlers拋出的異常的影響
當handlers拋出異常時,該異常會傳遞到本線程最外層的io.run,run_one,poll,poll_one,不會影響其他線程。捕獲該異常是程序員自己的責任。
例如:
boost::asio::io_service io_service;
Thread1,2,3,4()
{
for (;;)
{
try
{
io_service.run();
break; // run() exited normally
}
catch (my_exception& e)
{
// Deal with exception as appropriate.
}
}
}
Void func(void)
{
throw 1;
}
Thread5()
{
io_service.post(func);
}
注意這種情況下無需調(diào)用io_service.reset()。
這種情況下也不能調(diào)用reset,因為調(diào)用reset之前必須讓所有其他線程正在調(diào)用的io_service.run退出。(reset調(diào)用時不能有任何run,run_one,poll,poll_one正在運行)
Work
有些應(yīng)用程序希望在沒有pending的消息時,io.run也不退出。比如io.run運行于一個后臺線程,該線程在程序的異步請求發(fā)出之前就啟動了。
可以通過如下代碼實現(xiàn)這種需求:
main()
{
boost::asio::io_service io_service;
boost::asio::io_service::work work(io_service);
Create thread
Getchar();
}
Thread()
{
Io_service.run();
}
這種情況下,如果work不被析構(gòu),該線程永遠不會退出。在work不被析構(gòu)得情況下就讓其退出,可以調(diào)用io.stop。這將導(dǎo)致 io.run立刻退出,所有未完成的消息都將丟棄。已完成的消息(但尚未進入handler的)也不會調(diào)用其handler函數(shù)(由于在stop中設(shè)置了 stopped_= 1)。
如果希望所有發(fā)出的異步消息都正常處理之后io.run正常退出,work對象必須析構(gòu),或者顯式的刪除。
boost::asio::io_service io_service;
auto_ptr<boost::asio::io_service::work> work(
new boost::asio::io_service::work(io_service));
...
work.reset(); // Allow run() to normal exit.
work是一個很小的輔助類,只支持構(gòu)造函數(shù)和析構(gòu)函數(shù)。(還有一個get_io_service返回所關(guān)聯(lián)的io_service)
代碼如下:
inline io_service::work::work(boost::asio::io_service& io_service)
: io_service_(io_service)
{
io_service_.impl_.work_started();
}
inline io_service::work::work(const work& other)
: io_service_(other.io_service_)
{
io_service_.impl_.work_started();
}
inline io_service::work::~work()
{
io_service_.impl_.work_finished();
}
void work_started()
{
::InterlockedIncrement(&outstanding_work_);
}
// Notify that some work has finished.
void work_finished()
{
if (::InterlockedDecrement(&outstanding_work_) == 0)
stop();
}
可以看出構(gòu)造一個work時,outstanding_work_+1,使得io.run在完成所有異步消息后判斷outstanding_work_時不會為0,因此會繼續(xù)調(diào)用GetQueuedCompletionStatus并阻塞在這個函數(shù)上。
而析構(gòu)函數(shù)中將其-1,并判斷其是否為0,如果是,則post退出消息給GetQueuedCompletionStatus讓其退出。
因此work如果析構(gòu),則io.run會在處理完所有消息之后正常退出。work如果不析構(gòu),則io.run會一直運行不退出。如果用戶直接調(diào)用io.stop,則會讓io.run立刻退出。
特別注意的是,work提供了一個拷貝構(gòu)造函數(shù),因此可以直接在任意地方使用。對于一個io_service來說,有多少個work實例關(guān) 聯(lián),則outstanding_work_就+1了多少次,只有關(guān)聯(lián)到同一個io_service的work全被析構(gòu)之后,io.run才會在所有消息處 理結(jié)束之后正常退出。
strand
strand是另一個輔助類,提供2個接口dispatch和post,語義和io_service的dispatch和post類似。區(qū) 別在于,同一個strand所發(fā)出的dispatch和post絕對不會并行執(zhí)行,dispatch和post所包含的handlers也不會并行。因此 如果希望串行處理每一個tcp連接,則在accept之后應(yīng)該在該連接的數(shù)據(jù)結(jié)構(gòu)中構(gòu)造一個strand,并且所有dispatch/post(recv /send)操作都由該strand發(fā)出。strand的作用巨大,考慮如下場景:有多個thread都在執(zhí)行async_read_some,那么由于 線程調(diào)度,很有可能后接收到的包先被處理,為了避免這種情況,就只能收完數(shù)據(jù)后放入一個隊列中,然后由另一個線程去統(tǒng)一處理。
void connection::start()
{
socket_.async_read_some(boost::asio::buffer(buffer_),
strand_.wrap(
boost::bind(&connection::handle_read, shared_from_this(),
boost::asio::placeholders::error,
boost::asio::placeholders::bytes_transferred)));
}
不使用strand的處理方式:
前端tcp iocp收包,并且把同一個tcp連接的包放入一個list,如果list以前為空,則post一個消息給后端vnn iocp。后端vnn iocp收到post的消息后循環(huán)從list中獲取數(shù)據(jù),并且處理,直到list為空為止。處理結(jié)束后重新調(diào)用 GetQueuedCompletionStatus進入等待。如果前端tcp iocp發(fā)現(xiàn)list過大,意味著處理速度小于接收速度,則不再調(diào)用iocpRecv,并且設(shè)置標志,當vnn iocp thread處理完了當前所有積壓的數(shù)據(jù)包后,檢查這個標志,重新調(diào)用一次iocpRecv。
使用strand的處理方式:
前端tcp iocp收包,收到包后直接通過strand.post(on_recved)發(fā)給后端vnn iocp。后端vnn iocp處理完之后再調(diào)用一次strand.async_read_some。
這兩種方式我沒看出太大區(qū)別來。如果對數(shù)據(jù)包的處理的確需要阻塞操作,例如db query,那么使用后端iocp以及后端thread是值得考慮的。這種情況下,前端iocp由于僅用來異步收發(fā)數(shù)據(jù),因此1個thread就夠了。在 確定使用2級iocp的情況下,前者似乎更為靈活,也沒有增加什么開銷。
值得討論的是,如果后端多個thread都處于db query狀態(tài),那么實際上此時依然沒有thread可以提供數(shù)據(jù)處理服務(wù),因此2級iocp意義其實就在于在這種情況下,前端tcp iocp依然可以accept,以及recv第一次數(shù)據(jù),不會導(dǎo)致用戶connect不上的情況。在后端thread空閑之后會處理這期間的recv到的 數(shù)據(jù)并在此async_read_some。
如果是單級iocp(假定handlers沒有阻塞操作),多線程,那么strand的作用很明顯。這種情況下,很明顯應(yīng)該讓一個tcp連接的數(shù)據(jù)處理過程串行化。
Strand的實現(xiàn)原理
Strand內(nèi)部實現(xiàn)機制稍微有點復(fù)雜。每次發(fā)出strand請求(例如 async_read(strand_.wrap(funobj1))),strand再次包裹了一次成為funobj2。在async_read完成 時,系統(tǒng)調(diào)用funobj2,檢查是否正在執(zhí)行該strand所發(fā)出的完成函數(shù)(檢查該strand的一個標志位),如果沒有,則直接調(diào)用 funobj2。如果有,則檢查是否就是當前thread在執(zhí)行,如果是,則直接調(diào)用funobj2(這種情況可能發(fā)生在嵌套調(diào)用的時候,但并不產(chǎn)生同步 問題,就像同一個thread可以多次進入同一個critical_session一樣)。如果不是,則把該funobj2插入到strand內(nèi)部維護的 一個隊列中。