[bitbake-devel] [PATCH 10/16] hashserv: Use separate threads for answering requests and handling them

[Richard Purdie]

Experience with the prserv shows that having two threads, one accepting
and queueing connections and one handling the requests leads to much
more reliable behaviour than having everything in a single thread.

Signed-off-by: Richard Purdie <richard.purdie at linuxfoundation.org>
---
 lib/hashserv/__init__.py | 41 +++++++++++++++++++++++++++++++++++++++-
 1 file changed, 40 insertions(+), 1 deletion(-)

diff --git a/lib/hashserv/__init__.py b/lib/hashserv/__init__.py
index 544bc86b16..86aa7e9de6 100644
--- a/lib/hashserv/__init__.py
+++ b/lib/hashserv/__init__.py
@@ -10,6 +10,9 @@ import sqlite3
 import json
 import traceback
 import logging
+import socketserver
+import queue
+import threading
 from datetime import datetime
 
 logger = logging.getLogger('hashserv')
@@ -135,6 +138,41 @@ class HashEquivalenceServer(BaseHTTPRequestHandler):
             self.send_error(400, explain=traceback.format_exc())
             return
 
+class ThreadedHTTPServer(HTTPServer):
+    quit = False
+
+    def serve_forever(self):
+        self.requestqueue = queue.Queue()
+        self.handlerthread = threading.Thread(target=self.process_request_thread)
+        self.handlerthread.daemon = False
+
+        self.handlerthread.start()
+        super().serve_forever()
+
+    def process_request_thread(self):
+        while not self.quit:
+            try:
+                (request, client_address) = self.requestqueue.get(True)
+            except queue.Empty:
+                continue
+            if request is None:
+                continue
+            try:
+                self.finish_request(request, client_address)
+            except Exception:
+                self.handle_error(request, client_address)
+            finally:
+                self.shutdown_request(request)
+
+    def process_request(self, request, client_address):
+        self.requestqueue.put((request, client_address))
+
+    def server_close(self):
+        super().server_close()
+        self.quit = True
+        self.requestqueue.put((None, None))
+        self.handlerthread.join()
+
 def create_server(addr, dbname, prefix=''):
     class Handler(HashEquivalenceServer):
         pass
@@ -171,4 +209,5 @@ def create_server(addr, dbname, prefix=''):
         cursor.execute('CREATE INDEX IF NOT EXISTS outhash_lookup ON tasks_v2 (method, outhash)')
 
     logger.info('Starting server on %s', addr)
-    return HTTPServer(addr, Handler)
+
+    return ThreadedHTTPServer(addr, Handler)
-- 
2.20.1