Made indexer # threads controllable. (#366)

# Changes
- Added `num_threads` option to control fast indexer parallelism
- Don't use a thread pool for indexing if `num_threads == 1`

# Fixes
- Fixed `MixedLogReader` handling of Python `pathlib.Path` objects

Author: adamshapiro0

python/fusion_engine_client/analysis/data_loader.py

@@ -161,7 +161,7 @@ class DataLoader(object):
 
     logger = logging.getLogger('point_one.fusion_engine.analysis.data_loader')
 
-    def __init__(self, path=None, save_index=True, ignore_index=False):
+    def __init__(self, path=None, save_index=True, ignore_index=False, num_threads: int = None):
         """!
         @brief Create a new reader instance.
 
@@ -170,6 +170,8 @@ def __init__(self, path=None, save_index=True, ignore_index=False):
                future. See @ref FileIndex for details.
         @param ignore_index If `True`, ignore the existing index file and read from the `.p1log` binary file directly.
                If `save_index == True`, this will delete the existing file and create a new one.
+        @param num_threads The number of parallel threads to spawn during indexing. If `None`, defaults to the number
+               of available CPUs.
         """
         self.reader: MixedLogReader = None
 
@@ -185,9 +187,9 @@ def __init__(self, path=None, save_index=True, ignore_index=False):
 
         self._generate_index = save_index
         if path is not None:
-            self.open(path, save_index=save_index, ignore_index=ignore_index)
+            self.open(path, save_index=save_index, ignore_index=ignore_index, num_threads=num_threads)
 
-    def open(self, path, save_index=True, ignore_index=False):
+    def open(self, path, save_index=True, ignore_index=False, num_threads: int = None):
         """!
         @brief Open a FusionEngine binary file.
 
@@ -196,11 +198,13 @@ def open(self, path, save_index=True, ignore_index=False):
                future. See @ref FileIndex for details.
         @param ignore_index If `True`, ignore the existing index file and read from the `.p1log` binary file directly.
                If `save_index == True`, this will delete the existing file and create a new one.
+        @param num_threads The number of parallel threads to spawn during indexing. If `None`, defaults to the number
+               of available CPUs.
         """
         self.close()
 
         self.reader = MixedLogReader(input_file=path, save_index=save_index, ignore_index=ignore_index,
-                                     return_bytes=True, return_message_index=True)
+                                     return_bytes=True, return_message_index=True, num_threads=num_threads)
 
         # Read the first message (with P1 time) in the file to set self.t0.
         #

python/fusion_engine_client/parsers/fast_indexer.py

@@ -149,6 +149,9 @@ def fast_generate_index(
 
     @return The loaded or generated @ref FileIndex.
     """
+    if num_threads is None:
+        num_threads = cpu_count()
+
     file_size = os.stat(input_path).st_size
     _logger.debug(f'File size: {int(file_size/1024/1024)}MB')
 
@@ -191,9 +194,12 @@ def fast_generate_index(
     _logger.debug(f'Reads/thread: {blocks_per_thread}')
 
     # Create a threadpool.
-    with Pool(num_threads) as p:
-        # Kick off the threads to process with their args. Then concatenate their returned data.
-        index_raw = np.concatenate([o for o in p.starmap(_search_blocks_for_fe, args)])
+    if num_threads > 1:
+        with Pool(num_threads) as p:
+            # Kick off the threads to process with their args. Then concatenate their returned data.
+            index_raw = np.concatenate([o for o in p.starmap(_search_blocks_for_fe, args)])
+    else:
+        index_raw = _search_blocks_for_fe(*args[0])
 
     # Some messages may encapsulate other complete FE messages. Normally, these
     # are ignored. However, if a message straddles one of the processing blocks,

python/fusion_engine_client/parsers/mixed_log_reader.py

@@ -2,6 +2,7 @@
 
 import copy
 from datetime import datetime
+import io
 import os
 import sys
 
@@ -22,7 +23,8 @@ class MixedLogReader(object):
     logger = logging.getLogger('point_one.fusion_engine.parsers.mixed_log_reader')
 
     def __init__(self, input_file, warn_on_gaps: bool = False, show_progress: bool = False,
-                 save_index: bool = True, ignore_index: bool = False, max_bytes: int = None,
+                 save_index: bool = True, ignore_index: bool = False, num_threads: int = None,
+                 max_bytes: int = None,
                  time_range: TimeRange = None, message_types: Union[Iterable[MessageType], MessageType] = None,
                  source_ids: Optional[Iterable[int]] = None, return_header: bool = True,
                  return_payload: bool = True, return_bytes: bool = False, return_offset: bool = False,
@@ -41,6 +43,8 @@ def __init__(self, input_file, warn_on_gaps: bool = False, show_progress: bool =
                See @ref FileIndex for details. Ignored if `max_bytes` is specified.
         @param ignore_index If `True`, ignore the existing index file and read from the binary file directly. If
                `save_index == True`, this will delete the existing file and create a new one.
+        @param num_threads The number of parallel threads to spawn during indexing. If `None`, defaults to the number
+               of available CPUs.
         @param max_bytes If specified, read up to the maximum number of bytes.
         @param time_range An optional @ref TimeRange object specifying desired start and end time bounds of the data to
                be read. See @ref TimeRange for more details.
@@ -103,10 +107,10 @@ def __init__(self, input_file, warn_on_gaps: bool = False, show_progress: bool =
         self.start_time = datetime.now()
 
         # Open the file to be read.
-        if isinstance(input_file, str):
-            self.input_file = open(input_file, 'rb')
-        else:
+        if isinstance(input_file, io.IOBase):
             self.input_file = input_file
+        else:
+            self.input_file = open(input_file, 'rb')
 
         input_path = self.input_file.name
         self.file_size_bytes = os.stat(input_path).st_size
@@ -118,7 +122,8 @@ def __init__(self, input_file, warn_on_gaps: bool = False, show_progress: bool =
 
         # Open the companion index file if one exists, otherwise index the file.
         self._original_index = fast_indexer.fast_generate_index(input_path, force_reindex=ignore_index,
-                                                                save_index=save_index, max_bytes=max_bytes)
+                                                                save_index=save_index, max_bytes=max_bytes,
+                                                                num_threads=num_threads)
         self.next_index_elem = 0
         self.index = self._original_index
         self.filtered_message_types = False