arbiter-core-0.1.0.0
arbiter-core
Safe HaskellNone
LanguageGHC2024

Arbiter.Concurrency

Description

Per-job concurrency limits.

Declare which pool (if any) caps each job with a HasConcurrency instance, building the selector from noConcurrency / concurrencyBy / globalConcurrency / concurrencyByCase. The migration collects every pool a selector can reach and seeds it. The pool holds the limit.

Synopsis

Declaring a payload's pool

Source #class HasConcurrency payload where

A payload's per-job pool selection. Defaults to unbounded. Only capped payloads need an instance.

Minimal complete definition

Nothing

Instances

Instances details
HasConcurrency payload Source # 
Instance details

Defined in Arbiter.Core.Concurrency.Spec

Source #type ConcurrencyFor payload = Selector ConcurrencyPolicy payload (Maybe ConcurrencyKey)

A selective description of the concurrency key for a payload. Evaluation returns the job key. Static inspection returns the reachable pools.

Source #noConcurrency :: ConcurrencyFor payload

This payload is unbounded.

Source #concurrencyBy :: ConcurrencyPolicy -> (payload -> Text) -> ConcurrencyFor payload

Cap by a fixed pool, keyed by a per-job suffix (e.g. a tenant id).

Source #globalConcurrency :: ConcurrencyPolicy -> Text -> ConcurrencyFor payload

Cap by a fixed pool under one shared key (a single global pool).

Source #concurrencyByCase :: (Bounded k, Enum k, Eq k) => (payload -> k) -> (k -> ConcurrencyFor payload) -> ConcurrencyFor payload

Concurrency selectByCase. See selectByCase for the totality requirement on k.

Source #chooseWhen :: (payload -> Bool) -> Selector policy payload a -> Selector policy payload a -> Selector policy payload a

Select between two selectors with a job predicate. Policy collection inspects both branches.

Pools

Source #concurrencyPool :: Text -> Int32 -> ConcurrencyPolicy

A pool named prefix admitting at most limit concurrent jobs per key. The cap is floored at 1. Pause a pool via an override. The prefix must not contain :, the key separator. The migration enforces this.

Management and observability views

Source #data ConcurrencyPolicyView

A pool with its default and override limits plus live key and in-flight stats. The effective cap is override when set, else default.

Instances

Instances details
FromJSON ConcurrencyPolicyView Source # 
Instance details

Defined in Arbiter.Core.Concurrency.Stats

ToJSON ConcurrencyPolicyView Source # 
Instance details

Defined in Arbiter.Core.Concurrency.Stats

Eq ConcurrencyPolicyView Source # 
Instance details

Defined in Arbiter.Core.Concurrency.Stats

Generic ConcurrencyPolicyView Source # 
Instance details

Defined in Arbiter.Core.Concurrency.Stats

Associated Types

type Rep ConcurrencyPolicyView 
Instance details

Defined in Arbiter.Core.Concurrency.Stats

