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

fix: limit concurrent HTTP requests (backport for 0.2.x) #17

Merged
merged 3 commits into from
Jul 29, 2019
Merged

fix: limit concurrent HTTP requests (backport for 0.2.x) #17

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
200b25e
fix: limit concurrent HTTP requests
lidel Jul 26, 2019
91dcbb7
refactor: simplify callbacks
lidel Jul 26, 2019
a2e84f5
refactor: simplify callbacks
Jul 26, 2019
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
1 change: 1 addition & 0 deletions package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
"async": "^2.6.2",
"ipfs-http-client": "^33.0.2",
"multiaddr": "^6.1.0",
"p-queue": "^6.1.0",
"peer-id": "^0.12.2",
"peer-info": "^0.15.1"
},
Expand Down
51 changes: 19 additions & 32 deletions src/index.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,16 +4,14 @@ const dht = require('ipfs-http-client/src/dht')
const swarm = require('ipfs-http-client/src/swarm')
const refs = require('ipfs-http-client/src/files-regular/refs')
const defaultConfig = require('ipfs-http-client/src/utils/default-config')
const series = require('async/series')
const parallel = require('async/parallel')
const reflect = require('async/reflect')
const multiaddr = require('multiaddr')
const { default: PQueue } = require('p-queue')

const DEFAULT_MAX_TIMEOUT = 30e3 // 30 second default
const DEFAULT_IPFS_API = {
protocol: 'https',
port: 443,
host: 'ipfs.io'
host: 'node0.delegate.ipfs.io'
}

const DEFAULT_BOOSTRAP_NODES = [
Expand Down Expand Up @@ -50,6 +48,13 @@ class DelegatedContentRouting {

this.peerId = peerId
this.bootstrappers = bootstrappers || DEFAULT_BOOSTRAP_NODES.map((addr) => multiaddr(addr))

// limit concurrency to avoid request flood in web browser
// (backport of: https://github.com/libp2p/js-libp2p-delegated-content-routing/pull/16/)
this._httpQueue = new PQueue({ concurrency: 4 })
// sometimes refs requests take long time, they need separate queue
// to not suffocate regular bussiness
this._httpQueueRefs = new PQueue({ concurrency: 2 })
}

/**
Expand Down Expand Up @@ -77,46 +82,28 @@ class DelegatedContentRouting {

options.maxTimeout = options.maxTimeout || DEFAULT_MAX_TIMEOUT

this.dht.findProvs(key.toString(), {
timeout: `${options.maxTimeout}ms` // The api requires specification of the time unit (s/ms)
}, callback)
this._httpQueue.add(() =>
this.dht.findProvs(key.toString(), {
timeout: `${options.maxTimeout}ms` // The api requires specification of the time unit (s/ms)
}, callback)
).catch(callback)
Copy link
Member

Choose a reason for hiding this comment

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

I don't know exactly how promisify-es6 works under the hood, does it call callback as well as return a promise? I would have thought it did one or the other.

Maybe we can make this a bit easier to understand as well as eliminate any doubt?

    this._httpQueue.add(() =>
      this.dht.findProvs(key.toString(), {
        timeout: `${options.maxTimeout}ms` // The api requires specification of the time unit (s/ms)
      })
    ).then(res => callback(null, res), callback)

Copy link
Member Author

@lidel lidel Jul 26, 2019 •
edited
Loading

Choose a reason for hiding this comment

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

iiuc if callback is present, promisify-es6 will call function as-is; if not, it will return a Promise that injects callback as the last argument before calling the thing

promisify-es6/index.js#L31-L32

Should work the same, so I've switched to your version in 91dcbb7

}

/**
* Announce to the network that the delegated node can provide the given key.
*
* Currently this uses the following hack
* - call swarm.connect on the delegated node to us, to ensure we are connected
* - call refs --recursive on the delegated node, so it fetches the content
* - delegate is one of bootstrap nodes, so we are always connected to it
* - call refs on the delegated node, so it fetches the content
*
* @param {CID} key
* @param {function(Error)} callback
* @returns {void}
*/
provide (key, callback) {
const addrs = this.bootstrappers.map((addr) => {
return addr.encapsulate(`/p2p-circuit/ipfs/${this.peerId.toB58String()}`)
})

series([
(cb) => parallel(addrs.map((addr) => {
return reflect((cb) => this.swarm.connect(addr.toString(), cb))
}), (err, results) => {
if (err) {
return cb(err)
}

// only some need to succeed
const success = results.filter((res) => res.error == null)
if (success.length === 0) {
return cb(new Error('unable to swarm.connect using p2p-circuit'))
}
cb()
}),
(cb) => {
this.refs(key.toString(), { recursive: true }, cb)
}
], (err) => callback(err))
this._httpQueueRefs.add(() =>
this.refs(key.toString(), { recursive: false }, (err, res) => callback(err))
).catch(callback)
Copy link
Member

Choose a reason for hiding this comment

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

Same here!

Copy link
Member Author

Choose a reason for hiding this comment

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

That's odd: last test (provide) stop passing if I move callback call to the outside:

    this._httpQueueRefs.add(() =>
      this.refs(key.toString(), { recursive: false })
    ).then(res => callback(null, res), callback)

makes tests hang:

  6 passing (22s)
  1 failing

  1) DelegatedContentRouting
       provide
         should be able to register as a content provider to the delegate node:
     Error: Timeout of 20000ms exceeded. For async tests and hooks, ensure "done()" is called; if returning a Promise, ensure it resolves. (/home/lidel/project/js-libp2p-delegated-content-routing/test/index.spec.js)

Bit late on my end, so I can't tell if it is a bug somewhere, or just me being tired.
Thoughts?

}
}

Expand Down
4 changes: 2 additions & 2 deletions test/index.spec.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,14 +88,14 @@ describe('DelegatedContentRouting', function () {
expect(() => new DelegatedContentRouting()).to.throw()
})

it('should default to https://ipfs.io as the delegate', () => {
it('should default to https://node0.delegate.ipfs.io as the delegate', () => {
const router = new DelegatedContentRouting(selfId)

expect(router.api).to.include({
'api-path': '/api/v0/',
protocol: 'https',
port: 443,
host: 'ipfs.io'
host: 'node0.delegate.ipfs.io'
})
})

Expand Down