Performance Testing with the CompactRIO system using NiDAQmx gRPC commands

[bhver3593]

Hello, I am experiencing latency issues when sending NiDAQmx gRPC commands from a computer to a CompactRIO system hosting a gRPC device server over an Ethernet cable. The CompactRIO has been chosen as the hardware device for this system and cannot be changed.

I've created a C++ wrapper to interact with the NI 9221 analog input card. During initialization, the wrapper connects to the gRPC server address, configures the AI channels and clock timing, and starts the task. For card parameters, the sample mode is set to continuous sampling acquisition, with a clock sampling rate set to 100 Samples/second per channel and an internal buffer size of 1. After setup, I can successfully read samples from all 8 AI card channels.

To measure the latency when reading analog input samples from all 8 channels, I created a simple performance test where I set up and call the gRPC client command ReadAnalogF64 1000 times while collecting 1 sample per channel. I also added a sleep of 10 milliseconds between each function call. Here are my performance results:

Average elapsed time: 1593.70 usecs

Highest elapsed times (latency spikes):
  95586 usecs
  93211 usecs
  92410 usecs
  92927 usecs
  94396 usecs
  94477 usecs
  92120 usecs
  93263 usecs
  89610 usecs
  87642 usecs

Number of function calls between latency spikes:
  86
  94
  94
  94
  94
  94
  94
  94
  94
  94

To resolve this issue, I've adjusted the timeout parameter for the NIDAQmx gRPC command, as well as other read parameters, but the latency spikes persist without any gRPC or NiDAQmx errors messages. However, updating the set_deadline function of the grpc::ClientContext resulted in a significant increase in latency spikes and flooding of gRPC error messages.

Interestingly, the number of function calls between spikes seems to be related to the sleep duration. For example:

• With a 10 msec sleep, there are approximately 100 function calls between spikes
• With a 20 msec sleep, there are approximately 50 function calls between spikes.

Testing with different sleep durations suggests that these latency spikes occur at a consistent rate. I'm looking for insights into the cause of these latency spikes and potential solutions to resolve them.

[bkeryan]

Hi @bhver3593,

The problem may not be related to gRPC at all. If possible, try running your test application on the CompactRIO controller. This may rule out gRPC and grpc-device entirely.

It sounds like you're using a continuous buffered acquisition, as described in Continuous Acquisition and Generation with Finite Buffer Size. The DAQmx task allocates a circular buffer. The CompactRIO chassis driver writes to this buffer, and the DAQmxReadAnalogF64 function reads from this buffer.

For best performance, the buffer size should be larger than the number of samples per read. This is important because both the reader and writer lock the portion of the buffer that they are accessing. If both the reader and writer lock the entire buffer, then the reader can't access the buffer while the writer is accessing it. However, if this was the problem, I think you would probably also get buffer underflow or buffer overwrite errors (unless you are configuring the Overwrite Mode).

When you call DAQmxReadAnalogF64, the DAQmx buffer reader waits for the data to be available in the buffer. There are a couple of properties to control this wait: Wait Mode and Sleep Time. The default wait mode is "sleep" and the default sleep time is 1 ms, so this also doesn't entirely explain the problem, but it may be one of multiple factors. Try changing the wait mode to "poll" or "wait for interrupt" and see if that helps.

Note that DAQmx has an alternative to buffered mode that is optimized for latency: Hardware Timed Single Point. This mode disables the circular buffer, but the wait mode and sleep time still apply. Hardware Timed Single Point (HWTSP) is commonly used with an explicit wait, performed using the DAQmxWaitForNextSampleClock function.

[bhver3593]

Hello @bkeryan,

My team is working on moving our test application over to the cRIO, but encountering some dependency issues. Hoping to get a test script running on the cRIO that can identify if gRPC or the gRPC server is causing the latency issue.

Modified the card parameters you mentioned (increased buffer size, changed the wait mode to 'poll' and 'wait for interrupt', and used the DAQmxWaitForNextSampleClock function), but all of these still resulted in similar latency spikes. No new gRPC or NiDAQmx error messages were presented for all cases.

