exchangealgebra-0.5.0.0: src/ExchangeAlgebra/Simulate/Lite.hs
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE TypeFamilies #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE FlexibleInstances #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE RankNTypes #-}
{-# LANGUAGE TypeOperators #-}
{-# LANGUAGE MultiParamTypeClasses #-}
{-# LANGUAGE KindSignatures #-}
{-# LANGUAGE ExistentialQuantification #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE ConstraintKinds #-}
{-# LANGUAGE QuantifiedConstraints #-}
{-# LANGUAGE UndecidableInstances #-}
{-# LANGUAGE AllowAmbiguousTypes #-}
{- |
Module : ExchangeAlgebra.Simulate.Lite
Copyright : (c) Kaya Akagi. 2018-2026
Maintainer : yakagika@icloud.com
Released under the OWL license
== What this module is
@Simulate.Lite@ is a small, additive front-end for agent-based bookkeeping
simulations. It sits /beside/ the classic "ExchangeAlgebra.Simulate" engine
(which is left completely unchanged) and trades some of its flexibility for
much less boilerplate and a clearer, /bulk-synchronous parallel/ (BSP)
semantics.
A model is described by three things:
1. a __product-only HKD world__ @w@ (one record, sibling fields, no
nesting, no sum types) whose fields are tagged by a /role/
(@'InitT'@, @'RefT' s@, @'SnapT'@) via the 'HK' type family;
2. a list of __stages__ ('Stage'); each stage maps its agents to
/messages/ (t'Journal') purely from a read-only world snapshot; and
3. a t'SimSpec' bundling the term range, seed, the ledger field selector,
the stages, and a parallelism policy.
'runLite' drives the BSP loop: per term, per stage, it freezes the world to
a snapshot, runs every agent against /that same snapshot/, and then commits
the merged messages to the ledger in one shot.
== Correspondence with the classic "ExchangeAlgebra.Simulate"
+--------------------------+----------------------------------+----------------------------------+
| concept | classic @Simulate@ | @Simulate.Lite@ |
+==========================+==================================+==================================+
| world state | @StateSpace@ + @Updatable@ | product-only HKD @w@ + 'HK' |
| | instances per field | (no per-field class instances) |
+--------------------------+----------------------------------+----------------------------------+
| mutable cell access | @UpdatableSTRef@ plumbed by hand | 'HK' @('RefT' s)@ = @STRef s@, |
| | | generated by 'gInit' |
+--------------------------+----------------------------------+----------------------------------+
| read-only view | read each ref where needed | one 'gFreeze' = @w 'SnapT'@ |
+--------------------------+----------------------------------+----------------------------------+
| per-agent step | imperative @ST@ that may read | pure @w 'SnapT' -> ... -> |
| | /and write/ shared refs | Journal@ (a /message/) |
+--------------------------+----------------------------------+----------------------------------+
| term boundary update | bespoke @Updatable@ logic | declarative 'Field' rule |
| | | (Carry\/ResetEach\/UpdateEach) |
+--------------------------+----------------------------------+----------------------------------+
== BSP semantics (differs from the classic engine)
Within a single stage every agent observes the __same snapshot__, taken once
at the start of the stage. An agent /cannot/ see the messages emitted by
earlier agents in the same stage (intra-stage invisibility). Messages from a
stage become visible only to /later/ stages (same term) and to later terms,
because the ledger commit happens after the whole stage has run. This is the
BSP "superstep" rule; the classic engine, by contrast, lets an imperative
step read mutations made earlier in the same step. Models ported from the
classic engine must respect this: split a read-then-write dependency into two
stages.
== Determinism
Each agent's 'StdGen' is derived deterministically from
@('specSeed', term index, stage index, agent index)@ only — never from
wall-clock, thread scheduling, or the chosen 'Par' policy. Consequently a
'Sequential' run and a 'ParChunk' run of the same t'SimSpec' produce the same
observable ledger (see the @DET-2@ test), and re-running is reproducible
(@DET-1@).
== Reading HKD type errors
If a world field is given the wrong element type, GHC reports the mismatch
with the 'HK' family already reduced to the field representation, e.g.
annotating @wPrice :: 'HK' f Double@ but using it as a @String@ yields:
> • Couldn't match type ‘[Char]’ with ‘Double’
> Expected: HK SnapT String
> Actual: Double
> • In the ‘wPrice’ field of a record
Read the @Expected@/@Actual@ lines as "this field is a @String@ here but a
@Double@ was expected" — the family is shown reduced, so there is no
instance-resolution wall to wade through.
== Scope
World records must be __product-only and non-nested__ (the generic traversal
in this module only handles @M1@ \/ @:*:@ \/ @K1@). Ledger retention, spill
and compaction are not decided in this module: they are supplied
declaratively as a t'LedgerPolicy' ("ExchangeAlgebra.Simulate.Policy") and
applied by 'runLiteWithPolicy'. Snapshot-dependent term-boundary
recomputation and parallel speedup measurement remain out of scope.
-}
module ExchangeAlgebra.Simulate.Lite
( -- * Term-boundary field rules
Field(..)
, carry
, resetEach
, updateEach
-- * Role tags and the HK field family
, InitT
, RefT
, SnapT
, HK
-- * Generic world traversal
-- The GLite* classes are exported name-only: their primed methods are the
-- Generic-Rep plumbing (instances for M1/(:*:)/K1 live in this module);
-- user code only ever names the classes in constraints.
, GLiteInit
, GLiteFreeze
, GLiteCommit
, LiteWorld
, gInit
, gFreeze
, gCommit
-- * Stages
-- 'Stage' is exported name-only: build stages with the smart constructors
-- ('stageFor'/'stage'/'stageOf') and read the name via 'stageName'.
, Stage
, stageFor
, stage
, stageOf
, stageName
-- * Simulation specification
, Par(..)
, SimSpec(..)
, mkSimSpec
-- * Runner
, runLite
-- * Policy-driven runner
, runLiteWithPolicy
) where
import GHC.Generics
import Data.Kind (Type)
import Control.Monad (forM_, when)
import Control.Monad.ST (ST, runST, RealWorld, stToIO)
import Data.STRef (STRef, newSTRef, readSTRef, writeSTRef, modifySTRef')
import Data.Hashable (hash)
import Data.IORef (newIORef, readIORef, writeIORef)
import System.IO (Handle, IOMode(WriteMode), withFile)
import System.Random (StdGen, mkStdGen)
import Control.Parallel.Strategies (parListChunk, rdeepseq, using)
import Control.DeepSeq (deepseq)
import qualified Data.Binary as Binary
import qualified Data.HashMap.Strict as HM
import ExchangeAlgebra.Journal ( Journal
, Note
, HatVal
, HatBaseClass
, sigma
, toMap
, fromMap
, (.|)
, filterWithNote )
import ExchangeAlgebra.Algebra (Alg)
import qualified ExchangeAlgebra.Algebra as EA ((.+), compress)
import ExchangeAlgebra.Simulate (StateTime)
import ExchangeAlgebra.Simulate.Spill (defaultBinarySpillWriter, stepBackWith)
import ExchangeAlgebra.Simulate.Policy ( LedgerPolicy(..)
, Retention(..)
, Compaction(..)
, HasTermAxis(..) )
------------------------------------------------------------------
-- * Term-boundary field rules
------------------------------------------------------------------
-- | How a world field is updated at a /term boundary/ (between superstep
-- sweeps of all stages). The constructor also carries the field's initial
-- value, so a single @w 'InitT'@ value fully describes both the starting world
-- and its boundary dynamics.
--
-- Note: within a term, a field only changes through the ledger commit of a
-- stage (for the ledger field) or not at all (for parameter fields). The
-- 'Field' rule fires exactly once per term, after every stage has run.
data Field a
= Carry !a -- ^ Keep the current ref value across the boundary
-- (the usual choice for an accumulating ledger).
| ResetEach !a -- ^ Overwrite with this value at every boundary.
| UpdateEach !a (a -> a) -- ^ Start from this value; apply the function at
-- every boundary (e.g. a decaying price).
-- | Smart constructor for 'Carry'.
carry :: a -> Field a
carry = Carry
-- | Smart constructor for 'ResetEach'.
resetEach :: a -> Field a
resetEach = ResetEach
-- | Smart constructor for 'UpdateEach'.
updateEach :: a -> (a -> a) -> Field a
updateEach = UpdateEach
------------------------------------------------------------------
-- * Role tags and the HK field family
------------------------------------------------------------------
-- | Role tag: a field as an /initial value + boundary rule/ ('Field').
data InitT
-- | Role tag: a field as a mutable @'STRef' s@ cell during a run. The state
-- region @s@ lives only on this tag (snapshots and inits are region-free), so
-- stages stay plain (non-rank-2) functions.
data RefT s
-- | Role tag: a field as a bare, read-only value — the BSP snapshot view. No
-- @Identity@ wrapper, so @w 'SnapT'@ reads exactly like an ordinary record.
data SnapT
-- | @'HK' f a@ chooses the representation of a world field of element type @a@
-- under role tag @f@.
type family HK (f :: Type) a where
HK InitT a = Field a
HK (RefT s) a = STRef s a
HK SnapT a = a
------------------------------------------------------------------
-- * Generic world traversal (product-only)
------------------------------------------------------------------
--
-- Three hand-written classes over the GHC.Generics product structure
-- (@M1@ / @:*:@ / @K1@ only — no @barbies@ or other HKD dependency). The
-- per-field leaf instances pattern-match on the concrete field shape
-- (@Field a@, @STRef s a@, bare @a@) so GHC never has to invert 'HK'.
-- | Build the @'RefT' s@ world from the @'InitT'@ world by allocating one
-- 'STRef' per field, seeded with the field's initial value.
class GLiteInit s i o where
gInit' :: i x -> ST s (o x)
instance GLiteInit s i o => GLiteInit s (M1 t m i) (M1 t m o) where
gInit' (M1 a) = M1 <$> gInit' a
instance (GLiteInit s i1 o1, GLiteInit s i2 o2)
=> GLiteInit s (i1 :*: i2) (o1 :*: o2) where
gInit' (a :*: b) = (:*:) <$> gInit' a <*> gInit' b
instance GLiteInit s (K1 r (Field a)) (K1 r (STRef s a)) where
gInit' (K1 fld) = K1 <$> newSTRef (initialOf fld)
-- | The starting value carried by a 'Field', regardless of its boundary rule.
initialOf :: Field a -> a
initialOf (Carry a) = a
initialOf (ResetEach a) = a
initialOf (UpdateEach a _) = a
-- | Freeze the @'RefT' s@ world to a bare-value snapshot (@'SnapT'@). Each
-- field is a single 'readSTRef'; immutable structures (arrays, maps) are shared
-- O(1), not copied.
class GLiteFreeze s i o where
gFreeze' :: i x -> ST s (o x)
instance GLiteFreeze s i o => GLiteFreeze s (M1 t m i) (M1 t m o) where
gFreeze' (M1 a) = M1 <$> gFreeze' a
instance (GLiteFreeze s i1 o1, GLiteFreeze s i2 o2)
=> GLiteFreeze s (i1 :*: i2) (o1 :*: o2) where
gFreeze' (a :*: b) = (:*:) <$> gFreeze' a <*> gFreeze' b
instance GLiteFreeze s (K1 r (STRef s a)) (K1 r a) where
gFreeze' (K1 ref) = K1 <$> readSTRef ref
-- | Apply each field's 'Field' boundary rule to the live refs, zipping the
-- @'InitT'@ structure (rules + initial values) against the @'RefT' s@ structure.
class GLiteCommit s i r where
gCommit' :: i x -> r x -> ST s ()
instance GLiteCommit s i r => GLiteCommit s (M1 t m i) (M1 t m r) where
gCommit' (M1 a) (M1 b) = gCommit' a b
instance (GLiteCommit s i1 r1, GLiteCommit s i2 r2)
=> GLiteCommit s (i1 :*: i2) (r1 :*: r2) where
gCommit' (a :*: b) (c :*: d) = gCommit' a c >> gCommit' b d
instance GLiteCommit s (K1 k (Field a)) (K1 k (STRef s a)) where
gCommit' (K1 fld) (K1 ref) =
case fld of
Carry _ -> pure ()
ResetEach a0 -> writeSTRef ref a0
UpdateEach _ f -> modifySTRef' ref f
-- | Constraint bundling everything a world type @w@ must satisfy to be driven
-- by 'runLite' in the @s@ region. It is a single (method-less) class so that it
-- can be used under a @forall s.@ quantified constraint in 'runLite' (a bare
-- @ConstraintKinds@ tuple synonym cannot). A product-only @deriving Generic@
-- on @w@ is all the user has to supply; the catch-all instance discharges the
-- rest automatically.
class ( Generic (w InitT)
, Generic (w (RefT s))
, Generic (w SnapT)
, GLiteInit s (Rep (w InitT)) (Rep (w (RefT s)))
, GLiteFreeze s (Rep (w (RefT s))) (Rep (w SnapT))
, GLiteCommit s (Rep (w InitT)) (Rep (w (RefT s)))
) => LiteWorld w s
instance ( Generic (w InitT)
, Generic (w (RefT s))
, Generic (w SnapT)
, GLiteInit s (Rep (w InitT)) (Rep (w (RefT s)))
, GLiteFreeze s (Rep (w (RefT s))) (Rep (w SnapT))
, GLiteCommit s (Rep (w InitT)) (Rep (w (RefT s)))
) => LiteWorld w s
-- | Allocate the live (ref) world from the initial world.
gInit :: (Generic (w InitT), Generic (w (RefT s)), GLiteInit s (Rep (w InitT)) (Rep (w (RefT s))))
=> w InitT -> ST s (w (RefT s))
gInit w = to <$> gInit' (from w)
-- | Take a read-only snapshot of the live world.
gFreeze :: (Generic (w (RefT s)), Generic (w SnapT), GLiteFreeze s (Rep (w (RefT s))) (Rep (w SnapT)))
=> w (RefT s) -> ST s (w SnapT)
gFreeze w = to <$> gFreeze' (from w)
-- | Apply every field's term-boundary 'Field' rule to the live world.
gCommit :: (Generic (w InitT), Generic (w (RefT s)), GLiteCommit s (Rep (w InitT)) (Rep (w (RefT s))))
=> w InitT -> w (RefT s) -> ST s ()
gCommit wi wr = gCommit' (from wi) (from wr)
------------------------------------------------------------------
-- * Stages
------------------------------------------------------------------
-- | A BSP stage: a named, pure mapping from agents to /messages/. Every agent
-- runs against the same snapshot @w 'SnapT'@, the current term @t@, and a
-- deterministically derived 'StdGen'. The agent element type @a@ is
-- existential, so different stages may use different agent populations.
--
-- There are two constructors:
--
-- * @StageFor@ — the original, fully general stage. Each agent emits a
-- t'Journal' directly, so the stage body is free to attach /any/ notes
-- (including several different notes from one agent). Built with 'stageFor'
-- or 'stage'.
--
-- * @StageTagged@ — a /note-tagged/ stage whose note type is fixed to
-- @(tag, t)@ by construction. Each agent emits a bare 'Alg' and the runner
-- attaches the single note @(stTag, t)@ in __one place__ (@runStage@). This
-- removes the write-site note duplication of @alg '.|' (Tag, t)@ and ties
-- the stage's tag to its note type at compile time. Built with 'stageOf'.
--
-- == Which constructor to use
--
-- Prefer 'stageOf' for a stage that emits __exactly one note tag__: the tag is
-- written once (as @stTag@), the runner supplies it, and a downstream
-- @projWithNote [(Tag, t)]@ that names the wrong constructor is a type error
-- rather than a silently empty projection. Use 'stageFor'\/'stage' when a single
-- stage must emit __several different notes__ (e.g. a closing stage that posts
-- both @(Closing, t)@ and @(Carryover, t+1)@): such a stage cannot be expressed
-- as a single auto-attached tag, so it keeps returning a t'Journal' itself.
--
-- The two constructors are observationally interchangeable for a one-note
-- stage: @'stageOf' tag as f@ produces the same messages as
-- @'stageFor' (show tag) as (\\v t g a -> f v t g a '.|' (tag, t))@ (asserted by
-- the @stageOf auto-note@ sentinel). @StageTagged@ only moves the @'.|' (tag, t)@
-- from the stage body into the runner.
data Stage w t n v b where
StageFor :: { stName :: String
, stAgents :: [a]
, stRun :: w SnapT -> t -> StdGen -> a -> Journal n v b }
-> Stage w t n v b
StageTagged :: (Note tag)
=> { stTag :: tag
, stAgentsT :: [a]
, stRunAlg :: w SnapT -> t -> StdGen -> a -> Alg v b }
-> Stage w t (tag, t) v b
-- | Build a stage that runs once per element of the given agent list, each
-- agent emitting a t'Journal' directly.
stageFor :: String
-> [a]
-> (w SnapT -> t -> StdGen -> a -> Journal n v b)
-> Stage w t n v b
stageFor = StageFor
-- | Build a singleton stage with no per-agent fan-out and no random draw —
-- handy for aggregate/bookkeeping steps. Equivalent to
-- @'stageFor' name [()] (\\v t _ () -> f v t)@.
stage :: String
-> (w SnapT -> t -> Journal n v b)
-> Stage w t n v b
stage name f = StageFor name [()] (\v t _g () -> f v t)
-- | Build a __note-tagged__ stage: each agent emits a bare 'Alg' and the runner
-- attaches the single note @(stTag, t)@ once, in @runStage@. The note type is
-- fixed to @(tag, t)@ by the result type, so the write-site tag (the @tag@
-- argument) and any read-site @projWithNote [(tag, t)]@ are checked against the
-- same constructor by the type-checker — a stringly-typed mismatch becomes a
-- compile error instead of a silently empty projection.
--
-- Use this for any stage that emits __exactly one note tag__. For a stage that
-- must post __several different notes__ (e.g. @(Closing, t)@ together with
-- @(Carryover, t+1)@), keep using 'stageFor'\/'stage', which let the body return
-- a fully general t'Journal'.
--
-- Determinism is unaffected: the per-agent 'StdGen' is still derived from
-- @('specSeed', termIx, stageIx, agentIx)@ only (see @runStage@), and the note
-- attachment @'.|' (stTag, t)@ is a pure post-transform of each agent's 'Alg'.
stageOf :: (Note tag)
=> tag
-> [a]
-> (w SnapT -> t -> StdGen -> a -> Alg v b)
-> Stage w t (tag, t) v b
stageOf = StageTagged
-- | The display name of a stage: @stName@ for @StageFor@, @show stTag@ for
-- @StageTagged@ (the tag's 'Show' comes from its 'Note' superclass). Use this
-- instead of @stName@ when a stage may be either constructor.
stageName :: Stage w t n v b -> String
stageName (StageFor nm _ _) = nm
stageName (StageTagged tg _ _) = show tg
------------------------------------------------------------------
-- * Simulation specification
------------------------------------------------------------------
-- | Parallelism policy for running a stage's agents. The chunk size is fixed
-- (input-size-only), so the partitioning — and therefore the result — does not
-- depend on the runtime scheduler (@DET-1@).
data Par = Sequential -- ^ Run agents left-to-right, no sparks.
| ParChunk !Int -- ^ Evaluate agent messages in fixed-size chunks
-- in parallel ('parListChunk').
-- | A complete simulation description.
data SimSpec w t n v b = SimSpec
{ specTerms :: (t, t)
-- ^ Inclusive @(from, to)@ term range, a runtime value (no compile-time
-- @lastTerm@ constant required).
, specSeed :: Int
-- ^ Master seed; all per-agent generators derive from it deterministically.
, specLedger :: forall f. w f -> HK f (Journal n v b)
-- ^ The ledger field selector (just the record accessor), used polymorphically
-- across roles to read the snapshot ledger and to write the live one.
--
-- __Which field is the ledger is a model declaration, not inferred.__ The
-- committed-ledger role is conferred by /this selector alone/: 'runLite'
-- commits every stage's merged messages only to the field it returns, and
-- retention\/spill\/compaction ('runLiteWithPolicy') all act on that same
-- field. The product type only fixes the three roles
-- (@'InitT'@\/@'RefT'@\/@'SnapT'@); it does /not/ mark one field as the
-- ledger and the others as auxiliary — that split is a discipline this
-- selector expresses, not a type-level guarantee.
--
-- Consequence: if the world has more than one @'Journal' n v b@ field,
-- pointing the selector at the wrong one type-checks and fails __silently__.
-- Commits, eviction and the final projection are then all consistently
-- applied to the wrong ledger, so a test that reads back through the same
-- selector can still pass. Prefer __exactly one__ @'Journal'@ field per
-- world, or name fields so the intended ledger is unmistakable. (A future
-- @newtype Ledger@ wrapper could make this distinction type-level; until
-- then it is a convention.)
, specStages :: [Stage w t n v b]
-- ^ Stages, run in this declared order each term.
, specParallel :: Par
-- ^ Agent-level parallelism policy.
}
-- | Smart constructor for t'SimSpec' defaulting to 'Sequential'. Prefer this over
-- the raw record so that future, additive t'SimSpec' fields stay non-breaking.
mkSimSpec :: (t, t)
-> Int
-> (forall f. w f -> HK f (Journal n v b))
-> [Stage w t n v b]
-> SimSpec w t n v b
mkSimSpec terms seed ledger stages = SimSpec
{ specTerms = terms
, specSeed = seed
, specLedger = ledger
, specStages = stages
, specParallel = Sequential
}
------------------------------------------------------------------
-- * Runner (BSP)
------------------------------------------------------------------
-- | Run a t'SimSpec' over the given initial world and project the final
-- snapshot through a continuation.
--
-- The loop, per term @t@ in @['from'..'to']@ and per stage in declared order:
--
-- 1. take one snapshot @view@ of the live world ('gFreeze');
-- 2. run every agent against /that/ snapshot (intra-stage invisibility),
-- producing one message (t'Journal') each — sequentially or in fixed-size
-- parallel chunks per 'specParallel';
-- 3. /flatten-once commit/: fold all messages into a single journal once
-- with the public 'sigma' (which itself folds into a 'Data.HashMap',
-- skipping zero journals, then rebuilds one journal), and add it to the
-- ledger ref with a single 'modifySTRef''.
--
-- After all stages of a term, every field's term-boundary 'Field' rule fires
-- ('gCommit'). The final live world is frozen once and handed to the
-- continuation.
runLite :: forall w t n v b r.
( forall s. LiteWorld w s
, HatVal v, HatBaseClass b, Note n, Enum t, Ord t )
=> SimSpec w t n v b
-> w InitT
-> (w SnapT -> r)
-> r
runLite spec wInit k = runST $ do
wr <- gInitR wInit
let (from0, to0) = specTerms spec
terms = enumFromThenToInclusive from0 to0
stages = zip [0 ..] (specStages spec)
forM_ (zip [0 ..] terms) $ \(termIx, t) -> do
forM_ stages $ \(stageIx, st) -> do
view <- gFreezeR wr
let msgs = runStage spec view t termIx stageIx st
delta = sigma msgs id :: Journal n v b
modifySTRef' (specLedger spec wr) (\acc -> acc EA..+ delta)
-- term boundary: fire the Field rules exactly once per term, AFTER
-- all stages of the term have committed (BSP semantics, design S3).
gCommitR wInit wr
final <- gFreezeR wr
pure (k final)
where
-- Pin the generic-traversal dictionaries at the @s@ chosen by 'runST'.
gInitR :: forall s. LiteWorld w s => w InitT -> ST s (w (RefT s))
gInitR = gInit
gFreezeR :: forall s. LiteWorld w s => w (RefT s) -> ST s (w SnapT)
gFreezeR = gFreeze
gCommitR :: forall s. LiteWorld w s => w InitT -> w (RefT s) -> ST s ()
gCommitR = gCommit
-- | Inclusive @[from .. to]@ for an 'Enum'/'Ord' term type. Empty when
-- @from > to@.
enumFromThenToInclusive :: (Enum t, Ord t) => t -> t -> [t]
enumFromThenToInclusive from0 to0
| from0 > to0 = []
| otherwise = [from0 .. to0]
-- | Produce the messages of one stage, applying the parallelism policy. Each
-- agent gets a 'StdGen' derived purely from
-- @('specSeed', termIx, stageIx, agentIx)@, so the result is independent of the
-- evaluation order or the 'Par' policy.
--
-- Under 'ParChunk' the FIRST message is forced to normal form in the calling
-- thread before the remaining messages are sparked. Every message reaches the
-- shared snapshot @view@ through lazily-built index structures (the Journal's
-- note-axis index, each Alg's base index); if many sparks race to force those
-- shared thunks, the RTS's duplicate-work suspension can re-enter its own
-- blackhole and abort with a spurious @<<loop>>@ (observed intermittently at
-- @-N4@; the thunk graph is acyclic — sequential evaluation never loops).
-- Forcing one message first materialises the shared structure in a single
-- thread, so the sparks only evaluate agent-local work. Pure values: the
-- result is unchanged (DET-2 asserts exact equality).
runStage :: forall w t n v b.
(HatVal v, HatBaseClass b, Note n)
=> SimSpec w t n v b
-> w SnapT
-> t
-> Int -- ^ term index (0-based)
-> Int -- ^ stage index (0-based)
-> Stage w t n v b
-> [Journal n v b]
runStage spec view t termIx stageIx st =
let seed0 = specSeed spec
-- The per-agent messages, before applying the parallelism policy. Both
-- constructors derive each generator from the same coordinates only, so
-- DET is identical to the StageFor-only runner. 'StageTagged' attaches
-- the single note @(stTag, t)@ here — the ONLY place auto-tagging
-- happens — as a pure post-transform of each agent's bare 'Alg'.
msgs = case st of
StageFor _ agents f ->
[ f view t (deriveGen seed0 termIx stageIx agentIx) a
| (agentIx, a) <- zip [0 ..] agents ]
StageTagged tg agents g ->
[ g view t (deriveGen seed0 termIx stageIx agentIx) a .| (tg, t)
| (agentIx, a) <- zip [0 ..] agents ]
in case specParallel spec of
Sequential -> msgs
ParChunk c -> case msgs of
[] -> []
(m : ms) -> m `deepseq`
(m : (ms `using` parListChunk (max 1 c) rdeepseq))
-- | Deterministically derive an agent's generator from the seed and the
-- @(term, stage, agent)@ coordinates only — never from scheduling. Uses a hash
-- of the tuple so that distinct coordinates almost never collide and the value
-- is stable across runs and 'Par' policies.
deriveGen :: Int -> Int -> Int -> Int -> StdGen
deriveGen seed0 termIx stageIx agentIx =
mkStdGen (hash (seed0, termIx, stageIx, agentIx))
{-# INLINE deriveGen #-}
------------------------------------------------------------------
-- * Policy-driven runner
------------------------------------------------------------------
-- | Run a t'SimSpec' under a declarative t'LedgerPolicy', returning in 'IO'
-- (because spill writes a file). The BSP loop is /identical/ to 'runLite' — per
-- term, per stage: snapshot, run agents, flatten-once commit — and the term
-- range, seed, stages, parallelism and per-agent generators are all the same.
-- The only additions happen at each term boundary, /after/ the stages have
-- committed and the 'Field' rules have fired:
--
-- 1. __compaction__ — if @'compaction' = 'CompressClosedTerms'@, every entry
-- of a /closed/ term (term @<@ the current term @t@) is 'EA.compress'ed.
-- This is norm- and balance-preserving; only the within-term posting
-- sequence is collapsed. The in-progress term is never touched.
-- 2. __retention / spill__ — if @'retain' = 'RetainRecent' w@, terms with
-- @'termOf' n '<=' t - w@ are evicted from the in-memory ledger. If
-- @'spillTo' = 'Just' path@, each newly-closed-and-evicted term is first
-- appended to that binary file (compatible with 'defaultBinarySpillWriter',
-- so 'ExchangeAlgebra.Simulate.Policy.restoreLedger' can read it back).
--
-- Under 'ExchangeAlgebra.Simulate.Policy.defaultLedgerPolicy' nothing is
-- compacted, evicted or spilled, so the result is observationally equal to
-- @'runLite' spec wInit k@ (see the equivalence test).
--
-- __Data loss warning.__ @'spillTo' = 'Nothing'@ together with
-- @'RetainRecent' w@ /discards/ evicted terms — they are written nowhere and
-- cannot be restored. Use it only when the older history is genuinely not
-- needed; otherwise set @'spillTo' = 'Just' path@.
runLiteWithPolicy
:: forall w t n v b r.
( forall s. LiteWorld w s
, HatVal v, HatBaseClass b
, HasTermAxis n, TermOf n ~ t
, StateTime t
, Binary.Binary t, Binary.Binary (Journal n v b) )
=> LedgerPolicy
-> SimSpec w t n v b
-> w InitT
-> (w SnapT -> r)
-> IO r
runLiteWithPolicy pol spec wInit k = do
let (from0, to0) = specTerms spec
terms = zip [0 :: Int ..] (enumFromThenToInclusive from0 to0)
stages = zip [0 :: Int ..] (specStages spec)
window = case retain pol of
RetainAll -> Nothing
RetainRecent w -> Just (max 0 w)
-- High-water mark of the most recent term already spilled (so each closed
-- term is written to disk at most once). 'Nothing' = nothing spilled yet.
spilledRef <- newIORef (Nothing :: Maybe t)
wr <- stToIO (gInitR wInit)
let -- Run one full term (all stages, then Field rules). Mirrors 'runLite'.
runTerm :: Int -> t -> ST RealWorld ()
runTerm termIx t = do
forM_ stages $ \(stageIx, st) -> do
view <- gFreezeR wr
let msgs = runStage spec view t termIx stageIx st
delta = sigma msgs id :: Journal n v b
modifySTRef' (specLedger spec wr) (\acc -> acc EA..+ delta)
gCommitR wInit wr
-- Apply 'CompressClosedTerms' to entries strictly before term @t@.
compactClosed :: t -> ST RealWorld ()
compactClosed t = case compaction pol of
FullAudit -> pure ()
CompressClosedTerms ->
modifySTRef' (specLedger spec wr)
(compressClosedTerms t)
withMaybeSpillHandle (spillTo pol) $ \mh ->
forM_ terms $ \(termIx, t) -> do
stToIO (runTerm termIx t)
stToIO (compactClosed t)
-- retention / spill at the term boundary
case window of
Nothing -> pure ()
Just w -> do
-- Eviction boundary for a w-term resident window ending at t:
-- evict every term <= t - w. The step-back arithmetic is
-- single-sourced in ExchangeAlgebra.Simulate ('stepBackWith';
-- design-review C4). When the boundary is below the spec's
-- first term the delete predicate matches nothing — no
-- clamping needed.
let boundary = stepBackWith pred w t -- evict terms <= boundary
spilledHi <- readIORef spilledRef
-- spill newly-closed terms (spilledHi, boundary] before deleting
case mh of
Nothing -> pure ()
Just h -> when (firstUnspilled spilledHi <= boundary) $ do
ledger <- stToIO (readSTRef (specLedger spec wr))
let lo = firstUnspilled spilledHi
chunk = filterWithNote
(\nn _ -> let tt = termOf nn
in tt >= lo && tt <= boundary)
ledger
-- Suppress empty spill chunks: at early term boundaries the
-- eviction window may not yet cover any resident term, so the
-- filtered chunk is empty. Writing it would emit a zero-entry
-- record to the binary file (restored as a no-op, but still a
-- wasted write); skip it. The high-water mark is still advanced
-- below, so a later non-empty term in (lo, boundary] is not lost.
when (not (HM.null (toMap chunk))) $
defaultBinarySpillWriter h (lo, boundary) chunk
-- delete evicted terms from memory (whether or not spilled)
stToIO $ modifySTRef' (specLedger spec wr)
(filterWithNote (\nn _ -> termOf nn > boundary))
writeIORef spilledRef (Just boundary)
final <- stToIO (gFreezeR wr)
pure (k final)
where
gInitR :: forall s. LiteWorld w s => w InitT -> ST s (w (RefT s))
gInitR = gInit
gFreezeR :: forall s. LiteWorld w s => w (RefT s) -> ST s (w SnapT)
gFreezeR = gFreeze
gCommitR :: forall s. LiteWorld w s => w InitT -> w (RefT s) -> ST s ()
gCommitR = gCommit
-- The first term that has not yet been spilled, given the high-water mark.
firstUnspilled :: Maybe t -> t
firstUnspilled Nothing = fst (specTerms spec)
firstUnspilled (Just hi) = succ hi
-- | Open the spill file in 'WriteMode' if a path is given; otherwise run the
-- continuation with no handle. Truncate on open: one run writes one fresh file;
-- a stale file from an earlier run would otherwise be appended to and fail the
-- range checks in
-- 'ExchangeAlgebra.Simulate.Spill.readBinarySpillFileChecked'.
withMaybeSpillHandle :: Maybe FilePath -> (Maybe Handle -> IO a) -> IO a
withMaybeSpillHandle Nothing act = act Nothing
withMaybeSpillHandle (Just path) act = withFile path WriteMode (act . Just)
-- | Apply 'EA.compress' to the entry of every Note whose term is strictly
-- before @t@ (a /closed/ term), leaving the in-progress term untouched. Uses
-- 'toMap'\/'fromMap' so the traversal order does not affect the result.
compressClosedTerms
:: (HasTermAxis n, TermOf n ~ t, Ord t, HatVal v, HatBaseClass b)
=> t -> Journal n v b -> Journal n v b
compressClosedTerms t j =
fromMap (HM.mapWithKey
(\nn alg -> if termOf nn < t then EA.compress alg else alg)
(toMap j))