NetworkInspector.Sessions 0.20.0

NetworkInspector.Sessions

The engine behind a packet-analysis application: a Session reads one or more capture sources on background threads, parses every frame once, and lets any number of listeners (a packet list, a statistics view, a filtered view, an exporter) pull exactly the packets they need, while the capture is still being read.

Highlights

• Built for UIs and services. Every source and every listener runs on its own thread. Listeners are woken with coalesced notifications and pull by packet id; nothing is pushed or copied per listener.
• Small memory footprint. The session keeps only a packet-id → frame map, never parsed packets. A read re-parses the frame and can recycle a caller-owned packet. The map is packed per 64 packets: about 0.25 B per packet for one source and 0.5 B for two interleaved sources (8 B before). With a memory-mapped PCAPNG source a session over 500 000 packets retains 1.9 B per packet including the source's index, down from 32.3 B.
• Per-listener filters. Each listener can have its own filter expression; filtered reads prune candidates with the presence index before any packet is evaluated.
• Value caches. Record selected field values into RAM columns during the first parse, or add a cache later and let it fill in the background.
• Hot stack swap as a generation change. TryRestart replaces the protocol stack (new settings, new dissectors) without stopping running sources and re-parses every frame. The swap raises Generation (readable without a lock, paired with the stack in the immutable StackView), jobs declare whether they depend on the stack (JobStackBinding) and end with a JobEndReason, the caller chooses to cancel, wait for, or refuse on running jobs (RestartOptions), the result is a RestartReport, and Restarted notifies hosts on a thread per handler. A restart that fails after the swap leaves the session Faulted (a visible state with a way out), not half-swapped.
• Predictable errors. Operational calls are Try* methods with a SessionFailure (code plus message); shutdown is idempotent, waits for a running restart, and collects disposal errors.

Install

dotnet add package NetworkInspector.Sessions
dotnet add package NetworkInspector.Protocols

NetworkInspector.Core, NetworkInspector.Sources, and NetworkInspector.Filter come in as dependencies. NetworkInspector.Protocols provides the standard dissectors.

Quick Start

Read a capture in a session and print every packet from a listener thread:

using NetworkInspector.Core;
using NetworkInspector.Core.Settings;
using NetworkInspector.Protocols;
using NetworkInspector.Sessions;
using NetworkInspector.Sessions.Listeners;
using NetworkInspector.Sources.Pcapng;

StackBuilder builder = new(new SettingsManager(), new FrameInterfaceRegistry());
builder.RegisterStandardProtocols();
using Stack stack = builder.Build(); // the caller owns the initial stack; declare it before the session

using Session session = new(stack);
PrintingListener listener = new();

// The session owns the source after a successful add and starts it on its own thread.
if (!session.TryAddFrameSource(PcapngSource.Open("capture.pcapng"), out _, out SessionFailure? failure)
    || !session.TryAddListener(listener, out _, out failure)
    || !session.TryStart(out failure))
{
    throw new InvalidOperationException(failure?.Message); // SessionFailure is a record struct
}

session.WaitForCompletion(); // all source threads are done
listener.Done.Wait();        // the listener has seen every packet
session.Shutdown();
Console.WriteLine($"{listener.Count} packets");

sealed class PrintingListener : ISessionListener
{
    private readonly Packet?[] _Buffer = new Packet?[256]; // reused by every read

    public ManualResetEventSlim Done { get; } = new();

    public int Count { get; private set; }

    public string UiName => "printer";

    // Runs on this listener's own thread with the new packet-id window.
    public void OnNewPackets(ISessionReader session, int fromIndex, int toIndexExclusive)
    {
        for (int start = fromIndex; start < toIndexExclusive; start += _Buffer.Length)
        {
            int length = Math.Min(_Buffer.Length, toIndexExclusive - start);
            int read = session.ReadPackets(start, _Buffer.AsSpan(0, length));
            for (int i = 0; i < read; i++)
            {
                if (_Buffer[i] is Packet packet)
                {
                    Console.WriteLine($"#{packet.Id.Value} {packet.Info}");
                }
            }
        }

        Count += toIndexExclusive - fromIndex;
    }

