parent
7a87d86c32
commit
1a0de55b8c
|
@ -123,34 +123,21 @@ fediChordInit initConf serviceRunner = do
|
||||||
-- prepare for joining: start node cache writer thread
|
-- prepare for joining: start node cache writer thread
|
||||||
-- currently no masking is necessary, as there is nothing to clean up
|
-- currently no masking is necessary, as there is nothing to clean up
|
||||||
nodeCacheWriterThread <- forkIO $ nodeCacheWriter realNodeSTM
|
nodeCacheWriterThread <- forkIO $ nodeCacheWriter realNodeSTM
|
||||||
fediThreadsAsync <- if confEnableKChoices initConf
|
fediThreadsAsync <- do
|
||||||
then
|
|
||||||
-- TODO: k-choices way of joining
|
|
||||||
-- placeholder
|
|
||||||
runExceptT (kChoicesNodeJoin realNodeSTM ("foo", fromIntegral 3))
|
|
||||||
>> async (fediMainThreads serverSock realNodeSTM)
|
|
||||||
else do
|
|
||||||
-- without k-choices, just initialise a single vserver
|
|
||||||
firstVS <- nodeStateInit realNodeSTM 0
|
|
||||||
firstVSSTM <- newTVarIO firstVS
|
|
||||||
-- add vserver to list at RealNode
|
|
||||||
atomically . modifyTVar' realNodeSTM $ \rn -> rn { vservers = HMap.insert (getNid firstVS) firstVSSTM (vservers rn) }
|
|
||||||
-- try joining the DHT using one of the provided bootstrapping nodes
|
|
||||||
joinedState <- tryBootstrapJoining firstVSSTM
|
|
||||||
|
|
||||||
either (\err -> do
|
either (\err -> do
|
||||||
-- handle unsuccessful join
|
-- handle unsuccessful join
|
||||||
putStrLn $ err <> " Error joining, start listening for incoming requests anyways"
|
putStrLn $ err <> " Error joining, start listening for incoming requests anyways"
|
||||||
-- launch thread attempting to join on new cache entries
|
-- launch thread attempting to join on new cache entries
|
||||||
_ <- forkIO $ joinOnNewEntriesThread firstVSSTM
|
-- TODO: adjust joinOnNewEntriesThread to k-choices
|
||||||
|
--_ <- forkIO $ joinOnNewEntriesThread firstVSSTM
|
||||||
async (fediMainThreads serverSock realNodeSTM)
|
async (fediMainThreads serverSock realNodeSTM)
|
||||||
)
|
)
|
||||||
(\joinedNS -> do
|
(\_ -> do
|
||||||
-- launch main eventloop with successfully joined state
|
-- launch main eventloop with successfully joined state
|
||||||
putStrLn "successful join"
|
putStrLn "successful join"
|
||||||
async (fediMainThreads serverSock realNodeSTM)
|
async (fediMainThreads serverSock realNodeSTM)
|
||||||
)
|
)
|
||||||
joinedState
|
=<< tryBootstrapJoining realNodeSTM
|
||||||
pure (fediThreadsAsync, realNodeSTM)
|
pure (fediThreadsAsync, realNodeSTM)
|
||||||
|
|
||||||
|
|
||||||
|
@ -318,7 +305,7 @@ kChoicesJoinCost remainingOwnLoad ownCap segment =
|
||||||
fediChordBootstrapJoin :: Service s (RealNodeSTM s)
|
fediChordBootstrapJoin :: Service s (RealNodeSTM s)
|
||||||
=> LocalNodeStateSTM s -- ^ the local 'NodeState'
|
=> LocalNodeStateSTM s -- ^ the local 'NodeState'
|
||||||
-> (String, PortNumber) -- ^ domain and port of a bootstrapping node
|
-> (String, PortNumber) -- ^ domain and port of a bootstrapping node
|
||||||
-> IO (Either String (LocalNodeStateSTM s)) -- ^ the joined 'NodeState' after a
|
-> IO (Either String ()) -- ^ the joined 'NodeState' after a
|
||||||
-- successful join, otherwise an error message
|
-- successful join, otherwise an error message
|
||||||
fediChordBootstrapJoin nsSTM bootstrapNode = do
|
fediChordBootstrapJoin nsSTM bootstrapNode = do
|
||||||
-- can be invoked multiple times with all known bootstrapping nodes until successfully joined
|
-- can be invoked multiple times with all known bootstrapping nodes until successfully joined
|
||||||
|
@ -330,10 +317,10 @@ fediChordBootstrapJoin nsSTM bootstrapNode = do
|
||||||
liftIO . putStrLn $ "Trying to join on " <> show (getNid currentlyResponsible)
|
liftIO . putStrLn $ "Trying to join on " <> show (getNid currentlyResponsible)
|
||||||
-- 2. then send a join to the currently responsible node
|
-- 2. then send a join to the currently responsible node
|
||||||
liftIO $ putStrLn "send a bootstrap Join"
|
liftIO $ putStrLn "send a bootstrap Join"
|
||||||
joinResult <- liftIO $ requestJoin currentlyResponsible nsSTM
|
_ <- liftEither =<< liftIO (requestJoin currentlyResponsible nsSTM)
|
||||||
liftEither joinResult
|
pure ()
|
||||||
|
|
||||||
-- Periodically lookup own ID through a random bootstrapping node to discover and merge separated DHT clusters.
|
-- Periodically lookup own IDs through a random bootstrapping node to discover and merge separated DHT clusters.
|
||||||
-- Unjoined try joining instead.
|
-- Unjoined try joining instead.
|
||||||
convergenceSampleThread :: Service s (RealNodeSTM s) => RealNodeSTM s -> IO ()
|
convergenceSampleThread :: Service s (RealNodeSTM s) => RealNodeSTM s -> IO ()
|
||||||
convergenceSampleThread nodeSTM = forever $ do
|
convergenceSampleThread nodeSTM = forever $ do
|
||||||
|
@ -362,27 +349,41 @@ convergenceSampleThread nodeSTM = forever $ do
|
||||||
else pure ()
|
else pure ()
|
||||||
) >> pure ()
|
) >> pure ()
|
||||||
-- unjoined node: try joining through all bootstrapping nodes
|
-- unjoined node: try joining through all bootstrapping nodes
|
||||||
else tryBootstrapJoining nsSTM >> pure ()
|
else tryBootstrapJoining nodeSTM >> pure ()
|
||||||
|
|
||||||
let delaySecs = confBootstrapSamplingInterval . nodeConfig $ node
|
let delaySecs = confBootstrapSamplingInterval . nodeConfig $ node
|
||||||
threadDelay delaySecs
|
threadDelay delaySecs
|
||||||
|
|
||||||
|
|
||||||
-- | Try joining the DHT through any of the bootstrapping nodes until it succeeds.
|
-- | Try joining the DHT through any of the bootstrapping nodes until it succeeds.
|
||||||
tryBootstrapJoining :: Service s (RealNodeSTM s) => LocalNodeStateSTM s -> IO (Either String (LocalNodeStateSTM s))
|
tryBootstrapJoining :: Service s (RealNodeSTM s) => RealNodeSTM s -> IO (Either String ())
|
||||||
tryBootstrapJoining nsSTM = do
|
tryBootstrapJoining nodeSTM = do
|
||||||
bss <- atomically $ do
|
node <- readTVarIO nodeSTM
|
||||||
nsSnap <- readTVar nsSTM
|
let
|
||||||
realNodeSnap <- readTVar $ parentRealNode nsSnap
|
bss = bootstrapNodes node
|
||||||
pure $ bootstrapNodes realNodeSnap
|
conf = nodeConfig node
|
||||||
tryJoining bss
|
if confEnableKChoices conf
|
||||||
|
then tryJoining bss $ runExceptT . kChoicesNodeJoin nodeSTM
|
||||||
|
else do
|
||||||
|
firstVS <- nodeStateInit nodeSTM 0
|
||||||
|
firstVSSTM <- newTVarIO firstVS
|
||||||
|
joinResult <- tryJoining bss (fediChordBootstrapJoin firstVSSTM)
|
||||||
|
either
|
||||||
|
(pure . Left)
|
||||||
|
(\_ -> fmap Right . atomically . modifyTVar' nodeSTM $ (\rn -> rn
|
||||||
|
{ vservers = HMap.insert (getNid firstVS) firstVSSTM (vservers rn) }
|
||||||
|
)
|
||||||
|
)
|
||||||
|
(joinResult :: Either String ())
|
||||||
|
|
||||||
where
|
where
|
||||||
tryJoining (bn:bns) = do
|
tryJoining :: [(String, PortNumber)] -> ((String, PortNumber) -> IO (Either String ())) -> IO (Either String ())
|
||||||
j <- fediChordBootstrapJoin nsSTM bn
|
tryJoining (bn:bns) joinFunc = do
|
||||||
|
j <- joinFunc bn
|
||||||
case j of
|
case j of
|
||||||
Left err -> putStrLn ("join error: " <> err) >> tryJoining bns
|
Left err -> putStrLn ("join error: " <> err) >> tryJoining bns joinFunc
|
||||||
Right joined -> pure $ Right joined
|
Right joined -> pure $ Right ()
|
||||||
tryJoining [] = pure $ Left "Exhausted all bootstrap points for joining."
|
tryJoining [] _ = pure $ Left "Exhausted all bootstrap points for joining."
|
||||||
|
|
||||||
|
|
||||||
-- | Look up a key just based on the responses of a single bootstrapping node.
|
-- | Look up a key just based on the responses of a single bootstrapping node.
|
||||||
|
|
Loading…
Reference in a new issue