Filter out invalid messages that have the nanoseconds wrong for their cycle

Filter out messages with inconsistent nanoseconds and cycles

• Fix the checks in mbuf_probe gap_check to catch this case
• Add a filter to cps_recv
• Add an integration test
• extend fe_simulation to simulate a timing glitch that sets the wrong nano-seconds value

This is work towards #294

src/daqd/tests/test_daqd_timing_glitches_issue_294.py

+
+import json
+import os.path
+import urllib.request
+
+import integration
+
+integration.state.preserve_files()
+
+integration.Executable(name="fe_simulated_streams", hints=["../fe_stream_test/fe_simulation",], description="data stream generator")
+integration.Executable(name="local_dc", hints=['../local_dc'], description='local data concentrator')
+integration.Executable(name="cps_xmit", hints=['../pub_sub_stream'], description='data transport')
+integration.Executable(name="cps_recv", hints=['../pub_sub_stream'], description='data receiver')
+integration.Executable(name="daqd", hints=[], description="daqd")
+
+ini_dir = integration.state.temp_dir('ini_files')
+master_file = os.path.join(ini_dir, "master")
+testpoint_file = ""
+daqdrc_file = os.path.join(ini_dir, "daqdrc")
+
+integration.transform_text_file(input='daqdrc_live_test', output=daqdrc_file,
+                                substitutions=[('MASTER', master_file)])
+
+fe_simulated_streams = integration.Process("fe_simulated_streams",
+                                           ["-i", ini_dir,
+                                            "-M", master_file,
+                                            "-S",
+                                            "-m", "100",
+                                            "-k", "300",
+                                            "-D", "5,6,7,8,9",
+                                            "-admin", "127.0.0.1:10000",
+                                            ])
+
+local_dc_a = integration.Process("local_dc",
+                                 ["-b", "input_dc_a",
+                                  "-m", "100",
+                                  "-s", "mod5:5:2048"])
+
+local_dc_b = integration.Process("local_dc",
+                                 ["-b", "input_dc_b",
+                                  "-m", "100",
+                                  "-s", "mod6:6:2048 mod7:7:2048: mod8:8:2048 mod9:2048",
+                                  "-w", "20"])
+
+cps_xmit_a = integration.Process("cps_xmit",
+                               ["-b", "input_dc_a",
+                                "-m", "100",
+                                "-p", "tcp://127.0.0.1:9000",
+                                ])
+cps_xmit_b = integration.Process("cps_xmit",
+                                ["-b", "input_dc_b",
+                                 "-m", "100",
+                                 "-p", "tcp://127.0.0.1:9001",
+                                 ])
+
+cps_recv = integration.Process("cps_recv",
+                               ["-s", "tcp://127.0.0.1:9000 tcp://127.0.0.1:9001",
+                                "-b", "local_dc",
+                                "-m", "100",
+                                ])
+
+daqd = integration.Process("daqd", ["-c", daqdrc_file])
+
+
+def glitch_timing(dcuid):
+    def do_glitch():
+        payload = json.dumps({
+            'DeltaCycles': -1,
+            'Dcus': '{0}'.format(dcuid),
+        }).encode('utf-8')
+        req = urllib.request.Request("http://localhost:10000/api/v1/system/timing_glitch/",
+                               data=payload,
+                               headers={
+                                   'Content-Type': 'application/json',
+                               },
+                               method='POST')
+        with urllib.request.urlopen(req) as f:
+            obj = json.load(f)
+            if obj['status'] != 'ok':
+                raise RuntimeError('Unexpected response from fe_simulation')
+
+    return do_glitch
+
+
+# def daqd_not_running():
+#     state = daqd.state()
+#     return state not in (integration.Process.UNINITIALIZED, integration.Process.RUNNING)
+
+
+integration.Sequence(
+ [
+  integration.require_readable(files=["/dev/gpstime", "/dev/mbuf"], description="required device files"),
+  fe_simulated_streams.run,
+  integration.wait_tcp_server(port=10000),
+  local_dc_a.run,
+  local_dc_b.run,
+  integration.wait(1),
+  cps_xmit_a.run,
+  cps_xmit_b.run,
+  integration.wait(1),
+  cps_recv.run,
+  integration.wait(1),
+  daqd.run,
+  integration.wait_tcp_server(port=8088, timeout=15, pause_length=0.25),
+
+  integration.wait(12),
+  glitch_timing(5),
+  integration.wait(2),
+  integration.wait(0),
+  #integration.wait_for(predicate=daqd_not_running, description="daqd must fail", timeout=10),
+ ]
+)
+