    // Raised after the final OnNewPackets.
    public void OnAllSourcesCompleted(ISessionReader session) => Done.Set();

    // Packet ids restart at 0 after a restart (TryRestart).
    public void OnStackChanged(ISessionReader session) => Count = 0;
}

WaitForCompletion waits for the source threads only. Listener callbacks can still be running when it returns; OnAllSourcesCompleted is the reliable "everything delivered" signal. Never call Shutdown, Dispose, or the listener's own unsubscribe from inside a callback.

How to embed a session in an application (setup order, threads, recipes for packet lists, filtered lists, value caches, field observers, restart, shutdown): SESSION_GUIDE.md.

Lifecycle

graph TD
    Idle["Idle<br/>sources and listeners registered"]
    Running["Running<br/>source jobs active"]
    Restarting["Restarting<br/>stack swap + reparse"]
    Faulted["Faulted<br/>re-parse failed after the swap"]
    Stopped["Stopped<br/>all sources finished"]
    ShuttingDown["ShuttingDown<br/>teardown"]
    Idle -->|"TryStart, sources"| Running
    Idle -->|"TryStart, no sources"| Stopped
    Running -->|"last source completes"| Stopped
    Running -->|"TryRestart"| Restarting
    Stopped -->|"TryRestart"| Restarting
    Restarting -->|"reparse done, sources active"| Running
    Restarting -->|"reparse done, no sources"| Stopped
    Restarting -->|"frame cannot be re-read"| Faulted
    Faulted -->|"TryRestart"| Restarting
    Faulted -->|"Shutdown"| ShuttingDown
    Idle -->|"Shutdown"| ShuttingDown
    Running -->|"Shutdown"| ShuttingDown
    ShuttingDown -->|"cleanup complete"| Stopped

Shutdown on a Stopped session keeps the phase Stopped; listeners receive OnShuttingDown without a phase change.

Typical flow:

1. Create a Session with a Stack.
2. TryAddFrameSource (Idle only) and TryAddListener (Idle, Running, or Stopped; refused while a restart runs or after shutdown).
3. TryStart(out failure): starts the source and listener threads; listeners added later start right away.
4. WaitForCompletion(): blocks until all source jobs finish (optional timeout).
5. Shutdown() or Dispose(): cancels listeners and jobs, disposes jobs and sources.
6. Dispose the initial stack after the session. Stacks created by a TryRestart factory are owned by the session.

Key Types

Unless noted, types are in NetworkInspector.Sessions. Stream sources are wrapped in CachedFrameSource so their frames stay readable after NextFrame; random-access sources are wrapped only with FrameSourceAddOptions.CacheRandomAccess. The wrap holds the inner Frame and does not copy payload bytes.

Pull-Based Listeners

Producers set atomic notification flags per listener and wake its thread. The listener clears the flags, then pulls data from ISessionReader. Several events between two wake-ups coalesce into one callback. Only UiName and OnNewPackets must be implemented; the other callbacks have empty default implementations.

Callbacks of one wake-up run in the order of this table.

Per-Listener Filters

A listener can register a filter, which then applies to that listener's reads only:

// Expression: a bad expression registers nothing and explains why in filterFailure.
session.TryAddListener(listener, "tcp.port == 443", out ListenerInfo? info, out FilterError? filterFailure,
    out SessionFailure? failure);

// A NetworkInspector.Filter.Filter you compiled yourself:
session.TryAddListener(otherListener, myFilter, out info, out failure);

An empty or whitespace-only expression compiles to the always-match filter. A filter compiled against another stack (another session, or this session before a restart) is re-bound to the session stack with TryDerive; the session installs that copy and leaves your instance untouched. A filter that cannot be re-bound throws ArgumentException and registers nothing. FilterError is in NetworkInspector.Filter.Errors.

Reading Packets

A packet is read by looking up its frame (packet id → frame id and source, about 0.25–0.5 B per packet in a packed map), fetching the frame with FrameById, and parsing it again. Very large captures therefore stay cheap to browse: pair the session with memory-mapped file sources (PreloadBudget = null), whose frame index needs well under 2 B per frame (see the Sources README).

