sf-daq/sf_daq_broker
0
0
Fork 0
mirror of https://github.com/paulscherrerinstitute/sf_daq_broker.git synced 2026-08-12 00:00:28 +02:00
Code Issues Packages Projects Releases Wiki Activity

Refactor image assembly to expose socket creation

Browse Source
This commit is contained in:
babic_a
2020-11-17 13:52:06 +01:00
parent 28c0674b68
commit 8675e05c64
1 changed files with 11 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sf_daq_broker/detector/image_assembler.py
+11 -6
View File
@@ -23,12 +23,7 @@ class ImageAssembler(object):
raise ValueError("RamBuffer must have at least 5 slots.")
for module_id in range(self.n_modules):
# We use the PUSH/PULL mechanism to moderate disk throughput.
receiver = self.zmq_context.socket(zmq.PULL)
# No buffering on send side - receiver dictates reading speed.
receiver.setsockopt(zmq.RCVHWM, zmq_rcv_hwm)
# If in 1 second nothing was received we have a problem in reading.
receiver.setsockopt(zmq.RCVTIMEO, 1000)
receiver = get_pull_receiver(self.zmq_context, zmq_rcv_hwm=zmq_rcv_hwm)
receiver.connect("inproc://%s" % module_id)
@@ -97,3 +92,13 @@ class ImageAssembler(object):
self.receivers.clear()
def get_pull_receiver(context, zmq_rcv_hwm):
# We use the PUSH/PULL mechanism to moderate disk throughput.
receiver = context.socket(zmq.PULL)
# No buffering on send side - receiver dictates reading speed.
receiver.setsockopt(zmq.RCVHWM, zmq_rcv_hwm)
# If in 1 second nothing was received we have a problem in reading.
receiver.setsockopt(zmq.RCVTIMEO, 1000)
return receiver