別)
下圖是fastdds官方的圖，清晰地展示了dds支持的傳輸層:根據(jù)通信雙方的相對位置(跨機(jī)器、同機(jī)器跨進(jìn)程、同進(jìn)程)的不同，選擇合適的傳輸層，是通信中間件必須要考慮的事情?？鐧C(jī)器：udp、tcp跨機(jī)器通信，只能通過網(wǎng)絡(luò)， fastdds支持UDP和TCP。同機(jī)器，跨進(jìn)程：SHM、DATA-SHARING在同一個機(jī)器中，使用UDP和TCP進(jìn)行通信，當(dāng)然也是可以的。但是，從性能的角度來考慮，更推薦使用SHM，SHM和UDP/TCP相比有2點(diǎn)優(yōu)勢：1、減少系統(tǒng)調(diào)用次數(shù)共享內(nèi)存創(chuàng)建完成之后，后邊使用的時候直接是內(nèi)存操作，而UCP/TCP每次發(fā)送或者接收的的時候，都要通過系統(tǒng)調(diào)用。每次系統(tǒng)調(diào)用，都會有上下文的保存和恢復(fù)的動作，性能不友好。2、支持超長消息UDP/TCP受協(xié)議棧的影響，可能會對消息進(jìn)行分片和組裝，1500。SHM也不完全是優(yōu)點(diǎn)，也有自己的缺點(diǎn)，比如實(shí)現(xiàn)復(fù)雜：1、使用UDP/TCP來收發(fā)包，功能成熟，接口簡單，拿來就用。使用SHM進(jìn)行通信的話，很多功能需要自己實(shí)現(xiàn)，比如進(jìn)程間的同步(通過信號量來實(shí)現(xiàn))，buffer的管理(segment)等。同進(jìn)程：INTRA-PROCESS如果writer和reader在同一個進(jìn)程，那么可以使用這種傳輸方式，writer線程會直接調(diào)用reader的回調(diào)函數(shù)，兩者在同一個線程中。如下圖所示，使用命令gdb --args ./delivery_mechanisms pubsub -m INTRA-PROCESS，并對函數(shù)PubSubApp::on_data_available設(shè)置斷點(diǎn)，可以看到調(diào)用棧，使用INTRA-PROCESS的時候，subscriber的接收回調(diào)函數(shù)直接被發(fā)送線程調(diào)用，在同一個線程中。這是一種效率最高的方式，去掉了中間緩存的環(huán)節(jié)，減少了數(shù)據(jù)拷貝次數(shù)。在默認(rèn)情況下，同進(jìn)程的通信使用intra-process，同機(jī)器跨進(jìn)程使用SHM，跨機(jī)器使用UDP。同機(jī)器的不同進(jìn)程間通信可以使用SHM，也可以使用DATA-SHARING，那么兩者有什么區(qū)別呢？本文使用fastdds中的例子delivery_mechanisms，這個例子可以測試不同的傳輸層。本文先介紹SHM和DATA-SHARING傳輸層的數(shù)據(jù)收發(fā)流程，最后總結(jié)兩者之間的異同。1SHM收發(fā)流程簡單來說，如果要看SHM的發(fā)送流程，那么可以以函數(shù)SharedMemTransport::send為中心進(jìn)行梳理，通過gdb對這個函數(shù)設(shè)置斷點(diǎn)，可以查看函數(shù)的調(diào)用棧，閱讀函數(shù)的代碼，可以看函數(shù)的發(fā)送操作；如果要看SHM的接收過程，可以通過函數(shù)SharedMemChannelResource::perform_listen_operation來梳理，對于TCP、UDP、SHM來說，三種傳輸層都實(shí)現(xiàn)了自己的函數(shù)perform_listen_operation，這個函數(shù)監(jiān)聽接收數(shù)據(jù)。上圖中左邊是TranportDescriptor，右邊是Transport。用戶在創(chuàng)建傳輸層的時候，并不需要直接創(chuàng)建一個Transport，而是構(gòu)造一個傳輸層的描述符即可，fastdds內(nèi)部根據(jù)描述符來創(chuàng)建傳輸層。這是典型的設(shè)計模式中的建造者模式，當(dāng)構(gòu)造一個對象時，如果需要傳遞較多的參數(shù)，并且參數(shù)之間還有一些依賴關(guān)系，需要對參數(shù)的關(guān)系進(jìn)行檢查，那么構(gòu)造函數(shù)實(shí)現(xiàn)起來就會比較復(fù)雜，這個時候就可以使用建造者模式，在建造者中對參數(shù)進(jìn)行檢查，如果沒有問題則直接構(gòu)造對象，讓構(gòu)造函數(shù)只專注于對象的構(gòu)造，至于參數(shù)的檢查工作，放到構(gòu)造器中來完成。參數(shù)檢查與對象創(chuàng)建進(jìn)行了解耦，設(shè)計模式的主要出發(fā)點(diǎn)和落腳點(diǎn)就是解耦，但是解耦也不是盲目地將代碼進(jìn)行拆分，而是有意義的拆分：滿足模塊之間的邊界劃分，提升代碼的復(fù)用性，對修改關(guān)閉、對擴(kuò)展開放等。上圖中左圖的SenderResource負(fù)責(zé)發(fā)送數(shù)據(jù)，右圖的ChannelResource負(fù)責(zé)接收數(shù)據(jù)。SenderResource和ChannelResource都是ChannelResource，負(fù)責(zé)維護(hù)底層的通信channel，對于UDP、TCP來說，channel就是socket；對于SHM來說，channel就是共享內(nèi)存文件。ChannelResource中保存的是收發(fā)數(shù)據(jù)的通道，Transport管理ChannelResource。UDP、TCP和SHM的類有平行的關(guān)系，它們都有自己的TransportDescriptor、Tranport、SenderResource、ChannelResource類。通過一個基類或者接口派生出3個傳輸層，體現(xiàn)了c++中多態(tài)的使用。Locator用來描述一個端點(diǎn)，其中kind包括LOCATOR_KIND_TCPv4、LOCATOR_KIND_UDPv4、 LOCATOR_KIND_TCPv6、LOCATOR_KIND_UDPv6、LOCATOR_KIND_SHM；port就是通信的端口，比如7400、7410這些；address是地址，在UDP、TCP中就是ip地址。只要知道要發(fā)送對象的Locator信息，那么數(shù)據(jù)就可以發(fā)送出去。Locator { int32_t kind; uint32_t port; std::string address; }1.1發(fā)送下圖是對SharedMemTransport::send設(shè)置斷點(diǎn)，看到的調(diào)用棧。發(fā)送流程：①用戶調(diào)用DataWriter::write進(jìn)行發(fā)送，所以dds提供的接口是write，并不是publish。在我們實(shí)際工作中，往往會對dds接口進(jìn)行封裝，封裝成publish接口，publish更符合dds發(fā)布數(shù)據(jù)的語義，更直觀，好理解。②在DataWriterHistory中，將要發(fā)送的數(shù)據(jù)封裝到一個CacheChange_t中③在發(fā)送側(cè)，dds和rtps之間的接口類是dds側(cè)的DataWriterHistory和rtps側(cè)的WriterHistory④rtps中的BaseWriter進(jìn)行發(fā)送⑤RtpsParticipantImpl::sendSync進(jìn)行發(fā)送，SenderResource屬于Participant中的資源，在Participant中調(diào)用SenderResource的send函數(shù)，最終調(diào)用到SharedMemTransport的send然后我們再來看看SharedMemTransport::send函數(shù)內(nèi)部的實(shí)現(xiàn)：bool SharedMemTransport::send( const std::vectorNetworkBuffer buffers, uint32_t total_bytes, fastdds::rtps::LocatorsIterator* destination_locators_begin, fastdds::rtps::LocatorsIterator* destination_locators_end, const std::chrono::steady_clock::time_point max_blocking_time_point) { ... fastdds::rtps::LocatorsIterator it = *destination_locators_begin; bool ret = true; std::shared_ptrSharedMemManager::Buffer shared_buffer; try { while (it != *destination_locators_end) { if (IsLocatorSupported(*it)) { if (shared_buffer == nullptr) { shared_buffer = copy_to_shared_buffer(buffers, total_bytes, max_blocking_time_point); } ret = send(shared_buffer, *it); ... } ++it; } } ... return ret; }①對于第一次循環(huán)，需要通過函數(shù)copy_to_shared_buffer將數(shù)據(jù)保存到共享內(nèi)存中在函數(shù)copy_to_shared_buffer中打印shared_mem_segment_的信息，可以看到segment name是fastdds_aec0cadd732cebe2，這個就是發(fā)送數(shù)據(jù)要拷貝的共享內(nèi)存。所以說，使用共享內(nèi)存發(fā)送數(shù)據(jù)時，會進(jìn)行一次數(shù)據(jù)拷貝。將數(shù)據(jù)拷貝到共享內(nèi)存之后，就是通知接收者，通知接收者需要兩個步驟來實(shí)現(xiàn)：①構(gòu)造一個BufferDescrptor(簡稱BD)，并將BD發(fā)送到對應(yīng)的端口②喚醒監(jiān)聽者構(gòu)造一個BufferDescriptor:喚醒監(jiān)聽者：1.2接收對Subscriber的接收回調(diào)設(shè)置斷點(diǎn)，調(diào)用棧如下：fastdds SHM的writer和reader通過信號量進(jìn)行喚醒。怎么去查看相關(guān)的代碼，去確認(rèn)通過哪個信號量進(jìn)行喚醒的？（1）在gdb中對subscriber的數(shù)據(jù)接收函數(shù)設(shè)置斷點(diǎn)，如下函數(shù)eprosima::fastdds::examples::delivery_mechanisms::SubscriberApp::on_data_available當(dāng)該函數(shù)被調(diào)用時，通過info thread查看哪個這個函數(shù)在哪個進(jìn)程中被調(diào)用。如下圖所示，該函數(shù)在進(jìn)程6中被調(diào)用。（2）停止publisher，讓發(fā)布側(cè)不再寫數(shù)據(jù)，這個時候因為沒有數(shù)據(jù)，subscriber側(cè)收不到數(shù)據(jù)，那么一定是通過信號量在等待。此時再使用bt查看線程6的調(diào)用棧如下。使用boost庫中的信號量boost::interprocess::interprocess_semaphore，這個信號量是基于共享內(nèi)存的信號量。信號量屬于一個Port。從調(diào)用棧可以看出來，只有創(chuàng)建接收通道的時候，才會創(chuàng)建信號量。接收側(cè)通過這個信號量進(jìn)行wait，發(fā)送側(cè)喚醒的時候也是找到接收側(cè)的信號量進(jìn)行喚醒。所以，創(chuàng)建發(fā)送資源的時候不需要創(chuàng)建信號量。由上圖紅線可知，用于創(chuàng)建信號量的共享內(nèi)存創(chuàng)建在了端口共享內(nèi)存上。2DATA-SHARING收發(fā)流程DATA-SHARING不是一個標(biāo)準(zhǔn)的傳輸層，沒有TransportDescriptor、Transport、SenderResource、ChannelResource這些資源。2.1發(fā)送delivery mechanisms這個例子，publish的時候，并不是構(gòu)造了一個臨時的變量發(fā)送出去的，而是通過load_sample首先從底層申請了一塊內(nèi)存，而這塊內(nèi)存就是從data sharing的writer pool中申請的，這樣數(shù)據(jù)直接就是保存在這里的，payload pool就是從這里申請的，申請的是data sharing的writer pool，而writer pool就是基于共享內(nèi)存創(chuàng)建的。所以說，在發(fā)送時，沒有拷貝。std::shared_ptrIPayloadPool DataWriterImpl::get_payload_pool(){ if (!payload_pool_) { // Avoid calling the serialization size functors on PREALLOCATED mode fixed_payload_size_ = pool_config_.memory_policy == PREALLOCATED_MEMORY_MODE ? fastdds:傳輸層SHM和DATA-SHARING的區(qū)別
下圖是fastdds官方的圖，清晰地展示了dds支持的傳輸層:根據(jù)通信雙方的相對位置(跨機(jī)器、同機(jī)器跨進(jìn)程、同進(jìn)程)的不同，選擇合適的傳輸層，是通信中間件必須要考慮的事情?？鐧C(jī)器：udp、tcp跨機(jī)器通信，只能通過網(wǎng)絡(luò)， fastdds支持UDP和TCP。同機(jī)器，跨進(jìn)程：SHM、DATA-SHARING在同一個機(jī)器中，使用UDP和TCP進(jìn)行通信，當(dāng)然也是可以的。但是，從性能的角度來考慮，更推薦使用SHM，SHM和UDP/TCP相比有2點(diǎn)優(yōu)勢：1、減少系統(tǒng)調(diào)用次數(shù)共享內(nèi)存創(chuàng)建完成之后，后邊使用的時候直接是內(nèi)存操作，而UCP/TCP每次發(fā)送或者接收的的時候，都要通過系統(tǒng)調(diào)用。每次系統(tǒng)調(diào)用，都會有上下文的保存和恢復(fù)的動作，性能不友好。2、支持超長消息UDP/TCP受協(xié)議棧的影響，可能會對消息進(jìn)行分片和組裝，1500。SHM也不完全是優(yōu)點(diǎn)，也有自己的缺點(diǎn)，比如實(shí)現(xiàn)復(fù)雜：1、使用UDP/TCP來收發(fā)包，功能成熟，接口簡單，拿來就用。使用SHM進(jìn)行通信的話，很多功能需要自己實(shí)現(xiàn)，比如進(jìn)程間的同步(通過信號量來實(shí)現(xiàn))，buffer的管理(segment)等。同進(jìn)程：INTRA-PROCESS如果writer和reader在同一個進(jìn)程，那么可以使用這種傳輸方式，writer線程會直接調(diào)用reader的回調(diào)函數(shù)，兩者在同一個線程中。如下圖所示，使用命令gdb --args ./delivery_mechanisms pubsub -m INTRA-PROCESS，并對函數(shù)PubSubApp::on_data_available設(shè)置斷點(diǎn)，可以看到調(diào)用棧，使用INTRA-PROCESS的時候，subscriber的接收回調(diào)函數(shù)直接被發(fā)送線程調(diào)用，在同一個線程中。這是一種效率最高的方式，去掉了中間緩存的環(huán)節(jié)，減少了數(shù)據(jù)拷貝次數(shù)。在默認(rèn)情況下，同進(jìn)程的通信使用intra-process，同機(jī)器跨進(jìn)程使用SHM，跨機(jī)器使用UDP。同機(jī)器的不同進(jìn)程間通信可以使用SHM，也可以使用DATA-SHARING，那么兩者有什么區(qū)別呢？本文使用fastdds中的例子delivery_mechanisms，這個例子可以測試不同的傳輸層。本文先介紹SHM和DATA-SHARING傳輸層的數(shù)據(jù)收發(fā)流程，最后總結(jié)兩者之間的異同。1SHM收發(fā)流程簡單來說，如果要看SHM的發(fā)送流程，那么可以以函數(shù)SharedMemTransport::send為中心進(jìn)行梳理，通過gdb對這個函數(shù)設(shè)置斷點(diǎn)，可以查看函數(shù)的調(diào)用棧，閱讀函數(shù)的代碼，可以看函數(shù)的發(fā)送操作；如果要看SHM的接收過程，可以通過函數(shù)SharedMemChannelResource::perform_listen_operation來梳理，對于TCP、UDP、SHM來說，三種傳輸層都實(shí)現(xiàn)了自己的函數(shù)perform_listen_operation，這個函數(shù)監(jiān)聽接收數(shù)據(jù)。上圖中左邊是TranportDescriptor，右邊是Transport。用戶在創(chuàng)建傳輸層的時候，并不需要直接創(chuàng)建一個Transport，而是構(gòu)造一個傳輸層的描述符即可，fastdds內(nèi)部根據(jù)描述符來創(chuàng)建傳輸層。這是典型的設(shè)計模式中的建造者模式，當(dāng)構(gòu)造一個對象時，如果需要傳遞較多的參數(shù)，并且參數(shù)之間還有一些依賴關(guān)系，需要對參數(shù)的關(guān)系進(jìn)行檢查，那么構(gòu)造函數(shù)實(shí)現(xiàn)起來就會比較復(fù)雜，這個時候就可以使用建造者模式，在建造者中對參數(shù)進(jìn)行檢查，如果沒有問題則直接構(gòu)造對象，讓構(gòu)造函數(shù)只專注于對象的構(gòu)造，至于參數(shù)的檢查工作，放到構(gòu)造器中來完成。參數(shù)檢查與對象創(chuàng)建進(jìn)行了解耦，設(shè)計模式的主要出發(fā)點(diǎn)和落腳點(diǎn)就是解耦，但是解耦也不是盲目地將代碼進(jìn)行拆分，而是有意義的拆分：滿足模塊之間的邊界劃分，提升代碼的復(fù)用性，對修改關(guān)閉、對擴(kuò)展開放等。上圖中左圖的SenderResource負(fù)責(zé)發(fā)送數(shù)據(jù)，右圖的ChannelResource負(fù)責(zé)接收數(shù)據(jù)。SenderResource和ChannelResource都是ChannelResource，負(fù)責(zé)維護(hù)底層的通信channel，對于UDP、TCP來說，channel就是socket；對于SHM來說，channel就是共享內(nèi)存文件。ChannelResource中保存的是收發(fā)數(shù)據(jù)的通道，Transport管理ChannelResource。UDP、TCP和SHM的類有平行的關(guān)系，它們都有自己的TransportDescriptor、Tranport、SenderResource、ChannelResource類。通過一個基類或者接口派生出3個傳輸層，體現(xiàn)了c++中多態(tài)的使用。Locator用來描述一個端點(diǎn)，其中kind包括LOCATOR_KIND_TCPv4、LOCATOR_KIND_UDPv4、 LOCATOR_KIND_TCPv6、LOCATOR_KIND_UDPv6、LOCATOR_KIND_SHM；port就是通信的端口，比如7400、7410這些；address是地址，在UDP、TCP中就是ip地址。只要知道要發(fā)送對象的Locator信息，那么數(shù)據(jù)就可以發(fā)送出去。Locator { int32_t kind; uint32_t port; std::string address; }1.1發(fā)送下圖是對SharedMemTransport::send設(shè)置斷點(diǎn)，看到的調(diào)用棧。發(fā)送流程：①用戶調(diào)用DataWriter::write進(jìn)行發(fā)送，所以dds提供的接口是write，并不是publish。在我們實(shí)際工作中，往往會對dds接口進(jìn)行封裝，封裝成publish接口，publish更符合dds發(fā)布數(shù)據(jù)的語義，更直觀，好理解。②在DataWriterHistory中，將要發(fā)送的數(shù)據(jù)封裝到一個CacheChange_t中③在發(fā)送側(cè)，dds和rtps之間的接口類是dds側(cè)的DataWriterHistory和rtps側(cè)的WriterHistory④rtps中的BaseWriter進(jìn)行發(fā)送⑤RtpsParticipantImpl::sendSync進(jìn)行發(fā)送，SenderResource屬于Participant中的資源，在Participant中調(diào)用SenderResource的send函數(shù)，最終調(diào)用到SharedMemTransport的send然后我們再來看看SharedMemTransport::send函數(shù)內(nèi)部的實(shí)現(xiàn)：bool SharedMemTransport::send( const std::vectorNetworkBuffer buffers, uint32_t total_bytes, fastdds::rtps::LocatorsIterator* destination_locators_begin, fastdds::rtps::LocatorsIterator* destination_locators_end, const std::chrono::steady_clock::time_point max_blocking_time_point) { ... fastdds::rtps::LocatorsIterator it = *destination_locators_begin; bool ret = true; std::shared_ptrSharedMemManager::Buffer shared_buffer; try { while (it != *destination_locators_end) { if (IsLocatorSupported(*it)) { if (shared_buffer == nullptr) { shared_buffer = copy_to_shared_buffer(buffers, total_bytes, max_blocking_time_point); } ret = send(shared_buffer, *it); ... } ++it; } } ... return ret; }①對于第一次循環(huán)，需要通過函數(shù)copy_to_shared_buffer將數(shù)據(jù)保存到共享內(nèi)存中在函數(shù)copy_to_shared_buffer中打印shared_mem_segment_的信息，可以看到segment name是fastdds_aec0cadd732cebe2，這個就是發(fā)送數(shù)據(jù)要拷貝的共享內(nèi)存。所以說，使用共享內(nèi)存發(fā)送數(shù)據(jù)時，會進(jìn)行一次數(shù)據(jù)拷貝。將數(shù)據(jù)拷貝到共享內(nèi)存之后，就是通知接收者，通知接收者需要兩個步驟來實(shí)現(xiàn)：①構(gòu)造一個BufferDescrptor(簡稱BD)，并將BD發(fā)送到對應(yīng)的端口②喚醒監(jiān)聽者構(gòu)造一個BufferDescriptor:喚醒監(jiān)聽者：1.2接收對Subscriber的接收回調(diào)設(shè)置斷點(diǎn)，調(diào)用棧如下：fastdds SHM的writer和reader通過信號量進(jìn)行喚醒。怎么去查看相關(guān)的代碼，去確認(rèn)通過哪個信號量進(jìn)行喚醒的？（1）在gdb中對subscriber的數(shù)據(jù)接收函數(shù)設(shè)置斷點(diǎn)，如下函數(shù)eprosima::fastdds::examples::delivery_mechanisms::SubscriberApp::on_data_available當(dāng)該函數(shù)被調(diào)用時，通過info thread查看哪個這個函數(shù)在哪個進(jìn)程中被調(diào)用。如下圖所示，該函數(shù)在進(jìn)程6中被調(diào)用。（2）停止publisher，讓發(fā)布側(cè)不再寫數(shù)據(jù)，這個時候因為沒有數(shù)據(jù)，subscriber側(cè)收不到數(shù)據(jù)，那么一定是通過信號量在等待。此時再使用bt查看線程6的調(diào)用棧如下。使用boost庫中的信號量boost::interprocess::interprocess_semaphore，這個信號量是基于共享內(nèi)存的信號量。信號量屬于一個Port。從調(diào)用棧可以看出來，只有創(chuàng)建接收通道的時候，才會創(chuàng)建信號量。接收側(cè)通過這個信號量進(jìn)行wait，發(fā)送側(cè)喚醒的時候也是找到接收側(cè)的信號量進(jìn)行喚醒。所以，創(chuàng)建發(fā)送資源的時候不需要創(chuàng)建信號量。由上圖紅線可知，用于創(chuàng)建信號量的共享內(nèi)存創(chuàng)建在了端口共享內(nèi)存上。2DATA-SHARING收發(fā)流程DATA-SHARING不是一個標(biāo)準(zhǔn)的傳輸層，沒有TransportDescriptor、Transport、SenderResource、ChannelResource這些資源。2.1發(fā)送delivery mechanisms這個例子，publish的時候，并不是構(gòu)造了一個臨時的變量發(fā)送出去的，而是通過load_sample首先從底層申請了一塊內(nèi)存，而這塊內(nèi)存就是從data sharing的writer pool中申請的，這樣數(shù)據(jù)直接就是保存在這里的，payload pool就是從這里申請的，申請的是data sharing的writer pool，而writer pool就是基于共享內(nèi)存創(chuàng)建的。所以說，在發(fā)送時，沒有拷貝。std::shared_ptrIPayloadPool DataWriterImpl::get_payload_pool(){ if (!payload_pool_) { // Avoid calling the serialization size functors on PREALLOCATED mode fixed_payload_size_ = pool_config_.memory_policy == PREALLOCATED_MEMORY_MODE ?