MongoDB分片迁移原理与源码（4）

原创

云计算与数据库

修改于 2020-04-07 11:00:50

8430

修改于 2020-04-07 11:00:50

文章被收录于专栏：云计算与数据库云计算与数据库

MongoDB分片迁移原理与源码

异步删除数据

在from shard将迁移结果提交到config服务器成功后，from shard就会执行删除原数据的操作；如果迁移的参数"_waitForDelete"为false，则触发异步删除。"_waitForDelete"的默认参数即是false，即异步删除是默认设计。

将此次迁移的数据范围调用cleanUpRange()函数进行后续处理。

默认情况下，900s 以后开始清理 chunks 的数据，每次清理 128 个文档，每隔 20ms 删除一次。具体通过以下参数设置：

rangeDeleterBatchDelayMS: 删除每个 chunk 数据的时候分批次删除，每批之间间隔的时间，单位 ms，默认 20ms；
internalQueryExecYieldIterations: 默认为 128；
rangeDeleterBatchSize：每次删除数据的数量，默认即为0；为0时，则每次删除的数量为max(internalQueryExecYieldIterations，1)，
orphanCleanupDelaySecs: moveChunk 以后延迟删除数据的时间，单位 s ，默认 900 s

const ChunkRange range(_args.getMinKey(), _args.getMaxKey());

auto notification = [&] {
    auto const whenToClean = _args.getWaitForDelete() ? CollectionShardingRuntime::kNow
                                                      : CollectionShardingRuntime::kDelayed;
    UninterruptibleLockGuard noInterrupt(opCtx->lockState());
    AutoGetCollection autoColl(opCtx, getNss(), MODE_IS);
    return CollectionShardingRuntime::get(opCtx, getNss())->cleanUpRange(range, whenToClean);
}();

// 默认的异步删除时间
//MONGO_EXPORT_SERVER_PARAMETER(orphanCleanupDelaySecs, int, 900);  // 900s = 15m
auto CollectionShardingRuntime::cleanUpRange(ChunkRange const& range, CleanWhen when)
    -> CleanupNotification {
    Date_t time = (when == kNow) ? Date_t{} : Date_t::now() +
            stdx::chrono::seconds{orphanCleanupDelaySecs.load()};
    return _metadataManager->cleanUpRange(range, time);
}

再删除之前，还要判断是否满足没有任何基于该chunk的查询了：如果没有则放到删除队列中，等删除时间到了；如果还有查询，则放到另外一个孤儿文档队列，后续再删除；

auto MetadataManager::cleanUpRange(ChunkRange const& range, Date_t whenToDelete)
    -> CleanupNotification {
    stdx::lock_guard<stdx::mutex> lg(_managerLock);
    invariant(!_metadata.empty());

    auto* const activeMetadata = _metadata.back().get();
    auto* const overlapMetadata = _findNewestOverlappingMetadata(lg, range);

    if (overlapMetadata == activeMetadata) {
        return Status{ErrorCodes::RangeOverlapConflict,
                      str::stream() << "Requested deletion range overlaps a live shard chunk"};
    }

    if (rangeMapOverlaps(_receivingChunks, range.getMin(), range.getMax())) {
        return Status{ErrorCodes::RangeOverlapConflict,
                      str::stream() << "Requested deletion range overlaps a chunk being"
                                       " migrated in"};
    }

    if (!overlapMetadata) {
        //如果没有基于该chunk的查询了，则把该数据块放到删除队列中.
        const auto whenStr = (whenToDelete == Date_t{}) ? "immediate"_sd : "deferred"_sd;
        log() << "Scheduling " << whenStr << " deletion of " << _nss.ns() << " range "
              << redact(range.toString());
        return _pushRangeToClean(lg, range, whenToDelete);
    }

    log() << "Deletion of " << _nss.ns() << " range " << redact(range.toString())
          << " will be scheduled after all possibly dependent queries finish";

    //如果还有查询，则放到孤儿文档的队列中，后续再删除.
    auto& orphans = overlapMetadata->orphans;
    orphans.emplace_back(ChunkRange(range.getMin().getOwned(), range.getMax().getOwned()),
                         whenToDelete);
    return orphans.back().notification;
}

根据删除时间，则定是否放到最终的异步删除的任务线程中scheduleCleanup()

auto MetadataManager::_pushRangeToClean(WithLock lock, ChunkRange const& range, Date_t when)
    -> CleanupNotification {
    std::list<Deletion> ranges;
    ranges.emplace_back(ChunkRange(range.getMin().getOwned(), range.getMax().getOwned()), when);
    auto& notifn = ranges.back().notification;
    _pushListToClean(lock, std::move(ranges));
    return notifn;
}

void MetadataManager::_pushListToClean(WithLock, std::list<Deletion> ranges) {
    auto when = _rangesToClean.add(std::move(ranges));
    if (when) {
        scheduleCleanup(
            _executor, _nss, _metadata.back()->metadata.getCollVersion().epoch(), *when);
    }
    invariant(ranges.empty());
}

void scheduleCleanup(executor::TaskExecutor* executor,
                     NamespaceString nss,
                     OID epoch,
                     Date_t when) {
    LOG(1) << "Scheduling cleanup on " << nss.ns() << " at " << when;
    auto swCallbackHandle = executor->scheduleWorkAt(
        when, [ executor, nss = std::move(nss), epoch = std::move(epoch) ](auto&) {
            Client::initThreadIfNotAlready("Collection Range Deleter");
            auto uniqueOpCtx = Client::getCurrent()->makeOperationContext();
            auto opCtx = uniqueOpCtx.get();

            const int maxToDelete = std::max(int(internalQueryExecYieldIterations.load()), 1);

            MONGO_FAIL_POINT_PAUSE_WHILE_SET(suspendRangeDeletion);

            //执行真正的删除，但是每批只删除maxToDelete（默认128）个文档；每批间隔时间默认为rangeDeleterBatchDelayMS（20）毫秒。
            //最终删除调用的是collection->deleteDocument()删除集合文档的接口，完成文档删除
            auto next = CollectionRangeDeleter::cleanUpNextRange(opCtx, nss, epoch, maxToDelete);
            if (next) {
                scheduleCleanup(executor, std::move(nss), std::move(epoch), *next);
            }
        });

    if (!swCallbackHandle.isOK()) {
        log() << "Failed to schedule the orphan data cleanup task"
              << causedBy(redact(swCallbackHandle.getStatus()));
    }
}