ISessionReader offers three read shapes. All fill a caller-owned buffer; reads allocate only for empty slots, so reuse the buffer to recycle packets:

// 1. Packets only; ids are implied by the start index.
int read = reader.ReadPackets(fromIndex, packetBuffer);           // Span<Packet?>

// 2. Packets with their ids; always contiguous.
PacketRef[] refs = new PacketRef[256];
int readRefs = reader.ReadPackets(startId, refs, out PacketIdLayout layout);

// 3. Listener-bound, optionally filtered (listenerId = ListenerInfo.Id, NetworkInspector.Sessions.Ids).
bool ok = reader.TryReadPackets(listenerId, startId, refs, PacketReadMode.Matching,
    out int count, out PacketIdLayout matchLayout, out FilterError? filterError, out SessionFailure? readFailure);

ReadPackets and TryRestart read frames 64 packets at a time: the session maps the packets of a chunk, and for every source reads each run of consecutive frame ids with one range call (IRandomAccessFrameSource.FramesById), so the source reads its frames forward. Single-packet reads (TryGetPacket, TryGetFrame) keep FrameById. Measured against the build before range reads with the NetworkInspector.Profiling A/B harness: ReadPackets +5 % (the re-parse dominates a range read), TryGetFrame random +14 %, in order +10 %, four threads +197 %.

Matching scans from startId to the current PacketCount and keeps only what the filter accepts, so the layout becomes Gapped as soon as an id is skipped. A listener without a filter, or with the always-match filter, takes the unfiltered fast path. Otherwise the filter's presence-index candidate set prunes the range first.

TryReadPackets returns false with count == 0 and filterError set when the filter cannot produce a verdict (poisoned, a packet failed to evaluate, or not re-bound after a stack swap); All reads keep working. An unknown listener id returns SessionErrorCode.ListenerNotFound; disabled queries (restart, shutdown) return SessionErrorCode.QueriesDisabled.

The Matching scan can run past the window of the current OnNewPackets call. Keep a watermark (last returned id + 1) and start the next read at max(watermark, fromIndex); see SESSION_GUIDE.md section 7.3. Filtering never affects notifications: OnNewPackets always reports the unfiltered id window.

Export A Range

SessionExport.TryStart exports a packet range as a background job by the pull principle: the job thread pulls the packets through ISessionReader.TryVisitPackets, parses only what the exporter and the filter need (an export of a few fields builds no field tree), and commits the matching packets to the exporter. Ingest keeps running; a restart ends the job and the output holds what was committed.

if (SessionExport.TryStart(
        session,
        "export udp",
        view => CsvExporter.CreateBuilder()
            .ToFile(path)
            .WithContent(SelectedFields(view.Stack, "ip.src", "udp.length"))
            .Build(),
        0,
        session.PacketCount,
        "udp.length >= 56",
        out SessionExportJob? export,
        out SessionFailure? failure,
        out FilterError? filterError))
{
    export.Job.Join();                                  // or poll export.VisitedCount for a progress display
    (export.Exporter as IDisposable)?.Dispose();
}

TryVisitPackets(from, to, in options, view, visitor, …) is the same pull without a job: your visitor receives each packet of the range, parsed with your own ParseOptions, on your thread, under one stack lease per call. See SESSION_GUIDE.md section 8.6.

Value Caches

A session can fill a RAM ValueCache during the first parse (SessionOptions.ValueCache) and add caches at runtime with TryAddValueCache. Value-cache listeners receive OnNewRows on their own thread with the same coalesced window as OnNewPackets, then read columns from ReadOnlyValueCache. Runtime caches never record on the parse thread; they fill by skip-parsing frames in the background.

using NetworkInspector.Core.ValueCaches;
using NetworkInspector.Sessions;
using NetworkInspector.Sessions.ValueCaches;

sealed class UdpPortCacheListener : IValueCacheListener
{
    private int _RowWatermark;

