Method executeImpl

src/DaemonManager/DaemonJobServerBGThread.cpp:620–812

every failed call on BackgroundJob in this function will be retried on next time

/// every failed call on BackgroundJob in this function will be retried on next time
bool DaemonJobServerBGThread::executeImpl()
{
    if (suspended())
    {
        LOG_DEBUG(log, "thread suspended");
        std::unique_lock lock(bg_jobs_mutex);
        background_jobs.clear();
        return true;
    }

    Context & context = *getContext();
    std::shared_ptr<CnchTopologyMaster> topology_master = context.getCnchTopologyMaster();
    if (!topology_master)
    {
        LOG_ERROR(log, "Failed to get topology master");
        return false;
    }

    UInt64 milliseconds = 0;
    Stopwatch watch;
    BackgroundJobs background_jobs_clone;
    {
        std::shared_lock shared_lock(bg_jobs_mutex);
        background_jobs_clone = background_jobs;
    }

    milliseconds = watch.elapsedMilliseconds();
    if (milliseconds >= SLOW_EXECUTION_THRESHOLD_MS)
        LOG_DEBUG(log, "copy background jobs took {} ms.", milliseconds);

    watch.restart();
    std::unordered_map<UUID, StorageID> new_uuid_map = getUUIDsFromCatalog(*this);
    milliseconds = watch.elapsedMilliseconds();
    if (milliseconds >= SLOW_EXECUTION_THRESHOLD_MS)
        LOG_DEBUG(log, "getUUIDsFromCatalog with size {} took {} ms.", new_uuid_map.size(), milliseconds);

    std::map<String, UInt64> new_server_start_times = fetchServerStartTimes(context, *topology_master, log);
    if (new_server_start_times.empty())
    {
        LOG_WARNING(log, "There are network partition, skip this iteration");
        return false;
    }

    const std::vector<String> alive_servers = findAliveServers(new_server_start_times);
    watch.restart();
    UpdateResult update_res = getUpdateBGJobs(background_jobs_clone, new_uuid_map, alive_servers);
    milliseconds = watch.elapsedMilliseconds();
    if (milliseconds >= SLOW_EXECUTION_THRESHOLD_MS)
        LOG_DEBUG(log, "getUpdateBGJobs took {} ms.", milliseconds);

    const UUIDs & remove_uuids = update_res.remove_uuids;
    for (auto uuid : remove_uuids)
        LOG_DEBUG(log, "UUID: {} will be removed from background jobs", UUIDHelpers::UUIDToString(uuid));

    const UUIDs & add_uuids = update_res.add_uuids;
    for (auto uuid : add_uuids)
        LOG_DEBUG(log, "UUID: {} will be added into background jobs", UUIDHelpers::UUIDToString(uuid));