Add threading to host-side comms server

Author: solidpixel

lgl_android_install.py

@@ -606,19 +606,21 @@ def configure_server(conn: ADBConnect,
         profile_dir: The desired output directory path for timeline. Existing
             files in the directory may be overwritten.
     '''
+    verbose = False
+
     # Create a server instance
-    instance = server.CommsServer(0)
+    instance = server.CommsServer(0, verbose)
 
     if timeline_file:
         # Import late to avoid pulling in transitive deps when unused
         from lglpy.comms import service_gpu_timeline
-        service_tl = service_gpu_timeline.GPUTimelineService(timeline_file)
+        service_tl = service_gpu_timeline.GPUTimelineService(timeline_file, verbose)
         instance.register_endpoint(service_tl)
 
     if profile_dir:
         # Import late to avoid pulling in transitive deps when unused
         from lglpy.comms import service_gpu_profile
-        service_prof = service_gpu_profile.GPUProfileService(profile_dir)
+        service_prof = service_gpu_profile.GPUProfileService(profile_dir, verbose)
         instance.register_endpoint(service_prof)
 
     # Start it running

lglpy/comms/server.py

@@ -22,19 +22,23 @@
 # -----------------------------------------------------------------------------
 
 '''
-This module implements the server-side communications module that can accept
-client connections from a layer driver, and dispatch messages to registered
-service handler in the server.
-
-This module currently only accepts a single connection at a time and message
-handling is synchronous inside the server. It is therefore not possible to
-implement pseudo-host-driven event loops if the layer is using multiple
-services concurrently - this needs threads per service.
+This module implements the server-side of the communications module that can
+accept connections from client layer drivers running on the device. The
+protocol is service-based, and the server will dispatch messages to the
+registered service handler for each message channel.
+
+The server is multi-threaded, allowing multiple layers to concurrently access
+networked services provided by host-side implementations. However, within each
+client connection messages are handled synchronously by a single worker thread.
+It is therefore not possible to implement pseudo-host-driven event loops if a
+layer is using multiple services concurrently - this needs threads per service
+endpoint which is not yet implemented.
 '''
 
 import enum
 import socket
 import struct
+import threading
 from typing import Any, Optional
 
 
@@ -143,7 +147,7 @@ class CommsServer:
     Class listening for client connection from a layer and handling messages.
 
     This implementation is designed to run in a thread, so has a run method
-    that will setup and listen on the server socket.q
+    that will setup and listen on the server socket.
 
     This implementation only handles a single layer connection at a time, but
     can handle multiple connections serially without restarting.
@@ -173,7 +177,6 @@ def __init__(self, port: int, verbose: bool = False):
         self.register_endpoint(self)
 
         self.shutdown = False
-        self.sockd = None  # type: Optional[socket.socket]
 
         self.sockl = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
         self.sockl.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
@@ -185,6 +188,9 @@ def __init__(self, port: int, verbose: bool = False):
         # Work out which port was assigned if not user-defined
         self.port = self.sockl.getsockname()[1]
 
+        # Pool of worker threads
+        self.workers: list[threading.Thread] = []
+
     def register_endpoint(self, endpoint: Any) -> int:
         '''
         Register a new service endpoint with the server.
@@ -235,55 +241,55 @@ def run(self) -> None:
             if self.verbose:
                 print('Waiting for client connection')
             try:
-                self.sockd, _ = self.sockl.accept()
+                sockd, _ = self.sockl.accept()
                 if self.verbose:
                     print('  + Client connected')
 
-                self.run_client()
-
-                if self.verbose:
-                    print('  + Client disconnected')
-                self.sockd.close()
-                self.sockd = None
-
-            except ClientDropped:
-                if self.verbose:
-                    print('  + Client disconnected')
-                if self.sockd:
-                    self.sockd.close()
-                    self.sockd = None
+                thread = threading.Thread(target=self.run_client, args=(sockd,))
+                self.workers.append(thread)
+                thread.start()
 
             except OSError:
                 continue
 
         self.sockl.close()
 
-    def run_client(self) -> None:
+    def run_client(self, sockd: socket.socket) -> None:
         '''
         Enter client message handler run loop.
 
         Raises:
             ClientDropped: The client disconnected from the socket.
         '''
-        while not self.shutdown:
-            # Read the header
-            data = self.receive_data(Message.HEADER_LEN)
-            message = Message(data)
-
-            # Read the payload if there is one
-            if message.payload_size:
-                data = self.receive_data(message.payload_size)
-                message.add_payload(data)
-
-            # Dispatch to a service handler
-            endpoint = self.endpoints[message.endpoint_id]
-            response = endpoint.handle_message(message)
+        try:
+            while not self.shutdown:
+                # Read the header
+                data = self.receive_data(sockd, Message.HEADER_LEN)
+                message = Message(data)
+
+                # Read the payload if there is one
+                if message.payload_size:
+                    data = self.receive_data(sockd, message.payload_size)
+                    message.add_payload(data)
+
+                # Dispatch to a service handler
+                endpoint = self.endpoints[message.endpoint_id]
+                response = endpoint.handle_message(message)
+
+                # Send a response for all TX_RX messages
+                if message.message_type == MessageType.TX_RX:
+                    header = Response(message, response)
+                    self.send_data(sockd, header.get_header())
+                    self.send_data(sockd, response)
+
+        except ClientDropped:
+            pass
+
+        finally:
+            if self.verbose:
+                print('  + Client disconnected')
 
-            # Send a response for all TX_RX messages
-            if message.message_type == MessageType.TX_RX:
-                header = Response(message, response)
-                self.send_data(header.get_header())
-                self.send_data(response)
+            sockd.close()
 
     def stop(self) -> None:
         '''
@@ -294,14 +300,29 @@ def stop(self) -> None:
         if self.sockl is not None:
             self.sockl.close()
 
-        if self.sockd is not None:
-            self.sockd.shutdown(socket.SHUT_RDWR)
+        self.wait_for_workers()
 
-    def receive_data(self, size: int) -> bytes:
+    def wait_for_workers(self) -> None:
+        '''
+        Wait for workers
+        '''
+        # Wait for each worker - slightly convoluted logic as workers threads
+        # remove themselves from the list as they complete so there is a race
+        # between self.workers test and testing a worker in the loop
+        while self.workers:
+            try:
+                worker = self.workers[0]
+                worker.join()
+            except IndexError:
+                pass
+
+    @staticmethod
+    def receive_data(sockd: socket.socket, size: int) -> bytes:
         '''
         Fetch a fixed size packet from the socket.
 
         Args:
+            sockd: The data socket.
             size: The length of the packet in bytes.
 
         Returns:
@@ -310,31 +331,29 @@ def receive_data(self, size: int) -> bytes:
         Raises:
             ClientDropped: The client disconnected from the socket.
         '''
-        assert self.sockd is not None
-
         data = b''
         while len(data) < size:
-            new_data = self.sockd.recv(size - len(data))
+            new_data = sockd.recv(size - len(data))
             if not new_data:
                 raise ClientDropped()
             data = data + new_data
 
         return data
 
-    def send_data(self, data: bytes) -> None:
+    @staticmethod
+    def send_data(sockd: socket.socket, data: bytes) -> None:
         '''
         Send a fixed size packet to the socket.
 
         Args:
+            sockd: The data socket.
             data: The binary data to send.
 
         Raises:
             ClientDropped: The client disconnected from the socket.
         '''
-        assert self.sockd is not None
-
         while len(data):
-            sent_bytes = self.sockd.send(data)
+            sent_bytes = sockd.send(data)
             if not sent_bytes:
                 raise ClientDropped()
             data = data[sent_bytes:]