mirror of
https://github.com/graninas/Hydra.git
synced 2024-11-24 12:45:57 +03:00
Merge remote-tracking branch 'origin/master'
This commit is contained in:
Alexander Granin
commit
fe4c001d78
@ -1,29 +1,27 @@
|
||||
{-# LANGUAGE UndecidableInstances #-}
|
||||
{-# LANGUAGE GeneralizedNewtypeDeriving #-}
|
||||
module FTL where
|
||||
|
||||
import qualified Data.Map as Map
|
||||
import qualified Data.Set as Set
|
||||
|
||||
import FTLTypes
|
||||
import qualified Hydra.Domain as D
|
||||
import qualified Hydra.FTL as L
|
||||
import Hydra.Prelude
|
||||
import qualified Hydra.Runtime as R
|
||||
import Types
|
||||
|
||||
import qualified Control.Monad.IO.Unlift as UIO
|
||||
import qualified UnliftIO.Concurrent as UIO
|
||||
|
||||
import Hydra.FTLI ()
|
||||
|
||||
delayFactor :: Int
|
||||
delayFactor = 100
|
||||
|
||||
initState :: AppConfig -> STM AppState'
|
||||
initState :: L.StateL m => AppConfig -> m (AppState' m)
|
||||
initState cfg = do
|
||||
ne <- newTVar Set.empty
|
||||
nw <- newTVar Set.empty
|
||||
se <- newTVar Set.empty
|
||||
sw <- newTVar Set.empty
|
||||
ne <- L.newVar Set.empty
|
||||
nw <- L.newVar Set.empty
|
||||
se <- L.newVar Set.empty
|
||||
sw <- L.newVar Set.empty
|
||||
|
||||
let catalogue = Map.fromList
|
||||
[ (NorthEast, ne)
|
||||
@ -32,9 +30,9 @@ initState cfg = do
|
||||
, (SouthWest, sw)
|
||||
]
|
||||
|
||||
published <- newTVar Set.empty
|
||||
total <- newTVar 0
|
||||
pure $ AppState' catalogue total published cfg
|
||||
publised <- L.newVar Set.empty
|
||||
total <- L.newVar 0
|
||||
pure $ AppState' catalogue total publised cfg
|
||||
|
||||
getRandomMeteor :: L.RandomL m => Region -> m Meteor
|
||||
getRandomMeteor region = do
|
||||
@ -47,64 +45,76 @@ getRandomMilliseconds = L.getRandomInt (0, 3000)
|
||||
|
||||
withRandomDelay
|
||||
:: (L.ControlFlowL m, L.RandomL m)
|
||||
=> AppState' -> m () -> m ()
|
||||
=> AppState' t -> m () -> m ()
|
||||
withRandomDelay st action = do
|
||||
when (delaysEnabled' st)
|
||||
$ getRandomMilliseconds >>= \d -> L.delay $ d * dFactor' st
|
||||
action
|
||||
|
||||
publishMeteor :: AppState' -> Meteor -> STM ()
|
||||
publishMeteor :: L.StateL m => AppState' m -> Meteor -> m ()
|
||||
publishMeteor st meteor =
|
||||
modifyTVar (_channel' st) $ Set.insert meteor
|
||||
L.modifyVar (_channel' st) $ Set.insert meteor
|
||||
|
||||
meteorShower
|
||||
:: (UIO.MonadUnliftIO m, L.LoggerL m, L.RandomL m)
|
||||
=> AppState' -> Region -> m ()
|
||||
:: (Lang m)
|
||||
=> AppState' (L.Transaction m) -> Region -> m ()
|
||||
meteorShower st region = do
|
||||
meteor <- getRandomMeteor region
|
||||
when (doLogDiscovered' st) $ L.logInfo $ "New meteor discovered: " <> show meteor
|
||||
atomically $ publishMeteor st meteor
|
||||
L.transaction $ publishMeteor st meteor
|
||||
|
||||
trackMeteor
|
||||
:: (UIO.MonadUnliftIO m, L.LoggerL m)
|
||||
=> AppState' -> Meteor -> m ()
|
||||
:: (Lang m)
|
||||
=> AppState' (L.Transaction m) -> Meteor -> m ()
|
||||
trackMeteor st meteor = do
|
||||
let region = _region meteor
|
||||
case Map.lookup region (_catalogue' st) of
|
||||
Nothing -> L.logError $ "Region not found: " <> show region
|
||||
Just r -> do
|
||||
when (storeTrackedMeteors' st) $
|
||||
atomically $ modifyTVar r $ Set.insert meteor
|
||||
L.transaction $ L.modifyVar r $ Set.insert meteor
|
||||
when (doLogTracked' st) $ L.logInfo $ "New meteor tracked: " <> show meteor
|
||||
|
||||
meteorCounter :: (UIO.MonadUnliftIO m, L.LoggerL m) => AppState' -> m ()
|
||||
meteorCounter :: (Lang m) => AppState' (L.Transaction m) -> m ()
|
||||
meteorCounter st = do
|
||||
untracked <- atomically $ do
|
||||
ps <- readTVar (_channel' st)
|
||||
when (Set.null ps) retry
|
||||
writeTVar (_channel' st) Set.empty
|
||||
pure $ Set.toList ps
|
||||
untracked <- L.transaction $ do
|
||||
ps <- L.readVar (_channel' st)
|
||||
when (Set.null ps) L.retry
|
||||
L.writeVar (_channel' st) Set.empty
|
||||
pure $ Set.toList ps
|
||||
mapM_ (trackMeteor st) untracked
|
||||
|
||||
atomically $ modifyTVar (_totalMeteors' st) $ (+(length untracked))
|
||||
total <- readTVarIO (_totalMeteors' st)
|
||||
L.transaction $ L.modifyVar (_totalMeteors' st) $ (+(length untracked))
|
||||
total <- L.transaction $ L.readVar (_totalMeteors' st)
|
||||
|
||||
when (doLogTotal' st) $ L.logInfo $ "Total tracked: " <> show total
|
||||
|
||||
meteorsMonitoring :: (UIO.MonadUnliftIO m, L.ControlFlowL m, L.LoggerL m, L.RandomL m) => AppConfig -> m ()
|
||||
meteorsMonitoring cfg = do
|
||||
st <- atomically $ initState cfg
|
||||
meteorsMonitoring :: (Lang m, L.Transaction m ~ t) => AppConfig -> AppState' t -> m ()
|
||||
meteorsMonitoring cfg st = do
|
||||
_ <- L.forkProcess $ forever $ meteorCounter st
|
||||
_ <- L.forkProcess $ forever $ withRandomDelay st $ meteorShower st NorthEast
|
||||
_ <- L.forkProcess $ forever $ withRandomDelay st $ meteorShower st NorthWest
|
||||
_ <- L.forkProcess $ forever $ withRandomDelay st $ meteorShower st SouthEast
|
||||
_ <- L.forkProcess $ forever $ withRandomDelay st $ meteorShower st SouthWest
|
||||
|
||||
UIO.forkIO $ forever $ meteorCounter st
|
||||
UIO.forkIO $ forever $ withRandomDelay st $ meteorShower st NorthEast
|
||||
UIO.forkIO $ forever $ withRandomDelay st $ meteorShower st NorthWest
|
||||
UIO.forkIO $ forever $ withRandomDelay st $ meteorShower st SouthEast
|
||||
UIO.forkIO $ forever $ withRandomDelay st $ meteorShower st SouthWest
|
||||
|
||||
atomically $ do
|
||||
L.transaction $ do
|
||||
let maxTotal = fromMaybe 0 $ maxMeteors cfg
|
||||
total <- readTVar $ _totalMeteors' st
|
||||
when (maxTotal == 0 || total < maxTotal) retry
|
||||
total <- L.readVar $ _totalMeteors' st
|
||||
when (maxTotal == 0 || total < maxTotal) L.retry
|
||||
|
||||
scenario :: R.CoreRuntime -> AppConfig -> IO ()
|
||||
scenario coreRt cfg = void $ runReaderT (meteorsMonitoring cfg) coreRt
|
||||
scenario coreRt cfg = void $ do
|
||||
st <- atomically $ initState cfg
|
||||
runReaderT (runAppM $ meteorsMonitoring cfg st) coreRt
|
||||
|
||||
newtype AppM a = AppM { runAppM :: ReaderT R.CoreRuntime IO a }
|
||||
deriving (Functor, Applicative, Monad, L.ControlFlowL, L.LoggerL, L.RandomL, L.ProcessL)
|
||||
|
||||
class (L.StateL (L.Transaction m), L.Atomic m,
|
||||
L.StateL (L.Transaction m), L.LoggerL m, L.RandomL m, L.ControlFlowL m, L.ProcessL m) => Lang m
|
||||
instance Lang AppM
|
||||
|
||||
instance L.Atomic AppM where
|
||||
type Transaction AppM = STM
|
||||
transaction = AppM . atomically
|
||||
|
||||
|
||||
@ -3,28 +3,28 @@ module FTLTypes where
|
||||
import qualified Data.Map as Map
|
||||
import qualified Data.Set as Set
|
||||
|
||||
import qualified Hydra.Domain as D
|
||||
import Hydra.Prelude
|
||||
import qualified Hydra.Runtime as R
|
||||
import Hydra.FTL as L
|
||||
import Types
|
||||
|
||||
type Meteors' = TVar (Set.Set Meteor)
|
||||
type Meteors' m = L.StateVar m (Set.Set Meteor)
|
||||
|
||||
type Catalogue' = Map.Map Region Meteors'
|
||||
type Catalogue' m = Map.Map Region (Meteors' m)
|
||||
|
||||
data AppState' = AppState'
|
||||
{ _catalogue' :: Catalogue'
|
||||
, _totalMeteors' :: TVar Int
|
||||
, _channel' :: TVar (Set.Set Meteor)
|
||||
data AppState' m = AppState'
|
||||
{ _catalogue' :: Catalogue' m
|
||||
, _totalMeteors' :: L.StateVar m Int
|
||||
, _channel' :: L.StateVar m (Set.Set Meteor)
|
||||
, _config' :: AppConfig
|
||||
}
|
||||
|
||||
delaysEnabled' :: AppState' -> Bool
|
||||
|
||||
delaysEnabled' :: AppState' m -> Bool
|
||||
delaysEnabled' = enableDelays . _config'
|
||||
|
||||
dFactor' = delaysFactor . _config'
|
||||
|
||||
storeTrackedMeteors' :: AppState' -> Bool
|
||||
storeTrackedMeteors' :: AppState' m -> Bool
|
||||
storeTrackedMeteors' = storeTracked . _config'
|
||||
|
||||
doLogDiscovered' = logDiscovered . _config'
|
||||
|
||||
@ -4,3 +4,11 @@ import Hydra.Prelude
|
||||
|
||||
class Monad m => ControlFlowL m where
|
||||
delay :: Int -> m ()
|
||||
|
||||
instance ControlFlowL IO where
|
||||
delay = threadDelay
|
||||
{-# INLINE delay #-}
|
||||
|
||||
instance ControlFlowL (ReaderT e IO) where
|
||||
delay = lift . threadDelay
|
||||
{-# INLINE delay #-}
|
||||
|
||||
@ -1,10 +0,0 @@
|
||||
module Hydra.Core.ControlFlow.FTLI where
|
||||
|
||||
import Hydra.Prelude
|
||||
|
||||
import qualified Hydra.Core.FTL as L
|
||||
import qualified Hydra.Core.RLens as RLens
|
||||
import qualified Hydra.Core.Runtime as R
|
||||
|
||||
instance MonadIO m => L.ControlFlowL (ReaderT R.CoreRuntime m) where
|
||||
delay = liftIO . threadDelay
|
||||
@ -7,5 +7,5 @@ import Hydra.Core.Lang.FTL as X
|
||||
import Hydra.Core.Logger.FTL as X
|
||||
import Hydra.Core.Process.FTL as X
|
||||
import Hydra.Core.Random.FTL as X
|
||||
-- import Hydra.Core.State.FTL as X
|
||||
import Hydra.Core.State.Language as X
|
||||
import Hydra.Core.State.FTL as X
|
||||
-- import Hydra.Core.State.Language as X
|
||||
|
||||
@ -2,9 +2,7 @@ module Hydra.Core.FTLI
|
||||
( module X
|
||||
) where
|
||||
|
||||
import Hydra.Core.ControlFlow.FTLI as X
|
||||
import Hydra.Core.Lang.FTLI as X
|
||||
import Hydra.Core.Logger.FTLI as X
|
||||
import Hydra.Core.Process.FTLI as X
|
||||
import Hydra.Core.Random.FTLI as X
|
||||
import Hydra.Core.State.FTLI as X
|
||||
|
||||
@ -13,7 +13,7 @@ import Hydra.Core.Random.FTL as L
|
||||
import qualified Hydra.Core.State.Class as L
|
||||
import qualified Hydra.Core.State.Language as L
|
||||
|
||||
class (Monad m, L.ControlFlowL m, L.RandomL m, L.LoggerL m) => LangL m where
|
||||
class (Monad m, L.RandomL m, L.LoggerL m) => LangL m where
|
||||
evalStateAtomically :: L.StateL a -> m a
|
||||
|
||||
instance (Monad m, LangL m) => L.StateIO m where
|
||||
|
||||
@ -2,7 +2,6 @@ module Hydra.Core.Lang.FTLI where
|
||||
|
||||
import Hydra.Prelude
|
||||
|
||||
import Hydra.Core.ControlFlow.FTLI ()
|
||||
import qualified Hydra.Core.FTL as L
|
||||
import Hydra.Core.Logger.FTLI ()
|
||||
import Hydra.Core.Random.FTLI ()
|
||||
|
||||
@ -1,3 +1,4 @@
|
||||
{-# LANGUAGE FunctionalDependencies #-}
|
||||
{-# LANGUAGE FlexibleContexts #-}
|
||||
{-# LANGUAGE GADTs #-}
|
||||
{-# LANGUAGE MultiParamTypeClasses #-}
|
||||
@ -5,14 +6,35 @@
|
||||
|
||||
module Hydra.Core.Process.FTL where
|
||||
|
||||
import Hydra.Prelude
|
||||
import Hydra.Prelude hiding (atomically)
|
||||
|
||||
import qualified Hydra.Core.Domain as D
|
||||
import Control.Concurrent
|
||||
import Control.Concurrent.STM
|
||||
|
||||
class (Monad m) => ProcessL m where
|
||||
type ProcessHandle m :: * -> *
|
||||
forkProcess :: m a -> m (ProcessHandle m a)
|
||||
killProcess :: ProcessHandle m a -> m ()
|
||||
tryGetResult :: ProcessHandle m a -> m (Maybe a)
|
||||
awaitResult :: ProcessHandle m a -> m a
|
||||
|
||||
newtype H a = H { unH :: (ThreadId, TMVar (Either SomeException a)) }
|
||||
|
||||
-- class (Monad m) => ProcessL m where
|
||||
-- forkProcess :: m' a -> m (D.ProcessPtr a)
|
||||
-- killProcess :: D.ProcessPtr a -> m ()
|
||||
-- tryGetResult :: D.ProcessPtr a -> m (Maybe a)
|
||||
-- awaitResult :: D.ProcessPtr a -> m a
|
||||
instance ProcessL (ReaderT e IO) where
|
||||
type ProcessHandle (ReaderT e IO) = H
|
||||
forkProcess f = ReaderT $ \e -> do
|
||||
z <- newEmptyTMVarIO
|
||||
t <- forkIOWithUnmask $ \restore -> do
|
||||
x <- (restore $ runReaderT f e) `catch` (\e -> do
|
||||
atomically $ putTMVar z (Left e)
|
||||
throwM e)
|
||||
atomically $ putTMVar z (Right x)
|
||||
pure $ H (t,z)
|
||||
killProcess = ReaderT . const . killThread . fst . unH
|
||||
tryGetResult (H (_,e)) = ReaderT $ const $ atomically (tryReadTMVar e) >>=
|
||||
traverse (\case
|
||||
Left e -> throwM e
|
||||
Right x -> pure x)
|
||||
awaitResult (H (_,e)) = ReaderT $ const $ atomically (readTMVar e) >>= \case
|
||||
Left e -> throwM e
|
||||
Right x -> pure x
|
||||
|
||||
@ -1,30 +1,45 @@
|
||||
{-# LANGUAGE AllowAmbiguousTypes #-}
|
||||
{-# LANGUAGE FunctionalDependencies #-}
|
||||
{-# LANGUAGE ScopedTypeVariables #-}
|
||||
module Hydra.Core.State.FTL where
|
||||
|
||||
import Hydra.Prelude
|
||||
import Control.Concurrent.STM as STM
|
||||
|
||||
import qualified Hydra.Core.Domain as D
|
||||
-- import qualified Hydra.Core.Domain as D
|
||||
|
||||
-- class Monad m => StateL m where
|
||||
-- newVar :: a -> m (D.StateVar a)
|
||||
-- readVar :: D.StateVar a -> m a
|
||||
-- writeVar :: D.StateVar a -> a -> m ()
|
||||
-- retry :: m a
|
||||
class Monad m => StateL m where
|
||||
type StateVar m :: * -> *
|
||||
newVar :: a -> m (StateVar m a)
|
||||
readVar :: StateVar m a -> m a
|
||||
writeVar :: StateVar m a -> a -> m ()
|
||||
retry :: m a
|
||||
|
||||
-- -- | Modify variable with function.
|
||||
-- modifyVar :: StateL m => D.StateVar a -> (a -> a) -> m ()
|
||||
-- modifyVar var f = readVar var >>= writeVar var . f
|
||||
modifyVar :: StateL m => StateVar m a -> (a -> a) -> m ()
|
||||
modifyVar var f = readVar var >>= writeVar var . f
|
||||
{-# SPECIALIZE modifyVar :: TVar a -> (a -> a) -> STM () #-}
|
||||
|
||||
instance StateL STM where
|
||||
type StateVar STM = TVar
|
||||
newVar = newTVar
|
||||
{-# INLINE newVar #-}
|
||||
readVar = readTVar
|
||||
{-# INLINE readVar #-}
|
||||
writeVar = writeTVar
|
||||
{-# INLINE writeVar #-}
|
||||
retry = STM.retry
|
||||
{-# INLINE retry #-}
|
||||
|
||||
-- class StateIO m where
|
||||
-- atomically :: StateL a -> m a
|
||||
-- newVarIO :: a -> m (D.StateVar a)
|
||||
-- readVarIO :: D.StateVar a -> m a
|
||||
-- writeVarIO :: D.StateVar a -> a -> m ()
|
||||
--
|
||||
-- | Class that defines how can we run internal nested transaction in the
|
||||
-- current computation.
|
||||
class Atomic m where
|
||||
type Transaction m :: * -> *
|
||||
transaction :: (Transaction m) a -> m a
|
||||
|
||||
-- -- -- | Eval "delayed" logger: it will be written after successfull state operation.
|
||||
-- -- evalStmLogger :: L.LoggerL () -> StateL ()
|
||||
-- -- evalStmLogger action = liftF $ EvalStmLogger action id
|
||||
--
|
||||
-- -- instance L.Logger StateL where
|
||||
-- -- logMessage level = evalStmLogger . L.logMessage level
|
||||
-- -- logMessage level = evalStmLogger . L.logMessage level
|
||||
|
||||
@ -1,23 +0,0 @@
|
||||
module Hydra.Core.State.FTLI where
|
||||
|
||||
import Hydra.Prelude
|
||||
|
||||
import qualified Hydra.Core.Domain as D
|
||||
import qualified Hydra.Core.FTL as L
|
||||
import qualified Hydra.Core.RLens as RLens
|
||||
import qualified Hydra.Core.Runtime as R
|
||||
|
||||
import qualified Hydra.Core.State.Interpreter as Impl
|
||||
--
|
||||
-- instance L.StateL (ReaderT R.CoreRuntime STM) where
|
||||
-- newVar val = do
|
||||
-- coreRt <- ask
|
||||
-- r <- lift $ Impl.newVar' (coreRt ^. RLens.stateRuntime) val
|
||||
-- pure $ D.StateVar r
|
||||
-- readVar var = do
|
||||
-- coreRt <- ask
|
||||
-- lift $ Impl.readVar' (coreRt ^. RLens.stateRuntime) var
|
||||
-- writeVar var val = do
|
||||
-- coreRt <- ask
|
||||
-- lift $ Impl.writeVar' (coreRt ^. RLens.stateRuntime) var val
|
||||
-- retry = lift retry
|
||||
Loading…
Reference in New Issue
Block a user