tricu

An interpreted language for exploring Tree Calculus
Log | Files | Refs | README | LICENSE

Process.hs (8657B)

      1 {-# LANGUAGE ScopedTypeVariables #-}
      2 -- Direct-child processes, supervised by the existing structured task machinery.
      3 -- No shell expansion, detached waiter, implicit output collection or job control.
      4 module IOV2.Process (Spec(..), launch, validSpec, TerminationOps(..), terminateWith, cleanupAll) where
      5 
      6 import qualified IOV2.Resources as R
      7 import qualified IOV2.Tasks as S
      8 import qualified IOV2.Clock as Clock
      9 import Runtime (T(..), ofList, ofString, ofNumber)
     10 import Control.Concurrent (threadDelay)
     11 import Control.Concurrent.STM
     12 import Control.Exception
     13 import Control.Monad
     14 import Data.IORef
     15 import Data.List (nub)
     16 import qualified Data.ByteString as BS
     17 import System.Exit
     18 import System.IO
     19 import qualified System.Process as P
     20 import System.Posix.Signals (signalProcess, sigKILL)
     21 
     22 -- Nothing explicitly selects inherited cwd/environment. Just [] is an empty env.
     23 -- Limits count Unicode scalars, not OS-encoded bytes; exec may impose smaller limits.
     24 data Spec = Spec FilePath [String] (Maybe FilePath) (Maybe [(String,String)]) Integer
     25   deriving Show
     26 
     27 validSpec :: Spec -> Bool
     28 validSpec (Spec executable arguments directory environment grace) =
     29   not (null executable) && maybe True (not . null) directory &&
     30   length arguments <= 256 && length variables <= 256 &&
     31   all validText strings && sum (map length strings) <= 65536 &&
     32   all (\(key,_) -> not (null key) && '=' `notElem` key) variables &&
     33   length (nub (map fst variables)) == length variables &&
     34   grace >= 0 && grace <= 9223372036854775807
     35   where
     36     variables = maybe [] id environment
     37     strings = executable : arguments ++ maybe [] (:[]) directory ++ concatMap (\(k,v) -> [k,v]) variables
     38     validText s = length s <= 4096 && all (\c -> c /= '\0' && not (c >= '\xD800' && c <= '\xDFFF')) s
     39 
     40 -- Returns one ordinary task/join reference and parent-owned stdin/stdout/stderr.
     41 -- The private process handle is owned by the monitor child, never source-visible.
     42 -- Launch failures consume the internal join ticket before returning an error.
     43 launch :: R.Runtime -> S.Group -> Spec -> IO (Either R.Problem (S.TaskRef, [R.Token]))
     44 launch runtime parent spec
     45   | not (validSpec spec) = pure (Left (R.Problem "invalid-argument" "process/spawn" "" "none"))
     46   | otherwise = mask $ \restore -> do
     47       ready <- newEmptyTMVarIO
     48       spawned <- S.spawn parent False True [] (\child _ -> monitor ready child)
     49       case spawned of
     50         Left problem -> pure (Left problem)
     51         Right task -> do
     52           result <- restore (atomically (takeTMVar ready))
     53           case result of
     54             Right pipes -> pure (Right (task,pipes))
     55             Left problem -> do
     56               -- Failure is not acknowledged until the launched process is reaped.
     57               settled <- restore (S.join parent task)
     58               case settled of
     59                 Right (S.Completed _) -> pure (Left problem)
     60                 _ -> S.checkpoint parent >> R.resourceFault problem
     61   where
     62     Spec executable arguments directory environment grace = spec
     63     monitor ready child = do
     64       slot <- newIORef Nothing
     65       acquired <- R.acquire runtime (S.scopeOf child) $ do
     66         (input,output,errors,process) <- P.createProcess (P.proc executable arguments)
     67           { P.cwd = directory, P.env = environment, P.std_in = P.CreatePipe
     68           , P.std_out = P.CreatePipe, P.std_err = P.CreatePipe, P.close_fds = True }
     69         let handles = [h | Just h <- [input,output,errors]]
     70         raw <- newIORef handles
     71         let release = cleanupAll [releaseRaw raw, terminateAndReap process grace]
     72         writeIORef slot (Just (process,raw,handles))
     73         pure (R.Internal release)
     74       case acquired of
     75         Left problem -> atomically (putTMVar ready (Left problem)) >> pure (S.Completed Leaf)
     76         Right _ -> do
     77           Just (process,raw,handles) <- readIORef slot
     78           case handles of
     79             [input,output,errors] -> do
     80               pipes <- install raw [] [(True,input),(False,output),(False,errors)]
     81               case pipes of
     82                 Left problem -> atomically (putTMVar ready (Left problem)) >> pure (S.Completed Leaf)
     83                 Right tokens -> do
     84                   atomically (putTMVar ready (Right tokens))
     85                   -- This wait is body work, not a protected release callback.
     86                   status <- awaitExit process
     87                   pure (S.Completed (exitTree status))
     88             _ -> ioError (userError "process backend omitted requested pipes")
     89     install _ installed [] = pure (Right (reverse installed))
     90     install raw installed ((writing,h):rest) = do
     91       result <- R.acquire runtime (S.scopeOf parent) $ do
     92         hSetBinaryMode h True
     93         hSetBuffering h NoBuffering
     94         poisoned <- newIORef False
     95         let guarded action = mask $ \restore -> do
     96               failed <- readIORef poisoned
     97               if failed then R.resourceFault (R.Problem "invalid-state" "process/pipe" "prior operation failed; progress is uncertain" "unknown")
     98                 else restore action `onException` writeIORef poisoned True
     99         -- Publication is masked by acquire; thereafter registry release owns h.
    100         modifyIORef' raw (filter (/= h))
    101         pure $ if writing
    102           then R.Writer (\bytes -> guarded (BS.hPut h bytes >> pure (BS.length bytes))) (hClose h)
    103           else R.Reader (\count -> guarded (BS.hGetSome h count)) (hClose h)
    104       case result of
    105         Right token -> install raw (token:installed) rest
    106         Left problem -> do
    107           cleanupAll [R.closeResource runtime (S.scopeOf parent) token >>= either R.resourceFault pure | token <- installed]
    108           pure (Left problem)
    109 
    110 exitTree :: ExitCode -> T
    111 exitTree ExitSuccess = ofList [ofString "exit", ofNumber 0]
    112 exitTree (ExitFailure code)
    113   | code < 0 = ofList [ofString "signal", ofNumber (negate (toInteger code))]
    114   | otherwise = ofList [ofString "exit", ofNumber (toInteger code)]
    115 
    116 releaseRaw :: IORef [Handle] -> IO ()
    117 releaseRaw raw = atomicModifyIORef' raw (\handles -> ([],handles)) >>= cleanupAll . map hClose
    118 
    119 -- Attempt all releases, preserving the first exception. Resource settlement
    120 -- reports failure rather than silently acknowledging a partial cleanup.
    121 cleanupAll :: [IO ()] -> IO ()
    122 cleanupAll actions = do
    123   outcomes <- mapM try actions
    124   case [e | Left (e :: SomeException) <- outcomes] of
    125     [] -> pure ()
    126     [e] -> throwIO e
    127     e:rest -> R.resourceFault (R.Problem "cleanup-failed" "process/release"
    128       (take 240 (displayException e) ++ "; additional cleanup errors omitted: " ++ show (length rest)) "unknown")
    129 
    130 -- Injectable boundary for independent escalation/error-order tests. The owner
    131 -- must settle the sole wait body before entering this release protocol.
    132 data TerminationOps = TerminationOps
    133   { checkReaped :: IO Bool, requestTerminate :: IO (), forceTerminate :: IO ()
    134   , reap :: IO (), clock :: IO Integer, pause :: IO () }
    135 
    136 terminateWith :: TerminationOps -> Integer -> IO ()
    137 terminateWith ops grace = do
    138   exited <- checkReaped ops
    139   unless exited $ do
    140     start <- clock ops
    141     let wait remaining previous = do
    142           done <- checkReaped ops
    143           unless done $ do
    144             now <- clock ops
    145             -- Host samples are Word64. Account for wrap between nearby polls,
    146             -- rather than constructing an unreachable absolute deadline.
    147             let elapsed = (now - previous) `mod` 18446744073709551616
    148             if elapsed < remaining then pause ops >> wait (remaining-elapsed) now
    149               else cleanupAll [forceTerminate ops, reap ops]
    150     cleanupAll [requestTerminate ops, wait (grace*1000000) start]
    151 
    152 terminateAndReap :: P.ProcessHandle -> Integer -> IO ()
    153 terminateAndReap process = terminateWith TerminationOps
    154   { checkReaped = maybe False (const True) <$> P.getProcessExitCode process
    155   , requestTerminate = P.terminateProcess process
    156   , forceTerminate = do
    157       -- No other reaper remains: an unreaped child retains its PID. Never
    158       -- expose this handle for concurrent public wait/getPid operations.
    159       pid <- P.getPid process
    160       case pid of Just identity -> signalProcess sigKILL identity; Nothing -> pure ()
    161   , reap = void (awaitExit process)
    162   , clock = Clock.monotonicNanoseconds
    163   , pause = threadDelay 10000 }
    164 
    165 -- waitForProcess may block an RTS capability in a foreign wait (and cannot be
    166 -- relied on as a cancellation checkpoint). WNOHANG polling keeps the sole reaper
    167 -- in the task body and is interruptible without a detached host waiter.
    168 awaitExit :: P.ProcessHandle -> IO ExitCode
    169 awaitExit process = do
    170   status <- P.getProcessExitCode process
    171   case status of
    172     Just result -> pure result
    173     Nothing -> threadDelay 10000 >> awaitExit process