WorkerPool: Wait for previous task in _try_send_to_primary_thread

In order to prevent tasks from running in a non-main thread,
wait for the previous task inside _try_send_to_primary_thread,
then schedule the next task.

Closes: pytest-dev#96

src/execnet/gateway_base.py

@@ -345,7 +345,11 @@ def integrate_as_primary_thread(self):
             with self._running_lock:
                 if self._shuttingdown:
                     break
-                primary_thread_task_ready.clear()
+                # Only clear if _try_send_to_primary_thread has not
+                # yet set the next self._primary_thread_task reply
+                # after waiting for this one to complete.
+                if reply is self._primary_thread_task:
+                    primary_thread_task_ready.clear()
 
     def trigger_shutdown(self):
         with self._running_lock:
@@ -371,11 +375,19 @@ def _try_send_to_primary_thread(self, reply):
         # note that we should be called with _running_lock hold
         primary_thread_task_ready = self._primary_thread_task_ready
         if primary_thread_task_ready is not None:
-            if not primary_thread_task_ready.is_set():
-                self._primary_thread_task = reply
-                # wake up primary thread
-                primary_thread_task_ready.set()
-                return True
+            if (
+                primary_thread_task_ready.is_set()
+                and self._primary_thread_task is not None
+            ):
+                self._primary_thread_task.waitfinish()
+            self._primary_thread_task = reply
+            # wake up primary thread (it's okay if this is already set
+            # because we waited for the previous task to finish above
+            # and integrate_as_primary_thread will not clear it when
+            # it enters self._running_lock if it detects that a new
+            # task is available)
+            primary_thread_task_ready.set()
+            return True
         return False
 
     def spawn(self, func, *args, **kwargs):