| 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
class HasConcurrency payload where Source #
A payload's per-job pool selection. Defaults to unbounded. Only capped payloads need an instance.
Minimal complete definition
Nothing
Methods
concurrencyFor :: ConcurrencyFor payload Source #
Instances
| HasConcurrency payload Source # | |
Defined in Arbiter.Core.Concurrency.Spec Methods concurrencyFor :: ConcurrencyFor payload Source # | |
type ConcurrencyFor payload = Selector ConcurrencyPolicy payload (Maybe ConcurrencyKey) Source #
A selective description of the concurrency key for a payload. Evaluation returns the job key. Static inspection returns the reachable pools.
noConcurrency :: ConcurrencyFor payload Source #
This payload is unbounded.
concurrencyBy :: ConcurrencyPolicy -> (payload -> Text) -> ConcurrencyFor payload Source #
Cap by a fixed pool, keyed by a per-job suffix (e.g. a tenant id).
globalConcurrency :: ConcurrencyPolicy -> Text -> ConcurrencyFor payload Source #
Cap by a fixed pool under one shared key (a single global pool).
concurrencyByCase :: (Bounded k, Enum k, Eq k) => (payload -> k) -> (k -> ConcurrencyFor payload) -> ConcurrencyFor payload Source #
Concurrency selectByCase. See selectByCase for the totality requirement on k.
chooseWhen :: (payload -> Bool) -> Selector policy payload a -> Selector policy payload a -> Selector policy payload a Source #
Select between two selectors with a job predicate. Policy collection inspects both branches.
Pools
data ConcurrencyPolicy Source #
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
concurrencyPool :: Text -> Int32 -> ConcurrencyPolicy Source #
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.
policyPrefixOf :: AdmissionPolicy p => p -> Text Source #
Management and observability views
data ConcurrencyPolicyView Source #
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
data ConcurrencyKeyView Source #
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 Source # parseJSONList :: Value -> Parser [ConcurrencyKeyView] Source # | |||||
| ToJSON ConcurrencyKeyView Source # | |||||
Defined in Arbiter.Core.Concurrency.Stats Methods toJSON :: ConcurrencyKeyView -> Value Source # toEncoding :: ConcurrencyKeyView -> Encoding Source # toJSONList :: [ConcurrencyKeyView] -> Value Source # toEncodingList :: [ConcurrencyKeyView] -> Encoding Source # omitField :: ConcurrencyKeyView -> Bool Source # | |||||
| 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 # | |||||
| Eq ConcurrencyKeyView Source # | |||||
Defined in Arbiter.Core.Concurrency.Stats Methods (==) :: ConcurrencyKeyView -> ConcurrencyKeyView -> Bool # (/=) :: ConcurrencyKeyView -> ConcurrencyKeyView -> Bool # | |||||
| 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-Kcm43gPyDGYL5vaZ4cZIJF" '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)))))) | |||||
data ConcurrencyPolicyUpdate Source #
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 Methods parseJSON :: Value -> Parser ConcurrencyPolicyUpdate Source # parseJSONList :: Value -> Parser [ConcurrencyPolicyUpdate] Source # | |||||
| ToJSON ConcurrencyPolicyUpdate Source # | |||||
Defined in Arbiter.Core.Concurrency.Stats Methods toJSON :: ConcurrencyPolicyUpdate -> Value Source # toEncoding :: ConcurrencyPolicyUpdate -> Encoding Source # toJSONList :: [ConcurrencyPolicyUpdate] -> Value Source # toEncodingList :: [ConcurrencyPolicyUpdate] -> Encoding Source # | |||||
| 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 # | |||||
| Eq ConcurrencyPolicyUpdate Source # | |||||
Defined in Arbiter.Core.Concurrency.Stats Methods (==) :: ConcurrencyPolicyUpdate -> ConcurrencyPolicyUpdate -> Bool # (/=) :: ConcurrencyPolicyUpdate -> ConcurrencyPolicyUpdate -> Bool # | |||||
| 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-Kcm43gPyDGYL5vaZ4cZIJF" 'False) (C1 ('MetaCons "ConcurrencyPolicyUpdate" 'PrefixI 'True) (S1 ('MetaSel ('Just "overrideLimit") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 (Maybe (Maybe Int32))))) | |||||
Operations
updateConcurrencyPolicyOverrides :: MonadArbiter m => Text -> ConcurrencyPolicyUpdate -> m Int64 Source #
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.
setConcurrencyLimit :: MonadArbiter m => ConcurrencyPolicy -> Int32 -> m Int64 Source #
Override a declared pool's limit for every key under it. Returns rows affected.
clearConcurrencyLimit :: MonadArbiter m => ConcurrencyPolicy -> m Int64 Source #
Drop a pool's override. Its declared limit applies again. Returns rows affected.
pruneConcurrencyKeys :: (MonadArbiter m, RegistryTables (RegistryOf m)) => m Int64 Source #
Delete drained concurrency rows with no live job. The reaper runs this.
reconcileConcurrencyCounts :: (MonadArbiter m, RegistryTables (RegistryOf m)) => m Int64 Source #
Recompute the concurrency counts from live jobs, repairing any trigger drift.
listConcurrencyPolicies :: MonadArbiter m => m [ConcurrencyPolicyView] Source #
List every concurrency pool with its default/override limit and live key and in-flight aggregates.
listConcurrencyKeys :: MonadArbiter m => Text -> Int -> Int -> m [ConcurrencyKeyView] Source #
List a prefix's keys with effective cap and in-flight fill fraction, paginated.