mirror of
https://github.com/fluencelabs/js-libp2p
synced 2025-04-25 02:22:14 +00:00
refactor: connection manager (#511)
* refactor: initial refactor of the connection manager * fix: start/stop issues * fix: add tests and resolve pruning issues * chore: fix lint * test: move conn manager tests to node only for now * chore: apply suggestions from code review Co-Authored-By: Vasco Santos <vasco.santos@moxy.studio> * fix: assert min max connection options * test: fix assertion check for browser * docs: add api and config docs for conn manager
This commit is contained in:
Jacob Heun
parent
f1eb373235
commit
14a1955a78
30
doc/API.md
30
doc/API.md
@ -643,6 +643,36 @@ const handler = (msg) => {
|
||||
libp2p.pubsub.unsubscribe(topic, handler)
|
||||
```
|
||||
|
||||
### connectionManager.setPeerValue
|
||||
|
||||
Enables users to change the value of certain peers in a range of 0 to 1. Peers with the lowest values will have their Connections pruned first, if any Connection Manager limits are exceeded. See [./CONFIGURATION.md#configuring-connection-manager](./CONFIGURATION.md#configuring-connection-manager) for details on how to configure these limits.
|
||||
|
||||
`libp2p.connectionManager.setPeerValue(peerId, value)`
|
||||
|
||||
#### Parameters
|
||||
|
||||
| Name | Type | Description |
|
||||
|------|------|-------------|
|
||||
| peerId | `PeerId` | The peer to set the value for |
|
||||
| value | `number` | The value of the peer from 0 to 1 |
|
||||
|
||||
#### Returns
|
||||
|
||||
| Type | Description |
|
||||
|------|-------------|
|
||||
| `void` | |
|
||||
|
||||
#### Example
|
||||
|
||||
```js
|
||||
const topic = 'topic'
|
||||
const handler = (msg) => {
|
||||
// msg.data - pubsub data received
|
||||
}
|
||||
|
||||
libp2p.pubsub.unsubscribe(topic, handler)
|
||||
```
|
||||
|
||||
### metrics.global
|
||||
|
||||
A [`Stats`](#stats) object of tracking the global bandwidth of the libp2p node.
|
||||
|
||||
@ -15,10 +15,12 @@
|
||||
- [Examples](#examples)
|
||||
- [Basic setup](#basic-setup)
|
||||
- [Customizing Peer Discovery](#customizing-peer-discovery)
|
||||
- [Setup webrtc transport and discovery](#setup-webrtc-transport-and-discovery)
|
||||
- [Customizing Pubsub](#customizing-pubsub)
|
||||
- [Customizing DHT](#customizing-dht)
|
||||
- [Setup with Content and Peer Routing](#setup-with-content-and-peer-routing)
|
||||
- [Setup with Relay](#setup-with-relay)
|
||||
- [Configuring Connection Manager](#configuring-connection-manager)
|
||||
- [Configuring Metrics](#configuring-metrics)
|
||||
- [Configuration examples](#configuration-examples)
|
||||
|
||||
@ -412,6 +414,37 @@ const node = await Libp2p.create({
|
||||
})
|
||||
```
|
||||
|
||||
#### Configuring Connection Manager
|
||||
|
||||
The Connection Manager prunes Connections in libp2p whenever certain limits are exceeded. If Metrics are enabled, you can also configure the Connection Manager to monitor the bandwidth of libp2p and prune connections as needed. You can read more about what Connection Manager does at [./CONNECTION_MANAGER.md](./CONNECTION_MANAGER.md). The configuration values below show the defaults for Connection Manager. See [./CONNECTION_MANAGER.md](./CONNECTION_MANAGER.md#options) for a full description of the parameters.
|
||||
|
||||
```js
|
||||
const Libp2p = require('libp2p')
|
||||
const TCP = require('libp2p-tcp')
|
||||
const MPLEX = require('libp2p-mplex')
|
||||
const SECIO = require('libp2p-secio')
|
||||
|
||||
const node = await Libp2p.create({
|
||||
modules: {
|
||||
transport: [TCP],
|
||||
streamMuxer: [MPLEX],
|
||||
connEncryption: [SECIO]
|
||||
},
|
||||
connectionManager: {
|
||||
maxConnections: Infinity,
|
||||
minConnections: 0,
|
||||
pollInterval: 2000,
|
||||
defaultPeerValue: 1,
|
||||
// The below values will only be taken into account when Metrics are enabled
|
||||
maxData: Infinity,
|
||||
maxSentData: Infinity,
|
||||
maxReceivedData: Infinity,
|
||||
maxEventLoopDelay: Infinity,
|
||||
movingAverageInterval: 60000
|
||||
}
|
||||
})
|
||||
```
|
||||
|
||||
#### Configuring Metrics
|
||||
|
||||
Metrics are disabled in libp2p by default. You can enable and configure them as follows. Aside from enabled being `false` by default, the configuration options listed here are the current defaults.
|
||||
|
||||
20
doc/CONNECTION_MANAGER.md
Normal file
20
doc/CONNECTION_MANAGER.md
Normal file
@ -0,0 +1,20 @@
|
||||
# Connection Manager
|
||||
|
||||
The Connection Manager works with the Registrar to keep connections across libp2p within acceptable ranges, which can be configured. By default Connection Manager will monitor:
|
||||
- The total number of open connections
|
||||
- The latency/delay of the event loop
|
||||
|
||||
If Metrics are enabled for libp2p, see [./CONFIGURATION.md#configuring-metrics](./CONFIGURATION.md#configuring-metrics) on how to configure metrics, the Connection Manager can be used to prune connections when certain limits are exceeded.
|
||||
|
||||
The following is a list of available options for setting limits for the Connection Manager to enforce.
|
||||
|
||||
## Options
|
||||
- `maxConnections`: the maximum number of connections libp2p is willing to have before it starts disconnecting. Defaults to `Infinity`
|
||||
- `minConnections`: the minimum number of connections below which libp2p not activate preemptive disconnections. Defaults to `0`.
|
||||
- `maxData`: sets the maximum data — in bytes per second - (sent and received) this node is willing to endure before it starts disconnecting peers. Defaults to `Infinity`.
|
||||
- `maxSentData`: sets the maximum sent data — in bytes per second - this node is willing to endure before it starts disconnecting peers. Defaults to `Infinity`.
|
||||
- `maxReceivedData`: sets the maximum received data — in bytes per second - this node is willing to endure before it starts disconnecting peers. Defaults to `Infinity`.
|
||||
- `maxEventLoopDelay`: sets the maximum event loop delay (measured in milliseconds) this node is willing to endure before it starts disconnecting peers. Defaults to `Infinity`.
|
||||
- `pollInterval`: sets the poll interval (in milliseconds) for assessing the current state and determining if this peer needs to force a disconnect. Defaults to `2000` (2 seconds).
|
||||
- `movingAverageInterval`: the interval used to calculate moving averages (in milliseconds). Defaults to `60000` (1 minute). This must be an available interval configured in `Metrics`
|
||||
- `defaultPeerValue`: number between 0 and 1. Defaults to 1.
|
||||
@ -1,99 +0,0 @@
|
||||
# libp2p-connection-manager
|
||||
|
||||
> JavaScript connection manager for libp2p
|
||||
|
||||
**Note**: git history prior to merging into js-libp2p can be found in the original repository, https://github.com/libp2p/js-libp2p-connection-manager.
|
||||
|
||||
## Table of Contents
|
||||
|
||||
- [Install](#install)
|
||||
- [npm](#npm)
|
||||
- [Use in Node.js, a browser with browserify, webpack or any other bundler](##use-in-nodejs-or-in-the-browser-with-browserify-webpack-or-any-other-bundler)
|
||||
- [Usage](#usage)
|
||||
- [API](#api)
|
||||
- [Contribute](#contribute)
|
||||
- [License](#license)
|
||||
|
||||
## API
|
||||
|
||||
A connection manager manages the peers you're connected to. The application provides one or more limits that will trigger the disconnection of peers. These limits can be any of the following:
|
||||
|
||||
* number of connected peers
|
||||
* maximum bandwidth (sent / received or both)
|
||||
* maximum event loop delay
|
||||
|
||||
The connection manager will disconnect peers (starting from the less important peers) until all the measures are withing the stated limits.
|
||||
|
||||
A connection manager first disconnects the peers with the least value. By default all peers have the same importance (1), but the application can define otherwise. Once a peer disconnects the connection manager discards the peer importance. (If necessary, the application should redefine the peer state if the peer is again connected).
|
||||
|
||||
|
||||
### Create a ConnectionManager
|
||||
|
||||
```js
|
||||
const libp2p = // …
|
||||
const options = {…}
|
||||
const connManager = new ConnManager(libp2p, options)
|
||||
```
|
||||
|
||||
Options is an optional object with the following key-value pairs:
|
||||
|
||||
* **`maxPeers`**: number identifying the maximum number of peers the current peer is willing to be connected to before is starts disconnecting. Defaults to `Infinity`
|
||||
* **`maxPeersPerProtocol`**: Object with key-value pairs, where a key is the protocol tag (case-insensitive) and the value is a number, representing the maximum number of peers allowing to connect for each protocol. Defaults to `{}`.
|
||||
* **`minPeers`**: number identifying the number of peers below which this node will not activate preemptive disconnections. Defaults to `0`.
|
||||
* **`maxData`**: sets the maximum data — in bytes per second - (sent and received) this node is willing to endure before it starts disconnecting peers. Defaults to `Infinity`.
|
||||
* **`maxSentData`**: sets the maximum sent data — in bytes per second - this node is willing to endure before it starts disconnecting peers. Defaults to `Infinity`.
|
||||
* **`maxReceivedData`**: sets the maximum received data — in bytes per second - this node is willing to endure before it starts disconnecting peers. Defaults to `Infinity`.
|
||||
* **`maxEventLoopDelay`**: sets the maximum event loop delay (measured in miliseconds) this node is willing to endure before it starts disconnecting peers. Defaults to `Infinity`.
|
||||
* **`pollInterval`**: sets the poll interval (in miliseconds) for assessing the current state and determining if this peer needs to force a disconnect. Defaults to `2000` (2 seconds).
|
||||
* **`movingAverageInterval`**: the interval used to calculate moving averages (in miliseconds). Defaults to `60000` (1 minute).
|
||||
* **`defaultPeerValue`**: number between 0 and 1. Defaults to 1.
|
||||
|
||||
|
||||
### `connManager.start()`
|
||||
|
||||
Starts the connection manager.
|
||||
|
||||
### `connManager.stop()`
|
||||
|
||||
Stops the connection manager.
|
||||
|
||||
|
||||
### `connManager.setPeerValue(peerId, value)`
|
||||
|
||||
Sets the peer value for a given peer id. This is used to sort peers (in reverse order of value) to determine which to disconnect from first.
|
||||
|
||||
Arguments:
|
||||
|
||||
* peerId: B58-encoded string or [`peer-id`](https://github.com/libp2p/js-peer-id) instance.
|
||||
* value: a number between 0 and 1, which represents a scale of how valuable this given peer id is to the application.
|
||||
|
||||
### `connManager.peers()`
|
||||
|
||||
Returns the peers this connection manager is connected to.
|
||||
|
||||
Returns an array of [PeerInfo](https://github.com/libp2p/js-peer-info).
|
||||
|
||||
### `connManager.emit('limit:exceeded', limitName, measured)`
|
||||
|
||||
Emitted when a limit is exceeded. Limit names can be:
|
||||
|
||||
* `maxPeers`
|
||||
* `minPeers`
|
||||
* `maxData`
|
||||
* `maxSentData`
|
||||
* `maxReceivedData`
|
||||
* `maxEventLoopDelay`
|
||||
* a protocol tag string (lower-cased)
|
||||
|
||||
|
||||
### `connManager.emit('disconnect:preemptive', peerId)`
|
||||
|
||||
Emitted when a peer is about to be preemptively disconnected.
|
||||
|
||||
### `connManager.emit('disconnected', peerId)`
|
||||
|
||||
Emitted when a peer is disconnected (preemptively or note). If this peer reconnects, you will need to reset it's value, since the connection manager does not remember it.
|
||||
|
||||
### `connManager.emit('connected', peerId: String)`
|
||||
|
||||
Emitted when a peer connects. This is a good event to set the peer value, so you can get some control over who gets banned once a maximum number of peers is reached.
|
||||
@ -1,12 +1,14 @@
|
||||
'use strict'
|
||||
|
||||
const EventEmitter = require('events')
|
||||
const assert = require('assert')
|
||||
const mergeOptions = require('merge-options')
|
||||
const LatencyMonitor = require('latency-monitor').default
|
||||
const debug = require('debug')('libp2p:connection-manager')
|
||||
const retimer = require('retimer')
|
||||
|
||||
const defaultOptions = {
|
||||
maxPeers: Infinity,
|
||||
minPeers: 0,
|
||||
maxConnections: Infinity,
|
||||
minConnections: 0,
|
||||
maxData: Infinity,
|
||||
maxSentData: Infinity,
|
||||
maxReceivedData: Infinity,
|
||||
@ -16,170 +18,171 @@ const defaultOptions = {
|
||||
defaultPeerValue: 1
|
||||
}
|
||||
|
||||
class ConnectionManager extends EventEmitter {
|
||||
class ConnectionManager {
|
||||
/**
|
||||
* @constructor
|
||||
* @param {Libp2p} libp2p
|
||||
* @param {object} options
|
||||
* @param {Number} options.maxConnections The maximum number of connections allowed. Default=Infinity
|
||||
* @param {Number} options.minConnections The minimum number of connections to avoid pruning. Default=0
|
||||
* @param {Number} options.maxData The max data (in and out), per average interval to allow. Default=Infinity
|
||||
* @param {Number} options.maxSentData The max outgoing data, per average interval to allow. Default=Infinity
|
||||
* @param {Number} options.maxReceivedData The max incoming data, per average interval to allow.. Default=Infinity
|
||||
* @param {Number} options.maxEventLoopDelay The upper limit the event loop can take to run. Default=Infinity
|
||||
* @param {Number} options.pollInterval How often, in milliseconds, metrics and latency should be checked. Default=2000
|
||||
* @param {Number} options.movingAverageInterval How often, in milliseconds, to compute averages. Default=60000
|
||||
* @param {Number} options.defaultPeerValue The value of the peer. Default=1
|
||||
*/
|
||||
constructor (libp2p, options) {
|
||||
super()
|
||||
this._libp2p = libp2p
|
||||
this._options = Object.assign({}, defaultOptions, options)
|
||||
this._options.maxPeersPerProtocol = fixMaxPeersPerProtocol(this._options.maxPeersPerProtocol)
|
||||
this._registrar = libp2p.registrar
|
||||
this._peerId = libp2p.peerInfo.id.toString()
|
||||
this._options = mergeOptions(defaultOptions, options)
|
||||
assert(
|
||||
this._options.maxConnections > this._options.minConnections,
|
||||
'Connection Manager maxConnections must be greater than minConnections'
|
||||
)
|
||||
|
||||
debug('options: %j', this._options)
|
||||
|
||||
this._stats = libp2p.stats
|
||||
if (options && !this._stats) {
|
||||
throw new Error('No libp2p.stats')
|
||||
}
|
||||
this._metrics = libp2p.metrics
|
||||
|
||||
this._peerValues = new Map()
|
||||
this._peers = new Map()
|
||||
this._peerProtocols = new Map()
|
||||
this._peerCountPerProtocol = new Map()
|
||||
this._onStatsUpdate = this._onStatsUpdate.bind(this)
|
||||
this._onPeerConnect = this._onPeerConnect.bind(this)
|
||||
this._onPeerDisconnect = this._onPeerDisconnect.bind(this)
|
||||
|
||||
if (this._libp2p.isStarted()) {
|
||||
this._onceStarted()
|
||||
} else {
|
||||
this._libp2p.once('start', this._onceStarted.bind(this))
|
||||
}
|
||||
this._connections = new Map()
|
||||
this._timer = null
|
||||
this._checkMetrics = this._checkMetrics.bind(this)
|
||||
}
|
||||
|
||||
/**
|
||||
* Starts the Connection Manager. If Metrics are not enabled on libp2p
|
||||
* only event loop and connection limits will be monitored.
|
||||
*/
|
||||
start () {
|
||||
this._stats.on('update', this._onStatsUpdate)
|
||||
this._libp2p.on('connection:start', this._onPeerConnect)
|
||||
this._libp2p.on('connection:end', this._onPeerDisconnect)
|
||||
if (this._metrics) {
|
||||
this._timer = this._timer || retimer(this._checkMetrics, this._options.pollInterval)
|
||||
}
|
||||
|
||||
// latency monitor
|
||||
this._latencyMonitor = new LatencyMonitor({
|
||||
latencyCheckIntervalMs: this._options.pollInterval,
|
||||
dataEmitIntervalMs: this._options.pollInterval
|
||||
})
|
||||
this._onLatencyMeasure = this._onLatencyMeasure.bind(this)
|
||||
this._latencyMonitor.on('data', this._onLatencyMeasure)
|
||||
debug('started')
|
||||
}
|
||||
|
||||
/**
|
||||
* Stops the Connection Manager
|
||||
*/
|
||||
stop () {
|
||||
this._stats.removeListener('update', this._onStatsUpdate)
|
||||
this._libp2p.removeListener('connection:start', this._onPeerConnect)
|
||||
this._libp2p.removeListener('connection:end', this._onPeerDisconnect)
|
||||
this._latencyMonitor.removeListener('data', this._onLatencyMeasure)
|
||||
this._timer && this._timer.clear()
|
||||
this._latencyMonitor && this._latencyMonitor.removeListener('data', this._onLatencyMeasure)
|
||||
debug('stopped')
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the value of the given peer. Peers with lower values
|
||||
* will be disconnected first.
|
||||
* @param {PeerId} peerId
|
||||
* @param {number} value A number between 0 and 1
|
||||
*/
|
||||
setPeerValue (peerId, value) {
|
||||
if (value < 0 || value > 1) {
|
||||
throw new Error('value should be a number between 0 and 1')
|
||||
}
|
||||
if (peerId.toB58String) {
|
||||
peerId = peerId.toB58String()
|
||||
if (peerId.toString) {
|
||||
peerId = peerId.toString()
|
||||
}
|
||||
this._peerValues.set(peerId, value)
|
||||
}
|
||||
|
||||
_onceStarted () {
|
||||
this._peerId = this._libp2p.peerInfo.id.toB58String()
|
||||
}
|
||||
|
||||
_onStatsUpdate () {
|
||||
const movingAvgs = this._stats.global.movingAverages
|
||||
const received = movingAvgs.dataReceived[this._options.movingAverageInterval].movingAverage()
|
||||
/**
|
||||
* Checks the libp2p metrics to determine if any values have exceeded
|
||||
* the configured maximums.
|
||||
* @private
|
||||
*/
|
||||
_checkMetrics () {
|
||||
const movingAverages = this._metrics.global.movingAverages
|
||||
const received = movingAverages.dataReceived[this._options.movingAverageInterval].movingAverage()
|
||||
this._checkLimit('maxReceivedData', received)
|
||||
const sent = movingAvgs.dataSent[this._options.movingAverageInterval].movingAverage()
|
||||
const sent = movingAverages.dataSent[this._options.movingAverageInterval].movingAverage()
|
||||
this._checkLimit('maxSentData', sent)
|
||||
const total = received + sent
|
||||
this._checkLimit('maxData', total)
|
||||
debug('stats update', total)
|
||||
debug('metrics update', total)
|
||||
this._timer.reschedule(this._options.pollInterval)
|
||||
}
|
||||
|
||||
_onPeerConnect (peerInfo) {
|
||||
const peerId = peerInfo.id.toB58String()
|
||||
debug('%s: connected to %s', this._peerId, peerId)
|
||||
this._peerValues.set(peerId, this._options.defaultPeerValue)
|
||||
this._peers.set(peerId, peerInfo)
|
||||
this.emit('connected', peerId)
|
||||
this._checkLimit('maxPeers', this._peers.size)
|
||||
|
||||
protocolsFromPeerInfo(peerInfo).forEach((protocolTag) => {
|
||||
const protocol = this._peerCountPerProtocol[protocolTag]
|
||||
if (!protocol) {
|
||||
this._peerCountPerProtocol[protocolTag] = 0
|
||||
}
|
||||
this._peerCountPerProtocol[protocolTag]++
|
||||
|
||||
let peerProtocols = this._peerProtocols[peerId]
|
||||
if (!peerProtocols) {
|
||||
peerProtocols = this._peerProtocols[peerId] = new Set()
|
||||
}
|
||||
peerProtocols.add(protocolTag)
|
||||
this._checkProtocolMaxPeersLimit(protocolTag, this._peerCountPerProtocol[protocolTag])
|
||||
})
|
||||
}
|
||||
|
||||
_onPeerDisconnect (peerInfo) {
|
||||
const peerId = peerInfo.id.toB58String()
|
||||
debug('%s: disconnected from %s', this._peerId, peerId)
|
||||
this._peerValues.delete(peerId)
|
||||
this._peers.delete(peerId)
|
||||
|
||||
const peerProtocols = this._peerProtocols[peerId]
|
||||
if (peerProtocols) {
|
||||
Array.from(peerProtocols).forEach((protocolTag) => {
|
||||
const peerCountForProtocol = this._peerCountPerProtocol[protocolTag]
|
||||
if (peerCountForProtocol) {
|
||||
this._peerCountPerProtocol[protocolTag]--
|
||||
}
|
||||
})
|
||||
/**
|
||||
* Tracks the incoming connection and check the connection limit
|
||||
* @param {Connection} connection
|
||||
*/
|
||||
onConnect (connection) {
|
||||
const peerId = connection.remotePeer.toString()
|
||||
this._connections.set(connection.id, connection)
|
||||
if (!this._peerValues.has(peerId)) {
|
||||
this._peerValues.set(peerId, this._options.defaultPeerValue)
|
||||
}
|
||||
|
||||
this.emit('disconnected', peerId)
|
||||
this._checkLimit('maxConnections', this._connections.size)
|
||||
}
|
||||
|
||||
/**
|
||||
* Removes the connection from tracking
|
||||
* @param {Connection} connection
|
||||
*/
|
||||
onDisconnect (connection) {
|
||||
this._connections.delete(connection.id)
|
||||
this._peerValues.delete(connection.remotePeer.toString())
|
||||
}
|
||||
|
||||
/**
|
||||
* If the event loop is slow, maybe close a connection
|
||||
* @private
|
||||
* @param {*} summary The LatencyMonitor summary
|
||||
*/
|
||||
_onLatencyMeasure (summary) {
|
||||
this._checkLimit('maxEventLoopDelay', summary.avgMs)
|
||||
}
|
||||
|
||||
/**
|
||||
* If the `value` of `name` has exceeded its limit, maybe close a connection
|
||||
* @private
|
||||
* @param {string} name The name of the field to check limits for
|
||||
* @param {number} value The current value of the field
|
||||
*/
|
||||
_checkLimit (name, value) {
|
||||
const limit = this._options[name]
|
||||
debug('checking limit of %s. current value: %d of %d', name, value, limit)
|
||||
if (value > limit) {
|
||||
debug('%s: limit exceeded: %s, %d', this._peerId, name, value)
|
||||
this.emit('limit:exceeded', name, value)
|
||||
this._maybeDisconnectOne()
|
||||
}
|
||||
}
|
||||
|
||||
_checkProtocolMaxPeersLimit (protocolTag, value) {
|
||||
debug('checking protocol limit. current value of %s is %d', protocolTag, value)
|
||||
const limit = this._options.maxPeersPerProtocol[protocolTag]
|
||||
if (value > limit) {
|
||||
debug('%s: protocol max peers limit exceeded: %s, %d', this._peerId, protocolTag, value)
|
||||
this.emit('limit:exceeded', protocolTag, value)
|
||||
this._maybeDisconnectOne()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* If we have more connections than our maximum, close a connection
|
||||
* to the lowest valued peer.
|
||||
* @private
|
||||
*/
|
||||
_maybeDisconnectOne () {
|
||||
if (this._options.minPeers < this._peerValues.size) {
|
||||
if (this._options.minConnections < this._connections.size) {
|
||||
const peerValues = Array.from(this._peerValues).sort(byPeerValue)
|
||||
debug('%s: sorted peer values: %j', this._peerId, peerValues)
|
||||
const disconnectPeer = peerValues[0]
|
||||
if (disconnectPeer) {
|
||||
const peerId = disconnectPeer[0]
|
||||
debug('%s: lowest value peer is %s', this._peerId, peerId)
|
||||
debug('%s: forcing disconnection from %j', this._peerId, peerId)
|
||||
this._disconnectPeer(peerId)
|
||||
debug('%s: closing a connection to %j', this._peerId, peerId)
|
||||
for (const connection of this._connections.values()) {
|
||||
if (connection.remotePeer.toString() === peerId) {
|
||||
connection.close()
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
_disconnectPeer (peerId) {
|
||||
debug('preemptively disconnecting peer', peerId)
|
||||
this.emit('%s: disconnect:preemptive', this._peerId, peerId)
|
||||
const peer = this._peers.get(peerId)
|
||||
this._libp2p.hangUp(peer, (err) => {
|
||||
if (err) {
|
||||
this.emit('error', err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = ConnectionManager
|
||||
@ -187,32 +190,3 @@ module.exports = ConnectionManager
|
||||
function byPeerValue (peerValueEntryA, peerValueEntryB) {
|
||||
return peerValueEntryA[1] - peerValueEntryB[1]
|
||||
}
|
||||
|
||||
function fixMaxPeersPerProtocol (maxPeersPerProtocol) {
|
||||
if (!maxPeersPerProtocol) {
|
||||
maxPeersPerProtocol = {}
|
||||
}
|
||||
|
||||
Object.keys(maxPeersPerProtocol).forEach((transportTag) => {
|
||||
const max = maxPeersPerProtocol[transportTag]
|
||||
delete maxPeersPerProtocol[transportTag]
|
||||
maxPeersPerProtocol[transportTag.toLowerCase()] = max
|
||||
})
|
||||
|
||||
return maxPeersPerProtocol
|
||||
}
|
||||
|
||||
function protocolsFromPeerInfo (peerInfo) {
|
||||
const protocolTags = new Set()
|
||||
peerInfo.multiaddrs.forEach((multiaddr) => {
|
||||
multiaddr.protos().map(protocolToProtocolTag).forEach((protocolTag) => {
|
||||
protocolTags.add(protocolTag)
|
||||
})
|
||||
})
|
||||
|
||||
return Array.from(protocolTags)
|
||||
}
|
||||
|
||||
function protocolToProtocolTag (protocol) {
|
||||
return protocol.name.toLowerCase()
|
||||
}
|
||||
|
||||
12
src/index.js
12
src/index.js
@ -15,6 +15,7 @@ const { getPeerInfo, getPeerInfoRemote } = require('./get-peer-info')
|
||||
const { validate: validateConfig } = require('./config')
|
||||
const { codes } = require('./errors')
|
||||
|
||||
const ConnectionManager = require('./connection-manager')
|
||||
const Circuit = require('./circuit')
|
||||
const Dialer = require('./dialer')
|
||||
const Metrics = require('./metrics')
|
||||
@ -63,6 +64,7 @@ class Libp2p extends EventEmitter {
|
||||
onConnection: (connection) => {
|
||||
const peerInfo = this.peerStore.put(new PeerInfo(connection.remotePeer))
|
||||
this.registrar.onConnect(peerInfo, connection)
|
||||
this.connectionManager.onConnect(connection)
|
||||
this.emit('peer:connect', peerInfo)
|
||||
|
||||
// Run identify for every connection
|
||||
@ -74,6 +76,7 @@ class Libp2p extends EventEmitter {
|
||||
onConnectionEnd: (connection) => {
|
||||
const peerInfo = getPeerInfo(connection.remotePeer)
|
||||
this.registrar.onDisconnect(peerInfo, connection)
|
||||
this.connectionManager.onDisconnect(connection)
|
||||
|
||||
// If there are no connections to the peer, disconnect
|
||||
if (!this.registrar.getConnection(peerInfo)) {
|
||||
@ -88,6 +91,9 @@ class Libp2p extends EventEmitter {
|
||||
this.handle = this.handle.bind(this)
|
||||
this.registrar.handle = this.handle
|
||||
|
||||
// Create the Connection Manager
|
||||
this.connectionManager = new ConnectionManager(this, this._options.connectionManager)
|
||||
|
||||
// Setup the transport manager
|
||||
this.transportManager = new TransportManager({
|
||||
libp2p: this,
|
||||
@ -208,6 +214,7 @@ class Libp2p extends EventEmitter {
|
||||
log('libp2p is stopping')
|
||||
|
||||
try {
|
||||
this.connectionManager.stop()
|
||||
await Promise.all([
|
||||
this.pubsub && this.pubsub.stop(),
|
||||
this._dht && this._dht.stop(),
|
||||
@ -225,6 +232,7 @@ class Libp2p extends EventEmitter {
|
||||
this.emit('error', err)
|
||||
}
|
||||
}
|
||||
this._isStarted = false
|
||||
log('libp2p has stopped')
|
||||
}
|
||||
|
||||
@ -290,7 +298,7 @@ class Libp2p extends EventEmitter {
|
||||
*/
|
||||
hangUp (peer) {
|
||||
return Promise.all(
|
||||
this.registrar.connections.get(peer.toB58String()).map(connection => {
|
||||
this.registrar.connections.get(peer.toString()).map(connection => {
|
||||
return connection.close()
|
||||
})
|
||||
)
|
||||
@ -378,6 +386,8 @@ class Libp2p extends EventEmitter {
|
||||
_onDidStart () {
|
||||
this._isStarted = true
|
||||
|
||||
this.connectionManager.start()
|
||||
|
||||
this.peerStore.on('peer', peerInfo => {
|
||||
this.emit('peer:discovery', peerInfo)
|
||||
this._maybeConnect(peerInfo)
|
||||
|
||||
@ -74,7 +74,7 @@ class Registrar {
|
||||
assert(PeerInfo.isPeerInfo(peerInfo), 'peerInfo must be an instance of peer-info')
|
||||
assert(Connection.isConnection(conn), 'conn must be an instance of interface-connection')
|
||||
|
||||
const id = peerInfo.id.toB58String()
|
||||
const id = peerInfo.id.toString()
|
||||
const storedConn = this.connections.get(id)
|
||||
|
||||
if (storedConn) {
|
||||
@ -95,7 +95,7 @@ class Registrar {
|
||||
onDisconnect (peerInfo, connection, error) {
|
||||
assert(PeerInfo.isPeerInfo(peerInfo), 'peerInfo must be an instance of peer-info')
|
||||
|
||||
const id = peerInfo.id.toB58String()
|
||||
const id = peerInfo.id.toString()
|
||||
let storedConn = this.connections.get(id)
|
||||
|
||||
if (storedConn && storedConn.length > 1) {
|
||||
@ -106,7 +106,7 @@ class Registrar {
|
||||
topology.disconnect(peerInfo, error)
|
||||
}
|
||||
|
||||
this.connections.delete(peerInfo.id.toB58String())
|
||||
this.connections.delete(peerInfo.id.toString())
|
||||
}
|
||||
}
|
||||
|
||||
@ -118,7 +118,7 @@ class Registrar {
|
||||
getConnection (peerInfo) {
|
||||
assert(PeerInfo.isPeerInfo(peerInfo), 'peerInfo must be an instance of peer-info')
|
||||
|
||||
const connections = this.connections.get(peerInfo.id.toB58String())
|
||||
const connections = this.connections.get(peerInfo.id.toString())
|
||||
// TODO: what should we return
|
||||
return connections ? connections[0] : null
|
||||
}
|
||||
|
||||
133
test/connection-manager/index.spec.js
Normal file
133
test/connection-manager/index.spec.js
Normal file
@ -0,0 +1,133 @@
|
||||
'use strict'
|
||||
/* eslint-env mocha */
|
||||
|
||||
const chai = require('chai')
|
||||
chai.use(require('dirty-chai'))
|
||||
chai.use(require('chai-as-promised'))
|
||||
const { expect } = chai
|
||||
const sinon = require('sinon')
|
||||
|
||||
const { createPeer } = require('../utils/creators/peer')
|
||||
const mockConnection = require('../utils/mockConnection')
|
||||
const baseOptions = require('../utils/base-options.browser')
|
||||
|
||||
describe('Connection Manager', () => {
|
||||
let libp2p
|
||||
|
||||
afterEach(async () => {
|
||||
sinon.restore()
|
||||
libp2p && await libp2p.stop()
|
||||
})
|
||||
|
||||
it('should be able to create without metrics', async () => {
|
||||
[libp2p] = await createPeer({
|
||||
config: {
|
||||
modules: baseOptions.modules
|
||||
},
|
||||
started: false
|
||||
})
|
||||
|
||||
const spy = sinon.spy(libp2p.connectionManager, 'start')
|
||||
|
||||
await libp2p.start()
|
||||
expect(spy).to.have.property('callCount', 1)
|
||||
expect(libp2p.connectionManager._metrics).to.not.exist()
|
||||
})
|
||||
|
||||
it('should be able to create with metrics', async () => {
|
||||
[libp2p] = await createPeer({
|
||||
config: {
|
||||
modules: baseOptions.modules,
|
||||
metrics: {
|
||||
enabled: true
|
||||
}
|
||||
},
|
||||
started: false
|
||||
})
|
||||
|
||||
const spy = sinon.spy(libp2p.connectionManager, 'start')
|
||||
|
||||
await libp2p.start()
|
||||
expect(spy).to.have.property('callCount', 1)
|
||||
expect(libp2p.connectionManager._metrics).to.exist()
|
||||
})
|
||||
|
||||
it('should close lowest value peer connection when the maximum has been reached', async () => {
|
||||
const max = 5
|
||||
;[libp2p] = await createPeer({
|
||||
config: {
|
||||
modules: baseOptions.modules,
|
||||
connectionManager: {
|
||||
maxConnections: max
|
||||
}
|
||||
},
|
||||
started: false
|
||||
})
|
||||
|
||||
await libp2p.start()
|
||||
|
||||
sinon.spy(libp2p.connectionManager, '_maybeDisconnectOne')
|
||||
|
||||
// Add 1 too many connections
|
||||
const spies = new Map()
|
||||
await Promise.all([...new Array(max + 1)].map(async (_, index) => {
|
||||
const connection = await mockConnection()
|
||||
const spy = sinon.spy(connection, 'close')
|
||||
// The connections have the same remote id, give them random ones
|
||||
// so that we can verify the correct connection was closed
|
||||
sinon.stub(connection.remotePeer, 'toString').returns(index)
|
||||
const value = Math.random()
|
||||
spies.set(value, spy)
|
||||
libp2p.connectionManager.setPeerValue(connection.remotePeer, value)
|
||||
libp2p.connectionManager.onConnect(connection)
|
||||
}))
|
||||
|
||||
// get the lowest value
|
||||
const lowest = Array.from(spies.keys()).sort()[0]
|
||||
const lowestSpy = spies.get(lowest)
|
||||
|
||||
expect(libp2p.connectionManager._maybeDisconnectOne).to.have.property('callCount', 1)
|
||||
expect(lowestSpy).to.have.property('callCount', 1)
|
||||
})
|
||||
|
||||
it('should close connection when the maximum has been reached even without peer values', async () => {
|
||||
const max = 5
|
||||
;[libp2p] = await createPeer({
|
||||
config: {
|
||||
modules: baseOptions.modules,
|
||||
connectionManager: {
|
||||
maxConnections: max
|
||||
}
|
||||
},
|
||||
started: false
|
||||
})
|
||||
|
||||
await libp2p.start()
|
||||
|
||||
sinon.spy(libp2p.connectionManager, '_maybeDisconnectOne')
|
||||
|
||||
// Add 1 too many connections
|
||||
const spy = sinon.spy()
|
||||
await Promise.all([...new Array(max + 1)].map(async () => {
|
||||
const connection = await mockConnection()
|
||||
sinon.stub(connection, 'close').callsFake(() => spy())
|
||||
libp2p.connectionManager.onConnect(connection)
|
||||
}))
|
||||
|
||||
expect(libp2p.connectionManager._maybeDisconnectOne).to.have.property('callCount', 1)
|
||||
expect(spy).to.have.property('callCount', 1)
|
||||
})
|
||||
|
||||
it('should fail if the connection manager has mismatched connection limit options', async () => {
|
||||
await expect(createPeer({
|
||||
config: {
|
||||
modules: baseOptions.modules,
|
||||
connectionManager: {
|
||||
maxConnections: 5,
|
||||
minConnections: 6
|
||||
}
|
||||
},
|
||||
started: false
|
||||
})).to.eventually.rejected('maxConnections must be greater')
|
||||
})
|
||||
})
|
||||
@ -33,9 +33,16 @@ describe('libp2p.metrics', () => {
|
||||
})
|
||||
|
||||
it('should start/stop metrics on startup/shutdown when enabled', async () => {
|
||||
const config = { ...baseOptions }
|
||||
config.metrics = {
|
||||
enabled: true
|
||||
const config = {
|
||||
...baseOptions,
|
||||
connectionManager: {
|
||||
movingAverageIntervals: [10]
|
||||
},
|
||||
metrics: {
|
||||
enabled: true,
|
||||
computeThrottleMaxQueueSize: 1, // compute after every message
|
||||
movingAverageIntervals: [10]
|
||||
}
|
||||
}
|
||||
;[libp2p] = await createPeer({ started: false, config })
|
||||
|
||||
@ -51,11 +58,16 @@ describe('libp2p.metrics', () => {
|
||||
})
|
||||
|
||||
it('should record metrics on connections and streams when enabled', async () => {
|
||||
const config = { ...baseOptions }
|
||||
config.metrics = {
|
||||
enabled: true,
|
||||
computeThrottleMaxQueueSize: 1, // compute after every message
|
||||
movingAverageIntervals: [10]
|
||||
const config = {
|
||||
...baseOptions,
|
||||
connectionManager: {
|
||||
movingAverageIntervals: [10]
|
||||
},
|
||||
metrics: {
|
||||
enabled: true,
|
||||
computeThrottleMaxQueueSize: 1, // compute after every message
|
||||
movingAverageIntervals: [10]
|
||||
}
|
||||
}
|
||||
let remoteLibp2p
|
||||
;[libp2p, remoteLibp2p] = await createPeer({ number: 2, config })
|
||||
@ -88,11 +100,16 @@ describe('libp2p.metrics', () => {
|
||||
})
|
||||
|
||||
it('should move disconnected peers to the old peers list', async () => {
|
||||
const config = { ...baseOptions }
|
||||
config.metrics = {
|
||||
enabled: true,
|
||||
computeThrottleMaxQueueSize: 1, // compute after every message
|
||||
movingAverageIntervals: [10]
|
||||
const config = {
|
||||
...baseOptions,
|
||||
connectionManager: {
|
||||
movingAverageIntervals: [10]
|
||||
},
|
||||
metrics: {
|
||||
enabled: true,
|
||||
computeThrottleMaxQueueSize: 1, // compute after every message
|
||||
movingAverageIntervals: [10]
|
||||
}
|
||||
}
|
||||
let remoteLibp2p
|
||||
;[libp2p, remoteLibp2p] = await createPeer({ number: 2, config })
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user