From 9e931d09ba9a1269bf11a689e06e56f61dab7e21 Mon Sep 17 00:00:00 2001 From: Gregory Noma Date: Wed, 27 Oct 2021 14:33:37 +0000 Subject: SERVER-59662 Test killing time-series inserts --- src/mongo/db/commands/write_commands.cpp | 23 +++++++++++++++------- src/mongo/db/timeseries/bucket_catalog.cpp | 31 +++++++++++++++--------------- src/mongo/db/timeseries/bucket_catalog.h | 2 +- 3 files changed, 32 insertions(+), 24 deletions(-) (limited to 'src/mongo') diff --git a/src/mongo/db/commands/write_commands.cpp b/src/mongo/db/commands/write_commands.cpp index d794025b2c0..9be0940221a 100644 --- a/src/mongo/db/commands/write_commands.cpp +++ b/src/mongo/db/commands/write_commands.cpp @@ -770,10 +770,11 @@ public: invariant(batch->finished()); auto batchStatus = batch->getResult().getStatus(); tassert(5916402, - str::stream() << "Got unexpected error (" << batch->getResult().getStatus() + str::stream() << "Got unexpected error (" << batchStatus << ") preparing time-series bucket to be committed for " << ns() << ": " << redact(request().toBSON({})), batchStatus == ErrorCodes::TimeseriesBucketCleared || + batchStatus.isA() || batchStatus.isA()); docsToRetry->push_back(index); @@ -863,15 +864,21 @@ public: return left.get()->bucket() < right.get()->bucket(); }); + boost::optional abortStatus; + ScopeGuard batchGuard{[&] { + for (auto batch : batchesToCommit) { + if (batch.get()) { + bucketCatalog.abort(batch, abortStatus); + } + } + }}; + std::vector insertOps; std::vector updateOps; for (auto batch : batchesToCommit) { auto metadata = bucketCatalog.getMetadata(batch.get()->bucket()); if (!bucketCatalog.prepareCommit(batch)) { - for (auto batchToAbort : batchesToCommit) { - bucketCatalog.abort(batchToAbort); - } return false; } @@ -889,9 +896,7 @@ public: auto result = write_ops_exec::performAtomicTimeseriesWrites(opCtx, insertOps, updateOps); if (!result.isOK()) { - for (auto batch : batchesToCommit) { - bucketCatalog.abort(batch, result); - } + abortStatus = result; return false; } @@ -915,10 +920,14 @@ public: compressClosedBuckets = false; } if (!ret.canContinue) { + if (!ret.result.isOK()) { + abortStatus = ret.result.getStatus(); + } return false; } } + batchGuard.dismiss(); return true; } diff --git a/src/mongo/db/timeseries/bucket_catalog.cpp b/src/mongo/db/timeseries/bucket_catalog.cpp index 3b5da81a3bb..c6fecf6f47b 100644 --- a/src/mongo/db/timeseries/bucket_catalog.cpp +++ b/src/mongo/db/timeseries/bucket_catalog.cpp @@ -299,8 +299,6 @@ bool BucketCatalog::prepareCommit(std::shared_ptr batch) { batch->_prepareCommit(); _memoryUsage.fetchAndAdd(bucket->_memoryUsage - prevMemoryUsage); - bucket->_batches.erase(batch->_opId); - return true; } @@ -351,18 +349,20 @@ boost::optional BucketCatalog::finish( bucket.release(); auto lk = _lockExclusive(); - closedBucket = - ClosedBucket{ptr->_id, ptr->getTimeField().toString(), ptr->numMeasurements()}; - - // Only remove from _allBuckets and _idleBuckets. If it was marked full, we know that - // happened in BucketAccess::rollover, and that there is already a new open bucket for - // this metadata. - _markBucketNotIdle(ptr, false /* locked */); - { - stdx::lock_guard statesLk{_statesMutex}; - _bucketStates.erase(ptr->_id); + if (_allBuckets.contains(ptr)) { + closedBucket = + ClosedBucket{ptr->_id, ptr->getTimeField().toString(), ptr->numMeasurements()}; + + // Only remove from _allBuckets and _idleBuckets. If it was marked full, we know + // that happened in BucketAccess::rollover, and that there is already a new open + // bucket for this metadata. + _markBucketNotIdle(ptr, false /* locked */); + { + stdx::lock_guard statesLk{_statesMutex}; + _bucketStates.erase(ptr->_id); + } + _allBuckets.erase(ptr); } - _allBuckets.erase(ptr); } else { _markBucketIdle(bucket); } @@ -486,7 +486,7 @@ BucketCatalog::StripedMutex::ExclusiveLock BucketCatalog::_lockExclusive() const void BucketCatalog::_waitToCommitBatch(const std::shared_ptr& batch) { while (true) { BucketAccess bucket{this, batch->bucket()}; - if (!bucket) { + if (!bucket || batch->finished()) { return; } @@ -494,6 +494,7 @@ void BucketCatalog::_waitToCommitBatch(const std::shared_ptr& batch) if (!current) { // No other batches for this bucket are currently committing, so we can proceed. bucket->_preparedBatch = batch; + bucket->_batches.erase(batch->_opId); break; } @@ -1239,7 +1240,6 @@ void BucketCatalog::WriteBatch::_finish(const CommitInfo& info) { invariant(_commitRights.load()); invariant(!_active); _promise.emplaceValue(info); - _bucket = nullptr; } void BucketCatalog::WriteBatch::_abort(const boost::optional& status, @@ -1257,7 +1257,6 @@ void BucketCatalog::WriteBatch::_abort(const boost::optional& status, _promise.setError(status.value_or( Status{ErrorCodes::TimeseriesBucketCleared, str::stream() << "Time-series bucket " << bucketIdentification << "was cleared"})); - _bucket = nullptr; } class BucketCatalog::ServerStatus : public ServerStatusSection { diff --git a/src/mongo/db/timeseries/bucket_catalog.h b/src/mongo/db/timeseries/bucket_catalog.h index 2c2b1994443..9936a9be4c3 100644 --- a/src/mongo/db/timeseries/bucket_catalog.h +++ b/src/mongo/db/timeseries/bucket_catalog.h @@ -158,7 +158,7 @@ public: void _abort(const boost::optional& status, bool canAccessBucket); - Bucket* _bucket; + Bucket* const _bucket; OperationId _opId; std::shared_ptr _stats; -- cgit v1.2.1