Additionally, I tested a C++ card wrapper that communicates to the NI 9403 DIO card, where I call read and write commands to all 32 DIO card channels. For this performance test, I call the gRPC client command WriteDigitalU32 1000 times while writing 1 sample per channel. I also added a sleep of 10 milliseconds between each function call. Here are my performance results:

Average elapsed time: 2445.27 usecs

Highest elapsed times:
  100374 usecs
  94561 usecs
  99691 usecs
  99058 usecs
  92535 usecs
  95232 usecs
  89658 usecs
  95281 usecs
  92158 usecs
  89066 usecs
  98127 usecs
Number of calls between spikes:
  62
  88
  87
  87
  88
  88
  88
  88
  88
  88
  87

For this card, it seems there's a similar relationship between the sleep duration and the number of calls between latency spikes.

[bkeryan]

Some more thoughts:

What API are you using to sleep? If the client is on Windows and you are using the Win32 Sleep() function, the timer resolution is platform dependent and it may be as high as 16 ms per tick. If you are polling QueryPerformanceCounter (which I guess is not actually sleeping), the timer resolution is much more precise, but power management and multi-core CPUs have caused "time warps" with this in the past.

What happens if you remove the sleep and let DAQmxReadAnalogF64 do the sleeping for you?

Have you tried changing the sample rate? Do the latency spikes scale accordingly?

If running code on the chassis does not show latency spikes, then start looking at the gRPC communication:

• Capture the traffic and view it in Wireshark. Are there any TCP retries, checksum errors, etc.?
• Use netstat -s or a similar command to view the Linux kernel's TCP/IP statistics such as "segments retransmitted".

[bhver3593]

Hello @bkeryan,

Thank you for your suggestions. Unfortunately, we are unable to monitor traffic between systems. After discussing with the team, we created a test script for each of the 3 cases:

1. Linux computer sending NiDAQmx gRPC messages to the cRIO
2. Calling NiDAQmx gRPC commands directly on cRIO
3. Calling NiDAQmx C commands directly on the cRIO

For case 1, we are seeing these latency spikes occur at regular intervals, as mentioned above.
For case 2, we are having difficulty installing Python or C++ packages to get our test working.
For case 3, we are not seeing any latency spikes.

Provided is a Python test script I'm running that tests the performance of gRPC and TCP for Case 1. For gRPC, I call the read_digital_u32 NiDAQmx gRPC function call 1000x for the NI 9403 DIO card using a sleep of ~10 msecs between iterations:

NUM_ITERATIONS = 1000
MAX_DURATION_USECS = 5000
elapsed_times_usecs = []
highest_elapsed_times_usecs = []
iterations_between_spikes = []
num_iterations = 0
try:
    task = create_task()
    create_di_chan(task)
    start_task(task)

    for _ in range(NUM_ITERATIONS):
        num_iterations += 1
        start_time = time.time()
        read_digital_u32(task)
        duration = (time.time() - start_time) * 1e6
        elapsed_times_usecs += [duration]
        if(duration > MAX_DURATION_USECS):
            highest_elapsed_times_usecs += [duration]
            iterations_between_spikes += [num_iterations]
            num_iterations = 0

        time.sleep(0.010)

      print(f"Average elapsed time: {sum(elapsed_times_usecs) / NUM_ITERATIONS:.0f} usecs")
      print(f"Max elapsed time: {max(elapsed_times_usecs):.0f} usecs")
      print(f"Min elapsed time: {min(elapsed_times_usecs):.0f} usecs")
      print(f"\nElapsed times above {MAX_DURATION_USECS} usecs:")
      for high_time in highest_elapsed_times_usecs:
          print(f"    {high_time:.0f} usecs")
      print("Number of iterations between spikes:")
      for iterations in iterations_between_spikes:
          print(f"    {iterations}")

      stop_task(task)

Performance data:

Average elapsed time: 3658 usecs
Max elapsed time: 99059 usecs
Min elapsed time: 2045 usecs

Elapsed times above 5000 usecs:
    91226 usecs
    94503 usecs
    93307 usecs
    94502 usecs
    98626 usecs
    92309 usecs
    93335 usecs
    94933 usecs
    88284 usecs
    99059 usecs
    91093 usecs
    98812 usecs
    97029 usecs
Number of iterations between spikes:
    8
    80
    80
    80
    80
    80
    80
    81
    81
    81
    81
    81
    81

