IOV2NetworkTests.hs (12108B)
1 {-# LANGUAGE ScopedTypeVariables #-} 2 module IOV2NetworkTests (ioV2NetworkTests) where 3 import qualified IOV2.Resources as R 4 import qualified IOV2.Network as TCP 5 import qualified IOV2.Tasks as S 6 import qualified IOV2.Driver as V 7 import Runtime (T(..), ofString, ofNumber, ofList) 8 import qualified Network.Socket as N 9 import qualified Network.Socket.ByteString as NB 10 import qualified Data.ByteString as B 11 import Control.Concurrent 12 import Control.Concurrent.STM 13 import Control.Exception 14 import Control.Monad 15 import Data.IORef 16 import GHC.Conc (threadStatus, ThreadStatus(..)) 17 import System.Timeout (timeout) 18 import Test.Tasty 19 import Test.Tasty.HUnit 20 21 right :: Show e => Either e a -> IO a 22 right = either (\e -> assertFailure (show e) >> error "assertion") pure 23 category name result = case result of 24 Left p -> R.category p @?= name 25 _ -> assertFailure ("expected "++name) 26 withRuntime = bracket R.newRuntime (\r -> void (R.closeScope r 0)) 27 wait done = timeout 3000000 (readMVar done) >>= maybe (assertFailure "timeout" >> error "timeout") pure 28 fork action = do 29 done <- newEmptyMVar 30 worker <- forkFinally action (putMVar done) 31 pure (worker,done) 32 stop (worker,done) = killThread worker >> void (wait done) 33 -- Probing a live socket with an exclusive address query can WIN the lease and 34 -- make the intended reader/acceptor fail busy. Observe the actual worker instead. 35 waitBlocked worker = do 36 ready <- timeout 3000000 loop 37 case ready of Just () -> pure (); Nothing -> assertFailure "worker never blocked" 38 where loop = threadStatus worker >>= \status -> case status of 39 ThreadBlocked _ -> pure () 40 ThreadFinished -> assertFailure "worker finished instead of blocking" 41 ThreadDied -> assertFailure "worker died instead of blocking" 42 _ -> yield >> loop 43 loop4 = R.IPv4 (127,0,0,1) 0 44 loop6 = R.IPv6 (0,0,0,0,0,0,0,1) 0 0 45 connected r target = do 46 listener <- TCP.listenTCP r 0 target 16 >>= right 47 address <- R.socketAddress r 0 listener False >>= right 48 client <- TCP.connectTCP r 0 address >>= right 49 server <- R.acceptConnection r 0 listener 0 >>= right 50 pure (listener,client,server,address) 51 write r token bytes = do 52 n <- R.writeSome r 0 token bytes >>= right 53 if n == B.length bytes then pure () else write r token (B.drop n bytes) 54 readAll r scope token = do 55 bytes <- R.readSome r scope token 4096 >>= right 56 if B.null bytes then pure B.empty else B.append bytes <$> readAll r scope token 57 58 ioV2NetworkTests = localOption (mkTimeout 30000000) $ testGroup "IO v2 TCP" 59 [ testCase "real IPv4/IPv6 byte streams, addresses, EOF and independent write half-close" $ withRuntime $ \r -> 60 forM_ [loop4,loop6] $ \target -> do 61 (listener,client,server,address) <- connected r target 62 R.socketAddress r 0 server False >>= (@?= Right address) 63 R.socketAddress r 0 client True >>= (@?= Right address) 64 R.socketAddress r 0 listener True >>= category "wrong-kind" 65 R.readSome r 0 listener 1 >>= category "wrong-kind" 66 R.filePosition r 0 client >>= category "wrong-kind" 67 let bytes = B.pack ([0..255]++[255,0,13,10]) 68 write r client bytes 69 R.shutdownWrite r 0 client >>= (@?= Right ()) 70 R.shutdownWrite r 0 client >>= (@?= Right ()) 71 R.writeSome r 0 client (B.singleton 7) >>= category "invalid-state" 72 readAll r 0 server >>= (@?= bytes) 73 R.readSome r 0 server 1 >>= (@?= Right B.empty) 74 write r server (B.reverse bytes) 75 R.shutdownWrite r 0 server >>= (@?= Right ()) 76 readAll r 0 client >>= (@?= B.reverse bytes) 77 mapM_ (\token -> R.closeResource r 0 token >>= (@?= Right ())) [listener,client,server] 78 R.liveCounts r >>= (@?= (1,3)) 79 , testCase "duplex leases coexist, release only their own bit, and close waits for both" $ withRuntime $ \r -> do 80 enteredRead <- newEmptyMVar; enteredWrite <- newEmptyMVar 81 readGate <- newEmptyMVar; writeGate <- newEmptyMVar 82 released <- newIORef False 83 let methods = R.DuplexOps 84 (\_ -> putMVar enteredRead () >> takeMVar readGate >> pure B.empty) 85 (\bytes -> putMVar enteredWrite () >> takeMVar writeGate >> pure (B.length bytes)) 86 (pure ()) (pure loop4) (pure loop4) (writeIORef released True) 87 token <- R.acquire r 0 (pure (R.Duplex methods)) >>= right 88 bracket (fork (R.readSome r 0 token 1)) stop $ \(_,readDone) -> 89 bracket (fork (R.writeSome r 0 token (B.singleton 1))) stop $ \(_,writeDone) -> do 90 wait enteredRead; wait enteredWrite 91 R.readSome r 0 token 1 >>= category "busy" 92 R.writeSome r 0 token (B.singleton 2) >>= category "busy" 93 R.shutdownWrite r 0 token >>= category "busy" 94 R.socketAddress r 0 token False >>= category "busy" 95 atomically (R.reserveChildSTM r 0 [token]) >>= category "busy" 96 putMVar readGate () 97 wait readDone >>= right >>= (@?= Right B.empty) 98 -- Finishing a read must not clear the simultaneous write lease. 99 atomically (R.reserveChildSTM r 0 [token]) >>= category "busy" 100 R.writeSome r 0 token (B.singleton 3) >>= category "busy" 101 bracket (fork (R.closeResource r 0 token)) stop $ \(_,closed) -> do 102 threadDelay 20000 103 isEmptyMVar closed >>= (@?= True) 104 readIORef released >>= (@?= False) 105 putMVar writeGate () 106 wait writeDone >>= right >>= (@?= Right 1) 107 wait closed >>= right >>= (@?= Right ()) 108 readIORef released >>= (@?= True) 109 , testCase "real pending receive still permits send and shutdown-write" $ withRuntime $ \r -> do 110 (_,client,server,_) <- connected r loop4 111 bracket (fork (R.readSome r 0 server 1)) stop $ \(worker,done) -> do 112 waitBlocked worker 113 write r server (B.singleton 255) 114 R.shutdownWrite r 0 server >>= (@?= Right ()) 115 readAll r 0 client >>= (@?= B.singleton 255) 116 write r client (B.singleton 7) 117 wait done >>= right >>= (@?= Right (B.singleton 7)) 118 , testCase "cancelled receive poisons only that direction and drains its lease" $ withRuntime $ \r -> do 119 (_,client,server,_) <- connected r loop4 120 bracket (fork (R.readSome r 0 server 1)) stop $ \worker@(_,done) -> do 121 waitBlocked (fst worker) 122 stop worker 123 wait done >>= \outcome -> case outcome of Left _ -> pure (); _ -> assertFailure "receive resumed" 124 R.readSome r 0 server 1 >>= category "invalid-state" 125 write r server (B.singleton 8) 126 R.shutdownWrite r 0 server >>= (@?= Right ()) 127 readAll r 0 client >>= (@?= B.singleton 8) 128 , testCase "accepted connection transfers atomically and child settlement sends EOF" $ withRuntime $ \r -> do 129 (_,client,server,_) <- connected r loop4 130 (scope,tokens) <- atomically (R.reserveChildSTM r 0 [server]) >>= right 131 R.readSome r 0 server 1 >>= category "stale-handle" 132 write r client (B.pack [0,255]) 133 R.shutdownWrite r 0 client >>= (@?= Right ()) 134 case tokens of 135 [fresh] -> readAll r scope fresh >>= (@?= B.pack [0,255]) 136 _ -> assertFailure "missing transferred connection" 137 R.closeScope r scope >>= (@?= []) 138 R.readSome r 0 client 1 >>= (@?= Right B.empty) 139 , testCase "accept cancellation releases admission and keeps listener reusable" $ withRuntime $ \r -> do 140 listener <- TCP.listenTCP r 0 loop4 1 >>= right 141 address <- R.socketAddress r 0 listener False >>= right 142 scope <- R.childScope r 0 >>= right 143 bracket (fork (R.acceptConnection r scope listener scope)) stop $ \worker -> do 144 waitBlocked (fst worker) 145 R.acceptConnection r 0 listener 0 >>= category "busy" 146 stop worker 147 R.closeScope r scope >>= (@?= []) 148 client <- TCP.connectTCP r 0 address >>= right 149 accepted <- R.acceptConnection r 0 listener 0 >>= right 150 R.shutdownWrite r 0 client >>= (@?= Right ()) 151 R.readSome r 0 accepted 1 >>= (@?= Right B.empty) 152 , testCase "closed owner admission cannot accept or leak a new connection" $ withRuntime $ \r -> do 153 (listener,_,_,_) <- connected r loop4 154 child <- R.childScope r 0 >>= right 155 atomically (R.beginCloseScopeSTM r 0) 156 R.acceptConnection r child listener 0 >>= category "closed-scope" 157 R.liveCounts r >>= (@?= (2,6)) 158 , testCase "bind/refused errors and bounds remain ordinary with no leaked entries" $ withRuntime $ \r -> do 159 listener <- TCP.listenTCP r 0 loop4 1 >>= right 160 address <- R.socketAddress r 0 listener False >>= right 161 TCP.listenTCP r 0 address 1 >>= category "address-in-use" 162 TCP.listenTCP r 0 loop4 0 >>= category "invalid-bound" 163 TCP.listenTCP r 0 loop4 4097 >>= category "invalid-bound" 164 TCP.connectTCP r 0 loop4 >>= category "invalid-bound" 165 R.liveCounts r >>= (@?= (1,4)) 166 R.closeResource r 0 listener >>= (@?= Right ()) 167 TCP.connectTCP r 0 address >>= category "connection-refused" 168 R.liveCounts r >>= (@?= (1,3)) 169 , testCase "structured cancellation settles a real accept before scope release" $ withRuntime $ \r -> do 170 scheduler <- S.newScheduler r 171 root <- S.rootGroup scheduler 172 outcome <- S.runGroup root $ do 173 listener <- TCP.listenTCP r 0 loop4 1 >>= right 174 started <- newEmptyMVar 175 child <- S.spawn root True True [] (\group _ -> do 176 myThreadId >>= putMVar started 177 _ <- R.acceptConnection r (S.scopeOf group) listener (S.scopeOf group) >>= right 178 pure (S.Completed Leaf)) >>= right 179 takeMVar started >>= waitBlocked 180 S.cancel root child (ofString "stop") >>= (@?= Right ()) 181 S.join root child >>= right >>= \result -> case result of 182 S.Cancelled reason [] -> reason @?= ofString "stop" 183 _ -> assertFailure "cancellation lost" 184 R.liveCounts r >>= (@?= (1,4)) 185 pure (S.Completed Leaf) 186 case outcome of S.Completed Leaf -> pure (); _ -> assertFailure ("parent failed: " ++ show outcome) 187 R.liveCounts r >>= (@?= (0,0)) 188 , testCase "wire bounds reject noncanonical/excess address fields before acquisition" $ do 189 let v4 bytes port = ofList [ofString "ipv4",bytes,port] 190 v6 words scope = ofList [ofString "ipv6",words,scope,ofNumber 80] 191 zero4 = ofList (map ofNumber [127,0,0,1]) 192 zero6 = ofList (replicate 8 Leaf) 193 excess = foldr Fork (Fork (error "excess octet forced") (error "excess tail forced")) 194 (replicate 4 Leaf) 195 addresses = [v4 (ofList (map ofNumber [127,0,0,256])) (ofNumber 80), 196 v4 zero4 (ofNumber 65536), v4 excess (ofNumber 80), 197 v4 (ofList [Fork Leaf Leaf,Leaf,Leaf,Leaf]) (ofNumber 80), 198 v6 zero6 (ofNumber 4294967296), 199 v6 (ofList (ofNumber 65536:replicate 7 Leaf)) Leaf] 200 forM_ addresses $ \address -> withRuntime $ \r -> do 201 identity <- R.runtimeIdentity r 202 let owner = ofList [ofString "io.v2.scope",ofNumber identity,Leaf] 203 action = Fork (ofNumber 10) (Fork (ofString "tcp") (Fork (ofNumber 1) 204 (Fork (ofString "connect") (Fork owner address)))) 205 V.runInRuntime r action >>= \outcome -> case outcome of 206 V.Completed value -> value @?= Fork Leaf (Fork 207 (ofList (map ofString ["invalid-argument","tcp/connect","","none"])) Leaf) 208 _ -> assertFailure "bad address executed" 209 , testCase "blocked real write cancellation poisons writer; read direction survives" $ withRuntime $ \r -> 210 bracket (N.socketPair N.AF_UNIX N.Stream N.defaultProtocol) 211 (\(a,b) -> N.close a >> N.close b) $ \(socket,peer) -> do 212 N.setSocketOption socket N.SendBuffer 4096 213 token <- R.acquire r 0 (TCP.duplexBackend socket) >>= right 214 bracket (fork (forever (write r token (B.replicate 4096 7)))) stop $ \worker -> do 215 waitBlocked (fst worker) 216 stop worker 217 R.writeSome r 0 token (B.singleton 1) >>= category "invalid-state" 218 NB.sendAll peer (B.singleton 255) 219 R.readSome r 0 token 1 >>= (@?= Right (B.singleton 255)) 220 ]