type Rep ConcurrencyPolicyView = D1 ('MetaData "ConcurrencyPolicyView" "Arbiter.Core.Concurrency.Stats" "arbiter-core-0.1.0.0-inplace" 'False) (C1 ('MetaCons "ConcurrencyPolicyView" 'PrefixI 'True) ((S1 ('MetaSel ('Just "prefix") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Text) :*: (S1 ('MetaSel ('Just "defaultLimit") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Int32) :*: S1 ('MetaSel ('Just "overrideLimit") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 (Maybe Int32)))) :*: (S1 ('MetaSel ('Just "keyCount") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Int64) :*: (S1 ('MetaSel ('Just "totalInFlight") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Int64) :*: S1 ('MetaSel ('Just "maxInFlight") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 (Maybe Int32))))))
Show ConcurrencyPolicyView Source # 
Instance details

Defined in Arbiter.Core.Concurrency.Stats

type Rep ConcurrencyPolicyView Source # 
Instance details

Defined in Arbiter.Core.Concurrency.Stats

type Rep ConcurrencyPolicyView = D1 ('MetaData "ConcurrencyPolicyView" "Arbiter.Core.Concurrency.Stats" "arbiter-core-0.1.0.0-inplace" 'False) (C1 ('MetaCons "ConcurrencyPolicyView" 'PrefixI 'True) ((S1 ('MetaSel ('Just "prefix") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Text) :*: (S1 ('MetaSel ('Just "defaultLimit") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Int32) :*: S1 ('MetaSel ('Just "overrideLimit") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 (Maybe Int32)))) :*: (S1 ('MetaSel ('Just "keyCount") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Int64) :*: (S1 ('MetaSel ('Just "totalInFlight") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Int64) :*: S1 ('MetaSel ('Just "maxInFlight") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 (Maybe Int32))))))

Source #data ConcurrencyKeyView

A single key's in-flight count, its effective cap, and fill fraction.

Instances

Instances details
FromJSON ConcurrencyKeyView Source # 
Instance details

Defined in Arbiter.Core.Concurrency.Stats

ToJSON ConcurrencyKeyView Source # 
Instance details

Defined in Arbiter.Core.Concurrency.Stats

Eq ConcurrencyKeyView Source # 
Instance details

Defined in Arbiter.Core.Concurrency.Stats

Generic ConcurrencyKeyView Source # 
Instance details

Defined in Arbiter.Core.Concurrency.Stats

Associated Types

type Rep ConcurrencyKeyView 
Instance details

Defined in Arbiter.Core.Concurrency.Stats

type Rep ConcurrencyKeyView = D1 ('MetaData "ConcurrencyKeyView" "Arbiter.Core.Concurrency.Stats" "arbiter-core-0.1.0.0-inplace" 'False) (C1 ('MetaCons "ConcurrencyKeyView" 'PrefixI 'True) ((S1 ('MetaSel ('Just "concurrencyKey") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Text) :*: S1 ('MetaSel ('Just "concurrencyPrefix") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Text)) :*: (S1 ('MetaSel ('Just "inFlight") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Int32) :*: (S1 ('MetaSel ('Just "effectiveLimit") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Int32) :*: S1 ('MetaSel ('Just "fillFraction") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 (Maybe Double))))))
Show ConcurrencyKeyView Source # 
Instance details

Defined in Arbiter.Core.Concurrency.Stats

type Rep ConcurrencyKeyView Source # 
Instance details

Defined in Arbiter.Core.Concurrency.Stats

type Rep ConcurrencyKeyView = D1 ('MetaData "ConcurrencyKeyView" "Arbiter.Core.Concurrency.Stats" "arbiter-core-0.1.0.0-inplace" 'False) (C1 ('MetaCons "ConcurrencyKeyView" 'PrefixI 'True) ((S1 ('MetaSel ('Just "concurrencyKey") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Text) :*: S1 ('MetaSel ('Just "concurrencyPrefix") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Text)) :*: (S1 ('MetaSel ('Just "inFlight") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Int32) :*: (S1 ('MetaSel ('Just "effectiveLimit") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Int32) :*: S1 ('MetaSel ('Just "fillFraction") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 (Maybe Double))))))

Source #data ConcurrencyPolicyUpdate

A patch over a pool's override limit. Nothing leaves it unchanged, Just Nothing clears the override (reverts to the default), Just (Just v) sets it.

Instances

Instances details
FromJSON ConcurrencyPolicyUpdate Source #

Hand-written. A missing key leaves the field unchanged. An explicit null clears the override.

Instance details

Defined in Arbiter.Core.Concurrency.Stats

ToJSON ConcurrencyPolicyUpdate Source # 
Instance details

Defined in Arbiter.Core.Concurrency.Stats

Eq ConcurrencyPolicyUpdate Source # 
Instance details

Defined in Arbiter.Core.Concurrency.Stats

Generic ConcurrencyPolicyUpdate Source # 
Instance details

Defined in Arbiter.Core.Concurrency.Stats

Associated Types

type Rep ConcurrencyPolicyUpdate 
Instance details

Defined in Arbiter.Core.Concurrency.Stats

type Rep ConcurrencyPolicyUpdate = D1 ('MetaData "ConcurrencyPolicyUpdate" "Arbiter.Core.Concurrency.Stats" "arbiter-core-0.1.0.0-inplace" 'False) (C1 ('MetaCons "ConcurrencyPolicyUpdate" 'PrefixI 'True) (S1 ('MetaSel ('Just "overrideLimit") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 (Maybe (Maybe Int32)))))
Show ConcurrencyPolicyUpdate Source # 
Instance details

Defined in Arbiter.Core.Concurrency.Stats

type Rep ConcurrencyPolicyUpdate Source # 
Instance details

Defined in Arbiter.Core.Concurrency.Stats

type Rep ConcurrencyPolicyUpdate = D1 ('MetaData "ConcurrencyPolicyUpdate" "Arbiter.Core.Concurrency.Stats" "arbiter-core-0.1.0.0-inplace" 'False) (C1 ('MetaCons "ConcurrencyPolicyUpdate" 'PrefixI 'True) (S1 ('MetaSel ('Just "overrideLimit") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 (Maybe (Maybe Int32)))))

Operations

Source #updateConcurrencyPolicyOverrides :: MonadArbiter m => Text -> ConcurrencyPolicyUpdate -> m Int64

Apply a pool's override-limit patch on its policy row, retuning every key under the prefix live. Lowering it does not preempt in-flight jobs until they drain. Returns rows affected.

Source #setConcurrencyLimit :: MonadArbiter m => ConcurrencyPolicy -> Int32 -> m Int64

Override a declared pool's limit for every key under it. Returns rows affected.

Source #clearConcurrencyLimit :: MonadArbiter m => ConcurrencyPolicy -> m Int64

Drop a pool's override. Its declared limit applies again. Returns rows affected.

Source #pruneConcurrencyKeys :: (MonadArbiter m, RegistryTables (RegistryOf m)) => m Int64

Delete drained concurrency rows with no live job. The reaper runs this.

Source #reconcileConcurrencyCounts :: (MonadArbiter m, RegistryTables (RegistryOf m)) => m Int64

Recompute the concurrency counts from live jobs, repairing any trigger drift.

Source #listConcurrencyPolicies :: MonadArbiter m => m [ConcurrencyPolicyView]

List every concurrency pool with its default/override limit and live key and in-flight aggregates.

Source #listConcurrencyKeys :: MonadArbiter m => Text -> Int -> Int -> m [ConcurrencyKeyView]

List a prefix's keys with effective cap and in-flight fill fraction, paginated.