Speed up tests (#686)

* speed up tests

* fix fake socket test race

* fix fake socket setup order

* wait for vipc listener startup

* Revert "wait for vipc listener startup"

This reverts commit b96129a556.

* reverts

* down to 0.2s

* fix that

* don't need that

msgq/ipc_pyx.pyx

@@ -2,6 +2,7 @@
 # cython: c_string_encoding=ascii, language_level=3
 
 import sys
+import time
 from libcpp.string cimport string
 from libcpp.vector cimport vector
 from libcpp cimport bool
@@ -249,3 +250,11 @@ cdef class PubSocket:
 
   def all_readers_updated(self):
     return self.socket.all_readers_updated()
+
+  def wait_for_readers(self, double timeout=1.0, double interval=0.001):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+      if self.all_readers_updated():
+        return
+      time.sleep(interval)
+    raise TimeoutError("subscriber did not connect")

msgq/tests/test_messaging.py

@@ -17,8 +17,7 @@ class TestPubSubSockets(unittest.TestCase):
     sock = random_sock()
     pub_sock = msgq.pub_sock(sock)
     sub_sock = msgq.sub_sock(sock, conflate=False, timeout=None)
-
-    for _ in range(1000):
+    for _ in range(100):
       msg = random_bytes()
       pub_sock.send(msg)
       recvd = sub_sock.receive()
@@ -36,7 +35,6 @@ class TestPubSubSockets(unittest.TestCase):
         msg = random_bytes()
         pub_sock.send(msg)
         sent_msgs.append(msg)
-      time.sleep(0.1)
       recvd_msgs = msgq.drain_sock_raw(sub_sock)
       if conflate:
         assert len(recvd_msgs) == 1
@@ -48,7 +46,7 @@ class TestPubSubSockets(unittest.TestCase):
 
   def test_receive_timeout(self):
     sock = random_sock()
-    timeout_ms = 50
+    timeout_ms = 5
     sub_sock = msgq.sub_sock(sock, timeout=timeout_ms)
 
     start_time = time.monotonic()

msgq/tests/test_poller.py

@@ -1,5 +1,4 @@
 import unittest
-import time
 import msgq
 import concurrent.futures
 
@@ -30,7 +29,7 @@ class TestPoller(unittest.TestCase):
     with concurrent.futures.ThreadPoolExecutor() as e:
       poll = e.submit(poller)
 
-      time.sleep(0.1)  # Slow joiner syndrome
+      pub.wait_for_readers()
 
       # Send message
       pub.send(b"a")
@@ -52,13 +51,11 @@ class TestPoller(unittest.TestCase):
     with concurrent.futures.ThreadPoolExecutor() as e:
       poll = e.submit(poller)
 
-      time.sleep(0.1)  # Slow joiner syndrome
+      pub.wait_for_readers()
       c = msgq.Context()
       for _ in range(10):
         msgq.SubSocket().connect(c, SERVICE_NAME)
 
-      time.sleep(0.1)
-
       # Send message
       pub.send(b"a")
 
@@ -95,7 +92,7 @@ class TestPoller(unittest.TestCase):
     sub = msgq.SubSocket()
     sub.connect(context, SERVICE_NAME)
 
-    time.sleep(0.1)  # Slow joiner
+    pub.wait_for_readers()
 
     for i in range(1, 100):
       pub.send(b'a'*i)
@@ -127,7 +124,7 @@ class TestPoller(unittest.TestCase):
     sub = msgq.SubSocket()
     sub.connect(context, SERVICE_NAME, conflate=True)
 
-    time.sleep(0.1)  # Slow joiner
+    pub.wait_for_readers()
     pub.send(b'a')
     pub.send(b'b')
 

msgq/visionipc/tests/test_visionipc.py

@@ -99,5 +99,5 @@ class TestVisionIpc(unittest.TestCase):
     assert recv_buf is not None
     assert self.client.frame_id == 2
 
-    recv_buf = self.client.recv()
+    recv_buf = self.client.recv(timeout_ms=5)
     assert recv_buf is None

msgq/visionipc/visionipc_client.cc

@@ -14,9 +14,13 @@
 static int connect_to_vipc_server(const std::string &name, bool blocking) {
   const std::string ipc_path = get_ipc_path(name);
   int socket_fd = ipc_connect(ipc_path.c_str());
+  bool logged_retry = false;
   while (socket_fd < 0 && blocking) {
-    std::cout << "VisionIpcClient connecting" << std::endl;
-    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+    if (!logged_retry) {
+      std::cout << "VisionIpcClient connecting" << std::endl;
+      logged_retry = true;
+    }
+    std::this_thread::sleep_for(std::chrono::milliseconds(20));
     socket_fd = ipc_connect(ipc_path.c_str());
   }
   return socket_fd;

msgq/visionipc/visionipc_server.cc

@@ -105,7 +105,11 @@ void VisionIpcServer::listener(){
 
     VisionStreamType type = VisionStreamType::VISION_STREAM_MAX;
     int r = ipc_sendrecv_with_fds(false, fd, &type, sizeof(type), nullptr, 0, nullptr);
-    assert(r == sizeof(type));
+    if (r != sizeof(type)) {
+      close(fd);
+      if (should_exit) break;
+      continue;
+    }
 
     // send available stream types
     if (type == VisionStreamType::VISION_STREAM_MAX) {
@@ -184,7 +188,13 @@ void VisionIpcServer::send(VisionBuf * buf, VisionIpcBufExtra * extra, bool sync
 
 VisionIpcServer::~VisionIpcServer(){
   should_exit = true;
-  listener_thread.join();
+  if (listener_thread.joinable()) {
+    int sock = ipc_connect(get_ipc_path(name).c_str());
+    if (sock >= 0) {
+      close(sock);
+    }
+    listener_thread.join();
+  }
 
   // VisionBuf cleanup
   for (auto const& [type, buf] : buffers) {