Tasks: inline bounded same-entity hook outputs

Ingestion hooks always sent generated tasks through the priority RabbitMQ path, even when they targeted the same entity and would return to the same worker thread. Mirrored TTL data therefore added a confirmed publish, broker delivery, parse, and acknowledgement cycle for every source task.

Add a startup-configured bound for processing same-routing-key descendants through the complete TaskExecutor pipeline in a local FIFO. The default of zero preserves queue-only behavior; cross-entity work, overflow, and pending siblings after an inline failure continue through the priority queue.

A saturated benchmark using the real data-mirroring TTL hook and a three-second master-buffer flush improved median throughput from 67 to 207 entities/s and reduced completion time from 29.70s to 9.68s. It also eliminated 2,000 priority publish/deliver/ack cycles and reduced median Mongo updates from 2,314 to 2,004; the database coalescing benefit remains workload-dependent.

Author: xsedla1o

config/processing_core.yml

@@ -22,6 +22,10 @@ worker_processes: 1
 # external services via network)
 worker_threads: 16
 
+# Maximum number of same-entity tasks generated by ingestion hooks to process inline.
+# Set to 0 to send all generated tasks through the priority queue. Changes require a worker restart.
+max_inline_generated_tasks: 0
+
 # Path to directory with plug-in modules
 # (Relative path is evaluated relative to location of this configuration file)
 modules_dir: "../modules"

docs/configuration/processing_core.md

@@ -11,6 +11,7 @@ msg_broker:
   password: dp3_password
 worker_processes: 2
 worker_threads: 16
+max_inline_generated_tasks: 0
 modules_dir: "../dp3_modules"
 enabled_modules:
   - "module_one"
@@ -50,6 +51,22 @@ to utilize computational power of multiple CPUs (which Python cannot do well
 anyway due to the GIL), but to mask long I/O operations (e.g. queries to
 external services via network).
 
+## Inline generated tasks
+
+`max_inline_generated_tasks` is a non-negative integer that limits how many tasks generated by
+ingestion hooks may be processed inline for one task received from RabbitMQ. The default is `0`,
+which disables inline processing and sends every generated task through the priority queue.
+
+Only generated tasks with the same complete entity key (entity type and EID) as the received task
+are eligible. The limit counts generated children, not the received source task, and is shared by
+the whole chain of children and descendants. Tasks for other entities and same-entity tasks above
+the limit are sent through the priority queue. An overflow task receives a fresh allowance if it is
+later consumed from RabbitMQ.
+
+The setting is read when workers start, so changing it requires a worker restart. A small value such
+as `20` is recommended when first enabling the optimization; restore `0` and restart workers to
+return to queue-only processing.
+
 ## Modules directory
 
 Path to directory with plug-in (secondary) modules.

docs/hooks.md

@@ -193,8 +193,8 @@ Callable[[AnyEidT, DataPointTask], list[DataPointTask]]
 This hook is called once for a newly created entity, after `allow_entity_creation` has accepted creation and before the task's datapoints are staged into raw and master persistence.
 The callback receives the new `eid` together with the original `DataPointTask` that caused the entity to be created, so it can inspect the incoming datapoints through `task.data_points`.
 It should not assume that the entity's new state is already readable as a persisted master record, because the creation-triggering task has not been written through the normal persistence path yet.
