prevent cache invariant querying when not joined
This commit is contained in:
parent
abbe664ca1
commit
8d349212b4
|
@ -32,6 +32,7 @@ module Hash2Pub.DHTProtocol
|
||||||
, ackRequest
|
, ackRequest
|
||||||
, isPossibleSuccessor
|
, isPossibleSuccessor
|
||||||
, isPossiblePredecessor
|
, isPossiblePredecessor
|
||||||
|
, isJoined
|
||||||
, closestCachePredecessors
|
, closestCachePredecessors
|
||||||
)
|
)
|
||||||
where
|
where
|
||||||
|
@ -213,8 +214,8 @@ markCacheEntryAsVerified timestamp nid = RingMap . Map.adjust adjustFunc nid . g
|
||||||
|
|
||||||
-- | uses the successor and predecessor list of a node as an indicator for whether a
|
-- | uses the successor and predecessor list of a node as an indicator for whether a
|
||||||
-- node has properly joined the DHT
|
-- node has properly joined the DHT
|
||||||
isJoined_ :: LocalNodeState -> Bool
|
isJoined :: LocalNodeState -> Bool
|
||||||
isJoined_ ns = not . all null $ [successors ns, predecessors ns]
|
isJoined ns = not . all null $ [successors ns, predecessors ns]
|
||||||
|
|
||||||
-- | the size limit to be used when serialising messages for sending
|
-- | the size limit to be used when serialising messages for sending
|
||||||
sendMessageSize :: Num i => i
|
sendMessageSize :: Num i => i
|
||||||
|
@ -260,8 +261,8 @@ handleIncomingRequest nsSTM sendQ msgSet sourceAddr = do
|
||||||
-- ToDo: figure out what happens if not joined
|
-- ToDo: figure out what happens if not joined
|
||||||
QueryID -> Just <$> respondQueryID nsSTM msgSet
|
QueryID -> Just <$> respondQueryID nsSTM msgSet
|
||||||
-- only when joined
|
-- only when joined
|
||||||
Leave -> if isJoined_ ns then Just <$> respondLeave nsSTM msgSet else pure Nothing
|
Leave -> if isJoined ns then Just <$> respondLeave nsSTM msgSet else pure Nothing
|
||||||
Stabilise -> if isJoined_ ns then Just <$> respondStabilise nsSTM msgSet else pure Nothing
|
Stabilise -> if isJoined ns then Just <$> respondStabilise nsSTM msgSet else pure Nothing
|
||||||
)
|
)
|
||||||
-- for single part request, response starts with part number 1. For multipart requests, response starts with part number n+1.
|
-- for single part request, response starts with part number 1. For multipart requests, response starts with part number n+1.
|
||||||
|
|
||||||
|
|
|
@ -253,7 +253,9 @@ checkCacheSliceInvariants :: LocalNodeState
|
||||||
-> NodeCache
|
-> NodeCache
|
||||||
-> [NodeID] -- ^ list of middle IDs of slices not
|
-> [NodeID] -- ^ list of middle IDs of slices not
|
||||||
-- ^ fulfilling the invariant
|
-- ^ fulfilling the invariant
|
||||||
checkCacheSliceInvariants ns = checkPredecessorSlice jEntries (getNid ns) startBound lastPred <> checkSuccessorSlice jEntries (getNid ns) startBound lastSucc
|
checkCacheSliceInvariants ns
|
||||||
|
| isJoined ns = checkPredecessorSlice jEntries (getNid ns) startBound lastPred <> checkSuccessorSlice jEntries (getNid ns) startBound lastSucc
|
||||||
|
| otherwise = const []
|
||||||
where
|
where
|
||||||
jEntries = jEntriesPerSlice ns
|
jEntries = jEntriesPerSlice ns
|
||||||
lastPred = getNid <$> lastMay (predecessors ns)
|
lastPred = getNid <$> lastMay (predecessors ns)
|
||||||
|
@ -340,7 +342,7 @@ stabiliseThread nsSTM = forever $ do
|
||||||
-- try looking up additional neighbours if list too short
|
-- try looking up additional neighbours if list too short
|
||||||
forM_ [(length $ predecessors updatedNs)..(kNeighbours updatedNs)] (\_ -> do
|
forM_ [(length $ predecessors updatedNs)..(kNeighbours updatedNs)] (\_ -> do
|
||||||
ns' <- readTVarIO nsSTM
|
ns' <- readTVarIO nsSTM
|
||||||
nextEntry <- requestQueryID ns' $ pred . getNid $ atDef (toRemoteNodeState ns') (predecessors ns') (-1)
|
nextEntry <- requestQueryID ns' $ pred . getNid $ lastDef (toRemoteNodeState ns') (predecessors ns')
|
||||||
atomically $ do
|
atomically $ do
|
||||||
latestNs <- readTVar nsSTM
|
latestNs <- readTVar nsSTM
|
||||||
writeTVar nsSTM $ addPredecessors [nextEntry] latestNs
|
writeTVar nsSTM $ addPredecessors [nextEntry] latestNs
|
||||||
|
@ -348,7 +350,7 @@ stabiliseThread nsSTM = forever $ do
|
||||||
|
|
||||||
forM_ [(length $ successors updatedNs)..(kNeighbours updatedNs)] (\_ -> do
|
forM_ [(length $ successors updatedNs)..(kNeighbours updatedNs)] (\_ -> do
|
||||||
ns' <- readTVarIO nsSTM
|
ns' <- readTVarIO nsSTM
|
||||||
nextEntry <- requestQueryID ns' $ succ . getNid $ atDef (toRemoteNodeState ns') (successors ns') (-1)
|
nextEntry <- requestQueryID ns' $ succ . getNid $ lastDef (toRemoteNodeState ns') (successors ns')
|
||||||
atomically $ do
|
atomically $ do
|
||||||
latestNs <- readTVar nsSTM
|
latestNs <- readTVar nsSTM
|
||||||
writeTVar nsSTM $ addSuccessors [nextEntry] latestNs
|
writeTVar nsSTM $ addSuccessors [nextEntry] latestNs
|
||||||
|
@ -460,6 +462,9 @@ fediMessageHandler :: TQueue (BS.ByteString, SockAddr) -- ^ send queue
|
||||||
-> LocalNodeStateSTM -- ^ acting NodeState
|
-> LocalNodeStateSTM -- ^ acting NodeState
|
||||||
-> IO ()
|
-> IO ()
|
||||||
fediMessageHandler sendQ recvQ nsSTM = do
|
fediMessageHandler sendQ recvQ nsSTM = do
|
||||||
|
-- Read node state just once, assuming that all relevant data for this function does
|
||||||
|
-- not change.
|
||||||
|
-- Other functions are passed the nsSTM reference and thus can get the latest state.
|
||||||
nsSnap <- readTVarIO nsSTM
|
nsSnap <- readTVarIO nsSTM
|
||||||
-- handling multipart messages:
|
-- handling multipart messages:
|
||||||
-- Request parts can be insert into a map (key: (sender IP against spoofing, request ID), value: timestamp + set of message parts, handle all of them when size of set == parts) before being handled. This map needs to be purged periodically by a separate thread and can be protected by an MVar for fairness.
|
-- Request parts can be insert into a map (key: (sender IP against spoofing, request ID), value: timestamp + set of message parts, handle all of them when size of set == parts) before being handled. This map needs to be purged periodically by a separate thread and can be protected by an MVar for fairness.
|
||||||
|
|
Loading…
Reference in a new issue