Improve synchronisation for parent-child messages

This new model will have the parent process wait until IocInit() has
successfully completed before continuing with the test

The child process had need to do a queue.put() followed immediately by a
queue.get(). On most platforms the main process would interuupt by
pulling from the queue, but this was not guaranteed. To solve this using
queues would have required a second queue for signalling.

We work around this by using Pipes instead. These provide the
read-write semantics we want - the ability to wait for an explicit message
from the parent process - without needing a side-channel for more
signalling.

Hopefully this will fix the MacOS issues where it seemed caget was
returning values from a previous iteration of the tests, possibly
because the test process was running much faster than the child process
could keep up with.

Author: AlexanderWells-diamond

tests/test_records.py

@@ -293,7 +293,7 @@ class GetValueEnum(Enum):
     CAGET = 2
 
 
-def run_ioc(creation_func, initial_value, queue, set_enum):
+def run_ioc(creation_func, initial_value, conn, set_enum, get_enum):
     """Creates a record and starts the IOC. `initial_value` will be set on
     the record at different times based on the `set_enum` parameter."""
     kwarg = {}
@@ -316,15 +316,21 @@ def run_ioc(creation_func, initial_value, queue, set_enum):
     builder.LoadDatabase()
     softioc.iocInit(dispatcher)
 
+    conn.send("IOC started")
+
     if set_enum == SetValueEnum.SET_AFTER_INIT:
         out_rec.set(initial_value)
 
-    if queue is not None:
-        queue.put(out_rec.get())
+    if get_enum == GetValueEnum.GET:
+        conn.send(out_rec.get())
         # Some tests need to do a caput and then cause another .get() to happen
-        val = queue.get(timeout=TIMEOUT)
-        if val:
-            queue.put(out_rec.get())
+        if set_enum == SetValueEnum.CAPUT:
+            if conn.poll(TIMEOUT):
+                val = conn.recv()
+                if val is not None:
+                    conn.send(out_rec.get())
+
+    conn.close()
 
     # Keep process alive while main thread works.
     asyncio.run_coroutine_threadsafe(
@@ -343,17 +349,21 @@ def run_test_function(
 
     creation_func, initial_value, expected_value, expected_type = record_values
 
-    queue = None
-    if get_enum == GetValueEnum.GET:
-        queue = multiprocessing.Queue()
+    parent_conn, child_conn = multiprocessing.Pipe()
 
     ioc_process = multiprocessing.Process(
         target=run_ioc,
-        args=(creation_func, initial_value, queue, set_enum)
+        args=(creation_func, initial_value, child_conn, set_enum, get_enum)
     )
 
     ioc_process.start()
 
+    # Wait for message that IOC has started
+    if parent_conn.poll(TIMEOUT):
+        parent_conn.recv()
+    else:
+        pytest.fail("IOC process did not start before TIMEOUT expired")
+
     # Infer some required keywords from parameters
     put_kwargs = {}
     get_kwargs = {}
@@ -375,24 +385,30 @@ def run_test_function(
         _channel_cache.purge()
 
         if set_enum == SetValueEnum.CAPUT:
-            if queue:
-                queue.get(timeout=TIMEOUT)
+            if get_enum == GetValueEnum.GET:
+                if parent_conn.poll(TIMEOUT):
+                    parent_conn.recv()
+                else:
+                    pytest.fail("IOC did not provide initial record value")
             caput(
                 DEVICE_NAME + ":" + RECORD_NAME,
                 initial_value,
                 wait=True,
                 **put_kwargs)
 
-            if queue:
-                queue.put("Do another get!")
+            if get_enum == GetValueEnum.GET:
+                parent_conn.send("Do another get!")
             # Ensure IOC process has time to execute.
             # I saw failures on MacOS where it appeared the IOC had not
             # processed the put'ted value as the caget returned the same value
             # as was originally passed in.
             Yield(timeout=TIMEOUT)
 
         if get_enum == GetValueEnum.GET:
-            rec_val = queue.get(timeout=TIMEOUT)
+            if parent_conn.poll(TIMEOUT):
+                rec_val = parent_conn.recv()
+            else:
+                pytest.fail("IOC did not provide record value in queue")
         else:
             rec_val = caget(
                 DEVICE_NAME + ":" + RECORD_NAME,