Message being drop even in TCP mode on single host?

[chengguizi]

May I know if there is any mechanism in eCAL subscriber to drop messages, even if they are on TCPpubsub transport?

I am doing a publisher at 200Hz or so, and subscribing it at another process. I am using an embedded ARM64 machine. I get frame drops. Is there anything can be tuned so I can get back the reliability that the TCP promises?

My message size is not big, less than 100 bytes per message.

[chengguizi]

Im seeing some API around QoS being reliable and best effort. Does that have an effect?

[rex-schilasky]

Hi @chengguizi so your processes run on different hosts I assume and you get drops even you switched to the TCP layer ? As mentioned in the TCP layer documentation the TCP layer will ensure that a message will be transmitted but if you overload that point to point connection (sending with a frequency that is higher then the subscriber is able to process) it will still act like a best effort QoS.

And regarding the QoS API .. we defined an API but there is no implementation behind. Many QoS mechanisms like Reliability or Durability would complicate the implementation of eCAL's pub/sub pattern or implicate additional latency for queuing data or doing handshakes.

[chengguizi]

Im running it on the same host and different processes! I tried SHM and TCP. They both have drops.
My subscriber loop i didnt write any heavy work. Just a dummy python test code.

I have already made sure the publisher is publishing at least 4ms apart by injecting thread sleep.

It appears to me the subscriber fail to wake up from wait or something

[chengguizi]

I can understand on SHM dropping frames if the subscriber callback is still running, make sense.
But why TCP is dropping as well? Seems less reasonable, I would assume the kernel should have helped in buffering the TCP packets.

But i will double check my subscriber again, maybe the deserialisation causing unexpected long latency.

[chengguizi]

For a bit more information. Here is the little script I am testing with on the subscriber side

#!/usr/bin/env python3

import sys
import time

import capnp
import numpy as np

import ecal.core.core as ecal_core
from byte_subscriber import ByteSubscriber

capnp.add_import_hook(['../src/capnp'])

import imu_capnp as eCALImu

last_seq = None
last_ecal_time = None

def callback(topic_name, msg, time_ecal):

    global last_seq
    global last_ecal_time

    time_now = time.time_ns() / 1e3

    

    # need to remove the .decode() function within the Python API of ecal.core.subscriber ByteSubscriber
    
    with eCALImu.Imu.from_bytes(msg) as imuMsg:
            
        force_print = False
        if last_seq is None:
            pass
        else:
            expected_seq = last_seq + imuMsg.seqIncrement
            if imuMsg.header.seq != expected_seq:
                print(f"WARN: actual seq {imuMsg.header.seq} and expected {expected_seq} mismatch, last seq {last_seq}")
                force_print = True

        if force_print or imuMsg.header.seq % 100 == 0:
            decoded_time = time.time_ns() / 1e3
            print(f"seq = {imuMsg.header.seq}, lapse {(time_now - time_ecal)} us, decoding {decoded_time - time_now} us")
            print(f"time interval since the last ecal message at sender {time_ecal - last_ecal_time} us")

        last_seq = imuMsg.header.seq
        last_ecal_time = time_ecal

def main():  

    # print eCAL version and date
    print("eCAL {} ({})\n".format(ecal_core.getversion(), ecal_core.getdate()))
    
    # initialize eCAL API
    ecal_core.initialize(sys.argv, "test_imu_sub")
    
    # set process state
    ecal_core.set_process_state(1, 1, "I feel good")

    # create subscriber and connect callback
    sub = ByteSubscriber("S0/imu")
    sub.set_callback(callback)
    
    # idle main thread
    while ecal_core.ok():
        time.sleep(0.1)
    
    # finalize eCAL API
    ecal_core.finalize()

if __name__ == "__main__":
    main()

The result I am getting:

