From f1b0ab7e4b1cfeabaa8f981175e503d0c435ba0b Mon Sep 17 00:00:00 2001 From: Etienne Cordonnier Date: Thu, 21 Sep 2023 09:56:58 +0200 Subject: bitbake: bitbake-worker: add header with length of message The IPC mechanism between runqueue.py and bitbake-worker is currently not scalable: The data is sent with the format pickled-data, and bitbake-worker has no information about the size of the message. Therefore, the bitbake-worker is calling select() and read() in a loop, and then calling "self.queue.find(b"")" for each chunk received. This does not scale, because queue.find has a linear complexity relative to the size of the queue, and workerdata messages get very big e.g. for builds which reference a lot of files in SRC_URI. The number of chunks varies, but on my test system a lot of chunks of 65536 bytes are sent, and each iteration takes 0.1 seconds, making the transfer of the "workerdata" data very slow (on my test setup 35 seconds before this fix, and 1.5 seconds after this fix). This commit adds a 4 bytes header after , so that bitbake-worker knows how many bytes need to be received, and does not need to constantly search the whole queue for . (Bitbake rev: 595176d6be95a9c4718d3a40499d1eb576b535f5) Signed-off-by: Etienne Cordonnier Signed-off-by: Alexandre Belloni Signed-off-by: Richard Purdie --- bitbake/bin/bitbake-worker | 34 +++++++++++++++++++++++----------- 1 file changed, 23 insertions(+), 11 deletions(-) (limited to 'bitbake/bin') diff --git a/bitbake/bin/bitbake-worker b/bitbake/bin/bitbake-worker index 609e276fe2..eba9c562c7 100755 --- a/bitbake/bin/bitbake-worker +++ b/bitbake/bin/bitbake-worker @@ -433,18 +433,30 @@ class BitbakeWorker(object): while self.process_waitpid(): continue - def handle_item(self, item, func): - if self.queue.startswith(b"<" + item + b">"): - index = self.queue.find(b"") - while index != -1: - try: - func(self.queue[(len(item) + 2):index]) - except pickle.UnpicklingError: - workerlog_write("Unable to unpickle data: %s\n" % ":".join("{:02x}".format(c) for c in self.queue)) - raise - self.queue = self.queue[(index + len(item) + 3):] - index = self.queue.find(b"") + opening_tag = b"<" + item + b">" + if not self.queue.startswith(opening_tag): + return + + tag_len = len(opening_tag) + if len(self.queue) < tag_len + 4: + # we need to receive more data + return + header = self.queue[tag_len:tag_len + 4] + payload_len = int.from_bytes(header, 'big') + # closing tag has length (tag_len + 1) + if len(self.queue) < tag_len * 2 + 1 + payload_len: + # we need to receive more data + return + + index = self.queue.find(b"") + if index != -1: + try: + func(self.queue[(tag_len + 4):index]) + except pickle.UnpicklingError: + workerlog_write("Unable to unpickle data: %s\n" % ":".join("{:02x}".format(c) for c in self.queue)) + raise + self.queue = self.queue[(index + len(b"")):] def handle_cookercfg(self, data): self.cookercfg = pickle.loads(data) -- cgit v1.2.3-54-g00ecf