diff --git a/app/Main.hs b/app/Main.hs index 0ddeb36..4e1501d 100644 --- a/app/Main.hs +++ b/app/Main.hs @@ -12,6 +12,7 @@ import Invar.Worker qualified as W import Options qualified as O import PolicyInput qualified import Quality qualified +import Remote qualified import Scoring qualified import System.Console.GetOpt (OptDescr, usageInfo) import System.Environment (getArgs) @@ -35,6 +36,7 @@ main = do "use" : rest -> Use.run rest "train" : rest -> Training.run rest "policy" : rest -> PolicyInput.run rest + "remote" : rest -> Remote.run rest "infer" : "batch" : rest -> InferenceBatch.run rest "infer" : rest -> do (worker, planned, bound) <- parse rest @@ -43,7 +45,7 @@ main = do _ -> die usage usage :: String -usage = usageInfo "Usage: invar use prepare|plan|run|inspect|admit OPTIONS | invar infer OPTIONS | invar infer batch OPTIONS < calls.json | invar score OPTIONS | invar policy OPTIONS | invar train OPTIONS < tasks.json | invar evaluate OPTIONS < tasks.json | invar quality OPTIONS | invar inspect KIND OPTIONS | invar compare KIND OPTIONS\nUse the command's --help for its options. Inference selects --checkpoint, or an explicit --adapter with all four digest options. --checkpoint reads the published policy description. --worker-config is optional." options +usage = usageInfo "Usage: invar use prepare|plan|run|inspect|admit OPTIONS | invar infer OPTIONS | invar infer batch OPTIONS < calls.json | invar score OPTIONS | invar policy OPTIONS | invar train OPTIONS < tasks.json | invar evaluate OPTIONS < tasks.json | invar quality OPTIONS | invar inspect KIND OPTIONS | invar compare KIND OPTIONS | invar remote run|status|terminate OPTIONS\nUse the command's --help for its options. Inference selects --checkpoint, or an explicit --adapter with all four digest options. --checkpoint reads the published policy description. --worker-config is optional." options options :: [OptDescr (String, String)] options = O.descriptions [("python", "Python executable"), ("worker", "Inference worker script"), ("worker-config", "Optional worker launch configuration"), ("cache", "Pinned model cache"), ("adapter", "Explicit adapter file or native handoff directory"), ("checkpoint", "Checkpoint containing policy.json and adapter.safetensors")] ++ InferenceInput.options diff --git a/app/Remote.hs b/app/Remote.hs new file mode 100644 index 0000000..f0c2a1b --- /dev/null +++ b/app/Remote.hs @@ -0,0 +1,75 @@ +{-# LANGUAGE OverloadedStrings #-} + +module Remote (run) where + +import Control.Monad (unless) +import Data.Aeson (encode, object, (.=)) +import Data.ByteString.Lazy.Char8 qualified as Lazy +import Data.List (nub) +import Data.Text qualified as Text +import Invar.Remote qualified as R +import Invar.Remote.Control qualified as C +import Options qualified as O +import System.Console.GetOpt (OptDescr, usageInfo) +import System.Environment (getExecutablePath) +import System.Exit (die, exitWith) + +run :: [String] -> IO () +run ["--help"] = putStrLn usage +run [_, "--help"] = putStrLn usage +run ("run" : supplied) = do + let (named, command) = break (== "--") supplied + (declared, rest) = environment named + (target, values) <- case drop 1 command of + chosen : remaining -> pure (chosen, remaining) + [] -> die "remote run needs -- PROGRAM [ARGUMENT]..." + fields <- either die pure (O.parse runOptions rest) + pairs <- either die pure (traverse assignment declared) + unless (length (nub (map fst pairs)) == length pairs) (die "Duplicate environment names are not allowed") + root <- either die pure (O.required fields "instance") + chosen <- either die pure (O.required fields "nonce") + self <- getExecutablePath + R.run self (R.Launch root (Text.pack chosen) (O.optional fields "directory") pairs target values) >>= exitWith +run ("exec" : supplied) = do + let (named, command) = break (== "--") supplied + fields <- either die pure (O.parse execOptions named) + barrier <- either die pure (O.numeric fields "barrier") + report <- either die pure (O.numeric fields "status") + case drop 1 command of + target : values -> R.exec (fromInteger barrier) (fromInteger report) target values + [] -> die "remote exec needs -- PROGRAM [ARGUMENT]..." +run ("status" : supplied) = do + fields <- either die pure (O.parse statusOptions supplied) + root <- either die pure (O.required fields "instance") + R.status root >>= either die (Lazy.putStrLn . encode . C.describe) +run ("terminate" : supplied) = do + fields <- either die pure (O.parse terminateOptions supplied) + root <- either die pure (O.required fields "instance") + chosen <- either die pure (O.required fields "nonce") + request <- either die pure (C.Request (Text.pack chosen) <$> O.numeric fields "wrapper" <*> O.numeric fields "worker" <*> O.numeric fields "grace") + seconds <- either die pure (O.numeric fields "wait") + (written, answered, current) <- R.terminate root request seconds + Lazy.putStrLn (encode (object ["requested" .= written, "answer" .= C.answerName answered, "status" .= C.describe current])) +run _ = die usage + +environment :: [String] -> ([String], [String]) +environment ("--environment" : value : rest) = let (declared, remaining) = environment rest in (value : declared, remaining) +environment (other : rest) = fmap (other :) (environment rest) +environment [] = ([], []) + +assignment :: String -> Either String (String, String) +assignment value = case break (== '=') value of + (name@(_ : _), '=' : setting) -> Right (name, setting) + _ -> Left ("An environment entry is NAME=VALUE: " ++ value) + +runOptions, execOptions, statusOptions, terminateOptions :: [OptDescr (String, String)] +runOptions = O.descriptions [("instance", "Instance directory to create; it must not exist"), ("nonce", "Nonce the core chose for this launch"), ("directory", "Working directory of the worker")] +execOptions = O.descriptions [("barrier", "Descriptor to read the release from"), ("status", "Descriptor to report a failed start on")] +statusOptions = O.descriptions [("instance", "Instance directory")] +terminateOptions = O.descriptions [("instance", "Instance directory"), ("nonce", "Nonce of the instance"), ("wrapper", "Wrapper process id of the instance"), ("worker", "Worker process id of the instance"), ("grace", "Seconds between termination and kill"), ("wait", "Seconds to wait for the wrapper's confirmation")] + +usage :: String +usage = + usageInfo + "Usage: invar remote run --instance DIRECTORY --nonce NONCE [--directory PATH] [--environment NAME=VALUE]... -- PROGRAM [ARGUMENT]...\n invar remote status --instance DIRECTORY\n invar remote terminate --instance DIRECTORY --nonce NONCE --wrapper PID --worker PID --grace SECONDS --wait SECONDS\nrun starts the worker behind a start barrier with exactly the declared environment, forwards its stdout unchanged, and writes its control files in the instance directory; status and terminate print one JSON object. invar remote exec is the internal child of run." + (runOptions ++ terminateOptions) diff --git a/invar.cabal b/invar.cabal index 9594571..bb0c84e 100644 --- a/invar.cabal +++ b/invar.cabal @@ -28,7 +28,7 @@ library cc-options: -Wall -Wextra -Werror default-extensions: NoGeneralizedNewtypeDeriving hs-source-dirs: src - exposed-modules: Invar.Async.Plan, Invar.Tokenizer, Invar.Learn.Algorithm, Invar.Teacher, Invar.Async.Core, Invar.Async.Completion, Invar.Async.Entry, Invar.Async.Replay, Invar.Transcript, Invar.Canonical, Invar.Infer.Trajectory, Invar.Infer.Session, Invar.Infer.Replay, Invar.Resident.Owner, Invar.Journal, Invar.Runtime, Invar.Learn.Stream, Invar.Use.Measurement, Invar.Use.Decimal, Invar.Use.Confidence, Invar.Use.Plan, Invar.Use.Prepare, Invar.Use.Execution, Invar.Use.Statistics, Invar.Use, Invar.Score.Worker, Invar.Score.Resident, Invar.Score, Invar.Numerical, Invar.Spec.Request, Invar.Spec.Value, Invar.Spec.Program, Invar.Spec.Dependency, Invar.Spec.Operator, Invar.Spec.Evaluate, Invar.Spec.Artifact, Invar.Spec.Invocation, Invar.Spec.Evidence, Invar.Spec.Load, Invar.Construct, Invar.Literal, Invar.Store, Invar.Infer, Invar.Infer.Result, Invar.Infer.Observation, Invar.Infer.Invocation, Invar.Worker, Invar.Worker.Resident, Invar.Reward, Invar.Cohort, Invar.Schedule, Invar.Rollout, Invar.Learn, Invar.Learn.Advantage, Invar.Learn.Objective, Invar.Learn.Mismatch, Invar.Learn.Program, Invar.Learn.Wire, Invar.Learn.Protocol, Invar.Learn.Worker, Invar.Learn.Worker.Resident, Invar.Loop, Invar.Policy, Invar.Artifact, Invar.Workload, Invar.Evaluation, Invar.Quality, Invar.Learn.Report, Invar.Learn.Observation, Invar.Learn.State, Invar.History.Cohort, Invar.History.Trace, Invar.History.Generation, Invar.History.Schedule, Invar.History.Runtime, Invar.History.Initial, Invar.History + exposed-modules: Invar.Async.Plan, Invar.Tokenizer, Invar.Learn.Algorithm, Invar.Teacher, Invar.Async.Core, Invar.Async.Completion, Invar.Async.Entry, Invar.Async.Replay, Invar.Transcript, Invar.Canonical, Invar.Infer.Trajectory, Invar.Infer.Session, Invar.Infer.Replay, Invar.Resident.Owner, Invar.Journal, Invar.Runtime, Invar.Learn.Stream, Invar.Use.Measurement, Invar.Use.Decimal, Invar.Use.Confidence, Invar.Use.Plan, Invar.Use.Prepare, Invar.Use.Execution, Invar.Use.Statistics, Invar.Use, Invar.Score.Worker, Invar.Score.Resident, Invar.Score, Invar.Numerical, Invar.Spec.Request, Invar.Spec.Value, Invar.Spec.Program, Invar.Spec.Dependency, Invar.Spec.Operator, Invar.Spec.Evaluate, Invar.Spec.Artifact, Invar.Spec.Invocation, Invar.Spec.Evidence, Invar.Spec.Load, Invar.Construct, Invar.Literal, Invar.Store, Invar.Infer, Invar.Infer.Result, Invar.Infer.Observation, Invar.Infer.Invocation, Invar.Worker, Invar.Worker.Resident, Invar.Reward, Invar.Cohort, Invar.Schedule, Invar.Rollout, Invar.Learn, Invar.Learn.Advantage, Invar.Learn.Objective, Invar.Learn.Mismatch, Invar.Learn.Program, Invar.Learn.Wire, Invar.Learn.Protocol, Invar.Learn.Worker, Invar.Learn.Worker.Resident, Invar.Loop, Invar.Policy, Invar.Artifact, Invar.Workload, Invar.Evaluation, Invar.Quality, Invar.Learn.Report, Invar.Learn.Observation, Invar.Learn.State, Invar.History.Cohort, Invar.History.Trace, Invar.History.Generation, Invar.History.Schedule, Invar.History.Runtime, Invar.History.Initial, Invar.History, Invar.Remote, Invar.Remote.Control other-modules: Invar.Async.Completion.Internal, Invar.Infer.Trajectory.Internal, Invar.Infer.Records, Invar.Learn.Step, Invar.Spec.Measurement, Invar.Spec.Domain, Invar.Use.Encoding, Invar.Use.Contract, Invar.Use.Admission, Invar.Use.Finding, Invar.Numerical.Logarithm, Invar.Spec.UseContract, Invar.Spec.Obligation, Invar.Evidence.Encoding, Invar.Use.Observation, Invar.Spec.Use, Invar.Numerical.Distribution, Invar.Score.Execution, Invar.Score.Probe, Invar.Score.Resources, Invar.Spec.Score, Invar.Infer.Schema, Invar.Score.Program, Invar.Score.Output, Invar.Spec.Numerical, Invar.Quality.Summary, Invar.Reward.Decimal, Invar.Reward.Program, Invar.Materialization, Invar.Load, Invar.Infer.Load, Invar.Infer.Batch, Invar.Infer.Framing, Invar.Spec.Syntax, Invar.Spec.Decode, Invar.Spec.Encode, Invar.Construct.Types, Invar.Infer.Wire, Invar.Infer.Output, Invar.Infer.Model, Invar.Rollout.Internal, Invar.Rollout.Resident, Invar.Process, Invar.Process.Internal, Invar.Process.Resident, Invar.Resident, Invar.Resident.Group, Invar.Policy.Header, Invar.Policy.Description, Invar.Policy.Encoding, Invar.Json, Invar.Learn.Native, Invar.Learn.Observed, Invar.Learn.Codec, Invar.Learn.Adapter, Invar.Learn.Checkpoint, Invar.Learn.Checkpoint.Types, Invar.Learn.Checkpoint.MLX, Invar.Learn.Request, Invar.Learn.Gradient, Invar.Policy.File, Invar.Learn.Probability, Invar.Learn.Fingerprint, Invar.Float32, Invar.Digest, Invar.Learn.Trace, Invar.Learn.Report.Internal, Invar.Async.Recorded, Invar.Learn.Framing, Invar.Learn.Worker.Internal, Invar.Learn.Worker.Observation, Invar.Learn.Worker.Owner, Invar.Loop.Owner, Invar.History.Execution, Invar.History.Profile, Invar.History.Publication, Invar.History.Artifacts, Invar.Measurement.Duration test-suite request @@ -36,7 +36,7 @@ test-suite request type: exitcode-stdio-1.0 hs-source-dirs: test main-is: Main.hs - other-modules: TokenizerFixture, Tokenizers, Algorithms, Distillations, Teachers, Events, Journals, Replays, Recoveries, Restarts, Comparisons, Drivers, Generation, Logs, Evaluations, Traces, Steps, UseBernstein, UsePlan, UsePrepare, UseExecution, UseGeneric, Statistics, UseAdmission, UseEvidence, Uses, UseFixture, NumericalProbes, ScoreProbes, NumericalScores, Scores, Numerical, Model, History, Properties, Fixtures, Arguments, Values, Dependencies, Evaluation, Artifacts, Construction, Collections, Records, Literals, Invocations, Evidence, Loads, Store, Checkpoints, Inference, Results, Calls, BatchCalls, GroupedCalls, BatchedProtocol, ResidentFixture, ResidentCalls, ResidentWorkloads, ResidentDrivers, ResidentObservations, Streams, Rewards, Cohorts, Rollouts, Sessions, Learning, Advantages, Objectives, Mismatches, Updates, LearnerFixture, LearnerCalls, Loops, Schedules, Policies, Probabilities, Workloads, Reports, Quality, Observations, InferenceObservations + other-modules: Remotes, TokenizerFixture, Tokenizers, Algorithms, Distillations, Teachers, Events, Journals, Replays, Recoveries, Restarts, Comparisons, Drivers, Generation, Logs, Evaluations, Traces, Steps, UseBernstein, UsePlan, UsePrepare, UseExecution, UseGeneric, Statistics, UseAdmission, UseEvidence, Uses, UseFixture, NumericalProbes, ScoreProbes, NumericalScores, Scores, Numerical, Model, History, Properties, Fixtures, Arguments, Values, Dependencies, Evaluation, Artifacts, Construction, Collections, Records, Literals, Invocations, Evidence, Loads, Store, Checkpoints, Inference, Results, Calls, BatchCalls, GroupedCalls, BatchedProtocol, ResidentFixture, ResidentCalls, ResidentWorkloads, ResidentDrivers, ResidentObservations, Streams, Rewards, Cohorts, Rollouts, Sessions, Learning, Advantages, Objectives, Mismatches, Updates, LearnerFixture, LearnerCalls, Loops, Schedules, Policies, Probabilities, Workloads, Reports, Quality, Observations, InferenceObservations build-depends: invar, hedgehog >= 1.7 && < 1.8, @@ -44,13 +44,16 @@ test-suite request text >= 2.1 && < 2.2, directory >= 1.3 && < 1.4, filepath >= 1.5 && < 1.6, - unix >= 2.8 && < 2.9 + unix >= 2.8 && < 2.9, + process >= 1.6.26 && < 1.7, + cryptohash-sha256 >= 0.11.102.1 && < 0.12 + build-tool-depends: invar:invar executable invar import: defaults hs-source-dirs: app main-is: Main.hs - other-modules: Use, UseRuns, UsePlan, UsePrepare, UseExecution, UseInput, UseRun, Check, Located, Envelope, Scoring, Numerical, Options, InferenceInput, EvaluationInput, InferenceBatch, PolicyInput, NativeCodec, HistoryInput, Dataset, Training, Evaluation, Quality, Inspection, Comparison + other-modules: Use, UseRuns, UsePlan, UsePrepare, UseExecution, UseInput, UseRun, Check, Located, Envelope, Scoring, Numerical, Options, InferenceInput, EvaluationInput, InferenceBatch, PolicyInput, NativeCodec, HistoryInput, Dataset, Training, Evaluation, Quality, Inspection, Comparison, Remote build-depends: invar, text >= 2.1 && < 2.2, aeson >= 2.3.1 && < 2.4, cryptohash-sha256 >= 0.11.102.1 && < 0.12, directory >= 1.3 && < 1.4, filepath >= 1.5 && < 1.6, process >= 1.6.26 && < 1.7 source-repository head diff --git a/src/Invar/Remote.hs b/src/Invar/Remote.hs new file mode 100644 index 0000000..b4beb7b --- /dev/null +++ b/src/Invar/Remote.hs @@ -0,0 +1,229 @@ +{-# LANGUAGE OverloadedStrings #-} + +module Invar.Remote (Launch (..), run, exec, status, terminate) where + +import Control.Concurrent (MVar, forkIO, isEmptyMVar, newEmptyMVar, putMVar, takeMVar, threadDelay) +import Control.Exception (IOException, catch, throwIO, try) +import Control.Monad (unless, void, when) +import Crypto.Hash.SHA256 qualified as SHA256 +import Data.ByteString (ByteString) +import Data.ByteString qualified as Bytes +import Data.ByteString.Char8 qualified as Char +import Data.Maybe (isJust, isNothing) +import Data.Text (Text) +import Data.Text qualified as Text +import Data.Text.Encoding (decodeUtf8Lenient) +import GHC.Clock (getMonotonicTime) +import Invar.Artifact qualified as Artifact +import Invar.Remote.Control qualified as C +import Invar.Store qualified as Store +import Numeric.Natural (Natural) +import System.Directory (createDirectory, doesDirectoryExist, doesFileExist, removeFile) +import System.Exit (ExitCode (..), exitWith) +import System.FilePath (takeFileName, ()) +import System.IO (Handle, SeekMode (AbsoluteSeek), hClose, hFlush, hSetBinaryMode, stdout) +import System.IO.Error (catchIOError, isAlreadyExistsError, isDoesNotExistError, isEOFError) +import System.Posix.IO qualified as Posix +import System.Posix.IO.ByteString qualified as Raw +import System.Posix.Process (executeFile, getProcessID) +import System.Posix.Signals (Handler (Default, Ignore), Signal, installHandler, killProcess, sigHUP, sigPIPE, signalProcessGroup, softwareTermination) +import System.Posix.Temp (mkstemp) +import System.Posix.Time (epochTime) +import System.Posix.Types (Fd) +import System.Posix.Unistd (getSystemID, nodeName) +import System.Process (CreateProcess (..), ProcessHandle, StdStream (..), createProcess, getPid, getProcessExitCode, proc) + +-- | A remote launch: the instance directory and nonce the core chose, and the worker with exactly the declared +-- working directory and environment. +data Launch = Launch {instanceDirectory :: FilePath, launchNonce :: Text, directory :: Maybe FilePath, environment :: [(String, String)], program :: FilePath, arguments :: [String]} + +-- | Runs the worker behind the start barrier as a child of this process, forwards its stdout unchanged, and is the +-- only owner that signals and reaps it. `self` is this executable, started again as `invar remote exec`. +run :: FilePath -> Launch -> IO ExitCode +run self launch = do + let root = instanceDirectory launch + createDirectory root + held <- Posix.openFd (root "lock") Posix.ReadWrite Posix.defaultFileFlags {Posix.creat = Just 0o644, Posix.exclusive = True, Posix.cloexec = True} + Posix.setLock held (Posix.WriteLock, AbsoluteSeek, 0, 0) + name <- Text.pack . nodeName <$> getSystemID + own <- fromIntegral <$> getProcessID + let launched = C.Launching name (launchNonce launch) own + publish root "launching" (C.launchingFile launched) + void (installHandler sigHUP Ignore Nothing) + (barrier, release) <- Posix.createPipe + (report, reported) <- Posix.createPipe + Posix.setFdOption release Posix.CloseOnExec True + Posix.setFdOption report Posix.CloseOnExec True + let child = (proc self (["remote", "exec", "--barrier", show barrier, "--status", show reported, "--", program launch] ++ arguments launch)) {std_in = Inherit, std_out = CreatePipe, std_err = Inherit, cwd = directory launch, env = Just (environment launch), create_group = True} + spawned <- try (createProcess child) + Posix.closeFd barrier + Posix.closeFd reported + case spawned of + Left (problem :: IOException) -> do + publish root "failure" (C.failureFile (C.Failure launched (Text.pack (show problem)))) + pure (ExitFailure 125) + Right (_, Just output, _, handle) -> do + pid <- getPid handle >>= maybe (ioError (userError "The wrapper's child has no process id")) pure + let started = C.Instance launched (fromIntegral pid) + publish root "instance" (C.instanceFile started) + void (Raw.fdWrite release "1") + Posix.closeFd release + forwarded <- forward output + failure <- Posix.fdToHandle report >>= \source -> Bytes.hGetContents source + code <- supervise root started handle forwarded + (count, sha) <- takeMVar forwarded + at <- epochTime + if Bytes.null failure + then do + publish root "outcome" (C.outcomeFile (C.Outcome started (ending code) (fromIntegral (fromEnum at)) count sha)) + pure code + else do + publish root "failure" (C.failureFile (C.Failure launched (Text.pack (Char.unpack failure)))) + pure (ExitFailure 125) + Right _ -> ioError (userError "The worker's stdout pipe was not created") + where + ending ExitSuccess = C.Exited 0 + ending (ExitFailure code) + | code < 0 = C.Signalled (negate code) + | otherwise = C.Exited code + +-- | Forwards every byte of the worker's stdout to this process's stdout, and counts and digests all of it. When a +-- write fails because the connection dropped, it stops writing and keeps reading to the end. +forward :: Handle -> IO (MVar (Natural, String)) +forward source = do + done <- newEmptyMVar + hSetBinaryMode source True + hSetBinaryMode stdout True + let go context count open = do + chunk <- Bytes.hGetSome source 65536 + if Bytes.null chunk + then hClose source >> putMVar done (count, Artifact.hex (SHA256.finalize context)) + else do + writing <- if open then (Bytes.hPut stdout chunk >> hFlush stdout >> pure True) `catchIOError` const (pure False) else pure False + go (SHA256.update context chunk) (count + fromIntegral (Bytes.length chunk)) writing + void (forkIO (go SHA256.init 0 True)) + pure done + +-- | `Waiting` carries the number of the next refusal. +data Phase = Waiting Int | Terminating Double | Killed + +-- | The one loop that owns the child. It keeps the leader unreaped until the worker's stdout is drained and no +-- termination is under way, so the worker's process group keeps its id while any signal may still go to it. It +-- answers an admitted termination request by signalling the group, and kills the group when the grace ends. +supervise :: FilePath -> C.Instance -> ProcessHandle -> MVar (Natural, String) -> IO ExitCode +supervise root started handle forwarded = go (Waiting 1) + where + go phase = do + drained <- not <$> isEmptyMVar forwarded + exited <- if drained && settled phase then getProcessExitCode handle else pure Nothing + case exited of + Just code -> pure code + Nothing -> advance phase >>= \next -> threadDelay 20000 >> go next + settled (Terminating _) = False + settled _ = True + advance (Waiting count) = do + pending <- doesFileExist (root "terminate.json") + if not pending + then pure (Waiting count) + else do + encoded <- Bytes.readFile (root "terminate.json") + case C.admitRequest started encoded of + Right chosen -> do + send softwareTermination C.Terminate "terminated" + now <- getMonotonicTime + pure (Terminating (now + fromIntegral (C.grace chosen))) + Left why -> do + at <- epochTime + publish root ("refused." ++ show count) (C.refusalFile (C.Refusal started (decodeUtf8Lenient encoded) why (fromIntegral (fromEnum at)))) + removeFile (root "terminate.json") + pure (Waiting (count + 1)) + advance (Terminating deadline) = do + now <- getMonotonicTime + if now < deadline then pure (Terminating deadline) else Killed <$ send killProcess C.Kill "killed" + advance Killed = pure Killed + send :: Signal -> C.Signal -> String -> IO () + send number named record = do + delivered <- try (signalProcessGroup number (fromIntegral (C.worker started))) + case delivered of + Left (_ :: IOException) -> pure () + Right () -> do + at <- epochTime + publish root record (C.signalledFile (C.Sent started named (fromIntegral (fromEnum at)))) + +-- | The child started by `run`: it leads the process group it was created in, waits for the barrier, and replaces +-- itself with the worker. A failed replacement is written to the status descriptor, which is closed on success. +exec :: Fd -> Fd -> FilePath -> [String] -> IO () +exec barrier report target values = do + released <- (not . Bytes.null <$> Raw.fdRead barrier 1) `catchIOError` \problem -> if isEOFError problem then pure False else ioError problem + Posix.closeFd barrier + unless released (exitWith (ExitFailure 125)) + Posix.setFdOption report Posix.CloseOnExec True + void (installHandler sigHUP Default Nothing) + void (installHandler sigPIPE Default Nothing) + executeFile target False values Nothing `catchIOError` \problem -> do + void (Raw.fdWrite report (Char.pack (show problem))) + exitWith (ExitFailure 127) + +-- | What the wrapper wrote so far, read in the reverse order of writing so that every file read names files that +-- already existed, and whether a live wrapper still holds the instance. +status :: FilePath -> IO (Either String C.Observation) +status root = do + present <- doesDirectoryExist root + if not present + then pure (C.observe False (C.Files Nothing Nothing Nothing Nothing Nothing Nothing Nothing [])) + else do + holding <- alive root + outcome <- optional "outcome.json" + failure <- optional "failure.json" + killed <- optional "killed.json" + terminated <- optional "terminated.json" + asked <- optional "terminate.json" + refused <- refusals (1 :: Int) + published <- optional "instance.json" + launched <- optional "launching.json" + pure (C.observe holding (C.Files launched published failure outcome asked terminated killed refused)) + where + optional name = (Just <$> Bytes.readFile (root name)) `catchIOError` \problem -> if isDoesNotExistError problem then pure Nothing else ioError problem + refusals count = optional ("refused." ++ show count ++ ".json") >>= maybe (pure []) (\encoded -> (encoded :) <$> refusals (count + 1)) + +alive :: FilePath -> IO Bool +alive root = do + exists <- doesFileExist (root "lock") + if not exists + then pure False + else do + file <- Posix.openFd (root "lock") Posix.ReadWrite Posix.defaultFileFlags {Posix.cloexec = True} + conflict <- Posix.getLock file (Posix.WriteLock, AbsoluteSeek, 0, 0) + Posix.closeFd file + pure (isJust conflict) + +-- | Asks a live wrapper to terminate its worker and waits for the answer to this request. The request is written only +-- while the instance holds none: an admitted request stays in place, and a refused one is removed by the wrapper, so +-- this request is written once that refusal leaves room for it. Without a live wrapper nothing is written and nothing +-- is signalled. It reports whether this call wrote the request, the answer to this request alone, and the instance's +-- status. +terminate :: FilePath -> C.Request -> Int -> IO (Bool, C.Answer, C.Observation) +terminate root chosen seconds = go (seconds * 50) False + where + encoded = C.requestFile chosen + go remaining written = do + current <- status root >>= either (ioError . userError) pure + let answered = C.reply encoded current + if not (C.alive current) || answered `elem` [C.Confirmed, C.Refused] || remaining < (0 :: Int) + then pure (written, answered, current) + else do + wrote <- if answered == C.Unrequested && isNothing (C.asked current) then requested else pure False + threadDelay 20000 + go (remaining - 1) (written || wrote) + requested = (True <$ publish root "terminate" encoded) `catch` \problem@(Store.Failure _ cause) -> if isAlreadyExistsError cause then pure False else throwIO problem + +-- | Writes a control file under a staging name of its own, synchronizes it, renames it into place without replacing +-- anything, and synchronizes the directory. A staging file left by an interrupted writer never blocks another, and a +-- writer whose rename finds the name taken removes its own. +publish :: FilePath -> String -> ByteString -> IO () +publish root name encoded = do + (staged, written) <- mkstemp (root name ++ ".staging.") + Bytes.hPut written encoded + hClose written + void (Store.publish (Store.Location root (Char.pack (takeFileName staged)) (Char.pack (name ++ ".json")))) `catch` \problem@(Store.Failure phase _) -> + when (phase == Store.Rename) (removeFile staged) >> throwIO problem diff --git a/src/Invar/Remote/Control.hs b/src/Invar/Remote/Control.hs new file mode 100644 index 0000000..51615e3 --- /dev/null +++ b/src/Invar/Remote/Control.hs @@ -0,0 +1,317 @@ +{-# LANGUAGE OverloadedStrings #-} + +module Invar.Remote.Control ( + Launching (..), + Instance (..), + Failure (..), + Ending (..), + Outcome (..), + Signal (..), + Signalled (..), + Request (..), + Refusal (..), + Answer (..), + Files (..), + State (..), + Observation (..), + launchingFile, + instanceFile, + failureFile, + outcomeFile, + signalledFile, + requestFile, + refusalFile, + admitRequest, + answer, + reply, + answerName, + observe, + describe, +) where + +import Control.Monad (unless, when) +import Data.Aeson (Key, Object, Value (..), encode, object, parseJSON, withObject, (.:), (.=)) +import Data.Aeson.KeyMap qualified as Fields +import Data.Aeson.Types (Pair, Parser, parseEither) +import Data.ByteString (ByteString) +import Data.ByteString.Lazy qualified as Lazy +import Data.Either (isLeft, isRight) +import Data.Maybe (isJust, isNothing) +import Data.Text (Text) +import Data.Text qualified as Text +import Data.Text.Encoding (decodeUtf8Lenient, encodeUtf8) +import Invar.Json qualified as Json +import Numeric.Natural (Natural) + +data Launching = Launching {host :: Text, nonce :: Text, wrapper :: Natural} + deriving (Eq, Show) + +data Instance = Instance {launching :: Launching, worker :: Natural} + deriving (Eq, Show) + +data Failure = Failure {failed :: Launching, reason :: Text} + deriving (Eq, Show) + +data Ending = Exited Int | Signalled Int + deriving (Eq, Show) + +data Outcome = Outcome {ended :: Instance, ending :: Ending, endedAt :: Integer, bytes :: Natural, digest :: String} + deriving (Eq, Show) + +data Signal = Terminate | Kill + deriving (Eq, Show) + +data Signalled = Sent {target :: Instance, signal :: Signal, sentAt :: Integer} + deriving (Eq, Show) + +data Request = Request {requested :: Text, requestedWrapper :: Natural, requestedWorker :: Natural, grace :: Natural} + deriving (Eq, Show) + +data Refusal = Refusal {held :: Instance, refusedRequest :: Text, refusedBecause :: Text, refusedAt :: Integer} + deriving (Eq, Show) + +data Answer = Unrequested | Unanswered | Confirmed | Refused + deriving (Eq, Show) + +data Files = Files {launchingBytes, instanceBytes, failureBytes, outcomeBytes, requestBytes, terminatedBytes, killedBytes :: Maybe ByteString, refusedBytes :: [ByteString]} + +data State = Absent | Launched Launching | Started Instance | LaunchFailed (Maybe Instance) Failure | Ended Outcome + deriving (Eq, Show) + +data Observation = Observation {state :: State, asked :: Maybe ByteString, terminated :: Maybe Signalled, killed :: Maybe Signalled, refused :: [Refusal], alive :: Bool} + deriving (Eq, Show) + +launchingFile :: Launching -> ByteString +launchingFile = file . launchingValue + +instanceFile :: Instance -> ByteString +instanceFile = file . instanceValue + +failureFile :: Failure -> ByteString +failureFile = file . failureValue + +outcomeFile :: Outcome -> ByteString +outcomeFile = file . outcomeValue + +signalledFile :: Signalled -> ByteString +signalledFile = file . signalledValue + +requestFile :: Request -> ByteString +requestFile (Request chosen wrapperId workerId seconds) = file (object ["format" .= ("invar-remote-request-v1" :: Text), "nonce" .= chosen, "wrapper" .= wrapperId, "worker" .= workerId, "grace" .= seconds]) + +file :: Value -> ByteString +file = Lazy.toStrict . encode + +refusalFile :: Refusal -> ByteString +refusalFile = file . refusalValue + +-- | The request in `terminate.json` is admitted when it names this instance. The wrapper signals for an admitted +-- request, which then stays in place; it records a refused one with its own refusal and removes it, so that a request +-- naming another instance does not close the termination of this one. +admitRequest :: Instance -> ByteString -> Either Text Request +admitRequest chosen encoded = case decoded "invar-remote-request-v1" ["nonce", "wrapper", "worker", "grace"] parser encoded of + Left problem -> Left (Text.pack ("The termination request cannot be read: " ++ problem)) + Right value + | requested value == nonce (launching chosen) && requestedWrapper value == wrapper (launching chosen) && requestedWorker value == worker chosen -> Right value + | otherwise -> Left "The termination request names another instance" + where + parser fields = Request <$> fields .: "nonce" <*> fields .: "wrapper" <*> fields .: "worker" <*> fields .: "grace" + +-- | The answer to the request the instance holds. +answer :: Observation -> Answer +answer current = case asked current of + Nothing -> Unrequested + Just encoded + | isJust (terminated current) -> Confirmed + | refusedOf encoded current -> Refused + | otherwise -> Unanswered + +-- | The answer to the request with these bytes: refused when the wrapper refused them, the answer to the request the +-- instance holds when it holds these bytes, and otherwise unrequested, so a request never takes another's answer. +reply :: ByteString -> Observation -> Answer +reply encoded current + | refusedOf encoded current = Refused + | asked current == Just encoded = answer current + | otherwise = Unrequested + +refusedOf :: ByteString -> Observation -> Bool +refusedOf encoded current = any ((== decodeUtf8Lenient encoded) . refusedRequest) (refused current) + +answerName :: Answer -> Text +answerName value = case value of + Unrequested -> "unrequested" + Unanswered -> "unanswered" + Confirmed -> "confirmed" + Refused -> "refused" + +-- | Every control file, decoded and checked against each other. The files appear in a fixed order, and only a refused +-- request is removed, after its refusal, so a later file names exactly the identity of the earlier ones. +observe :: Bool -> Files -> Either String Observation +observe holding supplied = do + started <- traverse (decoded "invar-remote-launching-v1" launchingFields launchingParser) (launchingBytes supplied) + published <- traverse (decoded "invar-remote-instance-v1" instanceFields instanceParser) (instanceBytes supplied) + failure <- traverse (decoded "invar-remote-failure-v1" failureFields failureParser) (failureBytes supplied) + outcome <- traverse (decoded "invar-remote-outcome-v1" outcomeFields outcomeParser) (outcomeBytes supplied) + termination <- traverse (decoded "invar-remote-signalled-v1" signalledFields signalledParser) (terminatedBytes supplied) + escalation <- traverse (decoded "invar-remote-signalled-v1" signalledFields signalledParser) (killedBytes supplied) + refusals <- traverse (decoded "invar-remote-refused-v1" refusedFields refusalParser) (refusedBytes supplied) + when (isJust failure && isJust outcome) (Left "A remote instance records both a launch failure and an outcome") + current <- case started of + Nothing -> do + unless (isNothing published && isNothing failure && isNothing outcome && isNothing (requestBytes supplied) && isNothing termination && isNothing escalation && null refusals) (Left "Remote control files exist without a launch record") + pure Absent + Just first -> do + mapM_ (\value -> unless (launching value == first) (Left "The remote instance names another launch")) published + mapM_ (\value -> unless (failed value == first) (Left "The remote launch failure names another launch")) failure + case (published, failure, outcome) of + (_, Just value, _) -> pure (LaunchFailed published value) + (Just value, _, Just result) -> unless (ended result == value) (Left "The remote outcome names another instance") >> pure (Ended result) + (Nothing, _, Just _) -> Left "A remote outcome exists without its instance" + (Just value, _, Nothing) -> pure (Started value) + (Nothing, _, Nothing) -> pure (Launched first) + let named = case current of + Started value -> Just value + Ended result -> Just (ended result) + LaunchFailed value _ -> value + _ -> Nothing + signalled Terminate named termination + signalled Kill named escalation + when (isJust escalation && isNothing termination) (Left "A remote kill was recorded without its termination") + mapM_ (\_ -> answered named (requestBytes supplied) isRight) termination + mapM_ (\value -> unless (Just (held value) == named) (Left "A remote refusal names another instance") >> answered named (Just (encodeUtf8 (refusedRequest value))) isLeft) refusals + pure (Observation current (requestBytes supplied) termination escalation refusals holding) + where + signalled chosen named = mapM_ $ \value -> do + unless (Just (target value) == named) (Left "A remote signal record names another instance") + unless (signal value == chosen) (Left "A remote signal record holds another signal") + answered named request agrees = case (named, request) of + (Just chosen, Just encoded) -> unless (agrees (admitRequest chosen encoded)) (Left "A remote answer disagrees with its termination request") + _ -> Left "A remote answer exists without its instance and termination request" + +describe :: Observation -> Value +describe observation@(Observation current pending termination escalation refusals holding) = + object + [ "format" .= ("invar-remote-status-v1" :: Text) + , "state" .= stateName current + , "launching" .= fmap launchingValue (launched current) + , "instance" .= fmap instanceValue (instanceOf current) + , "failure" .= fmap failureValue (failureOf current) + , "outcome" .= fmap outcomeValue (outcomeOf current) + , "request" .= fmap decodeUtf8Lenient pending + , "terminated" .= fmap signalledValue termination + , "killed" .= fmap signalledValue escalation + , "refused" .= map refusalValue refusals + , "answer" .= answerName (answer observation) + , "wrapper_alive" .= holding + ] + where + stateName value = case value of + Absent -> "absent" :: Text + Launched _ -> "launching" + Started _ -> "started" + LaunchFailed _ _ -> "launch_failed" + Ended _ -> "ended" + launched value = case value of + Absent -> Nothing + Launched chosen -> Just chosen + Started chosen -> Just (launching chosen) + LaunchFailed _ chosen -> Just (failed chosen) + Ended chosen -> Just (launching (ended chosen)) + instanceOf value = case value of + Started chosen -> Just chosen + LaunchFailed chosen _ -> chosen + Ended chosen -> Just (ended chosen) + _ -> Nothing + failureOf value = case value of + LaunchFailed _ chosen -> Just chosen + _ -> Nothing + outcomeOf value = case value of + Ended chosen -> Just chosen + _ -> Nothing + +decoded :: Text -> [Key] -> (Object -> Parser value) -> ByteString -> Either String value +decoded format names parser encoded = Json.decode encoded >>= parseEither (withObject "remote control record" checked) + where + checked fields = do + Json.fields ("format" : names) fields + declared <- fields .: "format" + unless (declared == format) (fail ("Expected " <> show format)) + parser fields + +launchingFields, instanceFields, failureFields, outcomeFields, signalledFields, refusedFields :: [Key] +launchingFields = ["host", "nonce", "wrapper"] +instanceFields = launchingFields ++ ["worker", "group"] +failureFields = launchingFields ++ ["reason"] +outcomeFields = instanceFields ++ ["ending", "ended", "bytes", "sha256"] +signalledFields = instanceFields ++ ["signal", "at"] +refusedFields = instanceFields ++ ["request", "reason", "at"] + +launchingParser :: Object -> Parser Launching +launchingParser fields = Launching <$> fields .: "host" <*> fields .: "nonce" <*> fields .: "wrapper" + +instanceParser :: Object -> Parser Instance +instanceParser fields = do + chosen <- launchingParser fields + workerId <- fields .: "worker" + group <- fields .: "group" + unless (group == workerId) (fail "The worker must lead its own process group") + pure (Instance chosen workerId) + +failureParser :: Object -> Parser Failure +failureParser fields = Failure <$> launchingParser fields <*> fields .: "reason" + +outcomeParser :: Object -> Parser Outcome +outcomeParser fields = do + chosen <- instanceParser fields + result <- fields .: "ending" >>= withObject "remote ending" endingParser + sha <- fields .: "sha256" >>= Json.identity + Outcome chosen result <$> fields .: "ended" <*> fields .: "bytes" <*> pure sha + where + endingParser value = case Fields.toList value of + [("exited", code)] -> Exited <$> parseJSON code + [("signalled", number)] -> Signalled <$> parseJSON number + _ -> fail "A remote ending is either exited or signalled" + +signalledParser :: Object -> Parser Signalled +signalledParser fields = do + chosen <- instanceParser fields + named <- fields .: "signal" + sent <- case named :: Text of + "TERM" -> pure Terminate + "KILL" -> pure Kill + _ -> fail "Unknown remote signal" + Sent chosen sent <$> fields .: "at" + +refusalParser :: Object -> Parser Refusal +refusalParser fields = Refusal <$> instanceParser fields <*> fields .: "request" <*> fields .: "reason" <*> fields .: "at" + +refusalValue :: Refusal -> Value +refusalValue (Refusal chosen asking why at) = object (instancePairs chosen ++ ["format" .= ("invar-remote-refused-v1" :: Text), "request" .= asking, "reason" .= why, "at" .= at]) + +launchingValue :: Launching -> Value +launchingValue chosen = object (launchingPairs chosen ++ ["format" .= ("invar-remote-launching-v1" :: Text)]) + +launchingPairs :: Launching -> [Pair] +launchingPairs (Launching name chosen wrapperId) = ["host" .= name, "nonce" .= chosen, "wrapper" .= wrapperId] + +instanceValue :: Instance -> Value +instanceValue chosen = object (instancePairs chosen ++ ["format" .= ("invar-remote-instance-v1" :: Text)]) + +instancePairs :: Instance -> [Pair] +instancePairs (Instance chosen workerId) = launchingPairs chosen ++ ["worker" .= workerId, "group" .= workerId] + +failureValue :: Failure -> Value +failureValue (Failure chosen why) = object (launchingPairs chosen ++ ["format" .= ("invar-remote-failure-v1" :: Text), "reason" .= why]) + +outcomeValue :: Outcome -> Value +outcomeValue (Outcome chosen result at count sha) = + object (instancePairs chosen ++ ["format" .= ("invar-remote-outcome-v1" :: Text), "ending" .= endingValue result, "ended" .= at, "bytes" .= count, "sha256" .= sha]) + where + endingValue (Exited code) = object ["exited" .= code] + endingValue (Signalled number) = object ["signalled" .= number] + +signalledValue :: Signalled -> Value +signalledValue (Sent chosen sent at) = object (instancePairs chosen ++ ["format" .= ("invar-remote-signalled-v1" :: Text), "signal" .= signalName sent, "at" .= at]) + where + signalName Terminate = "TERM" :: Text + signalName Kill = "KILL" diff --git a/test/Main.hs b/test/Main.hs index abef52d..7ec1f61 100644 --- a/test/Main.hs +++ b/test/Main.hs @@ -44,6 +44,7 @@ import Properties (properties) import Quality (quality) import Records (records) import Recoveries (inspections, recoveries) +import Remotes (remotes) import Replays (replays) import Reports (reports) import ResidentCalls (residentCalls) @@ -78,5 +79,5 @@ import Workloads (workloads) main :: IO () main = do - outcomes <- traverse checkSequential [fixtures, properties, arguments, values, dependencies, evaluation, artifacts, construction, collections, records, literals, invocations, evidence, loads, store, checkpoints, inference, results, calls, batchCalls, groupedCalls, batchedProtocol, generation, traces, residentCalls, residentDrivers, residentObservations, rewards, cohorts, rollouts, sessions, learning, advantages, objectives, events, journals, replays, recoveries, inspections, restarts, comparisons, drivers, steps, tokenizers, algorithms, distillations, teachers, mismatches, updates, learnerCalls, loops, schedules, policies, probabilities, workloads, reports, quality, observations, inferenceObservations, numerical, scores, scoreProbes, numericalScores, numericalProbes, uses, useEvidence, useAdmission, useBernstein, usePlan, usePrepare, useExecution, useGeneric, statistics] + outcomes <- traverse checkSequential [fixtures, properties, arguments, values, dependencies, evaluation, artifacts, construction, collections, records, literals, invocations, evidence, loads, store, checkpoints, inference, results, calls, batchCalls, groupedCalls, batchedProtocol, generation, traces, residentCalls, residentDrivers, residentObservations, rewards, cohorts, rollouts, sessions, learning, advantages, objectives, events, journals, replays, recoveries, inspections, restarts, comparisons, drivers, steps, tokenizers, algorithms, distillations, teachers, mismatches, updates, learnerCalls, loops, schedules, policies, probabilities, workloads, reports, quality, observations, inferenceObservations, numerical, scores, scoreProbes, numericalScores, numericalProbes, uses, useEvidence, useAdmission, useBernstein, usePlan, usePrepare, useExecution, useGeneric, statistics, remotes] unless (and outcomes) exitFailure diff --git a/test/Remotes.hs b/test/Remotes.hs new file mode 100644 index 0000000..0261fa2 --- /dev/null +++ b/test/Remotes.hs @@ -0,0 +1,298 @@ +{-# LANGUAGE OverloadedStrings #-} + +module Remotes (remotes) where + +import Control.Concurrent (threadDelay) +import Control.Exception (IOException, try) +import Control.Monad (forM_, void) +import Crypto.Hash.SHA256 qualified as SHA256 +import Data.Aeson (Value (..), decodeStrict, encode, object, withObject, (.:), (.=)) +import Data.Aeson.KeyMap qualified as Fields +import Data.Aeson.Types (parseEither) +import Data.ByteString qualified as Bytes +import Data.ByteString.Char8 qualified as Char +import Data.ByteString.Lazy qualified as Lazy +import Data.Text (Text) +import Data.Text qualified as Text +import Data.Text.Encoding (decodeUtf8Lenient) +import Hedgehog +import Invar.Artifact qualified as Artifact +import Invar.Remote qualified as R +import Invar.Remote.Control qualified as C +import Numeric.Natural (Natural) +import System.Directory (doesFileExist, getTemporaryDirectory) +import System.Exit (ExitCode (..)) +import System.FilePath (()) +import System.IO (Handle, hClose, hPutStrLn) +import System.Posix.Signals (sigHUP, signalProcess) +import System.Posix.Temp (mkdtemp) +import System.Process (CreateProcess (..), ProcessHandle, StdStream (..), createProcess, proc, readProcessWithExitCode, waitForProcess) + +remotes :: Group +remotes = + Group + "Remote wrapper" + [ ("the worker starts only after its instance is published, and status reads absent, started, then ended", once barrier) + , ("a program that cannot start is a launch failure", once missing) + , ("stdout is forwarded unchanged, with a final line without a newline, and the outcome counts and digests it", once forwarded) + , ("after the reader goes away the wrapper keeps draining, survives, and counts every byte the worker wrote", once dropped) + , ("termination ends the worker's whole process group through the wrapper, which survives to record it, and a repeated request reports that confirmation", once grouped) + , ("a worker that ignores termination is killed after the grace", once escalated) + , ("a group member that ignores termination and holds stdout is killed after the leader exits", once orphaned) + , ("a hangup leaves the wrapper running and the worker finishes normally", once hangup) + , ("a request without a live wrapper is not written and signals nothing", once unconfirmed) + , ("a request naming another instance is refused and recorded, and a request naming this one still terminates it", once mismatched) + , ("control files are refused when they disagree, carry another format or field, or appear out of order", once refused) + ] + where + once = withTests 1 . property + +workspace :: PropertyT IO FilePath +workspace = evalIO (getTemporaryDirectory >>= \temporary -> mkdtemp (temporary "invar-remote.")) + +-- | Starts `invar remote run` with the worker's stdin and stdout as pipes; the environment names the instance as D. +launch :: FilePath -> String -> [String] -> PropertyT IO (Handle, Handle, ProcessHandle) +launch root nonce command = do + created <- evalIO (createProcess (proc "invar" (["remote", "run", "--instance", root, "--nonce", nonce, "--environment", "D=" ++ root, "--"] ++ command)) {std_in = CreatePipe, std_out = CreatePipe}) + case created of + (Just input, Just output, _, handle) -> pure (input, output, handle) + _ -> failure + +observed :: FilePath -> PropertyT IO C.Observation +observed root = evalIO (R.status root) >>= evalEither + +started :: FilePath -> PropertyT IO C.Instance +started root = do + waitUntil (doesFileExist (root "instance.json")) + current <- observed root + case C.state current of + C.Started chosen -> pure chosen + other -> annotateShow other >> failure + +waitUntil :: IO Bool -> PropertyT IO () +waitUntil holds = go (250 :: Int) + where + go remaining = do + done <- evalIO holds + if done then pure () else if remaining <= 0 then failure else evalIO (threadDelay 20000) >> go (remaining - 1) + +shell :: String -> [String] +shell script = ["/bin/sh", "-c", script] + +ended :: FilePath -> PropertyT IO C.Outcome +ended root = do + current <- observed root + C.alive current === False + case C.state current of + C.Ended outcome -> pure outcome + other -> annotateShow other >> failure + +barrier :: PropertyT IO () +barrier = do + root <- ( "instance") <$> workspace + before <- observed root + C.state before === C.Absent + (input, output, handle) <- launch root "barrier" (shell "test -f \"$D/instance.json\" && printf published; read line; printf ' %s' \"$line\"") + chosen <- started root + during <- observed root + C.alive during === True + evalIO (hPutStrLn input "released" >> hClose input) + evalIO (Bytes.hGetContents output) >>= (=== "published released") + evalIO (waitForProcess handle) >>= (=== ExitSuccess) + outcome <- ended root + C.ended outcome === chosen + C.ending outcome === C.Exited 0 + +missing :: PropertyT IO () +missing = do + root <- ( "instance") <$> workspace + (input, output, handle) <- launch root "missing" ["/nonexistent/program"] + evalIO (hClose input >> void (Bytes.hGetContents output)) + evalIO (waitForProcess handle) >>= (=== ExitFailure 125) + current <- observed root + case C.state current of + C.LaunchFailed (Just _) _ -> success + other -> annotateShow other >> failure + +forwarded :: PropertyT IO () +forwarded = do + root <- ( "instance") <$> workspace + (input, output, handle) <- launch root "forward" (shell "printf 'first\\nsecond\\nthird'; exit 3") + evalIO (hClose input) + received <- evalIO (Bytes.hGetContents output) + received === "first\nsecond\nthird" + evalIO (waitForProcess handle) >>= (=== ExitFailure 3) + outcome <- ended root + C.ending outcome === C.Exited 3 + (C.bytes outcome, C.digest outcome) === (fromIntegral (Bytes.length received), Artifact.hex (SHA256.hash received)) + +dropped :: PropertyT IO () +dropped = do + root <- ( "instance") <$> workspace + let size = 1048576 :: Int + (input, output, handle) <- launch root "dropped" (shell ("head -c " ++ show size ++ " /dev/zero")) + evalIO (hClose input) + first <- evalIO (Bytes.hGetSome output 16) + assert (not (Bytes.null first)) + evalIO (hClose output) + evalIO (waitForProcess handle) >>= (=== ExitSuccess) + outcome <- ended root + (C.bytes outcome, C.digest outcome) === (fromIntegral size, Artifact.hex (SHA256.hash (Bytes.replicate size 0))) + +-- | Runs `invar remote terminate` for an instance's nonce and process ids, and returns whether it wrote the request +-- and the answer to it. +terminated :: FilePath -> C.Instance -> Int -> PropertyT IO (Bool, String) +terminated root chosen = requestedFor root (C.nonce (C.launching chosen)) (C.wrapper (C.launching chosen)) (C.worker chosen) + +requestedFor :: FilePath -> Text -> Natural -> Natural -> Int -> PropertyT IO (Bool, String) +requestedFor root chosen wrapperId workerId grace = do + let arguments = ["remote", "terminate", "--instance", root, "--nonce", Text.unpack chosen, "--wrapper", show wrapperId, "--worker", show workerId, "--grace", show grace, "--wait", "5"] + (code, printed, _) <- evalIO (readProcessWithExitCode "invar" arguments "") + code === ExitSuccess + printedValue <- maybe failure pure (decodeStrict (Char.pack printed)) + evalEither (parseEither (withObject "terminate answer" (\fields -> (,) <$> fields .: "requested" <*> fields .: "answer")) printedValue) + +gone :: Int -> IO Bool +gone pid = either (\(_ :: IOException) -> True) (const False) <$> try (signalProcess 0 (fromIntegral pid)) + +grouped :: PropertyT IO () +grouped = do + root <- ( "instance") <$> workspace + (input, _, handle) <- launch root "group" (shell "sleep 300 & echo $! > \"$D.grandchild\"; wait") + chosen <- started root + waitUntil (doesFileExist (root ++ ".grandchild")) + grandchild <- evalIO (read . Char.unpack . Char.strip <$> Bytes.readFile (root ++ ".grandchild")) + terminated root chosen 1 >>= (=== (True, "confirmed")) + asked <- evalIO (Bytes.readFile (root "terminate.json")) + terminated root chosen 1 >>= (=== (False, "confirmed")) + requestedFor root (C.nonce (C.launching chosen)) (C.wrapper (C.launching chosen)) (C.worker chosen + 1) 1 >>= (=== (False, "unrequested")) + evalIO (Bytes.readFile (root "terminate.json")) >>= (=== asked) + evalIO (hClose input) + void (evalIO (waitForProcess handle)) + waitUntil (gone grandchild) + outcome <- ended root + C.ending outcome === C.Signalled 15 + current <- observed root + fmap C.signal (C.terminated current) === Just C.Terminate + +escalated :: PropertyT IO () +escalated = do + root <- ( "instance") <$> workspace + (input, _, handle) <- launch root "group" (shell "trap '' TERM; while :; do sleep 1; done") + chosen <- started root + terminated root chosen 1 >>= (=== (True, "confirmed")) + evalIO (hClose input) + void (evalIO (waitForProcess handle)) + outcome <- ended root + C.ending outcome === C.Signalled 9 + current <- observed root + fmap C.signal (C.killed current) === Just C.Kill + +orphaned :: PropertyT IO () +orphaned = do + root <- ( "instance") <$> workspace + (input, output, handle) <- launch root "orphan" (shell "( trap '' TERM; while :; do sleep 1; done ) & echo $! > \"$D.grandchild\"; wait") + chosen <- started root + waitUntil (doesFileExist (root ++ ".grandchild")) + grandchild <- evalIO (read . Char.unpack . Char.strip <$> Bytes.readFile (root ++ ".grandchild")) + terminated root chosen 1 >>= (=== (True, "confirmed")) + evalIO (hClose input >> void (Bytes.hGetContents output) >> void (waitForProcess handle)) + waitUntil (gone grandchild) + outcome <- ended root + C.ending outcome === C.Signalled 15 + current <- observed root + fmap C.signal (C.killed current) === Just C.Kill + +hangup :: PropertyT IO () +hangup = do + root <- ( "instance") <$> workspace + (input, output, handle) <- launch root "hangup" (shell "read line; printf '%s' \"$line\"") + chosen <- started root + evalIO (signalProcess sigHUP (fromIntegral (C.wrapper (C.launching chosen)))) + evalIO (threadDelay 200000) + during <- observed root + C.alive during === True + evalIO (hPutStrLn input "finished" >> hClose input) + evalIO (Bytes.hGetContents output) >>= (=== "finished") + evalIO (waitForProcess handle) >>= (=== ExitSuccess) + outcome <- ended root + C.ending outcome === C.Exited 0 + +unconfirmed :: PropertyT IO () +unconfirmed = do + root <- ( "instance") <$> workspace + (input, output, handle) <- launch root "group" (shell "read line") + chosen <- started root + evalIO (hClose input >> void (Bytes.hGetContents output) >> void (waitForProcess handle)) + terminated root chosen 1 >>= (=== (False, "unrequested")) + evalIO (doesFileExist (root "terminate.json")) >>= (=== False) + +mismatched :: PropertyT IO () +mismatched = do + root <- ( "instance") <$> workspace + (input, output, handle) <- launch root "group" (shell "read line") + chosen <- started root + requestedFor root (C.nonce (C.launching chosen)) (C.wrapper (C.launching chosen)) (C.worker chosen + 1) 1 >>= (=== (True, "refused")) + terminated root chosen 1 >>= (=== (True, "confirmed")) + evalIO (hClose input >> void (Bytes.hGetContents output) >> void (waitForProcess handle)) + outcome <- ended root + C.ending outcome === C.Signalled 15 + current <- observed root + map C.held (C.refused current) === [chosen] + fmap (C.admitRequest chosen) (C.asked current) === Just (Right (C.Request (C.nonce (C.launching chosen)) (C.wrapper (C.launching chosen)) (C.worker chosen) 1)) + +refused :: PropertyT IO () +refused = do + let launched = C.Launching "host" "nonce" 10 + other = C.Launching "host" "other" 10 + chosen = C.Instance launched 11 + outcome = C.Outcome chosen (C.Exited 0) 1 0 (Artifact.hex (SHA256.hash "")) + sent signal = C.Sent chosen signal 1 + files = C.Files Nothing Nothing Nothing Nothing Nothing Nothing Nothing [] + complete = files {C.launchingBytes = Just (C.launchingFile launched), C.instanceBytes = Just (C.instanceFile chosen), C.outcomeBytes = Just (C.outcomeFile outcome)} + asked = C.requestFile (C.Request "nonce" 10 11 2) + stray = C.requestFile (C.Request "nonce" 10 12 2) + refusal encoded = C.refusalFile (C.Refusal chosen (decodeUtf8Lenient encoded) "The termination request names another instance" 1) + confirmed = complete {C.requestBytes = Just asked, C.terminatedBytes = Just (C.signalledFile (sent C.Terminate))} + declined = complete {C.refusedBytes = [refusal stray]} + changed key value encoded = case decodeStrict encoded of + Just (Object fields) -> Lazy.toStrict (encode (Object (Fields.insert key value fields))) + _ -> encoded + dropped' key encoded = case decodeStrict encoded of + Just (Object fields) -> Lazy.toStrict (encode (Object (Fields.delete key fields))) + _ -> encoded + fmap C.state (C.observe False complete) === Right (C.Ended outcome) + fmap C.state (C.observe True files {C.launchingBytes = Just (C.launchingFile launched)}) === Right (C.Launched launched) + fmap C.answer (C.observe False confirmed) === Right C.Confirmed + fmap (\current -> map (`C.reply` current) [asked, stray, C.requestFile (C.Request "other" 10 11 2)]) (C.observe False confirmed) === Right [C.Confirmed, C.Unrequested, C.Unrequested] + fmap C.answer (C.observe False declined) === Right C.Unrequested + fmap (C.reply stray) (C.observe False declined) === Right C.Refused + fmap (\current -> (C.answer current, C.reply asked current)) (C.observe False declined {C.requestBytes = Just stray}) === Right (C.Refused, C.Unrequested) + fmap (\current -> (C.answer current, C.reply stray current)) (C.observe False confirmed {C.refusedBytes = [refusal stray]}) === Right (C.Confirmed, C.Refused) + fmap C.answer (C.observe False complete {C.requestBytes = Just asked}) === Right C.Unanswered + C.admitRequest chosen asked === Right (C.Request "nonce" 10 11 2) + assert (either (const True) (const False) (C.admitRequest chosen stray)) + assert (either (const True) (const False) (C.admitRequest chosen "{}")) + forM_ + [ complete {C.instanceBytes = Nothing} + , complete {C.launchingBytes = Nothing} + , complete {C.instanceBytes = Just (C.instanceFile (C.Instance other 11))} + , complete {C.failureBytes = Just (C.failureFile (C.Failure launched "failed"))} + , complete {C.killedBytes = Just (C.signalledFile (sent C.Kill))} + , complete {C.terminatedBytes = Just (C.signalledFile (sent C.Kill))} + , complete {C.terminatedBytes = Just (C.signalledFile (C.Sent (C.Instance launched 12) C.Terminate 1))} + , complete {C.instanceBytes = Just (changed "group" (Number 12) (C.instanceFile chosen))} + , complete {C.instanceBytes = Just (changed "format" (String "invar-remote-launching-v1") (C.instanceFile chosen))} + , complete {C.instanceBytes = Just (changed "extra" Null (C.instanceFile chosen))} + , complete {C.outcomeBytes = Just (dropped' "sha256" (C.outcomeFile outcome))} + , complete {C.outcomeBytes = Just (changed "ending" (object ["exited" .= (0 :: Int), "signalled" .= (9 :: Int)]) (C.outcomeFile outcome))} + , confirmed {C.requestBytes = Nothing} + , confirmed {C.requestBytes = Just stray} + , declined {C.refusedBytes = [refusal asked]} + , declined {C.refusedBytes = [C.refusalFile (C.Refusal (C.Instance launched 12) (decodeUtf8Lenient stray) "The termination request names another instance" 1)]} + , confirmed {C.refusedBytes = [refusal asked]} + , files {C.refusedBytes = [refusal stray]} + ] + $ \supplied -> case C.observe False supplied of + Left _ -> success + Right accepted -> annotateShow accepted >> failure