For the TCP connection, I used the same Python test with a similar byte size message being passed from the same Linux computer to the cRIO. This did not present with latency spikes:

Average elapsed time: 168 usecs
Max elapsed time: 512 usecs
Min elapsed time: 105 usecs

Elapsed times above 300 usecs:
    388 usecs
    512 usecs
    306 usecs
    314 usecs
    332 usecs
    309 usecs
    308 usecs
    304 usecs
Number of iterations between spikes:
    1
    48
    359
    13
    86
    55
    44
    494

Since the latency spikes related to the NiDAQmx gPRC functions are consistent and at regular intervals, our team has concluded it's related to our gRPC client configuration or the gRPC server configuration that was pre-set.
Other options we are exploring are gRPC streaming support or gRPC sideband.

Based on this information, are there other ways to address these latency spikes? Does someone have experience with setting gRPC server configurations and/or gRPC streaming support?

[bkeryan]

1. Linux computer sending NiDAQmx gRPC messages to the cRIO
2. Calling NiDAQmx gRPC commands directly on cRIO
3. Calling NiDAQmx C commands directly on the cRIO

For case 1, we are seeing these latency spikes occur at regular intervals, as mentioned above.
For case 2, we are having difficulty installing Python or C++ packages to get our test working.
For case 3, we are not seeing any latency spikes.

These are great steps towards isolating the problem.

By testing single-point DIO, you have removed analog input buffering from the picture. All of the DAQ-specific sources of latency I described in my previous reply are no longer possible. Your DIO test is explicitly starting the task, which eliminates another common source of latency (auto-start).

By testing the DAQmx C API running directly on the cRIO, you have confirmed that the issue is related to gRPC, NI gRPC Device Server, or networking in general.

For the TCP connection, I used the same Python test with a similar byte size message being passed from the same Linux computer to the cRIO. This did not present with latency spikes:

I'm not sure what you're describing here. Are you testing with a non-gRPC TCP client and server? That would rule out networking in general, and continue to point towards something specific about gRPC or NI gRPC Device Server.

Here are some ways to further isolate the problem:

• See if you can rule out the C Series hardware by sending a DAQmx RPC that does not access hardware. For example, with the task stopped and not reserved, setting a channel or timing attribute (such as Chan.Descr, AI.Max, DI.InvertLines, or SampClk.Rate) will not access hardware until you commit or start the task.
• See if you can rule out DAQmx by sending a non-DAQmx RPC, such as IsReservedByClient from https://github.com/ni/grpc-device/blob/main/source/protobuf/session_utilities.proto .

Note about support venue: I am a software developer at NI who has worked on both DAQmx and NI gRPC Device Server in the past. I am in touch with the Field Applications Engineer who has been helping you. Please continue to work closely with him so that he can reproduce your problem, escalate it as needed, and get it resolved.

[bhver3593]

Hello @bkeryan,

My apologies for the late reply. My team and I have been working to resolve the latency issue and have not had a chance to respond. I would like to provide an update on where we are at currently.

For identifying the gRPC performance issue, my team identified that these latency spikes are happening regardless of which API is being used, so it seems to be a non-DAQmx related issue. We added some logging to the gRPC Device Server and the latency spikes appear to be related to the TCP socket. For some reason, the server is not waking up for new data when the 100 millisecond latency occurs.

For the NI 9221 AI card wrapper that we developed in C++, we were able to completely eliminate latency spikes and significantly improve performance using gRPC data moniker streaming while referencing these documents:

• https://github.com/ni/grpc-device/wiki/Moniker-based-streaming-for-NI-Drivers
• https://github.com/ni/grpc-device/wiki/Moniker-streaming-using-gRPC-streams

Unfortunately for our NI 9403 DIO card wrapper, we have encountered some gRPC errors when setting up data streaming and reading digital/discrete signals. By default, it appears the DAQmx read functions for the DIO card are configured to finite sample acquisition. It seems that BeginReadDigitalU32() has to be called every time to retrieve another discrete sample during streaming, unlike the AI wrapper setup where continuous sampling using CfgSampClkTiming and BeginReadAnalogF64 functions are called on construction without issues. To resolve this, we tried adding continuous sampling to the DIO card wrapper but are continuing to receive gRPC errors related to "resource already in use".

