commit 02d44b105f80ed095fb1582fca963c1e7fef9f6b
parent dfd70d677a6c7ad941235a8ac4bb6199629ef5c1
Author: James Eversole <james@eversole.co>
Date: Wed, 14 Oct 2026 20:14:40 -0500
Scope asynchronous workers and sockets to their owning IO run
Diffstat:
3 files changed, 162 insertions(+), 39 deletions(-)
diff --git a/docs/host-abi.md b/docs/host-abi.md
@@ -259,3 +259,19 @@ Unrestricted read/write flags explicitly bypass the corresponding checks.
This is not a race-free filesystem sandbox: checking a pathname and subsequently
opening it are separate operations. Use trusted directories and do not allow
untrusted concurrent mutation of directory entries or symlinks.
+
+## Haskell IO run ownership
+
+A host IO run owns its asynchronous OS workers and open socket registry. Root
+completion and `abortIO` end the run; they do not leave detached workers running.
+Normal return, failure and external cancellation cancel and join outstanding OS
+workers before closing remaining registered sockets. Explicit socket close still
+invalidates that handle immediately. Task handles and socket handles are local
+to the run and cannot transfer resource ownership to another invocation.
+
+Cancellation does not execute pending tricu continuations or language-level
+cleanup actions, and it cannot roll back completed effects or partial writes.
+Worker cancellation waits for host cleanup; an uninterruptible OS operation can
+delay termination beyond a requested timeout. This is not a hard process-kill
+deadline. Socket operations report operational IO errors as Results, but host
+asynchronous cancellation is not converted into an ordinary Result.
diff --git a/src/IODriver.hs b/src/IODriver.hs
@@ -23,7 +23,8 @@ import System.Posix.Files (createLink)
import System.Posix.IO (openFd, closeFd, defaultFileFlags, OpenMode(ReadOnly))
import System.Posix.Unistd (fileSynchronise)
import qualified System.IO as IO
-import Control.Exception (try, catch, bracket, evaluate, IOException, SomeException)
+import Control.Exception (try, catch, bracket, bracketOnError, evaluate, IOException, SomeException,
+ SomeAsyncException, fromException, throwIO, mask_, finally)
import System.IO.Error (isDoesNotExistError, isPermissionError, isAlreadyExistsError)
import Data.List (isPrefixOf, isInfixOf)
import System.FilePath (addTrailingPathSeparator, takeDirectory)
@@ -40,7 +41,7 @@ import Data.Map.Strict (Map)
import qualified Data.Sequence as Seq
import Data.Sequence (Seq, (|>), ViewL(..))
import Data.Time.Clock (UTCTime, getCurrentTime, addUTCTime, diffUTCTime)
-import Control.Concurrent (threadDelay, forkIO)
+import Control.Concurrent (threadDelay, forkIOWithUnmask, killThread, ThreadId, MVar, newEmptyMVar, putMVar, readMVar)
import Control.Concurrent.STM (TVar, newTVarIO, atomically, readTVar, writeTVar, modifyTVar', retry)
import qualified Data.Set as Set
import Data.Set (Set)
@@ -207,6 +208,41 @@ data SocketRegistry = SocketRegistry
, sockNextId :: Integer
}
+-- Call with async exceptions masked from acquisition through registration.
+registerSocket :: TVar SocketRegistry -> NS.Socket -> IO SockId
+registerSocket registry sock = atomically $ do
+ SocketRegistry sockets next <- readTVar registry
+ let sid = SockId next
+ writeTVar registry (SocketRegistry (Map.insert sid sock sockets) (next + 1))
+ pure sid
+
+-- All OS workers belong to one run. Completed workers unregister immediately;
+-- cleanup cancels and joins the remaining workers before closing their sockets.
+type Workers = TVar (Map ThreadId (MVar ()))
+
+startWorker :: Workers -> IO () -> IO ()
+startWorker workers action = mask_ $ do
+ start <- newEmptyMVar
+ done <- newEmptyMVar
+ tid <- forkIOWithUnmask $ \unmask -> do
+ self <- readMVar start
+ unmask action `finally` do
+ atomically $ modifyTVar' workers (Map.delete self)
+ putMVar done ()
+ atomically $ modifyTVar' workers (Map.insert tid done)
+ putMVar start tid
+
+cleanupRun :: (TVar (Map TaskId T), TVar SocketRegistry, Workers) -> IO ()
+cleanupRun (_, sockets, workers) = do
+ pending <- atomically $ readTVar workers
+ mapM_ killThread (Map.keys pending)
+ mapM_ readMVar (Map.elems pending)
+ owned <- atomically $ do
+ SocketRegistry owned next <- readTVar sockets
+ writeTVar sockets (SocketRegistry Map.empty next)
+ pure owned
+ mapM_ (\sock -> NS.close sock `catch` (\(_ :: IOException) -> pure ())) owned
+
-- ---------------------------------------------------------------------------
-- Free-monad action AST
-- ---------------------------------------------------------------------------
@@ -802,21 +838,17 @@ stepMachine sockVar machine =
_ ->
finishValue machine invalidSleepResult
- ASocket -> do
- result <- try (NS.socket NS.AF_INET NS.Stream NS.defaultProtocol) :: IO (Either SomeException NS.Socket)
- case result of
- Left e ->
- finishValue machine (errResult ("io error: " ++ show e))
- Right sock -> do
+ ASocket -> mask_ $ do
+ -- Acquisition through registry insertion is one ownership transfer.
+ result <- try $ bracketOnError
+ (NS.socket NS.AF_INET NS.Stream NS.defaultProtocol) NS.close $ \sock -> do
NS.setSocketOption sock NS.ReuseAddr 1
- sid <- atomically $ do
- SocketRegistry m next <- readTVar sockVar
- let sid = SockId next
- writeTVar sockVar (SocketRegistry (Map.insert sid sock m) (next + 1))
- return sid
- finishValue machine (okResult (sockHandle sid))
+ registerSocket sockVar sock
+ case result of
+ Left (e :: IOException) -> finishValue machine (errResult ("io error: " ++ show e))
+ Right sid -> finishValue machine (okResult (sockHandle sid))
- ACloseSocket sockTree ->
+ ACloseSocket sockTree -> mask_ $
case decodeSockHandle sockTree of
Left _ -> finishValue machine invalidSocketHandleResult
Right sid -> do
@@ -855,7 +887,7 @@ stepMachine sockVar machine =
(Just (show port))
let serverAddr = head addrInfo
NS.bind sock (NS.addrAddress serverAddr)
- ) :: IO (Either SomeException ())
+ ) :: IO (Either IOException ())
case result of
Left e ->
finishValue machine (errResult ("io error: " ++ show e))
@@ -875,7 +907,7 @@ stepMachine sockVar machine =
case mSock of
Nothing -> finishValue machine invalidSocketHandleResult
Just sock -> do
- result <- try (NS.listen sock (fromIntegral backlog)) :: IO (Either SomeException ())
+ result <- try (NS.listen sock (fromIntegral backlog)) :: IO (Either IOException ())
case result of
Left e ->
finishValue machine (errResult ("io error: " ++ show e))
@@ -886,23 +918,19 @@ stepMachine sockVar machine =
case decodeSockHandle listenTree of
Left _ -> finishValue machine invalidSocketHandleResult
Right listenSid ->
- pure (AsyncAction (do
+ pure (AsyncAction (mask_ $ do
mListenSock <- atomically $ do
SocketRegistry m _ <- readTVar sockVar
return (Map.lookup listenSid m)
case mListenSock of
Nothing -> return (errResult "invalid socket handle")
Just listenSock -> do
- result <- try (NS.accept listenSock) :: IO (Either SomeException (NS.Socket, NS.SockAddr))
+ result <- try (NS.accept listenSock) :: IO (Either IOException (NS.Socket, NS.SockAddr))
case result of
Left e ->
return (errResult ("io error: " ++ show e))
Right (clientSock, addr) -> do
- clientSid <- atomically $ do
- SocketRegistry m next <- readTVar sockVar
- let sid = SockId next
- writeTVar sockVar (SocketRegistry (Map.insert sid clientSock m) (next + 1))
- return sid
+ clientSid <- registerSocket sockVar clientSock
let addrStr = case addr of
NS.SockAddrInet p h ->
let (a,b,c,d) = NS.hostAddressToTuple h
@@ -934,7 +962,7 @@ stepMachine sockVar machine =
(Just (show port))
let serverAddr = head addrInfo
NS.connect sock (NS.addrAddress serverAddr)
- ) :: IO (Either SomeException ())
+ ) :: IO (Either IOException ())
case result of
Left e ->
return (errResult ("io error: " ++ show e))
@@ -956,7 +984,7 @@ stepMachine sockVar machine =
Nothing -> finishValue machine invalidSocketHandleResult
Just sock ->
pure (AsyncAction (do
- result <- try (NSB.recv sock (fromIntegral maxBytes)) :: IO (Either SomeException BS.ByteString)
+ result <- try (NSB.recv sock (fromIntegral maxBytes)) :: IO (Either IOException BS.ByteString)
case result of
Left e ->
return (errResult ("io error: " ++ show e))
@@ -995,7 +1023,7 @@ stepMachine sockVar machine =
Nothing -> finishValue machine invalidSocketHandleResult
Just sock ->
pure (AsyncAction (do
- result <- try (NSB.send sock bs) :: IO (Either SomeException Int)
+ result <- try (NSB.send sock bs) :: IO (Either IOException Int)
case result of
Left e ->
return (errResult ("io error: " ++ show e))
@@ -1288,7 +1316,7 @@ data Scheduler = Scheduler
, schedulerAsyncCompleted :: TVar (Map TaskId T)
, schedulerCompleted :: Map TaskId (T, T)
, schedulerSockets :: TVar SocketRegistry
- , schedulerNextSockId :: Integer
+ , schedulerWorkers :: Workers
, schedulerFailure :: Maybe String
, schedulerYieldHook :: YieldStats -> IO ()
}
@@ -1302,7 +1330,6 @@ instance Show Scheduler where
++ ", schedulerAsyncCompleted = <tvar>"
++ ", schedulerCompleted = " ++ show (schedulerCompleted s)
++ ", schedulerSockets = <tvar>"
- ++ ", schedulerNextSockId = " ++ show (schedulerNextSockId s)
++ ", schedulerFailure = " ++ show (schedulerFailure s)
++ " }"
@@ -1313,8 +1340,8 @@ data YieldStats = YieldStats
, yieldTasks :: !Int
} deriving (Eq, Show)
-initialScheduler :: (YieldStats -> IO ()) -> TVar (Map TaskId T) -> TVar SocketRegistry -> Machine -> Scheduler
-initialScheduler yieldHook asyncVar sockVar mainMachine =
+initialScheduler :: (YieldStats -> IO ()) -> TVar (Map TaskId T) -> TVar SocketRegistry -> Workers -> Machine -> Scheduler
+initialScheduler yieldHook asyncVar sockVar workers mainMachine =
Scheduler
{ schedulerNextTaskId = 1
, schedulerRunnable = Seq.singleton (TaskId 0)
@@ -1324,7 +1351,7 @@ initialScheduler yieldHook asyncVar sockVar mainMachine =
, schedulerAsyncCompleted = asyncVar
, schedulerCompleted = Map.empty
, schedulerSockets = sockVar
- , schedulerNextSockId = 0
+ , schedulerWorkers = workers
, schedulerFailure = Nothing
, schedulerYieldHook = yieldHook
}
@@ -1492,8 +1519,11 @@ handleStep _ (ExecutionFailed message) scheduler =
pure scheduler { schedulerFailure = Just message }
handleStep taskId (AsyncAction ioAction machine) scheduler = do
- _ <- forkIO $ do
- result <- (Right <$> ioAction) `catch` \(e :: SomeException) -> pure (Left (show e))
+ startWorker (schedulerWorkers scheduler) $ do
+ result <- (Right <$> ioAction) `catch` \(e :: SomeException) ->
+ case fromException e :: Maybe SomeAsyncException of
+ Just _ -> throwIO e
+ Nothing -> pure (Left (show e))
atomically $ modifyTVar' (schedulerAsyncCompleted scheduler) (Map.insert taskId $
case result of
Right val -> val
@@ -1584,9 +1614,7 @@ runIOWithYieldStatsHook :: (YieldStats -> IO ()) -> IOPermissions -> T -> T -> T
runIOWithYieldStatsHook yieldHook perms env initialState action =
case checkIOSentinel action of
Left err -> pure (Left err)
- Right (_, action') -> do
- asyncVar <- newTVarIO Map.empty
- sockVar <- newTVarIO (SocketRegistry Map.empty 0)
+ Right (_, action') -> bracket acquire cleanupRun $ \(asyncVar, sockVar, workers) -> do
let initialMachine = Machine
{ machineRuntime = Runtime
{ rtPerms = perms
@@ -1596,7 +1624,14 @@ runIOWithYieldStatsHook yieldHook perms env initialState action =
, machineCurrent = action'
, machineFrames = []
}
- runScheduler (initialScheduler yieldHook asyncVar sockVar initialMachine)
+ runScheduler (initialScheduler yieldHook asyncVar sockVar workers initialMachine)
+
+ where
+ acquire = do
+ asyncVar <- newTVarIO Map.empty
+ sockVar <- newTVarIO (SocketRegistry Map.empty 0)
+ workers <- newTVarIO Map.empty
+ pure (asyncVar, sockVar, workers)
runIOWithEnv :: IOPermissions -> T -> T -> IO (Either String T)
runIOWithEnv perms env action = do
diff --git a/test/IOBoundaryTests.hs b/test/IOBoundaryTests.hs
@@ -2,9 +2,16 @@ module IOBoundaryTests (ioBoundaryTests) where
import IODriver
import Runtime
+import Control.Exception (bracket, try, IOException)
+import Control.Concurrent (threadDelay)
+import System.Posix.Files (createNamedPipe)
+import System.Posix.IO (openFd, closeFd, defaultFileFlags, OpenMode(ReadOnly), OpenFileFlags(..), fdRead)
import qualified Data.ByteString as BS
+import qualified Network.Socket as NS
import System.Directory (createDirectory, createDirectoryLink, doesFileExist)
+import System.IO.Error (isEOFError, isFullError, isAlreadyInUseError)
import System.IO.Temp (withSystemTempDirectory)
+import qualified System.Timeout as Timeout
import Test.Tasty
import Test.Tasty.HUnit
@@ -29,10 +36,75 @@ ioBoundaryTests = testGroup "IO ownership and path boundaries"
(envelope (act 21 (Fork (ofString target) (ofString "ok"))))
reply @?= Right (Fork (Stem Leaf) (Fork Leaf Leaf))
BS.readFile target >>= (@?= "ok")
-
+ , testCase "root completion closes unclosed listening sockets" $ socketExit (act 0 Leaf) (Just (Right Leaf))
+ , testCase "abort closes unclosed listening sockets" $ socketExit (act 2 (ofString "stop")) (Just (Left "execution aborted: stop"))
+ , testCase "timeout closes unclosed listening sockets" $ socketExit (act 63 (ofNumber 60000)) Nothing
+ , testCase "root completion joins blocked file workers" $ workerExit (act 0 Leaf) (Just (Right (Leaf, Leaf)))
+ , testCase "abort joins blocked file workers" $ workerExit (act 2 (ofString "stop")) (Just (Left "execution aborted: stop"))
+ , testCase "timeout joins blocked file workers" $ workerExit (act 63 (ofNumber 60000)) Nothing
+ , testCase "timeout interrupts blocked socket accept" $ socketExit
+ (act 74 (Fork (ofString "sock") (ofNumber 0))) Nothing
]
+-- The reader consumes one byte to prove the OS writer started, then stops
+-- draining until run exit. Cancellation must stop the blocked writer: draining
+-- afterward must reach EOF without receiving the entire payload.
+workerExit :: T -> Maybe (Either String (T, T)) -> Assertion
+workerExit final expected = withSystemTempDirectory "io-worker" $ \dir -> do
+ let fifo = dir ++ "/fifo"
+ size = 262144
+ createNamedPipe fifo 0o600
+ bracket (openFd fifo ReadOnly (defaultFileFlags { nonBlock = True })) closeFd $ \reader -> do
+ let readSome count = do
+ result <- try (fdRead reader count)
+ case result of
+ Right (bytes, _) -> pure bytes
+ Left (err :: IOException) | isEOFError err -> pure ""
+ | isFullError err -> threadDelay 1000 >> readSome count
+ | otherwise -> ioError err
+ started = do
+ bytes <- readSome 1
+ if null bytes then threadDelay 1000 >> started else pure ()
+ hook = do
+ ready <- Timeout.timeout 5000000 started
+ ready @?= Just ()
+ drain total = do
+ bytes <- readSome 65536
+ if null bytes then pure total else drain (total + length bytes)
+ payload = ofList (replicate size (ofNumber 120))
+ action = thenDo (act 60 (act 22 (Fork (ofString fifo) payload))) $
+ thenDo (act 62 Leaf) final
+ outcome <- Timeout.timeout 6000000 (runIOWithYieldHook hook unsafePerms Leaf Leaf (envelope action))
+ outcome @?= expected
+ drained <- Timeout.timeout 5000000 (drain 1)
+ case drained of
+ Nothing -> assertFailure "writer did not terminate"
+ Just count -> assertBool "worker continued writing after run exit" (count < size)
+
+-- SockId 0 belongs to this run. The independent probe keeps no driver handles.
+socketExit :: T -> Maybe (Either String T) -> Assertion
+socketExit final expected = do
+ port <- bracket (NS.socket NS.AF_INET NS.Stream NS.defaultProtocol) NS.close $ \sock -> do
+ NS.bind sock (NS.SockAddrInet 0 (NS.tupleToHostAddress (127,0,0,1)))
+ NS.SockAddrInet p _ <- NS.getSocketName sock
+ pure p
+ let handle = Fork (ofString "sock") (ofNumber 0)
+ action = thenDo (act 70 Leaf) $ thenDo
+ (act 72 (Fork handle (Fork (ofString "127.0.0.1") (ofNumber (fromIntegral port))))) $
+ thenDo (act 73 (Fork handle (ofNumber 4))) (thenDo (act 62 Leaf) final)
+ hook = bracket (NS.socket NS.AF_INET NS.Stream NS.defaultProtocol) NS.close $ \probe -> do
+ bound <- try (NS.bind probe (NS.SockAddrInet port (NS.tupleToHostAddress (127,0,0,1))))
+ case bound of
+ Left (err :: IOException) -> assertBool (show err) (isAlreadyInUseError err)
+ Right () -> assertFailure "driver did not bind its listening socket"
+ outcome <- Timeout.timeout 100000 (runIOWithYieldHook hook unsafePerms Leaf Leaf (envelope action))
+ fmap (fmap fst) outcome @?= expected
+ bracket (NS.socket NS.AF_INET NS.Stream NS.defaultProtocol) NS.close $ \probe ->
+ NS.bind probe (NS.SockAddrInet port (NS.tupleToHostAddress (127,0,0,1)))
+
act :: Integer -> T -> T
act tag = Fork (ofNumber tag)
envelope :: T -> T
envelope action = Fork (ofString "tricuIO") (Fork (ofNumber 1) action)
+thenDo :: T -> T -> T
+thenDo action next = act 1 (Fork action (Fork Leaf next))
