RasterTile / mailbox fix (#2443)
This commit is contained in:
@@ -20,15 +20,17 @@ void Mailbox::open(Scheduler& scheduler_) {
|
||||
std::lock_guard<std::recursive_mutex> receivingLock(receivingMutex);
|
||||
std::lock_guard<std::mutex> pushingLock(pushingMutex);
|
||||
|
||||
weakScheduler = scheduler_.makeWeakPtr();
|
||||
|
||||
if (closed) {
|
||||
return;
|
||||
}
|
||||
|
||||
weakScheduler = scheduler_.makeWeakPtr();
|
||||
|
||||
if (!queue.empty()) {
|
||||
auto guard = weakScheduler.lock();
|
||||
if (weakScheduler) weakScheduler->schedule(makeClosure(shared_from_this()));
|
||||
if (weakScheduler) {
|
||||
weakScheduler->schedule(makeClosure(shared_from_this()));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -45,6 +47,8 @@ void Mailbox::close() {
|
||||
std::lock_guard<std::mutex> pushingLock(pushingMutex);
|
||||
|
||||
closed = true;
|
||||
|
||||
weakScheduler = {};
|
||||
}
|
||||
|
||||
void Mailbox::abandon() {
|
||||
@@ -57,7 +61,7 @@ void Mailbox::abandon() {
|
||||
}
|
||||
|
||||
bool Mailbox::isOpen() const {
|
||||
return bool(weakScheduler);
|
||||
return bool(weakScheduler) && !closed;
|
||||
}
|
||||
|
||||
void Mailbox::push(std::unique_ptr<Message> message) {
|
||||
@@ -105,9 +109,6 @@ void Mailbox::receive() {
|
||||
}};
|
||||
std::lock_guard<std::recursive_mutex> receivingLock(receivingMutex);
|
||||
|
||||
auto guard = weakScheduler.lock();
|
||||
assert(weakScheduler);
|
||||
|
||||
if (closed) {
|
||||
state = State::Abandoned;
|
||||
return;
|
||||
@@ -126,8 +127,13 @@ void Mailbox::receive() {
|
||||
|
||||
(*message)();
|
||||
|
||||
// If there are more messages in the queue and the scheduler
|
||||
// is still active, create a new task to handle the next one
|
||||
if (!wasEmpty) {
|
||||
weakScheduler->schedule(makeClosure(shared_from_this()));
|
||||
auto guard = weakScheduler.lock();
|
||||
if (weakScheduler) {
|
||||
weakScheduler->schedule(makeClosure(shared_from_this()));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -100,12 +100,12 @@ protected:
|
||||
const GeometryTileData* getData() const;
|
||||
LayerRenderData* getLayerRenderData(const style::Layer::Impl&);
|
||||
|
||||
private:
|
||||
void markObsolete();
|
||||
|
||||
// Used to signal the worker that it should abandon parsing this tile as soon as possible.
|
||||
std::atomic<bool> obsolete{false};
|
||||
|
||||
private:
|
||||
void markObsolete();
|
||||
|
||||
const std::shared_ptr<Scheduler> threadPool;
|
||||
|
||||
const std::shared_ptr<Mailbox> mailbox;
|
||||
|
||||
@@ -17,6 +17,7 @@ namespace mbgl {
|
||||
RasterDEMTile::RasterDEMTile(const OverscaledTileID& id_, const TileParameters& parameters, const Tileset& tileset)
|
||||
: Tile(Kind::RasterDEM, id_),
|
||||
loader(*this, id_, parameters, tileset),
|
||||
threadPool(Scheduler::GetBackground()),
|
||||
mailbox(std::make_shared<Mailbox>(*Scheduler::GetCurrent())),
|
||||
worker(Scheduler::GetBackground(), ActorRef<RasterDEMTile>(*this, mailbox)) {
|
||||
encoding = tileset.encoding;
|
||||
@@ -31,7 +32,14 @@ RasterDEMTile::RasterDEMTile(const OverscaledTileID& id_, const TileParameters&
|
||||
}
|
||||
}
|
||||
|
||||
RasterDEMTile::~RasterDEMTile() = default;
|
||||
RasterDEMTile::~RasterDEMTile() {
|
||||
markObsolete();
|
||||
|
||||
// The bucket has resources that need to be released on the render thread.
|
||||
if (bucket) {
|
||||
threadPool->runOnRenderThread([bucket_{std::move(bucket)}]() {});
|
||||
}
|
||||
}
|
||||
|
||||
std::unique_ptr<TileRenderData> RasterDEMTile::createRenderData() {
|
||||
return std::make_unique<SharedBucketTileRenderData<HillshadeBucket>>(bucket);
|
||||
@@ -48,19 +56,23 @@ void RasterDEMTile::setMetadata(std::optional<Timestamp> modified_, std::optiona
|
||||
}
|
||||
|
||||
void RasterDEMTile::setData(const std::shared_ptr<const std::string>& data) {
|
||||
pending = true;
|
||||
++correlationID;
|
||||
worker.self().invoke(&RasterDEMTileWorker::parse, data, correlationID, encoding);
|
||||
if (!obsolete) {
|
||||
pending = true;
|
||||
++correlationID;
|
||||
worker.self().invoke(&RasterDEMTileWorker::parse, data, correlationID, encoding);
|
||||
}
|
||||
}
|
||||
|
||||
void RasterDEMTile::onParsed(std::unique_ptr<HillshadeBucket> result, const uint64_t resultCorrelationID) {
|
||||
bucket = std::move(result);
|
||||
loaded = true;
|
||||
if (resultCorrelationID == correlationID) {
|
||||
pending = false;
|
||||
if (!obsolete) {
|
||||
bucket = std::move(result);
|
||||
loaded = true;
|
||||
if (resultCorrelationID == correlationID) {
|
||||
pending = false;
|
||||
}
|
||||
renderable = static_cast<bool>(bucket);
|
||||
observer->onTileChanged(*this);
|
||||
}
|
||||
renderable = static_cast<bool>(bucket);
|
||||
observer->onTileChanged(*this);
|
||||
}
|
||||
|
||||
void RasterDEMTile::onError(std::exception_ptr err, const uint64_t resultCorrelationID) {
|
||||
@@ -125,4 +137,14 @@ void RasterDEMTile::setUpdateParameters(const TileUpdateParameters& params) {
|
||||
loader.setUpdateParameters(params);
|
||||
}
|
||||
|
||||
void RasterDEMTile::cancel() {
|
||||
markObsolete();
|
||||
}
|
||||
|
||||
void RasterDEMTile::markObsolete() {
|
||||
obsolete = true;
|
||||
pending = false;
|
||||
mailbox->abandon();
|
||||
}
|
||||
|
||||
} // namespace mbgl
|
||||
|
||||
@@ -86,9 +86,14 @@ public:
|
||||
void onParsed(std::unique_ptr<HillshadeBucket> result, uint64_t correlationID);
|
||||
void onError(std::exception_ptr, uint64_t correlationID);
|
||||
|
||||
void cancel() override;
|
||||
|
||||
private:
|
||||
void markObsolete();
|
||||
|
||||
TileLoader<RasterDEMTile> loader;
|
||||
|
||||
const std::shared_ptr<Scheduler> threadPool;
|
||||
std::shared_ptr<Mailbox> mailbox;
|
||||
Actor<RasterDEMTileWorker> worker;
|
||||
|
||||
@@ -98,6 +103,8 @@ private:
|
||||
// Contains the Bucket object for the tile. Buckets are render
|
||||
// objects and they get added by tile parsing operations.
|
||||
std::shared_ptr<HillshadeBucket> bucket;
|
||||
|
||||
bool obsolete = false;
|
||||
};
|
||||
|
||||
} // namespace mbgl
|
||||
|
||||
@@ -17,10 +17,18 @@ namespace mbgl {
|
||||
RasterTile::RasterTile(const OverscaledTileID& id_, const TileParameters& parameters, const Tileset& tileset)
|
||||
: Tile(Kind::Raster, id_),
|
||||
loader(*this, id_, parameters, tileset),
|
||||
threadPool(Scheduler::GetBackground()),
|
||||
mailbox(std::make_shared<Mailbox>(*Scheduler::GetCurrent())),
|
||||
worker(Scheduler::GetBackground(), ActorRef<RasterTile>(*this, mailbox)) {}
|
||||
|
||||
RasterTile::~RasterTile() = default;
|
||||
RasterTile::~RasterTile() {
|
||||
markObsolete();
|
||||
|
||||
// The bucket has resources that need to be released on the render thread.
|
||||
if (bucket) {
|
||||
threadPool->runOnRenderThread([bucket_{std::move(bucket)}]() {});
|
||||
}
|
||||
}
|
||||
|
||||
std::unique_ptr<TileRenderData> RasterTile::createRenderData() {
|
||||
return std::make_unique<SharedBucketTileRenderData<RasterBucket>>(bucket);
|
||||
@@ -37,19 +45,23 @@ void RasterTile::setMetadata(std::optional<Timestamp> modified_, std::optional<T
|
||||
}
|
||||
|
||||
void RasterTile::setData(const std::shared_ptr<const std::string>& data) {
|
||||
pending = true;
|
||||
++correlationID;
|
||||
worker.self().invoke(&RasterTileWorker::parse, data, correlationID);
|
||||
if (!obsolete) {
|
||||
pending = true;
|
||||
++correlationID;
|
||||
worker.self().invoke(&RasterTileWorker::parse, data, correlationID);
|
||||
}
|
||||
}
|
||||
|
||||
void RasterTile::onParsed(std::unique_ptr<RasterBucket> result, const uint64_t resultCorrelationID) {
|
||||
bucket = std::move(result);
|
||||
loaded = true;
|
||||
if (resultCorrelationID == correlationID) {
|
||||
pending = false;
|
||||
if (!obsolete) {
|
||||
bucket = std::move(result);
|
||||
loaded = true;
|
||||
if (resultCorrelationID == correlationID) {
|
||||
pending = false;
|
||||
}
|
||||
renderable = static_cast<bool>(bucket);
|
||||
observer->onTileChanged(*this);
|
||||
}
|
||||
renderable = static_cast<bool>(bucket);
|
||||
observer->onTileChanged(*this);
|
||||
}
|
||||
|
||||
void RasterTile::onError(std::exception_ptr err, const uint64_t resultCorrelationID) {
|
||||
@@ -78,4 +90,14 @@ void RasterTile::setUpdateParameters(const TileUpdateParameters& params) {
|
||||
loader.setUpdateParameters(params);
|
||||
}
|
||||
|
||||
void RasterTile::cancel() {
|
||||
markObsolete();
|
||||
}
|
||||
|
||||
void RasterTile::markObsolete() {
|
||||
obsolete = true;
|
||||
pending = false;
|
||||
mailbox->abandon();
|
||||
}
|
||||
|
||||
} // namespace mbgl
|
||||
|
||||
@@ -35,9 +35,14 @@ public:
|
||||
void onParsed(std::unique_ptr<RasterBucket> result, uint64_t correlationID);
|
||||
void onError(std::exception_ptr, uint64_t correlationID);
|
||||
|
||||
void cancel() override;
|
||||
|
||||
private:
|
||||
void markObsolete();
|
||||
|
||||
TileLoader<RasterTile> loader;
|
||||
|
||||
const std::shared_ptr<Scheduler> threadPool;
|
||||
std::shared_ptr<Mailbox> mailbox;
|
||||
Actor<RasterTileWorker> worker;
|
||||
|
||||
@@ -46,6 +51,8 @@ private:
|
||||
// Contains the Bucket object for the tile. Buckets are render
|
||||
// objects and they get added by tile parsing operations.
|
||||
std::shared_ptr<RasterBucket> bucket;
|
||||
|
||||
std::atomic<bool> obsolete{false};
|
||||
};
|
||||
|
||||
} // namespace mbgl
|
||||
|
||||
@@ -19,8 +19,6 @@ void Tile::setObserver(TileObserver* observer_) {
|
||||
observer = observer_;
|
||||
}
|
||||
|
||||
void Tile::cancel() {}
|
||||
|
||||
void Tile::setTriedCache() {
|
||||
triedOptional = true;
|
||||
observer->onTileChanged(*this);
|
||||
|
||||
@@ -68,7 +68,7 @@ public:
|
||||
virtual void setUpdateParameters(const TileUpdateParameters&) {}
|
||||
|
||||
// Mark this tile as no longer needed and cancel any pending work.
|
||||
virtual void cancel();
|
||||
virtual void cancel() = 0;
|
||||
|
||||
// Notifies this tile of the updated layer properties.
|
||||
//
|
||||
|
||||
@@ -62,8 +62,8 @@ private:
|
||||
std::atomic_bool aborted{false};
|
||||
};
|
||||
|
||||
// Allocated as a share_ptr so either the loader or request can outlive the
|
||||
// other and still see this.
|
||||
// Allocated as a `shared_ptr` so either the loader or
|
||||
// request can outlive the other and still see this.
|
||||
std::shared_ptr<Shared> shared;
|
||||
};
|
||||
|
||||
|
||||
@@ -59,11 +59,20 @@ TileLoader<T>::TileLoader(T& tile_,
|
||||
|
||||
template <typename T>
|
||||
TileLoader<T>::~TileLoader() {
|
||||
std::unique_lock<std::shared_mutex> lock(shared->requestLock);
|
||||
shared->aborted = true;
|
||||
tile.cancel();
|
||||
// Tasks may outlive this object, make sure that they are not
|
||||
// currently running and do not run if they are still pending.
|
||||
{
|
||||
std::unique_lock<std::shared_mutex> lock(shared->requestLock);
|
||||
shared->aborted = true;
|
||||
}
|
||||
|
||||
// Don't touch `tile` here, we may be called from its destructor
|
||||
// tile.cancel();
|
||||
|
||||
request.reset();
|
||||
};
|
||||
fileSource.reset();
|
||||
shared.reset();
|
||||
}
|
||||
|
||||
template <typename T>
|
||||
void TileLoader<T>::setNecessity(TileNecessity newNecessity) {
|
||||
|
||||
@@ -13,6 +13,12 @@ VectorTile::VectorTile(const OverscaledTileID& id_,
|
||||
: GeometryTile(id_, std::move(sourceID_), parameters),
|
||||
loader(*this, id_, parameters, tileset) {}
|
||||
|
||||
VectorTile::~VectorTile() {
|
||||
// Don't rely on `~TileLoader` to close, it's not safe to call there.
|
||||
// We're still calling a virtual method from a destructor, so any overrides will not be called.
|
||||
GeometryTile::cancel();
|
||||
}
|
||||
|
||||
void VectorTile::setNecessity(TileNecessity necessity) {
|
||||
loader.setNecessity(necessity);
|
||||
}
|
||||
@@ -27,6 +33,10 @@ void VectorTile::setMetadata(std::optional<Timestamp> modified_, std::optional<T
|
||||
}
|
||||
|
||||
void VectorTile::setData(const std::shared_ptr<const std::string>& data_) {
|
||||
if (obsolete) {
|
||||
return;
|
||||
}
|
||||
|
||||
GeometryTile::setData(data_ ? std::make_unique<VectorTileData>(data_) : nullptr);
|
||||
}
|
||||
|
||||
|
||||
@@ -11,6 +11,7 @@ class TileParameters;
|
||||
class VectorTile : public GeometryTile {
|
||||
public:
|
||||
VectorTile(const OverscaledTileID&, std::string sourceID, const TileParameters&, const Tileset&);
|
||||
~VectorTile() override;
|
||||
|
||||
void setNecessity(TileNecessity) final;
|
||||
void setUpdateParameters(const TileUpdateParameters&) final;
|
||||
|
||||
@@ -772,6 +772,8 @@ public:
|
||||
|
||||
std::unique_ptr<TileRenderData> createRenderData() override { return nullptr; }
|
||||
|
||||
void cancel() override {}
|
||||
|
||||
private:
|
||||
FakeTileSource& source;
|
||||
};
|
||||
|
||||
Reference in New Issue
Block a user