packages feed

lossy-chan (empty) → 0.1.0.0

raw patch · 10 files changed

+1436/−0 lines, 10 filesdep +asyncdep +basedep +criterion

Dependencies added: async, base, criterion, lossy-chan, stm, tasty, tasty-hunit, tasty-quickcheck, unagi-chan, weigh

Files

+ CHANGELOG.md view
@@ -0,0 +1,5 @@+# Revision history for lossy-chan++## 0.1.0.0 -- YYYY-mm-dd++* First version. Released on an unsuspecting world.
+ LICENSE view
@@ -0,0 +1,29 @@+Copyright (c) 2026, Well-Typed LLP+++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++    * Redistributions of source code must retain the above copyright+      notice, this list of conditions and the following disclaimer.++    * Redistributions in binary form must reproduce the above+      copyright notice, this list of conditions and the following+      disclaimer in the documentation and/or other materials provided+      with the distribution.++    * Neither the name of the copyright holder nor the names of its+      contributors may be used to endorse or promote products derived+      from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ README.md view
@@ -0,0 +1,130 @@+# `lossy-chan`++This package provides fast, bounded, and _lossy_ broadcast channels backed by a ring buffer.++As long as receivers keep up with senders, a `lossy-chan` acts as a regular channel. For example:++```haskell+(i, o) <- LC.new 64       -- buffer size+for_ [0..1000] $ \m -> do+    LC.send i m+    m' <- LC.receive o+    assertM (m' == m)+```++If a receiver lags behind, it loses access to the _oldest_ messages first. For example:++```haskell+(i, o) <- LC.new 64+for_ [0..64] (LC.send i)  -- sends 65 messages+m' <- LC.receive o+assertM (m' == 1)         -- 0 was dropped+```++A `lossy-chan` channel is split into an input and an output channel, both of which are thread-safe.+If you use a single output channel in two threads, the messages are split between the two.+If you want two output channls that both receive every message, you can _duplicate_ the output channel.+The new output channel receives messages starting where the duplicated channel left off. For example:++```haskell+(i, o1) <- LC.new 64+for_ [1,2] (LC.send i)++m1 <- LC.receive o1+assertM (m1 == 1)         -- receive the 1 on o1++o2 <- LC.dup o1           -- duplicate o1+m2 <- LC.receive o2       -- receive the 2 on o2+assertM (m2 == 2)++m2' <- LC.receive o1      -- receive the 2 on o1+assertM (m2' == 2)+```++The `lossy-chan` buffer uses access counts to ensure that messages can be garbage collected as soon as they've been received by every output channel. This relies on finalizers and is not guaranteed to be prompt. For example:++```haskell+(i, o) <- LC.new 64+LC.send i bigThing+void (LC.receive o)       -- bigThing can now be garbage collected+```++If the input channel is garbage collected, output channels can continue to receive values until they reach the end of the buffer, at which point `receive` will throw a `NoInChanError`. Alternatively, the GHC RTS may also throw a `BlockedIndefinitelyOnSTM` exception. For example:++```haskell+(i, o) <- LC.new 64+LC.send i 1++m1 <- LC.receive o+assertM (m1 == 1)         -- receive the 1++m2 <- LC.receive o        -- throws NoInChanError or BlockedIndefinitelyOnSTM+```++## Benchmarks++This section shows the results of time and space benchmarks that compare `lossy-chan` with various other channels implementations:++- The `base` benchmark uses `Control.Concurrent.Chan`.+- The `stm` benchmark uses `Control.Concurrent.STM.TChan`.+- The `unagi-chan` benchmark uses `Control.Concurrent.Chan.Unagi`.+- The `lossy-chan-N` benchmarks use `Control.Concurrent.LossyChan` with buffer size `N`.++The benchmark itself was adapted from the `unagi-chan`'s benchmark, and consists of some number of sender threads, each of which sends 10,000 messages, and some number of receiver threads, each of which tries to receive 10,000 messages.+Since `lossy-chan` channels may lose some messages, receiver threads terminate when they catch a `NoInChanError` or `BlockedIndefinitelyOnSTM` exception.++_This isn't a fair comparison_, since none of the other channel implementations is lossy. However, it should give you some idea of the performance and memory usage characteristics of `lossy-chan`.++### Time++The time benchmark, created using `criterion`, shows that a `lossy-chan` channel is roughly as fast as an `unagi-chan` channel, and both are significantly faster than the `base` and `stm` channels. Moreover, the buffer size doesn't appear to significanly affect the performance of a `lossy-chan` channel.++![Benchmark that compares lossy-chan with unagi-chan.](https://github.com/well-typed/lossy-chan/blob/ed2abb0ee3b55a549fce7f466f6a872c4bc6ec25/assets/benchmark.png?raw=true)++### Space++The space benchmark, created using `weigh`, shows that a `lossy-chan` channel has much lower maximum residency when compared to the other channel implementations, and that its maximum residency scales with the buffer size. This is expected from the use of a ring buffer.+++| Benchmark        | Case                   |  Allocated | Max Residency |+| :--------------- | :--------------------- | ---------: | ------------: |+| `base`           | 1 senders, 1 receivers | 25,604,448 |     8,348,840 |+|                  | 2 senders, 2 receivers | 51,208,176 |    12,275,048 |+|                  | 3 senders, 3 receivers | 76,811,928 |    16,699,824 |+|                  | 4 senders, 4 receivers | 96,154,064 |    24,617,800 |+| `stm`            | 1 senders, 1 receivers | 16,805,400 |     8,830,408 |+|                  | 2 senders, 2 receivers | 33,609,720 |     9,845,000 |+|                  | 3 senders, 3 receivers | 50,414,848 |    14,717,096 |+|                  | 4 senders, 4 receivers | 55,414,888 |    17,660,312 |+| `unagi-chan`     | 1 senders, 1 receivers | 18,439,984 |     4,540,792 |+|                  | 2 senders, 2 receivers | 36,862,432 |     4,541,464 |+|                  | 3 senders, 3 receivers | 55,277,096 |    10,573,080 |+|                  | 4 senders, 4 receivers | 73,698,768 |    10,573,984 |+| `lossy-chan-2`   | 1 senders, 1 receivers | 12,008,624 |         3,088 |+|                  | 2 senders, 2 receivers | 24,013,384 |         4,960 |+|                  | 3 senders, 3 receivers | 36,018,184 |         6,816 |+|                  | 4 senders, 4 receivers | 48,021,576 |         8,672 |+| `lossy-chan-4`   | 1 senders, 1 receivers | 12,008,928 |         3,168 |+|                  | 2 senders, 2 receivers | 24,014,392 |         5,040 |+|                  | 3 senders, 3 receivers | 36,018,456 |         6,896 |+|                  | 4 senders, 4 receivers | 48,021,664 |         8,752 |+| `lossy-chan-8`   | 1 senders, 1 receivers | 12,009,536 |         3,328 |+|                  | 2 senders, 2 receivers | 24,015,448 |         5,200 |+|                  | 3 senders, 3 receivers | 36,019,056 |         7,056 |+|                  | 4 senders, 4 receivers | 48,024,952 |         8,912 |+| `lossy-chan-16`  | 1 senders, 1 receivers | 12,010,472 |         3,648 |+|                  | 2 senders, 2 receivers | 24,017,080 |         5,520 |+|                  | 3 senders, 3 receivers | 36,019,840 |         7,376 |+|                  | 4 senders, 4 receivers | 48,028,224 |         9,232 |+| `lossy-chan-32`  | 1 senders, 1 receivers | 12,011,728 |         4,288 |+|                  | 2 senders, 2 receivers | 24,017,720 |         6,160 |+|                  | 3 senders, 3 receivers | 36,028,936 |         8,016 |+|                  | 4 senders, 4 receivers | 48,037,104 |         9,872 |+| `lossy-chan-64`  | 1 senders, 1 receivers | 12,014,464 |         5,568 |+|                  | 2 senders, 2 receivers | 24,030,232 |         7,440 |+|                  | 3 senders, 3 receivers | 36,041,456 |         9,296 |+|                  | 4 senders, 4 receivers | 48,067,576 |        11,152 |+| `lossy-chan-128` | 1 senders, 1 receivers | 12,021,448 |         8,128 |+|                  | 2 senders, 2 receivers | 24,047,128 |        10,000 |+|                  | 3 senders, 3 receivers | 36,065,520 |        11,856 |+|                  | 4 senders, 4 receivers | 48,083,512 |        13,712 |
+ benchmark/Benchmark/Chan.hs view
@@ -0,0 +1,97 @@+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Benchmark.Chan (+  Chan (..),+  lossyChan,+  baseChan,+  stmTChan,+  unagiChan,+) where++import Control.Concurrent.Chan qualified as Base+import Control.Concurrent.Chan.Unagi qualified as UC+import Control.Concurrent.LossyChan qualified as LC+import Control.Concurrent.STM.TChan qualified as STM+import Control.Monad.STM (atomically)+import Data.Kind (Type)+import Data.Word (Word32)++{- |+An abstraction over channel libraries.+-}+data Chan (i :: Type -> Type) (o :: Type -> Type)+  = MkChan+  { new :: forall a. IO (i a, o a)+  , dup :: forall a. (i a, o a) -> IO (o a)+  , send :: forall a. i a -> a -> IO ()+  , recv :: forall a. o a -> IO a+  }++{- |+The `Chan` instance for @lossy-chan@.+-}+lossyChan :: Word32 -> Chan LC.InChan LC.OutChan+lossyChan len =+  MkChan+    { new = LC.new len+    , dup = LC.dup . snd+    , send = LC.send+    , recv = LC.receive+    }+{-# INLINE lossyChan #-}++{- |+The `Chan` instance for "Control.Concurrent.Chan" from @base@.+-}+baseChan :: Chan Base.Chan Base.Chan+baseChan =+  MkChan+    { new = do i <- Base.newChan; pure (i, i)+    , dup = Base.dupChan . fst+    , send = Base.writeChan+    , recv = Base.readChan+    }+{-# INLINE baseChan #-}++{- |+The `Chan` instance for "Control.Concurrent.STM.TChan" from @stm@.+-}+stmTChan :: Chan STM.TChan STM.TChan+stmTChan =+  MkChan+    { new =+        atomically $ do+          i <- STM.newBroadcastTChan+          o <- STM.dupTChan i+          pure (i, o)+    , dup =+        \(i, _) ->+          atomically $+            STM.dupTChan i+    , send =+        \i x ->+          atomically $+            STM.writeTChan i x+    , recv =+        \o ->+          atomically $+            STM.readTChan o+    }+{-# INLINE stmTChan #-}++{- |+The `Chan` instance for "Control.Concurrent.Chan.Unagi" from @unagi-chan@.+-}+unagiChan :: Chan UC.InChan UC.OutChan+unagiChan =+  MkChan+    { new = UC.newChan+    , dup = UC.dupChan . fst+    , send = UC.writeChan+    , recv = UC.readChan+    }+{-# INLINE unagiChan #-}
+ benchmark/Benchmark/Simple.hs view
@@ -0,0 +1,61 @@+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TupleSections #-}++module Benchmark.Simple (+  SimpleBenchOpts (..),+  runSimpleBench,+) where++import Benchmark.Chan+import Control.Concurrent.Async (async, wait)+import Control.Concurrent.LossyChan (NoInChanError)+import Control.Exception (BlockedIndefinitelyOnSTM, Handler (..), catches)+import Control.Monad (void)+import Data.Foldable (for_)+import Data.Traversable (for)++{- |+Options for a simple benchmark.+-}+data SimpleBenchOpts+  = MkSimpleBenchOpts+  { messages :: !Int+  , senders :: !Int+  , receivers :: !Int+  }++{- |+Run a simple benchmark.+-}+runSimpleBench :: Chan i o -> SimpleBenchOpts -> IO ()+runSimpleBench MkChan{..} =+  \MkSimpleBenchOpts{..} -> do+    (i, o) <- new+    receiverAsyncs <-+      for [1 .. receivers] $ \r -> do+        o' <- if r == receivers then pure o else dup (i, o)+        async $+          handles $+            for_ [1 .. messages] $ \_m -> do+              void (recv o')+    _senderAsync <-+      for [1 .. senders] $ \s -> do+        async $+          for_ [1 .. messages] $ \m -> do+            void (send i (s, m))+    for_ receiverAsyncs wait+{-# INLINE runSimpleBench #-}+++handles :: IO () -> IO ()+handles = (`catches` handlers)+ where+  handlers :: [Handler ()]+  handlers =+    [ Handler (\(_e :: BlockedIndefinitelyOnSTM) -> pure ())+    , Handler (\(_e :: NoInChanError) -> pure ())+    ]+{-# INLINE handles #-}
+ benchmark/space.hs view
@@ -0,0 +1,38 @@+{-# LANGUAGE FunctionalDependencies #-}+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Main (main) where++import Benchmark.Chan+import Benchmark.Simple+import Data.Foldable (for_)+import Weigh (Column (..), Format (..), Weigh, io, mainWith, setColumns, setFormat, wgroup)++main :: IO ()+main = do+  mainWith $ do+    setColumns [Case, Allocated, Max]+    setFormat Markdown+    wgroup "base" $+      for_ [1 .. 4] $ \s ->+        weighChanWith baseChan s s+    wgroup "stm" $+      for_ [1 .. 4] $ \s ->+        weighChanWith stmTChan s s+    wgroup "unagi-chan" $+      for_ [1 .. 4] $ \s ->+        weighChanWith unagiChan s s+    for_ [2, 4, 8, 16, 32, 64, 128] $ \n ->+      wgroup ("lossy-chan-" <> show n) $+        for_ [1 .. 4] $ \s ->+          weighChanWith (lossyChan n) s s++weighChanWith :: Chan i o -> Int -> Int -> Weigh ()+weighChanWith chan senders receivers = do+  let label = show senders <> " senders, " <> show receivers <> " receivers"+  let action = runSimpleBench chan+  let opts = MkSimpleBenchOpts{messages = 100000, ..}+  io label action opts
+ benchmark/time.hs view
@@ -0,0 +1,48 @@+{-# LANGUAGE FunctionalDependencies #-}+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Main (main) where++import Benchmark.Chan+import Benchmark.Simple+import Criterion.Main (Benchmark, bench, bgroup, defaultMain, whnfIO)++main :: IO ()+main = do+  defaultMain . concat $+    [ [ bgroup+          "base"+          [ benchChanWith baseChan s r+          | s <- [1 .. 4]+          , let r = s+          ]+      , bgroup+          "stm"+          [ benchChanWith stmTChan s r+          | s <- [1 .. 4]+          , let r = s+          ]+      , bgroup+          "unagi-chan"+          [ benchChanWith unagiChan s r+          | s <- [1 .. 4]+          , let r = s+          ]+      ]+        <> [ bgroup ("lossy-chan-" <> show n) $+               [ benchChanWith (lossyChan n) s r+               | s <- [1 .. 4]+               , let r = s+               ]+           | n <- [2, 8, 32, 128]+           ]+    ]++benchChanWith :: Chan i o -> Int -> Int -> Benchmark+benchChanWith chan senders receivers =+  bench (show senders <> " senders, " <> show receivers <> " receivers") $+    whnfIO . runSimpleBench chan $+      MkSimpleBenchOpts{messages = 10000, ..}
+ lossy-chan.cabal view
@@ -0,0 +1,104 @@+cabal-version: 3.0+name: lossy-chan+version: 0.1.0.0+synopsis: Fast, bounded, and lossy broadcast channels backed by a ring buffer+description: Fast, bounded, and /lossy/ broadcast channels backed by a ring buffer.+license: BSD-3-Clause+license-file: LICENSE+author: Wen Kokke+maintainer: wen@well-typed.com+copyright: (c) 2026 Well-Typed+category: Concurrency+build-type: Simple+extra-doc-files:+  CHANGELOG.md+  README.md++tested-with:+  ghc ==9.4.8 || ==9.6.7 || ==9.8.4 || ==9.10.3 || ==9.12.4 || ==9.14.1++source-repository head+  type: git+  location: https://github.com/well-typed/lossy-chan.git++common language+  ghc-options:+    -Wall+    -Wcompat+    -Widentities+    -Wprepositive-qualified-module+    -Wredundant-constraints+    -Wunticked-promoted-constructors+    -Wunused-packages++  default-extensions:+    ImportQualifiedPost+    RecordWildCards++  default-language: Haskell2010++flag small-loop-counter+  description:+    Use Word8 instead of Word32 for the loop counter,+    for the purposes of testing loop counter overflows.++  default: False+  manual: True++library+  import: language+  hs-source-dirs: src+  exposed-modules: Control.Concurrent.LossyChan+  build-depends: base >=4.16 && <5++  if flag(small-loop-counter)+    cpp-options: -DLOSSY_CHAN_SMALL_LOOP_COUNTER++test-suite lossy-chan-test+  import: language+  type: exitcode-stdio-1.0+  hs-source-dirs: test+  main-is: Main.hs+  build-depends:+    base >=4.16 && <5,+    lossy-chan,+    tasty,+    tasty-hunit,+    tasty-quickcheck,++  if flag(small-loop-counter)+    cpp-options: -DLOSSY_CHAN_SMALL_LOOP_COUNTER++benchmark lossy-chan-benchmark-time+  import: language+  type: exitcode-stdio-1.0+  hs-source-dirs: benchmark+  main-is: time.hs+  other-modules:+    Benchmark.Chan+    Benchmark.Simple++  build-depends:+    async,+    base >=4.16 && <5,+    criterion,+    lossy-chan,+    stm,+    unagi-chan,++benchmark lossy-chan-benchmark-space+  import: language+  type: exitcode-stdio-1.0+  hs-source-dirs: benchmark+  main-is: space.hs+  other-modules:+    Benchmark.Chan+    Benchmark.Simple++  build-depends:+    async,+    base >=4.16 && <5,+    lossy-chan,+    stm,+    unagi-chan,+    weigh,
+ src/Control/Concurrent/LossyChan.hs view
@@ -0,0 +1,450 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE MagicHash #-}+{-# LANGUAGE UnboxedTuples #-}++module Control.Concurrent.LossyChan (+  InChan,+  OutChan,+  new,+  dup,+  send,+  receive,+  tryReceive,+  NoInChanError (..),+) where++import Control.Exception (Exception (..), mask)+import GHC.Conc (STM (..), TVar (..), atomically, newTVar, newTVarIO, orElse, readTVar, retry, throwSTM, writeTVar)+import GHC.Exts (Int#, MutVar#, RealWorld, SmallArray#, State#, TVar#, Weak#, Word32#, deRefWeak#, indexSmallArray#, isTrue#, mkWeak#, newMutVar#, newSmallArray#, newTVar#, readTVar#, unsafeFreezeSmallArray#, word2Int#, word32ToWord#, writeSmallArray#, writeTVar#, (+#), (==#))+import GHC.IO (IO (..))+import GHC.Word (Word32 (..))++#ifdef LOSSY_CHAN_SMALL_LOOP_COUNTER+-- If the small-loop-counter flag is set, use Word8.+import Data.Word (Word8)++type LoopCounter = Word8+#else+-- Otherwise, use Word32.+type LoopCounter = Word32+#endif++{- |+The position in a channel ring buffer.+-}+data ChanPos = MkChanPos+  { loop :: {-# UNPACK #-} !LoopCounter+  {- ^+  This value tracks the current loop around the channel ring buffer, which is+  used to determine whether or not a channel position refers to a live value.++  The `maxBound` of this type determines the resilience of the `OutChan`.+  If a receiver stops at loop @N@ and does not resume until the sender has+  written @`maxBound` - 1@ or @`maxBound`@ values and the loop counter has+  wrapped around all the way back to loop @N - 1 @ or @N@, then the receiver+  will treat its current position as live, which may cause it to skip some of+  the messages in the buffer.+  -}+  , next :: {-# UNPACK #-} !Word32+  {- ^+  This value tracks the next position that should be read (for receivers) or+  written to (for senders).+  -}+  }+  deriving (Eq, Show)++{- |+The initial position in a channel ring buffer.+-}+zero :: ChanPos+zero = MkChanPos{loop = 0, next = 0}+{-# INLINE zero #-}++{- |+Advance to the next channel position.+-}+advance :: Word32# -> ChanPos -> ChanPos+advance len# MkChanPos{..}+  | next < W32# len# - 1 = MkChanPos{next = next + 1, ..}+  | otherwise = MkChanPos{loop = loop + 1, next = 0}+{-# INLINE advance #-}++{- |+Get the latest of the current output position or the oldest live position.+-}+live :: ChanPos -> ChanPos -> ChanPos+live icPos ocPos+  -- The output channel is on the same loop as the input channel, and+  -- the position is behind (or just caught up with) the input channel.+  | loop ocPos == loop icPos && next ocPos <= next icPos = ocPos+  -- The output channel is one loop behind the input channel, and the+  -- position is aheaad of the input channel.+  | loop ocPos == loop icPos - 1 && next ocPos >= next icPos = ocPos+  -- The output channel is several loops behind the input channel.+  | otherwise = icPos{loop = loop icPos - 1}+{-# INLINE live #-}++data Entry a+  = Empty+  | Entry {seen :: {-# UNPACK #-} !Word, val :: !a}++{- |+The channel ring buffer that is used in `InChan`.++The channel ring buffer is an unboxed `SmallArray#` of unboxed `TVar#`. These+types are explicitly unboxed, because we cannot rely on an @UNPACK@ pragma to+unpack the @TVar@ inside of a @SmallArray@.+-}+data ChanBuf a = MkChanBuf+  { len# :: Word32#+  -- ^ The length of the channel ring buffer.+  , arr# :: SmallArray# (TVar# RealWorld (Entry a))+  -- ^ The array that backs the channel ring buffer.+  }++{- |+Create a new channel buffer.+-}+newChanBuf :: Word32 -> IO (ChanBuf a)+newChanBuf (W32# len#) = IO $ \s1# ->+  case word32ToInt# len# of+    ilen# ->+      case newTVar# Empty s1# of+        (# s2#, var0# #) ->+          case newSmallArray# (word32ToInt# len#) var0# s2# of+            (# s3#, mutarr# #) ->+              let go :: Int# -> State# RealWorld -> State# RealWorld+                  go i# = \s4# ->+                    if isTrue# (i# ==# ilen#)+                      then s4#+                      else case newTVar# Empty s4# of+                        (# s5#, vari# #) ->+                          case writeSmallArray# mutarr# i# vari# s5# of+                            s6# -> go (i# +# 1#) s6#+               in case unsafeFreezeSmallArray# mutarr# (go 1# s3#) of+                    (# s7#, arr# #) ->+                      (# s7#, MkChanBuf len# arr# #)++{- |+Read the value from the given position in a `ChanBuf`.+-}+takeChanBuf :: ChanBuf a -> Word -> Word32 -> STM (Maybe a)+takeChanBuf MkChanBuf{..} ocCount (W32# i#) =+  STM $ \s1# ->+    case indexSmallArray# arr# (word32ToInt# i#) of+      (# vari# #) ->+        case readTVar# vari# s1# of+          (# s2#, Empty #) ->+            (# s2#, Nothing #)+          (# s2#, Entry seen val #)+            | seen + 1 >= ocCount ->+                case writeTVar# vari# Empty s2# of+                  s3# -> (# s3#, Just val #)+            | otherwise ->+                case writeTVar# vari# Entry{seen = seen + 1, ..} s2# of+                  s3# -> (# s3#, Just val #)++{- |+Write the given value to a `ChanBuf` at the given position.+-}+writeChanBuf :: ChanBuf a -> Word32 -> Entry a -> STM ()+writeChanBuf MkChanBuf{..} (W32# i#) a =+  STM $ \s1# ->+    case indexSmallArray# arr# (word32ToInt# i#) of+      (# vari# #) ->+        case writeTVar# vari# a s1# of+          s2# ->+            (# s2#, () #)++{- |+Check the `ChanBuf` `Entry` at the given position. If the `seen` count is+greater than or equal to the new (decremented) output channel count, free+the entry.++This function returns `Just` `True` if the entry was freed, `Just` `False`+if the entry was not freed, and `Nothing` if the entry was uninitialised.+-}+tryFreeChanBufAt :: ChanBuf a -> Word -> Word32 -> STM (Maybe Bool)+tryFreeChanBufAt MkChanBuf{..} ocCount (W32# i#) =+  STM $ \s1# ->+    case indexSmallArray# arr# (word32ToInt# i#) of+      (# vari# #) ->+        case readTVar# vari# s1# of+          (# s2#, Empty #) ->+            (# s2#, Nothing #)+          (# s2#, Entry seen _val #)+            | seen >= ocCount ->+                case writeTVar# vari# Empty s2# of+                  s3# -> (# s3#, Just True #)+            | otherwise ->+                (# s2#, Just False #)++{- |+The channel state.+-}+data ChanState a = MkChanState+  { buffer :: {-# UNPACK #-} !(ChanBuf a)+  -- ^ The underlying channel ring buffer.+  , icPosVar :: {-# UNPACK #-} !(TVar ChanPos)+  -- ^ A `TVar` that holds the current input channel position.+  , ocCountVar :: {-# UNPACK #-} !(TVar Word)+  -- ^ A `TVar` that holds the current output channel count.+  }++{- |+An input channel.+-}+data InChan a = MkInChan+  { icState :: {-# UNPACK #-} !(ChanState a)+  {- ^+  The underlying channel state.++  This is shared between all input and output channels.+  -}+  , icLive :: {-# UNPACK #-} !Live+  {- ^+  When this value is dropped, the corresponding `IsLive`+  values in the output channels can be used to detect it.+  -}+  }++{- |+An output channel.+-}+data OutChan a = MkOutChan+  { ocState :: !(ChanState a)+  {- ^+  The underlying channel state.++  This is shared between all input and output channels.+  -}+  , ocIsLive :: !IsLive+  -- ^ A value used test whether the corresponding input channel is live.+  , ocPosVar :: !(TVar ChanPos)+  -- ^ A variable that holds this output channel's position.+  }++{- |+Free an output channel.++__Warning:__+This is the only STM operation that walks the channel buffer. Therefore, if+there's a lot of competition, this STM operation may repeatedly retry, which+would prevent the buffer entries from being garbage collected until the point+where the input channel position passed this output channel's position.+-}+freeOutChan :: OutChan a -> IO ()+freeOutChan MkOutChan{ocState = MkChanState{buffer = buffer@MkChanBuf{..}, ..}, ..} =+  atomically $ do+    -- Decrement the output channel counter.+    ocCount <- subtract 1 <$> readTVar ocCountVar+    writeTVar ocCountVar ocCount++    -- Walk the buffer to free values that have been seen by every remaining OutChan.+    icPos <- readTVar icPosVar+    ocPos <- live icPos <$> readTVar ocPosVar+    let go cp+          | icPos == cp = pure ()+          | otherwise = do+              mbSuccess <- tryFreeChanBufAt buffer ocCount (next cp)+              case mbSuccess of+                Nothing+                  | loop icPos == 0 && next icPos <= next cp -> go zero+                  | otherwise -> pure ()+                Just True -> go (advance len# cp)+                Just False -> pure ()+    go ocPos++{- |+Add a finalizer to a `TVar`.++Adapted from @mkWeakTVar in @stm@.+-}+addFinalizerTVar :: TVar a -> IO () -> IO ()+addFinalizerTVar t@(TVar t#) (IO finalizer) = IO $ \s1# ->+  case mkWeak# t# t finalizer s1# of+    (# s2, _ #) ->+      (# s2, () #)++{- |+@`new` len@ creates an input channel that is backed by a ring buffer with length @len@.++The ring buffer is represented as a @SmallArray@, so @len@ should be smaller than 128.+-}+new :: Word32 -> IO (InChan a, OutChan a)+new len = do+  -- Create the input channel.+  buffer <- newChanBuf len+  icPosVar <- newTVarIO zero+  ocCountVar <- newTVarIO 1+  icLive <- mkLive+  let icState = MkChanState{..}+  let ic = MkInChan{..}++  -- Create the output channel.+  ocPosVar <- newTVarIO zero+  let ocState = icState+  ocIsLive <- mkIsLive icLive+  let oc = MkOutChan{..}+  addFinalizerTVar ocPosVar (freeOutChan oc)++  pure (ic, oc)++{- |+@`send` i x@ sends the value @x@ over the input channel @i@.+-}+send :: InChan a -> a -> IO ()+send MkInChan{icState = MkChanState{buffer = buffer@MkChanBuf{..}, ..}} a =+  atomically $ do+    icPos@MkChanPos{..} <- readTVar icPosVar+    writeChanBuf buffer next Entry{seen = 0, val = a}+    writeTVar icPosVar (advance len# icPos)++{- |+@`dup` o@ duplicates the output channel @o@.+-}+dup :: OutChan a -> IO (OutChan a)+dup oc@MkOutChan{ocState = ocState@MkChanState{..}, ..} =+  -- This operation is masked to ensure that the finalizer is attached.+  mask $ \_restore -> do+    -- These two operations are performed atomically. Otherwise, if we're very+    -- unlucky with reorderings, we may get this interleaving between dup and+    -- receive on the same output channel:+    --+    -- ocCount <- readTVar ocCountVar          [receive]+    -- incrOcCountVar ocCountVar               [dup, (1)]+    -- newTVar =<< readTVar ocPosVar           [dup, (2)]+    -- writeTVar ocPosVar (advance ocPos)      [receive]+    -- takeChanBuf buffer ocCount (next ocPos) [receive]+    --+    -- The takeChanBuf operation may then decide, based on an outdated output+    -- channel count, to replace the Entry with Empty, which causes the new+    -- output channel position to point to a forgotten value.+    ocPosVar' <- atomically $ do+      -- (1) Increment the output channel counter.+      writeTVar ocCountVar . (+ 1) =<< readTVar ocCountVar++      -- (2) Duplicate the output channel position variable.+      newTVar =<< readTVar ocPosVar++    addFinalizerTVar ocPosVar' (freeOutChan oc)+    pure MkOutChan{ocPosVar = ocPosVar', ..}++{- |+The error thrown by `receive` and `tryReceive` when the corresponding `InChan`+has been garbage collected and the channel buffer is empty.+-}+data NoInChanError = ErrNoInChan+  deriving (Show)++instance Exception NoInChanError++{- |+@`receiveT` o@ receives a value from an output channel and+retries if none are available.++__Warning:__+If the corresponding input channel was garbage collected and+the buffer is empty, this function throws an `NoInChanError`.+-}+receiveT :: OutChan a -> STM a+receiveT MkOutChan{ocState = MkChanState{buffer = buffer@MkChanBuf{..}, ..}, ..} = do+  icPos <- readTVar icPosVar+  ocPos <- live icPos <$> readTVar ocPosVar+  receiveT_ icPos ocPos+ where+  receiveT_ icPos ocPos+    | icPos == ocPos = do+        -- If the input channel is still live...+        icLive <- isLive ocIsLive+        if icLive+          -- ...we retry.+          then retry+          -- Otherwise, throw an NoInChanError.+          else throwSTM ErrNoInChan+    | otherwise = do+        ocCount <- readTVar ocCountVar+        mbVal <- takeChanBuf buffer ocCount (next ocPos)+        case mbVal of+          Nothing -> retry+          Just val -> writeTVar ocPosVar (advance len# ocPos) >> pure val++{- |+@`receive` o@ receives a value from an output channel and+blocks if none is available.++__Warning:__+If the corresponding input channel was garbage collected and+the buffer is empty, this function throws an `NoInChanError`.+-}+receive :: OutChan a -> IO a+receive o =+  atomically (receiveT o)++{- |+@`tryReceive` o@ receives `Just` a value from an output channel+and returns `Nothing` if none are available.++__Warning:__+If the corresponding input channel was garbage collected and+the buffer is empty, this function throws an `NoInChanError`.+-}+tryReceive :: OutChan a -> IO (Maybe a)+tryReceive o = do+  atomically ((Just <$> receiveT o) `orElse` (pure Nothing))++{- |+This type is used to test for liveness.++The target should hold on to a value of this type, the tester can use+the corresponding `IsLive` to test whether the target is still live.+-}+data Live = MkLive (MutVar# RealWorld ())++{- |+Create a value of type `Live`.+-}+mkLive :: IO Live+mkLive =+  IO $ \s1# ->+    case newMutVar# () s1# of+      (# s2#, mv# #) ->+        (# s2#, MkLive mv# #)++{- |+This type is used to test for liveness.++See `Live`.+-}+data IsLive = MkIsLive (Weak# (MutVar# RealWorld ()))++{- |+Create an `IsLive` value that tests whether the given `Live` value is live.+-}+mkIsLive :: Live -> IO IsLive+mkIsLive (MkLive mv#) =+  IO $ \s1# ->+    case mkWeak# mv# mv# (\s3# -> (# s3#, () #)) s1# of+      (# s2#, wmv# #) ->+        (# s2#, MkIsLive wmv# #)++{- |+Test whether the `Live` value that corresponds to the given `IsLive`+value is still live.+-}+isLive :: IsLive -> STM Bool+isLive (MkIsLive wmv#) =+  STM $ \s1# ->+    case deRefWeak# wmv# s1# of+      (# s2#, flag#, _ #) ->+        case flag# of+          0# -> (# s2#, False #)+          _ -> (# s2#, True #)++{- |+The channel ring buffer is indexed by a `Word32`,+which we must convert to an `Int` to index the underlying `SmallArray#`.+-}+word32ToInt# :: Word32# -> Int#+word32ToInt# = \i# -> word2Int# (word32ToWord# i#)+{-# INLINE word32ToInt# #-}
+ test/Main.hs view
@@ -0,0 +1,474 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE InstanceSigs #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE TypeApplications #-}++module Main where++import Control.Concurrent.LossyChan qualified as LC+import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar)+import Control.Exception (ArrayException (IndexOutOfBounds), assert, handle, throw)+import Control.Monad (replicateM, unless, void)+import Data.Foldable (for_)+import Data.IORef (IORef, mkWeakIORef, modifyIORef', newIORef, readIORef)+import Data.Traversable (for)+import Data.Word (Word32, Word8)+import GHC.Stack (HasCallStack)+import System.Mem (performMajorGC)+import System.Mem.Weak (Weak)+import System.Timeout (timeout)+import Test.Tasty (TestTree, defaultMain, testGroup)+import Test.Tasty.HUnit (assertFailure, testCase, (@?=))+import Test.Tasty.QuickCheck (Arbitrary (..), Gen, Property, chooseBoundedIntegral, counterexample, forAll, ioProperty, label, oneof, testProperty)++main :: IO ()+main = do+  defaultMain $+    testGroup+      "Tests"+      [ unitTests 2+      , unitTests 8+      , unitTests 32+      , unitTests 128+      , propTests+      ]++{- |+Property-based tests.+-}+propTests :: TestTree+propTests =+  testGroup "Prop" $+    [ testProperty "lossless when keeping up with alternating sender and receiver" prop_LosslessWhenKeepingUpSimp+    , testProperty "lossless when keeping up with arbitrary trace" prop_LosslessWhenKeepingUpArb+    , testProperty "lose 1 when lagging by N+1" prop_Lose1WhenLaggingByNPlus1+    , testProperty "lose N when lagging by 2N+1" prop_LoseNWhenLaggingBy2NPlus1+    ]++{- |+Test that a @lossy-chan@ channel is lossless if the receiver keeps up.++This test uses a single sender and a single receiver that alternate sending and receiving.+-}+prop_LosslessWhenKeepingUpSimp :: Property+prop_LosslessWhenKeepingUpSimp =+  forAll (chooseBoundedIntegral (1, 128)) $ \n ->+    forAll (chooseBoundedIntegral (1, 2 * n * maxLoop)) $ \m ->+      label (if n >= m then "N >= M" else "N <  M") $+        ioProperty $ do+          (i, o) <- LC.new @Word32 n+          let xs = [0 .. m]+          and+            <$> ( for xs $ \x -> do+                    LC.send i x+                    mby <- LC.tryReceive o+                    pure (mby == Just x)+                )++{- |+Test that a @lossy-chan@ channel is lossless if all receivers keep up.++This test uses an arbitrary trace.+-}+prop_LosslessWhenKeepingUpArb :: TraceWithConfig -> Property+prop_LosslessWhenKeepingUpArb twc@(MkTraceWithConfig MkTraceConfig{..} _) =+  label ("with " <> nLabel <> ", " <> mLabel <> ", " <> oknLabel <> ", " <> olnLabel) $+    ioProperty (runTrace twc)+ where+  nLabel :: String+  nLabel+    | n >= 128 = "N >= 128"+    | n >= 64 = "N >= 64 "+    | n >= 32 = "N >= 32 "+    | n >= 16 = "N >= 16 "+    | n >= 8 = "N >= 8  "+    | n >= 4 = "N >= 4  "+    | n >= 2 = "N >= 2  "+    | otherwise = "N == 1  "+  mLabel :: String+  mLabel+    | m >= n * maxLoop = "M >= N * maxLoop"+    | otherwise = "M <  N * maxLoop"+  oknLabel :: String+  oknLabel+    | okn >= 2 = "multiple timely receivers"+    | otherwise = "one timely receiver"+  olnLabel :: String+  olnLabel+    | oln >= 2 = "multiple lagging receivers"+    | oln == 1 = "one lagging receiver"+    | otherwise = "no lagging receivers"++{- |+Test that after N+1 sends the first value received is the second value sent.+-}+prop_Lose1WhenLaggingByNPlus1 :: Property+prop_Lose1WhenLaggingByNPlus1 =+  forAll (chooseBoundedIntegral (1, 128)) $ \n ->+    ioProperty $ do+      (i, o) <- LC.new @Word32 n+      let xs = [0 .. n]+      for_ xs $ (LC.send i)+      mby <- LC.tryReceive o+      pure $+        counterexample ("Expected 1, but received " <> show mby) $+          mby == Just 1++{- |+Test that after N+1 sends the first value received is the second value sent.+-}+prop_LoseNWhenLaggingBy2NPlus1 :: Property+prop_LoseNWhenLaggingBy2NPlus1 =+  forAll (chooseBoundedIntegral (1, 128)) $ \n ->+    ioProperty $ do+      (i, o) <- LC.new @Word32 n+      let xs = [0 .. 2 * n]+      for_ xs $ (LC.send i)+      mby <- LC.tryReceive o+      pure $+        counterexample ("Expected " <> show (n + 1) <> ", but received " <> show mby) $+          mby == Just (n + 1)++{- |+Unit tests.+-}+unitTests :: Word32 -> TestTree+unitTests n =+  testGroup ("Unit (with N = " <> show n <> ")") $+    [ testCase "send i 1 >> receive o ==> 1" $ do+        (i, o) <- LC.new @Word32 n+        v <- LC.send i 1 >> LC.receive o+        v @?= 1+    , testCase "for [1,2] (\\x -> send i x >> receive 0) ==> [1,2]" $ do+        (i, o) <- LC.new @Int n+        x1 <- LC.send i 1 >> LC.receive o+        x1 @?= 1+        x2 <- LC.send i 2 >> LC.receive o+        x2 @?= 2+    , testCase "send i 1 >> receive o >> receive o ==> block" $ do+        (i, o) <- LC.new @Word32 n+        v <- LC.send i 1 >> LC.receive o+        v @?= 1+        w <- timeout 1000 (LC.receive o)+        w @?= Nothing+    , testCase "for_ [0..N] (send i) >> receive o ==> 1" $ do+        (i, o) <- LC.new @Word32 n+        v <- for_ [0 .. n] (LC.send i) >> LC.receive o+        v @?= 1+    , testCase "for_ [0..N*2] (send i) >> receive o ==> N + 1" $ do+        (i, o) <- LC.new @Word32 n+        v <- for_ [0 .. n * 2] (LC.send i) >> LC.receive o+        v @?= n + 1+    , testCase "for_ [0..N*L+1] (send i) >> receive o ==> N * (L - 1) + 1" $ do+        (i, o) <- LC.new @Word32 n+        v <- for_ [0 .. n * maxLoop] (LC.send i) >> LC.receive o+        v @?= n * (maxLoop - 1) + 1+    , testCase "send i 1 >> traverse receive [o1,o2] ==> [1,1]" $ do+        (i, o1) <- LC.new @Word32 n+        o2 <- LC.dup o1+        vs <- LC.send i 1 >> traverse LC.receive [o1, o2]+        vs @?= [1, 1]+    , testCase "early free" $ do+        p1 <- newEmptyMVar+        t1 <- mkTattleTale p1+        (i, o) <- LC.new n+        LC.send i (Left t1)+        LC.send i (Right ())+        void (LC.receive o)+        performMajorGC+        timeout 100000 (takeMVar p1)+          >>= maybe (assertFailure "t1 was not freed") pure+    , testCase "receive hangs if InChan is dropped" $ do+        let handleNoChan LC.ErrNoInChan = pure $ Left LC.ErrNoInChan+        (_, o) <- LC.new @() n+        performMajorGC+        v <- handle handleNoChan (Right <$> timeout 10000 (LC.receive o))+        case v of+          Left LC.ErrNoInChan -> pure ()+          Right _ -> assertFailure "ErrNoInChan exception not thrown."+    ]++--------------------------------------------------------------------------------+-- Maximum Loop Count+--------------------------------------------------------------------------------++{- |+The maximum loop value for a @ChanPos@.++This value is the `maxBound` of `Word8`. In the actual implementation,+`maxLoop` would be `maxBound` of `Word32`. To test the overflow cases,+build the library with @--flags=+small-loop-counter@.+-}+maxLoop :: Word32+maxLoop = fromIntegral (maxBound :: Word8)++--------------------------------------------------------------------------------+-- TattleTale+--------------------------------------------------------------------------------++{- |+A wrapper around an `IORef` and a `Weak` pointer.++Construct with `mkTattleTale` to get a value that pings an `MVar` when its finalizer runs.+-}+data TattleTale = TattleTale (IORef ()) (Weak (IORef ()))++{- |+Constructor for `TattleTale`.+-}+mkTattleTale :: MVar () -> IO TattleTale+mkTattleTale ping = do+  ref <- newIORef ()+  weakRef <- mkWeakIORef ref (putMVar ping ())+  pure $ TattleTale ref weakRef++--------------------------------------------------------------------------------+-- Arbitrary Traces+--------------------------------------------------------------------------------++{- |+This type holds a `Trace` with the `TraceConfig` used for its generation.+-}+data TraceWithConfig = MkTraceWithConfig TraceConfig Trace+  deriving (Show)++instance Arbitrary TraceWithConfig where+  arbitrary = do+    tc <- genTraceConfig+    t <- genTrace tc+    pure $ MkTraceWithConfig tc t++  shrink (MkTraceWithConfig tc t) = do+    tc' <- shrinkTraceConfig tc+    let t' = shrinkTrace tc' t+    pure $ MkTraceWithConfig tc' t'++{- |+This type holds the configuration for trace generation.+-}+data TraceConfig = MkTraceConfig+  { n :: !Word32+  -- ^ The size of the channel ring buffer.+  , m :: !Word32+  -- ^ The total number of messages to send.+  , okn :: !Int+  -- ^ The total number of receivers that must keep up.+  , oln :: !Int+  -- ^ The total number of receivers that may lag behind.+  }+  deriving (Show)++{- |+This type represents the operations on a @lossy-chan@ channel.+-}+data TraceOp+  = -- | @`Send` x t@ sends @x@ over the `LC.InChan`.+    Send !Word32+  | -- | @`ReceiveK` k x t@ receives @x@ over the @k@'th `LC.OutChan` that must keep up.+    ReceiveK !Int !Word32+  | -- | @`Receive` l t@ receives some value over the @l@'th `LC.OutChan` that may lag behind.+    ReceiveL !Int+  deriving (Show)++infixr 1 :>++{- |+This type represents a trace of the usage of a @lossy-chan@ channel.+-}+data Trace+  = TraceOp :> Trace+  | End++instance Show Trace where+  showsPrec p (to :> t) = showParen (p > 1) $ showsPrec 1 to . showString " :> " . showsPrec 1 t+  showsPrec _ End = showString "End"++{- |+This type holds the state of the system after at some point in a `Trace`.+-}+data TraceState = MkTraceState+  { ic :: !Word32+  -- ^ Number of messages sent over `LC.InChan`.+  , okcs :: ![Word32]+  -- ^ Number of messages received per `LC.OutChan` that must keep up.+  , olcs :: ![Word32]+  -- ^ Number of messages received per `LC.OutChan` that may lag behind.+  }+  deriving (Show)++{- |+The initial system state. No messages have been sent or received.+-}+initTraceState :: TraceConfig -> TraceState+initTraceState MkTraceConfig{..} =+  MkTraceState{ic = 0, okcs = replicate okn 0, olcs = replicate oln 0, ..}++{- |+Run the given trace.+-}+runTrace :: TraceWithConfig -> IO ()+runTrace (MkTraceWithConfig MkTraceConfig{..} t0) = do+  (i, o) <- LC.new n+  oks <- (o :) <$> replicateM (okn - 1) (LC.dup o)+  ols <- replicateM oln (LC.dup o)+  olxs <- newIORef (replicate oln Nothing)+  let go :: Trace -> IO ()+      go End = pure ()+      go (Send x :> t) = do+        LC.send i x+        go t+      go (ReceiveK k x :> t) = do+        let ok = oks !! k+        y <- LC.receive ok+        unless (x == y) $ error $ "Expected " <> show x <> ", but received " <> show y+        go t+      go (ReceiveL l :> t) = do+        let ol = ols !! l+        mbx <- (!! l) <$> readIORef olxs+        mby <- handle (\LC.ErrNoInChan -> pure Nothing) (LC.tryReceive ol)+        case (mbx, mby) of+          (Just x, Just y)+            | x == y -> error $ "Received duplicate message " <> show x+            | x > y -> error $ "Received " <> show x <> " before " <> show y+          _ -> pure ()+        modifyIORef' olxs ((const mby) `at` l)+        go t+  go t0++--------------------------------------------------------------------------------+-- Generation++{- |+Generate an arbitrary `TraceConfig`.+-}+genTraceConfig :: Gen TraceConfig+genTraceConfig = do+  n <- chooseBoundedIntegral (1, 128)+  m <- chooseBoundedIntegral (2 * n, 2 * n * maxLoop)+  okn <- chooseBoundedIntegral (2, 4)+  oln <- chooseBoundedIntegral (0, 2)+  pure MkTraceConfig{..}++{- |+Generate an arbitrary `Trace`.+-}+genTrace :: TraceConfig -> Gen Trace+genTrace tc =+  go (id, initTraceState tc)+ where+  go :: (Trace -> Trace, TraceState) -> Gen Trace+  go (tf, ts) = case step tc ts of+    [] -> assert (final tc ts) $ pure (tf End)+    op_ts's -> oneof [go (tf . (op :>), ts') | (op, ts') <- op_ts's]++{- |+List all possible traces for the `TraceConfig`.+-}+allTraces :: TraceConfig -> [Trace]+allTraces tc =+  go (id, initTraceState tc)+ where+  go :: (Trace -> Trace, TraceState) -> [Trace]+  go (tf, ts) = case step tc ts of+    [] -> assert (final tc ts) [tf End]+    op_ts's -> [t | (op, ts') <- op_ts's, t <- go (tf . (op :>), ts')]++{- |+Apply a `TraceOp` to the `TraceState`.+-}+tr :: TraceState -> TraceOp -> TraceState+tr MkTraceState{..} = \case+  Send _ -> MkTraceState{ic = ic + 1, ..}+  ReceiveK k _ -> MkTraceState{okcs = ((+ 1) `at` k) okcs, ..}+  ReceiveL l -> MkTraceState{olcs = ((+ 1) `at` l) olcs, ..}++{- |+Generate a list of valid `TraceOp`.+-}+step :: TraceConfig -> TraceState -> [(TraceOp, TraceState)]+step tc ts =+  assert (not (null op_ts's) || final tc ts) op_ts's+ where+  op_ts's = [(op, ts') | op <- ops tc ts, let ts' = tr ts op, valid tc ts']++{- |+Generate a list of possible `TraceOp`.+-}+ops :: TraceConfig -> TraceState -> [TraceOp]+ops MkTraceConfig{..} MkTraceState{..} =+  concat+    [ [Send ic]+    , [ReceiveK k (okcs !! k) | k <- [0 .. okn - 1]]+    , [ReceiveL l | l <- [0 .. oln - 1]]+    ]++final :: TraceConfig -> TraceState -> Bool+final MkTraceConfig{..} MkTraceState{..} =+  ic == m && all (\okc -> okc == m) okcs && all (\olc -> olc <= m) olcs++valid :: TraceConfig -> TraceState -> Bool+valid MkTraceConfig{..} MkTraceState{..} =+  and+    [ -- At most m messages are sent.+      ic <= m+    , -- The difference between the number of sent and the number of received+      -- messages by channels that must keep up is at least 0 and at most n.+      all (\okc -> let dm = asInt ic - asInt okc in 0 <= dm && dm <= asInt n) okcs+    , -- The difference between the number of sent and the number of received+      -- messages by channels that may lag behind is at least 0.+      all (\olc -> let dm = asInt ic - asInt olc in 0 <= dm) olcs+    ]+ where+  asInt :: Word32 -> Int+  asInt = fromIntegral++{- |+Apply the given function at the given index in the given list.+-}+at :: (HasCallStack) => (a -> a) -> Int -> [a] -> [a]+(_ `at` k) [] = throw (IndexOutOfBounds $ show k)+(f `at` 0) (x : xs) = f x : xs+(f `at` k) (x : xs) = x : (f `at` (k - 1)) xs++--------------------------------------------------------------------------------+-- Shrinking++{- |+Shrink a `TraceConfig`.+-}+shrinkTraceConfig :: TraceConfig -> [TraceConfig]+shrinkTraceConfig MkTraceConfig{..}+  | null tcs = []+  | otherwise = init tcs+ where+  tcs = do+    n' <- [1 .. n]+    m' <- [1 .. m]+    okn' <- [1 .. okn]+    oln' <- [0 .. oln]+    pure $ MkTraceConfig n' m' okn' oln'++{- |+Shrink a `Trace` to match the new `TraceConfig`.++__Warning:__+This function assumes that the given `TraceConfig` is smaller than the+`TraceConfig` that was used to generate the trace.+-}+shrinkTrace :: TraceConfig -> Trace -> Trace+shrinkTrace tc = go (initTraceState tc)+ where+  go :: TraceState -> Trace -> Trace+  go _ts End = End+  go ts (op :> t)+    | inBounds tc op, let ts' = tr ts op, valid tc ts = op :> go ts' t+    | otherwise = go ts t++{- |+Check whether a `TraceOp` is in bounds.+-}+inBounds :: TraceConfig -> TraceOp -> Bool+inBounds MkTraceConfig{..} = \case+  Send _ -> True+  ReceiveK k _ -> k < okn+  ReceiveL l -> l < oln