Add prototypes for benchmark testing. #94

YooSunYoung · 2023-09-22T15:17:42Z

There are 2 minimum prototype daemons with dummy workflows for benchmark testings.
This PR does not have benchmarking code. It only has unit tests of each prototypes.

Review points

prototype_mini.StopWatch is used to benchmark the process.
Three main applications, DataStreamListener/DataReduction/Visualization of prototypes.
Dummy workflows in workflows.

Minimum prototype

These are the minimum components of the real-time data reduction testing.

tests/prototypes/prototype_mini.py
tests/prototypes/parameters.py
tests/prototypes/random_data_providers.py
tests/prototypes/workflows.py.

prototype_mini can also be run as a script, python -m tests.prototypes.prototype_mini.
It simulates the data stream between async daemons.
The dummy workflow contains binning, transform coordinates, and histogram.

Kafka streaming tests

tests/prototypes/prototype_kafka.py contains the same daemons as prototype_mini.py but with real kafka api and protobuf serialization/deserialization process.
kafka broker should be running in the same machine in order to run this test.

Kafka related tests filter.

kafka related tests will not be included in the CI test (for now) since it requires kafka broker running in the same machine.
Those tests can be filtered by requesting kafka_test fixture.
kafka_test fixture will then check if there was a --kafka-test flag included as an argument of pytest command and then skip the test if it was not.
You can include those tests by pytest --kafka-test.

The reason for using `fixture` and not `mark`.

Adding extra marks is also an option to include/exclude marked tests.
It should be configured in the pytest.ini.
fixture was used for now since mark has limitation of how to decide whether it is skipping the test or not.

YooSunYoung · 2023-09-22T15:26:31Z

tests/prototypes/prototype_mini.py

+        ``Traceback`` of ``pytest`` did not show
+        where exactly the error is from in those cases.
+        It was resolved by using ``get_event_loop``,
+        or manually closing the event loop at the end of the test.


Test case to reproduce similar error:

async def some_coroutine() -> bool: import asyncio for _ in range(100): try: raise RuntimeError except RuntimeError: await asyncio.sleep(0.01) return True def wrong_way_of_running_async_calls(): import asyncio new_loop = asyncio.new_event_loop() return new_loop.run_until_complete(some_coroutine()) def test_foo(): assert wrong_way_of_running_async_calls() def test_foo1(): assert wrong_way_of_running_async_calls() def test_foo2(): assert wrong_way_of_running_async_calls()

YooSunYoung · 2023-09-22T16:06:52Z

Script for deleting all randomly generated topic with prefix of BEAMLIMETEST.

from confluent_kafka.admin import ClusterMetadata

def filter_beamlime_test_topics(cluster_meta: ClusterMetadata) -> list[str]:
    test_topic_prefix = "BEAMLIMETEST"
    return [topic for topic in cluster_meta.topics if topic.startswith(test_topic_prefix)]

if __name__ == "__main__":
    import time

    from confluent_kafka.admin import AdminClient
    
    admin = AdminClient({'bootstrap.servers': 'localhost:9092'})
    topic_list = filter_beamlime_test_topics(admin.list_topics())
    admin.list_topics().topics
    
    if topic_list:
        print("Deleting all existing topics: ", topic_list)
        admin.delete_topics(topic_list)

        while (topic_list:=filter_beamlime_test_topics(admin.list_topics())):
            time.sleep(1)
        
        print('Test topics all deleted.')
    else:
        print("No test topics to delete.")

nvaytet

I have to admit I don't understand how all the different parts come together here, I would have to spend much longer looking at this to understand it all.

However, I had a superficial look and tried to spot things/patterns that may seem odd.

About:

kafka related tests will not be included in the CI test (for now) since it requires kafka broker running in the same machine.

It would be nice to get this up and running soon, because there is a lot of code that pertains to Kafka that is currently not being automatically tested (we rely on the contributors to run those tests locally, which is quite dangerous).

nvaytet · 2023-09-25T13:01:45Z

tests/prototypes/prototype_kafka.py

+
+
+def provide_kafka_producer(broker_address: KafkaBootstrapServer) -> Producer:
+    return Producer({'bootstrap.servers': broker_address})


I think it's ok, but just making sure there is no typo here: 'bootstrap.servers' is the same key for both Producer and AdminClient.

Yes...! This key is minimum requirement so it raises an error if it's not correct.
I'm not sure how to handle these configuration at the moment...

tests/prototypes/prototype_kafka.py

nvaytet · 2023-09-25T13:05:11Z

tests/prototypes/prototype_kafka.py

