鴻蒙子系統解讀-分布式軟總線子系統初步研究（上）

本文作者：江蘇潤和軟體股份有限公司 郎建中

1.總體描述

1.1.總體介紹

裝置通信方式多種多樣(USB/WIFI/BT等)，不同通信方式使用差異很大且繁瑣，同時通信鍊路的融合共享和沖突無法處理，通信安全問題也不好保證。分布式軟總線緻力于實作近場裝置間統一的分布式通信能力管理，提供不區分鍊路的裝置發現和傳輸接口。目前實作能力包含：

l 服務釋出：服務釋出後周邊的裝置可以發現并使用服務。

l 資料傳輸：根據服務的名稱和裝置ID建立一個會話，就可以實作服務間的傳輸功能。

l 安全：提供通信資料的加密能力。

分布式軟總線是多種終端裝置的統一基座，為裝置之間的互聯互通提供了統一的分布式通信能力，能夠快速發現并連接配接裝置，高效地分發任務和傳輸資料。分布式軟總線架構示意圖如下：

1.2. 裝置發現

在分布式軟總線子系統中，裝置分為發現端和被發現端。

發現端：請求使用服務的裝置。一般指智慧屏裝置。

被發現端：釋出服務的裝置。一般指輕量裝置。

限制：目前必須保證發現端和被發現端處于同一個區域網路内。

（1） 發現端裝置，發起discover請求後，使用coap協定在區域網路内發送廣播。封包如下：

（2） 被發現端裝置使用****PublishService****接口釋出服務，接收端收到廣播後，發送coap協定單點傳播給發現端。封包格式如下：

（3） 發現端裝置收到封包會更新裝置資訊。

被發現端釋出服務的例子代碼如下：

發現的流程圖如下：

1.3. 資料傳輸

軟總線提供統一的基于Session的傳輸功能，業務可以通過sessionId收發資料或擷取其相關基本屬性。目前本項目隻實作被動接收Session連接配接的功能，業務可根據自身需要及Session自身屬性判斷是否接受此Session，如不接受，可以主動拒絕此連接配接。本項目暫未提供打開Session的相關能力。

下面是被發現端（服務提供端）向軟總線申請建立Session Server的代碼：

// 定義業務自身的業務名稱，會話名稱及相關回調

const char *g_moduleName = "BUSINESS_NAME";

const char *g_sessionName = "SESSION_NAME";

struct ISessionListener * g_sessionCallback= NULL;

 

// 回調實作：接收對方通過SendBytes發送的資料，此示例實作是接收到對端發送的資料後回複固定消息

void OnBytesReceivedTest(int sessionId, const void* data, unsigned int dataLen)

{

  printf("OnBytesReceivedTest\n");

  printf("Recv Data: %s\n", (char *)data);

  printf("Recv Data dataLen: %d\n", dataLen);

  char *testSendData = "Hello World, Hello!";

  ***\*SendBytes\****(sessionId, testSendData, strlen(testSendData));

  return;

}

 

// 回調實作：用于處理會話關閉後的相關業務操作，如釋放目前會話相關的業務資源，會話無需業務主動釋放

void OnSessionClosedEventTest(int sessionId)

{

  printf("Close session successfully, sessionId=%d\n", sessionId);

}

 

// 回調實作：用于處理會話打開後的相關業務操作。傳回值為0，表示接收；反之，非0表示拒絕。此示例表示隻接受其他裝置的同名會話連接配接

int OnSessionOpenedEventTest(int sessionId)

{

  if (strcmp(***\*GetPeerSessionName\****(sessionId), SESSION_NAME) != 0) {

​    printf("Reject the session which name is different from mine, sessionId=%d\n", sessionId);

​    return -1;

  }

  printf("Open session successfully, sessionId=%d\n", sessionId);

  return 0;

}

 

// 向SoftBus注冊業務會話服務及其回調

int StartSessionServer()

{

  if (g_sessionCallback == NULL) {

​    g_sessionCallback = (struct ISessionListener*)malloc(sizeof(struct ISessionListener));

  }

  if (g_sessionCallback == NULL) {

​    printf("Failed to malloc g_sessionCallback!\n");

​    return -1;

  }

  g_sessionCallback->onBytesReceived = OnBytesReceivedTest;

  g_sessionCallback->onSessionOpened = OnSessionOpenedEventTest;

  g_sessionCallback->onSessionClosed = OnSessionClosedEventTest;

  int ret = ***\*CreateSessionServer\****(g_moduleName, g_sessionName, g_sessionCallback);

  if (ret < 0) {

​    printf("Failed to create session server!\n");

​    free(g_sessionCallback);

​    g_sessionCallback = NULL;

  }

  return ret;

}

 

