feat: enhancing protocol management with mutex locks #2137
Conversation
size-limit report 📦
|
danisharora099
force-pushed
the
fix/peer-management
branch
2 times, most recently
from
4de00a5
to
aeb05cd
Compare
danisharora099
changed the title
danisharora099
force-pushed
the
fix/peer-management
branch
from
09028e7
to
a89e69f
Compare
| @@ -45,7 +45,7 @@ class Metadata extends BaseProtocol implements IMetadata { | |||
| pubsubTopicsToShardInfo(this.pubsubTopics) | |||
| ); | |||
|
|
|||
| const peer = await this.peerStore.get(peerId); | |||
| const peer = await this.libp2pComponents.peerStore.get(peerId); | |||
There was a problem hiding this comment.
we already had access to libp2pCompnents on the Base class, just turned it from private to protected and removed an additional arg of PeerStore
| @@ -18,7 +18,7 @@ export class ReliabilityMonitorManager { | |||
| public static createReceiverMonitor( | |||
| pubsubTopic: PubsubTopic, | |||
| getPeers: () => Peer[], | |||
| renewPeer: (peerId: PeerId) => Promise<Peer>, | |||
There was a problem hiding this comment.
in which cases it can be undefined? I expect it should be always provided
There was a problem hiding this comment.
If no new is found, this function will indeed return undefined:
const newPeer = await this.peerManager.findPeers(1);
if (newPeer.length === 0) {
this.log.error(
"Failed to find a new peer to replace the disconnected one"
);
return undefined;
}
| @@ -54,6 +54,9 @@ class FilterSDK extends BaseProtocolSDK implements IFilterSDK { | |||
|
|
|||
| await subscription.processIncomingMessage(wakuMessage, peerIdStr); | |||
| }, | |||
| async (error: Error) => { | |||
| @@ -1,4 +1,4 @@ | |||
| export const DEFAULT_KEEP_ALIVE = 30 * 1000; | |||
There was a problem hiding this comment.
Ideally we should be fine with increasing this number, not decreasing.
How did you reach this number?
| @@ -36,10 +35,6 @@ export type NetworkConfig = StaticSharding | AutoSharding; | |||
| * Options for using LightPush and Filter | |||
| */ | |||
| export type ProtocolUseOptions = { | |||
| /** | |||
There was a problem hiding this comment.
why these properties are removed?
There was a problem hiding this comment.
Not required for the backoff
| sortPeersByLatency | ||
| } from "@waku/utils/libp2p"; | ||
| import { Logger } from "@waku/utils"; | ||
| import { getPeersForProtocol, sortPeersByLatency } from "@waku/utils/libp2p"; |
There was a problem hiding this comment.
nit: these utils are used only in one place, let's move them in then
There was a problem hiding this comment.
let's do it in a different PR
| ); | ||
| this.updatePeers(updatedPeers); | ||
| await this.connectionManager.dropConnection(peerToDisconnect); | ||
| await this.peerManager.removePeer(peerToDisconnect); |
There was a problem hiding this comment.
maybe .replace operation should be implemented with a lock inside? this should be enough so we can to avoid usage of extra lib
There was a problem hiding this comment.
removePeer indeed has a lock inside
| { | ||
| payload: utf8ToBytes("Hello_World") | ||
| }, | ||
| { forceUseAllPeers: true } |
There was a problem hiding this comment.
so that PeerManager is forced to find all peers until numPeersToUse which awaits the process instead
Problem
The current implementation showcased a lot of inconsistencies around peer management for protocols.
Multiple weird behaviour were observed:
Solution
BaseProtocolclass to enhance performance and reliabilityFilterCoreclassasync-mutexfor all locks-related choresNotes
Contribution checklist:
!in title if breaks public API;