Table of Contents

Class AsyncWorkerData

Namespace
FishMMO.Server.Implementation
Assembly
FishMMO.Server.dll

Centralized async work queue.

Replaces fire-and-forget _ = SomeAsync(...) across all server systems with a bounded, backpressure-aware pool that runs work items concurrently while preserving FIFO order between items that share an entity key.

Design:

  • Work runs concurrently, capped by FishMMO.Server.Implementation.AsyncWorkerData.maxConcurrency. A slow item delays only the items ordered behind it.
  • Items sharing an entityKey run in the order they were enqueued, one at a time.
  • Bounded admission: Enqueue(Func<Task>, string) returns false once FishMMO.Server.Implementation.AsyncWorkerData.maxOutstandingItems items are accepted but unfinished.
  • Nothing ever executes on the calling thread — see FishMMO.Server.Implementation.AsyncWorkerData.DispatchUnordered(FishMMO.Server.Implementation.AsyncWorkerData.AsyncWorkItem).

Usage:

// Unordered — runs as soon as a concurrency slot is free:
asyncWorkerData.Enqueue(() => PersistInventoryAsync(dto));

// Ordered — this character's items run in enqueue order, one at a time:
asyncWorkerData.Enqueue(() => SaveCharacterAsync(charData), characterID);

Systems declare dependency via: [RequiresDataContainer(typeof(AsyncWorkerData))]

public class AsyncWorkerData : RuntimeDataContainer, IRuntimeDataContainer<INetworkManagerWrapper, ServerManager, NetworkConnection, IRuntimeDataContainer>, IServerComponent<INetworkManagerWrapper, ServerManager, NetworkConnection, IRuntimeDataContainer>, IAsyncWorkerData, IRuntimeDataContainer, IServerComponent
Inheritance
AsyncWorkerData
Implements
Inherited Members

Remarks

Why this is not a set of sequential worker loops. It used to be: N channels, one long-lived loop each, await item.Work() one item at a time. That made every worker a head-of-line queue — a single item that waited on something slow stalled every unrelated item routed to the same channel, for as long as it took. The waits are real and they are long: RunOnMainThreadAsync blocks for up to 30 seconds when the main-thread queue is not draining, ClaimCharacterSessionAsync backs off across five attempts, and any database call can stall. With eight loops it took eight such items to halt every save, session release, scene status write and routing decision on the process at once — while the thread pool sat idle, because none of those items were using a thread. They were awaiting.

Concurrency is what this pool should be bounding, not parallelism, so it bounds it directly with a semaphore and lets the thread pool schedule. Ordering is the one thing the loops genuinely provided, and it is preserved exactly where it was promised — per entity key — rather than as a side effect of which channel an item hashed to.

Pool sizing. FishMMO.Server.Implementation.AsyncWorkerData.maxConcurrency is deliberately well under the database connection pool (AppSettings.MaxPoolSize, 100 by default). A work item holds at most one connection at a time, so this caps connection demand from the pool with headroom left for the synchronous shutdown flush and the health checks, which do not come through here.

Properties

CompletedCount

Total number of work items processed since startup.

public long CompletedCount { get; }

Property Value

long

PendingCount

Current number of items accepted but not yet started. Useful for monitoring and diagnostics.

public int PendingCount { get; }

Property Value

int

Remarks

Accepted but not yet started. Items that are running are reported by neither this nor CompletedCount — they are in flight.

Methods

Clear()

Stops accepting new work. Everything already accepted is left to run.

public override void Clear()

Remarks

The channel-based implementation discarded the queue here, and this looked like a faithful port of that — but the contract was self-defeating. Clear has exactly one caller, RuntimeDataContainerRegistry.DeinitializeAll, which invokes it immediately before OnDeinitialize(). So "discard everything pending" only ever ran during shutdown, and what is pending during shutdown is precisely the work that must not be lost: the saves and session releases that the behaviours enqueued as they tore down moments earlier.

A dropped release is not a local loss. The character stays Online in the database until its lease expires, so after a restart the player is refused by every scene server for the next two minutes. Combat-logout bodies are the common case: FinalizeAllCombatLingers hands each one's save and release to this pool and then removes its token from SessionTokens, so the synchronous shutdown flush no longer covers them — this pool is the only thing that can release them.

Refusing new work still does the useful half: nothing further piles up while the process is going down, and OnDeinitialize()'s bounded wait flushes the backlog.

Enqueue(Func<Task>, long, string)

Enqueue an async work item with an entity key for ordered processing. Work items sharing the same entityKey are guaranteed to execute in FIFO order, one at a time; items with different keys proceed independently. Returns true if the item was accepted, false if the queue is full (backpressure).

public bool Enqueue(Func<Task> work, long entityKey, string callerName = null)

Parameters

work Func<Task>

The async work to execute.

entityKey long

Entity identifier for ordering (e.g., characterID), or 0 for none.

callerName string

Optional caller identifier for diagnostics.

Returns

bool

True if enqueued successfully.

Remarks

An entityKey of 0 means "no ordering requirement" and is treated exactly like the unkeyed overload. It is not an entity whose id happens to be zero, and callers that pass a default id are not asking to be serialized with each other.

Enqueue(Func<Task>, string)

Enqueue an async work item for processing. Returns true if the item was accepted, false if the queue is full (backpressure).

public bool Enqueue(Func<Task> work, string callerName = null)

Parameters

work Func<Task>

The async work to execute.

callerName string

Optional caller identifier for diagnostics.

Returns

bool

True if enqueued successfully.

InitializeOnce()

Prepares the concurrency gate and ordering lanes.

public override ServerComponentInitializationStatus InitializeOnce()

Returns

ServerComponentInitializationStatus

OnDeinitialize()

Stops accepting work and waits, bounded, for what is still running.

protected override void OnDeinitialize()