    public string UiName => "udp src ports";

    public void OnNewRows(ISessionReader session, ReadOnlyValueCache cache, int fromIndex, int toIndexExclusive)
    {
        if (!cache.TryGetSeries<ulong>("udp.srcport", out ReadOnlyValueCacheSeries<ulong> series))
        {
            return;
        }

        ValueCacheSeriesHandle<ulong> handle = series.Handle;
        foreach (ValueCacheRow<ulong> row in handle.EnumerateFrom(_RowWatermark))
        {
            Console.WriteLine($"packet {row.PacketId}: {row.Value}");
        }

        _RowWatermark = handle.Count;
    }
}

// Runtime cache, filled in the background:
session.TryAddValueCache(new UdpPortCacheListener(), new ValueCacheRequest { FieldNames = ["udp.srcport"] },
    out ValueCacheInfo? cacheInfo, out SessionFailure? cacheFailure);

// Ingest cache, filled during the first parse:
using Session ingestSession = new(stack, new SessionOptions
{
    ValueCache = new ValueCacheRequest { FieldNames = ["udp.srcport"] },
});

session.IngestValueCache is the read-only view of the ingest cache. Field names are validated when the request is bound (construction, TryAddValueCache, TryRestart). Usage of snapshots, watermarks, find, and SIMD scans: VALUECACHE_GUIDE.md.

Restart (Stack Swap)

TryRestart(stackFactory) replaces the protocol stack without stopping running sources. Every expected refusal is a result (false plus a SessionFailure), never an exception:

RestartOptions options = new() { JobPolicy = RestartJobPolicy.WaitThenCancel, JobWaitTimeout = TimeSpan.FromSeconds(10) };
if (!session.TryRestart(
        registry =>
        {
            // Use the registry passed in and a new SettingsManager: the session owns the new stack.
            StackBuilder newBuilder = new(new SettingsManager(), registry);
            newBuilder.RegisterStandardProtocols();
            return newBuilder.Build();
        },
        options,
        out RestartReport? report,
        out SessionFailure? failure))
{
    Console.WriteLine(failure.Value.Message);       // JobsRunning, InvalidPhase, RestartInProgress, the fault, ...
}

• Generation. The swap raises session.Generation (0 for the constructor stack) before reads are enabled again. session.CurrentStack is the stack with its generation in one immutable StackView; a view taken before a restart keeps naming the old stack. A consumer that holds field ids keeps the view next to them and re-resolves its field names when view.IsCurrent(session) is false. Polling Generation and QueriesEnabled shows stable, in progress and done without any callback. One read call parses on one generation.
• Jobs. TryAddJob declares a JobStackBinding (StackBound is the default; Independent opts out). A restart cancels stack-bound jobs (CancelNow, JobEndReason.StackReplaced), waits up to JobWaitTimeout (1 s … 1 h, default 30 s) and cancels the rest (WaitThenCancel), or is refused while they run (Refuse, JobsRunning). New stack-bound jobs are refused with RestartPending while a restart is pending; a restart from a stack-bound job's own thread is refused with RestartFromJob.
• Order. Claim, job check, factory, job phase, swap, re-parse, listener rebind, queries enabled, listeners woken, Restarted posted. A failing factory costs no job.
• Source threads wait at a parse gate while existing frames are re-parsed in id order (0 … N−1). TryRestart blocks until every frame is re-parsed (and, under WaitThenCancel, the wait); run it off a UI thread.
• Listeners receive OnStackChanged followed by OnNewPackets from 0.
• Every listener filter is re-bound with TryDerive (fresh state, empty match cache). A filter that cannot be re-bound is dropped, that listener's Matching reads report the bind error instead of returning everything, and the listener is named in RestartReport.FiltersNotRebound.
• The new ingest value cache and packet index are built before anything changes. A field the new stack lacks refuses the restart with ValueCacheUnknownField and leaves the session as it was.
• A frame that cannot be re-read makes TryRestart return false with FrameUnavailable and leaves the session Faulted: session.Fault keeps the failure, reads and new work are refused with SessionFaulted, ingest stays parked. Another TryRestart recovers (ids stay dense, ingest resumes); Shutdown completes.
• Restarted delivers a RestartNotification to every handler on its own thread (an Independent job named restart-event), never on the restarting thread; a blocking or throwing handler cannot influence another.
• A recycled Packet of an earlier stack of the same registry is rebound to the new stack, so a buffer of packets survives a restart without allocation. TryGetPacket(id, in options, view, recycle, …) parses with your observers under the session's lease and refuses options built for another generation (OptionsStackMismatch).
• The replaced stack is disposed as soon as it can safely be: right after the swap when no read is parsing on it, otherwise by the last read that ends (a lease counter, no lock; TryRestart does not wait for those reads). A stack the session created is never kept until Shutdown. A Packet handed out before the restart keeps its materialised fields; its still unmaterialised lazy containers may report a failed population (error child) after the old stack was disposed, so call MaterializeAll first or re-read after OnStackChanged.
• The factory must not call Shutdown, TryRestart, Dispose, or TryStart.

