large message Transmission Jetson Thor to windows

Tái hiện quá trình trao đổi giữa client Windows và server Jetson Thor bằng dds.xml, server.py và client.py, trước tiên với các message dưới 1 MB rồi với payload 2 MB. Bắt đầu bằng cách kiểm tra cấu hình FragmentSize và lỗi UDP được báo cáo, bao gồm cả thiết lập CYCLONEDDS_URI trong code. Được xem là hoàn tất khi một request lớn và response lớn hoàn tất thành công với một cấu hình hợp lệ được ghi lại.

windows 10 as client ,send message to jet son
jetson Thor Arm64 as server, receive message and return back to client
i hava build from c source ,and pip install cyclonedds on jetson,
when i send message less 1M ,it's ok， when I send message over 1M , i got some errors:

1. when i set FragmentSize to 64kB ,i got python: config: //CycloneDDS/Domain/General/FragmentSize/#text: 64kB: value out of range , i have no idea what is the max FragmentSize
2. so i common FragmentSize , but client error: 1784015961.173836 [0] 11748: ddsi_udp_conn_write to udp/192.168.0.101:47404 failed with retcode -58
3. on windows ,set CYCLONEDDS_URI not work ,so i set it in code os.environ
   blow is my code and config file

dds.xml:

<?xml version="1.0" encoding="UTF-8" ?>
<CycloneDDS xmlns="https://cdds.io/config"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="https://cdds.io/config https://raw.githubusercontent.com/eclipse-cyclonedds/cyclonedds/iceoryx/etc/cyclonedds.xsd">
    <Domain>
        <General>
            <Interfaces>
                    <NetworkInterface address="192.168.0.101"></NetworkInterface>
            </Interfaces>
            <MaxMessageSize>10MB</MaxMessageSize>

            <FragmentSize>64kB</FragmentSize>     
        </General>
        <Internal>
          <MaxSampleSize>100MB</MaxSampleSize>
          <Watermarks>
            <WhcHigh>50MB</WhcHigh>
          </Watermarks>
          <!-- socket buffer -->
          <SocketReceiveBufferSize min="10MB"/>
          <SocketSendBufferSize min="10MB"/>
        </Internal>
    </Domain>
</CycloneDDS>

server.py :

import json
from dataclasses import dataclass
from cyclonedds.domain import DomainParticipant
from cyclonedds.topic import Topic
from cyclonedds.pub import Publisher
from cyclonedds.sub import Subscriber, DataReader
from cyclonedds.pub import DataWriter
from cyclonedds.idl import IdlStruct

@dataclass
class StringMsg(IdlStruct):
    content: str

class ServiceDDS:
    def __init__(self, service_name, callback):
        self.service_name = service_name
        self.callback = callback
        self.participant = DomainParticipant()
        self.subscriber = Subscriber(self.participant)
        self.publisher = Publisher(self.participant)

        self.req_topic = Topic(self.participant, service_name + "_request", StringMsg)
        self.res_topic = Topic(self.participant, service_name + "_response", StringMsg)

        self.reader = DataReader(self.subscriber, self.req_topic)
        self.writer = DataWriter(self.publisher, self.res_topic)

        print(f"[Service] {service_name} started")



    def spin(self):
        while True:
            samples = self.reader.take()
            for sample in samples:
                if sample is None:
                    continue
                if not hasattr(sample, 'content'):
                    continue
                req = json.loads(sample.content)
                req_id = req["id"]
                data = req["data"]
                print(f"[Service] Received request req_id {req_id}, data length: {len(data)}")
                result = self.callback(data)
                res = {"id": req_id, "result": result}
                content = json.dumps(res)
                print(f"len(msg)==={len(content)}")
                msg = StringMsg(content)
                #print(f"len(msg)==={len(msg)}")
                #print(msg)
                self.writer.write(msg)
                print('write ok')
                #print(f"[Service] Sent response: {result}")

# handle message 
def process_big_string(s):
    print('have recv msg.... ')
    #return "C" * 60 * 1024 * 1024
    return s

if __name__ == "__main__":
    srv = ServiceDDS("demo_service", process_big_string)
    srv.spin()

client.py

import json
import uuid
import time,os
from dataclasses import dataclass
from cyclonedds.domain import DomainParticipant
from cyclonedds.topic import Topic
from cyclonedds.pub import Publisher
from cyclonedds.sub import Subscriber, DataReader
from cyclonedds.pub import DataWriter
from cyclonedds.idl import IdlStruct
from cyclonedds.util import duration

@dataclass
class StringMsg(IdlStruct):
    content: str

class ReqDDS:
    def __init__(self, service_name):
        ip_address = "192.168.0.3"
        os.environ["CYCLONEDDS_URI"] = f"""
            <Domain>
                <General>
                    <Interfaces>
                            <NetworkInterface address="192.168.0.3"></NetworkInterface>
                    </Interfaces>
                    <MaxMessageSize>10MB</MaxMessageSize>
                     
                    <!-- <FragmentSize>64kB</FragmentSize>  -->
                </General>
                <Internal>
  
                  <MaxSampleSize>100MB</MaxSampleSize>        
 
                  <Watermarks>
                    <WhcHigh>50MB</WhcHigh>
                  </Watermarks>
        
                  <!-- socket buffer -->
                    <SocketSendBufferSize min="20MB"/>
                    <SocketReceiveBufferSize min="20MB"/>
                </Internal>
            </Domain>
            """

        self.service_name = service_name
        self.participant = DomainParticipant()
        self.subscriber = Subscriber(self.participant)
        self.publisher = Publisher(self.participant)

        self.req_topic = Topic(self.participant, service_name + "_request", StringMsg)
        self.res_topic = Topic(self.participant, service_name + "_response", StringMsg)

        self.reader = DataReader(self.subscriber, self.res_topic)
        self.writer = DataWriter(self.publisher, self.req_topic)

        print(f"[Client] Created for {service_name}")

    def wait_for_service(self, timeout=43200):
        start = time.time()
        while time.time() - start < timeout:
            subs = self.writer.get_matched_subscriptions()
            if len(subs) > 0:
                print("[Client] Service available!")
                return True
            print("Client wait Service up!")
            time.sleep(1)
        print("[Client] wait Service not available!")
        return False

    def call(self, data):
        req_id = str(uuid.uuid4())
        req = {"id": req_id, "data": data}
        content = json.dumps(req)
        msg = StringMsg(content)
        self.writer.write(msg)

        # 等待响应
        while True:     
            samples = self.reader.take()
            for sample in samples:
                if sample is None:
                    continue
                res = json.loads(sample.content)
                if res["id"] == req_id:
                    return res["result"]
            time.sleep(1)

if __name__ == "__main__":
    client = ReqDDS("demo_service")
    if client.wait_for_service():
        # generate 2M message all is A
        big_data = "A" * 2 * 1024 * 1024
        #big_data = 'test message less 1M'
        print(f"[Client] Sending big string of length {len(big_data)}")
        while True:
            start_time = time.time()
            result = client.call(big_data)
            end_time = time.time()
            print(f"{result[15]}")
            print(f"[Client] Result (length of string): {len(result)}, time {end_time - start_time}")
            time.sleep(5)