DAQ Mode Introduces ~60ms Delay Every Few Readings – How to Disable Buffering for Real-Time Data? (XCPlite)

[mo-m2c]

I’m using pyxcp-0.22.30 in DAQ mode to read ECU measurements from the XCPlite (sin, ramp) thrugh UDP, localhost:5555, but I’m observing a consistent delay of ~60ms every few readings, likely due to buffering or ODT alignment. I need real-time data with minimal latency (no buffering).

RESULTS:

DAQ: [-89.47944519652036, 30800]
sw Δt = 0.061028 s <-- ~60ms
DAQ: [-87.9703023027843, 30801]
sw Δt = 0.000118 s
DAQ: [-86.32179748941594, 30802]
sw Δt = 0.000073 s
DAQ: [-84.67834102928106, 30803]
sw Δt = 0.000067 s
DAQ: [-82.96020441965837, 30804]
sw Δt = 0.061525 s <-- ~60ms
DAQ: [-81.06045530220246, 30805]
sw Δt = 0.000162 s
DAQ: [-79.12423173563438, 30806]
sw Δt = 0.000077 s
DAQ: [-77.11963286955341, 30807]
sw Δt = 0.000053 s
DAQ: [-75.06745117118118, 30808]
sw Δt = 0.061665 s <-- ~60ms
DAQ: [-72.85943196690958, 30809]
sw Δt = 0.000302 s
DAQ: [-70.60426521015314, 30810]
sw Δt = 0.000091 s
DAQ: [-68.25796963782166, 30811]
sw Δt = 0.000069 s
DAQ: [-65.83768589039937, 30812]
sw Δt = 0.061501 s <-- ~60ms
DAQ: [-63.409210402584485, 30813]

sw Δt represents the elapsed time between two consecutive measurements. In theory, XCPlite’s "mainloop" should sample every 10 ms, but in DAQ mode I’m seeing a consistent pattern of four readings:

1. Three samples arrive almost instantaneously (~0.05–0.1 ms apart, instead of 10 ms).
2. The fourth sample is delayed by roughly 60 ms.
   This repeats every four readings, which strongly suggests DAQ is buffering four 10 ms samples internally and then sending them in one burst.
   If the data were truly sent in real time, the ECU’s packet counter would jump by several counts on the delayed transmission (reflecting the missed 10 ms intervals), yet it only increments by one.

By contrast, when I switch to simple polling mode, the ramp signal advances by 100 counts per second (one sample every 10 ms), confirming that the ECU can deliver data at the required rate—so the buffering must be occurring within the DAQ/ODT stack.

Questions for the Developer: Is there a hidden config parameter to disable DAQ buffering entirely?

[mo-m2c]

"""
@file daq_test.py
@brief Script to run DAQ lists via XCP and print both hardware and software timestamp intervals.

Enables XCP timestamps, reconstructs 64-bit timestamp values from ts0/ts1,
and computes the true sample interval (hw Δt) in seconds between successive packets.
Also tracks the arrival interval in software (sw Δt) using perf_counter().
"""

import time
from pyxcp.cmdline import ArgumentParser
from pyxcp.daq_stim import DaqList, DaqOnlinePolicy

# -----------------------------------------------------------------------------
# Configuration: Define your DAQ lists here
# -----------------------------------------------------------------------------
# Replace HW_TIME_UNIT with the time unit from your A2L (in seconds per tick)
HW_TIME_UNIT = 1e-6  # seconds per timestamp tick

DAQ_LISTS = [
    DaqList(
        name="App1toRTE10msSiga",      # channel name
        event_num=0,                     # event channel on the slave
        stim=False,                      # disable stimulation
        enable_timestamps=True,         # enable hardware timestamp
        measurements=[
            ("channel1", 0x1CEF78, 0, "F64"),  # 64-bit float at address 0x1CEF78
            ("counter", 0x1CEF80, 0, "U16"),
        ],
        priority=0,                      # highest priority
        prescaler=0                     # no prescaling
    )
]

