1
0
mirror of https://github.com/ilyakooo0/urbit.git synced 2024-12-27 08:52:10 +03:00
urbit/pkg/king/lib/Vere/Serf.hs

477 lines
14 KiB
Haskell
Raw Normal View History

{-# OPTIONS_GHC -Wwarn #-}
2019-07-17 02:14:46 +03:00
{-
2019-07-17 02:14:46 +03:00
- TODO: `recvLen` is not big-endian safe.
-}
2019-09-04 01:17:20 +03:00
module Vere.Serf ( Serf, sStderr, SerfState(..), doJob
, run, shutdown, kill
, replay, bootFromSeq, snapshot
, collectFX
2019-07-22 00:24:07 +03:00
, Config(..), Flags, Flag(..)
) where
import UrbitPrelude hiding (fail)
import Arvo
import Control.Monad.Fail (fail)
import Data.Conduit
2019-07-12 04:16:40 +03:00
import Data.Void
2019-07-02 05:51:26 +03:00
import Noun
import System.Process
2019-07-12 04:16:40 +03:00
import Vere.Pier.Types
2019-07-22 00:24:07 +03:00
import Data.Bits (setBit)
2019-07-12 04:16:40 +03:00
import Data.ByteString (hGet)
import Data.ByteString.Unsafe (unsafeUseAsCString)
2019-07-12 04:16:40 +03:00
import Foreign.Marshal.Alloc (alloca)
import Foreign.Ptr (castPtr)
2019-07-12 22:24:44 +03:00
import Foreign.Storable (peek, poke)
import System.Directory (createDirectoryIfMissing)
2019-07-12 04:16:40 +03:00
import System.Exit (ExitCode)
import qualified Data.ByteString.Unsafe as BS
2019-07-17 02:14:46 +03:00
import qualified Data.Text as T
2019-07-21 23:30:30 +03:00
import qualified System.IO.Error as IO
import qualified System.IO as IO
import qualified Urbit.Time as Time
import qualified Vere.Log as Log
2019-07-22 00:24:07 +03:00
-- Serf Config -----------------------------------------------------------------
type Flags = [Flag]
data Flag
= DebugRam
| DebugCpu
| CheckCorrupt
| CheckFatal
| Verbose
| DryRun
| Quiet
| Hashless
| Trace
deriving (Eq, Ord, Show, Enum, Bounded)
compileFlags :: [Flag] -> Word
compileFlags = foldl' (\acc flag -> setBit acc (fromEnum flag)) 0
data Config = Config FilePath [Flag]
deriving (Show)
2019-08-28 15:22:56 +03:00
serf :: HasLogFunc e => Text -> RIO e ()
serf msg = logInfo $ display ("SERF: " <> msg)
2019-07-22 00:24:07 +03:00
2019-07-17 02:14:46 +03:00
-- Types -----------------------------------------------------------------------
data SerfState = SerfState
{ ssNextEv :: EventId
, ssLastMug :: Mug
}
deriving (Eq, Ord, Show)
2019-08-15 05:42:48 +03:00
ssLastEv :: SerfState -> EventId
ssLastEv = pred . ssNextEv
2019-09-04 01:17:20 +03:00
data Serf e = Serf
{ sendHandle :: Handle
, recvHandle :: Handle
2019-07-21 23:30:30 +03:00
, errThread :: Async ()
, process :: ProcessHandle
, sState :: MVar SerfState
2019-09-04 01:17:20 +03:00
, sStderr :: Text -> RIO e ()
}
data ShipId = ShipId Ship Bool
deriving (Eq, Ord, Show)
type Play = Maybe (EventId, Mug, ShipId)
data Plea
= PPlay Play
| PWork Work
| PDone EventId Mug FX
| PStdr EventId Cord
| PSlog EventId Word32 Tank
2019-06-01 03:21:44 +03:00
deriving (Eq, Show)
type ReplacementEv = Job
type WorkResult = (SerfState, FX)
type SerfResp = Either ReplacementEv WorkResult
2019-06-01 03:21:44 +03:00
data SerfExn
2019-06-01 03:21:44 +03:00
= BadComputeId EventId WorkResult
| BadReplacementId EventId ReplacementEv
| UnexpectedPlay EventId Play
| BadPleaAtom Atom
| BadPleaNoun Noun [Text] Text
| ReplacedEventDuringReplay EventId ReplacementEv
2019-07-16 03:01:45 +03:00
| ReplacedEventDuringBoot EventId ReplacementEv
| EffectsDuringBoot EventId FX
| SerfConnectionClosed
| UnexpectedPleaOnNewShip Plea
| InvalidInitialPlea Plea
2019-06-01 03:21:44 +03:00
deriving (Show)
2019-07-17 02:14:46 +03:00
-- Instances -------------------------------------------------------------------
instance Exception SerfExn
2019-06-01 03:21:44 +03:00
deriveNoun ''ShipId
2019-07-17 02:14:46 +03:00
deriveNoun ''Plea
2019-06-01 03:21:44 +03:00
-- Utils -----------------------------------------------------------------------
2019-09-04 01:17:20 +03:00
printTank :: HasLogFunc e => (Text -> RIO e ()) -> Word32 -> Tank -> RIO e ()
printTank log _pri tank =
(log . unlines . fmap unTape . wash (WashCfg 0 80)) tank
2019-08-28 15:22:56 +03:00
guardExn :: (Exception e, MonadIO m) => Bool -> e -> m ()
guardExn ok = io . unless ok . throwIO
2019-06-01 03:21:44 +03:00
2019-08-28 15:22:56 +03:00
fromRightExn :: (Exception e, MonadIO m) => Either a b -> (a -> e) -> m b
fromRightExn (Left m) exn = throwIO (exn m)
fromRightExn (Right x) _ = pure x
2019-07-17 02:14:46 +03:00
-- Process Management ----------------------------------------------------------
2019-09-04 01:17:20 +03:00
run :: HasLogFunc e => Config -> RAcquire e (Serf e)
2019-08-28 15:22:56 +03:00
run config = mkRAcquire (startUp config) tearDown
2019-09-04 01:17:20 +03:00
startUp :: HasLogFunc e => Config -> RIO e (Serf e)
startUp conf@(Config pierPath flags) = do
2019-08-28 15:22:56 +03:00
logTrace "STARTING SERF"
logTrace (displayShow conf)
(i, o, e, p) <- io $ do
(Just i, Just o, Just e, p) <- createProcess pSpec
pure (i, o, e, p)
ss <- newEmptyMVar
2019-07-21 23:30:30 +03:00
et <- async (readStdErr e)
2019-09-04 01:17:20 +03:00
pure (Serf i o et p ss serf)
2019-07-17 02:14:46 +03:00
where
diskKey = ""
2019-07-22 00:24:07 +03:00
config = show (compileFlags flags)
2019-07-21 23:30:30 +03:00
args = [pierPath, diskKey, config]
pSpec = (proc "urbit-worker" args)
2019-07-17 02:14:46 +03:00
{ std_in = CreatePipe
, std_out = CreatePipe
2019-07-21 23:30:30 +03:00
, std_err = CreatePipe
2019-07-17 02:14:46 +03:00
}
2019-08-28 15:22:56 +03:00
readStdErr :: e. HasLogFunc e => Handle -> RIO e ()
2019-07-21 23:30:30 +03:00
readStdErr h =
untilEOFExn $ do
2019-08-28 15:22:56 +03:00
ln <- io $ IO.hGetLine h
serf ("[stderr] " <> T.strip (pack ln))
2019-07-21 23:30:30 +03:00
where
eofMsg = "[Serf.readStdErr] serf stderr closed"
2019-07-22 00:24:07 +03:00
2019-08-28 15:22:56 +03:00
untilEOFExn :: RIO e () -> RIO e ()
2019-07-21 23:30:30 +03:00
untilEOFExn act = loop
where
2019-08-28 15:22:56 +03:00
loop :: RIO e ()
2019-07-21 23:30:30 +03:00
loop = do
2019-08-28 15:22:56 +03:00
env <- ask
res <- io $ IO.tryIOError $ runRIO env act
case res of
Left exn | IO.isEOFError exn -> logDebug eofMsg
Left exn -> io (IO.ioError exn)
Right () -> loop
2019-09-04 01:17:20 +03:00
tearDown :: HasLogFunc e => Serf e -> RIO e ()
tearDown serf = do
2019-08-28 15:22:56 +03:00
io $ terminateProcess (process serf)
void $ waitForExit serf
2019-08-13 08:56:31 +03:00
-- race_ waitThenKill (shutdownAndWait serf 0)
2019-07-22 00:24:07 +03:00
where
2019-08-13 08:56:31 +03:00
-- killedMsg =
-- "[Serf.tearDown]: Serf didn't die when asked, killing it"
2019-07-22 00:24:07 +03:00
2019-08-13 08:56:31 +03:00
-- waitThenKill = do
-- threadDelay 1000000
-- debug killedMsg
-- terminateProcess (process serf)
2019-09-04 01:17:20 +03:00
waitForExit :: HasLogFunc e => Serf e -> RIO e ExitCode
2019-08-28 15:22:56 +03:00
waitForExit = io . waitForProcess . process
2019-09-04 01:17:20 +03:00
kill :: HasLogFunc e => Serf e -> RIO e ExitCode
2019-08-28 15:22:56 +03:00
kill serf = io (terminateProcess $ process serf) >> waitForExit serf
2019-07-17 02:14:46 +03:00
2019-09-04 01:17:20 +03:00
_shutdownAndWait :: HasLogFunc e => Serf e -> Word8 -> RIO e ExitCode
2019-08-28 15:22:56 +03:00
_shutdownAndWait serf code = do
shutdown serf code
waitForExit serf
2019-07-17 02:14:46 +03:00
-- Basic Send and Receive Operations -------------------------------------------
2019-08-28 15:22:56 +03:00
withWord64AsByteString :: Word64 -> (ByteString -> RIO e a) -> RIO e a
2019-07-17 02:14:46 +03:00
withWord64AsByteString w k = do
2019-08-28 15:22:56 +03:00
env <- ask
io $ alloca $ \wp -> do
poke wp w
bs <- BS.unsafePackCStringLen (castPtr wp, 8)
runRIO env (k bs)
2019-07-17 02:14:46 +03:00
2019-09-04 01:17:20 +03:00
sendLen :: HasLogFunc e => Serf e -> Int -> RIO e ()
2019-07-17 02:14:46 +03:00
sendLen s i = do
w <- evaluate (fromIntegral i :: Word64)
withWord64AsByteString (fromIntegral i) (hPut (sendHandle s))
2019-09-04 01:17:20 +03:00
sendOrder :: HasLogFunc e => Serf e -> Order -> RIO e ()
2019-07-17 02:14:46 +03:00
sendOrder w o = do
logDebug $ display ("(sendOrder) " <> tshow o)
sendBytes w $ jamBS $ toNoun o
logDebug "(sendOrder) Done"
2019-07-17 02:14:46 +03:00
2019-09-04 01:17:20 +03:00
sendBytes :: HasLogFunc e => Serf e -> ByteString -> RIO e ()
2019-08-13 08:56:31 +03:00
sendBytes s bs = handle ioErr $ do
2019-07-17 02:14:46 +03:00
sendLen s (length bs)
hFlush (sendHandle s)
2019-08-13 08:56:31 +03:00
hack
2019-07-17 02:14:46 +03:00
hPut (sendHandle s) bs
hFlush (sendHandle s)
2019-08-13 08:56:31 +03:00
hack
where
2019-08-28 15:22:56 +03:00
ioErr :: IOError -> RIO e ()
2019-08-13 08:56:31 +03:00
ioErr _ = throwIO SerfConnectionClosed
-- TODO WHY DOES THIS MATTER?????
hack = threadDelay 10000
2019-09-04 01:17:20 +03:00
recvLen :: (MonadIO m, HasLogFunc e) => Serf e -> m Word64
2019-08-28 15:22:56 +03:00
recvLen w = io $ do
2019-07-17 02:14:46 +03:00
bs <- hGet (recvHandle w) 8
case length bs of
8 -> unsafeUseAsCString bs (peek . castPtr)
_ -> throwIO SerfConnectionClosed
2019-09-04 01:17:20 +03:00
recvBytes :: HasLogFunc e => Serf e -> Word64 -> RIO e ByteString
2019-08-28 15:22:56 +03:00
recvBytes serf =
io . hGet (recvHandle serf) . fromIntegral
2019-07-17 02:14:46 +03:00
2019-09-04 01:17:20 +03:00
recvAtom :: HasLogFunc e => Serf e -> RIO e Atom
2019-07-17 02:14:46 +03:00
recvAtom w = do
len <- recvLen w
bs <- recvBytes w len
pure (packAtom bs)
where
packAtom :: ByteString -> Atom
packAtom = view (from atomBytes)
cordText :: Cord -> Text
2019-07-22 21:10:27 +03:00
cordText = T.strip . unCord
2019-07-17 02:14:46 +03:00
2019-06-01 03:21:44 +03:00
--------------------------------------------------------------------------------
2019-09-04 01:17:20 +03:00
snapshot :: HasLogFunc e => Serf e -> SerfState -> RIO e ()
2019-08-15 05:42:48 +03:00
snapshot serf ss = sendOrder serf $ OSave $ ssLastEv ss
2019-07-17 02:14:46 +03:00
2019-09-04 01:17:20 +03:00
shutdown :: HasLogFunc e => Serf e -> Word8 -> RIO e ()
shutdown serf code = sendOrder serf (OExit code)
2019-07-17 02:14:46 +03:00
{-
TODO Find a cleaner way to handle `PStdr` Pleas.
2019-07-17 02:14:46 +03:00
-}
2019-09-04 01:17:20 +03:00
recvPlea :: HasLogFunc e => Serf e -> RIO e Plea
2019-07-17 02:14:46 +03:00
recvPlea w = do
logDebug "(recvPlea) Waiting"
2019-07-17 02:14:46 +03:00
a <- recvAtom w
logDebug "(recvPlea) Got atom"
2019-07-17 02:14:46 +03:00
n <- fromRightExn (cue a) (const $ BadPleaAtom a)
p <- fromRightExn (fromNounErr n) (\(p,m) -> BadPleaNoun (traceShowId n) p m)
2019-07-17 02:14:46 +03:00
2019-09-04 01:17:20 +03:00
case p of PStdr e msg -> do (sStderr w) (cordText msg)
recvPlea w
2019-09-04 01:17:20 +03:00
PSlog _ pri t -> do printTank (sStderr w) pri t
recvPlea w
_ -> do logTrace "recvPlea got something else"
pure p
2019-07-17 02:14:46 +03:00
{-
Waits for initial plea, and then sends boot IPC if necessary.
-}
2019-09-04 01:17:20 +03:00
handshake :: HasLogFunc e => Serf e -> LogIdentity -> RIO e SerfState
2019-07-17 02:14:46 +03:00
handshake serf ident = do
ss@SerfState{..} <- recvPlea serf >>= \case
PPlay Nothing -> pure $ SerfState 1 (Mug 0)
PPlay (Just (e, m, _)) -> pure $ SerfState e m
x -> throwIO (InvalidInitialPlea x)
2019-07-17 02:14:46 +03:00
when (ssNextEv == 1) $ do
2019-07-17 02:14:46 +03:00
sendOrder serf (OBoot ident)
pure ss
2019-07-17 02:14:46 +03:00
2019-09-04 01:17:20 +03:00
sendWork :: e. HasLogFunc e => Serf e -> Job -> RIO e SerfResp
sendWork w job =
do
sendOrder w (OWork job)
res <- loop
2019-08-28 15:22:56 +03:00
logTrace ("[sendWork] Got response")
pure res
where
eId = jobId job
2019-08-28 15:22:56 +03:00
produce :: WorkResult -> RIO e SerfResp
produce (ss@SerfState{..}, o) = do
guardExn (ssNextEv == (1+eId)) (BadComputeId eId (ss, o))
pure $ Right (ss, o)
2019-06-01 03:21:44 +03:00
2019-08-28 15:22:56 +03:00
replace :: ReplacementEv -> RIO e SerfResp
replace job = do
guardExn (jobId job == eId) (BadReplacementId eId job)
pure (Left job)
2019-08-28 15:22:56 +03:00
loop :: RIO e SerfResp
2019-06-01 03:21:44 +03:00
loop = recvPlea w >>= \case
PPlay p -> throwIO (UnexpectedPlay eId p)
PDone i m o -> produce (SerfState (i+1) m, o)
PWork work -> replace (DoWork work)
2019-09-04 01:17:20 +03:00
PStdr _ cord -> (sStderr w) (cordText cord) >> loop
PSlog _ pri t -> printTank (sStderr w) pri t >> loop
--------------------------------------------------------------------------------
2019-09-04 01:17:20 +03:00
doJob :: HasLogFunc e => Serf e -> Job -> RIO e (Job, SerfState, FX)
doJob serf job = do
sendWork serf job >>= \case
Left replaced -> doJob serf replaced
Right (ss, fx) -> pure (job, ss, fx)
2019-09-04 01:17:20 +03:00
bootJob :: HasLogFunc e => Serf e -> Job -> RIO e (Job, SerfState)
bootJob serf job = do
doJob serf job >>= \case
(job, ss, []) -> pure (job, ss)
(job, ss, fx) -> throwIO (EffectsDuringBoot (jobId job) fx)
2019-09-04 01:17:20 +03:00
replayJob :: HasLogFunc e => Serf e -> Job -> RIO e SerfState
replayJob serf job = do
sendWork serf job >>= \case
Left replace -> throwIO (ReplacedEventDuringReplay (jobId job) replace)
Right (ss, _) -> pure ss
--------------------------------------------------------------------------------
type BootSeqFn = EventId -> Mug -> Time.Wen -> Job
data BootExn = ShipAlreadyBooted
deriving stock (Eq, Ord, Show)
deriving anyclass (Exception)
2019-09-04 01:17:20 +03:00
bootFromSeq :: e. HasLogFunc e => Serf e -> BootSeq -> RIO e ([Job], SerfState)
2019-07-16 03:23:48 +03:00
bootFromSeq serf (BootSeq ident nocks ovums) = do
handshake serf ident >>= \case
ss@(SerfState 1 (Mug 0)) -> loop [] ss bootSeqFns
_ -> throwIO ShipAlreadyBooted
2019-07-16 03:01:45 +03:00
where
2019-08-28 15:22:56 +03:00
loop :: [Job] -> SerfState -> [BootSeqFn] -> RIO e ([Job], SerfState)
loop acc ss = \case
[] -> pure (reverse acc, ss)
2019-08-28 15:22:56 +03:00
x:xs -> do wen <- io Time.now
job <- pure $ x (ssNextEv ss) (ssLastMug ss) wen
(job, ss) <- bootJob serf job
loop (job:acc) ss xs
bootSeqFns :: [BootSeqFn]
bootSeqFns = fmap muckNock nocks <> fmap muckOvum ovums
2019-07-16 03:23:48 +03:00
where
muckNock nok eId mug _ = RunNok $ LifeCyc eId mug nok
muckOvum ov eId mug wen = DoWork $ Work eId mug wen ov
2019-07-17 02:14:46 +03:00
{-
The ship is booted, but it is behind. shove events to the worker
until it is caught up.
-}
2019-08-28 15:22:56 +03:00
replayJobs :: HasLogFunc e
2019-09-04 01:17:20 +03:00
=> Serf e -> SerfState -> ConduitT Job Void (RIO e) SerfState
replayJobs serf = go
where
2019-08-28 15:22:56 +03:00
go ss = await >>= maybe (pure ss) (lift . replayJob serf >=> go)
2019-09-04 01:17:20 +03:00
replay :: HasLogFunc e => Serf e -> Log.EventLog -> RIO e SerfState
replay serf log = do
ss <- handshake serf (Log.identity log)
runConduit $ Log.streamEvents log (ssNextEv ss)
.| toJobs (Log.identity log) (ssNextEv ss)
.| replayJobs serf ss
2019-08-28 15:22:56 +03:00
toJobs :: HasLogFunc e
=> LogIdentity -> EventId -> ConduitT ByteString Job (RIO e) ()
toJobs ident eId =
await >>= \case
2019-08-28 15:22:56 +03:00
Nothing -> lift $ logTrace "[toJobs] no more jobs"
Just at -> do yield =<< lift (fromAtom at)
lift $ logTrace $ display ("[toJobs] " <> tshow eId)
toJobs ident (eId+1)
where
2019-08-28 15:22:56 +03:00
isNock = eId <= fromIntegral (lifecycleLen ident)
2019-08-28 15:22:56 +03:00
fromAtom :: ByteString -> RIO e Job
fromAtom bs | isNock = do
noun <- cueBSExn bs
(mug, nok) <- fromNounExn noun
pure $ RunNok (LifeCyc eId mug nok)
fromAtom bs = do
noun <- cueBSExn bs
(mug, wen, ovm) <- fromNounExn noun
pure $ DoWork (Work eId mug wen ovm)
-- Collect Effects for Parsing -------------------------------------------------
2019-09-04 01:17:20 +03:00
collectFX :: HasLogFunc e => Serf e -> Log.EventLog -> RIO e ()
collectFX serf log = do
ss <- handshake serf (Log.identity log)
2019-07-17 02:14:46 +03:00
runConduit $ Log.streamEvents log (ssNextEv ss)
.| toJobs (Log.identity log) (ssNextEv ss)
.| doCollectFX serf ss
.| persistFX log
2019-08-28 15:22:56 +03:00
persistFX :: Log.EventLog -> ConduitT (EventId, FX) Void (RIO e) ()
persistFX log = loop
where
loop = await >>= \case
Nothing -> pure ()
Just (eId, fx) -> do
2019-08-29 03:26:59 +03:00
lift $ Log.writeEffectsRow log eId (jamBS $ toNoun fx)
loop
2019-08-28 15:22:56 +03:00
doCollectFX :: e. HasLogFunc e
2019-09-04 01:17:20 +03:00
=> Serf e -> SerfState -> ConduitT Job (EventId, FX) (RIO e) ()
doCollectFX serf = go
where
2019-08-28 15:22:56 +03:00
go :: SerfState -> ConduitT Job (EventId, FX) (RIO e) ()
go ss = await >>= \case
Nothing -> pure ()
Just jb -> do
-- jb <- pure $ replaceMug jb (ssLastMug ss)
2019-08-28 15:22:56 +03:00
(_, ss, fx) <- lift $ doJob serf jb
lift $ logTrace $ displayShow (jobId jb)
yield (jobId jb, fx)
go ss
replaceMug :: Job -> Mug -> Job
replaceMug jb mug =
case jb of
DoWork (Work eId _ w o) -> DoWork (Work eId mug w o)
RunNok (LifeCyc eId _ n) -> RunNok (LifeCyc eId mug n)