Skip to content

Async Engine

Jump to bottom
Eugene Palchukovsky edited this page Jul 23, 2026 · 2 revisions

Async Engine (Go and C++)

The Go SDK's asyncengine subpackage and the C++ SDK's openpit::asyncengine::TypedAsyncEngine turn an AccountSync engine into a concurrent facade with per-account ordering guarantees. These helpers are one possible integration pattern - applications may provide their own sharded queues, actor model, or equivalent dispatcher.

Both SDKs provide sharded and dynamic strategies, futures, observer callbacks, graceful stop, and hard stop. The Go examples below use goroutines and context.Context; the C++ surface uses worker threads, Future<T>, and std::chrono deadlines with the same account-pinning contract.

When to Use

Use a bundled async engine when the application:

  • builds the engine with AccountSync to allow concurrent access across many accounts;
  • wants the SDK to guarantee that no two operations for the same account ever run inside the engine concurrently;
  • prefers Future-style return values over manual channel/wait-group bookkeeping;
  • needs graceful and hard stop modes with deadlines.

If FullSync or NoSync is sufficient, or if you already have a custom per-account dispatcher you are happy with, you do not need this package.

Choosing Between Strategies

asyncengine ships two dispatch strategies. Each chosen at build time so the hot path stays branch-free in steady state.

Sharded

  • Go: asyncengine.NewBuilder(engine).Sharded(workers)

  • C++: openpit::asyncengine::MakeTypedAsyncEngine(engine, workers) or TypedBuilder<Driver>(driver).Sharded(workers)

  • A fixed number of worker queues created up front. Go runs one goroutine per shard; C++ runs one worker thread per shard.

  • The account ID is hashed (Fibonacci mix) into one of the shards.

  • Routing is one multiply-shift per submit. There is no per-account map lookup; synchronization is limited to ordering the submit against queue capacity and stop.

Strengths:

  • The cheapest hot path: ~constant per submit.
  • O(1) memory regardless of how many distinct accounts are active.
  • Predictable worker count.

Trade-offs:

  • A hot account saturates a single shard while the others stay idle.
  • No per-account observer signals (queue created/removed never fire).
  • Different accounts that hash to the same shard interleave through the same channel.

Pick Sharded when the active account set is broad and roughly balanced or when raw throughput per submit matters more than per-account isolation.

Dynamic

  • Go: asyncengine.NewBuilder(engine).Dynamic().MaxQueues(n).IdleCleanupAfter(d)

  • C++: TypedBuilder<Driver>(driver).Dynamic().MaxQueues(n).IdleCleanupAfter(d)

  • Per-account queue created on first submit.

  • A background cleanup worker retires queues that have been empty and untouched for IdleCleanupAfter.

  • MaxQueues caps the live queue count; submits for unknown accounts return ErrQueueLimit in Go or ErrorCode::QueueLimit in C++ past the cap. MaxQueues(0) removes the cap.

Strengths:

  • Full per-account isolation: a slow account does not starve others that happen to share a shard.
  • Per-account observer signals (OnQueueCreated, OnQueueRemoved, per-account latency).
  • Memory scales with the active set rather than the total population.

Trade-offs:

  • A synchronized map lookup on every submit.
  • A periodic cleanup worker.
  • Slightly higher per-account memory because each queue holds its own bounded buffer and worker.

Pick Dynamic when account activity is skewed, when you want per-account metrics, or when the population is large enough that statically allocating shards would be wasteful.

MaxQueues defaults to runtime.NumCPU() * 32, a value designed to be effectively non-restrictive on typical hosts while still bounding pathological growth (e.g. ephemeral per-request accounts). Override with MaxQueues(0) for unbounded growth.

Result Delivery: Future

Every queued operation returns a future (pkg/future in Go, openpit::asyncengine::Future<T> in C++), resolved exactly once by a worker or synchronously when submission fails before queueing.