TryUnsubscribe vs Shutdown

TryUnsubscribe(job) ends one job:

• Source job: stops reading; the source stays available for random access until Shutdown.
• Listener job: cancels the slot, calls OnUnsubscribed, and removes it.
• User job: cancels its CancellationToken.

Shutdown() cancels all sources, listeners, and jobs (user jobs from TryAddJob included), waits for them, and disposes everything. With a timeout, the sources and the user jobs share one time budget; a registration that races with Shutdown is either completed before it (and cancelled with the rest) or refused with InvalidPhase.

FrameSourceInfo.Stop() and ListenerInfo.Unsubscribe() delegate to TryUnsubscribe. It returns false with a SessionFailure for foreign jobs, terminal jobs, or during shutdown; a source that has not started can be removed in Idle. TryRemoveJob removes a terminal job (Completed, Cancelled, Failed) and returns false with JobStillRunning for a pending or running one.

Error Handling

• Operational Try* refusals return false and a SessionFailure.
• Null arguments throw ArgumentNullException. Other caller bugs and hard limits (empty UI names, a disposed session, exhausted ids) throw SessionException with a SessionErrorCode.
• QueriesEnabled is false while a restart rewrites packets, while the session is Faulted, and after shutdown. A failed read means "no row" only when it is true.
• Shutdown() throws AggregateException when disposing one or more items fails. Dispose() stores those errors in Session.ShutdownErrors instead of throwing.

Thread Safety

All public Session methods are thread-safe. First parses are serialized under one lock across source threads and get dense ids; re-parsing an announced id takes no lock. Counters use Interlocked, phase and flags use Volatile. Listeners and value caches may be added in Idle, Running, or Stopped; user jobs until Shutdown starts; frame sources only in Idle. Shutdown waits for a running restart, and later Shutdown callers wait for the first one. The registries (sources, listeners, value caches, user jobs) are copy-on-write snapshots that the per-frame paths read without a lock; their writers, TryStart, and the shutdown flag serialise on one short internal lock, so concurrent adds lose no entry, exactly one TryStart caller wins, and a source is either started or refused. GetJobs(), GetListeners(), GetFrameSources(), and GetValueCaches() return read-only views of such a snapshot (not the snapshot array); a view never changes, a later change gives a new one.

A user job (TryAddJob) must observe its token and must not call Shutdown, TryRestart, or WaitForCompletion once shutdown has begun: Shutdown waits for the job while it holds the lifecycle lock those calls need.

Each listener filter is single-threaded: Matching reads and restart re-binding share that listener's filter gate, All reads do not take it.

Stack lifetime: every read that parses outside the parse lock holds a lease on the stack it parses with (one interlocked increment and decrement, no lock, no allocation). Shutdown waits for outstanding leases within its time budget and then disposes the stack anyway. Errors reported while a replaced stack was disposed are thrown by Shutdown (once) with the other cleanup errors.

Learn More

• Session guide: embedding a session step by step
• Filter · Sources · Core · Root overview
• GitHub repository · Issue tracker

License

MIT License