tmapchan (empty) → 0.0.0

raw patch · 6 files changed

+332/−0 lines, 6 filesdep +basedep +containersdep +hashablesetup-changed

Dependencies added: base, containers, hashable, stm, unordered-containers

Files

LICENSE +30/−0
README.md +1/−0
Setup.hs +2/−0
src/Control/Concurrent/STM/TMapChan.hs +139/−0
src/Control/Concurrent/STM/TMapChan/Hash.hs +131/−0
tmapchan.cabal +29/−0

+ LICENSE view

@@ -0,0 +1,30 @@+Copyright Author name here (c) 2017++All rights reserved.++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 Author name here nor the names of other+      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+OWNER 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,1 @@+# tmapchan

+ Setup.hs view

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain

+ src/Control/Concurrent/STM/TMapChan.hs view

@@ -0,0 +1,139 @@+module Control.Concurrent.STM.TMapChan+  ( TMapChan (..)+  , newTMapChan+  , insert, insertMany, insertFirst+  , lookup, tryLookup, lookupAll+  , observe, tryObserve, observeAll+  , delete, deleteAll+  , -- * Utils+    getTChan+  , cloneAt, cloneAll, cloneAllUniquely+  ) where++import Prelude hiding (lookup)+import Data.Map.Lazy (Map)+import qualified Data.Map.Lazy as Map+import qualified Data.Set as Set+import Control.Monad (void, forM_)+import Control.Concurrent.STM (STM)+import Control.Concurrent.STM.TChan (TChan, newTChan, writeTChan, readTChan, tryReadTChan, peekTChan, tryPeekTChan, unGetTChan, cloneTChan)+import Control.Concurrent.STM.TVar (TVar, newTVar, readTVar, writeTVar, modifyTVar)+++newtype TMapChan k a = TMapChan {runTMapChan :: TVar (Map k (TChan a))}++newTMapChan :: STM (TMapChan k a)+newTMapChan = TMapChan <$> newTVar Map.empty++insert :: Ord k => TMapChan k a -> k -> a -> STM ()+insert t k a = do+  c <- getTChan t k+  writeTChan c a++insertMany :: Ord k => TMapChan k a -> k -> [a] -> STM ()+insertMany t k as = forM_ as (insert t k)++-- | Inserts the element to the head of the stack, to be viewed next+insertFirst :: Ord k => TMapChan k a -> k -> a -> STM ()+insertFirst t k a = do+  c <- getTChan t k+  unGetTChan c a++-- | Blocks until there's a value available to remove from the mutlimap+lookup :: Ord k => TMapChan k a -> k -> STM a+lookup t k = do+  c <- getTChan t k+  readTChan c++tryLookup :: Ord k => TMapChan k a -> k -> STM (Maybe a)+tryLookup t k = do+  c <- getTChan t k+  tryReadTChan c++lookupAll :: Ord k => TMapChan k a -> k -> STM [a]+lookupAll t k = do+  mNext <- tryLookup t k+  case mNext of+    Nothing -> pure []+    Just next -> (next:) <$> lookupAll t k++-- | Blocks until there's a vale available to view, without removing it+observe :: Ord k => TMapChan k a -> k -> STM a+observe t k = do+  c <- getTChan t k+  peekTChan c++tryObserve :: Ord k => TMapChan k a -> k -> STM (Maybe a)+tryObserve t k = do+  c <- getTChan t k+  tryPeekTChan c++observeAll :: Ord k => TMapChan k a -> k -> STM [a]+observeAll t k = do+  mNext <- tryObserve t k+  case mNext of+    Nothing -> pure []+    Just next -> ((:) next) <$> observeAll t k+++-- | Deletes the /next/ element in the map, if it exists. Doesn't block.+delete :: Ord k => TMapChan k a -> k -> STM ()+delete t k = void (tryLookup t k)++-- | Clears the queue at the key+deleteAll :: Ord k => TMapChan k a -> k -> STM ()+deleteAll t k = void (lookupAll t k)+++-- * Utils+++-- | Creates a new one if it doesn't already exist+getTChan :: Ord k => TMapChan k a -> k -> STM (TChan a)+getTChan (TMapChan t) k = do+  xs <- readTVar t+  case Map.lookup k xs of+    Nothing -> do+      c' <- newTChan+      writeTVar t (Map.insert k c' xs)+      pure c'+    Just c' -> pure c'++cloneAt :: Ord k+        => TMapChan k a+        -> k -- ^ key to clone /from/+        -> k -- ^ key to clone /to/+        -> STM ()+cloneAt t@(TMapChan xs) kFrom kTo = do+  as <- observeAll t kFrom+  c <- newTChan+  forM_ as (writeTChan c)+  modifyTVar xs (Map.insert kTo c)++-- | Clones all the content for every key, by the key.+cloneAll :: Ord k => TMapChan k a -> k -> STM ()+cloneAll t@(TMapChan xs) k = do+  as <- readTVar xs+  c <- newTChan+  forM_ (Map.keysSet as) $ \k' -> do+    as' <- observeAll t k'+    forM_ as' (writeTChan c)+  modifyTVar xs (Map.insert k c)++-- | Clones all the content from every channel, and inserts the unique subset of them to `k`, in the order of their Ord instance+cloneAllUniquely :: ( Ord k+                    , Ord a+                    )+                  => TMapChan k a+                  -> k+                  -> STM ()+cloneAllUniquely t@(TMapChan xs) k = do+  as <- readTVar xs+  c <- newTChan+  as' <- newTVar Set.empty+  forM_ (Map.keysSet as) $ \k' -> do+    as'More <- Set.fromList <$> observeAll t k'+    modifyTVar as' (Set.union as'More)+  as'All <- Set.toAscList <$> readTVar as'+  forM_ as'All (writeTChan c)+  modifyTVar xs (Map.insert k c)

+ src/Control/Concurrent/STM/TMapChan/Hash.hs view

@@ -0,0 +1,131 @@+module Control.Concurrent.STM.TMapChan.Hash where++import Prelude hiding (lookup)+import Data.Hashable (Hashable)+import Data.HashMap.Lazy (HashMap)+import qualified Data.HashMap.Lazy as HashMap+import qualified Data.HashSet as HashSet+import Control.Monad (void, forM_)+import Control.Concurrent.STM (STM)+import Control.Concurrent.STM.TChan (TChan, newTChan, writeTChan, readTChan, tryReadTChan, peekTChan, tryPeekTChan, unGetTChan, cloneTChan)+import Control.Concurrent.STM.TVar (TVar, newTVar, readTVar, writeTVar, modifyTVar)+++newtype TMapChan k a = TMapChan {runTMapChan :: TVar (HashMap k (TChan a))}+++newTMapChan :: STM (TMapChan k a)+newTMapChan = TMapChan <$> newTVar HashMap.empty++insert :: (Eq k, Hashable k) => TMapChan k a -> k -> a -> STM ()+insert t k a = do+  c <- getTChan t k+  writeTChan c a++insertMany :: (Eq k, Hashable k) => TMapChan k a -> k -> [a] -> STM ()+insertMany t k as = forM_ as (insert t k)++-- | Inserts the element to the head of the stack, to be viewed next+insertFirst :: (Eq k, Hashable k) => TMapChan k a -> k -> a -> STM ()+insertFirst t k a = do+  c <- getTChan t k+  unGetTChan c a++-- | Blocks until there's a value available to remove from the mutlimap+lookup :: (Eq k, Hashable k) => TMapChan k a -> k -> STM a+lookup t k = do+  c <- getTChan t k+  readTChan c++tryLookup :: (Eq k, Hashable k) => TMapChan k a -> k -> STM (Maybe a)+tryLookup t k = do+  c <- getTChan t k+  tryReadTChan c++lookupAll :: (Eq k, Hashable k) => TMapChan k a -> k -> STM [a]+lookupAll t k = do+  mNext <- tryLookup t k+  case mNext of+    Nothing -> pure []+    Just next -> (next:) <$> lookupAll t k++-- | Blocks until there's a vale available to view, without removing it+observe :: (Eq k, Hashable k) => TMapChan k a -> k -> STM a+observe t k = do+  c <- getTChan t k+  peekTChan c++tryObserve :: (Eq k, Hashable k) => TMapChan k a -> k -> STM (Maybe a)+tryObserve t k = do+  c <- getTChan t k+  tryPeekTChan c++observeAll :: (Eq k, Hashable k) => TMapChan k a -> k -> STM [a]+observeAll t k = do+  mNext <- tryObserve t k+  case mNext of+    Nothing -> pure []+    Just next -> ((:) next) <$> observeAll t k+++-- | Deletes the /next/ element in the map, if it exists. Doesn't block.+delete :: (Eq k, Hashable k) => TMapChan k a -> k -> STM ()+delete t k = void (tryLookup t k)++-- | Clears the queue at the key+deleteAll :: (Eq k, Hashable k) => TMapChan k a -> k -> STM ()+deleteAll t k = void (lookupAll t k)+++-- * Utils+++-- | Creates a new one if it doesn't already exist+getTChan :: (Eq k, Hashable k) => TMapChan k a -> k -> STM (TChan a)+getTChan (TMapChan t) k = do+  xs <- readTVar t+  case HashMap.lookup k xs of+    Nothing -> do+      c' <- newTChan+      writeTVar t (HashMap.insert k c' xs)+      pure c'+    Just c' -> pure c'++cloneAt :: (Eq k, Hashable k)+        => TMapChan k a+        -> k -- ^ key to clone /from/+        -> k -- ^ key to clone /to/+        -> STM ()+cloneAt t@(TMapChan xs) kFrom kTo = do+  as <- observeAll t kFrom+  c <- newTChan+  forM_ as (writeTChan c)+  modifyTVar xs (HashMap.insert kTo c)++-- | Clones all the content for every key, by the key.+cloneAll :: (Eq k, Hashable k) => TMapChan k a -> k -> STM ()+cloneAll t@(TMapChan xs) k = do+  as <- readTVar xs+  c <- newTChan+  forM_ (HashMap.keys as) $ \k' -> do+    as' <- observeAll t k'+    forM_ as' (writeTChan c)+  modifyTVar xs (HashMap.insert k c)++-- | Clones all the content from every channel, and inserts the unique subset of them to `k`, in an unspecified order.+cloneAllUniquely :: ( Eq k, Hashable k+                    , Eq a, Hashable a+                    )+                  => TMapChan k a+                  -> k+                  -> STM ()+cloneAllUniquely t@(TMapChan xs) k = do+  as <- readTVar xs+  c <- newTChan+  as' <- newTVar HashSet.empty+  forM_ (HashMap.keys as) $ \k' -> do+    as'More <- HashSet.fromList <$> observeAll t k'+    modifyTVar as' (HashSet.union as'More)+  as'All <- HashSet.toList <$> readTVar as'+  forM_ as'All (writeTChan c)+  modifyTVar xs (HashMap.insert k c)

+ tmapchan.cabal view

@@ -0,0 +1,29 @@+name:                tmapchan+version:             0.0.0+synopsis:            A time-ordered multimap which consumes values as you lookup+-- description:+homepage:            https://github.com/athanclark/tmapchan#readme+license:             BSD3+license-file:        LICENSE+author:              Athan Clark+maintainer:          athan.clark@gmail.com+copyright:           2017 Athan Clark+category:            Web+build-type:          Simple+extra-source-files:  README.md+cabal-version:       >=1.10++library+  hs-source-dirs:      src+  exposed-modules:     Control.Concurrent.STM.TMapChan+                       Control.Concurrent.STM.TMapChan.Hash+  build-depends:       base >= 4.8 && < 5+                     , containers+                     , hashable+                     , stm+                     , unordered-containers+  default-language:    Haskell2010++source-repository head+  type:     git+  location: https://github.com/athanclark/tmapchan