wip, some probes

This commit is contained in:
voidlizard 2024-11-16 09:01:05 +03:00
parent 08816dfc46
commit 49b16606aa
2 changed files with 24 additions and 3 deletions

View File

@ -592,10 +592,12 @@ data PSt =
downloadDispatcher :: forall e m . ( e ~ L4Proto
, MonadUnliftIO m
)
=> SomeBrains e
=> AnyProbe
-> SomeBrains e
-> PeerEnv e
-> m ()
downloadDispatcher brains env = flip runContT pure do
downloadDispatcher probe brains env = flip runContT pure do
pts <- newTVarIO ( mempty :: HashMap (Peer e) (Async (), PeerNonce) )
@ -632,6 +634,13 @@ downloadDispatcher brains env = flip runContT pure do
dupes <- newTVarIO ( mempty :: HashMap HashRef Int )
ContT $ withAsync $ forever $ pause @'Seconds 10 >> do
acceptReport probe =<< S.toList_ do
wip <- readTVarIO wip <&> HM.size
pn <- readTVarIO pts <&> HM.size
S.yield ( "wip", fromIntegral wip )
S.yield ( "peerThreads", fromIntegral pn )
ContT $ withAsync do
polling (Polling 10 10) (readTVarIO dupes <&> fmap (,60) . HM.keys) $ \h -> do
atomically $ modifyTVar dupes (HM.delete h)
@ -678,6 +687,11 @@ downloadDispatcher brains env = flip runContT pure do
where
manageThreads :: (HashRef -> STM ())
-> TVar (HashMap HashRef DCB)
-> TVar (HashMap (Peer e) (Async (), PeerNonce))
-> m ()
manageThreads onBlock wip pts = do
_pnum <- newTVarIO 1
@ -686,6 +700,10 @@ downloadDispatcher brains env = flip runContT pure do
forever $ (>> pause @'Seconds 10) $ withPeerM env do
debug "MANAGE THREADS"
acceptReport probe =<< S.toList_ do
n <- readTVarIO _psem <&> HM.size
S.yield ( "psem", fromIntegral n )
peers <- HM.fromList <$> do
pips <- getKnownPeers @e <&> HS.fromList
S.toList_ $ for_ pips $ \p -> do

View File

@ -1199,8 +1199,11 @@ runPeer opts = respawnOnError opts $ do
peerThread "pexLoop" (pexLoop @e brains tcp)
downloadProbe <- newSimpleProbe "Download"
addProbe downloadProbe
-- FIXME: new-download-loop
peerThread "downloadDispatcher" (downloadDispatcher (SomeBrains brains) env)
peerThread "downloadDispatcher" (downloadDispatcher downloadProbe (SomeBrains brains) env)
peerThread "fillPeerMeta" (fillPeerMeta tcp tcpProbeWait)