eclipse-cyclonedds / eclipse-cyclonedds/cyclonedds-python

large message Transmission Jetson Thor to windows

Open
#334 0 comments 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

Dominant language
Python
Stars
110
Forks
68
Avg merge
1h 8m
Merged PRs (30d)
1

Description

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)

Contributor guide

No contributing guide indexed for this repository

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

Reproduce the Windows client and Jetson Thor server exchange using dds.xml, server.py, and client.py, first with messages below 1 MB and then with the 2 MB payload. Start by checking the FragmentSize configuration and the reported UDP error, including the in-code CYCLONEDDS_URI setup. Done means a large request and response complete successfully with a documented valid configuration.

Written by the indexing model from the issue text.

Assessment

Tech stack
python
Domain
distributed-systems, networking
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Quiet
Clarity
Mostly clear
Newbie friendliness
45/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.