diff --git a/CHANGES.rst b/CHANGES.rst
index a94bc5cdb90795af6d111b283ec798c184753015..3e05d3ddf2b1adf3e01416205cd610c4cd47a4a4 100644
--- a/CHANGES.rst
+++ b/CHANGES.rst
@@ -17,6 +17,10 @@ Changelog
- Allow SoG pipeline to be tested with MDC events.
+- Fix race condition of multiple instances of the RAVEN pipeline by
+ including a GraceDB poll of the superevent state before updating joint
+ FAR.
+
2.0.2 "Flying Icarus" (2022-12-23)
----------------------------------
diff --git a/gwcelery/tasks/raven.py b/gwcelery/tasks/raven.py
index 5d46e55183324337f7fbe26974c9e4045abbab73..ecd8efb63d1349492d2a9ef11f487119339dcec0 100644
--- a/gwcelery/tasks/raven.py
+++ b/gwcelery/tasks/raven.py
@@ -261,30 +261,32 @@ def update_coinc_far(coinc_far_dict, superevent, ext_event):
external event dictionary
"""
+ # Get graceids
+ superevent_id = superevent['superevent_id']
+ ext_id = ext_event['graceid']
+
# Joint FAR isn't computed for SNEWS coincidence
# Choose SNEWS coincidence over any other type of coincidence
if ext_event['pipeline'] == 'SNEWS':
- superevent_id = superevent['superevent_id']
- ext_id = ext_event['graceid']
gracedb.update_superevent(superevent_id, em_type=ext_id,
time_coinc_far=None,
space_coinc_far=None)
return coinc_far_dict
+ # Get the latest info to prevent race condition
+ superevent_f = gracedb.get_superevent(superevent_id)
+
# Load needed variables
infty = float('inf')
new_time_far = coinc_far_dict['temporal_coinc_far']
new_space_far = coinc_far_dict['spatiotemporal_coinc_far']
# Map None to infinity to make logic easier
new_space_far_f = new_space_far if new_space_far else infty
- old_time_far = superevent['time_coinc_far']
+ old_time_far = superevent_f['time_coinc_far']
old_time_far_f = old_time_far if old_time_far else infty
- old_space_far = superevent['space_coinc_far']
+ old_space_far = superevent_f['space_coinc_far']
old_space_far_f = old_space_far if old_space_far else infty
- superevent_id = superevent['superevent_id']
- ext_id = ext_event['graceid']
-
if new_space_far_f < old_space_far_f or \
(new_time_far < old_time_far_f and old_space_far_f == infty):
gracedb.update_superevent(superevent_id, em_type=ext_id,
diff --git a/gwcelery/tests/test_tasks_raven.py b/gwcelery/tests/test_tasks_raven.py
index b8a8ecf5e6f75cd435d3a2661907e9c3987fa8e3..fe7fa1fcf9354ce7d68098baf331a5a3e5cc37f0 100644
--- a/gwcelery/tests/test_tasks_raven.py
+++ b/gwcelery/tests/test_tasks_raven.py
@@ -166,6 +166,30 @@ def mock_coinc_far(*args):
'overlap_integral': None}
+def mock_get_superevent(superevent_id):
+ if superevent_id == 'S1':
+ old_time_far = None
+ old_space_far = None
+ elif superevent_id == 'S2':
+ old_time_far = 1e-5
+ old_space_far = None
+ elif superevent_id == 'S3':
+ old_time_far = 1e-4
+ old_space_far = None
+ elif superevent_id == 'S4':
+ old_time_far = 1e-5
+ old_space_far = 1e-6
+ elif superevent_id == 'S5':
+ old_time_far = 1e-4
+ old_space_far = 1e-6
+ else:
+ old_time_far = 1e-5
+ old_space_far = None
+ return {'time_coinc_far': old_time_far,
+ 'space_coinc_far': old_space_far,
+ 'superevent_id': superevent_id}
+
+
@pytest.mark.parametrize(
'raven_search_results,graceid,tl,th,group',
[[[{'graceid': 'E1', 'pipeline': 'GRB'}], 'S1', -5, 1, 'CBC'],
@@ -189,6 +213,7 @@ def mock_coinc_far(*args):
@patch('gwcelery.tasks.gracedb.update_superevent')
@patch('gwcelery.tasks.gracedb.create_label.run')
@patch('gwcelery.tasks.external_skymaps.plot_overlap_integral.run')
+@patch('gwcelery.tasks.gracedb.get_superevent', mock_get_superevent)
def test_raven_pipeline(mock_plot_overlap_integral,
mock_create_label,
mock_update_superevent,
@@ -298,24 +323,23 @@ def test_preferred_superevent(raven_search_results, testnum):
@pytest.mark.parametrize(
- 'new_time_far,new_space_far,old_time_far,old_space_far,pipeline,result',
- [[1e-4, None, None, None, 'Fermi', True],
- [1e-4, 1e-3, None, None, 'Swift', True],
- [1e-4, None, 1e-5, None, 'AGILE', False],
- [1e-4, 1e-3, 1e-4, None, 'Fermi', True],
- [1e-4, 1e-3, 1e-5, 1e-6, 'Fermi', False],
- [1e-8, None, 1e-5, 1e-6, 'Swift', False],
- [1e-4, 1e-8, 1e-4, 1e-6, 'AGILE', True],
- [None, None, None, None, 'SNEWS', True]])
+ 'new_time_far,new_space_far,superevent_id,pipeline,result',
+ [[1e-4, None, 'S1', 'Fermi', True],
+ [1e-4, 1e-3, 'S1', 'Swift', True],
+ [1e-4, None, 'S2', 'AGILE', False],
+ [1e-4, 1e-3, 'S3', 'Fermi', True],
+ [1e-4, 1e-3, 'S4', 'Fermi', False],
+ [1e-8, None, 'S4', 'Swift', False],
+ [1e-4, 1e-8, 'S5', 'AGILE', True],
+ [None, None, 'S1', 'SNEWS', True]])
@patch('gwcelery.tasks.gracedb.update_superevent')
+@patch('gwcelery.tasks.gracedb.get_superevent', mock_get_superevent)
def test_update_superevent(mock_update_superevent,
new_time_far, new_space_far,
- old_time_far, old_space_far,
+ superevent_id,
pipeline, result):
- superevent = {'time_coinc_far': old_time_far,
- 'space_coinc_far': old_space_far,
- 'superevent_id': 'S100'}
+ superevent = {'superevent_id': superevent_id}
ext_event = {'graceid': 'E100',
'pipeline': pipeline}
coinc_far_dict = {'temporal_coinc_far': new_time_far,
@@ -323,7 +347,7 @@ def test_update_superevent(mock_update_superevent,
raven.update_coinc_far(coinc_far_dict, superevent, ext_event)
if result:
mock_update_superevent.assert_called_with(
- 'S100', em_type='E100',
+ superevent_id, em_type='E100',
time_coinc_far=new_time_far,
space_coinc_far=new_space_far)
else: