diff --git a/bbot/core/helpers/helper.py b/bbot/core/helpers/helper.py index f3d8e8064f..77f94d1e89 100644 --- a/bbot/core/helpers/helper.py +++ b/bbot/core/helpers/helper.py @@ -262,8 +262,13 @@ def run_in_executor_io(self, callback, *args, **kwargs): def run_in_executor_cpu(self, callback, *args, **kwargs): """ - Run short CPU-bound work that releases the GIL in a dedicated thread pool, - separate from I/O so it never queues behind long-running network calls. + Run short CPU-bound work in a dedicated thread pool, separate from I/O so it + never queues behind long-running network calls. + + This is a thread pool, so it only buys parallelism for work that releases the + GIL (yara). Pure-Python work still runs one call at a time and competes with + the event loop for the GIL, so keep it short. Anything long-running and + GIL-bound belongs in run_in_executor_mp() instead. Examples: Execute callback: @@ -275,7 +280,12 @@ def run_in_executor_cpu(self, callback, *args, **kwargs): async def run_in_executor_mp(self, callback, *args, **kwargs): """ Same as run_in_executor_io() except with a process pool executor. - Use only in cases where callback is CPU-bound. + + Use for CPU-bound work that holds the GIL long enough to stall the event loop, + or that needs crash/hang isolation from hostile input. Arguments and return + values are pickled, so prefer plain data over objects with big object graphs. + Workers are recycled periodically, so this is a poor fit for very high rates of + very short tasks. Includes a timeout (default 300s) to prevent indefinite hangs if a child process dies or the pool enters a broken state. On timeout, the entire pool is terminated and replaced so that stuck workers cannot accumulate and starve the scan. diff --git a/bbot/core/helpers/regex.py b/bbot/core/helpers/regex.py index 7a6ec0b9cc..ba7a4a83e1 100644 --- a/bbot/core/helpers/regex.py +++ b/bbot/core/helpers/regex.py @@ -97,9 +97,6 @@ def _finditer_multi(self, compiled_regexes, *args, **kwargs): def _finditer(self, compiled_regex, *args, **kwargs): return list(compiled_regex.finditer(*args, **kwargs)) - async def extract_params_html(self, *args, **kwargs): - return await self.parent_helper.run_in_executor_cpu(misc.extract_params_html, *args, **kwargs) - async def extract_emails(self, *args, **kwargs): return await self.parent_helper.run_in_executor_cpu(misc.extract_emails, *args, **kwargs) diff --git a/bbot/core/helpers/simhash.py b/bbot/core/helpers/simhash.py index e602c2923e..e4ceb24ab6 100644 --- a/bbot/core/helpers/simhash.py +++ b/bbot/core/helpers/simhash.py @@ -1,5 +1,7 @@ -import xxhash import re +import struct +import xxhash +from collections import Counter _non_word_re = re.compile(r"[^\w]+") @@ -13,10 +15,6 @@ def compute_simhash(text, bits=64, truncate=True, normalization_filter=None): """ Static method for computing a SimHash fingerprint. - Designed to be called via run_in_executor_cpu(): the work is short and the - input is truncated to ~3KB inside the helper, so a thread pool avoids the - pickle/spawn overhead of a process pool. - Args: text (str): The text to hash bits (int): Number of bits for the hash. Defaults to 64. @@ -87,19 +85,26 @@ def hash(self, text, truncate=True, normalization_filter=None): if normalization_filter: text = self._normalize_text(text, normalization_filter) - vector = [0] * self.bits features = self._get_features(text) - - for feature in features: - hv = self._hash_feature(feature) - for i in range(self.bits): - bit = (hv >> i) & 1 - vector[i] += 1 if bit else -1 - - # Final fingerprint + num_features = len(features) + + # Tally how many feature hashes have each bit set. Packing the hashes into a + # buffer lets Counter do the counting in C over a byte-strided slice; a plain + # per-bit Python loop costs 64 iterations per feature and dominates the function. + packed = struct.pack(f"<{num_features}Q", *[self._hash_feature(f) for f in features]) + + ones = [0] * 64 + for byte_pos in range(8): + counts = Counter(packed[byte_pos::8]) + for bit_pos in range(8): + mask = 1 << bit_pos + ones[byte_pos * 8 + bit_pos] = sum(count for value, count in counts.items() if value & mask) + + # A bit is set when it appears in at least half the features, i.e. when + # (ones - zeros) >= 0. Bits past 64 are never set, since the hash is 64 bits. fingerprint = 0 - for i, val in enumerate(vector): - if val >= 0: + for i in range(min(self.bits, 64)): + if 2 * ones[i] >= num_features: fingerprint |= 1 << i return fingerprint diff --git a/bbot/modules/dnsbrute_mutations.py b/bbot/modules/dnsbrute_mutations.py index b4784f6ea6..0ede75dae4 100644 --- a/bbot/modules/dnsbrute_mutations.py +++ b/bbot/modules/dnsbrute_mutations.py @@ -43,7 +43,10 @@ async def handle_event(self, event): async def get_parent_event(self, subdomain): start = time.time() - parent_host = await self.helpers.run_in_executor_cpu(self.helpers.closest_match, subdomain, self.parent_events) + # pass the hostnames only; the dict values are events, which are expensive to pickle + parent_host = await self.helpers.run_in_executor_mp( + self.helpers.closest_match, subdomain, list(self.parent_events) + ) elapsed = time.time() - start self.trace(f"{subdomain}: got closest match among {len(self.parent_events):,} parent events in {elapsed:.2f}s") return self.parent_events[parent_host] diff --git a/bbot/modules/virtualhost.py b/bbot/modules/virtualhost.py index 5b5a33fbb8..060c4e26cf 100644 --- a/bbot/modules/virtualhost.py +++ b/bbot/modules/virtualhost.py @@ -68,9 +68,6 @@ async def setup(self): async def _response_similarity(self, text_a, text_b, normalization_filter=None): """ Compute simhash similarity between two response bodies. - Runs in the CPU thread pool: simhash work is short and the input is - truncated to ~3KB inside compute_simhash, so the process-pool overhead - (pickle + IPC + worker lifecycle) costs more than the work itself. """ kwargs = {} if normalization_filter is not None: diff --git a/bbot/modules/waf_bypass.py b/bbot/modules/waf_bypass.py index fc2a6dc9b9..0dce939002 100644 --- a/bbot/modules/waf_bypass.py +++ b/bbot/modules/waf_bypass.py @@ -111,7 +111,7 @@ async def handle_event(self, event): return # Store a "simhash" (fuzzy hash) of the response data for later comparison - simhash = await self.helpers.run_in_executor_mp(compute_simhash, response.text) + simhash = await self.helpers.run_in_executor_cpu(compute_simhash, response.text) self.content_fingerprints[url] = { "simhash": simhash, @@ -157,7 +157,7 @@ async def check_ip(self, ip, source_domain, protected_domain, source_event): self.debug(f"Failed to get content through IP {ip} for URL {matching_url}") return None - bypass_simhash = await self.helpers.run_in_executor_mp(compute_simhash, bypass_response.text or "") + bypass_simhash = await self.helpers.run_in_executor_cpu(compute_simhash, bypass_response.text or "") if original_response["http_code"] != bypass_response.status_code: self.debug(f"Ignoring code difference {original_response['http_code']} != {bypass_response.status_code}") diff --git a/bbot/modules/wayback.py b/bbot/modules/wayback.py index e58f6d9a37..cd7863705a 100644 --- a/bbot/modules/wayback.py +++ b/bbot/modules/wayback.py @@ -396,7 +396,7 @@ async def _fetch_cdx(self, query): # parse JSON + extract URLs off the event loop # (CDX responses can contain 100k+ entries) try: - urls = await self.helpers.run_in_executor_cpu(_parse_cdx_response, r.text) + urls = await self.helpers.run_in_executor_mp(_parse_cdx_response, r.text) except Exception: urls = None if urls is None: @@ -477,7 +477,7 @@ async def query(self, query): collapsed_urls = 0 start_time = datetime.now() # consolidate URLs to cut down on garbage data (CPU-intensive, runs off the event loop) - parsed_urls = await self.helpers.run_in_executor_cpu( + parsed_urls = await self.helpers.run_in_executor_mp( self.helpers.validators.collapse_urls, urls, threshold=self.garbage_threshold,