{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE PatternSynonyms #-}
module Free.Agent.Bus
(
Bus,
busLogPath,
openBus,
closeBus,
withBus,
scribe,
scribeIO,
postLocal,
appendStoredPosts,
appendStoredPostsUnlocked,
readSince,
awaitSince,
runSeatBus,
)
where
import Circuit (close, companion, conjoint)
import Circuit.Agent (Name, Post (..), PostId, deliversTo, mkPost, sortNub)
import Circuit.Agent.Framing
( Log (..),
PostBody,
Snoc (..),
Stamped,
Uncons (..),
encodeLog,
frameStored,
readLogFile,
stamp,
stamped,
pattern Stamped,
)
import Circuit.Agent.Mark (Mark (..), isEscalate, isHalt, markGlyph, markOf)
import Circuit.Agent.Tensor (closeShardIO)
import Control.Concurrent (ThreadId, forkIO, killThread)
import Control.Concurrent.STM
( STM,
TMVar,
TQueue,
TVar,
atomically,
isEmptyTQueue,
newEmptyTMVar,
newTQueueIO,
newTVarIO,
putTMVar,
readTQueue,
readTVar,
retry,
takeTMVar,
writeTQueue,
writeTVar,
)
import Control.Exception (SomeException, bracket, displayException, try)
import Control.Monad (forever, unless)
import Data.ByteString qualified as BS
import Data.Foldable (traverse_)
import Data.List (maximum)
import Data.Text (Text)
import Data.Text qualified as T
import Data.Text.IO qualified as TIO
import Data.Time (UTCTime, getCurrentTime)
import Free.Agent.Seat (FreeSeat, interpretSeat)
import System.Directory (createDirectoryIfMissing, doesFileExist)
import System.FileLock (SharedExclusive (Exclusive), withFileLock)
import System.FilePath (takeDirectory, (<.>), (</>))
import System.IO (IOMode (AppendMode), withFile)
import Text.Printf (printf)
data Bus a = Bus
{
forall a. Bus a -> TVar (Log a)
busLog :: TVar (Log a),
forall a. Bus a -> TQueue (Stamped a, TMVar ())
busPending :: TQueue (Stamped a, TMVar ()),
forall a. Bus a -> FilePath
busPath :: FilePath,
forall a. Bus a -> ThreadId
busThread :: ThreadId
}
busLogPath :: Bus a -> FilePath
busLogPath :: forall a. Bus a -> FilePath
busLogPath = Bus a -> FilePath
forall a. Bus a -> FilePath
busPath
openBus :: (PostBody a) => FilePath -> IO (Bus a)
openBus :: forall a. PostBody a => FilePath -> IO (Bus a)
openBus FilePath
root = do
Bool -> FilePath -> IO ()
createDirectoryIfMissing Bool
True FilePath
root
let path :: FilePath
path = FilePath
root FilePath -> FilePath -> FilePath
</> FilePath
"log.jsonl"
exists <- FilePath -> IO Bool
doesFileExist FilePath
path
unless exists $ do
withFile path AppendMode (\Handle
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())
initial <- if exists then readLogFile path else pure (Log [])
tv <- newTVarIO initial
q <- newTQueueIO
tid <- forkIO (persistLoop path q)
pure (Bus tv q path tid)
closeBus :: Bus a -> IO ()
closeBus :: forall a. Bus a -> IO ()
closeBus = ThreadId -> IO ()
killThread (ThreadId -> IO ()) -> (Bus a -> ThreadId) -> Bus a -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Bus a -> ThreadId
forall a. Bus a -> ThreadId
busThread
withBus :: (PostBody a) => FilePath -> (Bus a -> IO b) -> IO b
withBus :: forall a b. PostBody a => FilePath -> (Bus a -> IO b) -> IO b
withBus FilePath
root = IO (Bus a) -> (Bus a -> IO ()) -> (Bus a -> IO b) -> IO b
forall a b c. IO a -> (a -> IO b) -> (a -> IO c) -> IO c
bracket (FilePath -> IO (Bus a)
forall a. PostBody a => FilePath -> IO (Bus a)
openBus FilePath
root) Bus a -> IO ()
forall a. Bus a -> IO ()
closeBus
scribe :: Bus a -> UTCTime -> Post a -> STM (Stamped a, TMVar ())
scribe :: forall a. Bus a -> UTCTime -> Post a -> STM (Stamped a, TMVar ())
scribe Bus a
bus UTCTime
ts Post a
p = do
log0 <- TVar (Log a) -> STM (Log a)
forall a. TVar a -> STM a
readTVar (Bus a -> TVar (Log a)
forall a. Bus a -> TVar (Log a)
busLog Bus a
bus)
let pid = Int -> PostId
forall a b. (Integral a, Num b) => a -> b
fromIntegral ([Stamped a] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length (Log a -> [Stamped a]
forall a. Log a -> [Stamped a]
unLog Log a
log0))
stored = (UTCTime, PostId) -> Post a -> Stamped a
forall a. (UTCTime, PostId) -> Post a -> Stamped a
Stamped (UTCTime
ts, PostId
pid) Post a
p
ack <- newEmptyTMVar
writeTVar (busLog bus) (snoc log0 stored)
writeTQueue (busPending bus) (stored, ack)
pure (stored, ack)
scribeIO :: Bus a -> Post a -> IO (Stamped a)
scribeIO :: forall a. Bus a -> Post a -> IO (Stamped a)
scribeIO Bus a
bus Post a
p = do
ts <- IO UTCTime
getCurrentTime
(stored, ack) <- atomically (scribe bus ts p)
atomically (takeTMVar ack)
pure stored
postLocal :: (PostBody a) => FilePath -> Post a -> IO (Stamped a)
postLocal :: forall a. PostBody a => FilePath -> Post a -> IO (Stamped a)
postLocal FilePath
root Post a
p = do
Bool -> FilePath -> IO ()
createDirectoryIfMissing Bool
True FilePath
root
let path :: FilePath
path = FilePath
root FilePath -> FilePath -> FilePath
</> FilePath
"log.jsonl"
exists <- FilePath -> IO Bool
doesFileExist FilePath
path
unless exists $ withFile path AppendMode (\Handle
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())
ts <- getCurrentTime
stored <- withFileLock (path <.> "lock") Exclusive $ \FileLock
_lock -> do
n <- Word8 -> ByteString -> Int
BS.count Word8
0x0A (ByteString -> Int) -> IO ByteString -> IO Int
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> FilePath -> IO ByteString
BS.readFile FilePath
path
let stored = (UTCTime, PostId) -> Post a -> Stamped a
forall a. (UTCTime, PostId) -> Post a -> Stamped a
Stamped (UTCTime
ts, Int -> PostId
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
n) Post a
p
appendStoredPostsUnlocked path [stored]
pure stored
writePings path [stored]
pure stored
appendStoredPosts :: (PostBody a) => FilePath -> [Stamped a] -> IO ()
appendStoredPosts :: forall a. PostBody a => FilePath -> [Stamped a] -> IO ()
appendStoredPosts FilePath
path [Stamped a]
posts =
FilePath -> SharedExclusive -> (FileLock -> IO ()) -> IO ()
forall a. FilePath -> SharedExclusive -> (FileLock -> IO a) -> IO a
withFileLock (FilePath
path FilePath -> FilePath -> FilePath
<.> FilePath
"lock") SharedExclusive
Exclusive ((FileLock -> IO ()) -> IO ()) -> (FileLock -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \FileLock
_lock ->
FilePath -> [Stamped a] -> IO ()
forall a. PostBody a => FilePath -> [Stamped a] -> IO ()
appendStoredPostsUnlocked FilePath
path [Stamped a]
posts
appendStoredPostsUnlocked :: (PostBody a) => FilePath -> [Stamped a] -> IO ()
appendStoredPostsUnlocked :: forall a. PostBody a => FilePath -> [Stamped a] -> IO ()
appendStoredPostsUnlocked FilePath
path [Stamped a]
posts =
FilePath -> IOMode -> (Handle -> IO ()) -> IO ()
forall r. FilePath -> IOMode -> (Handle -> IO r) -> IO r
withFile FilePath
path IOMode
AppendMode ((Handle -> IO ()) -> IO ()) -> (Handle -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Handle
h ->
(Stamped a -> IO ()) -> [Stamped a] -> IO ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (Handle -> Text -> IO ()
TIO.hPutStrLn Handle
h (Text -> IO ()) -> (Stamped a -> Text) -> Stamped a -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Stamped a -> Text
forall a. PostBody a => Stamped a -> Text
frameStored) [Stamped a]
posts
readSince :: Bus a -> [Name] -> PostId -> STM [Stamped a]
readSince :: forall a. Bus a -> [Text] -> PostId -> STM [Stamped a]
readSince Bus a
bus [Text]
names PostId
since = do
log0 <- TVar (Log a) -> STM (Log a)
forall a. TVar a -> STM a
readTVar (Bus a -> TVar (Log a)
forall a. Bus a -> TVar (Log a)
busLog Bus a
bus)
let posts = Log a -> [Stamped a]
forall a. Log a -> [Stamped a]
unLog Log a
log0
pure [s | s <- posts, snd (stamp s) >= since, deliversTo (stamped s) names]
awaitSince :: Bus a -> [Name] -> PostId -> STM [Stamped a]
awaitSince :: forall a. Bus a -> [Text] -> PostId -> STM [Stamped a]
awaitSince Bus a
bus [Text]
names PostId
since = do
found <- Bus a -> [Text] -> PostId -> STM [Stamped a]
forall a. Bus a -> [Text] -> PostId -> STM [Stamped a]
readSince Bus a
bus [Text]
names PostId
since
if null found then retry else pure found
runSeatBus :: Bus Text -> Name -> [Name] -> FreeSeat -> IO ()
runSeatBus :: Bus Text -> Text -> [Text] -> FreeSeat -> IO ()
runSeatBus Bus Text
bus Text
agentName [Text]
names FreeSeat
seat = PostId -> IO ()
loop PostId
0
where
sh :: AgentShard [Post Text] [Post Text]
sh = FreeSeat -> AgentShard [Post Text] [Post Text]
interpretSeat FreeSeat
seat
loop :: PostId -> IO ()
loop PostId
lastId = do
posts <- STM [Stamped Text] -> IO [Stamped Text]
forall a. STM a -> IO a
atomically (Bus Text -> [Text] -> PostId -> STM [Stamped Text]
forall a. Bus a -> [Text] -> PostId -> STM [Stamped a]
awaitSince Bus Text
bus [Text]
names PostId
lastId)
let marked = (Stamped Text -> Bool) -> [Stamped Text] -> Bool
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Bool
any (Post Text -> Bool
halts (Post Text -> Bool)
-> (Stamped Text -> Post Text) -> Stamped Text -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Stamped Text -> Post Text
forall r a. Stamped r a -> a
stamped) [Stamped Text]
posts
work = (Stamped Text -> Bool) -> [Stamped Text] -> [Stamped Text]
forall a. (a -> Bool) -> [a] -> [a]
filter (Bool -> Bool
not (Bool -> Bool) -> (Stamped Text -> Bool) -> Stamped Text -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Post Text -> Bool
halts (Post Text -> Bool)
-> (Stamped Text -> Post Text) -> Stamped Text -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Stamped Text -> Post Text
forall r a. Stamped r a -> a
stamped) [Stamped Text]
posts
selfHalt <- go work
unless (marked || selfHalt) $
loop (maximum (map (snd . stamp) posts) + 1)
go :: [Stamped Text] -> IO Bool
go [] = Bool -> IO Bool
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
False
go (Stamped Text
stored : [Stamped Text]
rest) = do
outs <- Stamped Text -> IO [Post Text]
processOne Stamped Text
stored
let keep Post Text
p =
let b :: Text
b = Text -> Text
T.strip (Post Text -> Text
forall a. Post a -> a
body Post Text
p)
in Bool -> Bool
not (Text -> Bool
T.null Text
b) Bool -> Bool -> Bool
&& Text
b Text -> Text -> Bool
forall a. Eq a => a -> a -> Bool
/= Text
"(empty)"
nonEmpty = (Post Text -> Bool) -> [Post Text] -> [Post Text]
forall a. (a -> Bool) -> [a] -> [a]
filter Post Text -> Bool
keep [Post Text]
outs
traverse_ (scribeIO bus) nonEmpty
if any ((== Just StandDown) . markOf) outs
then pure True
else go rest
halts :: Post Text -> Bool
halts Post Text
p = Bool -> (Mark -> Bool) -> Maybe Mark -> Bool
forall b a. b -> (a -> b) -> Maybe a -> b
maybe Bool
False (\Mark
m -> Mark -> Bool
isHalt Mark
m Bool -> Bool -> Bool
|| Mark -> Bool
isEscalate Mark
m) (Post Text -> Maybe Mark
markOf Post Text
p)
processOne :: Stamped Text -> IO [Post Text]
processOne Stamped Text
stored = do
er <-
forall e a. Exception e => IO a -> IO (Either e a)
try @SomeException (IO [Post Text] -> IO (Either SomeException [Post Text]))
-> IO [Post Text] -> IO (Either SomeException [Post Text])
forall a b. (a -> b) -> a -> b
$ do
let p :: Post Text
p = Stamped Text -> Post Text
forall r a. Stamped r a -> a
stamped Stamped Text
stored
parentId :: PostId
parentId = (UTCTime, PostId) -> PostId
forall a b. (a, b) -> b
snd (Stamped Text -> (UTCTime, PostId)
forall r a. Stamped r a -> r
stamp Stamped Text
stored)
(outs, _st) <- AgentShard [Post Text] [Post Text]
-> [Post Text] -> [Post Text] -> IO ([Post Text], [Post Text])
forall s a. Poles (Body (,) s (K IO)) a a -> a -> s -> IO (a, s)
closeShardIO AgentShard [Post Text] [Post Text]
sh [Post Text
p] []
pure [out {thread = sortNub (parentId : thread out)} | out <- outs]
case er of
Left SomeException
e -> do
let p :: Post Text
p = Stamped Text -> Post Text
forall r a. Stamped r a -> a
stamped Stamped Text
stored
exc :: Text
exc = FilePath -> Text
T.pack (SomeException -> FilePath
forall e. Exception e => e -> FilePath
displayException SomeException
e)
msg :: Text
msg = Mark -> Text
markGlyph Mark
Escalate Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" handler failed: " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
exc
_ <- Bus Text -> Post Text -> IO (Stamped Text)
forall a. Bus a -> Post a -> IO (Stamped a)
scribeIO Bus Text
bus (Text -> [Text] -> Text -> Post Text
forall a. Text -> [Text] -> a -> Post a
mkPost Text
agentName [Post Text -> Text
forall a. Post a -> Text
from Post Text
p] Text
msg)
pure []
Right [Post Text]
outs -> [Post Text] -> IO [Post Text]
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure [Post Text]
outs
persistLoop :: (PostBody a) => FilePath -> TQueue (Stamped a, TMVar ()) -> IO ()
persistLoop :: forall a.
PostBody a =>
FilePath -> TQueue (Stamped a, TMVar ()) -> IO ()
persistLoop FilePath
path TQueue (Stamped a, TMVar ())
q = IO () -> IO ()
forall (f :: * -> *) a b. Applicative f => f a -> f b
forever (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
pairs <- STM [(Stamped a, TMVar ())] -> IO [(Stamped a, TMVar ())]
forall a. STM a -> IO a
atomically (STM [(Stamped a, TMVar ())] -> IO [(Stamped a, TMVar ())])
-> STM [(Stamped a, TMVar ())] -> IO [(Stamped a, TMVar ())]
forall a b. (a -> b) -> a -> b
$ do
first <- TQueue (Stamped a, TMVar ()) -> STM (Stamped a, TMVar ())
forall a. TQueue a -> STM a
readTQueue TQueue (Stamped a, TMVar ())
q
rest <- drainQueue
pure (first : rest)
let posts = ((Stamped a, TMVar ()) -> Stamped a)
-> [(Stamped a, TMVar ())] -> [Stamped a]
forall a b. (a -> b) -> [a] -> [b]
map (Stamped a, TMVar ()) -> Stamped a
forall a b. (a, b) -> a
fst [(Stamped a, TMVar ())]
pairs
appendStoredPosts path posts
writePings path posts
atomically $ traverse_ (\(Stamped a
_, TMVar ()
ack) -> TMVar () -> () -> STM ()
forall a. TMVar a -> a -> STM ()
putTMVar TMVar ()
ack ()) pairs
where
drainQueue :: STM [(Stamped a, TMVar ())]
drainQueue = do
empty <- TQueue (Stamped a, TMVar ()) -> STM Bool
forall a. TQueue a -> STM Bool
isEmptyTQueue TQueue (Stamped a, TMVar ())
q
if empty
then pure []
else do
x <- readTQueue q
xs <- drainQueue
pure (x : xs)
writePings :: FilePath -> [Stamped a] -> IO ()
writePings :: forall a. FilePath -> [Stamped a] -> IO ()
writePings FilePath
path [Stamped a]
posts = ((Text, PostId) -> IO ()) -> [(Text, PostId)] -> IO ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (Text, PostId) -> IO ()
writeOne [(Text, PostId)]
recipients
where
root :: FilePath
root = FilePath -> FilePath
takeDirectory FilePath
path
pingFile :: Text -> FilePath
pingFile Text
name = FilePath
root FilePath -> FilePath -> FilePath
</> FilePath -> FilePath -> FilePath
forall r. PrintfType r => FilePath -> r
printf FilePath
".ping-%s" (Text -> FilePath
T.unpack Text
name)
recipients :: [(Text, PostId)]
recipients = (Stamped a -> [(Text, PostId)]) -> [Stamped a] -> [(Text, PostId)]
forall (t :: * -> *) a b. Foldable t => (a -> [b]) -> t a -> [b]
concatMap Stamped a -> [(Text, PostId)]
forall {a} {b} {a}. Stamped (a, b) (Post a) -> [(Text, b)]
postRecipients [Stamped a]
posts
postRecipients :: Stamped (a, b) (Post a) -> [(Text, b)]
postRecipients Stamped (a, b) (Post a)
stored =
case Post a -> [Text]
forall a. Post a -> [Text]
to (Stamped (a, b) (Post a) -> Post a
forall r a. Stamped r a -> a
stamped Stamped (a, b) (Post a)
stored) of
[] -> []
[Text
""] -> []
[Text]
ts -> [(Text
name, (a, b) -> b
forall a b. (a, b) -> b
snd (Stamped (a, b) (Post a) -> (a, b)
forall r a. Stamped r a -> r
stamp Stamped (a, b) (Post a)
stored)) | Text
name <- [Text]
ts]
writeOne :: (Text, PostId) -> IO ()
writeOne (Text
name, PostId
pid) =
FilePath -> Text -> IO ()
TIO.writeFile (Text -> FilePath
pingFile Text
name) (FilePath -> Text
T.pack (PostId -> FilePath
forall a. Show a => a -> FilePath
show PostId
pid))