added peer ping loop to remove inactive peers

This commit is contained in:
Dmitry Zuikov 2023-02-04 14:11:05 +03:00
parent fbf8bd27fb
commit d5d7c6fbb7
5 changed files with 68 additions and 10 deletions

View File

@ -7,11 +7,13 @@ import HBS2.Net.Proto.Types
class PeerLocator e l where class PeerLocator e l where
knownPeers :: forall m . (HasPeer e, MonadIO m) => l -> m [Peer e] knownPeers :: forall m . (HasPeer e, MonadIO m) => l -> m [Peer e]
addPeers :: forall m . (HasPeer e, MonadIO m) => l -> [Peer e] -> m () addPeers :: forall m . (HasPeer e, MonadIO m) => l -> [Peer e] -> m ()
delPeers :: forall m . (HasPeer e, MonadIO m) => l -> [Peer e] -> m ()
data AnyPeerLocator e = forall a . PeerLocator e a => AnyPeerLocator a data AnyPeerLocator e = forall a . PeerLocator e a => AnyPeerLocator a
instance HasPeer e => PeerLocator e (AnyPeerLocator e) where instance HasPeer e => PeerLocator e (AnyPeerLocator e) where
knownPeers (AnyPeerLocator l) = knownPeers l knownPeers (AnyPeerLocator l) = knownPeers l
addPeers (AnyPeerLocator l) = addPeers l addPeers (AnyPeerLocator l) = addPeers l
delPeers (AnyPeerLocator l) = addPeers l

View File

@ -26,5 +26,7 @@ instance Ord (Peer e) => PeerLocator e (StaticPeerLocator e) where
addPeers (StaticPeerLocator peers) new = do addPeers (StaticPeerLocator peers) new = do
liftIO $ atomically $ modifyTVar' peers (<> Set.fromList new) liftIO $ atomically $ modifyTVar' peers (<> Set.fromList new)
delPeers (StaticPeerLocator peers) del = do
liftIO $ atomically $ modifyTVar' peers (`Set.difference` Set.fromList del)

View File

@ -314,7 +314,10 @@ downloadFromWithPeer peer thisBkSize h = do
) )
if not (null catched) then do if not (null catched) then do
liftIO $ atomically $ modifyTVar (view peerDownloaded pinfo) (+chunksN) liftIO $ atomically do
modifyTVar (view peerDownloaded pinfo) (+chunksN)
writeTVar (view peerPingFailed pinfo) 0
else do else do
liftIO $ atomically $ modifyTVar (view peerErrors pinfo) succ liftIO $ atomically $ modifyTVar (view peerErrors pinfo) succ

View File

