automaton 1.7 → 1.8

raw patch · 8 files changed

+895/−5 lines, 8 filesdep +containersdep +freedep +sop-coredep ~basePVP ok

version bump matches the API change (PVP)

Dependencies added: containers, free, sop-core, time-domain

Dependency ranges changed: base

API changes (from Hackage documentation)

Files

• CHANGELOG.md +10/−0
• automaton.cabal +9/−2
• src/Data/Automaton/Schedule.hs +372/−0
• src/Data/Automaton/Schedule/Trans.hs +223/−0
• src/Data/Stream.hs +2/−2
• test/Automaton.hs +2/−0
• test/Automaton/Schedule.hs +273/−0
• test/Stream.hs +4/−1

@@ -1,5 +1,15 @@ # Revision history for automaton +## 1.8++* Added `Data.Automaton.Schedule` module with a new `MonadSchedule` class that+  works natively on `Automaton` values instead of monadic actions.+  Instances are provided for common monad transformers.+* Added `Data.Automaton.Schedule.Trans` module with `ScheduleT` and related transformers+  providing free waiting and scheduling effects+  (previously in `monad-schedule`).+* Removed the `monad-schedule` package dependency from `automaton`+ ## 1.7  * Add `safely`, `forever` and `foreverE` exception handling functions for streams

@@ -1,6 +1,6 @@ cabal-version: 3.0 name: automaton-version: 1.7+version: 1.8 synopsis: Effectful streams and automata in coalgebraic encoding description:   Effectful streams have an internal state and a step function.@@ -29,15 +29,19 @@ common opts   build-depends:     MonadRandom >=0.5,-    base >=4.16 && <4.22,+    base >=4.18 && <4.22,     changeset ^>=0.2,+    containers >=0.5,+    free >=5.1,     mmorph ^>=1.2,     mtl >=2.2 && <2.4,     profunctors ^>=5.6,     selective ^>=0.7,     semialign >=1.2 && <=1.4,     simple-affine-space ^>=0.2,+    sop-core ^>=0.5,     these >=1.1 && <=1.3,+    time-domain ^>=1.8,     transformers >=0.5,     witherable ^>=0.5, @@ -80,6 +84,8 @@     Data.Automaton     Data.Automaton.Filter     Data.Automaton.Recursive+    Data.Automaton.Schedule+    Data.Automaton.Schedule.Trans     Data.Automaton.Trans.Accum     Data.Automaton.Trans.Changeset     Data.Automaton.Trans.Except@@ -112,6 +118,7 @@     Automaton     Automaton.Except     Automaton.Filter+    Automaton.Schedule     Automaton.Trans.Accum     Automaton.Trans.Changeset     Stream