// 從SoftBus中删除業務會話服務及其回調

void StopSessionServer()

{

  int ret = ***\*RemoveSessionServer\****(g_moduleName, g_sessionName);

  if (ret < 0) {

​    printf("Failed to remove session server!\n");

​    return;

  }

  if (g_sessionCallback != NULL) {

​    free(g_sessionCallback);

​    g_sessionCallback = NULL;

  }

}
           

注意：上面的代碼中的StartSessionServer()函數調用應該是在被發現端使用PublishService()函數釋出服務後，在釋出成功的回調中調用。可以參考《分布式排程子系統》研究中的時序圖，大緻的如下：

上面圖中Module是指使用軟總線的任意子產品（比如：分布式排程子系統子產品），onPublishServiceDone()回調函數是在PushService()調用中設定的回調函數。在成功釋出後，軟總線會調用這個回調。然後Module可以在這個回調中調用上面代碼中的StartSessionServer()函數，在StartSessionServer()函數中，調用軟總線的接口CreateSessionServer()建立會話服務，等待其他裝置的會話連接配接。當其他裝置會話連接配接成功後，軟總線會首先調用OnSessionOpenedEventTest()函數，然後在資料傳輸完成後調用OnBytesReceivedTest()回調函數。Module可以在OnBytesReceivedTest()中處理資料協定格式的解析以及對于的Command指令字的功能調用。

2. 代碼目錄結構

分布式軟總線的代碼在foundation/communication/services/softbus_lite目錄下面，目錄結構如下：

authmanager：提供裝置認證機制和裝置知識庫管理

discovery：提供基于coap協定的裝置發現機制

os_adapter：作業系統适配層

trans_service：提供認證和資料傳輸通道

通過BUILD.gn的分析，我們知道整個softbus_lite目錄下的所有源碼檔案将被編譯到一個動态庫中。其他依賴軟總線的子產品在編譯的時候加上這個動态庫的依賴就可以了。例如：分布式排程子系統所在的foundation這個bin檔案的編譯就依賴這個動态庫。

3. 代碼分析

3.1. 裝置發現

裝置發現的代碼位于foundation/communication/services/softbus_lite/discovery目錄中。這個目錄下的檔案如下：

coap：目錄主要是負責COAP協定的部分

discovery_service：實作了輕量裝置端的服務釋出的能力

整個目錄結構如下：

前面我們介紹過，輕量裝置主要承擔服務釋出者，也就是被發現端的功能。被發現端主要是通過PublishService()這個函數釋出服務，然後在服務釋出成功後的回調中使用CreateSessionServer()函數來建立會話伺服器等待發現端的連接配接。這個章節我們主要從代碼分析PublishService()這個API的實作過程。

PublishService()函數的實作在discovery_service.c檔案中，基本上其他所有的c源代碼的實作都是為這個函數提供支撐的。

下面我們從PublishService()代碼分析開始，分析下整個discovery目錄下的源碼：

這個函數大緻可以分成7個部分，下面我們分别分析這個7個部分的代碼。

3.1.1. 權限檢查

SoftBusCheckPermission()函數實作在os_adapter目錄中。這個目錄結構如下：

os_adapter目錄是為了适配OS作業系統的。我們知道HarmonyOS最底層的作業系統可以是Linux或者是LiteOS。是以，為了适配不同的底層作業系統，os_adapter.c中封裝了一些函數，例如：SemCreate()、SemDelete()、SemWait()等，這些封裝的函數在不同的作業系統上用不同的方法進行了實作。例如：SemCreate()在LiteOS中使用了LOS_SemCreate()建立信号量，在Linux上用sem_init()這個Posix标準接口建立信号量。這裡不再多說，有興趣的同學可以看一下os_adapter.c的源碼。

source/L0/os_adapter.c：這個檔案是适配LiteOS的

source/L1/os_adapter.c：這個檔案是适配Linux的

我們看下SoftBusCheckPermission()函數的實作。在L0中的實作如下：

在LiteOS中基本沒做權限的判斷，隻是檢查參數是否有效。

在L1中的實作如下：

我們看到CheckPermission()函數，記得在Android中有類似的API來判斷權限的。應該是HarmonyOS在Linux上有實作這個功能，而參數SOFTBUS_PERMISSION “ohos.permission.DISTRIBUTED_DATASYNC” 這個跟Android下的權限子串也很相似。

3.1.2. 函數參數有效性校驗

第二部分是輸出參數有效性檢查。輸入參數總共是3個，分别是：

moduleName：調用者的子產品名稱子串

