Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

fix!: refactor connection manager to use a prioritised queue #1678

Merged
merged 9 commits into from
Apr 13, 2023
Merged

fix!: refactor connection manager to use a prioritised queue #1678

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
125878e
fix!: refactor connection manager to use a queue and add priority dia…
achingbrain Apr 6, 2023
3b66103
chore: consistency
achingbrain Apr 6, 2023
e2f406b
chore: apply suggestions from code review
achingbrain Apr 7, 2023
ee7cd99
fix: unhandled promise rejections in browsers
achingbrain Apr 7, 2023
6ff8ed5
chore: close superfluous connections when multiple dials succeed
achingbrain Apr 7, 2023
3d92f93
chore: restore interval, restore string allow/deny
achingbrain Apr 11, 2023
1e802d9
chore: update docs
achingbrain Apr 11, 2023
dec86ff
chore: update docs
achingbrain Apr 11, 2023
4c4ab3b
chore: update any-signal to fix memory leak
achingbrain Apr 13, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 5 additions & 51 deletions doc/CONFIGURATION.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,6 @@
- [Setup with Automatic Reservations](#setup-with-automatic-reservations)
- [Setup with Preconfigured Reservations](#setup-with-preconfigured-reservations)
- [Setup with Keychain](#setup-with-keychain)
- [Configuring Dialing](#configuring-dialing)
- [Configuring Connection Manager](#configuring-connection-manager)
- [Configuring Connection Gater](#configuring-connection-gater)
- [Outgoing connections](#outgoing-connections)
Expand Down Expand Up @@ -543,50 +542,12 @@ const node = await createLibp2p({
})
```

#### Configuring Dialing

Dialing in libp2p can be configured to limit the rate of dialing, and how long dials are allowed to take. The dialer configuration object should have the following properties:

| Name | Type | Description |
|------|------|-------------|
| maxParallelDials | `number` | How many multiaddrs we can dial in parallel. |
| maxAddrsToDial | `number` | How many multiaddrs is the dial allowed to dial for a single peer. |
| maxDialsPerPeer | `number` | How many multiaddrs we can dial per peer, in parallel. |
| dialTimeout | `number` | Second dial timeout per peer in ms. |
| resolvers | `object` | Dial [Resolvers](https://github.com/multiformats/js-multiaddr/blob/master/src/resolvers/index.js) for resolving multiaddrs |
| addressSorter | `(Array<Address>) => Array<Address>` | Sort the known addresses of a peer before trying to dial. |
| startupReconnectTimeout | `number` | When a node is restarted, we try to connect to any peers marked with the `keep-alive` tag up until to this timeout in ms is reached (default: 60000) |

The below configuration example shows how the dialer should be configured, with the current defaults:

```js
import { createLibp2p } from 'libp2p'
import { tcp } from '@libp2p/tcp'
import { mplex } from '@libp2p/mplex'
import { noise } from '@chainsafe/libp2p-noise'

import { dnsaddrResolver } from '@multiformats/multiaddr/resolvers'
import { publicAddressesFirst } from '@libp2p-utils/address-sort'
#### Configuring Connection Manager

const node = await createLibp2p({
transports: [tcp()],
streamMuxers: [mplex()],
connectionEncryption: [noise()],
dialer: {
maxParallelDials: 100,
maxAddrsToDial: 25,
maxDialsPerPeer: 4,
dialTimeout: 30e3,
resolvers: {
dnsaddr: dnsaddrResolver
},
addressSorter: publicAddressesFirst
}
```
The Connection Manager manages connections to peers in libp2p. It controls opening closing connections but also pruning connections when 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](https://libp2p.github.io/js-libp2p-interfaces/modules/_libp2p_interface_connection_manager.html). The configuration values below show the defaults for Connection Manager.

#### Configuring Connection Manager
See the [API docs](https://libp2p.github.io/js-libp2p/interfaces/index._internal_.ConnectionManagerConfig.html) for a full run list and discussion of all Connection Manager options.

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](https://libp2p.github.io/js-libp2p-interfaces/modules/_libp2p_interface_connection_manager.html). The configuration values below show the defaults for Connection Manager.

```js
import { createLibp2p } from 'libp2p'
Expand All @@ -600,14 +561,7 @@ const node = await createLibp2p({
connectionEncryption: [noise()],
connectionManager: {
maxConnections: Infinity,
minConnections: 0,
pollInterval: 2000,
achingbrain marked this conversation as resolved.
Show resolved Hide resolved
// The below values will only be taken into account when Metrics are enabled
maxData: Infinity,
achingbrain marked this conversation as resolved.
Show resolved Hide resolved
maxSentData: Infinity,
maxReceivedData: Infinity,
maxEventLoopDelay: Infinity,
movingAverageInterval: 60000
minConnections: 0
}
})
```
Expand Down Expand Up @@ -657,7 +611,7 @@ const node = await createLibp2p({
*
* Return true to prevent dialing the passed peer on the passed multiaddr.
*/
denyDialMultiaddr: (peerId: PeerId, multiaddr: Multiaddr) => Promise<boolean>
denyDialMultiaddr: (multiaddr: Multiaddr) => Promise<boolean>

/**
* denyInboundConnection tests whether an incipient inbound connection is allowed.
Expand Down
53 changes: 0 additions & 53 deletions doc/LIMITS.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,6 @@ This is important for [DoS](https://en.wikipedia.org/wiki/Denial-of-service_atta
- [Connection limits](#connection-limits)
- [Closing connections](#closing-connections)
- [Inbound connection threshold](#inbound-connection-threshold)
- [Data transfer and Event Loop limits](#data-transfer-and-event-loop-limits)
- [Stream limits](#stream-limits)
- [Mplex](#mplex)
- [Yamux](#yamux)
Expand Down Expand Up @@ -81,58 +80,6 @@ const node = await createLibp2pNode({
})
```

## Data transfer and Event Loop limits

If metrics are enabled the node will track the amount of data being sent to and from the network. If the rate of data sent is over the threshold connections will be trimmed to free up resources. The default rate is `Ininity` so this must be explicitly enabled.

Connections may also be trimmed if [event loop](https://nodejs.org/en/docs/guides/event-loop-timers-and-nexttick) latency exceeds the configured limit.

All fields are optional. The default values are defined in [src/connection-manager/index.ts](https://github.com/libp2p/js-libp2p/blob/master/src/connection-manager/index.ts) - please see that file for the current values.

```ts
const node = await createLibp2pNode({
metrics: {
enabled: true
},
connectionManager: {
/**
* If the node transfers more than this amount of data in bytes/second
* connections to untagged peers or those not in the allow list may be
* closed.
*
* It is bytes per second.
*/
maxData: number

/**
* If the node sends more than this amount of data in bytes/second
* connections to untagged peers or those not in the allow list may be
* closed.
*
* It is bytes per second.
*/
maxSentData: number

/**
* If the node receives more than this amount of data in bytes/second
* connections to untagged peers or those not in the allow list may be
* closed.
*
* It is bytes per second.
*/
maxReceivedData: number

/**
* If the event loop takes longer than this many ms to run, connections
* to untagged peers or those not in the allow list may be closed.
*
* It is milliseconds.
*/
maxEventLoopDelay: number
}
})
```

## Stream limits

libp2p stream multiplexers impose limits on the amount of streams that can be opened per connection, and also the amount of data that will be buffered for a given stream. The data should be consumed as fast as possible - if a stream's input buffer exceeds the limits set the stream will be reset.
Expand Down
103 changes: 103 additions & 0 deletions doc/migrations/v0.43-v0.44.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,103 @@
# Migrating to libp2p@44 <!-- omit in toc -->

A migration guide for refactoring your application code from libp2p v0.43.x to v0.44.0.

## Table of Contents <!-- omit in toc -->

- [Connection Manager](#connection-manager)
- [Connection Gater](#connection-gater)

## Connection Manager

The connection manager has been refactored to be simpler and faster. The changes
are mostly internal but some configuration options have changed.

**Before**

```js
import { createLibp2p } from 'libp2p'

const node = await createLibp2p({
connectionManager: {
// this has been renamed to `maxPeerAddrsToDial`
maxAddrsToDial: 20,
achingbrain marked this conversation as resolved.
Show resolved Hide resolved

// this has been renamed to `maxParallelDialsPerPeer`
maxDialsPerPeer: 20,
achingbrain marked this conversation as resolved.
Show resolved Hide resolved

// these should now be ranges expressed as MultiaddrFilters
allow: [
'/ip4/0.0.0.0/tcp/123'
],

// these should now be ranges expressed as MultiaddrFilters
deny: [
'/ip4/0.0.0.0/tcp/123'
]
Comment on lines +28 to +36
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we get a "allow peer" that allows any appropriate transports given a specific peerId?

I was trying to do that the other day in kubo and realized i needed a full multiaddr. I was trying to grant full-access to/from my helia node and couldn't figure out how to do it within an hour or so and i gave up.

It would be great to be able to do that in js-libp2p

Copy link
Member Author

@achingbrain achingbrain Apr 7, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm a little confused, access is allowed by default, it's only disallowed if a configured connection gater denys it or you're at maxConnections already.

The allow list above enables specific networks to breach connection limits - a DoS mitigation for when your node is being eclipsed.

It's almost certainly not the solution to your problem - maybe you can open an issue on the helia repo with a repro case?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

sure. I will re-open if I run into it.. essentially what I was trying to do was prefer my local node to others, and to allow connections to that one no matter what. it sounds like that may not be necessary

}
})
```

**After**

```js
import { createLibp2p } from 'libp2p'
import { MultiaddrFilter } from from '@multiformats/multiaddr'

const node = await createLibp2p({
connectionManager: {
// how many peers to dial at once while trying to ensure the node
// is above minConnections
autoDialConcurrency: 25,

// a low value allows user-initiated dials to take priority over
// auto dials
autoDialPriority: 0,
Comment on lines +53 to +55
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nice


// this was previously named `maxAddrsToDial`
maxPeerAddrsToDial: 20,
Comment on lines +57 to +58
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

PerConnection attempt? Or total(then we banlist the peer as bad??)

Might be good to call out which one and how in jsdoc, or in the name if possible.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Did you check the jsdoc for this config option? If it's not clear could you suggest an edit?


// this was previously named `maxDialsPerPeer`
maxParallelDialsPerPeer: 20
}
})
```

## Connection Gater

The `denyDialMultiaddr` method on the `ConnectionGater` interface no longer receives a peer id. This is to
support gating multiaddr dials for multiaddrs without peer ids.

If the peer id is known it will be present in the multiaddr.

**Before**

```js
import { createLibp2p } from 'libp2p'

const node = await createLibp2p({
connectionGater: {
denyDialMultiaddr: (peerId, multiaddr) => {
// allow/deny logic here
}
}
})
```

**After**

```js
import { createLibp2p } from 'libp2p'

const node = await createLibp2p({
connectionGater: {
denyDialMultiaddr: (multiaddr) => {
if (multiaddr.getPeerId() != null) {
// there is a peer id present in the multiaddr
}
Comment on lines +94 to +97
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

👍


// allow/deny logic here
}
}
})
```
4 changes: 2 additions & 2 deletions examples/chat/src/stream.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@ export function stdinToStream(stream) {
// Turn strings into buffers
(source) => map(source, (string) => uint8ArrayFromString(string)),
// Encode with length prefix (so receiving side knows how much data is coming)
lp.encode(),
(source) => lp.encode(source),
// Write to the stream (the sink)
stream.sink
)
Expand All @@ -26,7 +26,7 @@ export function streamToConsole(stream) {
// Read from the stream (the source)
stream.source,
// Decode length-prefixed data
lp.decode(),
(source) => lp.decode(source),
// Turn buffers into strings
(source) => map(source, (buf) => uint8ArrayToString(buf.subarray())),
// Sink function
Expand Down
6 changes: 3 additions & 3 deletions examples/delegated-routing/package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,11 +6,11 @@
"@chainsafe/libp2p-noise": "^11.0.0",
"ipfs-core": "^0.15.4",
"ipfs-http-client": "^58.0.1",
"libp2p": "../../",
"libp2p": "file:../../",
"@libp2p/delegated-content-routing": "^4.0.0",
"@libp2p/delegated-peer-routing": "^4.0.0",
"@libp2p/kad-dht": "^7.0.0",
"@libp2p/mplex": "^7.0.0",
"@libp2p/kad-dht": "^8.0.7",
"@libp2p/mplex": "^7.1.6",
"@libp2p/webrtc-star": "^6.0.0",
"@libp2p/websockets": "^5.0.0",
"react": "^17.0.2",
Expand Down
5 changes: 2 additions & 3 deletions examples/delegated-routing/src/libp2p-bundle.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,10 +22,9 @@ export default function Libp2pBundle ({peerInfo, peerBook}) {
return createLibp2p({
peerInfo,
peerBook,
// Lets limit the connection managers peers and have it check peer health less frequently
// Let's limit the number of connections the connection managers can have
connectionManager: {
maxPeers: 10,
pollInterval: 5000
maxConnections: 10
},
contentRouting: [
delegatedPeerRouting(client)
Expand Down
5 changes: 2 additions & 3 deletions examples/discovery-mechanisms/1.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@ import { tcp } from '@libp2p/tcp'
import { mplex } from '@libp2p/mplex'
import { noise } from '@chainsafe/libp2p-noise'
import { bootstrap } from '@libp2p/bootstrap'
import bootstrapers from './bootstrappers.js'
import bootstrappers from './bootstrappers.js'

;(async () => {
const node = await createLibp2p({
Expand All @@ -17,8 +17,7 @@ import bootstrapers from './bootstrappers.js'
connectionEncryption: [noise()],
peerDiscovery: [
bootstrap({
interval: 60e3,
list: bootstrapers
list: bootstrappers
})
]
})
Expand Down
4 changes: 2 additions & 2 deletions examples/libp2p-in-the-browser/package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,10 +11,10 @@
"dependencies": {
"@chainsafe/libp2p-noise": "^11.0.0",
"@libp2p/bootstrap": "^6.0.0",
"@libp2p/mplex": "^7.0.0",
"@libp2p/mplex": "^7.1.6",
"@libp2p/webrtc-star": "^6.0.0",
"@libp2p/websockets": "^5.0.0",
"libp2p": "../../"
"libp2p": "file:../../"
},
"devDependencies": {
"vite": "^2.8.6"
Expand Down
2 changes: 1 addition & 1 deletion examples/package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
"execa": "^6.1.0",
"fs-extra": "^10.1.0",
"it-to-buffer": "^3.0.1",
"libp2p": "../",
"libp2p": "file:../",
"p-defer": "^4.0.0",
"uint8arrays": "^4.0.0",
"which": "^2.0.1"
Expand Down
4 changes: 2 additions & 2 deletions examples/webrtc-direct/package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,8 +12,8 @@
"@libp2p/webrtc-direct": "^5.0.0",
"@chainsafe/libp2p-noise": "^11.0.0",
"@libp2p/bootstrap": "^5.0.0",
"@libp2p/mplex": "^7.0.0",
"libp2p": "../../",
"@libp2p/mplex": "^7.1.6",
"libp2p": "file:../../",
"wrtc": "^0.4.7"
},
"devDependencies": {
Expand Down
10 changes: 5 additions & 5 deletions package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,19 +136,19 @@
"@multiformats/mafmt": "^12.0.0",
"@multiformats/multiaddr": "^12.0.0",
"abortable-iterator": "^4.0.2",
"any-signal": "^3.0.0",
"any-signal": "^4.1.1",
"datastore-core": "^9.0.0",
"interface-datastore": "^8.0.0",
"it-all": "^2.0.0",
"it-drain": "^2.0.0",
"it-filter": "^2.0.0",
"it-first": "^2.0.0",
"it-parallel": "^3.0.0",
"it-handshake": "^4.1.2",
"it-length-prefixed": "^8.0.2",
"it-map": "^2.0.0",
"it-merge": "^2.0.0",
"it-pair": "^2.0.2",
"it-parallel": "^3.0.0",
"it-pb-stream": "^3.2.0",
"it-pipe": "^2.0.3",
"it-stream-types": "^1.0.4",
Expand All @@ -172,7 +172,7 @@
"devDependencies": {
"@chainsafe/libp2p-gossipsub": "^6.2.0",
"@chainsafe/libp2p-noise": "^11.0.0",
"@chainsafe/libp2p-yamux": "^3.0.3",
"@chainsafe/libp2p-yamux": "^3.0.8",
"@libp2p/bootstrap": "^6.0.0",
"@libp2p/daemon-client": "^5.0.0",
"@libp2p/daemon-server": "^4.1.0",
Expand All @@ -182,9 +182,9 @@
"@libp2p/interface-connection-encrypter-compliance-tests": "^4.0.0",
"@libp2p/interface-mocks": "^9.0.0",
"@libp2p/interop": "^7.0.0",
"@libp2p/kad-dht": "^8.0.0",
"@libp2p/kad-dht": "^8.0.7",
"@libp2p/mdns": "^7.0.0",
"@libp2p/mplex": "^7.0.0",
"@libp2p/mplex": "^7.1.6",
"@libp2p/pubsub": "^6.0.0",
"@libp2p/tcp": "^6.0.0",
"@libp2p/websockets": "^5.0.0",
Expand Down
Loading