Merge pull request #129 from mapbox/129-pending_requests_hanging
Map destruction hanging with pending requests
This commit is contained in:
@@ -4,6 +4,8 @@
|
||||
|
||||
#include <mbgl/storage/request.hpp>
|
||||
|
||||
#include <algorithm>
|
||||
|
||||
namespace node_mbgl {
|
||||
|
||||
////////////////////////////////////////////////////////////////////////////////////////////////
|
||||
@@ -42,15 +44,15 @@ NAN_METHOD(NodeFileSource::New) {
|
||||
|
||||
struct NodeFileSource::Action {
|
||||
const enum : bool { Add, Cancel } type;
|
||||
mbgl::Request *const request;
|
||||
mbgl::Resource const resource;
|
||||
};
|
||||
|
||||
NodeFileSource::NodeFileSource() :
|
||||
queue(new Queue(uv_default_loop(), [this](Action &action) {
|
||||
if (action.type == Action::Add) {
|
||||
processAdd(action.request);
|
||||
processAdd(action.resource);
|
||||
} else if (action.type == Action::Cancel) {
|
||||
processCancel(action.request);
|
||||
processCancel(action.resource);
|
||||
}
|
||||
}))
|
||||
{
|
||||
@@ -66,21 +68,50 @@ NodeFileSource::~NodeFileSource() {
|
||||
mbgl::Request* NodeFileSource::request(const mbgl::Resource& resource, uv_loop_t* loop, Callback callback) {
|
||||
auto req = new mbgl::Request(resource, loop, std::move(callback));
|
||||
|
||||
// This function can be called from any thread. Make sure we're executing the actual call in the
|
||||
// file source loop by sending it over the queue. It will be processed in processAction().
|
||||
queue->send(Action{ Action::Add, req });
|
||||
std::lock_guard<std::mutex> lock(observersMutex);
|
||||
auto it = observers.find(resource);
|
||||
if (it == observers.end()) {
|
||||
observers[resource] = { req };
|
||||
|
||||
// This function can be called from any thread. Make sure we're executing the actual call in the
|
||||
// file source loop by sending it over the queue. It will be processed in processAction().
|
||||
queue->send(Action{ Action::Add, resource });
|
||||
} else {
|
||||
it->second.emplace_back(req);
|
||||
}
|
||||
|
||||
return req;
|
||||
}
|
||||
|
||||
void NodeFileSource::cancel(mbgl::Request* req) {
|
||||
req->cancel();
|
||||
|
||||
// This function can be called from any thread. Make sure we're executing the actual call in the
|
||||
// file source loop by sending it over the queue. It will be processed in processAction().
|
||||
queue->send(Action{ Action::Cancel, req });
|
||||
std::lock_guard<std::mutex> lock(observersMutex);
|
||||
|
||||
auto observersIter = observers.find(req->resource);
|
||||
if (observersIter == observers.end()) {
|
||||
return;
|
||||
}
|
||||
|
||||
auto& observersList = observersIter->second;
|
||||
auto observersListIter = std::find(observersList.begin(), observersList.end(), req);
|
||||
if (observersListIter == observersList.end()) {
|
||||
return;
|
||||
}
|
||||
|
||||
observersList.erase(observersListIter);
|
||||
if (observersList.empty()) {
|
||||
observers.erase(observersIter);
|
||||
|
||||
// This function can be called from any thread. Make sure we're executing the actual call in the
|
||||
// file source loop by sending it over the queue. It will be processed in processAction().
|
||||
queue->send(Action{ Action::Cancel, req->resource });
|
||||
}
|
||||
|
||||
req->destruct();
|
||||
}
|
||||
|
||||
void NodeFileSource::processAdd(mbgl::Request *req) {
|
||||
void NodeFileSource::processAdd(const mbgl::Resource& resource) {
|
||||
NanScope();
|
||||
|
||||
// Make sure the loop stays alive as long as request is pending.
|
||||
@@ -89,35 +120,29 @@ void NodeFileSource::processAdd(mbgl::Request *req) {
|
||||
}
|
||||
|
||||
auto handle = NanObjectWrapHandle(this);
|
||||
auto requestHandle = NanNew<v8::Object>(NodeRequest::Create(handle, req));
|
||||
#if (NODE_MODULE_VERSION > NODE_0_10_MODULE_VERSION)
|
||||
const v8::UniquePersistent<v8::Object> requestPersistent(v8::Isolate::GetCurrent(), requestHandle);
|
||||
#else
|
||||
auto requestHandle = NanNew<v8::Object>(NodeRequest::Create(handle, resource));
|
||||
|
||||
v8::Persistent<v8::Object> requestPersistent;
|
||||
NanAssignPersistent(requestPersistent, requestHandle);
|
||||
#endif
|
||||
pending.emplace(req, std::move(requestPersistent));
|
||||
pending.emplace(resource, std::move(requestPersistent));
|
||||
|
||||
v8::Local<v8::Value> argv[] = { requestHandle };
|
||||
NanMakeCallback(handle, NanNew("request"), 1, argv);
|
||||
}
|
||||
|
||||
void NodeFileSource::processCancel(mbgl::Request *req) {
|
||||
void NodeFileSource::processCancel(const mbgl::Resource& resource) {
|
||||
NanScope();
|
||||
|
||||
auto it = pending.find(req);
|
||||
auto it = pending.find(resource);
|
||||
if (it == pending.end()) {
|
||||
// The response callback was already fired. There is no point in calling the cancelation
|
||||
// callback because the request is already completed.
|
||||
} else {
|
||||
#if (NODE_MODULE_VERSION > NODE_0_10_MODULE_VERSION)
|
||||
auto requestHandle = v8::Local<v8::Object>::New(v8::Isolate::GetCurrent(), it->second);
|
||||
it->second.Reset();
|
||||
#else
|
||||
auto requestHandle = NanNew<v8::Object>(it->second);
|
||||
#endif
|
||||
|
||||
// Dispose and remove the persistent handle
|
||||
#if (NODE_MODULE_VERSION <= NODE_0_10_MODULE_VERSION)
|
||||
NanDisposePersistent(it->second);
|
||||
#endif
|
||||
pending.erase(it);
|
||||
@@ -136,16 +161,15 @@ void NodeFileSource::processCancel(mbgl::Request *req) {
|
||||
// Set the request handle in the request wrapper handle to null
|
||||
ObjectWrap::Unwrap<NodeRequest>(requestHandle)->cancel();
|
||||
}
|
||||
|
||||
// Finally, destruct the request object
|
||||
req->destruct();
|
||||
}
|
||||
|
||||
void NodeFileSource::notify(mbgl::Request *req, const std::shared_ptr<const mbgl::Response>& response) {
|
||||
void NodeFileSource::notify(const mbgl::Resource& resource, const std::shared_ptr<const mbgl::Response>& response) {
|
||||
// First, remove the request, since it might be destructed at any point now.
|
||||
auto it = pending.find(req);
|
||||
auto it = pending.find(resource);
|
||||
if (it != pending.end()) {
|
||||
#if (NODE_MODULE_VERSION <= NODE_0_10_MODULE_VERSION)
|
||||
#if (NODE_MODULE_VERSION > NODE_0_10_MODULE_VERSION)
|
||||
it->second.Reset();
|
||||
#else
|
||||
NanDisposePersistent(it->second);
|
||||
#endif
|
||||
pending.erase(it);
|
||||
@@ -156,7 +180,23 @@ void NodeFileSource::notify(mbgl::Request *req, const std::shared_ptr<const mbgl
|
||||
}
|
||||
}
|
||||
|
||||
req->notify(response);
|
||||
std::vector<mbgl::Request*> observersList;
|
||||
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(observersMutex);
|
||||
|
||||
auto observersIter = observers.find(resource);
|
||||
if (observersIter == observers.end()) {
|
||||
return;
|
||||
}
|
||||
|
||||
observersList.swap(observersIter->second);
|
||||
observers.erase(observersIter);
|
||||
}
|
||||
|
||||
for (auto request : observersList) {
|
||||
request->notify(response);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -10,8 +10,10 @@
|
||||
#include <nan.h>
|
||||
#pragma GCC diagnostic pop
|
||||
|
||||
#include <map>
|
||||
#include <memory>
|
||||
#include <mutex>
|
||||
#include <unordered_map>
|
||||
#include <vector>
|
||||
|
||||
namespace node_mbgl {
|
||||
|
||||
@@ -36,21 +38,32 @@ public:
|
||||
void cancel(mbgl::Request*);
|
||||
|
||||
// visiblity?
|
||||
void notify(mbgl::Request*, const std::shared_ptr<const mbgl::Response>&);
|
||||
void notify(const mbgl::Resource&, const std::shared_ptr<const mbgl::Response>&);
|
||||
|
||||
private:
|
||||
struct Action;
|
||||
using Queue = util::AsyncQueue<Action>;
|
||||
|
||||
void processAdd(mbgl::Request*);
|
||||
void processCancel(mbgl::Request*);
|
||||
void processAdd(const mbgl::Resource&);
|
||||
void processCancel(const mbgl::Resource&);
|
||||
|
||||
private:
|
||||
#if (NODE_MODULE_VERSION > NODE_0_10_MODULE_VERSION)
|
||||
std::map<mbgl::Request*, const v8::UniquePersistent<v8::Object> &> pending;
|
||||
std::unordered_map<mbgl::Resource, v8::Persistent<v8::Object, v8::CopyablePersistentTraits<v8::Object>>, mbgl::Resource::Hash> pending;
|
||||
#else
|
||||
std::map<mbgl::Request*, v8::Persistent<v8::Object>> pending;
|
||||
std::unordered_map<mbgl::Resource, v8::Persistent<v8::Object>, mbgl::Resource::Hash> pending;
|
||||
#endif
|
||||
|
||||
// The observers list will hold pointers to all the requests waiting
|
||||
// for a particular resource. It is also used for coalescing requests,
|
||||
// so we don't ask for the same resources twice. The access must be
|
||||
// guarded by a mutex because the list is also accessed by a thread
|
||||
// from the mbgl::Map object and from the main thread when notifying
|
||||
// requests of completion. Concurrent access is specially needed when
|
||||
// canceling a request to avoid a deadlock (see #129).
|
||||
std::unordered_map<mbgl::Resource, std::vector<mbgl::Request*>, mbgl::Resource::Hash> observers;
|
||||
std::mutex observersMutex;
|
||||
|
||||
Queue *queue = nullptr;
|
||||
};
|
||||
|
||||
|
||||
@@ -35,35 +35,34 @@ NAN_METHOD(NodeRequest::New) {
|
||||
return NanThrowTypeError("Cannot create Request objects explicitly");
|
||||
}
|
||||
|
||||
auto request = reinterpret_cast<mbgl::Request *>(args[1].As<v8::External>()->Value());
|
||||
auto req = new NodeRequest(args[0]->ToObject(), request);
|
||||
auto resource = reinterpret_cast<mbgl::Resource*>(args[1].As<v8::External>()->Value());
|
||||
auto req = new NodeRequest(args[0]->ToObject(), *resource);
|
||||
req->Wrap(args.This());
|
||||
|
||||
NanReturnValue(args.This());
|
||||
}
|
||||
|
||||
v8::Handle<v8::Object> NodeRequest::Create(v8::Handle<v8::Object> source, mbgl::Request *request) {
|
||||
v8::Handle<v8::Object> NodeRequest::Create(v8::Handle<v8::Object> source, const mbgl::Resource& resource) {
|
||||
NanEscapableScope();
|
||||
|
||||
v8::Local<v8::Value> argv[] = { NanNew<v8::Object>(source), NanNew<v8::External>(request) };
|
||||
v8::Local<v8::Value> argv[] = { NanNew<v8::Object>(source),
|
||||
NanNew<v8::External>(const_cast<mbgl::Resource*>(&resource)) };
|
||||
auto instance = NanNew<v8::FunctionTemplate>(constructorTemplate)->GetFunction()->NewInstance(2, argv);
|
||||
|
||||
instance->ForceSet(NanNew("url"), NanNew(request->resource.url), v8::ReadOnly);
|
||||
instance->ForceSet(NanNew("kind"), NanNew<v8::Integer>(int(request->resource.kind)), v8::ReadOnly);
|
||||
instance->ForceSet(NanNew("url"), NanNew(resource.url), v8::ReadOnly);
|
||||
instance->ForceSet(NanNew("kind"), NanNew<v8::Integer>(int(resource.kind)), v8::ReadOnly);
|
||||
|
||||
return NanEscapeScope(instance);
|
||||
}
|
||||
|
||||
NAN_METHOD(NodeRequest::Respond) {
|
||||
auto nodeRequest = ObjectWrap::Unwrap<NodeRequest>(args.Holder());
|
||||
if (!nodeRequest->request) {
|
||||
if (!nodeRequest->resource) {
|
||||
return NanThrowError("Request has already been responded to, or was canceled.");
|
||||
}
|
||||
|
||||
auto source = ObjectWrap::Unwrap<NodeFileSource>(NanNew<v8::Object>(nodeRequest->source));
|
||||
|
||||
auto request = nodeRequest->request;
|
||||
nodeRequest->request = nullptr;
|
||||
auto resource = std::move(nodeRequest->resource);
|
||||
|
||||
if (args.Length() < 1) {
|
||||
return NanThrowTypeError("First argument must be an error object");
|
||||
@@ -76,7 +75,7 @@ NAN_METHOD(NodeRequest::Respond) {
|
||||
const NanUtf8String message { args[0]->ToString() };
|
||||
response->message = std::string { *message, size_t(message.length()) };
|
||||
|
||||
source->notify(request, response);
|
||||
source->notify(*resource, response);
|
||||
} else if (args.Length() < 2 || !args[1]->IsObject()) {
|
||||
return NanThrowTypeError("Second argument must be a response object");
|
||||
} else {
|
||||
@@ -120,7 +119,7 @@ NAN_METHOD(NodeRequest::Respond) {
|
||||
}
|
||||
|
||||
// Send the response object to the NodeFileSource object
|
||||
source->notify(request, response);
|
||||
source->notify(*resource, response);
|
||||
}
|
||||
|
||||
NanReturnUndefined();
|
||||
@@ -129,8 +128,8 @@ NAN_METHOD(NodeRequest::Respond) {
|
||||
////////////////////////////////////////////////////////////////////////////////////////////////
|
||||
// Instance
|
||||
|
||||
NodeRequest::NodeRequest(v8::Local<v8::Object> source_, mbgl::Request *request_)
|
||||
: request(request_) {
|
||||
NodeRequest::NodeRequest(v8::Local<v8::Object> source_, const mbgl::Resource& resource_)
|
||||
: resource(std::make_unique<mbgl::Resource>(resource_)) {
|
||||
NanAssignPersistent(source, source_);
|
||||
}
|
||||
|
||||
@@ -139,7 +138,7 @@ NodeRequest::~NodeRequest() {
|
||||
}
|
||||
|
||||
void NodeRequest::cancel() {
|
||||
request = nullptr;
|
||||
resource.reset();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -7,8 +7,11 @@
|
||||
#include <nan.h>
|
||||
#pragma GCC diagnostic pop
|
||||
|
||||
#include <mbgl/storage/resource.hpp>
|
||||
|
||||
#include <memory>
|
||||
|
||||
class NodeFileSource;
|
||||
namespace mbgl { class Request; }
|
||||
|
||||
namespace node_mbgl {
|
||||
|
||||
@@ -20,21 +23,21 @@ public:
|
||||
static NAN_METHOD(New);
|
||||
static NAN_METHOD(Respond);
|
||||
|
||||
static v8::Handle<v8::Object> Create(v8::Handle<v8::Object> source, mbgl::Request *request);
|
||||
static v8::Handle<v8::Object> Create(v8::Handle<v8::Object> source, const mbgl::Resource& resource);
|
||||
|
||||
static v8::Persistent<v8::FunctionTemplate> constructorTemplate;
|
||||
|
||||
////////////////////////////////////////////////////////////////////////////////////////////////
|
||||
// Instance
|
||||
public:
|
||||
NodeRequest(v8::Local<v8::Object> source, mbgl::Request *request);
|
||||
NodeRequest(v8::Local<v8::Object> source, const mbgl::Resource& resource);
|
||||
~NodeRequest();
|
||||
|
||||
void cancel();
|
||||
|
||||
private:
|
||||
v8::Persistent<v8::Object> source;
|
||||
mbgl::Request *request = nullptr;
|
||||
std::unique_ptr<mbgl::Resource> resource;
|
||||
};
|
||||
|
||||
}
|
||||
|
||||
@@ -104,7 +104,7 @@ test('gzip', function(t) {
|
||||
mbgl.once('message', function(msg) {
|
||||
if (msg.severity == 'ERROR') {
|
||||
t.ok(msg, 'emits error');
|
||||
t.equal(msg.class, 'ResourceLoader');
|
||||
t.equal(msg.class, 'Style');
|
||||
t.equal(msg.severity, 'ERROR');
|
||||
t.ok(msg.text.match(/pbf unknown field type exception/), 'error text matches');
|
||||
}
|
||||
|
||||
@@ -132,11 +132,7 @@ test('Map', function(t) {
|
||||
});
|
||||
});
|
||||
|
||||
t.skip('accepts a JSON stylesheet', { timeout: 1000 }, function(t) {
|
||||
mbgl.once('message', function(msg) {
|
||||
console.log(msg);
|
||||
});
|
||||
|
||||
t.test('accepts a JSON stylesheet', { timeout: 1000 }, function(t) {
|
||||
setup(fileSource, function(map) {
|
||||
t.doesNotThrow(function() {
|
||||
map.load(style);
|
||||
@@ -148,11 +144,7 @@ test('Map', function(t) {
|
||||
});
|
||||
});
|
||||
|
||||
t.skip('accepts a stringified stylesheet', { timeout: 1000 }, function(t) {
|
||||
mbgl.once('message', function(msg) {
|
||||
console.log(msg);
|
||||
});
|
||||
|
||||
t.test('accepts a stringified stylesheet', { timeout: 1000 }, function(t) {
|
||||
setup(fileSource, function(map) {
|
||||
t.doesNotThrow(function() {
|
||||
map.load(JSON.stringify(style));
|
||||
@@ -167,6 +159,144 @@ test('Map', function(t) {
|
||||
t.end();
|
||||
});
|
||||
|
||||
t.test('request coalesting', function(t) {
|
||||
var requestList = [];
|
||||
var fileSourceDoNotReply = new mbgl.FileSource();
|
||||
|
||||
// This file source will never reply to any
|
||||
// request other than "./fixtures/tiles.tilejson" which
|
||||
// will force the cancellation of the pending requests.
|
||||
fileSourceDoNotReply.request = function(req) {
|
||||
if (req.url != "./fixtures/tiles.tilejson") {
|
||||
requestList.push(req);
|
||||
return;
|
||||
}
|
||||
|
||||
fs.readFile(path.join('test', req.url), function(err, data) {
|
||||
req.respond(err, { data: data });
|
||||
});
|
||||
};
|
||||
|
||||
fileSourceDoNotReply.cancel = function() {
|
||||
requestList.pop();
|
||||
};
|
||||
|
||||
t.test('merge requests/cancels for the same resource', function(t) {
|
||||
var map1 = new mbgl.Map(fileSourceDoNotReply);
|
||||
var map2 = new mbgl.Map(fileSourceDoNotReply);
|
||||
var map3 = new mbgl.Map(fileSourceDoNotReply);
|
||||
var map4 = new mbgl.Map(fileSourceDoNotReply);
|
||||
var map5 = new mbgl.Map(fileSourceDoNotReply);
|
||||
|
||||
map1.load(style);
|
||||
map2.load(style);
|
||||
map3.load(style);
|
||||
map4.load(style);
|
||||
map5.load(style);
|
||||
|
||||
var renderCallback = function(err, data) {}
|
||||
|
||||
map1.render({}, renderCallback);
|
||||
map2.render({}, renderCallback);
|
||||
map3.render({}, renderCallback);
|
||||
map4.render({}, renderCallback);
|
||||
map5.render({}, renderCallback);
|
||||
|
||||
var checkCancel = function() {
|
||||
t.equal(requestList.length, 0);
|
||||
t.end();
|
||||
};
|
||||
|
||||
var checkCoalescing = function() {
|
||||
// Check if all the 5 vector tile
|
||||
// requests are going to merge in
|
||||
// a single request.
|
||||
t.equal(requestList.length, 1);
|
||||
|
||||
// This should cause the pending
|
||||
// resources to get canceled, but
|
||||
// only one cancel is emitted, because
|
||||
// all map objects requested the same
|
||||
// resource.
|
||||
map1.release();
|
||||
map2.release();
|
||||
map3.release();
|
||||
map4.release();
|
||||
map5.release();
|
||||
|
||||
setTimeout(checkCancel, 500);
|
||||
};
|
||||
|
||||
setTimeout(checkCoalescing, 500);
|
||||
});
|
||||
|
||||
var vectorTileRequestCount = 0;
|
||||
var fileSourceDelay = new mbgl.FileSource();
|
||||
|
||||
// This file source will add a little delay before
|
||||
// replying so we make sure that all the map objects
|
||||
// request the same resource will get a reply from the
|
||||
// same request.
|
||||
fileSourceDelay.request = function(req) {
|
||||
var timeout = 0;
|
||||
|
||||
if (req.url != "./fixtures/tiles.tilejson") {
|
||||
timeout = 500;
|
||||
t.ok(++vectorTileRequestCount == 1, "Should make only one request for tiles");
|
||||
}
|
||||
|
||||
var readFile = function() {
|
||||
fs.readFile(path.join('test', req.url), function(err, data) {
|
||||
req.respond(err, { data: data });
|
||||
});
|
||||
};
|
||||
|
||||
setTimeout(readFile, timeout);
|
||||
};
|
||||
|
||||
fileSourceDelay.cancel = function() {
|
||||
t.fail("Should never cancel");
|
||||
};
|
||||
|
||||
t.test('one request for the same resource notifies multiple map objects', function(t) {
|
||||
var map1 = new mbgl.Map(fileSourceDelay);
|
||||
var map2 = new mbgl.Map(fileSourceDelay);
|
||||
var map3 = new mbgl.Map(fileSourceDelay);
|
||||
var map4 = new mbgl.Map(fileSourceDelay);
|
||||
var map5 = new mbgl.Map(fileSourceDelay);
|
||||
|
||||
map1.load(style);
|
||||
map2.load(style);
|
||||
map3.load(style);
|
||||
map4.load(style);
|
||||
map5.load(style);
|
||||
|
||||
var renderFinishedCount = 0;
|
||||
var renderCallback = function(err, data) {
|
||||
// We make only one request for tiles, but that
|
||||
// should be enough for all these map objects because
|
||||
// they all want the same resource.
|
||||
if (++renderFinishedCount == 5) {
|
||||
map1.release();
|
||||
map2.release();
|
||||
map3.release();
|
||||
map4.release();
|
||||
map5.release();
|
||||
|
||||
t.end();
|
||||
}
|
||||
}
|
||||
|
||||
map1.render({}, renderCallback);
|
||||
map2.render({}, renderCallback);
|
||||
map3.render({}, renderCallback);
|
||||
map4.render({}, renderCallback);
|
||||
map5.render({}, renderCallback);
|
||||
});
|
||||
|
||||
t.end();
|
||||
});
|
||||
|
||||
t.test('render argument requirements', function(t) {
|
||||
var fileSource = new mbgl.FileSource();
|
||||
fileSource.request = function(req) {
|
||||
@@ -220,10 +350,10 @@ test('Map', function(t) {
|
||||
});
|
||||
});
|
||||
|
||||
t.skip('returns an error', function(t) {
|
||||
t.test('returns an error', function(t) {
|
||||
mbgl.on('message', function(msg) {
|
||||
t.ok(msg, 'emits error');
|
||||
t.equal(msg.class, 'ResourceLoader');
|
||||
t.equal(msg.class, 'Style');
|
||||
t.equal(msg.severity, 'ERROR');
|
||||
t.ok(msg.text.match(/Failed to load/), 'error text matches');
|
||||
});
|
||||
|
||||
2
vendor/mbgl
vendored
2
vendor/mbgl
vendored
Submodule vendor/mbgl updated: a96a251de9...14083700e5
Reference in New Issue
Block a user