Here is a simple Python script that mimics similar behaviors when trying to set up continuous sampling for a DIO card wrapper:

class DioWrapper:
    def __init__(self):
        self.read_task = self.create_task(name="read_digital_inputs")
        self.create_di_chan(self.read_task)
        self.cfg_samp_clk_timing(self.read_task)
        self.start_task(self.read_task)

        self.write_task = self.create_task(name="write_digital_outputs")
        self.create_do_chan(self.write_task)
        self.start_task(self.write_task)

    def __del__(self):
        self.stop_task(self.read_task)
        self.stop_task(self.write_task)
        self.clear_task(self.read_task)
        self.clear_task(self.write_task)

    def check_for_warning(self, response):
        """Print to console if the status indicates an error warning."""
        if response.status == 0:
            return

        message = client.GetErrorString(
            nidaqmx_types.GetErrorStringRequest(error_code=response.status)
        )
        if response.status > 0:
            print(f"Warning message: {message.error_string}, Warning status: {response.status}\n")
        elif response.status < 0:
            print(f"Error message: {message.error_string}, Error status: {response.status}\n")

    def create_task(self, name='my task'):
        response = client.CreateTask(
            nidaqmx_types.CreateTaskRequest(session_name=name)
        )
        self.check_for_warning(response)
        if response.new_session_initialized:
            self.stop_task(response.task)
            self.clear_task(response.task)

            response = client.CreateTask(
                nidaqmx_types.CreateTaskRequest(session_name=name)
            )
            self.check_for_warning(response)

        return response.task

    def start_task(self, task):
        response = client.StartTask(nidaqmx_types.StartTaskRequest(task=task))
        self.check_for_warning(response)

    def stop_task(self, task):
        response = client.StopTask(nidaqmx_types.StopTaskRequest(task=task))
        self.check_for_warning(response)

    def clear_task(self, task):
        response = client.ClearTask(nidaqmx_types.ClearTaskRequest(task=task))
        self.check_for_warning(response)

    def create_do_chan(self, task):
        response = client.CreateDOChan(
            nidaqmx_types.CreateDOChanRequest(
                task=task, lines=LINES_DO, line_grouping=nidaqmx_types.LINE_GROUPING_CHAN_FOR_ALL_LINES
            )
        )
        self.check_for_warning(response)

    def create_di_chan(self, task):
        response = client.CreateDIChan(
            nidaqmx_types.CreateDIChanRequest(
                task=task, lines=LINES_DI, line_grouping=nidaqmx_types.LINE_GROUPING_CHAN_FOR_ALL_LINES
            )
        )
        self.check_for_warning(response)

    def cfg_samp_clk_timing(self, task):
        response = client.CfgSampClkTiming(
            nidaqmx_types.CfgSampClkTimingRequest(
                task=task,
                rate=1000.0,
                active_edge=nidaqmx_types.EDGE1_RISING,
                sample_mode=nidaqmx_types.ACQUISITION_TYPE_CONT_SAMPS,
                samps_per_chan=10,
            )
        )
        self.check_for_warning(response)


def main():
    card = DioWrapper()


if __name__ == "__main__":
    main()

Error messages:

grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with:
        status = StatusCode.UNKNOWN
        details = "Resource requested by this task has already been reserved by a different task with conflicting settings.
Unreserve any other tasks using this device, or change their settings to be compatible with this task.

Task Name: write_digital_outputs

Status Code: -201105"
        debug_error_string = "UNKNOWN:Error received from peer  {grpc_message:"Resource requested by this task has already been reserved by a different task with conflicting settings.\nUnreserve any other tasks using this device, or change their settings to be compatible with this task.\n\nTask Name: write_digital_outputs\n\nStatus Code: -201105", grpc_status:2, created_time:"2025-03-31T16:31:55.621446184+00:00"}"

Based on the NI DAQmx C reference documents, it looks like the DAQmxReadDigitalU32 function supports continuous sampling. We are wondering if the NI 9403 DIO card supports continuous sampling when you configure some port lines as inputs and some port lines as outputs? If there is a correct way to set up continuous sampling for this wrapper?