git.iko.soy/distributed-process
on master help sign in
Fixed one more race condition

This was exposed by a test of closing the same connection twice, causing the connection to be prematurely closed. Closing the same connection still results in incorrect behaviour (still to be fixed), but at least now we get the right unexpected behaviour :)

commit230dee93587038a5024e64f448072f4901352ad4 authorEdsko de Vries <edsko@edsko.net> dateWed 2 May 2012, 16:39 (14 years ago) parent78147be
2 files changed+80−17
compare with master
unifiedsplit
network-transport-2/src/Network/Transport/TCP.hs+19−15 network-transport-2/tests/TestTransport.hs+61−2
network-transport-2/src/Network/Transport/TCP.hs+19−15
@@ -58,7 +58,7 @@ import Control.Concurrent.MVar ( MVar
5858import Control.Category ((>>>))
5959import Control.Applicative ((<*>), (*>), (<$>))
6060import Control.Monad (forM_, void, unless)
61−import Control.Monad.Error (MonadError, ErrorT(ErrorT), runErrorT, throwError)
61+import Control.Monad.Error (ErrorT(ErrorT), runErrorT)
6262import Control.Monad.IO.Class (liftIO)
6363import Control.Exception (IOException, bracket_)
6464import Data.ByteString (ByteString)
@@ -256,15 +256,16 @@ apiConnect :: LocalEndPoint -- ^ Local end point
256256apiConnect ourEndPoint theirAddress _ = runErrorT $ do
257257 theirEndPoint <- findRemoteEndPoint ourEndPoint theirAddress
258258 connId <- requestNewConnection (ourEndPoint, theirEndPoint)
259− return $ Connection { send = apiSend theirEndPoint connId
260− , close = apiClose (ourEndPoint, theirEndPoint) connId
259+ connAlive <- liftIO $ newMVar True
260+ return $ Connection { send = apiSend theirEndPoint connId connAlive
261+ , close = apiClose (ourEndPoint, theirEndPoint) connId connAlive
261262 }
262263
263264-- | Close a connection
264265--
265266-- TODO: We ignore errors during a close. Is that right?
266−apiClose :: Conn -> ConnectionId -> IO ()
267−apiClose (ourEndPoint, theirEndPoint) connId = void . tryIO $ do
267+apiClose :: Conn -> ConnectionId -> MVar Bool -> IO ()
268+apiClose (ourEndPoint, theirEndPoint) connId connAlive = void . tryIO $ do
268269 sendTo theirEndPoint [encodeInt32 CloseConnection, encodeInt32 connId]
269270 decreaseRemoteRefCount (ourEndPoint, theirEndPoint)
270271
@@ -272,8 +273,8 @@ apiClose (ourEndPoint, theirEndPoint) connId = void . tryIO $ do
272273--
273274-- TODO: we should have some per-connection state so that the user cannot send
274275-- to the connection after they closed it, or close it more than once
275−apiSend :: RemoteEndPoint -> ConnectionId -> [ByteString] -> IO (Either (FailedWith SendErrorCode) ())
276−apiSend theirEndPoint connId payload = runErrorT $
276+apiSend :: RemoteEndPoint -> ConnectionId -> MVar Bool -> [ByteString] -> IO (Either (FailedWith SendErrorCode) ())
277+apiSend theirEndPoint connId connAlive payload = runErrorT $
277278 failWithIO (FailedWith SendFailed . show) $
278279 sendTo theirEndPoint (encodeInt32 connId : prependLength payload)
279280
@@ -372,16 +373,19 @@ connectToRemoteEndPoint ourEndPoint theirMVar theirAddress = do
372373
373374-- | Request a new connection
374375requestNewConnection :: Conn -> ErrorT (FailedWith ConnectErrorCode) IO ConnectionId
375−requestNewConnection (ourEndPoint, theirEndPoint) = do
376− mcid <- failWithIO (failed . show) $ do
376+requestNewConnection (ourEndPoint, theirEndPoint) = ErrorT $ do
377+ -- We increment the refcount before sending the request, because if a
378+ -- CloseSocket request comes this way first we want to ignore it
379+ increaseRemoteRefCount (ourEndPoint, theirEndPoint)
380+ mcid <- tryIO $ do
377381 reply <- sendRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId
378382 decodeInt32 . BS.concat <$> takeMVar reply
379− case mcid of
380− Nothing ->
381− throwError (failed "Invalid integer")
382− Just cid -> do
383− liftIO $ increaseRemoteRefCount (ourEndPoint, theirEndPoint)
384− return cid
383+ case mcid of
384+ Left err -> do decreaseRemoteRefCount (ourEndPoint, theirEndPoint)
385+ return . Left $ failed (show err)
386+ Right Nothing -> do decreaseRemoteRefCount (ourEndPoint, theirEndPoint)
387+ return . Left $ failed "Invalid integer"
388+ Right (Just cid) -> do return . Right $ cid
385389 where
386390 failed = FailedWith ConnectFailed
387391
network-transport-2/tests/TestTransport.hs+61−2
@@ -1,7 +1,7 @@
11module TestTransport where
22
33import Prelude hiding (catch)
4−import Control.Concurrent (forkIO)
4+import Control.Concurrent (forkIO, threadDelay)
55import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar, readMVar)
66import Control.Monad (replicateM, replicateM_, when, guard, forM_)
77import Control.Monad.Reader (ReaderT, runReaderT, ask)
@@ -21,7 +21,6 @@ import System.Timeout (timeout)
2121-- Server that echoes messages straight back to the origin endpoint.
2222echoServer :: EndPoint -> IO ()
2323echoServer endpoint = do
24− tlog "Echo server"
2524 go Map.empty
2625 where
2726 go :: Map ConnectionId Connection -> IO ()
@@ -372,6 +371,64 @@ testParallelConnects transport numPings = do
372371
373372 takeMVar done
374373
374+-- Test that sending on a closed connection gives an error
375+testSendAfterClose :: Transport -> Int -> IO ()
376+testSendAfterClose transport _ = do
377+ server <- spawn transport echoServer
378+ clientDone <- newEmptyMVar
379+
380+ forkIO $ do
381+ Right endpoint <- newEndPoint transport
382+
383+ -- We request two lightweight connections
384+ Right conn1 <- connect endpoint server ReliableOrdered
385+ Right conn2 <- connect endpoint server ReliableOrdered
386+
387+ -- Close the second, but leave the first open; then output on the second
388+ -- connection (i.e., on a closed connection while there is still another
389+ -- connection open)
390+ close conn2
391+ send conn2 ["ping2"]
392+
393+ -- Now close the first connection, and output on it (i.e., output while
394+ -- there are no lightweight connection at all anymore)
395+ close conn1
396+ send conn1 ["ping1"]
397+
398+ threadDelay 100000
399+ putMVar clientDone ()
400+
401+ takeMVar clientDone
402+
403+-- Test that closing the same connection twice has no effect
404+testCloseTwice :: Transport -> Int -> IO ()
405+testCloseTwice transport _ = do
406+ server <- spawn transport echoServer
407+ clientDone <- newEmptyMVar
408+
409+ forkIO $ do
410+ Right endpoint <- newEndPoint transport
411+
412+ -- We request two lightweight connections
413+ Right conn1 <- connect endpoint server ReliableOrdered
414+ Right conn2 <- connect endpoint server ReliableOrdered
415+
416+ -- Close the second one twice
417+ close conn2
418+ close conn2
419+
420+ -- Then send a message on the first and close that too
421+ send conn1 ["ping"]
422+ close conn1
423+
424+ -- Verify expected response from the echo server
425+ -- TODO
426+
427+ threadDelay 1000000
428+ putMVar clientDone ()
429+
430+ takeMVar clientDone
431+
375432runTestIO :: String -> IO () -> IO ()
376433runTestIO description test = do
377434 putStr $ "Running " ++ show description ++ ": "
@@ -397,3 +454,5 @@ testTransport transport = flip runReaderT (transport, 10000) $ do
397454 runTest "CloseOneDirection" testCloseOneDirection
398455 runTest "CloseReopen" testCloseReopen
399456 runTest "ParallelConnects" testParallelConnects
457+ runTest "SendAfterClose" testSendAfterClose
458+ runTest "CloseTwice" testCloseTwice