aboutsummaryrefslogtreecommitdiffstats
path: root/lib/hashserv/client.py
diff options
context:
space:
mode:
authorJoshua Watt <jpewhacker@gmail.com>2019-09-17 08:37:11 -0500
committerRichard Purdie <richard.purdie@linuxfoundation.org>2019-09-18 17:48:25 +0100
commit2124eec3a5830afe8e07ffb6f2a0df6a417ac973 (patch)
treeec54ecf492198a28628a7c251b6a3ea524e430c4 /lib/hashserv/client.py
parent22082c7b3ca0cffcedb7d1d8c6681d35286376db (diff)
downloadbitbake-2124eec3a5830afe8e07ffb6f2a0df6a417ac973.tar.gz
bitbake: Rework hash equivalence
Reworks the hash equivalence server to address performance issues that were encountered with the REST mechanism used previously, particularly during the heavy request load encountered during signature generation. Notable changes are: 1) The server protocol is no longer HTTP based. Instead, it uses a simpler JSON over a streaming protocol link. This protocol has much lower overhead than HTTP since it eliminates the HTTP headers. 2) The hash equivalence server can either bind to a TCP port, or a Unix domain socket. Unix domain sockets are more efficient for local communication, and so are preferred if the user enables hash equivalence only for the local build. The arguments to the 'bitbake-hashserve' command have been updated accordingly. 3) The value to which BB_HASHSERVE should be set to enable a local hash equivalence server is changed to "auto" instead of "localhost:0". The latter didn't make sense when the local server was using a Unix domain socket. 4) Clients are expected to keep a persistent connection to the server instead of creating a new connection each time a request is made for optimal performance. 5) Most of the client logic has been moved to the hashserve module in bitbake. This makes it easier to share the client code. 6) A new bitbake command has been added called 'bitbake-hashclient'. This command can be used to query a hash equivalence server, including fetching the statistics and running a performance stress test. 7) The table indexes in the SQLite database have been updated to optimize hash lookups. This change is backward compatible, as the database will delete the old indexes first if they exist. 8) The server has been reworked to use python async to maximize performance with persistently connected clients. This requires Python 3.5 or later. Signed-off-by: Joshua Watt <JPEWhacker@gmail.com> Signed-off-by: Richard Purdie <richard.purdie@linuxfoundation.org>
Diffstat (limited to 'lib/hashserv/client.py')
-rw-r--r--lib/hashserv/client.py156
1 files changed, 156 insertions, 0 deletions
diff --git a/lib/hashserv/client.py b/lib/hashserv/client.py
new file mode 100644
index 000000000..2559bbb3f
--- /dev/null
+++ b/lib/hashserv/client.py
@@ -0,0 +1,156 @@
+# Copyright (C) 2019 Garmin Ltd.
+#
+# SPDX-License-Identifier: GPL-2.0-only
+#
+
+from contextlib import closing
+import json
+import logging
+import socket
+
+
+logger = logging.getLogger('hashserv.client')
+
+
+class HashConnectionError(Exception):
+ pass
+
+
+class Client(object):
+ MODE_NORMAL = 0
+ MODE_GET_STREAM = 1
+
+ def __init__(self):
+ self._socket = None
+ self.reader = None
+ self.writer = None
+ self.mode = self.MODE_NORMAL
+
+ def connect_tcp(self, address, port):
+ def connect_sock():
+ s = socket.create_connection((address, port))
+
+ s.setsockopt(socket.SOL_TCP, socket.TCP_NODELAY, 1)
+ s.setsockopt(socket.SOL_TCP, socket.TCP_QUICKACK, 1)
+ s.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
+ return s
+
+ self._connect_sock = connect_sock
+
+ def connect_unix(self, path):
+ def connect_sock():
+ s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+ # AF_UNIX has path length issues so chdir here to workaround
+ cwd = os.getcwd()
+ try:
+ os.chdir(os.path.dirname(path))
+ s.connect(os.path.basename(path))
+ finally:
+ os.chdir(cwd)
+ return s
+
+ self._connect_sock = connect_sock
+
+ def connect(self):
+ if self._socket is None:
+ self._socket = self._connect_sock()
+
+ self.reader = self._socket.makefile('r', encoding='utf-8')
+ self.writer = self._socket.makefile('w', encoding='utf-8')
+
+ self.writer.write('OEHASHEQUIV 1.0\n\n')
+ self.writer.flush()
+
+ # Restore mode if the socket is being re-created
+ cur_mode = self.mode
+ self.mode = self.MODE_NORMAL
+ self._set_mode(cur_mode)
+
+ return self._socket
+
+ def close(self):
+ if self._socket is not None:
+ self._socket.close()
+ self._socket = None
+ self.reader = None
+ self.writer = None
+
+ def _send_wrapper(self, proc):
+ count = 0
+ while True:
+ try:
+ self.connect()
+ return proc()
+ except (OSError, HashConnectionError, json.JSONDecodeError, UnicodeDecodeError) as e:
+ logger.warning('Error talking to server: %s' % e)
+ if count >= 3:
+ if not isinstance(e, HashConnectionError):
+ raise HashConnectionError(str(e))
+ raise e
+ self.close()
+ count += 1
+
+ def send_message(self, msg):
+ def proc():
+ self.writer.write('%s\n' % json.dumps(msg))
+ self.writer.flush()
+
+ l = self.reader.readline()
+ if not l:
+ raise HashConnectionError('Connection closed')
+
+ if not l.endswith('\n'):
+ raise HashConnectionError('Bad message %r' % message)
+
+ return json.loads(l)
+
+ return self._send_wrapper(proc)
+
+ def send_stream(self, msg):
+ def proc():
+ self.writer.write("%s\n" % msg)
+ self.writer.flush()
+ l = self.reader.readline()
+ if not l:
+ raise HashConnectionError('Connection closed')
+ return l.rstrip()
+
+ return self._send_wrapper(proc)
+
+ def _set_mode(self, new_mode):
+ if new_mode == self.MODE_NORMAL and self.mode == self.MODE_GET_STREAM:
+ r = self.send_stream('END')
+ if r != 'ok':
+ raise HashConnectionError('Bad response from server %r' % r)
+ elif new_mode == self.MODE_GET_STREAM and self.mode == self.MODE_NORMAL:
+ r = self.send_message({'get-stream': None})
+ if r != 'ok':
+ raise HashConnectionError('Bad response from server %r' % r)
+ elif new_mode != self.mode:
+ raise Exception('Undefined mode transition %r -> %r' % (self.mode, new_mode))
+
+ self.mode = new_mode
+
+ def get_unihash(self, method, taskhash):
+ self._set_mode(self.MODE_GET_STREAM)
+ r = self.send_stream('%s %s' % (method, taskhash))
+ if not r:
+ return None
+ return r
+
+ def report_unihash(self, taskhash, method, outhash, unihash, extra={}):
+ self._set_mode(self.MODE_NORMAL)
+ m = extra.copy()
+ m['taskhash'] = taskhash
+ m['method'] = method
+ m['outhash'] = outhash
+ m['unihash'] = unihash
+ return self.send_message({'report': m})
+
+ def get_stats(self):
+ self._set_mode(self.MODE_NORMAL)
+ return self.send_message({'get-stats': None})
+
+ def reset_stats(self):
+ self._set_mode(self.MODE_NORMAL)
+ return self.send_message({'reset-stats': None})