Speed up compute_simhash and correct executor placements

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.

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)
 

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
 

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]

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:

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}")

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,