@@ -25,17 +25,24 @@ import Test.HUnit (Assertion, assertFailure)
2525import Data.Map (Map)
2626import qualified Data.Map as Map
2727import Control.Category ((>>>))
28+import Control.Arrow (second)
2829import Control.Applicative ((<$>))
29−import Control.Exception (Exception, throwIO)
30−import Control.Concurrent (forkIO, threadDelay)
31−import Control.Concurrent.Chan (Chan, newChan, writeChan, readChan)
32−import Control.Monad (replicateM, void)
33−import Data.List (inits, delete)
30+import Control.Exception (throwIO)
31+import Control.Concurrent (forkIO, threadDelay, ThreadId, killThread)
32+import Control.Monad (replicateM, forever, guard)
33+import Control.Monad.State.Lazy (StateT, execStateT)
34+import Control.Monad.IO.Class (liftIO)
35+import Data.Maybe (isJust)
36+import Data.List (inits)
3437import Data.ByteString (ByteString)
3538import Data.ByteString.Char8 (pack)
36−import Data.Accessor (Accessor, accessor, (^:), (^.), (^=))
39+import Data.Accessor (Accessor, accessor, (^.))
40+import Data.Accessor.Monad.Trans.State (get, modify)
41+import qualified Data.Accessor.Container as DAC (mapDefault)
3742import qualified Data.ByteString as BSS (concat)
3843import qualified Text.PrettyPrint as PP
44+import Data.Unique (Unique, newUnique, hashUnique)
45+import Data.Concurrent.Queue.MichaelScott (newQ, pushL, tryPopR)
3946
4047import Network.Transport
4148import Network.Transport.TCP
@@ -82,165 +89,179 @@ data ScriptCmd =
8289
8390type Script = [ScriptCmd]
8491
85−verticalList :: [PP.Doc] -> PP.Doc
86−verticalList = PP.brackets . PP.vcat
92+--------------------------------------------------------------------------------
93+-- Execute and verify scripts --
94+--------------------------------------------------------------------------------
8795
88−eventsDoc :: Map Int [Event] -> PP.Doc
89−eventsDoc = verticalList . map aux . Map.toList
90− where
91− aux :: (Int, [Event]) -> PP.Doc
92− aux (i, evs) = PP.parens . PP.hsep . PP.punctuate PP.comma $ [PP.int i, verticalList (map (PP.text . show) evs)]
96+data Variable a = Value a | Variable Unique
97+ deriving Eq
9398
99+instance Show a => Show (Variable a) where
100+ show (Value x) = show x
101+ show (Variable u) = "<<" ++ show (hashUnique u) ++ ">>"
94102
95−instance Show Script where
96− show = ("\n" ++) . show . verticalList . map (PP.text . show)
103+data ExpEvent =
104+ ExpConnectionOpened (Variable ConnectionId)
105+ | ExpConnectionClosed (Variable ConnectionId)
106+ | ExpReceived (Variable ConnectionId) [ByteString]
107+ deriving Show
108+
109+type TargetAddress = EndPointAddress
97110
98−instance Show (Map Int [Event]) where
99− show = ("\n" ++) . show . eventsDoc
111+data RunState = RunState {
112+ _endPoints :: [EndPoint]
113+ , _connections :: [(TargetAddress, Connection, Variable ConnectionId)]
114+ , _expectedEvents :: Map EndPointAddress [ExpEvent]
115+ , _forwardingThreads :: [ThreadId]
116+ }
117+
118+initialRunState :: RunState
119+initialRunState = RunState {
120+ _endPoints = []
121+ , _connections = []
122+ , _expectedEvents = Map.empty
123+ , _forwardingThreads = []
124+ }
125+
126+verify :: (Transport, TransportInternals) -> Script -> IO (Either String ())
127+verify (transport, transportInternals) script = do
128+ allEvents <- newQ
129+
130+ let runScript :: Script -> StateT RunState IO ()
131+ runScript = mapM_ runCmd
132+
133+ runCmd :: ScriptCmd -> StateT RunState IO ()
134+ runCmd NewEndPoint = do
135+ mEndPoint <- liftIO $ newEndPoint transport
136+ case mEndPoint of
137+ Right endPoint -> do
138+ modify endPoints (snoc endPoint)
139+ tid <- liftIO $ forkIO (forward endPoint)
140+ modify forwardingThreads (tid :)
141+ Left err ->
142+ liftIO $ throwIO err
143+ runCmd (Connect i j) = do
144+ endPointA <- get (endPointAtIx i)
145+ endPointB <- address <$> get (endPointAtIx j)
146+ mConn <- liftIO $ connect endPointA endPointB ReliableOrdered defaultConnectHints
147+ case mConn of
148+ Right conn -> do
149+ connId <- Variable <$> liftIO newUnique
150+ modify connections (snoc (endPointB, conn, connId))
151+ modify (expectedEventsAt endPointB) (snoc (ExpConnectionOpened connId))
152+ Left err ->
153+ liftIO $ throwIO err
154+ runCmd (Close i) = do
155+ (target, conn, connId) <- get (connectionAt i)
156+ liftIO $ close conn
157+ modify (expectedEventsAt target) (snoc (ExpConnectionClosed connId))
158+ runCmd (Send i payload) = do
159+ (target, conn, connId) <- get (connectionAt i)
160+ mResult <- liftIO $ send conn payload
161+ case mResult of
162+ Right () -> return ()
163+ Left err -> liftIO $ throwIO err
164+ modify (expectedEventsAt target) (snoc (ExpReceived connId payload))
100165
101−--------------------------------------------------------------------------------
102−-- Execution --
103−--------------------------------------------------------------------------------
166+ forward :: EndPoint -> IO ()
167+ forward endPoint = forever $ do
168+ ev <- receive endPoint
169+ pushL allEvents (address endPoint, ev)
104170
105−-- | Execute a script
106−--
107−-- Execute ignores error codes reported back. Instead, we verify the events
108−-- that are posted
109−execScript :: (Transport, TransportInternals) -> Script -> IO (Map EndPointIx [Event], VerificationState)
110−execScript (transport, transportInternals) script = do
111− chan <- newChan
112− vst <- runScript chan script
113− evs <- collectAll chan
114− return (evs, vst)
115− where
116− runScript :: Chan (Maybe (EndPointIx, Event)) -> Script -> IO VerificationState
117− runScript chan = go [] []
118− where
119− go :: [EndPoint] -> [Either (TransportError ConnectErrorCode) Connection] -> Script -> IO VerificationState
120− go endPoints _conns [] = do
121− threadDelay 10000
122− writeChan chan Nothing
123− return (initialVerificationState (map address endPoints))
124− go endPoints conns (NewEndPoint : cmds) = do
125− endPoint <- throwIfLeft $ newEndPoint transport
126− let endPointIx = length endPoints
127− _tid <- forkIO $ forwardTo chan (endPointIx, endPoint)
128− threadDelay 10000
129− go (endPoint `snoc` endPoints) conns cmds
130− go endPoints conns (Connect fr to : cmds) = do
131− conn <- connect (endPoints !! fr) (address (endPoints !! to)) ReliableOrdered defaultConnectHints
171+ collectEvents :: RunState -> IO (Map EndPointAddress [Event])
172+ collectEvents st = do
132173 threadDelay 10000
133− go endPoints (conn `snoc` conns) cmds
134− go endPoints conns (Close connIx : cmds) = do
135− case conns !! connIx of
136− Left _err -> return ()
137− Right conn -> close conn
138− threadDelay 10000
139− go endPoints conns cmds
140− go endPoints conns (Send connIx payload : cmds) = do
141− case conns !! connIx of
142− Left _err -> return ()
143− Right conn -> void $ send conn payload
144− threadDelay 10000
145− go endPoints conns cmds
146− go endPoints conns (BreakAfterReads n i j : cmds) = do
147− sock <- socketBetween transportInternals (address (endPoints !! i)) (address (endPoints !! j))
148− scheduleReadAction sock n (putStrLn "Closing" >> sClose sock)
149− go endPoints conns cmds
150−
151− forwardTo :: Chan (Maybe (EndPointIx, Event)) -> (EndPointIx, EndPoint) -> IO ()
152− forwardTo chan (ix, endPoint) = go
153− where
154− go :: IO ()
155− go = do
156− ev <- receive endPoint
157− case ev of
158− EndPointClosed -> return ()
159− _ -> writeChan chan (Just (ix, ev)) >> go
160−
161− collectAll :: Chan (Maybe (EndPointIx, Event)) -> IO (Map EndPointIx [Event])
162− collectAll chan = go Map.empty
163− where
164− go :: Map Int [Event] -> IO (Map Int [Event])
165− go acc = do
166− mEv <- readChan chan
167− case mEv of
168− Nothing -> return $ Map.map reverse acc
169− Just (ix, ev) -> go (Map.alter (insertEvent ev) ix acc)
170−
171− insertEvent :: Event -> Maybe [Event] -> Maybe [Event]
172− insertEvent ev Nothing = Just [ev]
173− insertEvent ev (Just evs) = Just (ev : evs)
174+ mapM_ killThread (st ^. forwardingThreads)
175+ evs <- go []
176+ return (groupByKey evs)
177+ where
178+ go acc = do
179+ mEv <- tryPopR allEvents
180+ case mEv of
181+ Just ev -> go (ev : acc)
182+ Nothing -> return acc
183+
184+ st <- execStateT (runScript script) initialRunState
185+ actualEvents <- collectEvents st
186+
187+ let eventsMatch = and . map (uncurry match) $
188+ zip (Map.elems (st ^. expectedEvents))
189+ (Map.elems actualEvents)
190+
191+ return $ if eventsMatch
192+ then Right ()
193+ else Left ("Could not match " ++ show (st ^. expectedEvents)
194+ ++ " and " ++ show actualEvents)
174195
175196--------------------------------------------------------------------------------
176−-- Verification --
197+-- Match expected and actual events --
177198--------------------------------------------------------------------------------
178199
179−data VerificationState = VerificationState {
180− endPointAddrs :: [EndPointAddress]
181− , _connections :: [(SourceEndPointIx, TargetEndPointIx, ConnectionId)]
182− , _mayBreak :: [(SourceEndPointIx, TargetEndPointIx)]
183− }
184−
185−initialVerificationState :: [EndPointAddress] -> VerificationState
186−initialVerificationState addrs = VerificationState {
187− endPointAddrs = addrs
188− , _connections = []
189− , _mayBreak = []
190− }
191−
192−-- TODO: we currently have no way to verify addresses in ConnectionOpened
193−-- or EventConnectionLost (because we don't know the addresses of the endpoints)
194−verify :: VerificationState -> Script -> Map EndPointIx [Event] -> Maybe String
195−verify _st [] evs =
196− -- TODO: we should compare the error events against mayBreak, but we
197− -- cannot because we don't know the endpoint addresses
198− if removeErrorEvents (concat (Map.elems evs)) == []
199− then Nothing
200− else Just $ "Unexpected events: " ++ show evs
201−verify st (NewEndPoint : cmds) evs =
202− verify st cmds evs
203−verify st (Connect fr to : cmds) evs =
204− case destruct evs to of
205− Just (ConnectionOpened connId _rel _addr, evs') ->
206− verify (connections ^: snoc (fr, to, connId) $ st) cmds evs'
207− ev ->
208− Just $ "Missing (ConnectionOpened <<connId>> <<rel>> <<addr>>). Got " ++ show ev
209−verify st (Close connIx : cmds) evs =
210− let (_fr, to, connId) = st ^. connectionAt connIx in
211− case destruct evs to of
212− Just (ConnectionClosed connId', evs') | connId' == connId ->
213− verify st cmds evs'
214− ev ->
215− Just $ "Missing (ConnectionClosed " ++ show connId ++ "). Got " ++ show ev
216−verify st (Send connIx payload : cmds) evs =
217− let (fr, to, connId) = st ^. connectionAt connIx in
218− case destruct evs to of
219− Just (Received connId' payload', evs') | connId' == connId && BSS.concat payload == BSS.concat payload' ->
220− verify st cmds evs'
221− Just (ErrorEvent (TransportError (EventConnectionLost _addr) _), evs') | st ^. mayBreak fr to ->
222− verify st cmds evs'
223− ev ->
224− Just $ "Missing (Received " ++ show connId ++ " " ++ show payload ++ "). Got " ++ show ev
225−verify st (BreakAfterReads _n i j : cmds) evs =
226− verify (mayBreak i j ^= True $ st) cmds evs
227−
228−connections :: Accessor VerificationState [(SourceEndPointIx, TargetEndPointIx, ConnectionId)]
229−connections = accessor _connections (\cs st -> st { _connections = cs })
230−
231−connectionAt :: ConnectionIx -> Accessor VerificationState (SourceEndPointIx, TargetEndPointIx, ConnectionId)
232−connectionAt i = connections >>> listAccessor i
200+-- | Match a list of expected events to a list of actual events, taking into
201+-- account that events may be reordered
202+match :: [ExpEvent] -> [Event] -> Bool
203+match expected actual = or (map (isJust . flip unify actual) (reorder expected))
204+
205+-- | Match a list of expected events to a list of actual events, without doing
206+-- reordering
207+unify :: [ExpEvent] -> [Event] -> Maybe ()
208+unify [] [] = return ()
209+unify (ExpConnectionOpened connId : expected) (ConnectionOpened connId' _ _ : actual) = do
210+ subst <- unifyConnectionId connId connId'
211+ unify (apply subst expected) actual
212+unify (ExpConnectionClosed connId : expected) (ConnectionClosed connId' : actual) = do
213+ subst <- unifyConnectionId connId connId'
214+ unify (apply subst expected) actual
215+unify (ExpReceived connId payload : expected) (Received connId' payload' : actual) = do
216+ guard (BSS.concat payload == BSS.concat payload')
217+ subst <- unifyConnectionId connId connId'
218+ unify (apply subst expected) actual
219+unify _ _ = fail "Cannot unify"
220+
221+type Substitution a = Map Unique a
222+
223+-- | Match two connection IDs
224+unifyConnectionId :: Variable ConnectionId -> ConnectionId -> Maybe (Substitution ConnectionId)
225+unifyConnectionId (Variable x) connId = Just $ Map.singleton x connId
226+unifyConnectionId (Value connId') connId | connId == connId' = Just Map.empty
227+ | otherwise = Nothing
228+
229+-- | Apply a substitution
230+apply :: Substitution ConnectionId -> [ExpEvent] -> [ExpEvent]
231+apply subst = map applyEvent
232+ where
233+ applyEvent :: ExpEvent -> ExpEvent
234+ applyEvent (ExpConnectionOpened connId) = ExpConnectionOpened (applyVar connId)
235+ applyEvent (ExpConnectionClosed connId) = ExpConnectionClosed (applyVar connId)
236+ applyEvent (ExpReceived connId payload) = ExpReceived (applyVar connId) payload
237+
238+ applyVar :: Variable ConnectionId -> Variable ConnectionId
239+ applyVar (Value connId) = Value connId
240+ applyVar (Variable x) = case Map.lookup x subst of
241+ Just connId -> Value connId
242+ Nothing -> Variable x
243+
244+-- | Return all possible reorderings of a list of expected events
245+--
246+-- Events from different connections can be reordered, but events from the
247+-- same connection cannot.
248+reorder :: [ExpEvent] -> [[ExpEvent]]
249+reorder = go
250+ where
251+ go :: [ExpEvent] -> [[ExpEvent]]
252+ go [] = [[]]
253+ go (ev : evs) = concat [insert ev evs' | evs' <- reorder evs]
233254
234−mayBreak :: EndPointIx -> EndPointIx -> Accessor VerificationState Bool
235−mayBreak i j = accessor
236− (\st -> (i, j) `elem` _mayBreak st || (j, i) `elem` _mayBreak st)
237− (\b st -> if b then st { _mayBreak = (i, j) : _mayBreak st }
238− else st { _mayBreak = delete (i, j) . delete (j, i) $ _mayBreak st })
255+ insert :: ExpEvent -> [ExpEvent] -> [[ExpEvent]]
256+ insert ev [] = [[ev]]
257+ insert ev (ev' : evs')
258+ | connectionId ev == connectionId ev' = [ev : ev' : evs']
259+ | otherwise = (ev : ev' : evs') : [ev' : evs'' | evs'' <- insert ev evs']
239260
240−removeErrorEvents :: [Event] -> [Event]
241−removeErrorEvents [] = []
242−removeErrorEvents (ErrorEvent _ : evs) = removeErrorEvents evs
243−removeErrorEvents (ev : evs) = ev : removeErrorEvents evs
261+ connectionId :: ExpEvent -> Variable ConnectionId
262+ connectionId (ExpConnectionOpened connId) = connId
263+ connectionId (ExpConnectionClosed connId) = connId
264+ connectionId (ExpReceived connId _) = connId
244265
245266--------------------------------------------------------------------------------
246267-- Script generators --
@@ -472,21 +493,61 @@ testScriptGen transport scriptGen =
472493 forAll scriptGen $ \script ->
473494 morallyDubiousIOProperty $ do
474495 logShow script
475− (evs, vst) <- execScript transport script
476− return $ case verify vst script evs of
477− Nothing -> result { ok = Just True
478− }
479− Just err -> result { ok = Just False
480− , reason = '\n' : err ++ "\nAll events: " ++ show evs
496+ mErr <- verify transport script
497+ return $ case mErr of
498+ Right () -> result { ok = Just True }
499+ Left err -> result { ok = Just False
500+ , reason = '\n' : err ++ "\n"
481501 }
482502
483503testScript :: (Transport, TransportInternals) -> Script -> Assertion
484504testScript transport script = do
485505 logShow script
486− (evs, vst) <- execScript transport script
487− case verify vst script evs of
488− Just err -> assertFailure $ "Failed with script " ++ show script ++ ": " ++ err ++ "\nAll events: " ++ show evs
489− Nothing -> return ()
506+ mErr <- verify transport script
507+ case mErr of
508+ Left err -> assertFailure $ "Failed with script " ++ show script ++ ": " ++ err ++ "\n"
509+ Right () -> return ()
510+
511+--------------------------------------------------------------------------------
512+-- Accessors --
513+--------------------------------------------------------------------------------
514+
515+endPoints :: Accessor RunState [EndPoint]
516+endPoints = accessor _endPoints (\es st -> st { _endPoints = es })
517+
518+endPointAtIx :: EndPointIx -> Accessor RunState EndPoint
519+endPointAtIx i = endPoints >>> listAccessor i
520+
521+connections :: Accessor RunState [(TargetAddress, Connection, Variable ConnectionId)]
522+connections = accessor _connections (\cs st -> st { _connections = cs })
523+
524+connectionAt :: ConnectionIx -> Accessor RunState (TargetAddress, Connection, Variable ConnectionId)
525+connectionAt i = connections >>> listAccessor i
526+
527+expectedEvents :: Accessor RunState (Map EndPointAddress [ExpEvent])
528+expectedEvents = accessor _expectedEvents (\es st -> st { _expectedEvents = es })
529+
530+expectedEventsAt :: EndPointAddress -> Accessor RunState [ExpEvent]
531+expectedEventsAt addr = expectedEvents >>> DAC.mapDefault [] addr
532+
533+forwardingThreads :: Accessor RunState [ThreadId]
534+forwardingThreads = accessor _forwardingThreads (\ts st -> st { _forwardingThreads = ts })
535+
536+--------------------------------------------------------------------------------
537+-- Pretty-printing --
538+--------------------------------------------------------------------------------
539+
540+verticalList :: Show a => [a] -> PP.Doc
541+verticalList = PP.brackets . PP.vcat . map (PP.text . show)
542+
543+instance Show Script where
544+ show = ("\n" ++) . show . verticalList
545+
546+instance Show [Event] where
547+ show = ("\n" ++) . show . verticalList
548+
549+instance Show [ExpEvent] where
550+ show = ("\n" ++) . show . verticalList
490551
491552--------------------------------------------------------------------------------
492553-- Draw random values from probability distributions --
@@ -529,13 +590,6 @@ log = appendFile "log" . (++ "\n")
529590logShow :: Show a => a -> IO ()
530591logShow = log . show
531592
532−throwIfLeft :: Exception a => IO (Either a b) -> IO b
533−throwIfLeft p = do
534− mb <- p
535− case mb of
536− Left a -> throwIO a
537− Right b -> return b
538−
539593instance Arbitrary ByteString where
540594 arbitrary = do
541595 len <- chooseFrom' (NormalD { mean = 5, stdDev = 10 }) (0, 100)
@@ -548,8 +602,5 @@ listAccessor i = accessor (!! i) (error "listAccessor.set not defined")
548602snoc :: a -> [a] -> [a]
549603snoc x xs = xs ++ [x]
550604
551−destruct :: Map EndPointIx [Event] -> EndPointIx -> Maybe (Event, Map EndPointIx [Event])
552−destruct evs i =
553− case evs Map.! i of
554− [] -> Nothing
555− ev : evs' -> Just (ev, Map.insert i evs' evs)
605+groupByKey :: Ord a => [(a, b)] -> Map a [b]
606+groupByKey = Map.fromListWith (++) . map (second return)