git.iko.soy/distributed-process
on master help sign in
Code cleanup
commitdbe95569da87259022122d40a14dfb67fda8460f authorEdsko de Vries <edsko@edsko.net> dateMon 28 May 2012, 14:22 (14 years ago) parent553015e
5 files changed+97−106
compare with master
unifiedsplit
network-transport/src/Network/Transport/Internal.hs+19−14 network-transport/src/Network/Transport/Internal/TCP.hs+3−6 network-transport/src/Network/Transport/TCP.hs+70−81 network-transport/tests/TestTCP.hs+4−4 network-transport/tests/TestTransport.hs+1−1
network-transport/src/Network/Transport/Internal.hs+19−14
@@ -30,6 +30,7 @@ import Control.Exception ( IOException
3030 , Exception
3131 , catch
3232 , try
33+ , throw
3334 , throwIO
3435 )
3536import Control.Concurrent (ThreadId, forkIO)
@@ -48,13 +49,15 @@ encodeInt32 i32 =
4849 pokeByteOff p 0 (htonl . fromIntegral . fromEnum $ i32)
4950
5051-- | Deserialize 32-bit from network byte order
51−decodeInt32 :: Num a => ByteString -> Maybe a
52−decodeInt32 bs | BS.length bs /= 4 = Nothing
53−decodeInt32 bs = Just . BSI.inlinePerformIO $ do
54− let (fp, _, _) = BSI.toForeignPtr bs
55− withForeignPtr fp $ \p -> do
56− w32 <- peekByteOff p 0
57− return (fromIntegral . ntohl $ w32)
52+-- Throws an IO exception if this is not a valid integer.
53+decodeInt32 :: Num a => ByteString -> a
54+decodeInt32 bs
55+ | BS.length bs /= 4 = throw $ userError "decodeInt32: Invalid length"
56+ | otherwise = BSI.inlinePerformIO $ do
57+ let (fp, _, _) = BSI.toForeignPtr bs
58+ withForeignPtr fp $ \p -> do
59+ w32 <- peekByteOff p 0
60+ return (fromIntegral . ntohl $ w32)
5861
5962-- | Serialize 16-bit to network byte order
6063encodeInt16 :: Enum a => a -> ByteString
@@ -63,13 +66,15 @@ encodeInt16 i16 =
6366 pokeByteOff p 0 (htons . fromIntegral . fromEnum $ i16)
6467
6568-- | Deserialize 16-bit from network byte order
66−decodeInt16 :: Num a => ByteString -> Maybe a
67−decodeInt16 bs | BS.length bs /= 2 = Nothing
68−decodeInt16 bs = Just . BSI.inlinePerformIO $ do
69− let (fp, _, _) = BSI.toForeignPtr bs
70− withForeignPtr fp $ \p -> do
71− w16 <- peekByteOff p 0
72− return (fromIntegral . ntohs $ w16)
69+-- Throws an IO exception if this is not a valid integer
70+decodeInt16 :: Num a => ByteString -> a
71+decodeInt16 bs
72+ | BS.length bs /= 2 = throw $ userError "decodeInt16: Invalid length"
73+ | otherwise = BSI.inlinePerformIO $ do
74+ let (fp, _, _) = BSI.toForeignPtr bs
75+ withForeignPtr fp $ \p -> do
76+ w16 <- peekByteOff p 0
77+ return (fromIntegral . ntohs $ w16)
7378
7479-- | Prepend a list of bytestrings with their total length
7580prependLength :: [ByteString] -> [ByteString]
network-transport/src/Network/Transport/Internal/TCP.hs+3−6
@@ -27,8 +27,9 @@ import qualified Network.Socket as N ( HostName
2727 )
2828import qualified Network.Socket.ByteString as NBS (recv)
2929import Control.Concurrent (ThreadId)
30−import Control.Monad (liftM, forever)
30+import Control.Monad (forever)
3131import Control.Exception (SomeException, catch, bracketOnError, throwIO, mask_)
32+import Control.Applicative ((<$>))
3233import Data.ByteString (ByteString)
3334import qualified Data.ByteString as BS (length, concat, null)
3435import Data.Int (Int32)
@@ -82,11 +83,7 @@ recvWithLength sock = recvInt32 sock >>= recvExact sock
8283
8384-- | Receive a 32-bit integer
8485recvInt32 :: Num a => N.Socket -> IO a
85−recvInt32 sock = do
86− mi <- liftM (decodeInt32 . BS.concat) $ recvExact sock 4
87− case mi of
88− Nothing -> throwIO (userError "Invalid integer")
89− Just i -> return i
86+recvInt32 sock = decodeInt32 . BS.concat <$> recvExact sock 4
9087
9188-- | Close a socket, ignoring I/O exceptions
9289tryCloseSocket :: N.Socket -> IO ()
network-transport/src/Network/Transport/TCP.hs+70−81
@@ -73,7 +73,6 @@ import Control.Concurrent.MVar ( MVar
7373import Control.Category ((>>>))
7474import Control.Applicative ((<$>))
7575import Control.Monad (when, unless)
76−import Control.Monad.Error (ErrorT(..), runErrorT)
7776import Control.Exception ( IOException
7877 , SomeException
7978 , handle
@@ -374,7 +373,7 @@ data TransportInternals = TransportInternals
374373 -- | Find the socket between a local and a remote endpoint
375374 , socketBetween :: EndPointAddress
376375 -> EndPointAddress
377− -> IO (Either String N.Socket)
376+ -> IO N.Socket
378377 }
379378
380379--------------------------------------------------------------------------------
@@ -626,39 +625,38 @@ connectToSelf ourEndPoint = try . mapIOException connectFailed $ do
626625--------------------------------------------------------------------------------
627626
628627-- Find a socket between two endpoints
628+--
629+-- Throws an IO exception if the socket could not be found.
629630internalSocketBetween :: TCPTransport -- ^ Transport
630631 -> EndPointAddress -- ^ Local endpoint
631632 -> EndPointAddress -- ^ Remote endpoint
632− -> IO (Either String N.Socket)
633−internalSocketBetween transport ourAddress theirAddress = runErrorT $ do
634− ourEndPoint <- ErrorT $
635− withMVar (transportState transport) $ \st -> case st of
633+ -> IO N.Socket
634+internalSocketBetween transport ourAddress theirAddress = do
635+ ourEndPoint <- withMVar (transportState transport) $ \st -> case st of
636636 TransportClosed ->
637− return . Left $ "Transport closed"
637+ throwIO $ userError "Transport closed"
638638 TransportValid vst ->
639639 case vst ^. localEndPointAt ourAddress of
640− Nothing -> return . Left $ "Local endpoint not found"
641− Just ep -> return . Right $ ep
642− theirEndPoint <- ErrorT $
643− withMVar (localState ourEndPoint) $ \st -> case st of
640+ Nothing -> throwIO $ userError "Local endpoint not found"
641+ Just ep -> return ep
642+ theirEndPoint <- withMVar (localState ourEndPoint) $ \st -> case st of
644643 LocalEndPointClosed ->
645− return . Left $ "Local endpoint closed"
644+ throwIO $ userError "Local endpoint closed"
646645 LocalEndPointValid vst ->
647646 case vst ^. localConnectionTo theirAddress of
648− Nothing -> return . Left $ "Remote endpoint not found"
649− Just ep -> return . Right $ ep
650− ErrorT $
651− withMVar (remoteState theirEndPoint) $ \st -> case st of
652− RemoteEndPointInit _ ->
653− return . Left $ "Remote endpoint not yet initialized"
654− RemoteEndPointValid vst ->
655− return . Right $ remoteSocket vst
656− RemoteEndPointClosing _ vst ->
657− return . Right $ remoteSocket vst
658− RemoteEndPointClosed ->
659− return . Left $ "Remote endpoint closed"
660− RemoteEndPointInvalid _ ->
661− return . Left $ "Remote endpoint invalid"
647+ Nothing -> throwIO $ userError "Remote endpoint not found"
648+ Just ep -> return ep
649+ withMVar (remoteState theirEndPoint) $ \st -> case st of
650+ RemoteEndPointInit _ ->
651+ throwIO $ userError "Remote endpoint not yet initialized"
652+ RemoteEndPointValid vst ->
653+ return $ remoteSocket vst
654+ RemoteEndPointClosing _ vst ->
655+ return $ remoteSocket vst
656+ RemoteEndPointClosed ->
657+ throwIO $ userError "Remote endpoint closed"
658+ RemoteEndPointInvalid _ ->
659+ throwIO $ userError "Remote endpoint invalid"
662660
663661--------------------------------------------------------------------------------
664662-- Lower level functionality --
@@ -708,35 +706,25 @@ createLocalEndPoint transport = do
708706requestConnectionTo :: LocalEndPoint
709707 -> EndPointAddress
710708 -> IO (RemoteEndPoint, ConnectionId)
711−requestConnectionTo ourEndPoint theirAddress = go
709+requestConnectionTo ourEndPoint theirAddress = do
710+ (theirEndPoint, isNew) <- mapIOException connectFailed $
711+ findRemoteEndPoint ourEndPoint theirAddress BlockOnInit
712+ (RemoteEndPointValid . (remoteOutgoing ^: (+ 1)))
713+
714+ -- If this is a new endpoint, start the initialization process and wait
715+ forM_ isNew $ \resolved -> do
716+ void . forkEndPointThread ourEndPoint $ do
717+ let outgoing = 1
718+ setupRemoteEndPoint (ourEndPoint, theirEndPoint) outgoing
719+ readMVar resolved
720+
721+ reply <- mapIOException connectFailed $
722+ doRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId
723+ return (theirEndPoint, decodeInt32 . BS.concat $ reply)
712724 where
713− go = do
714− (theirEndPoint, isNew) <- mapIOException connectFailed $
715− findRemoteEndPoint ourEndPoint theirAddress True
716− (RemoteEndPointValid . (remoteOutgoing ^: (+ 1)))
717−
718− -- If this is a new endpoint, start the initialization process and wait
719− forM_ isNew $ \resolved -> do
720− void . forkEndPointThread ourEndPoint $ do
721− let outgoing = 1
722− setupRemoteEndPoint (ourEndPoint, theirEndPoint) outgoing
723− readMVar resolved
724−
725− reply <- mapIOException connectFailed $
726− doRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId
727− case decodeInt32 . BS.concat $ reply of
728− Nothing ->
729− throwIO (connectFailed $ userError "Invalid integer")
730− Just cid ->
731− return (theirEndPoint, cid)
732−
733725 connectFailed = TransportError ConnectFailed . show
734726
735727-- | Set up a remote endpoint
736−--
737−-- RELY: The state of the remote endpoint must be uninitialized.
738−-- GUARANTEE: Will only change the state to RemoteEndPointValid or
739−-- RemoteEndPointInvalid.
740728setupRemoteEndPoint :: EndPointPair -- ^ Local endpoint, remote endpoint
741729 -> Int -- ^ Initial (outgoing) refcount
742730 -> IO ()
@@ -751,9 +739,7 @@ setupRemoteEndPoint (ourEndPoint, theirEndPoint) outgoing = do
751739 , _remoteIncoming = IntSet.empty
752740 , sendOn = sendMany sock
753741 }
754− modifyMVar_ theirState $ \(RemoteEndPointInit resolved) -> do
755− putMVar resolved ()
756− return (RemoteEndPointValid vst)
742+ resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)
757743 return True
758744 Right (sock, ConnectionRequestInvalid) -> do
759745 -- We remove the endpoint from our local state again because the next
@@ -763,9 +749,7 @@ setupRemoteEndPoint (ourEndPoint, theirEndPoint) outgoing = do
763749 -- connection requests.
764750 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
765751 let err = invalidAddress "Invalid endpoint"
766− modifyMVar_ theirState $ \(RemoteEndPointInit resolved) -> do
767− putMVar resolved ()
768− return (RemoteEndPointInvalid err)
752+ resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)
769753 tryCloseSocket sock
770754 return False
771755 Right (sock, ConnectionRequestCrossed) -> do
@@ -775,9 +759,7 @@ setupRemoteEndPoint (ourEndPoint, theirEndPoint) outgoing = do
775759 return False
776760 Left err -> do
777761 removeRemoteEndPoint (ourEndPoint, theirEndPoint) -- See comment above
778− modifyMVar_ theirState $ \(RemoteEndPointInit resolved) -> do
779− putMVar resolved ()
780− return (RemoteEndPointInvalid err)
762+ resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)
781763 return False
782764
783765 -- If we get to this point without an exception, then
@@ -800,14 +782,12 @@ setupRemoteEndPoint (ourEndPoint, theirEndPoint) outgoing = do
800782 onError result = do
801783 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
802784 case result of
803− Left err -> do
804− tryPutMVar theirState (RemoteEndPointInvalid err)
805− return ()
785+ Left err ->
786+ void $ tryPutMVar theirState (RemoteEndPointInvalid err)
806787 Right (sock, _) -> do
807788 let err = failed "setupRemoteEndPoint failed"
808789 tryPutMVar theirState (RemoteEndPointInvalid err)
809790 tryCloseSocket sock
810− return ()
811791
812792 failed = TransportError ConnectFailed
813793 ourAddress = localAddress ourEndPoint
@@ -815,6 +795,21 @@ setupRemoteEndPoint (ourEndPoint, theirEndPoint) outgoing = do
815795 theirState = remoteState theirEndPoint
816796 invalidAddress = TransportError ConnectNotFound
817797
798+resolveInit :: EndPointPair -> RemoteState -> IO ()
799+resolveInit (ourEndPoint, theirEndPoint) newState =
800+ modifyMVar_ (remoteState theirEndPoint) $ \st -> case st of
801+ RemoteEndPointInvalid _ ->
802+ relyViolation (ourEndPoint, theirEndPoint) "resolveInit (invalid)"
803+ RemoteEndPointInit resolved -> do
804+ putMVar resolved ()
805+ return newState
806+ RemoteEndPointClosed ->
807+ throwIO $ userError "Remote endpoint closed"
808+ RemoteEndPointClosing _ _ ->
809+ relyViolation (ourEndPoint, theirEndPoint) "resolveInit (closing)"
810+ RemoteEndPointValid _ ->
811+ relyViolation (ourEndPoint, theirEndPoint) "resolveInit (valid)"
812+
818813-- | Establish a connection to a remote endpoint
819814socketToEndPoint :: EndPointAddress -- ^ Our address
820815 -> EndPointAddress -- ^ Their address
@@ -904,9 +899,6 @@ decodeEndPointAddress (EndPointAddress bs) =
904899
905900-- | Do a (blocking) remote request
906901--
907−-- RELY: Remote endpoint must be in valid or closed state.
908−-- GUARANTEE: Will not change the state of the remote endpoint.
909−--
910902-- May throw IO (user) exception if the local or the remote endpoint is closed,
911903-- or if the send fails.
912904doRemoteRequest :: EndPointPair -> ControlHeader -> IO [ByteString]
@@ -1147,7 +1139,7 @@ handleConnectionRequest transport sock = handle handleException $ do
11471139 go ourEndPoint theirAddress = do
11481140 mEndPoint <- handle ((>> return Nothing) . invalidEndPoint) $ do
11491141 (theirEndPoint, isNew) <-
1150− findRemoteEndPoint ourEndPoint theirAddress False
1142+ findRemoteEndPoint ourEndPoint theirAddress DontBlockOnInit
11511143 (throw $ userError "Already connected")
11521144
11531145 let ourAddress = localAddress ourEndPoint
@@ -1166,9 +1158,7 @@ handleConnectionRequest transport sock = handle handleException $ do
11661158 , _remoteIncoming = IntSet.empty
11671159 , sendOn = sendMany sock
11681160 }
1169− modifyMVar_ theirState $ \(RemoteEndPointInit resolved) -> do
1170− putMVar resolved ()
1171− return (RemoteEndPointValid vst)
1161+ resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)
11721162 sendMany sock [encodeInt32 ConnectionRequestAccepted]
11731163 return (Just theirEndPoint)
11741164 -- If we left the scope of the exception handler with a return value of
@@ -1186,15 +1176,17 @@ handleConnectionRequest transport sock = handle handleException $ do
11861176 tryIO $ sendMany sock [encodeInt32 ConnectionRequestInvalid]
11871177 tryCloseSocket sock
11881178
1179+data BlockOnInit = BlockOnInit | DontBlockOnInit
1180+
11891181-- | Find a remote endpoint. If the remote endpoint does not yet exist we
11901182-- create it in Init state and return the corresponding 'resolved' MVar.
11911183findRemoteEndPoint
1192− :: LocalEndPoint -- ^ Local endpoint
1193− -> EndPointAddress -- ^ Remote address
1194− -> Bool -- ^ Wait if remote endpoint is in Init state
1184+ :: LocalEndPoint
1185+ -> EndPointAddress
1186+ -> BlockOnInit
11951187 -> (ValidRemoteEndPointState -> RemoteState) -- ^ Apply if in Valid state
11961188 -> IO (RemoteEndPoint, Maybe (MVar ()))
1197−findRemoteEndPoint ourEndPoint theirAddress waitOnInit whenValid = go
1189+findRemoteEndPoint ourEndPoint theirAddress blockOnInit whenValid = go
11981190 where
11991191 go = do
12001192 (theirEndPoint, isNew) <- modifyMVar ourState $ \st -> case st of
@@ -1233,9 +1225,9 @@ findRemoteEndPoint ourEndPoint theirAddress waitOnInit whenValid = go
12331225 RemoteEndPointInvalid err ->
12341226 throwIO err
12351227 RemoteEndPointInit resolved ->
1236− if waitOnInit
1237− then readMVar resolved >> go
1238− else return (theirEndPoint, Nothing)
1228+ case blockOnInit of
1229+ BlockOnInit -> readMVar resolved >> go
1230+ DontBlockOnInit -> return (theirEndPoint, Nothing)
12391231 RemoteEndPointValid _ ->
12401232 return (theirEndPoint, Nothing)
12411233 RemoteEndPointClosing resolved _ ->
@@ -1248,9 +1240,6 @@ findRemoteEndPoint ourEndPoint theirAddress waitOnInit whenValid = go
12481240-- | Handle requests from a remote endpoint.
12491241--
12501242-- Returns only if the remote party closes the socket or if an error occurs.
1251−--
1252−-- RELY: The remote endpoint must not be invalid.
1253−-- GUARANTEE: May change the remote endpoint to RemoteEndPointClosed state.
12541243handleIncomingMessages :: EndPointPair -> IO ()
12551244handleIncomingMessages (ourEndPoint, theirEndPoint) = do
12561245 mSock <- withMVar theirState $ \st ->
network-transport/tests/TestTCP.hs+4−4
@@ -654,7 +654,7 @@ testUnidirectionalError nextPort = do
654654 takeMVar serverGotPing
655655
656656 -- Close the *outgoing* part of the socket only
657− Right sock <- socketBetween internals (address endpoint) theirAddr
657+ sock <- socketBetween internals (address endpoint) theirAddr
658658 N.shutdown sock N.ShutdownSend
659659
660660 -- At this point we cannot notice the problem yet so we shouldn't receive an event yet
@@ -670,7 +670,7 @@ testUnidirectionalError nextPort = do
670670 takeMVar serverGotPing
671671
672672 -- Again, close the outgoing part of the socket
673− Right sock' <- socketBetween internals (address endpoint) theirAddr
673+ sock' <- socketBetween internals (address endpoint) theirAddr
674674 N.shutdown sock' N.ShutdownSend
675675
676676 -- We now find the error when we attempt to close the connection
@@ -682,7 +682,7 @@ testUnidirectionalError nextPort = do
682682 takeMVar serverGotPing
683683
684684 -- We repeat once more.
685− Right sock'' <- socketBetween internals (address endpoint) theirAddr
685+ sock'' <- socketBetween internals (address endpoint) theirAddr
686686 N.shutdown sock'' N.ShutdownSend
687687
688688 -- Now we notice the problem when we try to connect
@@ -727,7 +727,7 @@ testInvalidCloseConnection nextPort = do
727727 Right _ <- connect endpoint theirAddr ReliableOrdered defaultConnectHints
728728
729729 -- Get a handle on the TCP connection and manually send an invalid CloseConnection request
730− Right sock <- socketBetween internals ourAddr theirAddr
730+ sock <- socketBetween internals ourAddr theirAddr
731731 sendMany sock [encodeInt32 CloseConnection, encodeInt32 (12345 :: Int)]
732732
733733 putMVar clientDone ()
network-transport/tests/TestTransport.hs+1−1
@@ -49,7 +49,7 @@ echoServer endpoint = do
4949 -- Ignore
5050 go cs
5151 ErrorEvent _ ->
52− fail (show event)
52+ putStrLn $ "Echo server received error event: " ++ show event
5353 EndPointClosed ->
5454 return ()
5555