lossy-chan 0.1.0.0 → 0.1.1.0

raw patch · 5 files changed

+142/−9 lines, 5 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +5/−1
• README.md +25/−1
• lossy-chan.cabal +1/−1
• src/Control/Concurrent/LossyChan.hs +82/−5
• test/Main.hs +29/−1

@@ -1,5 +1,9 @@ # Revision history for lossy-chan -## 0.1.0.0 -- YYYY-mm-dd+## 0.1.1.0 -- 2026-11-12++* Support creating a new output channel from the input channel.++## 0.1.0.0 -- 2026-11-11  * First version. Released on an unsuspecting world.

@@ -23,7 +23,7 @@  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.+If you want two output channels 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@@ -39,6 +39,30 @@  m2' <- LC.receive o1      -- receive the 2 on o1 assertM (m2' == 2)+```++Alternatively, you can create a new output channel from the input channel.+In this case, the new output channel receives messages starting from the oldest live message.+This is non-determinstic, as it depends on garbage collection. For example:++```haskell+(i, _o) <- LC.new 64++-- if GC runs here, there won't be any live output+-- channels when send runs and the 1 is discarded.++LC.send i 1++-- if GC runs here, _o's finalizer removes the 1.++o <- LC.out i++-- if GC runs from here onwards, the 1 is kept.++LC.send i 2++m <- LC.receive o+assertM (m `elem` [1,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:

@@ -1,6 +1,6 @@ cabal-version: 3.0 name: lossy-chan-version: 0.1.0.0+version: 0.1.1.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

@@ -7,13 +7,15 @@   OutChan,   new,   dup,+  out,   send,   receive,   tryReceive,   NoInChanError (..), ) where -import Control.Exception (Exception (..), mask)+import Control.Exception (Exception (..), mask_)+import Control.Monad (when) 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 (..))@@ -145,6 +147,20 @@                   s3# -> (# s3#, Just val #)  {- |+Read the value from the given position in a `ChanBuf`.+-}+isEmptyChanBuf :: ChanBuf a -> Word32 -> STM Bool+isEmptyChanBuf MkChanBuf{..} (W32# i#) =+  STM $ \s1# ->+    case indexSmallArray# arr# (word32ToInt# i#) of+      (# vari# #) ->+        case readTVar# vari# s1# of+          (# s2#, Empty #) ->+            (# s2#, True #)+          (# s2#, Entry{} #) ->+            (# s2#, False #)++{- | Write the given value to a `ChanBuf` at the given position. -} writeChanBuf :: ChanBuf a -> Word32 -> Entry a -> STM ()@@ -210,6 +226,9 @@  {- | An output channel.++An output channel has a finalizer that frees unreachable values in the ring buffer.+The worst-case complexity of this operation is $O(N)$ where $N$ is the length of the ring buffer. -} data OutChan a = MkOutChan   { ocState :: !(ChanState a)@@ -292,21 +311,27 @@  {- | @`send` i x@ sends the value @x@ over the input channel @i@.++__Warning:__ If there are no output channels, the value is discarded. -} 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)+    ocCount <- readTVar ocCountVar+    when (ocCount > 0) $ 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@.++The new output channel starts receiving messages from the same position as the duplicated channel. -} 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+  mask_ $ 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:@@ -329,6 +354,58 @@      addFinalizerTVar ocPosVar' (freeOutChan oc)     pure MkOutChan{ocPosVar = ocPosVar', ..}++{- |+@`out` i@ creates a new output channel from the input channel @i@.++This worst-case complexity of this operation is $O(N)$ where $N$ is the length of the ring buffer.+-}+out :: InChan a -> IO (OutChan a)+out MkInChan{icState = icState@MkChanState{..}, ..} =+  -- This operation is masked to ensure that the finalizer is attached.+  mask_ $ do+    -- See `dup` for an explanation for the use of `atomically`.+    ocPosVar <- atomically $ do+      -- (1) Increment the output channel counter.+      writeTVar ocCountVar . (+ 1) =<< readTVar ocCountVar++      -- (2) Set the output channel position variable.+      icPos <- readTVar icPosVar+      newTVar icPos{loop = loop icPos - 1}++    -- Create the output channel.+    let ocState = icState+    ocIsLive <- mkIsLive icLive+    let oc = MkOutChan{..}+    addFinalizerTVar ocPosVar (freeOutChan oc)++    -- Advance to the next non-Empty position.+    advanceToNonEmpty oc++    -- Return the output channel.+    pure oc++{- |+Find the next non-`Empty` position in the channel buffer.+-}+advanceToNonEmpty :: OutChan a -> IO ()+advanceToNonEmpty MkOutChan{ocState = MkChanState{buffer = buffer@MkChanBuf{..}, ..}, ..} = do+  let go = do+        continue <-+          atomically $ do+            icPos <- readTVar icPosVar+            ocPos <- live icPos <$> readTVar ocPosVar+            if icPos == ocPos+              then pure False+              else do+                isEmpty <- isEmptyChanBuf buffer (next ocPos)+                if isEmpty+                  then writeTVar ocPosVar (advance len# ocPos) >> pure True+                  else pure False+        when continue go+  go++--  {- | The error thrown by `receive` and `tryReceive` when the corresponding `InChan`

@@ -5,6 +5,7 @@  module Main where +import Control.Concurrent (threadDelay) import Control.Concurrent.LossyChan qualified as LC import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar) import Control.Exception (ArrayException (IndexOutOfBounds), assert, handle, throw)@@ -135,7 +136,34 @@ unitTests :: Word32 -> TestTree unitTests n =   testGroup ("Unit (with N = " <> show n <> ")") $-    [ testCase "send i 1 >> receive o ==> 1" $ do+    [ testCase "out continues from oldest live value" $ do+        (i, o1) <- LC.new @Word32 n+        for_ [1, 2] (LC.send i)++        -- receive 1; is no longer live+        m1 <- LC.receive o1+        m1 @?= 1++        -- create o2+        o2 <- LC.out i++        -- receive 2 on o1; remains live+        m2 <- LC.receive o1+        m2 @?= 2++        -- receive 2 on o2+        m2' <- LC.receive o2+        m2' @?= 2+    , testCase "send discards message if there are no output channels" $ do+        (i, _) <- LC.new @Word32 n+        performMajorGC+        threadDelay 1000000+        LC.send i 1+        o <- LC.out i+        LC.send i 2+        m2 <- LC.receive o+        m2 @?= 2+    , testCase "send i 1 >> receive o ==> 1" $ do         (i, o) <- LC.new @Word32 n         v <- LC.send i 1 >> LC.receive o         v @?= 1