@ -1,15 +1,24 @@
{-# Language TemplateHaskell #-} {-# Language TemplateHaskell #-}
{-# Language AllowAmbiguousTypes #-}
module PeerInfo where module PeerInfo where
import HBS2.Prelude.Plated import HBS2.Actors.Peer
import HBS2.Net.Proto.Sessions
import HBS2.Net.Messaging.UDP
import HBS2.Clock import HBS2.Clock
import HBS2.Defaults import HBS2.Defaults
import HBS2.Net.Messaging.UDP
import HBS2.Net.PeerLocator
import HBS2.Net.Proto.Peer
import HBS2.Net.Proto.Sessions
import HBS2.Net.Proto.Types
import HBS2.Prelude.Plated
import HBS2.System.Logger.Simple
import Data.Foldable
import Lens.Micro.Platform import Lens.Micro.Platform
import Control.Concurrent.STM.TVar import Control.Concurrent.STM.TVar
import Control.Concurrent.STM
import Control.Monad
import Prettyprinter
data PeerInfo e = data PeerInfo e =
PeerInfo PeerInfo
@ -20,6 +29,7 @@ data PeerInfo e =
, _peerLastWatched :: TVar TimeSpec , _peerLastWatched :: TVar TimeSpec
, _peerDownloaded :: TVar Int , _peerDownloaded :: TVar Int
, _peerDownloadedLast :: TVar Int , _peerDownloadedLast :: TVar Int
, _peerPingFailed :: TVar Int
} }
deriving stock (Generic,Typeable) deriving stock (Generic,Typeable)
@ -35,6 +45,7 @@ newPeerInfo = liftIO do
<*> newTVarIO 0 <*> newTVarIO 0
<*> newTVarIO 0 <*> newTVarIO 0
<*> newTVarIO 0 <*> newTVarIO 0
<*> newTVarIO 0
type instance SessionData e (PeerInfo e) = PeerInfo e type instance SessionData e (PeerInfo e) = PeerInfo e
@ -48,3 +59,36 @@ instance Expires (SessionKey UDP (PeerInfo UDP)) where
expiresIn = const (Just 600) expiresIn = const (Just 600)
peerPingLoop :: forall e m . ( HasPeerLocator e m
, HasPeer e
, HasNonces (PeerHandshake e) m
, Nonce (PeerHandshake e) ~ PingNonce
, Request e (PeerHandshake e) m
, Sessions e (PeerHandshake e) m
, Sessions e (PeerInfo e) m
, Pretty (Peer e)
, MonadIO m
)
=> m ()
peerPingLoop = forever do
pause @'Minutes 2 -- FIXME: defaults
debug "peerPingLoop"
pl <- getPeerLocator @e
pips <- knownPeers @e pl
for_ pips $ \p -> do
npi <- newPeerInfo
pfails <- fetch True npi (PeerInfoKey p) (view peerPingFailed)
liftIO $ atomically $ modifyTVar pfails succ
sendPing @e p
pause @'Seconds 2 -- NOTE: it's okay?
fnum <- liftIO $ readTVarIO pfails
when (fnum > 3) do -- FIXME: hardcode!
warn $ "removing peer" <+> pretty p <+> "for not responding to our pings"
delPeers pl [p]
expire (PeerInfoKey p)

View File

@ -28,6 +28,7 @@ import HBS2.System.Logger.Simple qualified as Log
import RPC import RPC
import BlockDownload import BlockDownload
import PeerInfo
import Data.Maybe import Data.Maybe
import Crypto.Saltine (sodiumInit) import Crypto.Saltine (sodiumInit)
@ -87,10 +88,10 @@ main = do
sodiumInit sodiumInit
setLogging @DEBUG (set loggerTr ("[debug] " <>)) setLogging @DEBUG (set loggerTr ("[debug] " <>))
setLogging @INFO defLog setLogging @INFO (set loggerTr ("[info] " <>))
setLogging @ERROR defLog setLogging @ERROR (set loggerTr ("[error] " <>))
setLogging @WARN defLog setLogging @WARN (set loggerTr ("[warn] " <>))
setLogging @NOTICE defLog setLogging @NOTICE (set loggerTr ("[notice] " <>))
withSimpleLogger runCLI withSimpleLogger runCLI
@ -293,10 +294,14 @@ runPeer opts = Exception.handle myException $ do
subscribe @UDP AnyKnownPeerEventKey $ \(KnownPeerEvent p d) -> do subscribe @UDP AnyKnownPeerEventKey $ \(KnownPeerEvent p d) -> do
addPeers pl [p] addPeers pl [p]
npi <- newPeerInfo
pfails <- fetch True npi (PeerInfoKey p) (view peerPingFailed)
liftIO $ atomically $ writeTVar pfails 0
debug $ "Got authorized peer!" <+> pretty p debug $ "Got authorized peer!" <+> pretty p
<+> pretty (AsBase58 (view peerSignKey d)) <+> pretty (AsBase58 (view peerSignKey d))
void $ liftIO $ async $ withPeerM env do void $ liftIO $ async $ withPeerM env do
pause @'Seconds 1 pause @'Seconds 1
debug "sending first peer announce" debug "sending first peer announce"
@ -307,6 +312,8 @@ runPeer opts = Exception.handle myException $ do
debug "sending local peer announce" debug "sending local peer announce"
request localMulticast (PeerAnnounce @UDP pnonce) request localMulticast (PeerAnnounce @UDP pnonce)
as <- liftIO $ async $ withPeerM env (peerPingLoop @UDP)
as <- liftIO $ async $ withPeerM env (blockDownloadLoop denv) as <- liftIO $ async $ withPeerM env (blockDownloadLoop denv)
rpc <- liftIO $ async $ withPeerM env $ forever $ do rpc <- liftIO $ async $ withPeerM env $ forever $ do