From 07a02b31fd80748ab11d7e30fbf2a3d2e59b1426 Mon Sep 17 00:00:00 2001 From: Joshua Watt Date: Thu, 25 Jun 2020 09:21:07 -0500 Subject: bitbake: hashserv: Chunkify large messages The hash equivalence client and server can occasionally send messages that are too large for the server to fit in the receive buffer (64 KB). To prevent this, support is added to the protocol to "chunkify" the stream and break it up into manageable pieces that the server can each side can back together. Ideally, this would be negotiated by the client and server, but it's currently hard coded to 32 KB to prevent the round-trip delay. (Bitbake rev: e27a28c1e40e886ee68ba4b99b537ffc9c3577d4) Signed-off-by: Joshua Watt Signed-off-by: Richard Purdie --- bitbake/lib/hashserv/__init__.py | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) (limited to 'bitbake/lib/hashserv/__init__.py') diff --git a/bitbake/lib/hashserv/__init__.py b/bitbake/lib/hashserv/__init__.py index c3318620f5..f95e8f43f1 100644 --- a/bitbake/lib/hashserv/__init__.py +++ b/bitbake/lib/hashserv/__init__.py @@ -6,12 +6,20 @@ from contextlib import closing import re import sqlite3 +import itertools +import json UNIX_PREFIX = "unix://" ADDR_TYPE_UNIX = 0 ADDR_TYPE_TCP = 1 +# The Python async server defaults to a 64K receive buffer, so we hardcode our +# maximum chunk size. It would be better if the client and server reported to +# each other what the maximum chunk sizes were, but that will slow down the +# connection setup with a round trip delay so I'd rather not do that unless it +# is necessary +DEFAULT_MAX_CHUNK = 32 * 1024 def setup_database(database, sync=True): db = sqlite3.connect(database) @@ -66,6 +74,20 @@ def parse_address(addr): return (ADDR_TYPE_TCP, (host, int(port))) +def chunkify(msg, max_chunk): + if len(msg) < max_chunk - 1: + yield ''.join((msg, "\n")) + else: + yield ''.join((json.dumps({ + 'chunk-stream': None + }), "\n")) + + args = [iter(msg)] * (max_chunk - 1) + for m in map(''.join, itertools.zip_longest(*args, fillvalue='')): + yield ''.join(itertools.chain(m, "\n")) + yield "\n" + + def create_server(addr, dbname, *, sync=True): from . import server db = setup_database(dbname, sync=sync) -- cgit v1.2.3-54-g00ecf