230dee9 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Fixed one more race condition

230dee9Edsko de Vries14 years ago

network-transport-2/src/Network/Transport/TCP.hs+19−15

compare with master

unifiedsplit

@@ -58,7 +58,7 @@ import Control.Concurrent.MVar ( MVar

58import Control.Category ((>>>))

59import Control.Applicative ((<*>), (*>), (<$>))

60import Control.Monad (forM_, void, unless)

61−import Control.Monad.Error (MonadError, ErrorT(ErrorT), runErrorT, throwError)

61+import Control.Monad.Error (ErrorT(ErrorT), runErrorT)

62import Control.Monad.IO.Class (liftIO)

63import Control.Exception (IOException, bracket_)

64import Data.ByteString (ByteString)

@@ -256,15 +256,16 @@ apiConnect :: LocalEndPoint -- ^ Local end point

256apiConnect ourEndPoint theirAddress _ = runErrorT $ do

257 theirEndPoint <- findRemoteEndPoint ourEndPoint theirAddress

258 connId <- requestNewConnection (ourEndPoint, theirEndPoint)

259+ connAlive <- liftIO $ newMVar True

259− return $ Connection { send = apiSend theirEndPoint connId

260+ return $ Connection { send = apiSend theirEndPoint connId connAlive

260− , close = apiClose (ourEndPoint, theirEndPoint) connId

261+ , close = apiClose (ourEndPoint, theirEndPoint) connId connAlive

261 }

262 }

262

263

263-- | Close a connection

264-- | Close a connection

264--

265--

265-- TODO: We ignore errors during a close. Is that right?

266-- TODO: We ignore errors during a close. Is that right?

266−apiClose :: Conn -> ConnectionId -> IO ()

267+apiClose :: Conn -> ConnectionId -> MVar Bool -> IO ()

267−apiClose (ourEndPoint, theirEndPoint) connId = void . tryIO $ do

268+apiClose (ourEndPoint, theirEndPoint) connId connAlive = void . tryIO $ do

268 sendTo theirEndPoint [encodeInt32 CloseConnection, encodeInt32 connId]

269 sendTo theirEndPoint [encodeInt32 CloseConnection, encodeInt32 connId]

269 decreaseRemoteRefCount (ourEndPoint, theirEndPoint)

270 decreaseRemoteRefCount (ourEndPoint, theirEndPoint)

270

271

@@ -272,8 +273,8 @@ apiClose (ourEndPoint, theirEndPoint) connId = void . tryIO $ do

272--

273--

273-- TODO: we should have some per-connection state so that the user cannot send

274-- TODO: we should have some per-connection state so that the user cannot send

274-- to the connection after they closed it, or close it more than once

275-- to the connection after they closed it, or close it more than once

275−apiSend :: RemoteEndPoint -> ConnectionId -> [ByteString] -> IO (Either (FailedWith SendErrorCode) ())

276+apiSend :: RemoteEndPoint -> ConnectionId -> MVar Bool -> [ByteString] -> IO (Either (FailedWith SendErrorCode) ())

276−apiSend theirEndPoint connId payload = runErrorT $

277+apiSend theirEndPoint connId connAlive payload = runErrorT $

277 failWithIO (FailedWith SendFailed . show) $

278 failWithIO (FailedWith SendFailed . show) $

278 sendTo theirEndPoint (encodeInt32 connId : prependLength payload)

279 sendTo theirEndPoint (encodeInt32 connId : prependLength payload)

279

280

@@ -372,16 +373,19 @@ connectToRemoteEndPoint ourEndPoint theirMVar theirAddress = do

372

373

373-- | Request a new connection

374-- | Request a new connection

374requestNewConnection :: Conn -> ErrorT (FailedWith ConnectErrorCode) IO ConnectionId

375requestNewConnection :: Conn -> ErrorT (FailedWith ConnectErrorCode) IO ConnectionId

375−requestNewConnection (ourEndPoint, theirEndPoint) = do

376+requestNewConnection (ourEndPoint, theirEndPoint) = ErrorT $ do

376− mcid <- failWithIO (failed . show) $ 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

377 reply <- sendRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId

381 reply <- sendRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId

378 decodeInt32 . BS.concat <$> takeMVar reply

382 decodeInt32 . BS.concat <$> takeMVar reply

379− case mcid of

383+ case mcid of

380− Nothing ->

384+ Left err -> do decreaseRemoteRefCount (ourEndPoint, theirEndPoint)

385+ return . Left $ failed (show err)

386+ Right Nothing -> do decreaseRemoteRefCount (ourEndPoint, theirEndPoint)

381− throwError (failed "Invalid integer")

387+ return . Left $ failed "Invalid integer"

382− Just cid -> do

388+ Right (Just cid) -> do return . Right $ cid

383− liftIO $ increaseRemoteRefCount (ourEndPoint, theirEndPoint)

384− return cid

385 where

389 where

386 failed = FailedWith ConnectFailed

390 failed = FailedWith ConnectFailed

387

391

network-transport-2/tests/TestTransport.hs+61−2

@@ -1,7 +1,7 @@

1module TestTransport where

3import Prelude hiding (catch)

4−import Control.Concurrent (forkIO)

4+import Control.Concurrent (forkIO, threadDelay)

5import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar, readMVar)

6import Control.Monad (replicateM, replicateM_, when, guard, forM_)

7import Control.Monad.Reader (ReaderT, runReaderT, ask)

@@ -21,7 +21,6 @@ import System.Timeout (timeout)

21-- Server that echoes messages straight back to the origin endpoint.

22echoServer :: EndPoint -> IO ()

23echoServer endpoint = do

24− tlog "Echo server"

25 go Map.empty

24 go Map.empty

26 where

25 where

27 go :: Map ConnectionId Connection -> IO ()

26 go :: Map ConnectionId Connection -> IO ()

@@ -372,6 +371,64 @@ testParallelConnects transport numPings = do

372

371

373 takeMVar done

372 takeMVar done

374

373

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+

375runTestIO :: String -> IO () -> IO ()

432runTestIO :: String -> IO () -> IO ()

376runTestIO description test = do

433runTestIO description test = do

377 putStr $ "Running " ++ show description ++ ": "

434 putStr $ "Running " ++ show description ++ ": "

@@ -397,3 +454,5 @@ testTransport transport = flip runReaderT (transport, 10000) $ do

397 runTest "CloseOneDirection" testCloseOneDirection

454 runTest "CloseOneDirection" testCloseOneDirection

398 runTest "CloseReopen" testCloseReopen

455 runTest "CloseReopen" testCloseReopen

399 runTest "ParallelConnects" testParallelConnects

456 runTest "ParallelConnects" testParallelConnects

457+ runTest "SendAfterClose" testSendAfterClose

458+ runTest "CloseTwice" testCloseTwice