-The hook may return a list of `DataPointTask` objects, and those tasks are queued back into the ingestion pipeline, where they trigger the usual hooks again.
-Because they originate from ingestion, they are pushed to the **priority** task queue.
+The hook may return a list of `DataPointTask` objects, and those tasks re-enter the ingestion pipeline, where they trigger the usual hooks again.
+Depending on `max_inline_generated_tasks`, same-entity outputs may be processed inline by the current worker; outputs for other entities and same-entity overflow are pushed to the **priority** task queue.
 This registration also supports `refresh=` and `may_change=` for recomputation during module-config refresh; see [Refresh-on-config-change behavior for ingestion hooks](#refresh-on-config-change-behavior-for-ingestion-hooks).
 
 Real usage examples:
@@ -258,8 +258,9 @@ If `task.delete` is `True`, `on_task_start` still runs and entity deletion is pe
 
 #### Returned tasks recurse through the same pipeline
 
-Any `DataPointTask` returned by `on_entity_creation` or `on_new_attr` is sent back to the main task queue and later processed again by `TaskExecutor.process_task`.
-In other words, module-generated datapoints re-enter DP3 exactly like primary datapoints from the API, so they can trigger `on_task_start`, `allow_entity_creation`, `on_entity_creation`, `on_new_attr`, and later snapshot or updater hooks.
+Any `DataPointTask` returned by `on_entity_creation` or `on_new_attr` is processed again by `TaskExecutor.process_task`.
+Depending on worker configuration, a bounded number of same-entity tasks may run inline, while cross-entity tasks and same-entity overflow pass through the priority queue first.
+In either case, module-generated datapoints re-enter the complete ingestion pipeline, so they can trigger `on_task_start`, `allow_entity_creation`, `on_entity_creation`, `on_new_attr`, and later snapshot or updater hooks.
 
 ## Snapshot-time hooks: periodic processing over stored data
 
@@ -610,17 +611,28 @@ This creates a feedback loop:
 
 ```text
 hook returns DataPointTask(s)
-  -> task queue
+  -> inline processing or task queue
   -> TaskExecutor.process_task
   -> ingestion hooks run again
   -> data reaches master/raw storage
   -> later snapshot / updater cycles can see it
 ```
 
-The queueing path differs slightly by hook family.
-Ingestion hooks such as `on_entity_creation` and `on_new_attr` push returned tasks to the priority queue.
-Snapshot hooks and updater hooks push returned tasks to the normal task queue.
-`scheduler_register` callbacks do not have an automatic task-return path.
+The re-entry path differs slightly by hook family.
+For ingestion hooks such as `on_entity_creation` and `on_new_attr`, workers may process a bounded
+number of same-entity outputs inline when `max_inline_generated_tasks` is positive. The source task
+does not count toward the limit, and the allowance is shared by all descendants generated while
+processing it. Outputs for another entity and same-entity outputs above the limit use the priority
+queue. Inline processing still calls the complete `TaskExecutor.process_task` pipeline; it does not
+make buffered writes immediately visible through independent database reads.
+
+With the default `max_inline_generated_tasks: 0`, all ingestion-hook outputs use the priority queue,
+matching queue-only behavior. The value is read at worker startup. Enabling inline processing can
+change interleaving with tasks already assigned to the worker and removes the JSON serialization and
+RabbitMQ boundary for eligible outputs, so hooks must not depend on either boundary.
+
+Snapshot hooks and updater hooks push returned tasks to the normal task queue and are not eligible for
+this inline path. `scheduler_register` callbacks do not have an automatic task-return path.
 
 
 ## Type of `eid`

dp3/task_processing/task_distributor.py

@@ -6,6 +6,7 @@
 import sys
 import threading
 import time
+from collections import deque
 from functools import partial
 
 from dp3.common.config import PlatformConfig
@@ -66,6 +67,17 @@ def __init__(
         # List of worker threads for processing the update requests
         self._worker_threads = []
         self.num_threads = platform_config.config.get("processing_core.worker_threads", 8)
+        self.max_inline_generated_tasks = platform_config.config.get(
+            "processing_core.max_inline_generated_tasks", 0
+        )
+        if (
+            isinstance(self.max_inline_generated_tasks, bool)
+            or not isinstance(self.max_inline_generated_tasks, int)
+            or self.max_inline_generated_tasks < 0
+        ):
+            raise ValueError(
+                "processing_core.max_inline_generated_tasks must be a non-negative integer"
+            )
 
         # Internal queues for each worker
         self._queues = [queue.Queue(10) for _ in range(self.num_threads)]
@@ -274,7 +286,7 @@ def _worker_func(self, thread_index):
             # Process the task
             start_time = time.time()
             try:
-                created, new_tasks = self.task_executor.process_task(task)
+                created = self._process_task_chain(task)
             except Exception:
                 self.log.error(f"Error has occurred during processing task: {task}")
                 raise
@@ -292,7 +304,61 @@ def _worker_func(self, thread_index):
                     )
                 )
 
+    def _process_task_chain(self, source_task: DataPointTask) -> bool:
+        """Process one broker task and its bounded same-key generated-task chain."""
+        created, new_tasks = self.task_executor.process_task(source_task)
+        if self.max_inline_generated_tasks == 0:
             self.push_new_tasks(new_tasks)
+            return created
+
+        source_key = source_task.routing_key()
+        pending = deque()
+        admitted_count = 0
+        inline_count = 0
+        cross_key_count = 0
+        overflow_count = 0
+
+        def partition_tasks(tasks: list[DataPointTask]) -> None:
+            nonlocal admitted_count, cross_key_count, overflow_count
+
+            tasks_to_publish = []
+            for task in tasks:
+                if task.routing_key() != source_key:
+                    cross_key_count += 1
+                    tasks_to_publish.append(task)
+                elif admitted_count < self.max_inline_generated_tasks:
+                    admitted_count += 1
+                    pending.append(task)
+                else:
+                    overflow_count += 1
+                    tasks_to_publish.append(task)
+
+            if tasks_to_publish:
+                self.push_new_tasks(tasks_to_publish)
+
+        partition_tasks(new_tasks)
+        try:
+            while pending:
+                task = pending.popleft()
+                _, new_tasks = self.task_executor.process_task(task)
+                inline_count += 1
+                partition_tasks(new_tasks)
+        except Exception:
+            if pending:
+                self.push_new_tasks(list(pending))
+            raise
+
+        if inline_count:
+            self.log.debug(
+                "Generated-task chain for %s: processed %d inline, "
+                "published %d cross-key and %d overflow tasks",
+                source_key,
+                inline_count,
+                cross_key_count,
+                overflow_count,
+            )
+
+        return created
 
     def push_new_tasks(self, new_tasks):
         """Push new tasks (resulting from hooks) to the priority queue.

dp3/template/app/config/processing_core.yml

@@ -22,6 +22,10 @@ worker_processes: 1
 # external services via network)
 worker_threads: 16
 
+# Maximum number of same-entity tasks generated by ingestion hooks to process inline.
+# Set to 0 to send all generated tasks through the priority queue. Changes require a worker restart.
+max_inline_generated_tasks: 0
+
 # Path to directory with plug-in modules
 # (Relative path is evaluated relative to location of this configuration file)
 modules_dir: "../modules"