info：PublishInfo結構體，釋出的資訊

cb：釋出成功或者失敗的回調函數

上面的代碼可以看到分别對moduleName是否為空，子串的長度，info裡面的publishId、dataLen等進行了有效性檢查。如果有問題，那麼會調用 ****PublishCallback****()來回調到cb裡面的失敗回調函數，并且給出了出錯碼。

需要注意的是我們在代碼中看到 info->medium必須為COAP，也就是目前隻支援這個方案。其他的方案如下：

USB，BLE的方案估計以後會支援吧。

3.1.3. 建立信号量

SemCreate()函數前面已經介紹過，實作是在os_adapter.c中，分為L0和L1兩個實作分别對應LiteOS和Linux。功能是建立一個信号量。這個信号量的作用是在一個程序中的不同子產品之間保持互斥。也就是下面的SemWait()函數會等待信号量為一個非0的值，然後将信号量減1，如果有其他子產品搶先執行了SemWait()也就是信号量為0，那麼目前子產品會一直等到其他子產品執行SemPost()将信号量+1變成1後才能執行。這個互斥的意義是在SemWait()調用到SemPost()調用之間的所有操作保證不會有其他子產品的幹擾。

3.1.4. 初始化服務

我們看一下InitService()的代碼：

A、是否已經初始化過了

g_isServiceInit全局變量顯示是否已經初始化過了，如果已經被别的子產品調用PublishService()，那麼這個變量的值将為1，那麼不需要第二次初始化了，直接傳回。

B、初始化Common Manager（初始化g_deviceInfo結構體）

InitCommonManager()實作如下：

InitLocalDeviceInfo()函數主要是把g_deviceInfo結構體初始化好。g_deviceInfo中幾個主要的成員如下：

deviceName：對于L0裝置為DEV_L0

version：固定為 “1.0.0”

deviceId：通過函數GetDeviceIdFromFile()調用取得。這個函數會從"/storage/data/softbus/deviceid"檔案中讀取，如果讀取不到，那麼使用随機數字元串組成deviceId，然後再寫入到上面的檔案中。有興趣的同學可以看下這個函數的實作。

C、為内部使用的資料結構配置設定記憶體

g_publishModule 這個全局變量儲存所有釋出服務的子產品的資訊數組。這個數組的元素的資料結構如下：

上面的資料結構中的成員内容基本都是從PublishInfo結構體來的：

D、注冊wifi Callback

RegisterWifiCallback()函數的實作如下：

這個函數很簡單，就是将callback函數指派給全局變量g_wifiCallback。這裡的callback就是WifiEventTrigger()函數。下面會具體分析 g_wifiCallback在哪裡使用。

E、COAP初始化，注冊TCP/IP協定棧的處理，注冊session的底層socket的處理【重點】

在InitService()函數中，調用了CoapInit()函數，代碼如下：

我們再看NSTACKX_Init()的代碼：

繼續看CoapInitDiscovery()的代碼：

這裡大緻分三個部分：

⑴ CoapInitSocket()

CoapInitSocket()函數調用了CoapCreateUdpServer()函數，建立了一個UDP的socket，并綁定到COAP_DEFAULT_PORT（5684）端口上面上面。這個socket指派給****g_serverFd****，在下面還會使用。

CoapCreateUdpServer()函數的實作基本上都是用socket族的标準接口實作，首先是socket()建立一個socket，然後用bind()綁定到指定的IP+PORT。

⑵ CoapInitWifiEvent()

CoapInitWifiEvent()函數大緻分為兩個部分：

1、CreateMsgQue()建立一個消息隊列，使用RegisterWifiEvent()向wifi_lite注冊事件回調函數。g_coapEventHandler.OnWifiConnectionChanged 就是當wifi連接配接狀态發生改變（例如：裝置連接配接到wifi後）被回調。

我們看一下CoapConnectionChangedHandler()這個回調函數：

再看下CoapWriteMsgQueue()函數代碼：

是以這個函數的功能就是向消息隊列中放置了一個消息。這個消息包含了wifi的狀态和一個回調函數。

這個回調函數是CoapHandleWifiEvent()，代碼如下：

是以這個回調函數就是調用g_wifiCallback。前面我們介紹過，g_wifiCallback被設定成了WifiEventTrigger()函數。最終當wifi狀态改變後，會調用WifiEventTrigger()函數。

備注：RegisterWifiEvent()的實作在./vendor/hisi/hi3861/hi3861_adapter/hals/communication/wifi_lite/wifiservice/source/wifi_device.c中。

2、建立wifi消息隊列的處理線程（任務）CoapWifiEventThread()

