Class ModulePerfSession

java.lang.Object
ai.pipestream.module.pipelineprobe.moduletest.ModulePerfSession
@ApplicationScoped public class ModulePerfSession extends Object
Live, interactive load generator for one module's process() interface, driven through the dev process-once door. You Start it, then Ramp-up / Ramp-down concurrency through the geometric levels 1→2→4→8→16→32 while polling live metrics; Stop ends it.

Measures the module's raw ModuleProcessor.process(T) compute, not transport or fixture I/O: payload acquisition happens outside the timed window, so each timed call is just the door round-trip + the processor. The buffer knob trades memory for loader involvement — buffer ≥ corpus caches every wrapped doc once (true raw baseline, ~1GB for the heaviest corpus); smaller buffers stream via a bounded queue and the loaderStarved stat tells you if the loader ever gated throughput.

Per ramp level: an ConcurrentHistogram (lock-free concurrent record, non-destructive copy() for polling) gives min/max/mean/p50/p95/p99; level docs/sec is completions ÷ elapsed-in-level. Every call is also recorded to a Micrometer Timer so the same data lands in Grafana/LGTM.

One session at a time — start(String, int) stops any prior run.

  • Constructor Summary

    Constructors
    Constructor
    Description
    ModulePerfSession()
    Creates the module perf session.

  • Method Summary

    Modifier and Type
    Method
    Description
    Map<String,Object>
    pause()
    Pause the load without ending the session: workers drain to zero but the scaffold's provisioned state (plan, index) stays live — so downstream (OSM's indexing consumer) can drain its backlog against a valid plan and be measured cleanly.
    Map<String,Object>
    rampDown()
    Step concurrency down one geometric level (a no-op when stopped; stays paused when already paused).
    Map<String,Object>
    rampUp()
    Step concurrency up one geometric level (a no-op when stopped; resumes at the first level when paused).
    Map<String,Object>
    snapshot()
    Non-destructive live snapshot of the current level + lifetime totals.
    Map<String,Object>
    start(String module, int buffer)
    Default-options overload — see start(String, int, String).
    Map<String,Object>
    start(String module, int buffer, String strategy)
    Start a fresh session for module, replacing any prior run.
    Map<String,Object>
    stop()
    Stop the session: signal workers to finish, interrupt the streaming loader, shut down the worker pool, and run the scaffold cleanup (swallowing any cleanup failure with a warning).

    Methods inherited from class Object

    clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

  • Constructor Details

    • ModulePerfSession

      public ModulePerfSession()
      Creates the module perf session. Collaborators are injected by the container; instances are managed by CDI.

  • Method Details

    • start

      public Map<String,Object> start(String module, int buffer)
      Default-options overload — see start(String, int, String).
      Parameters:
      module - the module key to load and drive
      buffer - payload supply size selecting cached vs streaming mode
      Returns:
      the opening session snapshot

    • start

      public Map<String,Object> start(String module, int buffer, String strategy)
      Start a fresh session for module, replacing any prior run. buffer picks the payload supply mode: >= corpus size caches every wrapped doc once (full in-memory baseline); smaller values stream through a bounded queue of that capacity while a background loader cycles the whole corpus. Begins the load at the first ramp level.
      Parameters:
      module - the module key to load and drive
      buffer - payload supply size; >= corpus size caches every wrapped doc in memory, smaller values stream via a bounded queue of this capacity
      strategy - opensearch-sink only: indexing strategy (nested / separate_indices / chunk_combined; null = chunk_combined)
      Returns:
      the opening session snapshot
      Throws:
      IllegalArgumentException - if no corpus dataset feeds the module
      IllegalStateException - if no docs could be loaded for the module

    • pause

      public Map<String,Object> pause()
      Pause the load without ending the session: workers drain to zero but the scaffold's provisioned state (plan, index) stays live — so downstream (OSM's indexing consumer) can drain its backlog against a valid plan and be measured cleanly. Resume with rampUp().
      Returns:
      the session snapshot after pausing (concurrency dropped to zero)

    • rampUp

      public Map<String,Object> rampUp()
      Step concurrency up one geometric level (a no-op when stopped; resumes at the first level when paused).
      Returns:
      the session snapshot at the new ramp level

    • rampDown

      public Map<String,Object> rampDown()
      Step concurrency down one geometric level (a no-op when stopped; stays paused when already paused).
      Returns:
      the session snapshot at the new ramp level

    • snapshot

      public Map<String,Object> snapshot()
      Non-destructive live snapshot of the current level + lifetime totals.
      Returns:
      a map of session state, current-level latency percentiles and throughput, and lifetime disposition counts

    • stop

      public Map<String,Object> stop()
      Stop the session: signal workers to finish, interrupt the streaming loader, shut down the worker pool, and run the scaffold cleanup (swallowing any cleanup failure with a warning). Safe to call when no session is running.
      Returns:
      a final summary map with the module, total docs processed, and the lifetime disposition breakdown