并發(fā)編程高性能計算【免費下載鏈接】oneTBBoneAPI Threading Building Blocks (oneTBB)項目地址https://gitcode.com/gh_mirrors/on/oneTBB點擊查看免費下載本篇技術(shù)指南圍繞 oneAPI Threading Building BlocksoneTBB的 Flow Graph 接口展開講解如何用**節(jié)點node與邊edge**表達數(shù)據(jù)流圖與依賴圖兩種典型的圖并行應(yīng)用前者讓消息沿邊流動、逐節(jié)點加工后者用continue_msg在節(jié)點之間建立部分排序。讀完本文你將掌握function_node、input_node、continue_node、broadcast_node等預(yù)定義節(jié)點類型的構(gòu)造與連接方式理解并發(fā)限制concurrency limit如何控制任務(wù)生成并能獨立寫出可運行的圖并行程序。圖并行用節(jié)點和邊描述計算除了循環(huán)并行l(wèi)oop parallelismoneTBB 還支持圖并行g(shù)raph parallelism。通過 Flow Graph 接口你可以構(gòu)建高度可擴展的圖但也可能構(gòu)建出完全串行執(zhí)行的圖——是否并行、并行程度如何取決于你如何連接節(jié)點與設(shè)置并發(fā)參數(shù)。圖并行的核心模型是節(jié)點node代表一個計算單元通常攜帶一個用戶定義的 body 對象函數(shù)對象或 lambda 表達式邊edge代表節(jié)點之間的通信通道消息message沿邊在節(jié)點之間流動的數(shù)據(jù)任務(wù)task當節(jié)點收到消息時運行時庫生成一個任務(wù)來對這條消息執(zhí)行其 body 對象。try_put將消息送入節(jié)點節(jié)點處理完后其輸出會沿邊自動傳遞給后繼節(jié)點。消息在圖中沿連接節(jié)點的邊不斷流動最終由終端節(jié)點匯總結(jié)果。這種圖并行支持定義在oneapi::tbb::flow命名空間中接口聲明位于頭文件 flow_graph.h。兩類圖應(yīng)用數(shù)據(jù)流圖與依賴圖Flow Graph 接口可以表達兩類典型的圖應(yīng)用流式/數(shù)據(jù)流data flow應(yīng)用一個值序列依次流過圖中的各個節(jié)點每個節(jié)點對到達的值做變換。例如函數(shù) F 生成序列中的每個值G 對值求平方H 對值求立方J 將平方和立方的結(jié)果累加到全局和。當序列中所有值處理完畢sum等于 1 到 10 的平方與立方之和。在這類圖中值真正沿邊流動——一個節(jié)點的輸出就是其后繼節(jié)點的輸入。依賴dependence應(yīng)用用圖來建立步驟之間的部分排序。例如制作花生醬果凍三明治必須先拿到面包才能往面包上抹花生醬或果凍必須先抹花生醬才能收好花生醬罐必須先抹果凍才能收好果凍罐必須兩面都抹好才能把兩片面包合在一起。這是一個部分排序——先抹花生醬還是先抹果凍沒有關(guān)系先做完三明治再收罐子也沒有關(guān)系。在依賴圖中資源面包、果凍罐的共享只是隱含的圖中顯式表達的只有步驟之間要求的執(zhí)行順序例如“往一片面包上抹果凍”必須在“收好果凍罐”之前完成。Flow Graph 接口不僅能表達上述兩類圖還能表達更復(fù)雜的圖含環(huán)cycles、條件分支conditionals、緩沖buffering等。當你的應(yīng)用以圖的形式表達后運行時庫會自動生成任務(wù)充分挖掘圖中蘊含的并行性。例如在平方/立方示例中兩個不同的值可能被并行平方同一個值也可能被并行平方與立方在三明治示例中往一片面包抹花生醬可與往另一片面包抹果凍并行執(zhí)行。接口只聲明“哪些操作可以合法地并行執(zhí)行”運行時庫負責(zé)在運行時決定“哪些操作實際被并行執(zhí)行”。圖簡單數(shù)據(jù)流圖——值從生成節(jié)點流出經(jīng)平方/立方節(jié)點變換最終累加到全局和圖制作三明治的依賴圖——邊只表達步驟之間的順序約束不傳遞實際數(shù)據(jù)數(shù)據(jù)流圖實戰(zhàn)平方、立方與求和文檔 Data_Flow_Graph.rst 給出了“簡單數(shù)據(jù)流圖”的完整實現(xiàn)。圖中最左側(cè)的節(jié)點生成 1 到 10 的整數(shù)并分發(fā)給兩個后繼一個對每個值求平方并向下游傳遞另一個對每個值求立方并向下游傳遞最右側(cè)的節(jié)點同時接收兩個中間節(jié)點的輸出把每個值累加到運行總和sum中。程序運行結(jié)束后sum等于 1 到 10 的平方與立方之和。第一版實現(xiàn)兩個function_node與一個求和節(jié)點int sum 0; graph g; function_node int, int squarer( g, unlimited, [](const int v) { return v*v; } ); function_node int, int cuber( g, unlimited, [](const int v) { return v*v*v; } ); function_node int, int summer( g, 1, - int { return sum v; } ); make_edge( squarer, summer ); make_edge( cuber, summer ); for ( int i 1; i 10; i ) { squarer.try_put(i); cuber.try_put(i); } g.wait_for_all(); cout Sum is sum \n;該實現(xiàn)創(chuàng)建了三個function_nodesquarer對輸入值求平方cuber對輸入值求立方summer把每個到達的值累加到全局變量sum。由于squarer和cuber是**無副作用side-effect free**的創(chuàng)建時使用unlimited并發(fā)限制允許多個調(diào)用并行執(zhí)行而summer通過引用更新全局變量sum并行執(zhí)行不安全因此使用并發(fā)限制1保證其調(diào)用嚴格串行。原圖里的生成節(jié)點 F 在這里被實現(xiàn)為一個循環(huán)向squarer和cuber各try_put一次。第二版改進引入broadcast_node合并兩條輸入路徑一個輕微的改進是引入broadcast_node——它把收到的任何消息廣播給所有后繼節(jié)點從而把循環(huán)里的兩次try_put合并為一次broadcast_nodeint b(g); make_edge( b, squarer ); make_edge( b, cuber ); for ( int i 1; i 10; i ) { b.try_put(i); } g.wait_for_all();第三版優(yōu)化用input_node替代顯式循環(huán)更好的方案是引入input_node。顧名思義input_node只發(fā)送消息、不接收消息其構(gòu)造函數(shù)接受兩個參數(shù)template typename Body input_node( graph g, Body body)body 是一個函數(shù)對象或 lambda 表達式其函數(shù)運算符簽名如下Output Body::operator()( oneapi::tbb::flow_control fc );運行時庫會反復(fù)調(diào)用src_body中的函數(shù)運算符直到 body 內(nèi)部調(diào)用fc.stop()為止。你可以用input_node替換示例中的生成循環(huán)input_node int src( g, src_body(10) ); make_edge( src, squarer ); make_edge( src, cuber ); src.activate(); g.wait_for_all();注意input_node創(chuàng)建后需要調(diào)用src.activate()才會開始生成消息。完整的最終實現(xiàn)如下class src_body { const int my_limit; int my_next_value; public: src_body(int l) : my_limit(l), my_next_value(1) {} int operator()( oneapi::tbb::flow_control fc ) { if ( my_next_value my_limit ) { return my_next_value; } else { fc.stop(); return int(); } } }; int main() { int sum 0; graph g; function_node int, int squarer( g, unlimited, [](const int v) { return v*v; } ); function_node int, int cuber( g, unlimited, [](const int v) { return v*v*v; } ); function_node int, int summer( g, 1, - int { return sum v; } ); make_edge( squarer, summer ); make_edge( cuber, summer ); input_node int src( g, src_body(10) ); make_edge( src, squarer ); make_edge( src, cuber ); src.activate(); g.wait_for_all(); cout Sum is sum \n; }這個最終版本完整保留了原數(shù)據(jù)流圖中的所有節(jié)點和邊。在這個簡單示例里用input_node相比顯式循環(huán)并無明顯優(yōu)勢但input_node能感知下游節(jié)點的行為在更復(fù)雜的圖中可以限制內(nèi)存使用可參考 create_token_based_system.rst 中基于令牌的系統(tǒng)設(shè)計。從源碼實現(xiàn)看flow_control是input_node與運行時協(xié)作的關(guān)鍵機制在 flow_graph.h 的input_node內(nèi)部實現(xiàn)中運行時先調(diào)用(*my_body)(control)再根據(jù)control.is_pipeline_stopped判斷本次輸出是否有效、是否繼續(xù)下一次調(diào)用——這與文檔中“反復(fù)調(diào)用直到fc.stop()”的行為完全對應(yīng)。依賴圖實戰(zhàn)continue_node與部分排序依賴圖是數(shù)據(jù)流圖的一個特例節(jié)點之間傳遞的消息類型是oneapi::tbb::flow::continue_msg。與一般數(shù)據(jù)流圖不同依賴圖中的節(jié)點不會為每條收到的消息都生成一個任務(wù)它們知道自己有多少個前驅(qū)會計數(shù)收到的消息只有當計數(shù)等于前驅(qū)總數(shù)時才生成一個任務(wù)來執(zhí)行 body。以“簡單依賴圖”為例函數(shù) A 必須在任何其他計算開始前完成B 必須在 C 和 D 開始前完成E 必須在 D 和 F 開始前完成。這是一個部分排序因為例如 B 和 E 之間、C 和 F 之間都沒有顯式的順序約束。實現(xiàn)時用continue_node作為節(jié)點、continue_msg作為消息。continue_node的構(gòu)造函數(shù)接受兩個參數(shù)template typename Body continue_node( graph g, Body body)第一個參數(shù)是它所屬的圖第二個是函數(shù)對象或 lambda 表達式。與function_node不同continue_node總是被假定為無限并發(fā)只要依賴滿足就立即生成任務(wù)。typedef continue_node continue_msg node_t; typedef const continue_msg msg_t; int main() { oneapi::tbb::flow::graph g; node_t A(g, [](msg_t){ a(); } ); node_t B(g, [](msg_t){ b(); } ); node_t C(g, [](msg_t){ c(); } ); node_t D(g, [](msg_t){ d(); } ); node_t E(g, [](msg_t){ e(); } ); node_t F(g, [](msg_t){ f(); } ); make_edge(A, B); make_edge(B, C); make_edge(B, D); make_edge(A, E); make_edge(E, D); make_edge(E, F); A.try_put( continue_msg() ); g.wait_for_all(); return 0; }異步執(zhí)行與等待語義依賴圖的執(zhí)行時間線表明D 的執(zhí)行不會在 B 和 E 都完成之前開始而線程在wait_for_all等待期間可以參與執(zhí)行 oneTBB 工作池中的其他任務(wù)。以下幾點對理解依賴圖以及整個 flow graph至關(guān)重要圖內(nèi)所有執(zhí)行都是異步的A.try_put在遞增計數(shù)器并生成一個執(zhí)行 A 的 body 的任務(wù)后迅速把控制權(quán)交還調(diào)用線程body 任務(wù)執(zhí)行 lambda 表達式然后如果有后繼向所有后繼節(jié)點try_put一個continue_msg。只有wait_for_all會阻塞而且即使在等待期間調(diào)用線程也可能被用于執(zhí)行 oneTBB 工作池中的任務(wù)而不是空轉(zhuǎn)。并行程度取決于可用線程數(shù)上述時間線展示的是線程充足、所有可并發(fā)執(zhí)行的任務(wù)都能并行運行的情況。如果線程更少部分生成的任務(wù)只能等待線程可用后再執(zhí)行。continue_msg的定義可直接在源碼中找到——flow_graph.h 中聲明了class continue_msg {};這是一個不含數(shù)據(jù)的標記類型僅用于觸發(fā)后繼節(jié)點的依賴計數(shù)。核心機制節(jié)點、邊、圖對象與并發(fā)限制節(jié)點graph_node、sender與receiver節(jié)點是繼承自oneapi::tbb::flow::graph_node的類通常還繼承oneapi::tbb::flow::senderT、oneapi::tbb::flow::receiverT或兩者兼有。節(jié)點對到達的消息執(zhí)行操作可能產(chǎn)生零個或多個輸出消息有些節(jié)點需要多條輸入消息或產(chǎn)生多條輸出消息。雖然你可以通過繼承g(shù)raph_node、sender、receiver自定義節(jié)點類型但更常見的是直接使用flow_graph.h中預(yù)定義的節(jié)點類型來構(gòu)圖。最基礎(chǔ)的預(yù)定義類型是function_node——一個一進一出的簡單函數(shù)其構(gòu)造函數(shù)接受三個參數(shù)template typename Body function_node(graph g, size_t concurrency, Body body)參數(shù)說明Bodybody 對象的類型g節(jié)點所屬的圖concurrency節(jié)點的并發(fā)限制控制同一時刻最多有多少個節(jié)點調(diào)用并行執(zhí)行取值從 1串行到unlimited無限body用戶定義的函數(shù)對象或 lambda 表達式對輸入消息應(yīng)用后生成輸出消息單節(jié)點示例graph g; function_node int, int n( g, 1, []( int v ) - int { cout v; spin_for( v ); cout v; return v; } );構(gòu)造完成后可以通過邊連接其他節(jié)點或直接調(diào)用n.try_put(...)向節(jié)點投遞消息最后調(diào)用g.wait_for_all()等待所有消息處理完畢n.try_put( 1 ); n.try_put( 2 ); n.try_put( 3 ); g.wait_for_all();當function_node n以并發(fā)限制 1 創(chuàng)建時收到消息序列 1、2、3 后節(jié)點先生成任務(wù)對 1 執(zhí)行 body該任務(wù)完成后才生成任務(wù)處理 2依此類推。try_put不會阻塞等待任務(wù)生成——如果節(jié)點當前無法立即生成任務(wù)處理消息消息會被緩沖在節(jié)點中當并發(fā)限制允許時再生成任務(wù)處理下一條緩沖的消息。并發(fā)限制從串行到無限并行如果把并發(fā)限制改為oneapi::tbb::flow::unlimitedfunction_node int, int n( g, oneapi::tbb::flow::unlimited, []( int v ) - int { cout v; spin_for( v ); cout v; return v; } );運行時庫會在消息到達時立即生成任務(wù)無論此前已生成多少任務(wù)。也可以使用任意具體數(shù)值如 4、8把并發(fā)限制在最多 4 個或 8 個。務(wù)必記住生成任務(wù) ≠ 創(chuàng)建線程。圖可能生成大量任務(wù)但執(zhí)行這些任務(wù)只使用庫線程池中可用數(shù)量的線程——系統(tǒng)只有一個線程時即使任務(wù)再多也只能串行執(zhí)行。邊用make_edge建立消息通道邊是有方向的消息通道通過make_edge(p, s)創(chuàng)建參數(shù)分別是前驅(qū)predecessorp和后繼successors。下面的例子在第一個節(jié)點打印數(shù)值后將其傳遞給第二個節(jié)點求平方再打印graph g; function_node int, int n( g, unlimited, []( int v ) - int { cout v; spin_for( v ); cout v; return v; } ); function_node int, int m( g, 1, []( int v ) - int { v * v; cout v; spin_for( v ); cout v; return v; } ); make_edge( n, m ); n.try_put( 1 ); n.try_put( 2 ); n.try_put( 3 ); g.wait_for_all();make_edge建立了從n到m的邊。n無限并發(fā)其調(diào)用可全部并行m并發(fā)限制為 1其調(diào)用被串行化。由于存在從n到m的邊n返回的每個值都會被運行時庫自動傳遞給m。圖對象graph的生命周期管理概念上一個 flow graph 是節(jié)點與邊的集合。每個節(jié)點恰好屬于一個圖邊只允許連接同一圖中的節(jié)點。圖對象graph g;代表這個集合用于執(zhí)行整圖操作等待圖中所有相關(guān)任務(wù)完成wait_for_all、重置所有節(jié)點狀態(tài)reset、取消所有節(jié)點執(zhí)行cancel等graph g; g.wait_for_all();上面的調(diào)用立即返回因為這是個沒有任何節(jié)點和邊的平凡圖。需要注意圖對象不擁有其節(jié)點你必須確保圖對象的生命周期長于所有加入圖中的節(jié)點以及任何與該圖相關(guān)的活動銷毀圖對象前應(yīng)調(diào)用wait_for_all確保所有活動完成參見 Graph_Object.rst 中的建議即使使用智能指針也要注意節(jié)點與圖的析構(gòu)順序確保節(jié)點不會先于圖被刪除。節(jié)點到任務(wù)的映射以“雙節(jié)點圖”為例n無限并發(fā)、m并發(fā)限制為 1。三次try_put生成三個任務(wù)每個任務(wù)把 lambda 表達式 λ? 應(yīng)用到一個輸入消息上因為n無限并發(fā)只要線程足夠這三個任務(wù)就能并發(fā)執(zhí)行。n的每個任務(wù)完成后把輸出投給后繼m由于m并發(fā)限制為 1它不會立即生成全部任務(wù)而是按消息到達順序依次串行生成任務(wù)執(zhí)行其 body λ?。全部任務(wù)完成后wait_for_all返回。時間線圖例可參考 Mapping_Nodes2Tasks.rst 中的execution_timeline2node.jpg插圖任務(wù)由try_put快速派發(fā)、body 執(zhí)行后向下游投遞結(jié)果、wait_for_all阻塞期間調(diào)用線程參與工作池任務(wù)執(zhí)行——這與依賴圖的時間線語義完全一致。預(yù)定義節(jié)點類型總覽flow_graph.h提供了豐富的預(yù)定義節(jié)點類型足以覆蓋絕大多數(shù)構(gòu)圖需求詳見 Predefined_Node_Types.rst 及 Developer Reference 中對每個節(jié)點的詳細描述預(yù)定義節(jié)點類型說明input_node單輸出節(jié)點輸出類型泛型。激活后執(zhí)行用戶 body 生成輸出僅當下游接受前一次輸出時才再次調(diào)用 body否則臨時緩沖前一次輸出直到被下游接受。function_node單輸入單輸出節(jié)點輸出廣播給所有后繼輸入/輸出類型均泛型。執(zhí)行用戶 body并發(fā)級別與緩沖策略可控每個輸入恰好產(chǎn)生一個輸出。continue_node單輸入單輸出節(jié)點輸出廣播給所有后繼。輸入類型為continue_msg收到 N 個continue_msg后執(zhí)行用戶 bodyN 等于前驅(qū)數(shù)量加上構(gòu)造時指定的偏移量。multifunction_node單輸入多輸出節(jié)點輸入類型泛型、多個輸出類型泛型。執(zhí)行用戶 body并發(fā)級別與緩沖策略可控body 可在每個輸出端口產(chǎn)生零個或多個消息。broadcast_node單輸入單輸出節(jié)點把收到的每條消息廣播給所有后繼輸入輸出同型不緩沖消息。buffer_node、queue_node、priority_queue_node、sequencer_node單輸入單輸出節(jié)點緩沖消息并只把輸出發(fā)送給一個后繼順序按節(jié)點類型而異。這些節(jié)點與廣播類節(jié)點不同只發(fā)給單個后繼。join_node多輸入單輸出節(jié)點多個泛型輸入、輸出為這些類型的 tuple。節(jié)點從每個輸入端口各取一條消息組合成 tuple 廣播給所有后繼組合策略可選用排隊queueing、保留reserving或標簽匹配tag-matching。split_node單輸入多輸出節(jié)點輸入為 tupletuple 中每種類型對應(yīng)一個輸出端口收到 tuple 后把每個元素輸出到對應(yīng)端口。write_once_node、overwrite_node單輸入單輸出節(jié)點緩沖單條消息并廣播給所有后繼廣播后保留最后一條消息供未來后繼使用。write_once_node只接受第一條消息overwrite_node接受所有消息并以后者替換前者。limiter_node多輸入單輸出節(jié)點輸出廣播給所有后繼主輸入/輸出同型。廣播消息時遞增內(nèi)部計數(shù)器達到用戶設(shè)定的閾值后停止廣播可通過特殊輸入端口調(diào)整計數(shù)允許繼續(xù)廣播不緩沖消息。indexer_node多輸入單輸出節(jié)點輸出廣播給所有后繼。輸入為類型列表輸出為tagged_msg標簽標識消息來自哪個端口消息到達輸入端口時逐一廣播。composite_node端口數(shù)可為 0、1 或多個的節(jié)點把一組其他節(jié)點打包維護指向邊界端口的引用 tuple從而可用composite_node的對應(yīng)端口建立原本需要從內(nèi)部真實節(jié)點建立的邊。async_node預(yù)覽特性允許 flow graph 與用戶或另一運行時管理的外部活動通信接收泛型消息、調(diào)用用戶 body 把消息提交給外部活動外部活動可通過專用接口返回泛型結(jié)果并投給async_node的所有后繼。小結(jié)數(shù)據(jù)流圖讓真實數(shù)據(jù)沿邊流動適合流水線式、每個值逐節(jié)點變換的計算可用function_node 循環(huán)、broadcast_node或input_node三種方式實現(xiàn)源端input_node還能感知下游壓力、控制內(nèi)存。依賴圖用continue_msg傳遞“完成”信號continue_node對前驅(qū)消息計數(shù)、達到前驅(qū)總數(shù)才執(zhí)行適合表達部分排序約束如任務(wù) D 必須等 B 與 E 都完成后才能開始。并發(fā)限制是從 1串行到unlimited的標度決定節(jié)點可并行的調(diào)用數(shù)量任務(wù)數(shù)不等于線程數(shù)實際并行度受線程池可用線程約束。圖內(nèi)執(zhí)行全部異步try_put快速返回只有wait_for_all阻塞且等待線程仍可參與工作池任務(wù)執(zhí)行。若需深入可繼續(xù)閱讀 Basic_Flow_Graph_concepts.rst圖對象、節(jié)點、邊、消息傳遞協(xié)議、緩沖與保留語義、Flow_Graph_Tips.rst 以及 estimate_flow_graph_performance.rst完整的接口定義可在 flow_graph.h 中查閱。贊分享并發(fā)編程高性能計算【免費下載鏈接】oneTBBoneAPI Threading Building Blocks (oneTBB)項目地址https://gitcode.com/gh_mirrors/on/oneTBB點擊查看免費下載相關(guān)推薦ChatGPT Shortcut 瀏覽器擴展使用指南側(cè)邊欄、顯示模式與 AltShiftS 快捷鍵ChatGPT Shortcut 瀏覽器擴展使用指南側(cè)邊欄、顯示模式與 AltShiftS 快捷鍵 導(dǎo)讀 本文基于 ChatGPT ShortcutAAI 應(yīng)用提示工程人工智能前端OpenCore Legacy Patcher 實戰(zhàn)讓舊 Mac 免費升級新版 macOSOpenCore Legacy Patcher 實戰(zhàn)讓舊 Mac 免費升級新版 macOS OpenCore Legacy PatcherOCLP是一個操作系統(tǒng)固件驅(qū)動開發(fā)3 條命令本地跑通 FLUX.1文生圖、圖片編輯與補全的免費完整實戰(zhàn)3 條命令本地跑通 FLUX.1文生圖、圖片編輯與補全的免費完整實戰(zhàn) 本文帶你用 3 條命令跑通 Black Forest Labs 開源的 FLUX.1 推人工智能大模型本地部署媒體生成上一篇大麥搶票自動化工具從零配置到完整跑通只需10分鐘下一篇Fontello與TypeScript類型定義創(chuàng)建類型安全的圖標組件創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考