--merge-every-nth counts total messages, not pulses?

[SimonHeybrock]

Not sure this is what is going on, but I have a file with 3 NXevent_data groups. When I use --merge-every-nth=2 it will only pass the first two to the data reduction workflow. I would have expected it to load two pulse from each group?

This is for the Loki@Larmo file.

[jokasimr]

This is for the Loki@Larmo file

This file? ess.loki.data.loki_tutorial_sample_run_60339().

Yes the current version emits a new message every n-th message received.
I does seem more sensible to always emit messages containing a whole number of pulses.

But the details are a bit fuzzy, especially in the presence of several NXevent_data groups.
If merge-every-nth=2, do we emit a new message when we have received two pulses from each of the detectors?
Or do we emit a new message when we have received two pulses from any one of the detectors?

[YooSunYoung]

I'm not sure if we can automate them at all.
Because we never know how many detectors are flushing how many messages per pulse,
and the order of messages are also not guaranteed (most likely to be sorted since we only have one consumer).

We can set another parameter that tells the handler how many messages are expected per pulse,
but then it's easier to just multiply it by the number of pulses we want and then use it as merge-every-nth.

Or, we can maybe wait for all detector sources to receive message with a certain timeout...?

[SimonHeybrock]

I think we should probable look at the event_time_zero of the messages (or just the wallclock time in Beamlime?), and instead of merging every N frames, merge messages for X seconds?

[YooSunYoung]

I think we should probable look at the event_time_zero of the messages (or just the wallclock time in Beamlime?), and instead of merging every N frames, merge messages for X seconds?

We already have an option max-seconds-between-messages.

[SimonHeybrock]

What does that do? It is not clear from the name.

[YooSunYoung]

What does that do? It is not clear from the name.

If given, the handler accumulate messages for max-seconds-between-messages seconds.

It's passed to this function:

beamlime/src/beamlime/applications/handlers.py

Lines 62 to 79 in f3263c3

	def maxcount_or_maxtime(maxcount: Number, maxtime: Number):
	if maxcount <= 0:
	raise ValueError("maxcount must be positive")
	if maxtime <= 0:
	raise ValueError("maxtime must be positive")
	count = 0
	last = time.time()
	def run():
	nonlocal count, last
	count += 1
	if count >= maxcount or time.time() - last >= maxtime:
	count = 0
	last = time.time()
	return True
	return run

Actually, the helper message seems wrong...

what would you call it then...?

[jokasimr]

ev44 has a field that determines order, so in principle we can handle messages arriving out of order. But it might be a bit tricky.
Do we have more than one Kafka partition per topic? If we don't then messages are guaranteed to be ordered.

[YooSunYoung]

Do we have more than one Kafka partition per topic

Yes, I asked ECDC and all topics has(and will have) multiple partitions as recommended for performance. So they are not guaranteed to be ordered.
We cannot guarantee the order unless we only flush the data after a while, since we can't wait for a missing piece, cause we simply can't determine if anything's missing.

But in reality, most of event data will arrive pretty much in order.

[jokasimr]

Not sure how to do this right now, so putting it back into the selected pile.