Delayed postprocessing refactor with new monitoring variables

[denizergonul]

Description

• Delayed postprocessing algorithm is refactored because it was missing a critical consideration in timeout case. That is, timeout counts should be per item, a single count is not enough to represent all items' waiting times, which can vary.

• 4 new monitoring variables are added.

  uint64 num_postprocess_late_arrivals; // Number of data arrivals for an already closed post-processing window
  uint64 max_postprocess_tick_diff_to_newest; // Maximum tick difference between late arrival and the newest item in the buffer since last monitoring
  uint64 max_postprocess_tick_diff_to_next_window_start; // Maximum tick difference between late arrival and the next processing window start since last monitoring
  uint64 max_postprocess_tick_diff_to_last_processed; // Maximum tick difference between late arrival and the last processed item since last monitoring

• Added _ms suffix to wall-clock attributes of delayed postprocessing.

PS: These changes are meant for the next release. (Base branch will be changed to develop.)

datahandlinglibs dte/delayed_pp_monitoring
appmodel dte/delayedpp_attrs
daqsystemtest dte/delayedpp_attrs
snbmodules dte/delayedpp_attrs
ehn1-daqconfigs dte/delayedpp_attrs

Preliminary:

In the configuration, if DataHandlerConf.post_processing_delay_ticks > 0, postprocessing will be delayed.
This feature is introduced for data streams that may arrive out of order; such data is stored in sorted buffers.
Using this delay, we allow time for late data to arrive and be sorted so that it can still be processed in order.

If delay ticks is set, DHL spawns a dedicated thread pprocsched that runs the scheduler coroutine. The coroutine is woken up whenever data arrives.

In addition to delay ticks, one can also set DataHandlerConf.post_processing_delay_max_wait > 0. This enables the timeout feature of delayed postprocessing. With this feature, the coroutine wakes up not only when data arrives, but also if no data arrives within the specified duration.

One can also set DataHandlerConf.post_processing_delay_min_wait to enforce a minimum wait time before the next processing iteration.

Delayed postprocessing algorithm:

1. Return if buffer is empty.

2. Handle first cycle.

3. Return if postprocessing window is already closed.

   A (postprocessing) window is a subset of the buffer which will be processed.

   • It typically is only a part of the buffer.
   • In data arrival case, it can contain, at most, the entire buffer except this arrival.
   • In timeout case, it can even contain the entire buffer.

   Next window start is where to start processing in the next iteration. If it is greater than the newest item in the buffer, it means that the window was already closed.
   This happens if the entire buffer was already processed in a previous iteration and no newer data arrived since then.
   e.g. Buffer: {1, 3}. We already processed {1, 3}. Now {2} arrives. We notice this because next window start is {4} (last processed + 1) and 4 > 3.

4. In data arrival case, return if not enough time passed since last postprocessing.

5. Return if next window start cannot be found in the buffer. (Known issue with composite keys.)

6. Traverse the buffer and on the go:
   6.1. In timeout case, update (add 1 to) the timeout count of the item.
   6.2. Check if it is too early to process this item. If so, mark this point as window end and break the loop. (Window end will be the next window start.)
   6.3. Otherwise, process the item and remove its timeout count entry.

   Individual timeout count entries keep track of timeout counts of not-yet-processed items. This is required because not every item waits in the buffer the same amount of time. This number will be used to decide if this item can be processed.

   An item can be processed if it is "old enough" relative to the newest timestamp. The age difference must be bigger than the delay ticks we configure; otherwise, it is too early to process. Timeouts artificially make items older; that is why, on top of the (real) age difference, we add the virtual age of the item.

   Virtual age of an item is calculated by multiplying its timeout count by the max wait we configure.

7. Only in the timeout case, it is possible that the entire buffer was processed.
   7.1. If so, the loop didn't break and window end wasn't set. Set window end to last processed + 1.
   7.2. Otherwise, traverse the rest of the buffer to update the remaining timeout count s.

8. Make window end the next window start.

Addresses issue #116

Type of change

• New feature or enhancement (non-breaking change which adds functionality)
• Bug fix (non-breaking change which fixes an issue)

Testing checklist

• Unit tests pass (e.g. dbt-build --unittest)
• Minimal system quicktest passes (pytest -s minimal_system_quick_test.py)
• Full set of integration tests pass (dunedaq_integtest_bundle.sh)

May need to also clone: trigger, fdreadoutlibs and fdreadoutmodules.

dbt-build --unittest datahandlinglibs

The following configuration parameters should be set for TP, TA and/or TC. Numbers can be different.

post_processing_delay_ticks = 312500;
post_processing_delay_min_wait = 1;
post_processing_delay_max_wait = 5;

Example monitoring page

Further checks

• Performance comparison
• Code is commented where needed, particularly in hard-to-understand areas
• Code style is correct (dbt-build --lint, and/or see https://dune-daq-sw.readthedocs.io/en/latest/packages/styleguide/)
• If applicable, new tests have been added or an issue has been opened to tackle that in the future.
  (Indicate issue here: # (issue))

[roland-sipos]

TLOG promotion and mutex on fast path.