class DaqToTerminal(DaqOnlinePolicy):
    """
    Policy printing hardware and software Δt between DAQ packets.
    """
    def __init__(self, daq_lists):
        """
        Constructor initializes DAQ lists and timestamp trackers.

        :param daq_lists: List of DaqList instances.
        """
        super().__init__(daq_lists)
        self._last_hw_ts = None       # last hardware timestamp (ticks)
        self._last_sw_time = None     # last software timestamp (perf_counter)
        self._user_callback = None

    def add_callback(self, callback):
        """
        Register user callback to process payloads.

        :param callback: Function taking payload list.
        """
        self._user_callback = callback

    def initialize(self):
        """
        Called once just before DAQ starts.
        """
        print("DaqToTerminal -> initialize()")
        super().initialize()

    def on_daq_list(self, daq_list, ts0, ts1, payload):
        """
        Callback for each incoming DAQ packet.

        Computes and prints the hardware Δt between successive packets
        and software Δt between callback invocations.

        :param daq_list: Index of the DAQ list.
        :param ts0: Lower 32 bits of hardware timestamp ticks.
        :param ts1: Upper 32 bits of hardware timestamp ticks.
        :param payload: Measurement values per DaqList definition.
        """
        # Reconstruct 64-bit hardware timestamp
        hw_ts = (ts1 << 32) | ts0
        if self._last_hw_ts is not None:
            dt_hw = (hw_ts - self._last_hw_ts) * HW_TIME_UNIT
            #print(f"hw Δt = {dt_hw:.6f} s")
        self._last_hw_ts = hw_ts

        # Software arrival interval
        sw_now = time.perf_counter()
        if self._last_sw_time is not None:
            dt_sw = sw_now - self._last_sw_time
            print(f"sw Δt = {dt_sw:.6f} s")
        self._last_sw_time = sw_now

        # Forward to user callback
        if self._user_callback:
            self._user_callback(payload)

    def finalize(self):
        """
        Called when DAQ is stopped or session ends.
        """
        print("DaqToTerminal -> finalize()")
        super().finalize()

# -----------------------------------------------------------------------------
# User callback example
# -----------------------------------------------------------------------------
def send2matlab(payload: list):
    """
    Example callback: print payload for MATLAB.

    :param payload: List of measurement values.
    """
    print(f"DAQ: {payload}")

# -----------------------------------------------------------------------------
# Main execution
# -----------------------------------------------------------------------------
if __name__ == "__main__":
    ap = ArgumentParser(description="DAQ test printing hardware and software Δt")
    daq_policy = DaqToTerminal(DAQ_LISTS)
    daq_policy.add_callback(send2matlab)

    with ap.run(policy=daq_policy) as x:
        x.connect()
        x.cond_unlock("DAQ")
        print("Setting up DAQ lists...")
        daq_policy.setup()
        print("Starting DAQ lists...")
        daq_policy.start()

        # Stream data for specified duration
        duration = 1  # seconds
        print(f"Streaming for {duration} seconds...\n")
        time.sleep(duration)

        print("\nStopping DAQ...")
        daq_policy.stop()
        print("DAQ stopped.")
        x.disconnect()

[christoph2]

@mo-m2c
Ok, your code is basically working, now I start my investigations...

[christoph2]

@mo-m2c
Well, I think it's a documentation issue in the first place (shame on me);

• ts0 is the timestamp created by pyxcp.
• ts1 is the DAQ timestamp (if available and enabled) from your ECU/simulator (contained in the first ODT).

i.e. these are counters from different sources.

[mo-m2c]

@christoph2
While timestamped data (ts1) works well for post-test analysis and plotting, I am experiencing significant issues during real-time operation. The buffering behavior creates noticeable stuttering when streaming data to MATLAB/Simulink scopes, making real-time visualization unreliable. This becomes particularly problematic when using the live data stream for HIL test cases that dynamically adjust inputs based on measurements.

[christoph2]

The DaqOnlinePolicy has no buffering itself -- I rather suspect Python for timing issues.
A multi-threaded TCP/IP transport-layer written in C++ is on my todo-list, resp. exists in parts (I did some experiments with PTP).

[mo-m2c]

If I let the DAQ run for a full minute, the data I’m getting is already about 20 seconds behind real time. If my buffering theory is right, that lag makes real-time HIL testing basically impossible. The simulation would always be chasing outdated measurements. Is there a way to drop or skip any stale/buffered DAQ packets so I only process the freshest sample on each 10 ms cycle?

[christoph2]

@mo-m2c
Well, the ODTs that constitute a DAQ list are collected until a DAQ list is filled and then immediately handed over to on_daq_list of the choosen policy. There is no DAQ list buffering required and implemented.
Both timestamp values are not directly related to the wall clock; the value delivered by XCPlite is (as far as I remember) the time since program start.

Have you also tryied the DaqRecorder policy?

A closer look at the timestamps

pyxcp_ts = np.diff(df.ts0)
xcplite_ts = np.diff(df.ts1)

revealed nothing unusual for me.

[mo-m2c]

After running some tests, I’ve noticed that XCPlite’s timing is heavily influenced by the system’s CPU load. I see jitter on the sine and ramp signals when the machine is busy (e.g.: during Teams presentation). While the ramp is designed for 100 iterations/second (10ms cycleTime), actual performance reaches only ~70 iterations/second because of simulation timing inconsistencies. I will test it on other system to confirm the XCPlite performance issue.

But still, if we can eliminate the recurring burst behavior (the ~60 ms delay every fourth sample) it would already be a significant improvement for real-time HIL testing.