+        import time
+
+        admin.create_topics([NewTopic(topic)])
+        time.sleep(0.1)


Add a comment as to why the sleep is needed?

tests/prototypes/prototype_kafka.py

nvaytet · 2023-09-26T07:06:36Z

tests/prototypes/prototype_mini.py

+        """
+        Collect all coroutines of daemons and schedule them into the event loop.
+
+        Notes


nvaytet · 2023-09-26T07:12:44Z

tests/prototypes/random_data_providers.py

+
+
+def provide_time_coords(rng: RNG, ef_rate: EventFrameRate) -> TimeCoords:
+    dummy_zeros = [zr for zr in range(int(13620492e11), int(13620492e11) + ef_rate)]


where does the number 13620492e11 come from?

It's just a random datetime I picked... 2013-02-28T12:00:00 : D.... Actually event_time_zero was not used in this dummy workflow. Maybe it should be...

tests/prototypes/random_data_providers.py

nvaytet · 2023-09-26T07:19:46Z

tests/prototypes/workflows.py

+    return ReducedData(
+        binned.transform_coords(
+            ['L', 'wavelength'],
+            graph={


Is that the same graph as on L49? Can we avoid the duplication?
Maybe this requires c_a, c_b, and c_c to also be part of the graph, but maybe that is a good thing?

Yes probably...?
so the idea was to keep workflow_script from L36 to L58 to be a through workflow script from the input to the output,
and the rest is granular steps of that workflow, probably run by sciline later.
So they are meant to be duplicated.
But the graph might be separated and re-used I agree.

nvaytet · 2023-09-26T07:20:53Z

tests/prototypes/workflows.py

+    c_a: ConstantA = default_c_a,
+    c_b: ConstantB = default_c_b,
+    c_c: ConstantC = default_c_c,


Not sure about having default values here?

YooSunYoung · 2023-09-26T14:01:51Z

It would be nice to get this up and running soon, because there is a lot of code that pertains to Kafka that is currently not being automatically tested (we rely on the contributors to run those tests locally, which is quite dangerous).

Yes I agree.
I briefly searched how to run an integration test with kafka stream, and it might be just as simple as setting a secret in the github repository to access to a running kafka cluster for testing.
For now, it's always being used for benchmark so I will not spend too much time on it. I opened an issue #97 though...!

SimonHeybrock · 2023-10-05T09:37:38Z

tests/prototypes/workflows.py

+    graph = provide_coord_transform_graph()
+
+    transformed = binned.transform_coords(['L', 'wavelength'], graph=graph)
+    return transformed.hist(wavelength=histogram_bin_size).sum('L')


For instrument like DREAM we need thousands of bins. This will run out of memory, instead we will have to use something like

return transformed.bins.concat('L').hist(wavelength=histogram_bin_size)

YooSunYoung · 2023-10-12T14:32:43Z

tests/prototypes/workflows.py

+            'frame_offset': lambda event_time_zero: event_time_zero - first_pulse_time,
+            'time_offset_pivot': time_offset_pivot,
+            'tof': tof_from_time_offset,
+            'wavelength': wavelength_from_tof,


@SimonHeybrock I added converting event_time_offset, event_time_zero to tof in this graph.
Or should I just use scippneutron ...?

Haven't run the whole set of benchmarks yet though.

Yes, we absolutely should use scippneutron. If we start replicating logic here we will run into big problems, since things will never be in sync.

YooSunYoung · 2023-10-12T14:36:12Z

tests/prototypes/prototype_mini.py

+    def process_first_data(self, data: Events) -> None:
+        sample_event = data[0]
+        first_pulse_time = sample_event.coords['event_time_zero'][0]
+        self.workflow.providers[FirstPulseTime] = lambda: first_pulse_time


DataReductionApp now has separate process_first_data method so that it can retrieve the FirstPulseTime (which is the event_time_zero of the first incoming data in this case).

And then it updates the workflow providers to return the retrieved value.

SimonHeybrock

I mainly looked at the places where data is touched/transformed, I believe @nvaytet took care of reviewing the other code?

SimonHeybrock · 2023-10-16T11:08:30Z

tests/prototypes/prototype_kafka.py

+        from streaming_data_types.eventdata_ev44 import deserialise_ev44
+
+        data = deserialise_ev44(self.merge_bytes(data_list))
+        event_zeros = np.full(len(data.pixel_id), data.reference_time[0])


Initializing a full numpy array and copying into a variable is a bit inefficient, should directly init a scipp variable. Use scipp.full?

SimonHeybrock · 2023-10-16T11:14:27Z

tests/prototypes/workflows.py

+            'frame_offset': lambda event_time_zero: event_time_zero - first_pulse_time,
+            'time_offset_pivot': time_offset_pivot,
+            'tof': tof_from_time_offset,
+            'wavelength': wavelength_from_tof,


Yes, we absolutely should use scippneutron. If we start replicating logic here we will run into big problems, since things will never be in sync.

SimonHeybrock · 2023-10-16T11:15:17Z

tests/prototypes/workflows.py

+    frame_rate: FrameRate,
+) -> Histogrammed:
+    merged = sc.concat(da_list, dim='event')
+    pixel_ids = sc.arange(dim='pixel_id', start=0, stop=num_pixels)


This will be used for every processed chunk, can we avoid recreating it all the time?

This script is for processing all data at once, not per chunk.
But as it seems confusing, I just deleted it...!

SimonHeybrock · 2023-10-16T11:15:54Z

tests/prototypes/workflows.py

+    pixel_ids = sc.arange(dim='pixel_id', start=0, stop=num_pixels)
+    binned = merged.group(pixel_ids)
+
+    graph = provide_coord_transform_graph(frame_rate)


Why is this recreated every time?

SimonHeybrock · 2023-10-16T11:17:28Z

tests/prototypes/workflows.py

+
+def provide_workflow(
+    num_pixels: NumPixels, histogram_binsize: HistogramBinSize, frame_rate: FrameRate
+) -> Workflow:


All the code here seems to be duplicating what is in workflow_script above. Are both used?

No it was more for showing what it is doing.
I thought it was useful since this workflow doesn't have a nice documentation with graphs... but maybe it is more confusing than useful. I'll remove it.

Co-authored-by: Neil Vaytet <[email protected]>

…s into the graph

Co-authored-by: Neil Vaytet <[email protected]>

…op timestamp

…with VisualizationDaemon

…ni prototype

YooSunYoung

I'm not sure if this is what you meant... @SimonHeybrock but I replaced the existing graph with frame unwrapping helper from scippneutron package.

YooSunYoung · 2023-10-31T14:55:24Z

tests/prototypes/workflows.py

+
+def provide_workflow(
+    num_pixels: NumPixels, histogram_binsize: HistogramBinSize, frame_rate: FrameRate
+) -> Workflow:


No it was more for showing what it is doing.
I thought it was useful since this workflow doesn't have a nice documentation with graphs... but maybe it is more confusing than useful. I'll remove it.

YooSunYoung · 2023-10-31T14:57:04Z

tests/prototypes/workflows.py

+    frame_rate: FrameRate,
+) -> Histogrammed:
+    merged = sc.concat(da_list, dim='event')
+    pixel_ids = sc.arange(dim='pixel_id', start=0, stop=num_pixels)


This script is for processing all data at once, not per chunk.
But as it seems confusing, I just deleted it...!

YooSunYoung commented Sep 22, 2023

View reviewed changes

This was referenced Sep 22, 2023

Prototypes for benchmarking #91

Closed

Rough Sketch #68

Closed

nvaytet reviewed Sep 26, 2023

View reviewed changes

SimonHeybrock reviewed Oct 5, 2023

View reviewed changes

YooSunYoung commented Oct 12, 2023

View reviewed changes

YooSunYoung requested a review from SimonHeybrock October 16, 2023 08:27

SimonHeybrock reviewed Oct 16, 2023

View reviewed changes

Base automatically changed from provider-arguments-hashable to main October 31, 2023 09:11

YooSunYoung and others added 18 commits October 31, 2023 15:54

Add prototypes for benchmark testing.

3c7651a

Add minimum prototype

410a17e

Fix string fomatting

ccd1a01

Update random data provider

e23a826

Update workflow to use transform_coords.

55edaaa

Update tests/prototypes/prototype_kafka.py

fd6dfa4

Co-authored-by: Neil Vaytet <[email protected]>

Update tests/prototypes/prototype_kafka.py

7f5b490

Co-authored-by: Neil Vaytet <[email protected]>

Update tests/prototypes/prototype_kafka.py

45693b9

Co-authored-by: Neil Vaytet <[email protected]>

Update tests/prototypes/prototype_kafka.py

fcb82b2

Co-authored-by: Neil Vaytet <[email protected]>

Update tests/prototypes/random_data_providers.py

399035f

Co-authored-by: Neil Vaytet <[email protected]>

Wait for topic to be created with retry helper

892391d

Update tests/prototypes/prototype_mini.py

3b243bd

Co-authored-by: Neil Vaytet <[email protected]>

Separate coordinate transformation graph and integrate constant value…

e9b27f5

…s into the graph

Apply automatic formatting

77120ab

Add docstring about retrieving type arguments

3b7771d

Add indicator for duplicating description in docstring.

28a0b31

Rename parent class and remove unnecessary inheritance

c15c816

Update tests/prototypes/prototype_mini.py

44d6bf8

Co-authored-by: Neil Vaytet <[email protected]>

YooSunYoung and others added 20 commits October 31, 2023 15:54

Update tests/prototypes/prototype_mini.py

b8090e3

Co-authored-by: Neil Vaytet <[email protected]>

Rename daemon apps

167c72c

Update async event loop handling

e9cc12c

Update data counting in kafka listener

0f1d17c

Add duration property to the stop watch instead of providing start/st…

9ee04ec

…op timestamp

Adapt singleton provider changes.

728fd98

Move lap counts and benchmark report to StopWatch instead of bind it …

4e039ab

…with VisualizationDaemon

Update stop watch target count test according to the update of the mi…

6521bdb

…ni prototype

Update time coordinate to provide realistic time offset zero

ecbc99c

Update dummy event dumping function

13b1b29

Fix function argument

4d4d981

Fix function argument

936b23c

Add event_time_zero, event_time_offset to tof in the workflow

f173b27

Avoid using too much memory

2baa5af

Add wrap-up method in the data reduction app

ed678c0

Fix type hint

4db7ae0

Add scipy to the testing environment requirements

2b2d774

Add scippneutron to extra dependency

35bd4a1

Remove unecessary workflow script

50715a8

Update workflow to use frame unwrapping from scippneutron package

f9593b8

YooSunYoung force-pushed the prototypes branch from 36d2040 to f9593b8 Compare October 31, 2023 15:45

YooSunYoung commented Oct 31, 2023

View reviewed changes

SimonHeybrock approved these changes Nov 1, 2023

View reviewed changes

YooSunYoung merged commit 3590141 into main Nov 1, 2023

YooSunYoung deleted the prototypes branch November 1, 2023 08:17

This was referenced Nov 9, 2023

Include frame-unwrapping in benchmarking #88

Closed

Data Stream Simulator #20

Closed

Data Stream Listener data binning. #47

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add prototypes for benchmark testing. #94

Add prototypes for benchmark testing. #94

YooSunYoung commented Sep 22, 2023 •

edited

Loading

YooSunYoung Sep 22, 2023

YooSunYoung commented Sep 22, 2023

nvaytet left a comment

nvaytet Sep 25, 2023

YooSunYoung Sep 26, 2023

nvaytet Sep 25, 2023

nvaytet Sep 26, 2023

nvaytet Sep 26, 2023

YooSunYoung Sep 26, 2023

nvaytet Sep 26, 2023

YooSunYoung Sep 26, 2023

nvaytet Sep 26, 2023

YooSunYoung commented Sep 26, 2023

SimonHeybrock Oct 5, 2023

YooSunYoung Oct 12, 2023

SimonHeybrock Oct 16, 2023

YooSunYoung Oct 12, 2023 •

edited

Loading

SimonHeybrock left a comment

SimonHeybrock Oct 16, 2023

SimonHeybrock Oct 16, 2023

SimonHeybrock Oct 16, 2023

YooSunYoung Oct 31, 2023

SimonHeybrock Oct 16, 2023

SimonHeybrock Oct 16, 2023

YooSunYoung Oct 31, 2023

YooSunYoung left a comment

YooSunYoung Oct 31, 2023

YooSunYoung Oct 31, 2023



		def provide_kafka_producer(broker_address: KafkaBootstrapServer) -> Producer:
		return Producer({'bootstrap.servers': broker_address})



		def provide_time_coords(rng: RNG, ef_rate: EventFrameRate) -> TimeCoords:
		dummy_zeros = [zr for zr in range(int(13620492e11), int(13620492e11) + ef_rate)]

Add prototypes for benchmark testing. #94

Add prototypes for benchmark testing. #94

Conversation

YooSunYoung commented Sep 22, 2023 • edited Loading

Review points

Minimum prototype

Kafka streaming tests

Kafka related tests filter.

The reason for using fixture and not mark.

Choose a reason for hiding this comment

YooSunYoung commented Sep 22, 2023

nvaytet left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

YooSunYoung commented Sep 26, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

YooSunYoung Oct 12, 2023 • edited Loading

Choose a reason for hiding this comment

SimonHeybrock left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

YooSunYoung left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

YooSunYoung commented Sep 22, 2023 •

edited

Loading

The reason for using `fixture` and not `mark`.

YooSunYoung Oct 12, 2023 •

edited

Loading