Go operations that mirror one return value use future.Future[T]; tuple-shaped operations use future.Future2[A, B]. C++ typed operations return Future<StartOutcome>, Future<ExecuteOutcome>, or the concrete result type. AsyncRequest::Execute uses PairFuture for its reservation-or-rejects pair.

Go future operations:

  • f.Await(ctx) - block until resolved or ctx fires. Future[T] yields (T, error); Future2[A, B] yields (A, B, error).
  • f.Done() - non-blocking check.
  • f.TryGet() - non-blocking read.
  • f.Wait() - channel that closes on resolution (for select).

The future and the wrapped engine objects are decoupled: cancelling the context passed to Await does not stop the underlying engine call; it only stops the caller from waiting.

In C++, Await() returns the value or throws the carried async error on the waiting thread; Await(timeout) returns std::nullopt when the timeout expires. The underlying task continues in both cases.

Threading

AsyncEngine requires the wrapped engine to be built with AccountSync. Per-account correctness holds in both strategies: no two operations for the same account ever run inside the engine concurrently, and within one account every queued operation runs in submit order.

Parallelism across different accounts depends on the strategy. Dynamic gives full per-account isolation, so distinct accounts are always processed in parallel. Sharded gives per-shard serialization: distinct accounts that hash to the same shard share one worker and are processed one after another (a hot account can block others on its shard). Neither relaxes the per-account invariant above.

The Threading Contract of the engine itself is respected end-to-end.

AsyncRequest and AsyncReservation keep the same per-account queue across the boundary: calling Execute on a started request, or Commit on a reservation, re-enters the same per-account chain.

The C++ typed facade is non-copyable but move-constructible. Its dispatch state keeps a stable address, so already-issued request and reservation wrappers stay valid after a move. Move assignment is disabled because replacing a live target could invalidate wrappers created by that target. Keep the resulting engine alive until all wrappers are released.

Stop Modes

  • Go StopGraceful(ctx) refuses new submissions and waits for every already-queued task to run to completion. Returns ctx.Err() if ctx fires before workers drain; the engine is then partially stopped and StopHard may be invoked to complete the shutdown.
  • Go StopHard(ctx) refuses new submissions, aborts every task that has not yet started with ErrStopped, and waits for the currently-running task in each worker to finish. Returns ctx.Err() if ctx fires before the in-flight task in each worker finishes.
  • C++ StopGraceful(timeout) and StopHard(timeout) have the same queue behavior and return true when workers drained before the deadline. A hard stop resolves queued tasks with ErrorCode::Stopped.

When the builder was wired via AccountSyncReadyEngineBuilder.BuildAsync, the underlying *Engine is released when the stop completes successfully (returns nil). If StopGraceful or StopHard returns ctx.Err() (timeout), the engine is not yet released; the caller must complete the shutdown (call StopHard) before the release happens. When the caller used asyncengine.NewBuilder(engine) and did not wire WithStopUnderlying, the engine handle remains owned by the caller.

Observer

asyncengine.Observer in Go and openpit::asyncengine::Observer in C++ are optional diagnostic interfaces. The default is NoopObserver. Wire only the methods you need - the package itself has zero external observability dependencies (no OpenTelemetry, Prometheus, or logging in the import graph), so you decide where the signals go.

Available callbacks: OnEnqueue, OnDequeue, OnComplete, OnSlowSubmit, OnQueueFullBlocked, OnQueueCreated, OnQueueRemoved, OnSubmitCancelled.

Observer callbacks are diagnostic only. In C++, exceptions thrown by an observer are ignored so observability cannot change queue or task semantics.

Example

package main

import (
 "context"
 "log"
 "time"

 "go.openpit.dev/openpit"
 "go.openpit.dev/openpit/model"
 "go.openpit.dev/openpit/param"
 "go.openpit.dev/openpit/pretrade/policies"
)

