eclipse-cyclonedds / eclipse-cyclonedds/cyclonedds-python
[Question] About the usage of shared memory in docker
Nobody has claimed this yet.
- Dominant language
- Python
- Stars
- 110
- Forks
- 68
- Avg merge
- 1h 8m
- Merged PRs (30d)
- 1
Description
Hi,
I'm trying to setup environment and other thins to be able to transfer messages using shared memory. First of all, here is my Image class:
Image class
@dataclass
class MainCameraImage(IdlStruct):
img: array[uint8, 4147200]
shape: array[uint64, 3]
@classmethod
def from_numpy(cls, img):
assert len(img.shape) == 3
seq_img = img.tobytes()
return MainCameraImage(seq_img, tuple(img.shape))
def to_numpy(self):
shape = self.shape
return np.frombuffer(self.img, dtype=np.uint8).reshape(shape)
Note that image array has a fixed size. Then I implement publisher and subscriber:
Publisher
import cv2
import time
import numpy as np
import fire
from cyclonedds.core import Listener, Qos, Policy
from cyclonedds.util import duration
from cyclonedds.internal import dds_infinity
from cyclonedds.domain import DomainParticipant
from cyclonedds.topic import Topic
from cyclonedds.pub import DataWriter
from dds_data_structures import MainCameraImage, Image, CameraState,DeviceState
def stream(dev=0, resize_to=None, output_topic_names=['camera_images', 'camera_images_vis', 'camera_images_clr'],
data_type='MainCameraImage', cap_gstreamer=True):
data_type = MainCameraImage
participant = DomainParticipant(0)
writers = []
# unknown, disabled, enabled, error
qos = Qos(
# livelines: automatic
# deadline: dds_infinity
# reliability: reliable
# durability: volatile
# history: keep last
Policy.Liveliness.Automatic(lease_duration=duration(seconds=10)),
Policy.Deadline(deadline=dds_infinity),
Policy.Reliability.Reliable(max_blocking_time=duration(seconds=1)),
Policy.Durability.Volatile,
Policy.History.KeepLast(1)
)
for topic_name in output_topic_names:
topic_out = Topic(participant, topic_name, data_type)
writers.append(DataWriter(participant, topic_out, qos))
frame = np.random.randint(0, 256, (1280, 1080, 3), dtype=np.uint8)
print(frame.shape, frame.dtype)
while True:
time_start = time.time()
ret, frame = 1, frame
if not ret:
print('Error: Unable to read frame')
break
if resize_to is not None:
frame = cv2.resize(frame, resize_to[::-1])
for writer in writers:
writer.write(data_type.from_numpy(frame))
loop_time = time.time() - time_start
print(f'Streaming Camera with topic {output_topic_names[0]}, {loop_time=}', end='\r')
if __name__ == '__main__':
fire.Fire(stream)
Subscriber
import cv2
import time
import numpy as np
import fire
from cyclonedds.core import Listener, Qos, Policy
from cyclonedds.util import duration
from cyclonedds.internal import dds_infinity
from cyclonedds.domain import DomainParticipant
from cyclonedds.topic import Topic
from cyclonedds.sub import DataReader
from dds_data_structures import MainCameraImage
def receive(topic_name='camera_images'):
data_type = MainCameraImage
participant = DomainParticipant(0)
qos = Qos(
# livelines: automatic
# deadline: dds_infinity
# reliability: reliable
# durability: volatile
# history: keep last
Policy.Liveliness.Automatic(lease_duration=duration(seconds=10)),
Policy.Deadline(deadline=dds_infinity),
Policy.Reliability.Reliable(max_blocking_time=duration(seconds=1)),
Policy.Durability.Volatile,
Policy.History.KeepLast(1)
)
topic_in = Topic(participant, topic_name, data_type)
reader = DataReader(participant, topic_in, qos)
while True:
time_start = time.time()
img = reader.read()
if len(img) == 0:
continue
img = img[0].to_numpy()
loop_time = time.time() - time_start
print(f'Received img {img.shape=}, {loop_time=}', end='\r')
if __name__ == '__main__':
fire.Fire(receive)
I've checked the requirements to the QoS for both publisher and subscriber here (as far as I can see they should be identical).
Next step I'm running iceoryx roudi in one docker container with --net=host --ipc=host -v /dev:/dev -v /tmp:/tmp. I also set the environment variable CYCLONEDDS_URI to the config file. I'm using default config file:
CYclonedds uri config
<?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/master/etc/cyclonedds.xsd">
<Domain Id="any">
<General>
<Interfaces>
<NetworkInterface autodetermine="true" priority="default" multicast="default" />
</Interfaces>
<AllowMulticast>default</AllowMulticast>
</General>
<SharedMemory>
<Enable>true</Enable>
<LogLevel>info</LogLevel>
</SharedMemory>
</Domain>
</CycloneDDS>
Finally I run above scripts in the docker containers with the same net and ipc options. They work fine and subscriber receives messages from publisher. What I'm trying to understand:
- How to make sure that messages are being sent via shared memory?
ipcs -mshows only firefox and vs code processes (I've checked pids via `ps -p ). Should my subscriber and publisher be there if everything works as expected?
Contributor guide
No contributing guide indexed for this repository
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Research direction
Start with the linked CycloneDDS shared-memory limitations page, the CYCLONEDDS_URI XML configuration, and the inline publisher and subscriber entry points. Run the containers with the shown network, IPC, and volume options, then inspect the documented diagnostics and shared-memory state; the documentation is done when it explains how to verify the transport and whether the processes should appear in ipcs -m.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- docker, python
- Domain
- distributed-systems, infrastructure
- Issue type
- Documentation
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 25/100