concurrent-hashtable (empty) → 0.1.0
raw patch · 9 files changed
+960/−0 lines, 9 filesdep +QuickCheckdep +asyncdep +atomic-primopssetup-changed
Dependencies added: QuickCheck, async, atomic-primops, base, concurrent-hashtable, containers, criterion, dictionary-type, hashable, random, stm, unordered-containers, vector
Files
- ChangeLog.md +3/−0
- LICENSE +30/−0
- README.md +3/−0
- Setup.hs +2/−0
- benchmark/Main.hs +340/−0
- concurrent-hashtable.cabal +92/−0
- src/Data/HashTable.hs +30/−0
- src/Data/HashTable/Internal.hs +342/−0
- test/Spec.hs +118/−0
+ ChangeLog.md view
@@ -0,0 +1,3 @@+# Changelog for concurrent-hashtable++## Unreleased changes
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Peter Robinson (c) 2019++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 Peter Robinson 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,3 @@+# A thread-safe hash table for multi-cores++You can find benchmarks and more information about the internals of this package [here](https://lowerbound.io/blog/2019-10-24_concurrent_hash_table_performance.html).
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ benchmark/Main.hs view
@@ -0,0 +1,340 @@+----------------------------------------------------------------------+-- |+-- Module : Main+-- Copyright : (c) Peter Robinson+-- License : BSD3 (see the file LICENSE)+--+-- Maintainer : Peter Robinson <pwr@lowerbound.io>+-- Stability : provisional+-- Portability : non-portable+--+-- EXAMPLE USAGE:+--+-- > range=4 expon=6 stack bench +RTS -N8 -RTS --benchmark-arguments='--output results-32.html --csv results-32.csv'+--+-- The above will run all benchmarks with 10^6 requests split among 32 threads+-- and keys chosen from a range 10^4. You need to change the integer after "-N"+-- according to the number of cores that you have. Saves the results in files -- --- results-32.html and results-32.csv.+--+----------------------------------------------------------------------+{-# LANGUAGE BangPatterns, ScopedTypeVariables, FlexibleInstances, FlexibleContexts, MultiParamTypeClasses, AllowAmbiguousTypes #-}++module Main where++import Data.Dictionary+import Data.Dictionary.Request++import Control.Concurrent.Async+import Control.Concurrent.Chan+import Control.Concurrent.MVar+import Control.Concurrent+import Control.Exception+import Control.Monad+import System.Random+import Data.IORef+import Control.Concurrent.STM+import Control.Monad.STM+import Control.Monad.ST+import Criterion.Main+import System.Environment+import Data.Atomics+import Prelude hiding (lookup)++import qualified Data.HashTable as H+import qualified Data.IntMap.Strict as M+import qualified Data.HashMap.Strict as UO++-- Dictionary data type instances for all the test containers:++newtype MVarHashMap a = MVarHashMap (MVar (UO.HashMap Int a))+instance Dictionary (MVarHashMap Int) Int IO where+ runRequest (Lookup k) (MVarHashMap m) = do+ mymap <- readMVar m+ case UO.lookup k mymap of+ Nothing -> return False+ Just _ -> return True+ runRequest (Insert k a) (MVarHashMap m) =+ modifyMVar m $ \mymap -> do+ let mymap' = UO.insert k a mymap+ return (mymap',UO.null mymap')+ runRequest (Delete k) (MVarHashMap m) =+ modifyMVar m $ \mymap -> do+ let mymap' = UO.delete k mymap+ return (mymap',UO.null mymap')++newtype IORefPrimOps a = IORefPrimOps (IORef (UO.HashMap Int a))+instance Dictionary (IORefPrimOps Int) Int IO where+ runRequest (Lookup k) (IORefPrimOps m) = do+ mymap <- readIORef m+ case UO.lookup k mymap of+ Nothing -> return False+ Just _ -> return True+ runRequest (Insert k a) (IORefPrimOps m) =+ atomicModifyIORefCAS m $ \mymap ->+ let !mymap' = UO.insert k a mymap in+ (mymap',UO.null mymap')+ runRequest (Delete k) (IORefPrimOps m) =+ atomicModifyIORefCAS m $ \mymap ->+ let !mymap' = UO.delete k mymap in+ (mymap',UO.null mymap')++newtype IORefPrimOpsHashMap a = IORefPrimOpsHashMap (IORef (UO.HashMap Int a))+instance Dictionary (IORefPrimOpsHashMap Int) Int IO where+ runRequest (Lookup k) (IORefPrimOpsHashMap m) = do+ mymap <- readIORef m+ case UO.lookup k mymap of+ Nothing -> return False+ Just _ -> return True+ runRequest (Insert k a) (IORefPrimOpsHashMap m) =+ atomicModifyIORefCAS m $ \mymap ->+ let !mymap' = UO.insert k a mymap in+ (mymap',UO.null mymap')+ runRequest (Delete k) (IORefPrimOpsHashMap m) =+ atomicModifyIORefCAS m $ \mymap ->+ let !mymap' = UO.delete k mymap in+ (mymap',UO.null mymap')++newtype IORefHashMap a = IORefHashMap (IORef (UO.HashMap Int a))+instance Dictionary (IORefHashMap Int) Int IO where+ runRequest (Lookup k) (IORefHashMap m) = do+ mymap <- readIORef m+ case UO.lookup k mymap of+ Nothing -> return False+ Just _ -> return True+ runRequest (Insert k a) (IORefHashMap m) =+ atomicModifyIORef' m $ \mymap ->+ let !mymap' = UO.insert k a mymap in+ (mymap',UO.null mymap')+ runRequest (Delete k) (IORefHashMap m) =+ atomicModifyIORef' m $ \mymap ->+ let !mymap' = UO.delete k mymap in+ (mymap',UO.null mymap')++newtype TVarHashMap a = TVarHashMap (TVar (UO.HashMap Int a))+instance Dictionary (TVarHashMap Int) Int IO where+ runRequest (Lookup k) (TVarHashMap m) = do+ mymap <- readTVarIO m+ case UO.lookup k mymap of+ Nothing -> return False+ Just _ -> return True+ runRequest (Insert k a) (TVarHashMap m) =+ atomically $ stateTVar m $ \mymap ->+ let !mymap' = UO.insert k a mymap in+ (UO.null mymap',mymap')+ runRequest (Delete k) (TVarHashMap m) =+ atomically $ stateTVar m $ \mymap ->+ let !mymap' = UO.delete k mymap in+ (UO.null mymap',mymap')++instance Dictionary (H.HashTable Int Int) Int IO where+ runRequest (Lookup k) s = do+ r <- H.lookup s k+ case r of+ Nothing -> return False+ Just _ -> return True+ runRequest (Insert k a) s = H.insert s k a+ runRequest (Delete k) s = H.delete s k+++main = do+ -- use environment variables until we figure out how to avoid interfering with criterions cmdargs handling...+ mThreads <- lookupEnv "threads"+ mRange <- lookupEnv "range"+ mExpon <- lookupEnv "expon"+ mThreshold <- lookupEnv "threshold"+ mResizers <- lookupEnv "numresizers"+ let expon = case mExpon of Just e -> read e+ Nothing -> 5 -- ^ total number of requests+ let numRequests = 10^expon+ let range = (1,10^(case mRange of Just r -> read r+ Nothing -> expon `div` 2)) -- ^ range for keys+ let numThreads = case mThreads of Just t -> read t+ Nothing -> 4 -- ^ the number of requests lists we split the list into+ let threshold = case mThreshold of Just t -> read t+ Nothing -> 0.75 -- ^ threshold on load for resizing+ let numResizers = case mResizers of Just t -> read t+ Nothing -> 8 -- ^ number of worker threads for resizing+ print "Parameters: "+ putStrLn $ "number of spawned threads: " ++ show numThreads+ putStrLn $ "number of requests: " ++ show numRequests+ putStrLn $ "key range: " ++ show range+ putStrLn $ "load threshold: " ++ show threshold+ runBench numThreads numRequests range threshold++runBench :: Int -> Int -> (Int,Int) -> Float -> IO ()+runBench numThreads numRequests range threshold = do+ let genTests = generateTests (numRequests `div` numThreads)+ let genTestsForThreads i = generateTests (numRequests `div` i)++ let mkCHT :: IO (H.HashTable Int Int)+ mkCHT = H.newWithDefaults 10++ let mkIOPrimOpsMap :: IO (IORefPrimOps Int)+ mkIOPrimOpsMap =+ IORefPrimOps <$> newIORef (UO.empty :: UO.HashMap Int Int)++ let mkIOPrimOpsHashMap :: IO (IORefPrimOpsHashMap Int)+ mkIOPrimOpsHashMap =+ IORefPrimOpsHashMap <$> newIORef (UO.empty :: UO.HashMap Int Int)++ let mkIORefMap :: IO (IORefHashMap Int)+ mkIORefMap =+ IORefHashMap <$> newIORef (UO.empty :: UO.HashMap Int Int)++ let mkTVarMap :: IO (TVarHashMap Int)+ mkTVarMap =+ TVarHashMap <$> newTVarIO (UO.empty :: UO.HashMap Int Int)++ let mkMVarMap :: IO (MVarHashMap Int)+ mkMVarMap =+ MVarHashMap <$> newMVar (UO.empty :: UO.HashMap Int Int)++ numProcs <- getNumCapabilities+ print ("number of available cores: ",numProcs)+ threadDelay 3000000++ let evaluate tests ds = do+ res <- forConcurrently tests+ (mapM $ \r -> runRequest r ds)+ print $ length $ filter (==True) $ concat res+ return $! filter (==True) $ concat res+++ -- Generate set of tests for different workloads+ !tests_90_5_5 <- replicateM numThreads $ genTests (0.9,0.05) range+ !tests_80_10_10 <- replicateM numThreads $ genTests (0.8,0.1) range+ !tests_70_15_15 <- replicateM numThreads $ genTests (0.7,0.15) range+ !tests_60_20_20 <- replicateM numThreads $ genTests (0.6,0.2) range+ !tests_50_25_25 <- replicateM numThreads $ genTests (0.5,0.25) range+ !tests_40_30_30 <- replicateM numThreads $ genTests (0.4,0.30) range+ -- !tests_30_35_35 <- replicateM numThreads $ genTests (0.3,0.35) range+ !tests_20_40_40 <- replicateM numThreads $ genTests (0.2,0.40) range+ -- !tests_10_45_45 <- replicateM numThreads $ genTests (0.1,0.45) range+ !tests_0_50_50 <- replicateM numThreads $ genTests (0,0.5) range+ -- let !all_95_25_25 = concat tests_95_25_25+ let !all_90_5_5 = concat tests_90_5_5+ let !all_80_10_10 = concat tests_80_10_10+ let !all_70_15_15 = concat tests_70_15_15+ let !all_60_20_20 = concat tests_60_20_20+ let !all_50_25_25 = concat tests_50_25_25+ let !all_40_30_30 = concat tests_40_30_30+ -- let !all_30_35_35 = concat tests_30_35_35+ let !all_20_40_40 = concat tests_20_40_40+ -- let !all_10_45_45 = concat tests_10_45_45+ let !all_0_50_50 = concat tests_0_50_50+ print "--------------------------------------------"+ print "90 5 5:"+ print $ "# of lookup requests: "++ show (length $ filter isLookup all_90_5_5)+ print $ "# of insert requests: "++ show (length $ filter isInsert all_90_5_5)+ print $ "# of delete requests: "++ show (length $ filter isDelete all_90_5_5)+ print "--------------------------------------------"+ print "80 10 10:"+ print $ "# of lookup requests: "++ show (length $ filter isLookup all_80_10_10)+ print $ "# of insert requests: "++ show (length $ filter isInsert all_80_10_10)+ print $ "# of delete requests: "++ show (length $ filter isDelete all_80_10_10)+ print "--------------------------------------------"+ print "70 15 15:"+ print $ "# of lookup requests: "++ show (length $ filter isLookup all_70_15_15)+ print $ "# of insert requests: "++ show (length $ filter isInsert all_70_15_15)+ print $ "# of delete requests: "++ show (length $ filter isDelete all_70_15_15)+ print "--------------------------------------------"+ print "60 20 20:"+ print $ "# of lookup requests: "++ show (length $ filter isLookup all_60_20_20)+ print $ "# of insert requests: "++ show (length $ filter isInsert all_60_20_20)+ print $ "# of delete requests: "++ show (length $ filter isDelete all_60_20_20)+ print "--------------------------------------------"+ print "50 25 25:"+ print $ "# of lookup requests: "++ show (length $ filter isLookup all_50_25_25)+ print $ "# of insert requests: "++ show (length $ filter isInsert all_50_25_25)+ print $ "# of delete requests: "++ show (length $ filter isDelete all_50_25_25)+ print "--------------------------------------------"+ print "40 30 30:"+ print $ "# of lookup requests: "++ show (length $ filter isLookup all_40_30_30)+ print $ "# of insert requests: "++ show (length $ filter isInsert all_40_30_30)+ print $ "# of delete requests: "++ show (length $ filter isDelete all_40_30_30)+ print "--------------------------------------------"+ print "20 40 40:"+ print $ "# of lookup requests: "++ show (length $ filter isLookup all_20_40_40)+ print $ "# of insert requests: "++ show (length $ filter isInsert all_20_40_40)+ print $ "# of delete requests: "++ show (length $ filter isDelete all_20_40_40)+ print "--------------------------------------------"+ print "0 50 50:"+ print $ "# of lookup requests: "++ show (length $ filter isLookup all_0_50_50)+ print $ "# of insert requests: "++ show (length $ filter isInsert all_0_50_50)+ print $ "# of delete requests: "++ show (length $ filter isDelete all_0_50_50)+ print "--------------------------------------------"+ defaultMain+ -- run benchmark for different workloads:+ [ bgroup "0% Lookups; 50% Inserts; 50% Deletes"+ [+ bench "IORef atomicModifyIORef HashMap" $ nfIO (mkIORefMap >>= evaluate tests_0_50_50)+ , bench "IORef atomic-primops IntMap" $ nfIO (mkIOPrimOpsMap >>= evaluate tests_0_50_50)+ , bench "IORef atomic-primops HashMap" $ nfIO (mkIOPrimOpsHashMap >>= evaluate tests_0_50_50)+ , bench "TVar HashMap" $ nfIO (mkTVarMap >>= evaluate tests_0_50_50)+ , bench "MVar HashMap" $ nfIO (mkMVarMap >>= evaluate tests_0_50_50)+ , bench "Concurrent HashTable" $ nfIO (mkCHT >>= evaluate tests_0_50_50)+ ]+ , bgroup "20% Lookups; 40% Inserts; 40% Deletes"+ [+ bench "IORef atomicModifyIORef HashMap" $ nfIO (mkIORefMap >>= evaluate tests_20_40_40)+ , bench "IORef atomic-primops IntMap" $ nfIO (mkIOPrimOpsMap >>= evaluate tests_20_40_40)+ , bench "IORef atomic-primops HashMap" $ nfIO (mkIOPrimOpsHashMap >>= evaluate tests_20_40_40)+ , bench "TVar HashMap" $ nfIO (mkTVarMap >>= evaluate tests_20_40_40)+ , bench "MVar HashMap" $ nfIO (mkMVarMap >>= evaluate tests_20_40_40)+ , bench "Concurrent HashTable" $ nfIO (mkCHT >>= evaluate tests_20_40_40)+ ]+ , bgroup "40% Lookups; 30% Inserts; 30% Deletes"+ [+ bench "IORef atomicModifyIORef HashMap" $ nfIO (mkIORefMap >>= evaluate tests_40_30_30)+ , bench "IORef atomic-primops IntMap" $ nfIO (mkIOPrimOpsMap >>= evaluate tests_40_30_30)+ , bench "IORef atomic-primops HashMap" $ nfIO (mkIOPrimOpsHashMap >>= evaluate tests_40_30_30)+ , bench "TVar HashMap" $ nfIO (mkTVarMap >>= evaluate tests_40_30_30)+ , bench "MVar HashMap" $ nfIO (mkMVarMap >>= evaluate tests_40_30_30)+ , bench "Concurrent HashTable" $ nfIO (mkCHT >>= evaluate tests_40_30_30)+ ]+ , bgroup "50% Lookups; 25% Inserts; 25% Deletes"+ [+ bench "IORef atomicModifyIORef HashMap" $ nfIO (mkIORefMap >>= evaluate tests_50_25_25)+ , bench "IORef atomic-primops IntMap" $ nfIO (mkIOPrimOpsMap >>= evaluate tests_50_25_25)+ , bench "IORef atomic-primops HashMap" $ nfIO (mkIOPrimOpsHashMap >>= evaluate tests_50_25_25)+ , bench "TVar HashMap" $ nfIO (mkTVarMap >>= evaluate tests_50_25_25)+ , bench "MVar HashMap" $ nfIO (mkMVarMap >>= evaluate tests_50_25_25)+ , bench "Concurrent HashTable" $ nfIO (mkCHT >>= evaluate tests_50_25_25)+ ]+ , bgroup "60% Lookups; 20% Inserts; 20% Deletes"+ [+ bench "IORef atomicModifyIORef HashMap" $ nfIO (mkIORefMap >>= evaluate tests_60_20_20)+ , bench "IORef atomic-primops IntMap" $ nfIO (mkIOPrimOpsMap >>= evaluate tests_60_20_20)+ , bench "IORef atomic-primops HashMap" $ nfIO (mkIOPrimOpsHashMap >>= evaluate tests_60_20_20)+ , bench "TVar HashMap" $ nfIO (mkTVarMap >>= evaluate tests_60_20_20)+ , bench "MVar HashMap" $ nfIO (mkMVarMap >>= evaluate tests_60_20_20)+ , bench "Concurrent HashTable" $ nfIO (mkCHT >>= evaluate tests_60_20_20)+ ]+ , bgroup "70% Lookups; 15% Inserts; 15% Deletes"+ [+ bench "IORef atomicModifyIORef HashMap" $ nfIO (mkIORefMap >>= evaluate tests_70_15_15)+ , bench "IORef atomic-primops IntMap" $ nfIO (mkIOPrimOpsMap >>= evaluate tests_70_15_15)+ , bench "IORef atomic-primops HashMap" $ nfIO (mkIOPrimOpsHashMap >>= evaluate tests_70_15_15)+ , bench "TVar HashMap" $ nfIO (mkTVarMap >>= evaluate tests_70_15_15)+ , bench "MVar HashMap" $ nfIO (mkMVarMap >>= evaluate tests_70_15_15)+ , bench "Concurrent HashTable" $ nfIO (mkCHT >>= evaluate tests_70_15_15)+ ]+ , bgroup "80% Lookups; 10% Inserts; 10% Deletes"+ [+ bench "IORef atomicModifyIORef HashMap" $ nfIO (mkIORefMap >>= evaluate tests_80_10_10)+ , bench "IORef atomic-primops IntMap" $ nfIO (mkIOPrimOpsMap >>= evaluate tests_80_10_10)+ , bench "IORef atomic-primops HashMap" $ nfIO (mkIOPrimOpsHashMap >>= evaluate tests_80_10_10)+ , bench "TVar HashMap" $ nfIO (mkTVarMap >>= evaluate tests_80_10_10)+ , bench "MVar HashMap" $ nfIO (mkMVarMap >>= evaluate tests_80_10_10)+ , bench "Concurrent HashTable" $ nfIO (mkCHT >>= evaluate tests_80_10_10)+ ]+ , bgroup "90% Lookups; 5% Inserts; 5% Deletes"+ [+ bench "IORef atomicModifyIORef HashMap" $ nfIO (mkIORefMap >>= evaluate tests_90_5_5)+ , bench "IORef atomic-primops IntMap" $ nfIO (mkIOPrimOpsMap >>= evaluate tests_90_5_5)+ , bench "IORef atomic-primops HashMap" $ nfIO (mkIOPrimOpsHashMap >>= evaluate tests_90_5_5)+ , bench "TVar HashMap" $ nfIO (mkTVarMap >>= evaluate tests_90_5_5)+ , bench "MVar HashMap" $ nfIO (mkMVarMap >>= evaluate tests_90_5_5)+ , bench "Concurrent HashTable" $ nfIO (mkCHT >>= evaluate tests_90_5_5)+ ]+ ]
+ concurrent-hashtable.cabal view
@@ -0,0 +1,92 @@+cabal-version: 1.12++-- This file has been generated from package.yaml by hpack version 0.31.2.+--+-- see: https://github.com/sol/hpack+--+-- hash: 52e913bb36c726771c6a1d52ff86abb22578db4f327354ec03258c126f660c66++name: concurrent-hashtable+version: 0.1.0+synopsis: Thread-safe hash tables for multi-cores!+description: Please see the README on GitHub at <https://github.com/pwrobinson/concurrent-hashtable#readme>+category: Concurrency+homepage: https://github.com/pwrobinson/concurrent-hashtable#readme+bug-reports: https://github.com/pwrobinson/concurrent-hashtable/issues+author: Peter Robinson+maintainer: pwr@lowerbound.io+copyright: 2019 Peter Robinson+license: BSD3+license-file: LICENSE+build-type: Simple+extra-source-files:+ README.md+ ChangeLog.md++source-repository head+ type: git+ location: https://github.com/pwrobinson/concurrent-hashtable++library+ exposed-modules:+ Data.HashTable+ Data.HashTable.Internal+ other-modules:+ Paths_concurrent_hashtable+ hs-source-dirs:+ src+ build-depends:+ async >=2.2.2 && <3+ , atomic-primops >=0.8.3 && <2+ , base >=4.7 && <5+ , containers >=0.6.0.1 && <1+ , hashable >=1.2.7.0 && <2+ , random >=1.1 && <2+ , stm >=2.4.5.1 && <3+ , vector >=0.12.0.3 && <1+ default-language: Haskell2010++test-suite concurrent-hashtable-test+ type: exitcode-stdio-1.0+ main-is: Spec.hs+ other-modules:+ Paths_concurrent_hashtable+ hs-source-dirs:+ test+ ghc-options: -threaded -rtsopts -with-rtsopts=-N+ build-depends:+ QuickCheck >=2.13.2 && <3+ , async >=2.2.2 && <3+ , atomic-primops >=0.8.3 && <2+ , base >=4.7 && <5+ , concurrent-hashtable+ , containers >=0.6.0.1 && <1+ , dictionary-type+ , hashable >=1.2.7.0 && <2+ , random >=1.1 && <2+ , stm >=2.4.5.1 && <3+ , vector >=0.12.0.3 && <1+ default-language: Haskell2010++benchmark mainbench+ type: exitcode-stdio-1.0+ main-is: Main.hs+ other-modules:+ Paths_concurrent_hashtable+ hs-source-dirs:+ benchmark+ ghc-options: -threaded -rtsopts -with-rtsopts=-N -eventlog -O2+ build-depends:+ async >=2.2.2 && <3+ , atomic-primops >=0.8.3 && <2+ , base >=4.7 && <5+ , concurrent-hashtable+ , containers >=0.6.0.1 && <1+ , criterion >=1.5.6.0 && <2+ , dictionary-type+ , hashable >=1.2.7.0 && <2+ , random >=1.1 && <2+ , stm >=2.4.5.1 && <3+ , unordered-containers >=0.2.10.0 && <1+ , vector >=0.12.0.3 && <1+ default-language: Haskell2010
+ src/Data/HashTable.hs view
@@ -0,0 +1,30 @@+----------------------------------------------------------------------+-- |+-- Module : Data.HashTable+-- Copyright : (c) Peter Robinson+-- License : BSD3 (see the file LICENSE)+--+-- Maintainer : Peter Robinson <pwr@lowerbound.io>+-- Stability : provisional+-- Portability : non-portable (requires concurrency, stm)+--+-- You can find benchmarks and more information about the internals of this package here: <https://lowerbound.io/blog/2019-10-24_concurrent_hash_table_performance.html>+----------------------------------------------------------------------++module Data.HashTable(+ HashTable,+ Chain,+ _itemsTV,+ new,+ newWithDefaults,+ mkDefaultConfig,+ Config(..),+ lookup,+ insert,+ insertIfNotExists,+ delete,+ getAssocs+ )+where+import Data.HashTable.Internal+import Prelude hiding (lookup)
+ src/Data/HashTable/Internal.hs view
@@ -0,0 +1,342 @@+----------------------------------------------------------------------+-- |+-- Module : Data.HashTable.Internal+-- Copyright : (c) Peter Robinson+-- License : BSD3 (see the file LICENSE)+--+-- Maintainer : Peter Robinson <pwr@lowerbound.io>+-- Stability : provisional+-- Portability : non-portable (requires concurrency, stm)+--+-- You can find benchmarks and more information about the internals of this package here: <https://lowerbound.io/blog/2019-10-24_concurrent_hash_table_performance.html>+----------------------------------------------------------------------+{-# LANGUAGE MultiParamTypeClasses, ScopedTypeVariables #-}+module Data.HashTable.Internal+where++import Control.Concurrent.STM+import Control.Concurrent+import Control.Concurrent.Async+import Data.IORef+import Data.Atomics+import Control.Exception+import Control.Monad+import Data.Hashable+import System.Random+import System.IO.Unsafe+import Data.Maybe+import qualified Data.List as L+import Data.Vector(Vector,(!))+import qualified Data.Vector as V+import Prelude hiding (lookup,readList)++-- Used internally.+data MigrationStatus = NotStarted | Ongoing | Finished+ deriving(Show,Eq)++-- | Used for chain-hashing.+data Chain k v = Chain+ { _itemsTV :: TVar [(k,v)] -- ^ stores the+ , _migrationStatusTV :: TVar MigrationStatus -- ^ used internally for resizing+ }+ deriving (Eq)++newChainIO =+ Chain <$> newTVarIO []+ <*> newTVarIO NotStarted++-- | A thread-safe hash table that supports dynamic resizing.+data HashTable k v = HashTable+ { _chainsVecTV :: TVar (Vector (Chain k v)) -- ^ vector of linked lists+ , _totalLoad :: IORef Int+ , _config :: Config k+ }++-- | Configuration options that may affect the performance of the hash table+data Config k = Config+ { _scaleFactor :: Float -- ^ scale factor for resizing+ , _threshold :: Float -- ^ load threshold for initiating resizing.+ , _numResizeWorkers :: Int -- ^ maximum number of worker threads used during resizing+ , _hashFunc :: k -> Int+ }++instance Show (Config k) where+ show cfg = "Config " ++ show (_scaleFactor cfg)+ ++ show (_threshold cfg)+ ++ show (_numResizeWorkers cfg)++-- | Default configuration: scale factor = 2.0; resizing threshold = 0.75;+-- number of worker threads for resizing = 'getNumCapabilities';+-- hash function = use 'hashWithSalt' with a random salt+mkDefaultConfig :: Hashable k => IO (Config k)+mkDefaultConfig = do+ numCPUs <- getNumCapabilities+ salt <- randomIO :: IO Int+ return $ Config+ { _scaleFactor = 2.0+ , _threshold = 0.75+ , _numResizeWorkers = numCPUs+ , _hashFunc = hashWithSalt salt+ }+++-- | Creates a new hash table with an initial size. See 'newWithDefaults' for more details.+-- You probably either want to use 'newWithDefaults' instead or+-- something like this:+-- > mkDefaultConfig { _field = myValue } >>= new 10+new :: (Eq k) => Int -- ^ Initial size of the hash table+ -> Config k -> IO (HashTable k v)+new size config = do+ chainsVec <- V.replicateM size newChainIO+ HashTable <$> newTVarIO chainsVec+ <*> newIORef 0+ <*> return config++-- | Creates a new hash table with the given initial vector size, scale factor 2.0, a resizing load threshold of 0.75, and we use as many threads for resizing as we have cores available. This will use a hash function with a (single) random salt, so if you need security, you MUST supply your own hash function. To be replaced by universal hashing in future versions.+newWithDefaults :: (Eq k,Hashable k) => Int -- ^ Initial size of the hash table+ -> IO (HashTable k v)+newWithDefaults size = mkDefaultConfig >>= new size++-- | Returns the size of the vector representing the hash table.+{-# INLINABLE readSizeIO #-}+readSizeIO :: HashTable k v -> IO Int+readSizeIO ht = do+ chainsVec <- readTVarIO $ _chainsVecTV ht+ return $ V.length chainsVec++-- | Returns the size of the vector representing the hash table.+{-# INLINABLE readSize #-}+readSize :: HashTable k v -> STM Int+readSize ht = do+ chainsVec <- readTVar $ _chainsVecTV ht+ return $ V.length chainsVec+-- | Resizes the hash table by scaling it according to the _scaleFactor in the configuration.++resize :: (Eq k)+ => HashTable k v -> IO ()+resize ht = do+ chainsVec <- readTVarIO $ _chainsVecTV ht+ let size1 = V.length chainsVec+ alreadyResizing <- do+ hasStarted <- atomically $ do+ migrating <- readTVar (_migrationStatusTV $ chainsVec ! 0)+ if migrating `elem` [Ongoing,Finished] then+ return True+ else do+ writeTVar (_migrationStatusTV $ chainsVec ! 0) Ongoing+ return False+ size2 <- readSizeIO ht+ return (hasStarted || (size1 /= size2))+ unless alreadyResizing $ do+ let lists = V.toList chainsVec+ let oldSize = V.length chainsVec+ let numWorkers = _numResizeWorkers $ _config ht+ let numSlices = min numWorkers (oldSize `div` numWorkers)+ let sliceLength = oldSize `div` numSlices+ let restLength = oldSize - ((numSlices-1)*sliceLength)+ let vecSlices = [ V.unsafeSlice+ (i*sliceLength)+ (if i==numSlices-1 then restLength+ else sliceLength)+ chainsVec+ | i <- [0..numSlices-1]]+ let (scale :: Float) = _scaleFactor (_config ht)+ let (newSize::Int) = round $ (fromIntegral oldSize) * scale+ newVec <- V.replicateM newSize newChainIO+ forConcurrently_ vecSlices (V.mapM_ (migrate newVec newSize))+ `catch` (\(e::AssertionFailed) -> do+ debug "ERROR in resize; this should never happen..."+ throw e)+ debug "finished copying over nodes..."+ -- replace old vector with new one:+ atomically $ writeTVar (_chainsVecTV ht) newVec+ debug "replaced old array with new one..."+ -- now unblock threads that are blocking on the "migrate" field of any of the old chains:+ forConcurrently_ vecSlices $+ V.mapM_ (\chain ->+ atomically $ writeTVar (_migrationStatusTV chain) Finished)+ debug "woke up blocked threads..."+ where+ migrate newVec newSize chain = do+ let idx = 0+ -- debug ("starting to copy nodes of list # ",idx)+ atomically $ writeTVar (_migrationStatusTV chain) Ongoing+ -- debug ("updated list status # ",idx)+ listOfNodes <- readTVarIO (_itemsTV chain)+ -- debug ("done copying nodes of list # ",idx)+ sequence_ [ do let newIndex = (_hashFunc (_config ht) k) `mod` newSize+ let newChain = newVec ! newIndex+ newList <- readTVarIO (_itemsTV newChain)+ atomically $+ writeTVar (_itemsTV newChain) ((k,v):newList)+ | (k,v) <- listOfNodes ]+ -- debug ("finished migrate for list ",idx)++-- | Lookup the value for a key in the hash table.+{-# INLINABLE lookup #-}+lookup :: (Eq k) => HashTable k v -> k -> IO (Maybe v)+lookup htable k = do+ chain <- readChainForKeyIO htable k+ list <- readTVarIO (_itemsTV chain)+ return $ L.lookup k list+++-- | Used internally. An action to be executed atomically for the given chain.+type STMAction k v a = TVar [(k,v)] -> STM (Maybe a)++-- | Used internally by 'insert', 'insertIfNotExists', 'delete', 'update'+genericModify :: (Eq k)+ => HashTable k v+ -> k -- ^ key+ -> STMAction k v a+ -> IO a+genericModify htable k stmAction = do+ chain <- readChainForKeyIO htable k+ result <- atomically $ do+ migrationStatus <- readTVar (_migrationStatusTV chain)+ case migrationStatus of+ Ongoing -> retry -- block until resizing is done+ Finished -> return Nothing -- resizing already done; try again+ NotStarted -> -- no resizing happening at the moment+ stmAction (_itemsTV chain)+ case result of+ Nothing -> genericModify htable k stmAction+ Just v -> return v+++insert :: (Eq k)+ => HashTable k v+ -> k -- ^ key+ -> v -- ^ value+ -> IO Bool+insert htable k v = do+ result <- genericModify htable k $ \tvar -> do+ list <- readTVar tvar+ case L.lookup k list of+ Nothing -> do+ writeTVar tvar ((k,v):list)+ return $ Just True+ Just _ -> do -- entry was already there, so we overwrite it+ writeTVar tvar ((k,v) : deleteFirstKey k list)+ return $ Just False+ if result then do+ atomicallyChangeLoad htable 1+ return True+ else+ return False++++-- | Inserts a key and value pair into the hash table only if the key does not exist yet. Returns 'True' if the insertion was successful, i.e., the hash table did not contain this key before. To get the same behaviour as 'Data.Map.insert', use 'insert'. If you want to update an entry only if it exists, use 'update'.+insertIfNotExists :: (Eq k)+ => HashTable k v+ -> k -- ^ key+ -> v -- ^ value+ -> IO Bool+insertIfNotExists htable k v = do+ result <- genericModify htable k $ \tvar -> do+ list <- readTVar tvar+ case L.lookup k list of+ Nothing -> do+ writeTVar tvar ((k,v):list)+ return $ Just True+ Just _ -> return $ Just False+ if result then do+ atomicallyChangeLoad htable 1+ return True+ else+ return False++-- Deletes the entry for the given key from the hash table. Returns `True` if and only if an entry was deleted from the table.+delete :: (Eq k)+ => HashTable k v+ -> k -- ^ key of entry that will be deleted+ -> IO Bool+delete htable k = do+ result <- genericModify htable k $ \tvar -> do+ list <- readTVar tvar+ case L.lookup k list of+ Nothing ->+ return $ Just False+ Just _ -> do+ writeTVar tvar $ deleteFirstKey k list+ return $ Just True+ if result then do+ atomicallyChangeLoad htable (-1)+ return True+ else+ return False++-- | Atomically increment/decrement the table load value by adding the provided integer value to the current value.+atomicallyChangeLoad :: (Eq k)+ => HashTable k v+ -> Int -- ^ increment/decrement value+ -> IO ()+atomicallyChangeLoad htable incr = do+ totalLoad <- atomicModifyIORefCAS (_totalLoad htable) $+ \l -> (l+incr,l+incr)+ size <- readSizeIO htable+ when ((fromIntegral totalLoad / fromIntegral size)+ >= _threshold (_config htable)) $ do+ chain0 <- readChainForIndexIO htable 0+ migrationStatus <- readTVarIO (_migrationStatusTV chain0)+ when (migrationStatus == NotStarted) $+ void $ forkIO (resize htable)++-- Atomically retrieves list of key-value pairs. If there is a lot of contention going on, this may be very inefficient.+getAssocs :: (Eq k)+ => HashTable k v -> STM [(k,v)]+getAssocs htable = do+ chainsVec <- readTVar $ _chainsVecTV htable+ let len = V.length chainsVec+ let getItemsForChain k = do+ chain <- readChainForIndex htable k+ readTVar (_itemsTV chain)+ msum <$> mapM getItemsForChain [ k | k <- [0..len-1]]++{-# INLINABLE deleteFirstKey #-}+deleteFirstKey :: Eq a => a -> [(a,b)] -> [(a,b)]+deleteFirstKey _ [] = []+deleteFirstKey x (y:ys) = if x == fst y then ys else y : deleteFirstKey x ys++{-# INLINABLE readChainForKeyIO #-}+-- | Get the chain for the given key.+readChainForKeyIO :: HashTable k v -> k -> IO (Chain k v)+readChainForKeyIO htable k = do+ size1 <- readSizeIO htable+ let index = (_hashFunc (_config htable) k) `mod` size1+ chain <- readChainForIndexIO htable index+ size2 <- readSizeIO htable+ if size1 /= size2 then+ readChainForKeyIO htable k+ else+ return chain++{-# INLINABLE readChainForKey #-}+-- | Get the chain for the given key.+readChainForKey :: HashTable k v -> k -> STM (Chain k v)+readChainForKey htable k = do+ size1 <- readSize htable+ let index = (_hashFunc (_config htable) k) `mod` size1+ readChainForIndex htable index++-- | Get the chain for the given index (warning: bounds are not checked)+{-# INLINABLE readChainForIndexIO #-}+readChainForIndexIO :: HashTable k v -> Int -> IO (Chain k v)+readChainForIndexIO htable idx = do+ chainsVec <- readTVarIO $ _chainsVecTV htable+ return $ chainsVec ! idx++-- | Get the chain for the given index (warning: bounds are not checked)+{-# INLINABLE readChainForIndex #-}+readChainForIndex :: HashTable k v -> Int -> STM (Chain k v)+readChainForIndex htable idx = do+ chainsVec <- readTVar $ _chainsVecTV htable+ return $ chainsVec ! idx++{-# INLINABLE debug #-}+debug :: Show a => a -> IO ()+debug a = return () {- do+ -- tid <- myThreadId+ -- print a+ -- appendFile ("thread" ++ show tid) (show a) -}
+ test/Spec.hs view
@@ -0,0 +1,118 @@+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE ScopedTypeVariables #-}++import Control.Monad+import Data.IORef+import Data.Maybe+import qualified Data.Map.Strict as M+import Test.QuickCheck as QC+import Test.QuickCheck.Monadic as QC+import Prelude hiding (lookup,elems)+import System.IO+import qualified Data.Set as Set++import qualified Data.HashTable as H+import Control.Concurrent.STM+import Data.Hashable+import Data.Dictionary+import Data.Dictionary.Request+import qualified Data.Vector as V++hasDuplicates :: (Ord a) => [a] -> Bool+hasDuplicates list = length list /= length set+ where set = Set.fromList list++-- | Restrict the keys to be ints from small positive range+newtype BoundedInt = BoundedInt Int+ deriving (Eq,Ord)++instance Show BoundedInt where+ show (BoundedInt i) = show i++instance Hashable BoundedInt where+ hashWithSalt a (BoundedInt i) = hashWithSalt a i++instance QC.Arbitrary BoundedInt where+ arbitrary = BoundedInt <$> QC.choose (1,100)++type TestMap = M.Map BoundedInt BoundedInt+instance Dictionary (IORef TestMap) BoundedInt IO where+ runRequest (Lookup k) m = do+ mymap <- readIORef m+ case M.lookup k mymap of+ Nothing -> return False+ Just _ -> return True+ runRequest (Insert k a) m =+ atomicModifyIORef' m $ \mymap ->+ let s1 = M.size mymap in+ let mymap' = M.insert k a mymap in+ let s2 = M.size mymap' in+ (mymap',s1/=s2)+ runRequest (Update k a) m =+ atomicModifyIORef' m $ \mymap ->+ case M.lookup k mymap of+ Nothing -> (mymap,False)+ Just _ ->+ let s1 = M.size mymap in+ let mymap' = M.insert k a mymap in+ let s2 = M.size mymap' in+ (mymap',s1/=s2)+ runRequest (Delete k) m =+ atomicModifyIORef' m $ \mymap ->+ let s1 = M.size mymap in+ let mymap' = M.delete k mymap in+ let s2 = M.size mymap' in+ (mymap',s1/=s2)+++type TestChainHashTable = H.HashTable BoundedInt BoundedInt+instance Dictionary TestChainHashTable BoundedInt IO where+ runRequest (Lookup k) s = do+ r <- H.lookup s k+ case r of+ Nothing -> return False+ Just _ -> return True+ runRequest (Insert k a) s = H.insert s k a+ runRequest (Update k a) s = H.insert s k a+ runRequest (Delete k) s = H.delete s k++instance (QC.Arbitrary k) => QC.Arbitrary (Request k) where+ arbitrary = QC.oneof+ [ Insert <$> QC.arbitrary <*> QC.arbitrary+ , Lookup <$> QC.arbitrary+ , Delete <$> QC.arbitrary+ ]++prop :: IORef TestMap -> TestChainHashTable -> Property+prop ioref chainTable = monadicIO $ do+ (r :: Request BoundedInt) <- pick arbitrary+ run $ appendFile "request_sequence.txt" (show r ++ "\n")+ run $ runRequest r ioref+ run $ runRequest r chainTable++ mapAssocs <- run ( M.assocs <$> readIORef ioref)+ -- test if all entries in the Map are also in the Hash Table:+ resChain <- run $ sequence+ [ do r <- H.lookup chainTable k+ return $ (isJust r) && (fromJust r == a)+ | (k,a) <- mapAssocs ]+ list2 <- run $ atomically $ H.getAssocs chainTable+ -- test if there aren't any duplicates in the Hash Table:+ let res2 = (not . hasDuplicates) list2+ mymap <- run $ readIORef ioref+ -- test if all entries in the Hash Table are also in the Map:+ let res3 = [ isJust $ M.lookup (fst k) mymap | k <- list2 ]+ assert $ and resChain && res2 && and res3+++main :: IO ()+main = do+ ioref <- newIORef (M.empty :: M.Map BoundedInt BoundedInt)+ (ctable :: H.HashTable BoundedInt BoundedInt) <- H.newWithDefaults 10+ writeFile "request_sequence.txt" ""+ quickCheckWith stdArgs{ maxSuccess = 50000 } $ prop ioref ctable+ print "------- Map ASSOCS List ----"+ print =<< (M.assocs <$> readIORef ioref)+ print "--------Hashtable AFTER --------"+ print =<< atomically (H.getAssocs ctable)