lattest-lib 0.1.0.0 → 0.1.1

raw patch · 7 files changed

+121/−21 lines, 7 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• CHANGELOG.md +3/−0
• lattest-lib.cabal +2/−1
• src/Lattest/Adapter/StandardAdapters.hs +4/−3
• src/Lattest/Exec/ADG/SplitGraph.hs too large to diff
• src/Lattest/Model/BoundedMonad.hs +4/−0
• src/Lattest/Streams/Synchronized.hs +6/−17
• src/System/IO/Streams/Synchronized/Attoparsec.hs +102/−0

@@ -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 

@@ -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

@@ -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.

file too large to diff

@@ -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] 

@@ -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

@@ -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+++