diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,9 @@ ## Unreleased +## 0.1.1 - 2026-07-15 +Bugfix patch: adds instance OrdTraversable FreeLattice, and properly kills the reading thread before its socket is closed. + ## 0.1.0.0 - 2026-07-14 First release diff --git a/lattest-lib.cabal b/lattest-lib.cabal --- a/lattest-lib.cabal +++ b/lattest-lib.cabal @@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack name: lattest-lib -version: 0.1.0.0 +version: 0.1.1 synopsis: Model-based testing framework description: Welcome to Lattest, the latest attested lattice testing tech! . @@ -50,6 +50,7 @@ Lattest.Util.ReportUtils Lattest.Util.STSJSONParser Lattest.Util.Utils + System.IO.Streams.Synchronized.Attoparsec other-modules: Lattest.Model.Internal.NonDeterministic Lattest.Model.Symbolic.Internal.Boute diff --git a/src/Lattest/Adapter/StandardAdapters.hs b/src/Lattest/Adapter/StandardAdapters.hs --- a/src/Lattest/Adapter/StandardAdapters.hs +++ b/src/Lattest/Adapter/StandardAdapters.hs @@ -74,7 +74,7 @@ import Debug.Trace(trace) -- FIXME find a better alternative -import GHC.Conc(forkIO, newTVarIO, TVar, retry, atomically, writeTVar, readTVar, STM, orElse) +import GHC.Conc(forkIO, newTVarIO, TVar, retry, atomically, writeTVar, readTVar, STM, orElse, killThread) import Network.Socket(HostName, PortNumber) import Network.Utils (niceSocketsDo, connectTCP) @@ -566,11 +566,12 @@ connectSocketAdapterWith settings = niceSocketsDo $ do socket <- connectTCP (hostName settings) (portNumber settings) (actionBytes, inputCommandBytes) <- socketToStreams socket - forkedActionBytes <- fromInputStreamBuffered actionBytes + (threadid, forkedActionBytes) <- fromInputStreamBuffered actionBytes return $ Adapter { inputCommandsToSut = inputCommandBytes, actionsFromSut = forkedActionBytes, - close = Socket.gracefulClose socket 1000 + -- 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. diff --git a/src/Lattest/Exec/ADG/SplitGraph.hs b/src/Lattest/Exec/ADG/SplitGraph.hs # file too large to diff: src/Lattest/Exec/ADG/SplitGraph.hs diff --git a/src/Lattest/Model/BoundedMonad.hs b/src/Lattest/Model/BoundedMonad.hs --- a/src/Lattest/Model/BoundedMonad.hs +++ b/src/Lattest/Model/BoundedMonad.hs @@ -190,6 +190,10 @@ instance OM.OrdFunctor FreeLattice where ordMap f (FreeLattice x) = FreeLattice $ Set.map (Set.map f) x +instance OM.OrdTraversable FreeLattice where + ordTraverse f (FreeLattice x) = FreeLattice <$> ordTraverse (ordTraverse f) x + ordSequenceA (FreeLattice x) = FreeLattice <$> ordTraverse ordSequenceA x + instance Ord a => JoinSemiLattice (FreeLattice a) where (FreeLattice x) \/ (FreeLattice y) = FreeLattice $ Set.map Set.unions $ nAryCartesianProduct $ Set.fromList [x,y] diff --git a/src/Lattest/Streams/Synchronized.hs b/src/Lattest/Streams/Synchronized.hs --- a/src/Lattest/Streams/Synchronized.hs +++ b/src/Lattest/Streams/Synchronized.hs @@ -1,3 +1,4 @@ +{-# LANGUAGE TupleSections #-} module Lattest.Streams.Synchronized ( TInputStream, read, @@ -26,16 +27,14 @@ import Control.Concurrent.STM.TQueue(TQueue, newTQueueIO, writeTQueue, readTQueue, isEmptyTQueue, unGetTQueue) import Control.Concurrent.STM.TMVar(TMVar, takeTMVar, isEmptyTMVar) import Control.Concurrent.STM.TVar(newTVarIO, readTVar, writeTVar) -import GHC.IO.Exception (ioe_location, ioe_errno) -import Control.Exception(throwIO, Exception, catchJust) +import Control.Exception(throwIO, Exception) import Control.Monad (void, when) import Data.List(singleton) -import GHC.Conc (atomically, forkIO) +import GHC.Conc (atomically, forkIO, ThreadId) import System.IO.Streams (InputStream, OutputStream, makeOutputStream, connect) import qualified System.IO.Streams as Streams (write) import Control.Monad.Extra((||^)) -import Foreign.C.Error (eBADF, Errno (Errno)) data TInputStream a = TInputStream { tRead :: STM (Maybe a), @@ -141,22 +140,12 @@ -- a synchronized input stream that reads from the given input stream. A monitoring thread reads the given input stream and buffers the inputs on the background -- TODO support a bound for the buffer -fromInputStreamBuffered :: InputStream a -> IO (TInputStream a) +fromInputStreamBuffered :: InputStream a -> IO (ThreadId, TInputStream a) fromInputStreamBuffered is = do buffer <- newTQueueIO bufferOS <- makeOutputStream $ atomically . writeTQueue buffer -- an intermediate output stream to write to the queue - void $ forkIO $ catchSocketClosed $ connect is bufferOS -- move items from the original adapter into the buffer in a separate thread - fromBuffer buffer - where - -- catches "threadWait: invalid argument (Bad file descriptor)" - -- which is thrown when the socket is closed while this thread is running, e.g. on Adaptor.close() - -- The alternative solution is to keep track of the threadID given by forkIO, - -- and explicitly stop the thread before closing the socket. - catchSocketClosed forkedThread = catchJust threadwaitinvalid forkedThread pure - threadwaitinvalid e - | ioe_location e == "threadWait" - , fmap Errno (ioe_errno e) == Just eBADF = Just () - | otherwise = Nothing + threadid <- forkIO $ connect is bufferOS -- move items from the original adapter into the buffer in a separate thread + (threadid,) <$> fromBuffer buffer mapUnbuffered :: (a -> b) -> (b -> a) -> TInputStream a -> TInputStream b -- we need an inverse to push a's back to the original TInputStream of b's mapUnbuffered f f' tis = TInputStream { -- FIXME either remove or wrap in withClosedState diff --git a/src/System/IO/Streams/Synchronized/Attoparsec.hs b/src/System/IO/Streams/Synchronized/Attoparsec.hs new file mode 100644 --- /dev/null +++ b/src/System/IO/Streams/Synchronized/Attoparsec.hs @@ -0,0 +1,102 @@ +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE OverloadedStrings #-} + +module System.IO.Streams.Synchronized.Attoparsec ( +parseFromStream, +parserToInputStream +) +where + +import Control.Exception(Exception) +import Control.Monad(unless) +import Data.String(IsString) +import Control.Monad.Extra((||^), (&&^)) +import qualified Data.Attoparsec.ByteString as C8(parse,feed, Parser) +import Data.Attoparsec.Types(Parser,IResult(..)) +import Data.ByteString(ByteString) +import qualified Data.ByteString as C8(null) +import Data.List(intercalate) + +import Lattest.Streams.Synchronized (TInputStream,makeTInputStream,hasInput) +import qualified Lattest.Streams.Synchronized as Streams (read, unRead) +import Control.Concurrent.STM.TVar(TVar, newTVarIO, writeTVar, readTVar) +import Control.Concurrent.STM(STM, throwSTM) + +parseFromStream :: TVar (Bool, IResult ByteString (Maybe r)) -> Parser ByteString (Maybe r) -> TInputStream ByteString -> STM (Maybe r) +parseFromStream stateVar parser is = do + parseFromStream' stateVar True is + (closed, state) <- readTVar stateVar + writeTVar stateVar (closed, C8.parse parser "") + if closed + then return Nothing + else case state of + (Done "" r) -> return r + other -> + let (residual,ctx,message) = errorContext other + in throwSTM $ ParseException (toMsg ctx ++ message ++ ", residual [" ++ show residual ++ "]") + where + toMsg [] = "" + toMsg xs = "[parsing " ++ intercalate "/" xs ++ "] " + +parseFromStream' :: TVar (Bool, IResult ByteString (Maybe r)) -> Bool -> TInputStream ByteString -> STM () +parseFromStream' stateVar blockUntilFinished is = do + (closed, r) <- readTVar stateVar + if closed + then return () + else doParse r >>= writeTVar stateVar + where + doParse :: IResult ByteString (Maybe r) -> STM (Bool, IResult ByteString (Maybe r)) + doParse r = do + stop <- return (isFinished r) ||^ (return (not blockUntilFinished) &&^ (not <$> hasInput is)) + if stop + then do + r' <- unread' r + return (False, r') + else do + m <- Streams.read is + case m of + Nothing -> do + r' <- unread' r + return (True, r') + Just s -> if C8.null s + then doParse r + else doParse $! C8.feed r s + unread rest = unless (C8.null rest) $ Streams.unRead rest is + + unread' (Fail rest ctx e) = unread rest >> return (Fail "" ctx e) + unread' (Done rest result) = unread rest >> return (Done "" result) + unread' partial = return partial + +errorContext :: IsString a => IResult a r -> (a, [String], String) +errorContext (Fail residual ctx msg) = (residual, ctx, msg) +errorContext (Partial _) = ("", [], "") +errorContext (Done residual _) = (residual, [],"") + +isFinished :: IResult a b -> Bool +isFinished (Partial _) = False +isFinished _ = True + +newtype ParseException = ParseException String + +instance Exception ParseException +instance Show ParseException where + show (ParseException message) = "Parse error: " ++ message + +-- Convert a stream of bytes to a stream of actions parsed from those bytes, using the provided parser. The stream ends if either the parser yields +-- a Nothing or if the underlying stream yields a Nothing +-- If the parser yields a Nothing, then all bytes up to that Nothing are consumed from the underlying stream, but if the underyling stream yields a +-- Nothing, then all bytes since the last Just result are unread. +parserToInputStream :: C8.Parser (Maybe r) + -> TInputStream ByteString + -> IO (TInputStream r) +parserToInputStream parser is = do + stateVar <- newTVarIO (False, C8.parse parser "") + makeTInputStream (parseFromStream stateVar parser is) (parserHasInput stateVar) + where + parserHasInput stateVar = do + parseFromStream' stateVar False is + (closed, state) <- readTVar stateVar + return $ closed || isFinished state + + +