fluencelabs/js-libp2p
1
0
Fork 0
mirror of https://github.com/fluencelabs/js-libp2p synced 2025-06-15 02:01:21 +00:00
Code Issues Projects Releases Wiki Activity

feat: auto dial discovered peers (#349)

Browse Source
This commit is contained in:
Jacob Heun
2019-04-11 12:44:58 +02:00
committed by GitHub
parent 8b627797e2
commit 01aa44724e
12 changed files with 430 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.aegir.js
View File

@ -79,7 +79,7 @@ const after = (done) => {
}
module.exports = {
bundlesize: { maxSize: '217kB' },
bundlesize: { maxSize: '218kB' },
hooks: {
pre: before,
post: after

60
PEER_DISCOVERY.md Normal file
View File

@ -0,0 +1,60 @@
# Peer Discovery and Auto Dial
**Synopsis**:
* All peers discovered are emitted via `peer:discovery` so applications can take any desired action.
* Libp2p defaults to automatically connecting to new peers, when under the [ConnectionManager](https://github.com/libp2p/js-libp2p-connection-manager) low watermark (minimum peers).
* Applications can disable this via the `peerDiscovery.autoDial` config property, and handle connections themselves.
* Applications who have not disabled this should **never** connect on peer discovery. Applications should use the `peer:connect` event if they wish to take a specific action on new peers.
## Scenarios
In any scenario, if a peer is discovered it should be added to the PeerBook. This ensures that even if we don't dial to a node when we discover it, we know about it in the event that it becomes known as a provider for something we need. The scenarios listed below detail what actions the auto dialer will take when peers are discovered.
### 1. Joining the network
The node is new and needs to join the network. It currently has 0 peers.
**Discovery Mechanisms**: [Ambient Discovery](#ambient-discovery)
### Action to take
Connect to discovered peers. This should have some degree of concurrency limiting. While the case should be low, if we immediately discover more peers than our high watermark we should avoid dialing them all.
### 2. Connected to some
The node is connected to other nodes. The current number of connections is less than the desired low watermark.
**Discovery Mechanisms**: [Ambient Discovery](#ambient-discovery) and [Active Discovery](#active-discovery)
### Action to take
Connect to discovered peers. This should have some degree of concurrency limiting. The concurrency may need to be modified to reflect the current number of peers connected. The more peers we have, the lower the concurrency may need to be.
### 3. Connected to enough
**Discovery Mechanisms**: [Ambient Discovery](#ambient-discovery) and [Active Discovery](#active-discovery)
### Action to take
None. If we are connected to enough peers, the low watermark, we should not connect to discovered peers. As other peers discover us, they may connect to us based on their current scenario.
For example, a long running node with adequate peers is on an MDNS network. A new peer joins the network and both become aware of each other. The new peer should be the peer that dials, as it has too few peers. The existing node has no reason to dial the new peer, but should keep a record of it in case it later becomes an important node due to its contents/capabilities.
Avoiding dials above the low watermark also allows for a pool of connections to be reserved for application specific actions, such as connecting to a specific content provider via a DHT query to find that content (ipfs-bitswap).
### 4. Connected to too many
The node has more connections than it wants. The current number of connections is greater than the high watermark.
[WIP Connection Manager v2 spec](https://github.com/libp2p/specs/pull/161)
**Discovery Mechanisms**: [Ambient Discovery](#ambient-discovery) and [Active Discovery](#active-discovery)
### Action to take
None, the `ConnectionManager` will automatically prune connections.
## Discovery Mechanisms
Means of which a libp2p node discovers other peers.
### Active Discovery
Through active use of the libp2p network, a node may discovery peers.
* Content/Peer routing (DHT, delegated, etc) provider and peer queries
* DHT random walk
* Rendezvous servers
### Ambient Discovery
Leveraging known addresses, or network discovery mechanisms, a node may discover peers outside of the bounds of the libp2p network.
* Bootstrap
* MDNS
* proximity based (bluetooth, sound, etc)

16
README.md
View File

@ -160,6 +160,7 @@ class Node extends libp2p {
// libp2p config options (typically found on a config.json)
config: { // The config object is the part of the config that can go into a file, config.json.
peerDiscovery: {
autoDial: true, // Auto connect to discovered peers (limited by ConnectionManager minPeers)
mdns: { // mdns options
interval: 1000, // ms
enabled: true
@ -305,6 +306,9 @@ Required keys in the `options` object:
> Peer has been discovered.
If `autoDial` is `true`, applications should **not** attempt to connect to the peer
unless they are performing a specific action. See [peer discovery and auto dial](./PEER_DISCOVERY.md) for more information.
- `peer`: instance of [PeerInfo][]
##### `libp2p.on('peer:connect', (peer) => {})`
@ -509,18 +513,6 @@ Some available network protectors:
> npm run test:browser
```
#### Run interop tests
```sh
N/A
```
#### Run benchmark tests
```sh
N/A
```
### Packages
List of packages currently in existence for libp2p

2
package.json
View File

@ -45,7 +45,7 @@
"libp2p-connection-manager": "~0.0.2",
"libp2p-floodsub": "~0.15.8",
"libp2p-ping": "~0.8.5",
"libp2p-switch": "~0.42.7",
"libp2p-switch": "~0.42.8",
"libp2p-websockets": "~0.12.2",
"mafmt": "^6.0.7",
"multiaddr": "^6.0.6",

80
src/config.js
View File

@ -12,14 +12,7 @@ const transport = s.union([
}),
'function'
])
const optionsSchema = s(
{
connectionManager: 'object?',
datastore: 'object?',
peerInfo: 'object',
peerBook: 'object?',
modules: s({
const modulesSchema = s({
connEncryption: optional(list([s('object|function')])),
// this is hacky to simulate optional because interface doesnt work correctly with it
// change to optional when fixed upstream
@ -34,36 +27,73 @@ const optionsSchema = s(
return v > 0 ? true : 'ERROR_EMPTY'
}
})])
})
const configSchema = s({
peerDiscovery: s('object', {
autoDial: true
}),
config: s({
peerDiscovery: 'object?',
relay: s({
enabled: 'boolean',
hop: optional(s({
enabled: 'boolean',
active: 'boolean'
},
{ enabled: false, active: false }))
}, { enabled: true, hop: {} }),
}, {
// HOP defaults
enabled: false,
active: false
}))
}, {
// Relay defaults
enabled: true
}),
// DHT config
dht: s({
kBucketSize: 'number',
enabled: 'boolean?',
validators: 'object?',
selectors: 'object?',
randomWalk: optional(s({
enabled: 'boolean?', // disabled waiting for https://github.com/libp2p/js-libp2p-kad-dht/issues/86
enabled: 'boolean?',
queriesPerPeriod: 'number?',
interval: 'number?',
timeout: 'number?'
}, { enabled: false, queriesPerPeriod: 1, interval: 30000, timeout: 10000 })),
validators: 'object?',
selectors: 'object?'
}, { enabled: false, kBucketSize: 20, enabledDiscovery: false }),
}, {
// random walk defaults
enabled: false, // disabled waiting for https://github.com/libp2p/js-libp2p-kad-dht/issues/86
queriesPerPeriod: 1,
interval: 30000,
timeout: 10000
}))
}, {
// DHT defaults
enabled: false,
kBucketSize: 20,
enabledDiscovery: false
}),
// Experimental config
EXPERIMENTAL: s({
pubsub: 'boolean'
}, { pubsub: false })
}, { relay: {}, dht: {}, EXPERIMENTAL: {} })
},
{ config: {}, modules: {} }
)
}, {
// Experimental defaults
pubsub: false
})
}, {
relay: {},
dht: {},
EXPERIMENTAL: {}
})
const optionsSchema = s({
connectionManager: s('object', {
minPeers: 25
}),
datastore: 'object?',
peerInfo: 'object',
peerBook: 'object?',
modules: modulesSchema,
config: configSchema
})
module.exports.validate = (opts) => {
const [error, options] = optionsSchema.validate(opts)
@ -78,5 +108,9 @@ module.exports.validate = (opts) => {
}
}
if (options.config.peerDiscovery.autoDial === undefined) {
options.config.peerDiscovery.autoDial = true
}
return options
}

172
src/index.js
View File

@ -45,23 +45,23 @@ class Node extends EventEmitter {
super()
// validateConfig will ensure the config is correct,
// and add default values where appropriate
_options = validateConfig(_options)
this._options = validateConfig(_options)
this.datastore = _options.datastore
this.peerInfo = _options.peerInfo
this.peerBook = _options.peerBook || new PeerBook()
this.datastore = this._options.datastore
this.peerInfo = this._options.peerInfo
this.peerBook = this._options.peerBook || new PeerBook()
this._modules = _options.modules
this._config = _options.config
this._modules = this._options.modules
this._config = this._options.config
this._transport = [] // Transport instances/references
this._discovery = [] // Discovery service instances/references
// create the switch, and listen for errors
this._switch = new Switch(this.peerInfo, this.peerBook, _options.switch)
this._switch = new Switch(this.peerInfo, this.peerBook, this._options.switch)
this._switch.on('error', (...args) => this.emit('error', ...args))
this.stats = this._switch.stats
this.connectionManager = new ConnectionManager(this, _options.connectionManager)
this.connectionManager = new ConnectionManager(this, this._options.connectionManager)
// Attach stream multiplexers
if (this._modules.streamMuxer) {
@ -165,6 +165,16 @@ class Node extends EventEmitter {
log.error(err)
this.emit('error', err)
})
// Once we start, emit and dial any peers we may have already discovered
this.state.on('STARTED', () => {
this.peerBook.getAllArray().forEach((peerInfo) => {
this.emit('peer:discovery', peerInfo)
this._maybeConnect(peerInfo)
})
})
this._peerDiscovered = this._peerDiscovered.bind(this)
}
/**
@ -352,45 +362,21 @@ class Node extends EventEmitter {
this._switch.transport.add(ws.tag || ws.constructor.name, ws)
}
// all transports need to be setup before discover starts
if (this._modules.peerDiscovery) {
each(this._modules.peerDiscovery, (D, _cb) => {
let config = {}
// detect which multiaddrs we don't have a transport for and remove them
const multiaddrs = this.peerInfo.multiaddrs.toArray()
if (D.tag &&
this._config.peerDiscovery &&
this._config.peerDiscovery[D.tag]) {
config = this._config.peerDiscovery[D.tag]
multiaddrs.forEach((multiaddr) => {
if (!multiaddr.toString().match(/\/p2p-circuit($|\/)/) &&
!this._transport.find((transport) => transport.filter(multiaddr).length > 0)) {
this.peerInfo.multiaddrs.delete(multiaddr)
}
// If not configured to be enabled/disabled then enable by default
const enabled = config.enabled == null ? true : config.enabled
// If enabled then start it
if (enabled) {
let d
if (typeof D === 'function') {
d = new D(Object.assign({}, config, { peerInfo: this.peerInfo }))
} else {
d = D
}
d.on('peer', (peerInfo) => this.emit('peer:discovery', peerInfo))
this._discovery.push(d)
d.start(_cb)
} else {
_cb()
}
}, cb)
} else {
})
cb()
}
},
(cb) => {
if (this._dht) {
this._dht.start(() => {
this._dht.on('peer', (peerInfo) => this.emit('peer:discovery', peerInfo))
this._dht.on('peer', this._peerDiscovered)
cb()
})
} else {
@ -403,18 +389,14 @@ class Node extends EventEmitter {
}
cb()
},
// Peer Discovery
(cb) => {
// detect which multiaddrs we don't have a transport for and remove them
const multiaddrs = this.peerInfo.multiaddrs.toArray()
multiaddrs.forEach((multiaddr) => {
if (!multiaddr.toString().match(/\/p2p-circuit($|\/)/) &&
!this._transport.find((transport) => transport.filter(multiaddr).length > 0)) {
this.peerInfo.multiaddrs.delete(multiaddr)
}
})
if (this._modules.peerDiscovery) {
this._setupPeerDiscovery(cb)
} else {
cb()
}
}
], (err) => {
if (err) {
log.error(err)
@ -428,16 +410,17 @@ class Node extends EventEmitter {
_onStopping () {
series([
(cb) => {
if (this._modules.peerDiscovery) {
// stop all discoveries before continuing with shutdown
return parallel(
parallel(
this._discovery.map((d) => {
return (_cb) => d.stop(() => { _cb() })
d.removeListener('peer', this._peerDiscovered)
return (_cb) => d.stop((err) => {
log.error('an error occurred stopping the discovery service', err)
_cb()
})
}),
cb
)
}
cb()
},
(cb) => {
if (this._floodSub) {
@ -447,6 +430,7 @@ class Node extends EventEmitter {
},
(cb) => {
if (this._dht) {
this._dht.removeListener('peer', this._peerDiscovered)
return this._dht.stop(cb)
}
cb()
@ -468,6 +452,86 @@ class Node extends EventEmitter {
this.state('done')
})
}
/**
* Handles discovered peers. Each discovered peer will be emitted via
* the `peer:discovery` event. If auto dial is enabled for libp2p
* and the current connection count is under the low watermark, the
* peer will be dialed.
*
* TODO: If `peerBook.put` becomes centralized, https://github.com/libp2p/js-libp2p/issues/345,
* it would be ideal if only new peers were emitted. Currently, with
* other modules adding peers to the `PeerBook` we have no way of knowing
* if a peer is new or not, so it has to be emitted.
*
* @private
* @param {PeerInfo} peerInfo
*/
_peerDiscovered (peerInfo) {
peerInfo = this.peerBook.put(peerInfo)
if (!this.isStarted()) return
this.emit('peer:discovery', peerInfo)
this._maybeConnect(peerInfo)
}
/**
* Will dial to the given `peerInfo` if the current number of
* connected peers is less than the configured `ConnectionManager`
* minPeers.
* @private
* @param {PeerInfo} peerInfo
*/
_maybeConnect (peerInfo) {
// If auto dialing is on, check if we should dial
if (this._config.peerDiscovery.autoDial === true && !peerInfo.isConnected()) {
const minPeers = this._options.connectionManager.minPeers || 0
if (minPeers > Object.keys(this._switch.connection.connections).length) {
log('connecting to discovered peer')
this._switch.dialer.connect(peerInfo, (err) => {
err && log.error('could not connect to discovered peer', err)
})
}
}
}
/**
* Initializes and starts peer discovery services
*
* @private
* @param {function(Error)} callback
*/
_setupPeerDiscovery (callback) {
for (const DiscoveryService of this._modules.peerDiscovery) {
let config = {
enabled: true // on by default
}
if (DiscoveryService.tag &&
this._config.peerDiscovery &&
this._config.peerDiscovery[DiscoveryService.tag]) {
config = { ...config, ...this._config.peerDiscovery[DiscoveryService.tag] }
}
if (config.enabled) {
let discoveryService
if (typeof DiscoveryService === 'function') {
discoveryService = new DiscoveryService(Object.assign({}, config, { peerInfo: this.peerInfo }))
} else {
discoveryService = DiscoveryService
}
discoveryService.on('peer', this._peerDiscovered)
this._discovery.push(discoveryService)
}
}
each(this._discovery, (d, cb) => {
d.start(cb)
}, callback)
}
}
module.exports = Node

60
test/config.spec.js
View File

@ -58,6 +58,56 @@ describe('configuration', () => {
}).to.throw('ERROR_EMPTY')
})
it('should add defaults to config', () => {
const options = {
peerInfo,
modules: {
transport: [ WS ],
peerDiscovery: [ Bootstrap ],
dht: DHT
}
}
const expected = {
peerInfo,
connectionManager: {
minPeers: 25
},
modules: {
transport: [ WS ],
peerDiscovery: [ Bootstrap ],
dht: DHT
},
config: {
peerDiscovery: {
autoDial: true
},
EXPERIMENTAL: {
pubsub: false
},
dht: {
kBucketSize: 20,
enabled: false,
randomWalk: {
enabled: false,
queriesPerPeriod: 1,
interval: 30000,
timeout: 10000
}
},
relay: {
enabled: true,
hop: {
active: false,
enabled: false
}
}
}
}
expect(validateConfig(options)).to.deep.equal(expected)
})
it('should add defaults to missing items', () => {
const options = {
peerInfo,
@ -78,6 +128,9 @@ describe('configuration', () => {
const expected = {
peerInfo,
connectionManager: {
minPeers: 25
},
modules: {
transport: [ WS ],
peerDiscovery: [ Bootstrap ],
@ -85,6 +138,7 @@ describe('configuration', () => {
},
config: {
peerDiscovery: {
autoDial: true,
bootstrap: {
interval: 1000,
enabled: true
@ -180,6 +234,9 @@ describe('configuration', () => {
}
const expected = {
peerInfo,
connectionManager: {
minPeers: 25
},
modules: {
transport: [WS],
dht: DHT
@ -188,6 +245,9 @@ describe('configuration', () => {
EXPERIMENTAL: {
pubsub: false
},
peerDiscovery: {
autoDial: true
},
relay: {
enabled: true,
hop: {

61
test/peer-discovery.node.js
View File

@ -68,12 +68,17 @@ describe('peer discovery', () => {
(cb) => ss.stop(cb)
], done)
})
afterEach(() => {
sinon.restore()
})
}
describe('module registration', () => {
it('should enable by default a module passed as an object', (done) => {
const mockDiscovery = {
on: sinon.stub(),
removeListener: sinon.stub(),
start: sinon.stub().callsArg(0),
stop: sinon.stub().callsArg(0)
}
@ -94,6 +99,7 @@ describe('peer discovery', () => {
it('should enable by default a module passed as a function', (done) => {
const mockDiscovery = {
on: sinon.stub(),
removeListener: sinon.stub(),
start: sinon.stub().callsArg(0),
stop: sinon.stub().callsArg(0)
}
@ -116,6 +122,7 @@ describe('peer discovery', () => {
it('should enable module by configutation', (done) => {
const mockDiscovery = {
on: sinon.stub(),
removeListener: sinon.stub(),
start: sinon.stub().callsArg(0),
stop: sinon.stub().callsArg(0),
tag: 'mockDiscovery'
@ -151,6 +158,7 @@ describe('peer discovery', () => {
it('should disable module by configutation', (done) => {
const mockDiscovery = {
on: sinon.stub(),
removeListener: sinon.stub(),
start: sinon.stub().callsArg(0),
stop: sinon.stub().callsArg(0),
tag: 'mockDiscovery'
@ -186,6 +194,7 @@ describe('peer discovery', () => {
it('should register module passed as function', (done) => {
const mockDiscovery = {
on: sinon.stub(),
removeListener: sinon.stub(),
start: sinon.stub().callsArg(0),
stop: sinon.stub().callsArg(0)
}
@ -223,6 +232,7 @@ describe('peer discovery', () => {
it('should register module passed as object', (done) => {
const mockDiscovery = {
on: sinon.stub(),
removeListener: sinon.stub(),
start: sinon.stub().callsArg(0),
stop: sinon.stub().callsArg(0),
tag: 'mockDiscovery'
@ -256,9 +266,10 @@ describe('peer discovery', () => {
enabled: false
},
peerDiscovery: {
autoDial: true,
mdns: {
enabled: true,
interval: 1e3, // discover quickly
interval: 200, // discover quickly
// use a random tag to prevent CI collision
serviceTag: crypto.randomBytes(10).toString('hex')
}
@ -295,6 +306,7 @@ describe('peer discovery', () => {
enabled: false
},
peerDiscovery: {
autoDial: true,
webRTCStar: {
enabled: true
}
@ -331,9 +343,10 @@ describe('peer discovery', () => {
enabled: false
},
peerDiscovery: {
autoDial: true,
mdns: {
enabled: true,
interval: 1e3, // discovery quickly
interval: 200, // discovery quickly
// use a random tag to prevent CI collision
serviceTag: crypto.randomBytes(10).toString('hex')
},
@ -369,6 +382,7 @@ describe('peer discovery', () => {
setup({
config: {
peerDiscovery: {
autoDial: true,
mdns: {
enabled: false
},
@ -382,7 +396,7 @@ describe('peer discovery', () => {
randomWalk: {
enabled: true,
queriesPerPeriod: 1,
interval: 1000, // start the query sooner
interval: 200, // start the query sooner
timeout: 3000
}
}
@ -419,4 +433,45 @@ describe('peer discovery', () => {
})
})
})
describe('auto dial', () => {
setup({
connectionManager: {
minPeers: 1
},
config: {
peerDiscovery: {
autoDial: true,
mdns: {
enabled: false
},
webRTCStar: {
enabled: false
},
bootstrap: {
enabled: true,
list: []
}
},
dht: {
enabled: false
}
}
})
it('should only dial when the peer count is below the low watermark', (done) => {
const bootstrap = nodeA._discovery[0]
sinon.stub(nodeA._switch.dialer, 'connect').callsFake((peerInfo) => {
nodeA._switch.connection.connections[peerInfo.id.toB58String()] = []
})
bootstrap.emit('peer', nodeB.peerInfo)
bootstrap.emit('peer', nodeC.peerInfo)
// Only nodeB should get dialed
expect(nodeA._switch.dialer.connect.callCount).to.eql(1)
expect(nodeA._switch.dialer.connect.getCall(0).args[0]).to.eql(nodeB.peerInfo)
done()
})
})
})

30
test/transports.browser.js
View File

@ -4,6 +4,7 @@
const chai = require('chai')
chai.use(require('dirty-chai'))
chai.use(require('chai-checkmark'))
const expect = chai.expect
const PeerInfo = require('peer-info')
const PeerId = require('peer-id')
@ -413,29 +414,19 @@ describe('transports', () => {
it('node1 hangUp node2', (done) => {
node1.hangUp(peer2, (err) => {
expect(err).to.not.exist()
setTimeout(check, 500)
function check () {
const peers = node1.peerBook.getAll()
expect(Object.keys(peers)).to.have.length(1)
expect(node1._switch.connection.getAll()).to.have.length(0)
done()
}
})
})
it('create a third node and check that discovery works', function (done) {
this.timeout(10 * 1000)
let counter = 0
function check () {
if (++counter === 3) {
expect(node1._switch.connection.getAll()).to.have.length(1)
expect(node2._switch.connection.getAll()).to.have.length(1)
done()
}
}
const expectedPeers = [
node1.peerInfo.id.toB58String(),
node2.peerInfo.id.toB58String()
]
PeerId.create((err, id3) => {
expect(err).to.not.exist()
@ -444,13 +435,18 @@ describe('transports', () => {
const ma3 = '/ip4/127.0.0.1/tcp/14444/ws/p2p-websocket-star/p2p/' + id3.toB58String()
peer3.multiaddrs.add(ma3)
node1.on('peer:discovery', (peerInfo) => node1.dial(peerInfo, check))
node2.on('peer:discovery', (peerInfo) => node2.dial(peerInfo, check))
// 2 connects and 1 start
expect(3).checks(done)
const node3 = new Node({
peerInfo: peer3
})
node3.start(check)
node3.on('peer:connect', (peerInfo) => {
expect(expectedPeers).to.include(peerInfo.id.toB58String()).mark()
})
node3.start((err) => {
expect(err).to.not.exist().mark()
})
})
})
})

18
test/transports.node.js
View File

@ -438,6 +438,7 @@ describe('transports', () => {
},
config: {
peerDiscovery: {
autoDial: false,
[wstar.discovery.tag]: {
enabled: true
}
@ -452,7 +453,13 @@ describe('transports', () => {
},
(cb) => createNode([
'/ip4/0.0.0.0/tcp/0'
], (err, node) => {
], {
config: {
peerDiscovery: {
autoDial: false
}
}
}, (err, node) => {
expect(err).to.not.exist()
nodeTCP = node
node.handle('/echo/1.0.0', echo)
@ -460,7 +467,13 @@ describe('transports', () => {
}),
(cb) => createNode([
'/ip4/127.0.0.1/tcp/25022/ws'
], (err, node) => {
], {
config: {
peerDiscovery: {
autoDial: false
}
}
}, (err, node) => {
expect(err).to.not.exist()
nodeWS = node
node.handle('/echo/1.0.0', echo)
@ -479,6 +492,7 @@ describe('transports', () => {
},
config: {
peerDiscovery: {
autoDial: false,
[wstar.discovery.tag]: {
enabled: true
}

1
test/utils/bundle-browser.js
View File

@ -61,6 +61,7 @@ class Node extends libp2p {
},
config: {
peerDiscovery: {
autoDial: true,
webRTCStar: {
enabled: true
},

1
test/utils/bundle-nodejs.js
View File

@ -56,6 +56,7 @@ class Node extends libp2p {
},
config: {
peerDiscovery: {
autoDial: true,
mdns: {
interval: 10000,
enabled: false