IOV2TaskTests.hs (16456B)
1 {-# LANGUAGE ScopedTypeVariables #-} 2 module IOV2TaskTests (ioV2TaskTests) where 3 4 import qualified IOV2.Resources as R 5 import qualified IOV2.Tasks as S 6 import Runtime 7 import Control.Concurrent hiding (yield) 8 import Control.Exception 9 import Control.Monad hiding (join) 10 import Data.IORef 11 import qualified Data.ByteString as BS 12 import qualified System.Timeout as Timeout 13 import Test.Tasty 14 import Test.Tasty.HUnit 15 16 right :: Show e => Either e a -> IO a 17 right (Right x) = pure x 18 right (Left e) = assertFailure (show e) >> error "assertion failed" 19 wait :: MVar a -> IO a 20 wait var = Timeout.timeout 3000000 (takeMVar var) >>= maybe (assertFailure "timeout" >> error "timeout") pure 21 completed :: S.Outcome -> Assertion 22 completed (S.Completed value) = value @?= Leaf 23 completed value = assertFailure (show value) 24 rejected :: String -> Either R.Problem a -> Assertion 25 rejected name (Left e) = R.category e @?= name 26 rejected name _ = assertFailure ("expected " ++ name) 27 withGroup :: (R.Runtime -> S.Scheduler -> S.Group -> IO S.Outcome) -> IO S.Outcome 28 withGroup action = do 29 runtime <- R.newRuntime 30 scheduler <- S.newScheduler runtime 31 group <- S.rootGroup scheduler 32 result <- S.runGroup group (action runtime scheduler group) 33 R.liveCounts runtime >>= (@?= (0,0)) 34 S.liveCounts scheduler >>= (@?= (0,0,0)) 35 pure result 36 37 ioV2TaskTests :: TestTree 38 ioV2TaskTests = localOption (mkTimeout 30000000) $ testGroup "IO v2 structured tasks" 39 [ testCase "normal exit joins borrowing child before ancestor release" $ do 40 released <- newIORef False 41 entered <- newEmptyMVar 42 finish <- newEmptyMVar 43 result <- newEmptyMVar 44 _ <- forkIO $ withGroup (\runtime _ group -> do 45 token <- R.acquire runtime 0 (pure (R.Reader (const (pure BS.empty)) (writeIORef released True))) >>= right 46 _ <- S.spawn group False False [] $ \child _ -> do 47 putMVar entered () 48 wait finish 49 readIORef released >>= (@?= False) 50 R.readSome runtime (S.scopeOf child) token 1 >>= (@?= Right BS.empty) 51 pure (S.Completed Leaf) 52 pure (S.Completed Leaf)) >>= putMVar result 53 wait entered 54 tryTakeMVar result >>= \value -> case value of Nothing -> pure (); _ -> assertFailure "scope returned before child" 55 readIORef released >>= (@?= False) 56 putMVar finish () 57 wait result >>= completed 58 readIORef released >>= (@?= True) 59 , testCase "joined outcome is consumed once, ordinary errors remain values" $ 60 withGroup (\_ _ group -> do 61 ref <- S.spawn group False True [] (\_ _ -> pure (S.Completed (Fork Leaf Leaf))) >>= right 62 S.join group ref >>= right >>= \outcome -> case outcome of 63 S.Completed value -> value @?= Fork Leaf Leaf 64 _ -> assertFailure "domain value became terminal failure" 65 S.join group ref >>= rejected "stale-task" 66 S.discard group ref >>= rejected "stale-task" 67 pure (S.Completed Leaf)) >>= completed 68 , testCase "supervision is installed before execution and preserves lazy diagnostic" $ 69 withGroup (\_ _ group -> do 70 ref <- S.spawn group True True [] (\_ _ -> pure (S.Failed (error "diagnostic forced") [])) >>= right 71 S.join group ref >>= right >>= \outcome -> case outcome of 72 S.Failed _ [] -> pure () 73 _ -> assertFailure "supervised failure lost" 74 pure (S.Completed Leaf)) >>= completed 75 , testCase "uncontained fault interrupts body, cancels sibling and joins its lease" $ do 76 siblingReady <- newEmptyMVar 77 parentReady <- newEmptyMVar 78 blocked <- newEmptyMVar 79 resumed <- newIORef False 80 released <- newIORef False 81 outcome <- withGroup $ \runtime _ group -> do 82 _ <- S.spawn group False False [] $ \child _ -> do 83 token <- R.acquire runtime (S.scopeOf child) (pure (R.Writer 84 (\_ -> putMVar siblingReady () >> takeMVar blocked) 85 (writeIORef released True))) >>= right 86 _ <- R.writeSome runtime (S.scopeOf child) token (BS.singleton 1) 87 writeIORef resumed True 88 pure (S.Completed Leaf) 89 _ <- S.spawn group False False [] $ \_ _ -> do 90 wait siblingReady 91 wait parentReady 92 pure (S.Failed (ofString "child fault") []) 93 putMVar parentReady () 94 takeMVar blocked 95 writeIORef resumed True 96 pure (S.Completed Leaf) 97 case outcome of S.Failed diagnostic [] -> diagnostic @?= ofString "child fault"; _ -> assertFailure "fault lost" 98 readIORef resumed >>= (@?= False) 99 readIORef released >>= (@?= True) 100 , testCase "explicit cancellation settles blocked acquisition and reports reason" $ 101 withGroup (\runtime _ group -> do 102 entered <- newEmptyMVar 103 block <- newEmptyMVar 104 resumed <- newIORef False 105 ref <- S.spawn group True True [] (\child _ -> do 106 _ <- R.acquire runtime (S.scopeOf child) (putMVar entered () >> takeMVar block) 107 writeIORef resumed True 108 pure (S.Completed Leaf)) >>= right 109 wait entered 110 S.cancel group ref (ofString "shutdown") >>= (@?= Right ()) 111 S.join group ref >>= right >>= \outcome -> case outcome of 112 S.Cancelled reason [] -> reason @?= ofString "shutdown" 113 _ -> assertFailure "cancellation not observed" 114 readIORef resumed >>= (@?= False) 115 pure (S.Completed Leaf)) >>= completed 116 , testCase "transfer publication gives child fresh tokens and rejects old aliases" $ 117 withGroup (\runtime _ group -> do 118 released <- newIORef False 119 ready <- newEmptyMVar 120 finish <- newEmptyMVar 121 token <- R.acquire runtime 0 (pure (R.Reader (const (pure BS.empty)) (writeIORef released True))) >>= right 122 ref <- S.spawn group False True [token] (\child tokens -> do 123 case tokens of 124 [fresh] -> R.readSome runtime (S.scopeOf child) fresh 1 >>= (@?= Right BS.empty) 125 _ -> assertFailure "bad fresh tokens" 126 putMVar ready () 127 wait finish 128 pure (S.Completed Leaf)) >>= right 129 wait ready 130 R.readSome runtime 0 token 1 >>= rejected "stale-handle" 131 readIORef released >>= (@?= False) 132 putMVar finish () 133 S.join group ref >>= right >>= completed 134 readIORef released >>= (@?= True) 135 pure (S.Completed Leaf)) >>= completed 136 , testCase "failed transfer does not start callback or allocate task history" $ 137 withGroup (\runtime scheduler group -> do 138 started <- newIORef False 139 token <- R.acquire runtime 0 (pure (R.Reader (const (pure BS.empty)) (pure ()))) >>= right 140 S.spawn group False True [token,token] (\_ _ -> writeIORef started True >> pure (S.Completed Leaf)) >>= rejected "duplicate-handle" 141 readIORef started >>= (@?= False) 142 S.liveCounts scheduler >>= (@?= (0,0,0)) 143 R.readSome runtime 0 token 1 >>= (@?= Right BS.empty) 144 pure (S.Completed Leaf)) >>= completed 145 , testCase "repeated joins and discards leave no result archive" $ 146 withGroup (\_ scheduler group -> do 147 replicateM_ 500 $ do 148 ref <- S.spawn group False True [] (\_ _ -> pure (S.Completed (error "value forced"))) >>= right 149 S.join group ref >>= right >>= \outcome -> case outcome of S.Completed _ -> pure (); _ -> assertFailure "failed" 150 S.liveCounts scheduler >>= (@?= (0,0,0)) 151 replicateM_ 500 $ do 152 ref <- S.spawn group False True [] (\_ _ -> pure (S.Completed (error "discard forced"))) >>= right 153 S.discard group ref >>= (@?= Right ()) 154 pure (S.Completed Leaf)) >>= completed 155 , testCase "cancellation preserves primary and attaches all release errors" $ 156 withGroup (\runtime _ group -> do 157 ready <- newEmptyMVar 158 blocked <- newEmptyMVar 159 ref <- S.spawn group True True [] (\child _ -> do 160 replicateM_ 24 $ void $ R.acquire runtime (S.scopeOf child) 161 (pure (R.Reader (const (pure BS.empty)) (ioError (userError "release")))) >>= right 162 putMVar ready () 163 takeMVar blocked 164 pure (S.Completed Leaf)) >>= right 165 wait ready 166 S.cancel group ref (error "reason forced") >>= (@?= Right ()) 167 S.join group ref >>= right >>= \outcome -> case outcome of 168 S.Cancelled _ errors -> do 169 length errors @?= 17 170 R.detail (last errors) @?= "8" 171 _ -> assertFailure "primary cancellation lost" 172 pure (S.Completed Leaf)) >>= completed 173 , testCase "explicit cancellation of leased read/write remains sticky after backend catches wakeup" $ 174 withGroup (\runtime _ group -> do 175 forM_ [True, False] $ \reading -> do 176 entered <- newEmptyMVar 177 blocked <- newEmptyMVar 178 resumed <- newIORef False 179 released <- newIORef False 180 let operation = (putMVar entered () >> takeMVar blocked) 181 `catch` (\(_ :: AsyncException) -> pure ()) 182 backend = if reading 183 then R.Reader (\_ -> operation >> pure BS.empty) (writeIORef released True) 184 else R.Writer (\_ -> operation >> pure 1) (writeIORef released True) 185 ref <- S.spawn group True True [] (\child _ -> do 186 token <- R.acquire runtime (S.scopeOf child) (pure backend) >>= right 187 if reading then void (R.readSome runtime (S.scopeOf child) token 1) 188 else void (R.writeSome runtime (S.scopeOf child) token (BS.singleton 1)) 189 S.checkpoint child 190 writeIORef resumed True 191 pure (S.Completed Leaf)) >>= right 192 wait entered 193 S.cancel group ref Leaf >>= (@?= Right ()) 194 S.join group ref >>= right >>= \outcome -> case outcome of 195 S.Cancelled Leaf [] -> pure () 196 _ -> assertFailure "backend neutralized cancellation" 197 readIORef resumed >>= (@?= False) 198 readIORef released >>= (@?= True) 199 pure (S.Completed Leaf)) >>= completed 200 , testCase "cancellation versus completion linearizes to one terminal outcome" $ 201 withGroup (\_ _ group -> do 202 replicateM_ 200 $ do 203 ready <- newEmptyMVar 204 finish <- newEmptyMVar 205 cancelled <- newEmptyMVar 206 ref <- S.spawn group True True [] (\_ _ -> putMVar ready () >> wait finish >> pure (S.Completed Leaf)) >>= right 207 wait ready 208 _ <- forkIO (S.cancel group ref Leaf >>= putMVar cancelled) 209 putMVar finish () 210 request <- wait cancelled 211 outcome <- S.join group ref >>= right 212 case (request, outcome) of 213 (Right (), S.Cancelled Leaf []) -> pure () 214 (Left e, S.Completed Leaf) -> assertBool "completion won" (R.category e `elem` ["stale-task", "completed-task"]) 215 _ -> assertFailure "inconsistent completion/cancellation race" 216 pure (S.Completed Leaf)) >>= completed 217 , testCase "self/ancestor joins reject without consuming the ticket" $ 218 withGroup (\_ _ group -> do 219 reference <- newEmptyMVar 220 checked <- newEmptyMVar 221 ref <- S.spawn group True True [] (\child _ -> do 222 self <- wait reference 223 S.join child self >>= rejected "join-cycle" 224 result <- S.nested child (\inner -> S.join inner self >>= rejected "join-cycle" >> pure (S.Completed Leaf)) 225 putMVar checked () 226 pure result) >>= right 227 putMVar reference ref 228 wait checked 229 S.join group ref >>= right >>= completed 230 pure (S.Completed Leaf)) >>= completed 231 , testCase "implicit scope joins participate in wait-cycle detection" $ do 232 outcome <- withGroup $ \_ _ group -> do 233 aRef <- newEmptyMVar 234 bRef <- newEmptyMVar 235 aReady <- newEmptyMVar 236 b <- S.spawn group False True [] (\child _ -> do 237 a <- wait aRef 238 wait aReady 239 -- A's descendant already waits on B, or will do so after this claim. 240 result <- S.join child a 241 case result of Left e -> R.category e @?= "join-cycle"; Right _ -> pure () 242 pure (S.Completed Leaf)) >>= right 243 putMVar bRef b 244 a <- S.spawn group False True [] (\child _ -> do 245 _ <- S.spawn child False False [] (\grandchild _ -> do 246 target <- wait bRef 247 putMVar aReady () 248 result <- S.join grandchild target 249 case result of Left e -> R.category e @?= "join-cycle"; Right _ -> pure () 250 pure (S.Completed Leaf)) >>= right 251 pure (S.Completed Leaf)) >>= right 252 putMVar aRef a 253 pure (S.Completed Leaf) 254 case outcome of 255 S.Completed Leaf -> pure () -- Explicit join rejected the last edge. 256 S.Failed message [] -> message @?= ofString "task join cycle" 257 _ -> assertFailure "cycle was not settled" 258 , testCase "cancellation waits for lexical release and preserves its error" $ 259 withGroup (\runtime _ group -> do 260 entered <- newEmptyMVar 261 proceed <- newEmptyMVar 262 interrupted <- newIORef False 263 ref <- S.spawn group True True [] (\child _ -> S.nested child (\inner -> do 264 _ <- R.acquire runtime (S.scopeOf inner) (pure (R.Reader (const (pure BS.empty)) 265 ((putMVar entered () >> takeMVar proceed >> ioError (userError "release")) 266 `catch` (\(e :: AsyncException) -> writeIORef interrupted True >> throwIO e)))) >>= right 267 pure (S.Completed Leaf))) >>= right 268 wait entered 269 S.cancel group ref (ofString "shutdown") >>= (@?= Right ()) 270 putMVar proceed () 271 S.join group ref >>= right >>= \outcome -> case outcome of 272 S.Cancelled reason errors -> do 273 reason @?= ofString "shutdown" 274 length errors @?= 1 275 _ -> assertFailure "lexical cleanup/cancellation lost" 276 readIORef interrupted >>= (@?= False) 277 pure (S.Completed Leaf)) >>= completed 278 , testCase "concurrent ticket claims have exactly one consumer" $ 279 withGroup (\_ _ group -> do 280 ready <- newEmptyMVar 281 finish <- newEmptyMVar 282 loser <- newEmptyMVar 283 target <- S.spawn group True True [] (\_ _ -> putMVar ready () >> wait finish >> pure (S.Completed Leaf)) >>= right 284 wait ready 285 let consumer child _ = do 286 result <- S.join child target 287 case result of 288 Right (S.Completed Leaf) -> pure (S.Completed (Stem Leaf)) 289 Left e -> do 290 R.category e @?= "stale-task" 291 putMVar loser () 292 pure (S.Completed Leaf) 293 _ -> assertFailure "unexpected join outcome" >> pure (S.Failed Leaf []) 294 a <- S.spawn group True True [] consumer >>= right 295 b <- S.spawn group True True [] consumer >>= right 296 wait loser 297 putMVar finish () 298 x <- S.join group a >>= right 299 y <- S.join group b >>= right 300 case (x,y) of 301 (S.Completed Leaf, S.Completed (Stem Leaf)) -> pure () 302 (S.Completed (Stem Leaf), S.Completed Leaf) -> pure () 303 _ -> assertFailure "ticket consumed more or less than once" 304 pure (S.Completed Leaf)) >>= completed 305 , testCase "lexical fault preserves every release diagnostic without duplicate attachment" $ do 306 outcome <- withGroup $ \runtime _ group -> S.nested group $ \inner -> do 307 replicateM_ 3 $ void $ R.acquire runtime (S.scopeOf inner) 308 (pure (R.Reader (const (pure BS.empty)) (ioError (userError "lexical release")))) >>= right 309 pure (S.Failed (ofString "lexical fault") []) 310 case outcome of 311 S.Failed message errors -> do 312 message @?= ofString "lexical fault" 313 length errors @?= 3 314 _ -> assertFailure "lexical failure lost" 315 , testCase "nested lexical scope cancellation settles inner resources" $ do 316 ready <- newEmptyMVar 317 blocked <- newEmptyMVar 318 released <- newIORef False 319 outcome <- withGroup $ \runtime _ group -> do 320 _ <- S.spawn group False False [] (\_ _ -> wait ready >> pure (S.Failed Leaf [])) 321 S.nested group $ \inner -> do 322 _ <- R.acquire runtime (S.scopeOf inner) (pure (R.Reader (const (pure BS.empty)) (writeIORef released True))) >>= right 323 putMVar ready () 324 takeMVar blocked 325 pure (S.Completed Leaf) 326 case outcome of S.Failed Leaf [] -> pure (); _ -> assertFailure "nested interruption changed primary" 327 readIORef released >>= (@?= True) 328 ]