| Safe Haskell | None |
|---|---|
| Language | GHC2024 |
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
- class HasConcurrency payload where
- concurrencyFor :: ConcurrencyFor payload
- type ConcurrencyFor payload = Selector ConcurrencyPolicy payload (Maybe ConcurrencyKey)
- noConcurrency :: ConcurrencyFor payload
- concurrencyBy :: ConcurrencyPolicy -> (payload -> Text) -> ConcurrencyFor payload
- globalConcurrency :: ConcurrencyPolicy -> Text -> ConcurrencyFor payload
- concurrencyByCase :: (Bounded k, Enum k, Eq k) => (payload -> k) -> (k -> ConcurrencyFor payload) -> ConcurrencyFor payload
- chooseWhen :: (payload -> Bool) -> Selector policy payload a -> Selector policy payload a -> Selector policy payload a
- data ConcurrencyPolicy
- concurrencyPool :: Text -> Int32 -> ConcurrencyPolicy
- policyPrefixOf :: AdmissionPolicy p => p -> Text
- data ConcurrencyPolicyView = ConcurrencyPolicyView {
- prefix :: Text
- defaultLimit :: Int32
- overrideLimit :: Maybe Int32
- keyCount :: Int64
- totalInFlight :: Int64
- maxInFlight :: Maybe Int32
- data ConcurrencyKeyView = ConcurrencyKeyView {}
- data ConcurrencyPolicyUpdate = ConcurrencyPolicyUpdate {
- overrideLimit :: Maybe (Maybe Int32)
- updateConcurrencyPolicyOverrides :: MonadArbiter m => Text -> ConcurrencyPolicyUpdate -> m Int64
- setConcurrencyLimit :: MonadArbiter m => ConcurrencyPolicy -> Int32 -> m Int64
- clearConcurrencyLimit :: MonadArbiter m => ConcurrencyPolicy -> m Int64
- pruneConcurrencyKeys :: (MonadArbiter m, RegistryTables (RegistryOf m)) => m Int64
- reconcileConcurrencyCounts :: (MonadArbiter m, RegistryTables (RegistryOf m)) => m Int64
- listConcurrencyPolicies :: MonadArbiter m => m [ConcurrencyPolicyView]
- listConcurrencyKeys :: MonadArbiter m => Text -> Int -> Int -> m [ConcurrencyKeyView]
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
Methods
Source #concurrencyFor :: ConcurrencyFor payload
Instances
| HasConcurrency payload Source # | |
Defined in Arbiter.Core.Concurrency.Spec Methods Source #concurrencyFor :: ConcurrencyFor payload | |
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 #data ConcurrencyPolicy
A concurrency pool. At most cpLimit jobs share a key under cpPrefix. The
default is seeded. An operator override on the pool takes precedence.
Instances
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.
Source #policyPrefixOf :: AdmissionPolicy p => p -> Text
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.
Constructors
| ConcurrencyPolicyView | |
Fields
| |
Instances
Source #data ConcurrencyKeyView
A single key's in-flight count, its effective cap, and fill fraction.
Constructors
| ConcurrencyKeyView | |
Fields
| |
Instances
| FromJSON ConcurrencyKeyView Source # | |||||
Defined in Arbiter.Core.Concurrency.Stats Methods #parseJSON :: Value -> Parser ConcurrencyKeyView #parseJSONList :: Value -> Parser [ConcurrencyKeyView] | |||||
| ToJSON ConcurrencyKeyView Source # | |||||
Defined in Arbiter.Core.Concurrency.Stats Methods #toJSON :: ConcurrencyKeyView -> Value #toEncoding :: ConcurrencyKeyView -> Encoding #toJSONList :: [ConcurrencyKeyView] -> Value #toEncodingList :: [ConcurrencyKeyView] -> Encoding #omitField :: ConcurrencyKeyView -> Bool | |||||
| Eq ConcurrencyKeyView Source # | |||||
Defined in Arbiter.Core.Concurrency.Stats Methods #(==) :: ConcurrencyKeyView -> ConcurrencyKeyView -> Bool #(/=) :: ConcurrencyKeyView -> ConcurrencyKeyView -> Bool | |||||
| Generic ConcurrencyKeyView Source # | |||||
Defined in Arbiter.Core.Concurrency.Stats Associated Types
Methods #from :: ConcurrencyKeyView -> Rep ConcurrencyKeyView x #to :: Rep ConcurrencyKeyView x -> ConcurrencyKeyView | |||||
| Show ConcurrencyKeyView Source # | |||||
Defined in Arbiter.Core.Concurrency.Stats Methods #showsPrec :: Int -> ConcurrencyKeyView -> ShowS #show :: ConcurrencyKeyView -> String #showList :: [ConcurrencyKeyView] -> ShowS | |||||
| type Rep ConcurrencyKeyView Source # | |||||
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.
Constructors
| ConcurrencyPolicyUpdate | |
Fields
| |
Instances
| FromJSON ConcurrencyPolicyUpdate Source # | Hand-written. A missing key leaves the field unchanged. An explicit | ||||
Defined in Arbiter.Core.Concurrency.Stats | |||||
| ToJSON ConcurrencyPolicyUpdate Source # | |||||
Defined in Arbiter.Core.Concurrency.Stats Methods #toJSON :: ConcurrencyPolicyUpdate -> Value #toEncoding :: ConcurrencyPolicyUpdate -> Encoding #toJSONList :: [ConcurrencyPolicyUpdate] -> Value | |||||
| Eq ConcurrencyPolicyUpdate Source # | |||||
Defined in Arbiter.Core.Concurrency.Stats Methods #(==) :: ConcurrencyPolicyUpdate -> ConcurrencyPolicyUpdate -> Bool #(/=) :: ConcurrencyPolicyUpdate -> ConcurrencyPolicyUpdate -> Bool | |||||
| Generic ConcurrencyPolicyUpdate Source # | |||||
Defined in Arbiter.Core.Concurrency.Stats Associated Types
Methods #from :: ConcurrencyPolicyUpdate -> Rep ConcurrencyPolicyUpdate x #to :: Rep ConcurrencyPolicyUpdate x -> ConcurrencyPolicyUpdate | |||||
| Show ConcurrencyPolicyUpdate Source # | |||||
Defined in Arbiter.Core.Concurrency.Stats Methods #showsPrec :: Int -> ConcurrencyPolicyUpdate -> ShowS #show :: ConcurrencyPolicyUpdate -> String #showList :: [ConcurrencyPolicyUpdate] -> ShowS | |||||
| type Rep ConcurrencyPolicyUpdate Source # | |||||
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.