@@ -0,0 +1,372 @@+{-# LANGUAGE OverloadedLists #-}++{- | -+This module defines the 'MonadSchedule' class for running several automata concurrently,+and provides instances for common monad transformers.++The central abstraction is:++@'schedule' :: 'NonEmpty' ('Automaton' m a b) -> 'Automaton' m a b@++This takes a non-empty collection of automata and interleaves their outputs,+yielding one output per tick by cycling through the automata in some+monad-specific order (e.g. round-robin for 'Identity', first-available for 'IO').+For a free simulated-time scheduling monad transformer, see 'ScheduleT' in "Data.Automaton.Schedule.Trans".++== Relationship to @monad-schedule@++This class replaces the @MonadSchedule@ class from the @monad-schedule@ package+(which worked on monadic actions @m a@ directly) with one that works natively on 'Automaton' values.+The free waiting effect 'ScheduleT' from the @monad-schedule@ has been moved to @automaton@.+-}+module Data.Automaton.Schedule where++-- base+import Control.Arrow+import Control.Concurrent (forkIO, newEmptyMVar, putMVar, takeMVar)+import Control.Monad (forM_, guard, replicateM_)+import Control.Monad.IO.Class (MonadIO)+import Control.Monad.Identity (Identity (..))+import Data.Bifunctor qualified as Bifunctor+import Data.Foldable1 (Foldable1 (foldrMap1))+import Data.Function ((&))+import Data.Functor ((<&>))+import Data.Functor.Compose (Compose (..))+import Data.Kind (Type)+import Data.List qualified as List+import Data.List.NonEmpty as N+import Data.Maybe (maybeToList)+import Data.Tuple (swap)++-- transformers+import Control.Monad.Trans.Accum (AccumT (..), runAccumT)+import Control.Monad.Trans.Class (MonadTrans (..))+import Control.Monad.Trans.Except (ExceptT (..))+import Control.Monad.Trans.Maybe (MaybeT (..))+import Control.Monad.Trans.Reader (ReaderT (..))+import Control.Monad.Trans.State.Strict (StateT (..), get)+import Control.Monad.Trans.Writer.CPS qualified as CPS+import Control.Monad.Trans.Writer.Lazy qualified as Lazy+import Control.Monad.Trans.Writer.Strict qualified as Strict++-- sop-core+import Data.SOP (HCollapse (hcollapse), HSequence (htraverse'), I (..), K (..), NP (..), SListI, hmap, hzipWith)++-- containers+import Data.IntMap.Strict (IntMap)+import Data.IntMap.Strict qualified as IM+import Data.Sequence (Seq, ViewL (..), viewl)+import Data.Sequence qualified as Seq+import Data.Set qualified as S++-- mmorph+import Control.Monad.Morph (MFunctor)++-- witherable+import Witherable ((<&?>))++-- changeset+import Control.Monad.Trans.Changeset (ChangesetT (..))+import Data.Monoid.RightAction (RightAction)++-- time-domain+import Data.TimeDomain (TimeDifference (..))++-- automaton+import Data.Automaton (Automaton (..), arrM, constM, feedback, hoistS, initialised, liftS, reactimate, withAutomaton_)+import Data.Automaton qualified as Automaton+import Data.Automaton.Schedule.Trans (ScheduleT, SkipT, runScheduleS, runSkipS, scheduleS)+import Data.Automaton.Trans.Except (exceptS)+import Data.Automaton.Trans.Maybe (maybeExit, runMaybeS)+import Data.Automaton.Trans.Reader (readerS, runReaderS)+import Data.Automaton.Trans.State (modify, runStateS__)+import Data.Stream (StreamT (..), concatS)+import Data.Stream.Optimized (OptimizedStreamT (Stateful), toStreamT)+import Data.Stream.Result++{- | Class of monads that support running several 'Automaton's concurrently,+interleaving their outputs into a single 'Automaton'.++The semantics of 'schedule' depend on the monad:++* For 'Identity': round-robin (each automaton advances exactly once per cycle).+* For 'IO': all automata run in separate threads; results are delivered as soon+ as they are produced.+* For transformer stacks: defined compositionally by the individual instances.++The first input may be broadcast to an arbitrary number (1 for 'Identity', all for 'IO') of automata,+but subsequent inputs must be delivered to one automaton each.+-}+class MonadSchedule m where+  -- | Run a nonempty list of automata concurrently.+  schedule :: NonEmpty (Automaton m a b) -> Automaton m a b++{- | Start all streams in the background and send their values to a shared 'MVar'.++The first input is broadcast to all automata,+the following inputs are only broadcast to one each.+-}+instance MonadSchedule IO where+  schedule automata = proc a -> do+    (output, input) <- initialised startStreams -< a+    arrM $ uncurry putMVar -< (input, a)+    arrM takeMVar -< output+    where+      startStreams a0 = do+        output <- newEmptyMVar+        input <- newEmptyMVar+        forkIO $ replicateM_ (N.length automata - 1) $ putMVar input a0+        forM_ automata $ \automaton -> forkIO $ reactimate $ constM (takeMVar input) >>> automaton >>> arrM (putMVar output)+        return (output, input)++instance (Monad m, MonadSchedule m) => MonadSchedule (ReaderT r m) where+  schedule =+    fmap runReaderS+      >>> schedule+      >>> readerS++{- | Schedule automata in 'ExceptT'.++When any automaton throws an exception, all others are stopped immediately.+This is intentional: a thrown exception signals that the computation cannot+continue, so it is consistent to stop all peers.++To let all automata run to completion before propagating an exception, lift+them into a monad that does not short-circuit on exceptions (e.g. wrap the+exception type in 'Either' and post-process the results).+-}+instance (Monad m, MonadSchedule m) => MonadSchedule (ExceptT e m) where+  schedule =+    fmap exceptS+      >>> schedule+      >>> withAutomaton_ (fmap sequenceA >>> ExceptT)++{- | Schedule automata in 'MaybeT'.++When any automaton returns 'Nothing', all others are stopped immediately and the+combined automaton also returns 'Nothing'. This is intentional: 'Nothing'+signals termination, so it is consistent to stop all peers.++To let all automata run to completion before stopping, convert the 'MaybeT'+automata to base-monad automata producing @'Maybe' b@ values and post-process+the results.+-}+instance (Monad m, MonadSchedule m) => MonadSchedule (MaybeT m) where+  schedule =+    fmap runMaybeS+      >>> schedule+      >>> withAutomaton_ (fmap sequenceA >>> MaybeT)++{- | A monad transformer for scheduling automata that all need to run to+completion.++Like 'MaybeT', each automaton can signal termination by returning 'Nothing'.+Unlike 'MaybeT', the combined automaton does __not__ stop when one automaton+finishes; it waits until __all__ scheduled automata have finished.++While any automaton is still running, finished automata contribute no outputs.+Once every automaton has finished, the combined automaton itself terminates.+-}+newtype FinalizeT m a = FinalizeT+  { getFinalizeT :: MaybeT m a+  -- ^ Unwrap 'FinalizeT' to the underlying 'MaybeT'.+  }+  deriving newtype (Functor, Applicative, Monad, MonadIO, MonadTrans, MFunctor)++instance (Monad m, MonadSchedule m) => MonadSchedule (FinalizeT m) where+  schedule automata =+    automata+      & N.zip [1 ..]+      & fmap (\(i, automaton) -> runMaybeS (hoistS getFinalizeT automaton) <&> maybe (Left i) Right)+      & schedule+      & (>>> haveAllFinished)+      & liftS+      & (>>> maybeExit)+      & fmap maybeToList+      & Automaton.concatS+      & hoistS FinalizeT+    where+      allN = S.fromAscList [1 .. N.length automata]+      haveAllFinished = Automaton.unfold S.empty $ \input is -> case input of+        Left i -> let is' = S.insert i is in Result is' $ if is' == allN then Nothing else Just Nothing+        Right b -> Result is $ Just $ Just b++instance (Monoid w, Monad m, MonadSchedule m) => MonadSchedule (CPS.WriterT w m) where+  schedule =+    fmap (withAutomaton_ (CPS.runWriterT >>> fmap (\(Result s a, w) -> Result s (a, w))))+      >>> schedule+      >>> withAutomaton_ (fmap (\(Result s (a, w)) -> (Result s a, w)) >>> CPS.writerT)++instance (Monoid w, Monad m, MonadSchedule m) => MonadSchedule (Strict.WriterT w m) where+  schedule =+    fmap (withAutomaton_ (Strict.runWriterT >>> fmap (\(Result s a, w) -> Result s (a, w))))+      >>> schedule+      >>> withAutomaton_ (fmap (\(Result s (a, w)) -> (Result s a, w)) >>> Strict.WriterT)++instance (Monoid w, Monad m, MonadSchedule m) => MonadSchedule (Lazy.WriterT w m) where+  schedule =+    fmap (withAutomaton_ (Lazy.runWriterT >>> fmap (\(Result s a, w) -> Result s (a, w))))+      >>> schedule+      >>> withAutomaton_ (fmap (\(Result s (a, w)) -> (Result s a, w)) >>> Lazy.WriterT)++-- | This will share the accumulated log from the past with all automata+instance (Monoid w, Monad m, MonadSchedule m) => MonadSchedule (AccumT w m) where+  schedule =+    fmap (withAutomaton_ (runAccumT >>> ReaderT >>> CPS.writerT))+      >>> schedule+      >>> withAutomaton_ (CPS.runWriterT >>> runReaderT >>> AccumT)++-- | This will share the accumulated state from the past with all automata+instance (Monoid w, RightAction w s, Monad m, MonadSchedule m) => MonadSchedule (ChangesetT s w m) where+  schedule =+    fmap (withAutomaton_ (getChangesetT >>> ReaderT >>> fmap swap >>> CPS.writerT))+      >>> schedule+      >>> withAutomaton_ (CPS.runWriterT >>> fmap swap >>> runReaderT >>> ChangesetT)++{- | Cycle through all automata in a round-robin fashion.++On each tick of the combined 'Automaton', exactly one of the component+automata is stepped. The automata are advanced in the order they appear in+the input 'NonEmpty' list, cycling indefinitely.+-}+instance MonadSchedule Identity where+  schedule =+    fmap (getAutomaton >>> toStreamT)+      >>> foldrMap1 buildStreams consStreams+      >>> roundRobinStreams+      >>> fmap N.toList+      >>> concatS+      >>> Stateful+      >>> Automaton+    where+      buildStreams :: StreamT m b -> Streams m b+      buildStreams StreamT {state, step} =+        Streams+          { states = I state :* Nil+          , steps = Step (ResultStateT step) :* Nil+          }++      consStreams :: StreamT m b -> Streams m b -> Streams m b+      consStreams StreamT {state, step} Streams {states, steps} =+        Streams+          { states = I state :* states+          , steps = Step (ResultStateT step) :* steps+          }++-- The order of outputs matches the order of inputs: 'foldrMap1' places the+-- first input element at the head of the 'NP', so 'hnonemptycollapse' extracts+-- outputs in the original order.++{- | Step all streams in a 'Streams' bundle simultaneously and collect the+results into a 'NonEmpty' list, preserving the input order.+-}+roundRobinStreams :: (Functor m, Applicative m) => Streams m b -> StreamT m (NonEmpty b)+roundRobinStreams Streams {states, steps} =+  StreamT+    { state = states+    , step = \s ->+        s+          & hzipWith (\Step {getStep} (I s) -> getResultStateT getStep s <&> RunningResult & Compose) steps+          & htraverse' getCompose+          <&> ( \results ->+                  Result+                    (results & hmap (getRunningResult >>> resultState >>> I))+                    (results & hmap (getRunningResult >>> output >>> K) & hnonemptycollapse)+              )+    }++-- | Collapse a non-empty n-ary product of constant functors into a 'NonEmpty' list.+hnonemptycollapse :: (SListI as) => NP (K b) (a ': as) -> NonEmpty b+hnonemptycollapse (K a :* as) = a :| hcollapse as++-- | A nonempty list of 'StreamT's, unzipped into their states and their steps.+data Streams m b+  = forall state (states :: [Type]).+  (SListI states) =>+  Streams+  { states :: NP I (state ': states)+  , steps :: NP (Step m b) (state ': states)+  }++-- | One step of a stream, with the state type argument going last, so it is usable with sop-core.+newtype Step m b state = Step {getStep :: ResultStateT state m b}++-- | The result of a stream, with the type arguments swapped, so it's usable with sop-core+newtype RunningResult b state = RunningResult {getRunningResult :: Result state b}++instance (Monad m, MonadSchedule m) => MonadSchedule (SkipT m) where+  schedule = fmap runSkipS >>> schedule >>> fmap maybeToList >>> Automaton.concatS >>> liftS++-- | Each scheduled automaton must eventually produce an output or a diff greater than 'zero', otherwise this will loop indefinitely.+instance (Show diff, Ord diff, TimeDifference diff, Monad m, MonadSchedule m) => MonadSchedule (ScheduleT diff m) where+  schedule automata = automata & N.zip [1 ..] & fmap instrument & schedule & backpressure & scheduleS & Automaton.concatS+    where+      nAutomata = List.length automata+      instrument :: (Int, Automaton (ScheduleT diff m) a b) -> Automaton m (a, diff) (Int, Either (Maybe diff) b)+      instrument (i, automaton) = flip runStateS__ mempty $ proc (a, globalTime) -> do+        localTime <- constM get -< ()++        if globalTime < localTime+          -- We are ahead of the consensus time, skip this tick instead of emitting+          -- Each automaton ticks once per round-robin cycle over all N automata,+          -- so each input 'a' is delivered to each automaton exactly once per cycle.+          then do+            returnA -< (i, Left Nothing)+          else do+            diffOrOutput <- liftS $ runScheduleS automaton -< a+            case diffOrOutput of+              Left diffNew -> do+                arrM modify -< (`add` diffNew)+              _ -> returnA -< ()+            returnA -< (i, Bifunctor.first Just diffOrOutput)++      -- Each tick of 'schedule' advances exactly one of the N instrumented automata.+      -- 'backpressure' feeds the current consensus time back as the second input+      -- so that automata which are ahead of the consensus will skip their tick.+      backpressure :: Automaton m (a, diff) (Int, Either (Maybe diff) b) -> Automaton m a (Either diff [b])+      backpressure scheduled = feedback (IM.fromAscList $ (,Seq.Empty) <$> [1 .. nAutomata], mempty) $ proc (a, (queues, lastGlobalTime)) -> do+        (i, outputOrDiffMaybe) <- scheduled -< (a, lastGlobalTime)+        let (output, queues') = popOutput $ enqueueOutput i outputOrDiffMaybe queues+        returnA -< (output, (queues', lastGlobalTime & either add (const id) output))++      enqueueOutput :: Int -> Either (Maybe diff) b -> IntMap (Seq (Either diff b)) -> IntMap (Seq (Either diff b))+      enqueueOutput i = \case+        Left Nothing -> id+        Left (Just diff) -> IM.insertWith (<>) i $ Seq.singleton $ Left diff+        Right b -> IM.insertWith (<>) i $ Seq.singleton $ Right b++      analyseQueue :: IntMap (Seq (Either diff b)) -> Maybe (IntMap (Seq (Either diff b)), (IntMap diff, [b]))+      analyseQueue =+        let peekOutput i queuei = do+              case viewl queuei of+                -- Queue empty: we cannot yet determine a consensus output for this+                -- automaton, so we abort and wait for more input from 'schedule'.+                -- This conservatively waits even if some other automata already+                -- have outputs queued; a more aggressive variant could return+                -- those partial results early.+                EmptyL -> lift Nothing+                -- New diff enqueued.+                Left diff :< queuei' -> do+                  modify $ Bifunctor.first $ IM.insert i diff+                  pure queuei'+                -- New output.+                Right b :< queuei' -> do+                  modify $ Bifunctor.second (b :)+                  pure queuei'+         in flip runStateT (IM.empty, []) . IM.traverseWithKey peekOutput++      pushDiffsBack :: IntMap diff -> IntMap (Seq (Either diff b)) -> IntMap (Seq (Either diff b))+      pushDiffsBack diffs = IM.unionWith (<>) $ diffs <&> pure . Left++      popOutput :: IntMap (Seq (Either diff b)) -> (Either diff [b], IntMap (Seq (Either diff b)))+      popOutput queues = case analyseQueue queues of+        Nothing -> (Right [], queues) -- Queues weren't full yet, need to wait for more input+        Just (queues', (diffs, bs)) -> case N.nonEmpty bs of+          Nothing ->+            if IM.null diffs+              then (Right [], queues') -- No diffs or outputs enqueued+              else+                let minDiff = minimum diffs+                    adjustedDiffs = diffs <&?> \diff -> guard (diff /= minDiff) >> Just (diff `difference` minDiff)+                 in (Left minDiff, pushDiffsBack adjustedDiffs queues')+          Just bs -> (Right $ List.reverse $ toList bs, pushDiffsBack diffs queues')

@@ -0,0 +1,223 @@+{-# LANGUAGE DeriveFunctor #-}++{- |+This module supplies a general-purpose monad transformer that adds a syntactical+"delay", or "waiting" side effect, used for time-based scheduling of automata.++'ScheduleT' is used as the monad for automata that need to express timing+information: an automaton in @'ScheduleT' diff m@ can @'wait' diff@ to signal+how long until its next output.++The 'Data.Automaton.Schedule.MonadSchedule' instance for 'ScheduleT' interleaves+several such automata in time order.+-}+module Data.Automaton.Schedule.Trans (module Data.Automaton.Schedule.Trans) where++-- base+import Control.Concurrent (threadDelay)+import Data.Functor.Classes (Eq1 (..), liftEq)+import Data.Functor.Identity (Identity (..))+import Data.Ord (comparing)++-- transformers+import Control.Monad.IO.Class (MonadIO, liftIO)++-- free+import Control.Monad.Trans.Free (FreeF (..), FreeT (..), iterT, liftF, runFreeT)++-- time-domain++import Control.Monad.Trans.Class (MonadTrans (..))+import Control.Monad.Trans.Reader (ReaderT (..))+import Data.Automaton (Automaton, handleAutomaton)+import Data.Stream (StreamT (..))+import Data.Stream.Result (Result (..))+import Data.TimeDomain (TimeDifference (..))++-- * Waiting action++-- | A functor implementing a syntactical "waiting" action.+data Wait diff a = Wait+  { getDiff :: diff+  -- ^ The duration to wait.+  , awaited :: a+  -- ^ The encapsulated value.+  }+  deriving (Functor, Eq, Show)++instance (Eq diff) => Eq1 (Wait diff) where+  liftEq eq (Wait diff1 a) (Wait diff2 b) = diff1 == diff2 && eq a b++{- | Compare by the time difference, regardless of the value.++Note that this would not give a lawful 'Ord' instance since we do not compare+the @a@.+-}+compareWait :: (Ord diff) => Wait diff a -> Wait diff a -> Ordering+compareWait = comparing getDiff++-- * 'ScheduleT'++{- |+Values in @ScheduleT diff m@ are delayed computations with side effects in @m@.+Delays can occur between any two side effects, with lengths specified by a @diff@+value.++These delays don't have any semantics on their own; semantics can be given with+'runScheduleT'.++The 'Data.Automaton.Schedule.MonadSchedule' instance for 'ScheduleT' interprets+delays as logical time and interleaves several such computations in order of+their next scheduled time.+-}+type ScheduleT diff = FreeT (Wait diff)++-- | 'ScheduleT' over the 'Identity' monad.+type Schedule diff = ScheduleT diff Identity++-- | The side effect that waits for a specified amount.+wait :: (Monad m) => diff -> ScheduleT diff m ()+wait diff = FreeT $ pure $ Free $ Wait diff $ pure ()++{- | Supply a semantic meaning to 'Wait'.+For every occurrence of @Wait diff@ in the @ScheduleT diff m a@ value,+a waiting action is executed, depending on @diff@.+-}+runScheduleT :: (Monad m) => (diff -> m ()) -> ScheduleT diff m a -> m a+runScheduleT waitAction = iterT $ \(Wait n ma) -> waitAction n >> ma++{- | Run a 'ScheduleT' value, ignoring the waiting actions.++Usually, you would apply this function after having scheduled several automata together with 'schedule',+and you want to get the final result of the schedule without caring about the timing.+-}+evalScheduleT :: (Monad m) => ScheduleT diff m a -> m a+evalScheduleT = runScheduleT $ const $ pure ()++-- | Run a 'Schedule' value, ignoring the waiting actions.+evalSchedule :: Schedule diff a -> a+evalSchedule = runIdentity . evalScheduleT++{- | Run a 'ScheduleT' value in a 'MonadIO',+interpreting the times as milliseconds.+-}+runScheduleIO ::+  (MonadIO m, Integral n) =>+  ScheduleT n m a ->+  m a+runScheduleIO = runScheduleT waitms++{- | Formally execute all waiting actions,+returning the final value and all moments when the schedule would have waited.+-}+execScheduleT :: (Monad m) => ScheduleT diff m a -> m (a, [diff])+execScheduleT action = do+  free <- runFreeT action+  case free of+    Pure a -> pure (a, [])+    Free (Wait diff cont) -> do+      (a, diffs) <- execScheduleT cont+      pure (a, diff : diffs)++{- | Break down the steps of an 'Automaton' in 'ScheduleT' into waiting+effects and returning values.++Each tick either produces a @'Right' b@ (a regular output) or a+@'Left' diff@ (a wait duration), exposing the internal scheduling information+to the caller.  The dual of 'scheduleS'.+-}+runScheduleS :: (Functor m, Monad m) => Automaton (ScheduleT diff m) a b -> Automaton m a (Either diff b)+runScheduleS = handleAutomaton $ \StreamT {state, step} ->+  StreamT+    { state = step state+    , step = \s -> ReaderT $ \a -> do+        oneStep <- runFreeT $ runReaderT s a+        pure $ case oneStep of+          Pure (Result s' b) -> Result (step s') (Right b)+          Free (Wait diff cont) -> Result (lift cont) (Left diff)+    }++{- | Embed an automaton that produces @'Either' diff b@ values into+'ScheduleT', interpreting @'Left' diff@ as a 'wait' instruction.++The dual of 'runScheduleS': whenever the inner automaton emits @'Left' diff@,+'scheduleS' calls @'wait' diff@ and then re-runs the step with the same+input until a @'Right' b@ is produced.+-}+scheduleS :: (Monad m) => Automaton m a (Either diff b) -> Automaton (ScheduleT diff m) a b+scheduleS = handleAutomaton $ \StreamT {state, step} ->+  let step' s = ReaderT $ \a -> do+        Result s' eitherDiffB <- lift $ runReaderT (step s) a+        case eitherDiffB of+          Right b -> pure $ Result s' b+          Left diff -> do+            wait diff+            runReaderT (step' s') a+   in StreamT+        { state+        , step = step'+        }++-- * The symbolic effect of skipping one step of an automaton++{- | A monad transformer that adds the ability to __skip__ one output step.++An automaton in @'SkipT' m@ may call 'skip' to signal that it wants to+defer its output to the next step.  'MonadSchedule' for 'SkipT' uses this to+implement cooperative, non-preemptive round-robin scheduling: an automaton+that has not yet skipped enough times simply skips instead of emitting an+output, giving other automata a chance to catch up.++See also 'runSkipS' and 'skip'.+-}+newtype SkipT m a = SkipT+  { getSkipT :: FreeT Identity m a+  -- ^ Unwrap 'SkipT' to the underlying free-monad transformer.+  }+  deriving newtype (Functor, Applicative, Monad, MonadTrans, MonadIO)++-- | 'SkipT' specialised to 'Identity': a pure automaton with skip steps.+type Yield = SkipT Identity++{- | Run an 'Automaton' in 'SkipT', exposing skipped steps as 'Nothing'.++Each tick of the result automaton corresponds to one tick of the input+automaton.  If the input automaton called 'skip' on that tick the result is+'Nothing'; otherwise it is 'Just' the output.+-}+runSkipS :: (Functor m, Monad m) => Automaton (SkipT m) a b -> Automaton m a (Maybe b)+runSkipS = handleAutomaton $ \StreamT {state, step} ->+  StreamT+    { state = step state+    , step = \s -> ReaderT $ \a -> do+        oneTick <- runFreeT $ getSkipT $ runReaderT s a+        pure $ case oneTick of+          Pure (Result s' b) -> Result (step s') (Just b)+          Free (Identity cont) -> Result (lift $ SkipT cont) Nothing+    }++-- | Signal that the current step should be skipped, deferring output to the next tick.+skip :: (Monad m) => SkipT m ()+skip = SkipT $ liftF $ pure ()++-- | Run a 'SkipT' action, discarding all 'skip' steps.+runSkipT :: (Monad m) => SkipT m a -> m a+runSkipT = iterT runIdentity . getSkipT++-- | Run a 'SkipT' action, executing @action@ for every 'skip' step.+runSkipTWith :: (Monad m) => m () -> SkipT m a -> m a+runSkipTWith action = iterT (\ima -> action >> runIdentity ima) . getSkipT++-- | Run a pure 'Yield' computation, discarding all skipped steps.+runYield :: Yield a -> a+runYield = runIdentity . runSkipT++-- * Helper functions++-- | Wait for the given number of milliseconds.+waitms :: (MonadIO m, Integral n) => n -> m ()+waitms = liftIO . threadDelay . (* 1000) . fromIntegral++-- | Check whether a time difference is zero (i.e. equal to its own difference with itself).+isZero :: (Eq diff, TimeDifference diff) => diff -> Bool+isZero diff = diff `difference` diff == diff

@@ -263,8 +263,8 @@ reactimate StreamT {state, step} = go state   where     go s = do-      Result s' () <- step s-      go s'+      Result s' unit <- step s+      unit `seq` go s' {-# INLINE reactimate #-}  -- | Run a stream, collecting the outputs in a lazy, infinite list.

@@ -29,6 +29,7 @@ -- automaton import Automaton.Except import Automaton.Filter+import Automaton.Schedule import Automaton.Trans.Accum import Automaton.Trans.Changeset import Data.Automaton@@ -84,6 +85,7 @@     , Automaton.Filter.tests     , Automaton.Trans.Accum.tests     , Automaton.Trans.Changeset.tests+    , Automaton.Schedule.tests     ]  inMaybe :: Automaton Maybe (Maybe a) a

@@ -0,0 +1,273 @@+module Automaton.Schedule where++-- base+import Control.Category ((>>>))+import Control.Concurrent (threadDelay, yield)+import Control.Monad (replicateM)+import Control.Monad.IO.Class (MonadIO (..))+import Control.Monad.Identity (Identity (runIdentity))+import Data.Foldable (Foldable (..))+import Data.Functor (($>))+import Data.List qualified as List+import Data.List.NonEmpty (NonEmpty (..))+import Data.Maybe (fromJust, isJust)+import Data.Monoid (Sum (..))++-- quickcheck+import Test.QuickCheck+import Test.QuickCheck.Monadic (monadicIO, run)++-- transformers+import Control.Monad.Trans.Class (MonadTrans (..))+import Control.Monad.Trans.Except (runExceptT, throwE)+import Control.Monad.Trans.Maybe (MaybeT (..))+import Control.Monad.Trans.Writer.Strict (WriterT (..), tell)++-- mmorph+import Control.Monad.Morph (MFunctor (..))++-- changeset+import Control.Monad.Changeset.Class (MonadChangeset (..))+import Control.Monad.Trans.Changeset (Count (Increment), changeSingle, runChangeset)++-- tasty+import Test.Tasty (TestTree, testGroup)++-- tasty-quickcheck+import Test.Tasty.QuickCheck (testProperty)++-- tasty-hunit+import Test.Tasty.HUnit (testCase, (@?=))++-- automaton+import Data.Automaton (Automaton, accumulateWith, arrM, constM, embed, hoistS)+import Data.Automaton qualified as Automaton+import Data.Automaton.Schedule (FinalizeT (..), MonadSchedule, schedule)+import Data.Automaton.Schedule.Trans (ScheduleT, runScheduleT, runSkipS, runYield, skip, wait)+import Data.Automaton.Trans.Maybe (runMaybeS)+import Data.Stream.Result (Result (..))+import Data.TimeDomain (Seconds (..))++tests =+  testGroup+    "Schedule"+    [ testGroup+        "FinalizeT"+        [ testCase "FinalizeT stops when the single automaton completes" $ do+            -- A single automaton counting down from 3 should produce 3 outputs then stop.+            -- We lower to Identity via runMaybeS so outputs are not discarded on termination.+            let outputs = embedFinalize (schedule $ pure (make 3))+            length outputs @?= 3+        , testCase "FinalizeT waits for all automata to complete" $ do+            -- Two automata: A counts down from 2, B from 3.+            -- Identity schedule runs both in lockstep; outputs are interleaved then flattened.+            -- Total outputs: 2 + 3 = 5+            let outputs = embedFinalize (schedule $ make 2 :| [make 3])+            length outputs @?= 5+        , testCase "FinalizeT does not stop when the first automaton completes" $ do+            -- Unlike MaybeT, FinalizeT keeps running while any automaton is still alive.+            -- A (2 steps) finishes first; B (3 steps) finishes last.+            let maybeOutputs = embedMaybe (schedule $ makeMaybe 2 :| [makeMaybe 3])+                finalizeOutputs = embedFinalize (schedule $ make 2 :| [make 3])+            -- MaybeT stops when A finishes: both emit 2 outputs before MaybeT short-circuits+            length maybeOutputs @?= 4+            -- FinalizeT keeps going until B also finishes: 2 + 3 = 5 outputs+            length finalizeOutputs @?= 5+        , testCase "FinalizeT round-robin output order" $ do+            -- Two automata: A emits 10, 20 then stops; B emits 100, 200, 300 then stops.+            -- Identity schedule steps both simultaneously, then concatS flattens:+            --   tick 1: [10, 100], tick 2: [20, 200], tick 3: [300] (A done, B still live)+            -- Outputs: [10, 100, 20, 200, 300]+            let mkCounting startVal stepVal n =+                  Automaton.unfoldM (0 :: Int) $ const $ \k ->+                    if k >= n+                      then FinalizeT $ MaybeT $ pure Nothing+                      else FinalizeT $ MaybeT $ pure $ Just $ Result (k + 1) (startVal + stepVal * k :: Int)+                outputs = embedFinalize (schedule $ mkCounting 10 10 2 :| [mkCounting 100 100 3])+            outputs @?= [10, 100, 20, 200, 300]+        ]+    , testGroup+        "SkipT"+        [ testCase "SkipT skips an output step" $ do+            let output = runIdentity $ embed (runSkipS $ constM (waitSkip 5 $> (5 :: Int)) >>> accumulateWith (+) 0) $ replicate 10 ()+            output @?= [Nothing, Nothing, Nothing, Nothing, Just 5, Nothing, Nothing, Nothing, Nothing, Just 10]+        , testCase "schedule waits chronologically (mirrored)" $ do+            let output = runIdentity $ embed (runSkipS $ constM (waitSkip 3 $> (3 :: Int)) >>> accumulateWith (+) 0) $ replicate 10 ()+            output @?= [Nothing, Nothing, Just 3, Nothing, Nothing, Just 6, Nothing, Nothing, Just 9, Nothing]+        ]+    , testGroup+        "Yield"+        [ testCase "schedule waits chronologically" $ do+            let output = runYield $ embed (schedule $ (\n -> constM (waitSkip n $> n) >>> accumulateWith (+) 0) <$> 3 :| [5]) $ replicate 10 ()+            output @?= [3, 5, 6, 9, 10, 12, 15, 15, 18, 20]+        , testCase "schedule waits chronologically (mirrored)" $ do+            let output = runYield $ embed (schedule $ (\n -> constM (waitSkip n $> n) >>> accumulateWith (+) 0) <$> 5 :| [3]) $ replicate 10 ()+            output @?= [3, 5, 6, 9, 10, 12, 15, 15, 18, 20]+        ]+    , scheduleTests+        "ScheduleT IO busy"+        id+        (const yield)+    , scheduleTests+        "ScheduleT IO with delay"+        id+        (liftIO . threadDelay . (* 10) . fromIntegral)+    , scheduleTests+        "ScheduleT Identity"+        (pure . runIdentity)+        (const $ pure ())+    , testGroup+        "ChangesetT"+        [ testCase "Single automaton is unchanged" $ do+            let output = flip runChangeset (0 :: Int) $ flip embed (replicate 5 ()) $ schedule $ pure $ constM $ changeSingle Increment >> current+            output @?= ([1, 2, 3, 4, 5], 5)+        , testCase "Two automata see global state" $ do+            let output = flip runChangeset (0 :: Int) $ flip embed (replicate 10 ()) $ schedule $ constM (changeSingle Increment >> pure (-1)) :| [constM current]+            output+              @?= (+                    [ -1+                    , 0 -- First tick of both automata: Second one doesn't yet see the log of the other+                    , -1+                    , 1 -- Second joint tick: Log from the first reaches the second automaton+                    , -1+                    , 2+                    , -1+                    , 3+                    , -1+                    , 4+                    ]+                  , 5+                  )+        , testCase "Two automata see global state (mirrored)" $ do+            let output = flip runChangeset (0 :: Int) $ flip embed (replicate 10 ()) $ schedule $ constM current :| [constM (changeSingle Increment >> pure (-1))]+            output+              @?= (+                    [ 0 -- First tick of both automata: Second one doesn't yet see the log of the other+                    , -1+                    , 1 -- Second joint tick: Log from the first reaches the second automaton+                    , -1+                    , 2+                    , -1+                    , 3+                    , -1+                    , 4+                    , -1+                    ]+                  , 5+                  )+        ]+    , testGroup+        "MaybeT"+        [ testCase "MaybeT stops all automata when one returns Nothing" $ do+            -- Two automata: one finite (stops after 3 steps), one infinite.+            -- When the finite one stops, the whole scheduled automaton stops too.+            let finite =+                  Automaton.unfoldM (3 :: Int) $ const $ \n ->+                    if n <= 0+                      then MaybeT $ pure Nothing+                      else pure $ Result (n - 1) n+                infinite = constM $ pure (0 :: Int)+                output = runIdentity $ embed (runMaybeS (schedule $ finite :| [infinite])) $ replicate 10 ()+            -- Stops as soon as finite returns Nothing+            output @?= [Just 3, Just 0, Just 2, Just 0, Just 1, Just 0, Nothing, Nothing, Nothing, Nothing]+        , testCase "MaybeT all-Nothing immediately gives Nothing" $ do+            let alwaysNothing = constM (MaybeT $ pure Nothing) :: Automaton (MaybeT Identity) () Int+                output = runIdentity $ embed (runMaybeS (schedule $ alwaysNothing :| [alwaysNothing])) (replicate 5 ())+            output @?= [Nothing, Nothing, Nothing, Nothing, Nothing]+        ]+    , testGroup+        "ExceptT"+        [ testCase "ExceptT stops all automata when one throws" $ do+            -- One automaton throws after 3 steps, the other never throws.+            let throwing =+                  Automaton.unfoldM (3 :: Int) $ const $ \n ->+                    if n <= 0+                      then throwE ("done" :: String)+                      else pure $ Result (n - 1) n+                nonthrowing = constM $ pure (0 :: Int)+                output = runIdentity $ runExceptT $ embed (schedule $ throwing :| [nonthrowing]) $ replicate 10 ()+            -- Should terminate with Left, not run all 10 steps+            case output of+              Left _ -> pure ()+              Right xs -> fail $ "Expected Left but got Right with " <> (show (length xs) <> " elements")+        ]+    ]+  where+    waitSkip n = replicateM (n - 1) skip+    -- \| Automaton in FinalizeT that counts down from n, outputting each value, then terminates.+    make :: Int -> Automaton (FinalizeT Identity) () Int+    make n = Automaton.unfoldM n $ const $ \k ->+      if k <= 0+        then FinalizeT $ MaybeT $ pure Nothing+        else FinalizeT $ MaybeT $ pure $ Just $ Result (k - 1) k+    -- \| Automaton in MaybeT that counts down from n, outputting each value, then terminates.+    makeMaybe :: Int -> Automaton (MaybeT Identity) () Int+    makeMaybe n = Automaton.unfoldM n $ const $ \k ->+      if k <= 0+        then MaybeT $ pure Nothing+        else MaybeT $ pure $ Just $ Result (k - 1) k+    -- \| Run a FinalizeT automaton and collect all outputs produced before it terminates.+    -- Uses runMaybeS to lower to Identity, so the outputs are not discarded on termination.+    embedFinalize :: Automaton (FinalizeT Identity) () b -> [b]+    embedFinalize automaton =+      fmap fromJust+        . takeWhile isJust+        . runIdentity+        $ embed (runMaybeS $ hoistS getFinalizeT automaton) (replicate 100 ())+    -- \| Run a MaybeT automaton and collect all outputs produced before it terminates.+    embedMaybe :: Automaton (MaybeT Identity) () b -> [b]+    embedMaybe automaton =+      fmap fromJust+        . takeWhile isJust+        . runIdentity+        $ embed (runMaybeS automaton) (replicate 100 ())++scheduleTests ::+  (Monad m, MonadSchedule m) =>+  String ->+  (forall a. m a -> IO a) ->+  (Seconds Integer -> m ()) ->+  TestTree+scheduleTests name runProperty interpretSchedule =+  testGroup+    name+    [ testCase "schedule waits chronologically" $ do+        output <- runProperty $ runScheduleT interpretSchedule $ embed (schedule $ mkClock <$> 3 :| [5]) $ replicate 10 ()+        output @?= [3, 5, 6, 9, 10, 12, 15, 15, 18, 20]+    , testCase "schedule chronologically (mirrored)" $ do+        output <- runProperty $ runScheduleT interpretSchedule $ embed (schedule $ mkClock <$> 5 :| [3]) $ replicate 10 ()+        output @?= [3, 5, 6, 9, 10, 12, 15, 15, 18, 20]+    , testProperty "multiple automata schedule correctly" $ within 1_000_000 $ \(diffss :: NonEmpty Diffs) -> monadicIO $ do+        output <- fmap (List.nub . snd) $ run $ runProperty $ runScheduleT interpretSchedule $ runWriterT $ runMaybeT $ embed (hoistS (hoist lift) (schedule $ runningClock <$> diffss) >>> arrM (lift . tell . pure)) $ repeat ()+        pure $ output === take (length output) (List.nub $ List.sort $ concatMap accDiffs diffss)+    ]+  where+    mkClock n = constM (wait n $> n) >>> accumulateWith (+) (0 :: Seconds Integer)++-- These instances are needed because 'time-domain' does not yet provide+-- 'Semigroup'/'Monoid' for 'Integer'. Remove once time-domain is updated.+deriving via (Sum Integer) instance Semigroup Integer++deriving via (Sum Integer) instance Monoid Integer++type Diffs = [Positive (Seconds Integer)]++type Times = [Seconds Integer]++accDiffs :: Diffs -> Times+accDiffs = drop 1 . scanl (\t (Positive dt) -> t + dt) 0 . toList++interpretDiffs :: (Monad m) => Diffs -> Automaton (MaybeT (ScheduleT (Seconds Integer) m)) () (Seconds Integer)+interpretDiffs diffs0 = Automaton.unfoldM (toList diffs0) $ const $ \case+  [] -> MaybeT $ pure Nothing+  (Positive diff : diffs) -> lift (wait diff) >> pure (Result diffs diff)++runningClock :: (Monad m) => Diffs -> Automaton (MaybeT (ScheduleT (Seconds Integer) m)) () (Seconds Integer)+runningClock diffs = interpretDiffs diffs >>> accumulateWith (+) 0++-- 'NonEmpty' has no 'Arbitrary' instance in QuickCheck as of 2.14.+instance (Arbitrary a) => Arbitrary (NonEmpty a) where+  arbitrary = (:|) <$> arbitrary <*> arbitrary++instance (Arbitrary a) => Arbitrary (Seconds a) where+  arbitrary = Seconds <$> arbitrary

@@ -20,7 +20,7 @@ import Test.Tasty.HUnit (testCase, (@?=))  -- automaton-import Data.Stream (StreamT (..), constM, foreverExceptE, handleExceptT, handleWriterT, mmap, snapshot, streamToList, unfold, unfold_)+import Data.Stream (StreamT (..), concatS, constM, foreverExceptE, handleExceptT, handleWriterT, mmap, snapshot, streamToList, unfold, unfold_) import Data.Stream.Except qualified as StreamExcept import Data.Stream.Optimized qualified as StreamOptimized import Data.Stream.Result@@ -87,6 +87,9 @@              in take 10 (runIdentity $ streamToList $ StreamOptimized.toStreamT $ StreamExcept.foreverE 0 recursive)                   @?= [0, 0, 1, 0, 1, 2, 0, 1, 2, 3]         ]+    , testCase "concatS" $+        let stream = concatS $ unfold 0 (\n -> Result (n + 1) [n, n + 10])+         in take 10 (runIdentity $ streamToList stream) @?= [0, 10, 1, 11, 2, 12, 3, 13, 4, 14]     ]  nats :: (Applicative m) => StreamT m Int