{-# LANGUAGE CPP #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE DeriveDataTypeable #-}
{-# LANGUAGE DeriveGeneric #-}
module Main where
import Control.Concurrent.MVar
import Control.Concurrent.STM.TQueue
( newTQueueIO
, readTQueue
, writeTQueue
)
import Control.Exception (SomeException)
import Control.DeepSeq (NFData)
import Control.Distributed.Process hiding (call, send, catch, sendChan, wrapMessage)
import Control.Distributed.Process.Node
import Control.Distributed.Process.Extras hiding (__remoteTable, monitor)
import Control.Distributed.Process.Async hiding (check)
import Control.Distributed.Process.ManagedProcess hiding (reject, Message)
import qualified Control.Distributed.Process.ManagedProcess.Server.Priority as P (Message)
import Control.Distributed.Process.ManagedProcess.Server.Priority hiding (Message)
import qualified Control.Distributed.Process.ManagedProcess.Server.Gen as Gen
( dequeue
, continue
, lift
)
import Control.Distributed.Process.SysTest.Utils
import Control.Distributed.Process.Extras.Time
import Control.Distributed.Process.Extras.Timer hiding (runAfter)
import Control.Distributed.Process.Serializable()
import Control.Monad
import Control.Monad.Catch (catch)
import Data.Binary
import Data.Either (rights)
import Data.List (isInfixOf)
import Data.Maybe (isNothing, isJust)
import Data.Typeable (Typeable)
import Test.Tasty (TestTree, testGroup)
import Test.Tasty.HUnit (testCase)
import TestUtils
import ManagedProcessCommon
import qualified Network.Transport as NT
import GHC.Generics (Generic)
-- utilities
server :: Process (ProcessId, (MVar ExitReason))
server = mkServer Terminate
mkServer :: UnhandledMessagePolicy
-> Process (ProcessId, (MVar ExitReason))
mkServer policy =
let s = standardTestServer policy
p = s `prioritised` ([] :: [DispatchPriority ()])
in do
exitReason <- liftIO $ newEmptyMVar
pid <- spawnLocal $ do
catch ((pserve () (statelessInit Infinity) p >> stash exitReason ExitNormal)
`catchesExit` [
(\_ msg -> do
mEx <- unwrapMessage msg :: Process (Maybe ExitReason)
case mEx of
Nothing -> return Nothing
Just r -> stash exitReason r >>= return . Just
)
])
(\(e :: SomeException) -> stash exitReason $ ExitOther (show e))
return (pid, exitReason)
explodingServer :: ProcessId
-> Process (ProcessId, MVar ExitReason)
explodingServer pid =
let srv = explodingTestProcess pid
pSrv = srv `prioritised` ([] :: [DispatchPriority s])
in do
exitReason <- liftIO newEmptyMVar
spid <- spawnLocal $ do
catch (pserve () (statelessInit Infinity) pSrv >> stash exitReason ExitNormal)
(\(e :: SomeException) -> do
-- say "died in handler..."
stash exitReason $ ExitOther (show e))
return (spid, exitReason)
data GetState = GetState
deriving (Typeable, Generic, Show, Eq)
instance Binary GetState where
instance NFData GetState where
data MyAlarmSignal = MyAlarmSignal
deriving (Typeable, Generic, Show, Eq)
instance Binary MyAlarmSignal where
instance NFData MyAlarmSignal where
mkPrioritisedServer :: Process ProcessId
mkPrioritisedServer =
let p = procDef `prioritised` ([
prioritiseInfo_ (\MyAlarmSignal -> setPriority 10)
, prioritiseCast_ (\(_ :: String) -> setPriority 2)
, prioritiseCall_ (\(cmd :: String) -> (setPriority (length cmd)) :: Priority ())
] :: [DispatchPriority [Either MyAlarmSignal String]]
) :: PrioritisedProcessDefinition [(Either MyAlarmSignal String)]
in spawnLocal $ pserve () (initWait Infinity) p
where
initWait :: Delay
-> InitHandler () [Either MyAlarmSignal String]
initWait d () = do
() <- expect
return $ InitOk [] d
procDef :: ProcessDefinition [(Either MyAlarmSignal String)]
procDef =
defaultProcess {
apiHandlers = [
handleCall (\s GetState -> reply (reverse s) s)
, handleCall (\s (cmd :: String) -> reply () ((Right cmd):s))
, handleCast (\s (cmd :: String) -> continue ((Right cmd):s))
]
, infoHandlers = [
handleInfo (\s (sig :: MyAlarmSignal) -> continue ((Left sig):s))
]
, unhandledMessagePolicy = Drop
, timeoutHandler = \_ _ -> stop $ ExitOther "timeout"
} :: ProcessDefinition [(Either MyAlarmSignal String)]
mkOverflowHandlingServer :: (PrioritisedProcessDefinition Int ->
PrioritisedProcessDefinition Int)
-> Process ProcessId
mkOverflowHandlingServer modIt =
let p = procDef `prioritised` ([
prioritiseCall_ (\GetState -> setPriority 99 :: Priority Int)
, prioritiseCast_ (\(_ :: String) -> setPriority 1)
] :: [DispatchPriority Int]
) :: PrioritisedProcessDefinition Int
in spawnLocal $ pserve () (initWait Infinity) (modIt p)
where
initWait :: Delay
-> InitHandler () Int
initWait d () = return $ InitOk 0 d
procDef :: ProcessDefinition Int
procDef =
defaultProcess {
apiHandlers = [
handleCall (\s GetState -> reply s s)
, handleCast (\s (_ :: String) -> continue $ s + 1)
]
} :: ProcessDefinition Int
launchStmServer :: CallHandler () String String -> Process StmServer
launchStmServer handler = do
(inQ, replyQ) <- liftIO $ do
cIn <- newTQueueIO
cOut <- newTQueueIO
return (cIn, cOut)
let procDef = statelessProcess {
externHandlers = [
handleCallExternal
(readTQueue inQ)
(writeTQueue replyQ)
handler
]
, apiHandlers = [
action (\() -> stop_ ExitNormal)
]
}
let p = procDef `prioritised` ([
prioritiseCast_ (\() -> setPriority 99 :: Priority ())
, prioritiseCast_ (\(_ :: String) -> setPriority 100)
] :: [DispatchPriority ()]
) :: PrioritisedProcessDefinition ()
pid <- spawnLocal $ pserve () (statelessInit Infinity) p
return $ StmServer pid inQ replyQ
launchStmOverloadServer :: Process (ProcessId, ControlPort String)
launchStmOverloadServer = do
cc <- newControlChan :: Process (ControlChannel String)
let cp = channelControlPort cc
let procDef = statelessProcess {
externHandlers = [
handleControlChan_ cc (\(_ :: String) -> continue_)
]
, apiHandlers = [
handleCast (\s sp -> sendChan sp () >> continue s)
]
}
let p = procDef `prioritised` ([
prioritiseCast_ (\() -> setPriority 99 :: Priority ())
] :: [DispatchPriority ()]
) :: PrioritisedProcessDefinition ()
pid <- spawnLocal $ pserve () (statelessInit Infinity) p
return (pid, cp)
data Foo = Foo deriving (Show)
launchFilteredServer :: ProcessId -> Process (ProcessId, ControlPort (SendPort Int))
launchFilteredServer us = do
cc <- newControlChan :: Process (ControlChannel (SendPort Int))
let cp = channelControlPort cc
let procDef = defaultProcess {
externHandlers = [
handleControlChan cc (\s (p :: SendPort Int) -> sendChan p s >> continue s)
]
, apiHandlers = [
handleCast (\s sp -> sendChan sp () >> continue s)
, handleCall_ (\(s :: String) -> return s)
, handleCall_ (\(i :: Int) -> return i)
]
, unhandledMessagePolicy = DeadLetter us
} :: ProcessDefinition Int
let p = procDef `prioritised` ([
prioritiseCast_ (\() -> setPriority 1 :: Priority ())
, prioritiseCall_ (\(_ :: String) -> setPriority 100 :: Priority String)
] :: [DispatchPriority Int]
) :: PrioritisedProcessDefinition Int
let rejectUnchecked =
rejectApi Foo :: Int -> P.Message String String -> Process (Filter Int)
let p' = p {
filters = [
store (+1)
, ensure (>0) -- a bit pointless, but we're just checking the API
, check $ api_ (\(s :: String) -> return $ "checked-" `isInfixOf` s) rejectUnchecked
, check $ info (\_ (_ :: MonitorRef, _ :: ProcessId) -> return False) $ reject Foo
, refuse ((> 10) :: Int -> Bool)
]
}
pid <- spawnLocal $ pserve 0 (\c -> return $ InitOk c Infinity) p'
return (pid, cp)
testStupidInfiniteLoop :: TestResult Bool -> Process ()
testStupidInfiniteLoop result = do
let def = statelessProcess {
apiHandlers = [
handleCast (\_ sp -> eval $ do q <- processQueue
m <- Gen.dequeue
Gen.lift $ sendChan sp (length q, m)
Gen.continue)
]
, infoHandlers = [
handleInfo (\_ (m :: String) -> eval $ do enqueue (wrapMessage m)
Gen.continue)
]
} :: ProcessDefinition ()
let prio = def `prioritised` []
pid <- spawnLocal $ pserve () (statelessInit Infinity) prio
-- this message should create an infinite loop
send pid "fooboo"
(sp, rp) <- newChan :: Process (SendPort (Int, Maybe Message), ReceivePort (Int, Maybe Message))
cast pid sp
(i, m) <- receiveChan rp
cast pid sp
(i', m') <- receiveChan rp
stash result $ (i == 1 && isJust m && i' == 0 && isNothing m')
testFilteringBehavior :: TestResult Bool -> Process ()
testFilteringBehavior result = do
us <- getSelfPid
(sp, rp) <- newChan
(pid, cp) <- launchFilteredServer us
mRef <- monitor pid
sendControlMessage cp sp
r <- receiveChan rp :: Process Int
when (r > 1) $ stash result False >> die "we're done..."
Left _ <- safeCall pid "bad-input" :: Process (Either ExitReason String)
send pid (mRef, us) -- server doesn't like this, dead letters it...
-- back to us
void $ receiveWait [ matchIf (\(m, p) -> m == mRef && p == us) return ]
sendControlMessage cp sp
r2 <- receiveChan rp :: Process Int
when (r2 < 3) $ stash result False >> die "we're done again..."
-- server also doesn't like this, and sends it right back (via \DeadLetter us/)
send pid (25 :: Int)
m <- receiveWait [ matchIf (== 25) return ] :: Process Int
stash result $ m == 25
kill pid "done"
testServerSwap :: TestResult Bool -> Process ()
testServerSwap result = do
us <- getSelfPid
let def2 = statelessProcess { apiHandlers = [ handleCast (\s (i :: Int) -> send us (i, i+1) >> continue s)
, handleCall_ (\(i :: Int) -> return (i * 5))
]
, unhandledMessagePolicy = Drop -- otherwise `call` would fail
}
let def = statelessProcess
{ apiHandlers = [ handleCall_ (\(m :: String) -> return m) ]
, infoHandlers = [ handleInfo (\s () -> become def2 s) ]
} `prioritised` []
pid <- spawnLocal $ pserve () (statelessInit Infinity) def
m1 <- call pid "hello there"
let a1 = m1 == "hello there"
send pid () --changeover
m2 <- callTimeout pid "are you there?" (seconds 5) :: Process (Maybe String)
let a2 = isNothing m2
cast pid (45 :: Int)
res <- receiveWait [ matchIf (\(i :: Int) -> i == 45) (return . Left)
, match (\(_ :: Int, j :: Int) -> return $ Right j) ]
let a3 = res == (Right 46)
m4 <- call pid (20 :: Int) :: Process Int
let a4 = m4 == 100
stash result $ a1 && a2 && a3 && a4
testSafeExecutionContext :: TestResult Bool -> Process ()
testSafeExecutionContext result = do
let t = (asTimeout $ seconds 5)
(sigSp, rp) <- newChan
(wp, lp) <- newChan
let def = statelessProcess
{ apiHandlers = [ handleCall_ (\(m :: String) -> stranded rp wp Nothing >> return m) ]
, infoHandlers = [ handleInfo (\s (m :: String) -> stranded rp wp (Just m) >> continue s) ]
, exitHandlers = [ handleExit (\_ s (_ :: String) -> continue s) ]
} `prioritised` []
let spec = def { filters = [
safe (\_ (_ :: String) -> True)
, apiSafe (\_ (_ :: String) (_ :: Maybe String) -> True)
]
}
pid <- spawnLocal $ pserve () (statelessInit Infinity) spec
send pid "hello" -- pid can't process this as it's stuck waiting on rp
sleep $ seconds 3
exit pid "ooops" -- now we force an exit signal once the receiveWait finishes
sendChan sigSp () -- and allow the receiveWait to complete
send pid "hi again"
-- at this point, "hello" should still be in the backing queue/mailbox
sleep $ seconds 3
-- We should still be seeing "hello", since the 'safe' block saved us from
-- losing a message when we handled and swallowed the exit signal.
-- We should not see "hi again" until after "hello" has been processed
h <- receiveChanTimeout t lp
-- say $ "first response: " ++ (show h)
let a1 = h == (Just "hello")
sleep $ seconds 3
-- now we should have "hi again" waiting in the mailbox...
sendChan sigSp () -- we must release the handler a second time...
h2 <- receiveChanTimeout t lp
-- say $ "second response: " ++ (show h2)
let a2 = h2 == (Just "hi again")
void $ spawnLocal $ call pid "reply-please" >>= sendChan wp
-- the call handler should be stuck waiting on rp
Nothing <- receiveChanTimeout (asTimeout $ seconds 2) lp
-- now let's force an exit, then release the handler to see if it runs again...
exit pid "ooops2"
sleep $ seconds 2
sendChan sigSp ()
h3 <- receiveChanTimeout t lp
-- say $ "third response: " ++ (show h3)
let a3 = h3 == (Just "reply-please")
stash result $ a1 && a2 && a3
where
stranded :: ReceivePort () -> SendPort String -> Maybe String -> Process ()
stranded gate chan str = do
-- say $ "stranded with " ++ (show str)
void $ receiveWait [ matchChan gate return ]
sleep $ seconds 1
case str of
Nothing -> return ()
Just s -> sendChan chan s
testExternalTimedOverflowHandling :: TestResult Bool -> Process ()
testExternalTimedOverflowHandling result = do
(pid, cp) <- launchStmOverloadServer -- default 10k mailbox drain limit
wrk <- spawnLocal $ mapM_ (sendControlMessage cp . show) ([1..500000] :: [Int])
sleep $ milliSeconds 250 -- give the worker time to start spamming the server...
(sp, rp) <- newChan
cast pid sp -- tell the server we're expecting a reply
-- it might take "a while" for us to get through the first 10k messages
-- from our chatty friend wrk, before we finally get our control message seen
-- by the reader/listener loop, and in fact timing wise we don't even know when
-- our message will arrive, since we're racing with wrk to communicate with
-- the server. It's important therefore to give sufficient time for the right
-- conditions to occur so that our message is finally received and processed,
-- yet we don't want to lock up the build for 10-20 mins either. This value
-- of 30 seconds seems like a reasonable compromise.
answer <- receiveChanTimeout (asTimeout $ seconds 30) rp
stash result $ answer == Just ()
kill wrk "done"
kill pid "done"
testExternalCall :: TestResult Bool -> Process ()
testExternalCall result = do
let txt = "hello stm-call foo"
srv <- launchStmServer (\st (msg :: String) -> reply msg st)
echoStm srv txt >>= stash result . (== Right txt)
killProc srv "done"
testTimedOverflowHandling :: TestResult Bool -> Process ()
testTimedOverflowHandling result = do
pid <- mkOverflowHandlingServer (\s -> s { recvTimeout = RecvTimer $ within 3 Seconds })
wrk <- spawnLocal $ mapM_ (cast pid . show) ([1..500000] :: [Int])
sleep $ seconds 1 -- give the worker time to start spamming us...
cast pid "abc" -- just getting in line here...
st <- call pid GetState :: Process Int
-- the result of GetState is a list of messages in reverse insertion order
stash result $ st > 0
kill wrk "done"
kill pid "done"
testOverflowHandling :: TestResult Bool -> Process ()
testOverflowHandling result = do
pid <- mkOverflowHandlingServer (\s -> s { recvTimeout = RecvMaxBacklog 100 })
wrk <- spawnLocal $ mapM_ (cast pid . show) ([1..50000] :: [Int])
sleep $ seconds 1
cast pid "abc" -- just getting in line here...
st <- call pid GetState :: Process Int
-- the result of GetState is a list of messages in reverse insertion order
stash result $ st > 0
kill wrk "done"
kill pid "done"
testInfoPrioritisation :: TestResult Bool -> Process ()
testInfoPrioritisation result = do
pid <- mkPrioritisedServer
-- the server (pid) is configured to wait for () during its init
-- so we can fill up its mailbox with String messages, and verify
-- that the alarm signal (which is prioritised *above* these)
-- actually gets processed first despite the delivery order
cast pid "hello"
cast pid "prioritised"
cast pid "world"
-- note that these have to be a "bare send"
send pid MyAlarmSignal
-- tell the server it can move out of init and start processing messages
send pid ()
st <- call pid GetState :: Process [Either MyAlarmSignal String]
-- the result of GetState is a list of messages in reverse insertion order
case head st of
Left MyAlarmSignal -> stash result True
_ -> stash result False
testUserTimerHandling :: TestResult Bool -> Process ()
testUserTimerHandling result = do
us <- getSelfPid
let p = (procDef us) `prioritised` ([
prioritiseInfo_ (\MyAlarmSignal -> setPriority 100)
] :: [DispatchPriority ()]
) :: PrioritisedProcessDefinition ()
pid <- spawnLocal $ pserve () (statelessInit Infinity) p
cast pid ()
expect >>= stash result . (== MyAlarmSignal)
kill pid "goodbye..."
where
procDef :: ProcessId -> ProcessDefinition ()
procDef us =
statelessProcess {
apiHandlers = [
handleCast (\s () -> evalAfter (seconds 5) MyAlarmSignal s)
]
, infoHandlers = [
handleInfo (\s (sig :: MyAlarmSignal) -> send us sig >> continue s)
]
, unhandledMessagePolicy = Drop
} :: ProcessDefinition ()
testCallPrioritisation :: TestResult Bool -> Process ()
testCallPrioritisation result = do
pid <- mkPrioritisedServer
asyncRefs <- (mapM (callAsync pid)
["first", "the longest", "commands", "we do prioritise"])
:: Process [Async ()]
-- NB: This sleep is really important - the `init' function is waiting
-- (selectively) on the () signal to go, and if it receives this *before*
-- the async worker has had a chance to deliver the longest string message,
-- our test will fail. Such races are /normal/ given that the async worker
-- runs in another process and delivery order between multiple processes
-- is undefined (and in practise, partially depenendent on the scheduler)
sleep $ seconds 1
send pid ()
_ <- mapM wait asyncRefs :: Process [AsyncResult ()]
st <- call pid GetState :: Process [Either MyAlarmSignal String]
let ms = rights st
stash result $ ms == ["we do prioritise", "the longest", "commands", "first"]
tests :: NT.Transport -> IO TestTree
tests transport = do
localNode <- newLocalNode transport initRemoteTable
return $ testGroup "" [
testGroup "basic server functionality matches un-prioritised processes" [
testCase "basic call with explicit server reply"
(delayedAssertion
"expected a response from the server"
localNode (Just "foo") (testBasicCall $ wrap server))
, testCase "basic call with implicit server reply"
(delayedAssertion
"expected n * 2 back from the server"
localNode (Just 4) (testBasicCall_ $ wrap server))
, testCase "basic deferred call handling"
(delayedAssertion "expected a response sent via replyTo"
localNode (AsyncDone "Hello There") testDeferredCallResponse)
, testCase "basic cast with manual send and explicit server continue"
(delayedAssertion
"expected pong back from the server"
localNode (Just "pong") (testBasicCast $ wrap server))
, testCase "cast and explicit server timeout"
(delayedAssertion
"expected the server to stop after the timeout"
localNode (Just $ ExitOther "timeout") (testControlledTimeout $ wrap server))
, testCase "unhandled input when policy = Terminate"
(delayedAssertion
"expected the server to stop upon receiving unhandled input"
localNode (Just $ ExitOther "UnhandledInput")
(testTerminatePolicy $ wrap server))
, testCase "unhandled input when policy = Drop"
(delayedAssertion
"expected the server to ignore unhandled input and exit normally"
localNode Nothing (testDropPolicy $ wrap (mkServer Drop)))
, testCase "unhandled input when policy = DeadLetter"
(delayedAssertion
"expected the server to forward unhandled messages"
localNode (Just ("UNSOLICITED_MAIL", 500 :: Int))
(testDeadLetterPolicy $ \p -> mkServer (DeadLetter p)))
, testCase "incoming messages are ignored whilst hibernating"
(delayedAssertion
"expected the server to remain in hibernation"
localNode True (testHibernation $ wrap server))
, testCase "long running call cancellation"
(delayedAssertion "expected to get AsyncCancelled"
localNode True (testKillMidCall $ wrap server))
, testCase "server rejects call"
(delayedAssertion "expected server to send CallRejected"
localNode (ExitOther "invalid-call") (testServerRejectsMessage $ wrap server))
, testCase "simple exit handling"
(delayedAssertion "expected handler to catch exception and continue"
localNode Nothing (testSimpleErrorHandling $ explodingServer))
, testCase "alternative exit handlers"
(delayedAssertion "expected handler to catch exception and continue"
localNode Nothing (testAlternativeErrorHandling $ explodingServer))
]
, testGroup "Prioritised Mailbox Handling" [
testCase "Info Message Prioritisation"
(delayedAssertion "expected the info handler to be prioritised"
localNode True testInfoPrioritisation)
, testCase "Call Message Prioritisation"
(delayedAssertion "expected the longest strings to be prioritised"
localNode True testCallPrioritisation)
, testCase "Size-Based Mailbox Overload Management"
(delayedAssertion "expected the server loop to stop reading the mailbox"
localNode True testOverflowHandling)
, testCase "Timeout-Based Mailbox Overload Management"
(delayedAssertion "expected the server loop to stop reading the mailbox"
localNode True testTimedOverflowHandling)
]
, testGroup "Advanced Server Interactions" [
testCase "using callSTM to manage non-CH interactions"
(delayedAssertion
"expected the server to reply back via the TQueue"
localNode True testExternalCall)
, testCase "Timeout-Based Overload Management with Control Channels"
(delayedAssertion "expected the server loop to reply"
localNode True testExternalTimedOverflowHandling)
, testCase "Complex pre/before filters"
(delayedAssertion "expected verifiable filter actions"
localNode True testFilteringBehavior)
, testCase "Firing internal timeouts"
(delayedAssertion "expected our info handler to run after the timeout"
localNode True testUserTimerHandling)
, testCase "Creating 'Safe' Handlers"
(delayedAssertion "expected our handler to run on the old message"
localNode True testSafeExecutionContext)
, testCase "Swapping ProcessDefinitions at runtime"
(delayedAssertion "expected our handler to exist in the new handler list"
localNode True testServerSwap)
, testCase "Accessing the internal process implementation"
(delayedAssertion "it should allow us to modify the internal q"
localNode True testStupidInfiniteLoop)
]
]
main :: IO ()
main = testMain $ tests