lattest-lib-0.1.1: src/Lattest/Adapter/StandardAdapters.hs
{- |
This module contains building blocks for constructing out-of-the-box 'Adapter's.
Note that adapters do not observe inputs by default. In other words, an input command
from a test controller does not automatically lead to an input transition in the specification
model. Use 'acceptingInputs' to observe all sent inputs, or 'acceptingInputsWithIncompletenessAsFailures'
to also observe failures in processing inputs in the adapter. The latter is useful if the processing
of inputs is defined partially, in which case some inputs cannot be sent to the adapter.
-}
-- TODO also expose a generic parsing transformation which takes a parser
module Lattest.Adapter.StandardAdapters (
-- * Type Definition
Adapter, -- re-export so that a user of the library who wants to set up a testing experiment doesn't need to import the Adapter module at all
-- * Ready-to-use Adapters
-- ** Pure adapters
-- | Adapters which are defined in haskell itself.
pureMealyAdapter,
pureAdapter,
-- ** Socket Adapters
-- *** Settings
SocketSettings(..),
baseSocketSettings,
-- *** Base Socket Adapters
connectSocketAdapter,
connectSocketAdapterWith,
-- *** Socket Adapters with JSON
connectJSONSocketAdapter,
connectJSONSocketAdapterWith,
connectJSONSocketAdapterAcceptingInputs,
connectJSONSocketAdapterAcceptingInputsWith,
-- * Transformations on Adapters
-- ** Encoding and Decoding
encodeUtf8,
decodeUtf8,
endecodeUtf8,
encodeJSONTestChoices,
parseJSONActionsFromSut,
-- ** Observing Inputs
acceptingInputs,
acceptingInputsWithIncompletenessAsFailures,
-- ** Timing and Quiescence
withQuiescence,
withQuiescenceMillis,
withInputDelay,
withInputDelayMillis,
withSuccessiveInputDelay,
withSuccessiveInputDelayMillis,
-- ** Adapters with Data Parameters
asSymbolicSuspAdapter,
connectJSONSocketAdapterSTSwithQuiescence
)
where
import Lattest.Adapter.Adapter as Adap (Adapter(..),parseActionsFromSut,mapTestChoices,mapActionsFromSut, map)
import Lattest.Model.Alphabet(TestChoice, choiceToActs, IOAct(..), Suspended(..), IOSuspAct, asSuspended, IOSuspGateValue, GateValue(..), SuspendedIF, isOutput, fromOutput, IFAct, InputAttempt(..))
import Lattest.Util.IOUtils(statefulIO', doAfter, ifM_, waitUntil)
import Lattest.Util.Utils(flipCoin, takeRandom)
import Control.Concurrent.STM.TMVar(newEmptyTMVarIO, tryReadTMVar, writeTMVar, readTMVar, isEmptyTMVar, putTMVar)
import Control.Concurrent.STM.TQueue(newTQueueIO, readTQueue, writeTQueue, isEmptyTQueue)
import Control.Concurrent.Thread.Delay(delay)
import Control.Exception(handle,PatternMatchFail)
import Control.Monad(forever,void)
import Control.Monad.Extra ((||^), (&&^))
import Data.Aeson(fromJSON,encode,Result(Error, Success), FromJSON, ToJSON)
import Data.Aeson.Parser(jsonNoDup)
import Data.Bits.Utils(c2w8)
import Data.ByteString (ByteString)
import Data.ByteString.Lazy(toStrict,snoc)
import Data.List(singleton)
import Data.Time.Clock(getCurrentTime,addUTCTime,diffUTCTime,NominalDiffTime,secondsToNominalDiffTime,nominalDiffTimeToSeconds)
import Debug.Trace(trace) -- FIXME find a better alternative
import GHC.Conc(forkIO, newTVarIO, TVar, retry, atomically, writeTVar, readTVar, STM, orElse, killThread)
import Network.Socket(HostName, PortNumber)
import Network.Utils (niceSocketsDo, connectTCP)
import System.IO.Streams as Streams (makeOutputStream, OutputStream, write, writeTo)
import System.IO.Streams.Combinators(contramap)
import System.IO.Streams.Network(socketToStreams)
import Lattest.Streams.Synchronized(TInputStream, duplicate, fromBuffer, mergeBufferedWith, mapUnbuffered, fromTMVar, readAll, fromInputStreamBuffered, makeTInputStream, hasInput)
import qualified Lattest.Streams.Synchronized as Streams (read, unRead)
import System.Random(RandomGen)
import qualified Data.Attoparsec.ByteString.Char8 as Parse
import qualified Data.Map as Map (Map, filterWithKey, null, lookup, toList)
import qualified Data.Text as Text (pack, unpack)
import qualified Data.Text.Encoding as Encoding (decodeUtf8With, encodeUtf8)
import qualified Data.Text.Encoding.Error as Encoding (lenientDecode)
import qualified Network.Socket as Socket(gracefulClose)
-- | Take an adapter that sends raw 'ByteString's, and transform it to an adapter that sends 'String's encoded in utf-8.
encodeUtf8 :: Adapter act ByteString -> IO (Adapter act String)
encodeUtf8 = mapTestChoices $ Encoding.encodeUtf8 . Text.pack
-- | Take an adapter that receives raw 'ByteString's, and transform it to an adapter that receives 'String's decoded from utf-8.
decodeUtf8 :: Adapter ByteString i -> IO (Adapter String i)
decodeUtf8 = mapActionsFromSut $ Text.unpack . Encoding.decodeUtf8With Encoding.lenientDecode
-- | Take an adapter that sends and receives raw 'ByteString's, and transform it to an adapter that sends and receives 'String's, encoded in utf-8 and decoded from utf-8.
endecodeUtf8 :: Adapter ByteString ByteString -> IO (Adapter String String)
endecodeUtf8 = Adap.map (Encoding.encodeUtf8 . Text.pack) (Text.unpack . Encoding.decodeUtf8With Encoding.lenientDecode)
encodeJSON :: (ToJSON i) => i -> ByteString
encodeJSON = toStrict . flip snoc (c2w8 '\n') . encode -- encode as strict ByteString and append a newline as separator
-- | Take an adapter that sends raw 'ByteString's, and transform it to an adapter that sends any type encoded in JSON.
encodeJSONTestChoices :: (ToJSON i) => Adapter act ByteString -> IO (Adapter act i)
encodeJSONTestChoices = mapTestChoices encodeJSON
-- | Take an adapter that receives raw 'ByteString's, and transform it to an adapter that receives any type decoded from JSON.
parseJSONActionsFromSut :: (FromJSON act) => Adapter ByteString i -> IO (Adapter act i)
parseJSONActionsFromSut adap = do
valueAdap <- parseActionsFromSut (Parse.skipSpace *> jsonNoDup) adap
mapActionsFromSut (fromResult . fromJSON) valueAdap
where
fromResult (Success s) = s
fromResult (Error e) = Debug.Trace.trace e $ undefined e -- TODO handle error case
-- transform an adapter to transfer input commands to observed actions
--
-- architecture:
-- input--combinedInputOS---fduplicate---adapInputOS---\
-- \ \
-- | \
-- loopbackInputOS |
-- v v
-- loopbackActionOS +------+
-- [loopbackBuffer] | adap |
-- loopbackActionIS +------+
-- | |
-- v |
-- <--output--combinedActionIS--fmerge <--adapActionIS---/
loopbackAdapter ::
Adapter o ic
-> (OutputStream i -> OutputStream ic -> IO (OutputStream ic))
-> (TInputStream (IOAct i o) -> TInputStream (IOAct i o) -> IO (TInputStream (IOAct i o)))
-> IO (Adapter (IOAct i o) ic)
loopbackAdapter adap fduplicate fmerge = do
loopbackBuffer <- newEmptyTMVarIO
loopbackActionOS <- makeOutputStream $ atomically . putTMVar loopbackBuffer
loopbackInputOS <- actionToInputOS loopbackActionOS -- map type i to type IOAct i o
let adapInputOS = inputCommandsToSut adap
combinedInputOS <- fduplicate loopbackInputOS adapInputOS
loopbackActionIS <- fromTMVar loopbackBuffer
let adapActionIS = outputToActionIS adap -- map type o to IOAct i o
-- loopback first: in case of a quick response to an input, the merging thread may see the input and output at the same time.
combinedActionIS <- fmerge loopbackActionIS adapActionIS -- the merge buffer lives in here implicitly
return $ Adapter {
inputCommandsToSut = combinedInputOS,
actionsFromSut = combinedActionIS,
close = close adap
}
outputToActionIS :: Adapter a i1 -> TInputStream (IOAct i2 a)
outputToActionIS adap = mapUnbuffered Out (error "acceptingInputs buffer from SUT does not support pushback") (actionsFromSut adap)
actionToInputOS :: TestChoice a1 a2 => OutputStream a2 -> IO (OutputStream a1)
actionToInputOS actionOS = streamSequence actionOS >>= contramap choiceToActs
-- direct pushbacks to the streams below is not needed, the merge buffer will handle pushbacks instead
streamSequence :: OutputStream a -> IO (OutputStream [a])
streamSequence s = makeOutputStream $ doMaybeSequenceCmd $ Streams.writeTo s
doMaybeSequenceCmd :: (Maybe a -> IO ()) -> Maybe [a] -> IO ()
doMaybeSequenceCmd writeMaybeAct Nothing = writeMaybeAct Nothing
doMaybeSequenceCmd writeMaybeAct (Just xs) = mapM_ (writeMaybeAct . Just) xs
{- |
Transform a given Adapter to accept all inputs. In other words, every selected input command directly becomes an observed input action.
Every observed output from the given Adapter becomes an output action.
-}
acceptingInputs :: Adapter o i -> IO (Adapter (IOAct i o) i)
acceptingInputs adap = loopbackAdapter adap duplicate (mergeBufferedWith mergeActions)
-- merging is done by reading an input, followed by all outputs that are then available. If no input is available, then read an output instead
mergeActions :: TInputStream (IOAct i o) -> TInputStream (IOAct i o) -> STM [Maybe (IOAct i o)]
mergeActions loopbackActionIS adapActionIS = do
hasLoopbackAction <- hasInput loopbackActionIS
if hasLoopbackAction
then do
inp <- Streams.read loopbackActionIS
outs <- readAll adapActionIS
return (inp:outs)
else do
hasAdapAction <- hasInput adapActionIS
if hasAdapAction
then singleton <$> Streams.read adapActionIS
else retry
{- |
Transform a given Adapter to accept all inputs, but if the given Adapter would call an undefined case in a partial function, then interpret this
as an observed input failure. Any input command that does not call undefined cases of partial functions directly become observed input action.
Every observed output from the given Adapter becomes an output action.
Input failures are useful for Adapters written in pure haskell. Instead of only writing complete functions to process input commands, where unexpected
input commands may e.g. lead to an error output, the adapter may also just process input commands with partial functions instead, avoiding some boilerplate code.
-}
acceptingInputsWithIncompletenessAsFailures :: Adapter o i -> IO (Adapter (IFAct i o) i)
acceptingInputsWithIncompletenessAsFailures adap = do
blockAdapActionsTVar <- newTVarIO False -- during duplication, block observation of actions from the adapter
loopbackAdapter adap (duplicateHandlingIncompleteness blockAdapActionsTVar) (mergeBufferedWith $ mergeActionsPaused blockAdapActionsTVar)
where
mergeActionsPaused :: TVar Bool -> TInputStream (IOAct i o) -> TInputStream (IOAct i o) -> STM [Maybe (IOAct i o)]
mergeActionsPaused blockAdapActionsTVar loopbackActionIS adapActionIS = do
blockAdapActions <- readTVar blockAdapActionsTVar
if blockAdapActions
then singleton <$> Streams.read loopbackActionIS -- adap actions are blocked, so observe just the loopback actions
else mergeActions loopbackActionIS adapActionIS -- adap actions are not blocked, merge actions as normal
duplicateHandlingIncompleteness :: TVar Bool -> OutputStream (InputAttempt i) -> OutputStream i -> IO (OutputStream i)
duplicateHandlingIncompleteness isAdapOutputBlocked loopbackInputOS adapInputOS = makeOutputStream $ \mi -> do
case mi of
Nothing -> Streams.write Nothing loopbackInputOS >> Streams.write Nothing adapInputOS
Just i -> do
atomically $ writeTVar isAdapOutputBlocked True
inputSucceeded <- attemptInputToAdap adapInputOS i
let mInputAction = Just $ InputAttempt (i,inputSucceeded)
Streams.write mInputAction loopbackInputOS
atomically $ writeTVar isAdapOutputBlocked False
attemptInputToAdap :: OutputStream i -> i -> IO Bool
attemptInputToAdap adapInputOS i = handle (patternMatchHandler i) (Streams.write (Just i) adapInputOS >> return True)
patternMatchHandler :: i -> PatternMatchFail -> IO Bool
patternMatchHandler _ _ = return False
-- | adapter which, after every input, directly sends the corresponding sequence of outputs
pureMealyAdapter :: (state -> i -> state) -> (state -> i -> [act]) -> state -> IO (Adapter act i)
-- FIXME this should also implicitly observe the accepted inputs, for which it probably needs to send either (IOAct i o) or (i/o) instead of an abstract act
pureMealyAdapter transitionFunction outputFunction initialState = do
stateVar <- newTVarIO initialState
outputQueue <- newTQueueIO
is <- fromBuffer outputQueue
os <- makeOutputStream $ \mi -> do
case mi of
Nothing -> do
atomically $ writeTQueue outputQueue Nothing
Just i -> do
os <- transduce stateVar i
atomically $ mapM_ (writeTQueue outputQueue . Just) os
return $ Adapter {
inputCommandsToSut = os,
actionsFromSut = is,
close = return ()
}
where
transduce stateVar i = atomically $ do
q <- readTVar stateVar
writeTVar stateVar (transitionFunction q i)
return $ outputFunction q i
{-|
Adapter which, after every action, has the given probability of producing non-deterministically one of the corresponding (non-timeout) output transitions if any is available.
After receiving a Nothing input, an output will be produced, which may be a timeout.
-}
pureAdapter :: (Ord i, Ord o, RandomGen g) => g -> Double -> (state -> Map.Map (IOAct i o) state) -> state -> IO (Adapter (SuspendedIF i o) (Maybe i))
pureAdapter g p transitionFunction initialState = do
let ((g',q), outs) = randomOutputTransitions transitionFunction g initialState False -- immediately take some outputs at the start
statefulAdapter <- statefulIO' (processInput transitionFunction) (g', q)
queue <- newTQueueIO
atomically $ mapM_ (writeTQueue queue . Just) outs
inputStream <- fromBuffer queue
outputStream <- makeOutputStream $ \mi -> do
case mi of
Nothing -> do
atomically $ writeTQueue queue Nothing
Just i -> do
os <- statefulAdapter i
atomically $ mapM_ (writeTQueue queue . Just) os
return $ Adapter {
inputCommandsToSut = outputStream,
actionsFromSut = inputStream,
close = return ()
}
where
--processInput :: (Ord i, Ord o, RandomGen g) => (q -> Map.Map (IOAct i o) q) -> (g, q) -> Maybe i -> ((g, q), [Suspended o])
processInput t (g', q) Nothing = randomOutputTransitions t g' q True
processInput t (g', q) (Just i) = case Map.lookup (In i) (t q) of
Just q' -> prependInput (InputAttempt (i, True)) $ randomOutputTransitions t g' q' False
Nothing -> ((g', q), [In $ InputAttempt (i, False)])
--randomOutputTransitions :: RandomGen g => (q -> Map.Map (IOAct i o) q) -> g -> q -> Bool -> ((g, q), [Suspended o])
randomOutputTransitions t g'' q isAfterNoInput = let (g', q', outs) = randomOutputTransitions' t g'' q [] isAfterNoInput in ((g', q'), reverse outs)
--randomOutputTransitions' :: RandomGen g => (q -> Map.Map (IOAct i o) q) -> g -> q -> [Suspended o] -> Bool -> (g, q, [Suspended o])
randomOutputTransitions' t g''' q outs isAfterNoInput =
let ts = Map.filterWithKey (\k _ -> isOutput k) (t q)
in if Map.null ts -- if no outputs are available at all,
then if isAfterNoInput then (g''', q, Out Quiescence : outs) else (g''', q, outs) -- then stop producing actions (meaning a timeout or just no more actions, depending on whether a "Nothing" input was previously received)
else let (produceOut, g') = if isAfterNoInput then (True, g''') else flipCoin g''' p -- else decide whether to produce more actions. This is mandatory if a "Nothing" input was previously received, otherwise flip a coin
in if not produceOut
then (g', q, outs)
else -- pick a random output, and continue randomly picking more outputs
let ((o, q'), g'') = takeRandom g' $ Map.toList ts
in randomOutputTransitions' t g'' q' (Out (OutSusp $ fromOutput o) : outs) False
prependInput i (q, acts) = (q, In i:acts)
-- | Transform the given Adapter by introducing quiescence (timeout observations). See 'withQuiescence', where the waiting time given in milliseconds.
withQuiescenceMillis :: Int -> Adapter (IOAct i o) i -> IO (Adapter (IOSuspAct i o) (Maybe i))
withQuiescenceMillis timeoutMillis = withQuiescence $ secondsToNominalDiffTime $ 0.001 * realToFrac timeoutMillis
{-|
Transform the given Adapter by introducing quiescence (timeout observations). Inputs can be sent as 'Just' inputs, and 'IOAct' observations can be
done as in the provided adapter, where the outputs in the 'IOAct' are wrapped in 'SuspAct' to denote that they are real outputs. Additionally:
* An artificial 'Quiescence' output observation is made after the provided timeout value. This timeout is measured since the last 'Just' input or
'SuspOut' output has been received.
* An artificial 'Nothing' input can be made, which indicates waiting for an output. Sending a 'Nothing' will block until any output is received,
which may be a real 'SuspAct' output or 'Quiescence'. This is guaranteed to happen within the given timeout value, give or take a few milliseconds
for processing.
-}
withQuiescence :: NominalDiffTime -> Adapter (IOAct i o) i -> IO (Adapter (IOSuspAct i o) (Maybe i))
withQuiescence timeoutDiff adap = do
lastObservationTime <- newEmptyTMVarIO -- time of the last observed action. Nothing if observing hasn't started yet.
isProcessingObservation <- newTVarIO False
observedQueue <- newTQueueIO
let ensureObservationTime = do -- start observing, if this hasn't already been done yet
isEmpty <- atomically $ isEmptyTMVar lastObservationTime
ifM_ isEmpty $ do
currentTime <- getCurrentTime
atomically $ writeTMVar lastObservationTime currentTime
updateObservationTime currentTime = do -- if the current time is past the stored observation time, then update that observation time to now
mLastTime <- tryReadTMVar lastObservationTime
case mLastTime of
Nothing -> writeTMVar lastObservationTime currentTime
Just lastTime -> ifM_ (lastTime < currentTime) $ writeTMVar lastObservationTime currentTime
getWaitTimeMicros currentTime = do -- the number of microseconds until the target timeout is reached
lastTime <- readTMVar lastObservationTime -- blocking, in case of Nothing
let targetTime = addUTCTime timeoutDiff lastTime
currentTime' = max currentTime lastTime -- lastTime > currentTime can occur if the caller waited too long after retrieving the
-- currentTime, in particular when blocking on reading lastObservationTime
return $ ceiling $ 1000000 * nominalDiffTimeToSeconds (diffUTCTime targetTime currentTime')
waitUntilQuiescence = do -- wait until the target timeout. Blocks if there is no target timeout yet.
currentTime <- getCurrentTime
waitTimeMicros <- atomically $ getWaitTimeMicros currentTime
delay waitTimeMicros
quiescenceMonitor = forever $ do -- background task that first wait until action monitoring starts, then continuously waits until a timeout
-- is reached and sets the quiescence state to true
waitUntilQuiescence
currentTime <- getCurrentTime
--quiIsSet <- atomically $ do
atomically $ do
additionalWaitTime <- getWaitTimeMicros currentTime
let quiescent = additionalWaitTime <= (0 :: Int)
ifM_ quiescent $ do
writeTQueue observedQueue (Just $ Out Quiescence)
updateObservationTime currentTime
actMonitor = do --background task that waits for outputs, updates the observation time and unsets the quiescence state
mAct <- atomically $ do
writeTVar isProcessingObservation True
Streams.read $ actionsFromSut adap
currentTime <- getCurrentTime
continue <- atomically $ do
-- observation has been made
updateObservationTime currentTime -- update the observation time
writeTVar isProcessingObservation False
case mAct of
Nothing -> do
writeTQueue observedQueue Nothing -- action adapter closed, pass on the Nothing once and then stop
return False
Just act -> do
writeTQueue observedQueue $ Just $ asSuspended act -- pass the action to the timeout adapter
return True
ifM_ continue actMonitor -- repeat until close
hasObservation = readTVar isProcessingObservation ||^ (not <$> isEmptyTQueue observedQueue) ||^ hasInput (actionsFromSut adap)
actionsFromSut' <- makeTInputStream (readTQueue observedQueue) hasObservation
inputCommandsToSut' <- makeOutputStream $ \mInCmd -> do
ensureObservationTime
case mInCmd of
Just (Just inCmd) -> Streams.write (Just inCmd) $ inputCommandsToSut adap
Just Nothing -> atomically $ waitUntil $ not <$> isEmptyTQueue observedQueue -- Just Nothing input means waiting, on an output or until a timeout
Nothing -> Streams.write Nothing $ inputCommandsToSut adap -- Nothing means closing the adapter, forward this to the underlying adapter
_ <- forkIO quiescenceMonitor
_ <- forkIO actMonitor
return $ Adapter {
inputCommandsToSut = inputCommandsToSut',
actionsFromSut = actionsFromSut',
close = ensureObservationTime >> close adap
}
-- | Transform the given Adapter by introducing a short delay before every provided input. See 'withInputDelay', with the delay time given in milliseconds.
withInputDelayMillis :: Int -> Adapter (IOAct i o) i' -> IO (Adapter (IOAct i o) i')
withInputDelayMillis timeDelayMillis = withInputDelay $ secondsToNominalDiffTime $ 0.001 * realToFrac timeDelayMillis
{-|
Transform the given Adapter by introducing a short delay before passing on any provided input. If, during that delay, an observation is made, then
that input is _not_ passed on.
This may be used to slow down a tester which performs inputs too fast for observation responses to occur, both after sending an input and after
making an observation.
-}
withInputDelay :: NominalDiffTime -> Adapter (IOAct i o) i' -> IO (Adapter (IOAct i o) i')
withInputDelay timeDelayDiff adap = do
lastObservationTime <- newEmptyTMVarIO
inputBlocked <- newTVarIO True
discardInput <- newTVarIO False
updateLastObservationTime <- newTVarIO False
let updateObservationTime currentTime = do -- if the current time is past the stored input time, then update that observation time to now
writeTVar inputBlocked True
mLastTime <- tryReadTMVar lastObservationTime
case mLastTime of
Nothing -> writeTMVar lastObservationTime currentTime
Just lastTime -> ifM_ (lastTime < currentTime) $ writeTMVar lastObservationTime currentTime
getWaitTimeMicros currentTime = do -- the number of microseconds until the target timeout is reached
lastTime <- readTMVar lastObservationTime -- should never be nothing, the unblocker ensures this
let targetTime = addUTCTime timeDelayDiff lastTime
currentTime' = max currentTime lastTime -- lastTime > currentTime can occur if the caller waited too long after retrieving the
-- currentTime, in particular when blocking on reading lastObservationTime
return $ ceiling $ 1000000 * nominalDiffTimeToSeconds (diffUTCTime targetTime currentTime')
waitUntilDelay = do
currentTime <- getCurrentTime
waitTimeMicros <- atomically $ getWaitTimeMicros currentTime
ifM_ (waitTimeMicros > 0) $ do
delay waitTimeMicros
waitUntilDelay
unblocker = do
-- this exists only because STM-monads cannot wait for a given delay, so this is a monitor that waits for the right TVar state to do so
atomically $ waitUntil $ readTVar inputBlocked &&^ (not <$> readTVar updateLastObservationTime)
waitUntilDelay
atomically $ writeTVar inputBlocked False
unblocker
observationTimeUpdater = do
-- this exists only because STM-monads cannot fetch the current time, so this is a monitor that waits for the right TVar state to do so
atomically $ waitUntil $ readTVar updateLastObservationTime
currentTime <- getCurrentTime
atomically $ do
updateObservationTime currentTime
writeTVar updateLastObservationTime False
observationTimeUpdater
waitUntilUnblocked = waitUntil $ not <$> readTVar inputBlocked
readFromSut = do
writeTVar updateLastObservationTime True -- signal the observationTimeUpdater to record the current time
writeTVar discardInput True -- an observation was made: discard any pending inputs
Streams.read $ actionsFromSut adap -- continue as usual
void $ forkIO unblocker
void $ forkIO observationTimeUpdater
inputCommandsToSut' <- makeOutputStream $ \mInCmd -> do
case mInCmd of
Just inCmd -> do
currentTime <- getCurrentTime
atomically $ do
-- sending an input command is strictly not an observation, but treat it as an observation as well: reset the observation time
updateObservationTime currentTime
writeTVar inputBlocked True
writeTVar discardInput False
discard <- atomically $ do
waitUntilUnblocked
readTVar discardInput
-- the previous atomic block set discardInput to False, send the input iff it wasn't set to True (due to an observation) in the meantime
ifM_ (not discard) $ Streams.write (Just inCmd) $ inputCommandsToSut adap
Nothing -> Streams.write Nothing $ inputCommandsToSut adap -- Nothing means closing the adapter, forward this to the underlying adapter
actionsFromSut' <- makeTInputStream readFromSut (hasInput $ actionsFromSut adap)
return $ Adapter {
inputCommandsToSut = inputCommandsToSut',
actionsFromSut = actionsFromSut',
close = close adap -- FIXME this should also close the forked threads
}
-- | Transform the given Adapter by introducing a short delay after every provided input. See 'withSuccessiveInputDelay', with the delay time given in milliseconds.
withSuccessiveInputDelayMillis :: Int -> Adapter (IOAct i o) i' -> IO (Adapter (IOAct i o) i')
withSuccessiveInputDelayMillis timeDelayMillis = withSuccessiveInputDelay $ secondsToNominalDiffTime $ 0.001 * realToFrac timeDelayMillis
{-|
Transform the given Adapter by introducing a short delay after every provided input. After an input is provided, then observing the adapter (or
even calling `hasObservation`) will block until the specified time duration has passed, or until an output is observed, or until the action stream
of the adapter closes, whichever comes first.
This may be used to slow down a tester which performs successive inputs too fast for observation responses to occur.
More precisely, this may be needed in case of a state with incoming transitions for inputs, and outgoing transitions for /both/ inputs and
outputs. Without input delay, the tester may perform one input to enter the state, immediately observe that input, and immediately perform the
second input from that state, whereas the wrapped adapter may receive the first input and respond with an output, followed by the second input.
The wrapped adapter and the tester then observe different traces of actions.
Note: this function does /not/ help to resolve states with an incoming transition for an /output/, and transitions for both inputs and output. To
resolve that situation, use `withInputDelay` instead.
-}
withSuccessiveInputDelay :: NominalDiffTime -> Adapter (IOAct i o) i' -> IO (Adapter (IOAct i o) i')
withSuccessiveInputDelay timeDelayDiff adap = do
-- FIXME loads of code duplication with withInputDelay, deduplicate this
lastInputTime <- newEmptyTMVarIO
observationBlocked <- newTVarIO False
let updateInputTime currentTime = do -- if the current time is past the stored input time, then update that observation time to now
mLastTime <- tryReadTMVar lastInputTime
case mLastTime of
Nothing -> writeTMVar lastInputTime currentTime
Just lastTime -> ifM_ (lastTime < currentTime) $ writeTMVar lastInputTime currentTime
getWaitTimeMicros currentTime = do -- the number of microseconds until the target timeout is reached
lastTime <- readTMVar lastInputTime -- should never be nothing, the unblocker ensures this
let targetTime = addUTCTime timeDelayDiff lastTime
currentTime' = max currentTime lastTime -- lastTime > currentTime can occur if the caller waited too long after retrieving the
-- currentTime, in particular when blocking on reading lastObservationTime
return $ ceiling $ 1000000 * nominalDiffTimeToSeconds (diffUTCTime targetTime currentTime')
waitUntilDelay = do
currentTime <- getCurrentTime
waitTimeMicros <- atomically $ getWaitTimeMicros currentTime
ifM_ (waitTimeMicros > 0) $ do
delay waitTimeMicros
waitUntilDelay
unblocker = do
-- this exists only because STM-monads cannot wait for a given delay, so this is a monitor that waits for the right TVar state to do so
atomically $ waitUntil $ readTVar observationBlocked
waitUntilDelay
atomically $ writeTVar observationBlocked False
unblocker
waitUntilUnblocked = waitUntil $ not <$> readTVar observationBlocked
waitUntilOutputOrClosed =
doAfter (hasInput $ actionsFromSut adap) $ do -- retry if there is no action present (and hence specifically no output)
mAct <- Streams.read $ actionsFromSut adap -- 'peek' an action (putting the action back later, because we need to peek more)
case mAct of
Nothing -> return () -- the stream closed, waiting has finished
Just act -> do
case act of
Out _ -> writeTVar observationBlocked False -- we've found an output, so waiting has finished
In _ -> waitUntilOutputOrClosed -- recursively peek more, which will either find an output or retry on an absent action
Streams.unRead act $ actionsFromSut adap -- now put the 'peeked' action back on the queue
readFromSut = do
waitUntilUnblocked `orElse` waitUntilOutputOrClosed
Streams.read $ actionsFromSut adap
hasObservation = readTVar observationBlocked ||^ hasInput (actionsFromSut adap)
void $ forkIO unblocker
inputCommandsToSut' <- makeOutputStream $ \mInCmd -> do
case mInCmd of
Just inCmd -> do
-- waiting is only necessary if the user of the adapter sends a second input without observing an action after the first input
atomically $ waitUntil $ not <$> readTVar observationBlocked
currentTime <- getCurrentTime
atomically $ do
updateInputTime currentTime
writeTVar observationBlocked True
Streams.write (Just inCmd) $ inputCommandsToSut adap
Nothing -> Streams.write Nothing $ inputCommandsToSut adap -- Nothing means closing the adapter, forward this to the underlying adapter
actionsFromSut' <- makeTInputStream readFromSut hasObservation
return $ Adapter {
inputCommandsToSut = inputCommandsToSut',
actionsFromSut = actionsFromSut',
close = close adap -- FIXME this should also close the forked threads
}
--------------------
-- socket adapter --
--------------------
-- | Settings for a socket Adapter.
data SocketSettings act i = SocketSettings {
hostName :: HostName,
portNumber :: PortNumber
}
-- | Default settings for a socket Adapter.
baseSocketSettings :: SocketSettings act i
baseSocketSettings = SocketSettings {
hostName = "127.0.0.1",
portNumber = 2929
}
-- | Create an adapter by connecting to a server socket, with the default settings.
connectSocketAdapter :: IO (Adapter ByteString ByteString)
connectSocketAdapter = connectSocketAdapterWith baseSocketSettings
-- | Create an adapter by connecting to a server socket, with the given settings.
connectSocketAdapterWith :: SocketSettings act i -> IO (Adapter ByteString ByteString)
connectSocketAdapterWith settings = niceSocketsDo $ do
socket <- connectTCP (hostName settings) (portNumber settings)
(actionBytes, inputCommandBytes) <- socketToStreams socket
(threadid, forkedActionBytes) <- fromInputStreamBuffered actionBytes
return $ Adapter {
inputCommandsToSut = inputCommandBytes,
actionsFromSut = forkedActionBytes,
-- when the adapter is closed, we first kill the thread, to prevent it from throwing errors when we kill its socket
close = killThread threadid >> Socket.gracefulClose socket 1000
}
-- | Create an adapter by connecting to a server socket, with the default settings, and sending inputs and reading outputs in JSON format.
connectJSONSocketAdapter :: (ToJSON i, FromJSON o) => IO (Adapter o i)
connectJSONSocketAdapter = connectJSONSocketAdapterWith baseSocketSettings
-- | Create an adapter by connecting to a server socket, with the given settings, and sending inputs and reading outputs in JSON format.
connectJSONSocketAdapterWith :: (ToJSON i, FromJSON o) => SocketSettings act i -> IO (Adapter o i)
connectJSONSocketAdapterWith settings = do
rawAdap <- connectSocketAdapterWith settings
parsingAdap <- parseJSONActionsFromSut rawAdap
encodeJSONTestChoices parsingAdap
-- | Create an adapter by connecting to a server socket, with the default settings, and sending inputs and reading outputs in JSON format, observing any input as accepted.
connectJSONSocketAdapterAcceptingInputs :: (ToJSON i, FromJSON o) => IO (Adapter (IOAct i o) i)
connectJSONSocketAdapterAcceptingInputs = connectJSONSocketAdapter >>= acceptingInputs
-- | Create an adapter by connecting to a server socket, with the given settings, and sending inputs and reading outputs in JSON format, observing any input as accepted.
connectJSONSocketAdapterAcceptingInputsWith :: (ToJSON i, FromJSON o) => SocketSettings act i -> IO (Adapter (IOAct i o) i)
connectJSONSocketAdapterAcceptingInputsWith settings = connectJSONSocketAdapterWith settings >>= acceptingInputs
-- | Transform the given I/O-Adapter (for action `IOAct` by interpreting the input and output actions as gate values with data parameters.
asSymbolicSuspAdapter :: Adapter (IOSuspAct (GateValue i) (GateValue o)) (Maybe (GateValue i)) -> IO (Adapter (IOSuspGateValue i o) (Maybe (GateValue i)))
asSymbolicSuspAdapter = mapActionsFromSut ioSuspActGateToSuspGateValue
where
ioSuspActGateToSuspGateValue (In (GateValue i cs)) = GateValue (In i) cs
ioSuspActGateToSuspGateValue (Out (OutSusp (GateValue o cs))) = GateValue (Out (OutSusp o)) cs
ioSuspActGateToSuspGateValue (Out Quiescence) = GateValue (Out Quiescence) []
-- | Create an adapter by connecting to a server socket, with the default settings, and sending inputs and reading outputs with data, in JSON format, observing any input as accepted.
connectJSONSocketAdapterSTSwithQuiescence :: (ToJSON i, FromJSON o) => Int -> IO (Adapter (IOSuspGateValue i o) (Maybe (GateValue i)))
connectJSONSocketAdapterSTSwithQuiescence millis = connectJSONSocketAdapterAcceptingInputs >>= withQuiescenceMillis millis >>= asSymbolicSuspAdapter