From 914e07a412a3105651188064eba64644173e1750 Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Fri, 5 Jun 2020 22:07:47 +0200 Subject: [PATCH] change NodeCache protection to STM - putting the NodeCache behind an IORef had been chose because those could've been read non-blocking - the same is true for TVars. The performance characteristics are likely worse, but at the advantage of composability within STM monads --- src/Hash2Pub/DHTProtocol.hs | 4 ++-- src/Hash2Pub/FediChord.hs | 19 +++++++------------ src/Hash2Pub/FediChordTypes.hs | 5 ++--- test/FediChordSpec.hs | 1 - 4 files changed, 11 insertions(+), 18 deletions(-) diff --git a/src/Hash2Pub/DHTProtocol.hs b/src/Hash2Pub/DHTProtocol.hs index 8857597..e30fe26 100644 --- a/src/Hash2Pub/DHTProtocol.hs +++ b/src/Hash2Pub/DHTProtocol.hs @@ -27,6 +27,7 @@ module Hash2Pub.DHTProtocol import Control.Concurrent.Async import Control.Concurrent.STM +import Control.Concurrent.STM.TVar import Control.Concurrent.STM.TBQueue import Control.Concurrent.STM.TQueue import Control.Exception @@ -35,7 +36,6 @@ import qualified Data.ByteString as BS import Data.Either (rights) import Data.Foldable (foldl', foldr') import Data.Functor.Identity -import Data.IORef import Data.IP (IPv6, fromHostAddress6, toHostAddress6) import Data.List (sortBy) @@ -272,7 +272,7 @@ requestQueryID :: LocalNodeState -- ^ NodeState of the querying node -- 5. repeat until FOUND (problem: new entries not necessarily already in cache, explicitly compare with closer results) -- TODO: deal with lookup failures requestQueryID ns targetID = do - firstCacheSnapshot <- readIORef . nodeCacheRef $ ns + firstCacheSnapshot <- readTVarIO . nodeCacheSTM $ ns queryIdLookupLoop firstCacheSnapshot ns targetID -- | like 'requestQueryID, but allows passing of a custom cache, e.g. for joining diff --git a/src/Hash2Pub/FediChord.hs b/src/Hash2Pub/FediChord.hs index 43de152..795772b 100644 --- a/src/Hash2Pub/FediChord.hs +++ b/src/Hash2Pub/FediChord.hs @@ -70,7 +70,6 @@ import Crypto.Hash import qualified Data.ByteArray as BA import qualified Data.ByteString as BS import qualified Data.ByteString.UTF8 as BSU -import Data.IORef import Data.IP (IPv6, fromHostAddress6, toHostAddress6) import Data.Typeable (Typeable (..), typeOf) @@ -96,7 +95,7 @@ fediChordInit conf = do -- Separated from 'fediChordInit' to be usable in tests. nodeStateInit :: FediChordConf -> IO LocalNodeState nodeStateInit conf = do - cacheRef <- newIORef initCache + cacheSTM <- newTVarIO initCache q <- atomically newTQueue let containedState = RemoteNodeState { @@ -109,7 +108,7 @@ nodeStateInit conf = do } initialState = LocalNodeState { nodeState = containedState - , nodeCacheRef = cacheRef + , nodeCacheSTM = cacheSTM , cacheWriteQueue = q , successors = [] , predecessors = [] @@ -172,15 +171,11 @@ fediChordJoin cacheSnapshot nsSTM = do -- | cache updater thread that waits for incoming NodeCache update instructions on -- the node's cacheWriteQueue and then modifies the NodeCache as the single writer. cacheWriter :: LocalNodeStateSTM -> IO () -cacheWriter nsSTM = do - ns <- readTVarIO nsSTM - let writeQueue' = cacheWriteQueue ns - forever $ do - f <- atomically $ readTQueue writeQueue' - let - refModifier :: NodeCache -> (NodeCache, ()) - refModifier nc = (f nc, ()) - atomicModifyIORef' (nodeCacheRef ns) refModifier +cacheWriter nsSTM = + forever $ atomically $ do + ns <- readTVar nsSTM + cacheModifier <- readTQueue $ cacheWriteQueue ns + modifyTVar' (nodeCacheSTM ns) cacheModifier -- | Receives UDP packets and passes them to other threads via the given TQueue. -- Shall be used as the single receiving thread on the server socket, as multiple diff --git a/src/Hash2Pub/FediChordTypes.hs b/src/Hash2Pub/FediChordTypes.hs index 2feea08..1f28aea 100644 --- a/src/Hash2Pub/FediChordTypes.hs +++ b/src/Hash2Pub/FediChordTypes.hs @@ -47,7 +47,6 @@ import Crypto.Hash import qualified Data.ByteArray as BA import qualified Data.ByteString as BS import qualified Data.ByteString.UTF8 as BSU -import Data.IORef import Data.IP (IPv6, fromHostAddress6, toHostAddress6) import Data.Typeable (Typeable (..), typeOf) @@ -127,7 +126,7 @@ data RemoteNodeState = RemoteNodeState data LocalNodeState = LocalNodeState { nodeState :: RemoteNodeState -- ^ represents common data present both in remote and local node representations - , nodeCacheRef :: IORef NodeCache + , nodeCacheSTM :: TVar NodeCache -- ^ EpiChord node cache with expiry times for nodes , cacheWriteQueue :: TQueue (NodeCache -> NodeCache) -- ^ cache updates are not written directly to the 'nodeCache' but queued and @@ -206,7 +205,7 @@ instance NodeState LocalNodeState where toRemoteNodeState = nodeState -- | defining Show instances to be able to print NodeState for debug purposes -instance Typeable a => Show (IORef a) where +instance Typeable a => Show (TVar a) where show x = show (typeOf x) instance Typeable a => Show (TQueue a) where diff --git a/test/FediChordSpec.hs b/test/FediChordSpec.hs index 629c7c2..36ae2de 100644 --- a/test/FediChordSpec.hs +++ b/test/FediChordSpec.hs @@ -4,7 +4,6 @@ module FediChordSpec where import Control.Exception import Data.ASN1.Parse (runParseASN1) import qualified Data.ByteString as BS -import Data.IORef import qualified Data.Map.Strict as Map import Data.Maybe (fromJust) import qualified Data.Set as Set