seq = 2100, lapse 373.0 us, decoding 136.0 us
time interval since the last ecal message at sender 1645 us
seq = 2200, lapse 409.75 us, decoding 191.0 us
time interval since the last ecal message at sender 1630 us
WARN: actual seq 2296 and expected 2295 mismatch, last seq 2294
seq = 2296, lapse 1789.0 us, decoding 800.0 us
time interval since the last ecal message at sender 3773 us
seq = 2300, lapse 384.5 us, decoding 156.25 us
time interval since the last ecal message at sender 1469 us
seq = 2400, lapse 323.75 us, decoding 121.5 us
time interval since the last ecal message at sender 1239 us
seq = 2500, lapse 405.75 us, decoding 177.0 us
time interval since the last ecal message at sender 1676 us
WARN: actual seq 2569 and expected 2568 mismatch, last seq 2567
seq = 2569, lapse 1441.75 us, decoding 935.0 us
time interval since the last ecal message at sender 29164 us
seq = 2600, lapse 344.25 us, decoding 129.75 us
time interval since the last ecal message at sender 1235 us
seq = 2700, lapse 398.0 us, decoding 155.0 us
time interval since the last ecal message at sender 1519 us
seq = 2800, lapse 390.25 us, decoding 133.0 us
time interval since the last ecal message at sender 1241 us
seq = 2900, lapse 436.75 us, decoding 144.75 us
time interval since the last ecal message at sender 1488 us
WARN: actual seq 2971 and expected 2970 mismatch, last seq 2969
seq = 2971, lapse 1327.5 us, decoding 659.0 us
time interval since the last ecal message at sender 3233 us
seq = 3000, lapse 803.0 us, decoding 187.5 us
time interval since the last ecal message at sender 1904 us
seq = 3100, lapse 419.75 us, decoding 129.75 us
time interval since the last ecal message at sender 1799 us
seq = 3200, lapse 335.25 us, decoding 127.0 us
time interval since the last ecal message at sender 1225 us
seq = 3300, lapse 450.5 us, decoding 143.5 us
time interval since the last ecal message at sender 1629 us
WARN: actual seq 3332 and expected 3331 mismatch, last seq 3330
seq = 3332, lapse 1192.0 us, decoding 480.0 us
time interval since the last ecal message at sender 24899 us
seq = 3400, lapse 307.0 us, decoding 149.5 us
time interval since the last ecal message at sender 1593 us
seq = 3500, lapse 298.75 us, decoding 112.25 us
time interval since the last ecal message at sender 1201 us
seq = 3600, lapse 250.75 us, decoding 108.75 us
time interval since the last ecal message at sender 1229 us
seq = 3700, lapse 364.25 us, decoding 146.0 us
time interval since the last ecal message at sender 1610 us
WARN: actual seq 3711 and expected 3710 mismatch, last seq 3709
seq = 3711, lapse 1170.25 us, decoding 566.75 us
time interval since the last ecal message at sender 3806 us
seq = 3800, lapse 391.5 us, decoding 154.75 us
time interval since the last ecal message at sender 1554 us
WARN: actual seq 3807 and expected 3806 mismatch, last seq 3805
seq = 3807, lapse 1362.0 us, decoding 879.0 us
time interval since the last ecal message at sender 10542 us
seq = 3900, lapse 407.0 us, decoding 154.5 us
time interval since the last ecal message at sender 1197 us
CTCPReaderLayer - TCPPubSub (Error) -SubscriberSession 127.0.0.1:34700->radxa-cm3-io:40057: Error reading header length: End of file

[chengguizi]

I can observe that whenever missing message happens, there is larger delays in ecal_time. I believer that timestamp is written by the sender?

WARN: actual seq 3711 and expected 3710 mismatch, last seq 3709
seq = 3711, lapse 1170.25 us, decoding 566.75 us
time interval since the last ecal message at sender 3806 us

WARN: actual seq 3807 and expected 3806 mismatch, last seq 3805
seq = 3807, lapse 1362.0 us, decoding 879.0 us
time interval since the last ecal message at sender 10542 us

Tested on TCP transport

[rex-schilasky]

Hi, yes the timestamp is written by the publisher. In the publisher send call you can set your own time stamp as optional function parameter otherwise the eCAL::Time is used (what is system time is standard use cases).

[rex-schilasky]

So you are running on a single host and you get dropped messages caused by a slow subscription I assume. Can you please try out the new synchronized publishing API ? A synchronized send (send with a acknowledge timeout) will wait for the subscribers callback function execution before sending new messages.

class ProtoPublisher(MessagePublisher):
  """Spezialized publisher that sends out protobuf messages
  """

...

  def send_sync(self, msg, time, ack_timeout_ms):
    return self.c_publisher.send_sync(msg.SerializeToString(), time, ack_timeout_ms)

[chengguizi]

Sure, does this API work for capnproto also? I am using that currently. Also, is this synchronized send a different concept from the memfile_ack_timeout (knowing it only compatible with SHM layer)?

[rex-schilasky]

It's a publisher API so it works independent from the serialization format you use. The concept is exactly the same like setting the memfile_ack_timeout in the ecal.ini file. Using the ecal.ini file would configure this behavior as default for all publisher on that host.

[chengguizi]

Got it. One particular question: if I set memfile_ack_timeout to non-zero, is there a difference to

• do the normal CPublisher::Send

  ecal/ecal/core/src/pubsub/ecal_publisher.cpp

  Line 354 in 1607c9f

  size_t CPublisher::Send(const void* const buf_, const size_t len_, const long long time_ /* = -1 */) const

• compared to try to call CPublisher::SendSynchronized instead

  ecal/ecal/core/src/pubsub/ecal_publisher.cpp

  Line 384 in 1607c9f

  size_t CPublisher::SendSynchronized(const void* const buf_, size_t len_, long long time_, long long acknowledge_timeout_ms_) const

May I know if you are implying that the first one does not repsect the setting of memfile_ack_timeout in the ecal.ini file? If it is respeced, should I be better off using the Send API togeher with ecal.ini config?

[rex-schilasky]

If you set memfile_ack_timeout in the global ecal.ini configuration file then for every shm memory pub/sub connection this timeout is used as default for the handshake back from all subscribers to the matching publisher. That means in this case

size_t CPublisher::Send(const void* const buf_, const size_t len_, const long long time_ /* = -1 */) const

is acting like

size_t CPublisher::SendSynchronized(const void* const buf_, size_t len_, long long time_, memfile_ack_timeout) const

But CPublisher::SendSynchronized is always overwriting this default global behavior. You could do a call like this in theory

CPublisher::SendSynchronized(mybuffer, mybuffer_length, -1 /* = use system time*/, 0 /* = acknowledge_timeout_ms_*);

that would then act like CPublisher:Send() finally without any handshake (makes not that much sense to do that, but anyway).

[chengguizi]

Got it! So I am already using handshake mechanism as i do make the memfile_ack_timeout non-zero. Thanks!