func main() {
 // Build an AccountSync engine and wrap it into an async facade in one
 // chain. BuildAsync is only available on the AccountSync builder; for
 // FullSync or NoSync engines the bundled async helper is not used.
 asyncBuilder, err := openpit.NewEngineBuilder().
  AccountSync().
  Builtin(policies.BuildOrderValidation()).
  Builtin(
   policies.BuildRateLimit().BrokerBarrier(
    policies.RateLimitBrokerBarrier{
     Limit: policies.RateLimit{
      MaxOrders: 100,
      Window:    time.Second,
     },
    },
   ),
  ).
  BuildAsync()
 if err != nil {
  log.Fatal(err)
 }

 // Pick a dispatch strategy. Use Sharded for cheap routing across a
 // balanced account population; use Dynamic for per-account isolation
 // and per-account metrics.
 async, err := asyncBuilder.Dynamic().
  MaxQueues(0).
  IdleCleanupAfter(5 * time.Minute).
  Build()
 if err != nil {
  log.Fatal(err)
 }
 defer func() {
  if err := async.StopGraceful(context.Background()); err != nil {
   log.Printf("StopGraceful: %v", err)
  }
 }()

 // Submit a start-stage call. The future resolves once the worker has
 // executed the call. AsyncRequest.Execute and Close are queued in the
 // same per-account chain so AccountSync is never violated.
 usd, err := param.NewAsset("USD")
 if err != nil {
  log.Fatal(err)
 }
 aapl, err := param.NewAsset("AAPL")
 if err != nil {
  log.Fatal(err)
 }
 order := model.NewOrder()
 op := order.EnsureOperationView()
 op.SetInstrument(param.NewInstrument(aapl, usd))
 op.SetAccountID(param.NewAccountIDFromUint64(99224416))
 op.SetSide(param.SideBuy)
 price, err := param.NewPriceFromString("185")
 if err != nil {
  log.Fatal(err)
 }
 qty, err := param.NewQuantityFromString("100")
 if err != nil {
  log.Fatal(err)
 }
 op.SetTradeAmount(param.NewQuantityTradeAmount(qty))
 op.SetPrice(price)

 request, rejects, err := async.StartPreTrade(
  context.Background(),
  order,
 ).Await(context.Background())
 if err != nil {
  log.Fatal(err)
 }
 if request == nil {
  // Rejected at the start stage; inspect rejects.
  _ = rejects
  return
 }

 // The async request preserves AccountSync across the Start - Execute
 // boundary by routing Execute through the same per-account queue. The
 // future yields the same (reservation, rejects, error) tuple the
 // synchronous main stage returns.
 reservation, rejects, err := request.Execute(
  context.Background(),
 ).Await(context.Background())
 if err != nil {
  log.Fatal(err)
 }
 if reservation == nil {
  _ = rejects
  return
 }

 if _, err := reservation.CommitAndClose(
  context.Background(),
 ).Await(context.Background()); err != nil {
  log.Fatal(err)
 }
}

Submitting Caller-Owned Work

Go Submit(ctx, accountID, fn) and C++ Submit(accountId, std::function<void()>, timeout) enqueue caller-owned work into the same per-account queue. Use this to run client-side work atomically with respect to engine calls on the same account - for example, to persist an order before Execute, or update a strategy book after Commit.

Splitting a logical transaction into two Submit calls "surfaces" between them: tasks for the same account from elsewhere in the system can interleave between the two halves. To keep that boundary closed, bundle the work into a single Submit whose fn does both halves.

Lifecycle Notes

  • Aborted tasks resolve their future with ErrStopped in Go or ErrorCode::Stopped in C++. For reservation methods that "and close" (Close, CommitAndClose, RollbackAndClose) the underlying reservation is still released as a safety net even on abort.
  • An aborted AsyncRequest.Execute releases the underlying request before resolving the future. The caller does not need a separate cleanup path.
  • The Go ctx or C++ timeout passed to a submit method controls how long the producer waits for queue space. Once queued, the worker runs or aborts the task.

Related Pages

Clone this wiki locally