CoapWifiEventThread()函數的代碼如下：

這個函數主要就是從wifi的消息隊列中擷取消息，然後執行消息中的回調函數，也就是WifiEventTrigger()函數。

⑶ CreateCoapListenThread()

這個函數的代碼如下：

這個函數主要就是建立了一個線程CoapReadHandle，用于處理COAP_DEFAULT_PORT端口上的UDP socket的資料（也就是基于COAP協定的discover廣播消息），代碼如下：

這個函數的核心是調用了HandleReadEvent()。從UDP socket中讀取資料，并解析COAP協定。代碼如下：

上面的代碼中 CoapSocketRecv()就是調用recvfrom()從socket中讀取資料。然後調用COAP_SoftBusDecode()函數對COAP協定進行解析，解析的内容放在decodePacket結構中。最後調用PostServiceDiscover()對智慧屏發送的DISCOVER消息進行回應。這裡代碼不再展開，有興趣同學自己讀一下。

F、調用CoapWriteMsgQueue()觸發擷取wifi的IP位址，并啟動總線

CoapWriteMsgQueue()函數的代碼如下：

這個函數就是向wifi的消息隊列寫入一個消息，強制觸發消息回調函數的執行。前面我們介紹過，消息回調函數就是WifiEventTrigger()。

流程圖如下：

WifiEventTrigger()函數的代碼如下：

這個函數的主要工作是：

1、通過CoapGetIp()擷取本地裝置wifi連接配接後的IP位址，并放入到deviceInfo->deviceIp中。後續會使用。

CoapGetIp()函數會循環執行，每次sleep 10ms，一直到能拿到IP位址。CoapGetWifiIp()函數有兩個版本，分别對應LiteOS和Linux，有興趣的同學可以自己看下代碼。

2、BusManager()函數啟動軟總線。

StartBus()函數主要完成兩個任務：

a、建立認證服務

b、建立會話服務

SelectSessionLoop()就是任務線程，會根據子產品注冊的sessionServer的數組監聽所有的session的資料通信情況，并處理資料。這裡不再展開，有興趣同學可以自己看下代碼。

注意：上層子產品注冊的session将保持在g_sessionMgr->sessionMap_[]數組中。

G、向COAP中注冊裝置資訊

這個函數主要完成g_localDeviceInfo資料結構的初始化，有興趣的同學可以自己看一下。

整個InitService的流程圖如下：

小結：

初始化服務調用InitService()函數實作了如下的功能：

1、初始化了g_deviceInfo結構體，包括：deviceName，deviceId，deviceIp。

2、注冊wifi_lite的event監控事件，當wifi鍊路發生變化的情況下（例如：裝置wifi連接配接成功），擷取目前裝置wifi的IP位址，并放入到g_deviceInfo->deviceIp中，為下面TCP/IP協定棧的初始化做準備。

3、初始化UDP協定socket綁定在COAP_DEFAULT_PORT端口上，監聽COAP的discover消息，并進行處理。

4、根據deviceIp初始化TCP協定socket，并且啟動一個監聽任務，處理連接配接請求，并進行auth的校驗。

5、根據deviceIp初始化TCP協定socket，并且啟動一個監聽任務，處理session的會話請求。

3.1.5. 将Publish資訊加入到Module清單

AddPublishModule()函數，将把moduleName和info（PublishInfo結構）中的内容加入到g_publishModule全局數組中。前面有過介紹，有興趣同學可以看下這個函數的具體實作。

3.1.6. 注冊COAP服務

這部分代碼我們主要看下CoapRegisterDefualtService()函數實作：

代碼中的 info->devicePort 就是基于TCP的認證服務的socket綁定的端口号（在StartBus()函數中指派的）。而serviceData就是 “port:%d”的子串。

上面的代碼主要是把g_localDeviceInfo.serverData指派成 “port:auth_port”這樣的子串。

3.1.7. 回調釋出成功

這裡很簡單，就是調用PublishCallback()執行cb中的釋出成功的回調函數。例如在分布式排程子系統中，使用這個回調函數可以繼續進行session server的建立。

總結

1、裝置發現部分代碼（主要是輕量裝置側）主入口函數為PublishService()函數。

2、PublishService()函數會檢查軟總線的服務是否已經初始化過，如果沒有則會初始化軟總線的所有服務，包括：a、基于UDP的COAP協定discover發現服務。b、wifi裝置狀态監聽服務。c、基于TCP的認證服務。d、基于TCP的session會話管理服務。

3、PublishService()然後會把子產品的資訊加入g_publishModule全局數組中。

4、回調子產品的釋出成功回調函數。

整體流程圖如下：