OpenHarmony-v4.0.1-Release/s

/*
 * Copyright (c) 2021 Huawei Device Co., Ltd.
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "single_ver_sync_task_context.h"

#include <algorithm>
#include "db_common.h"
#include "db_dfx_adapter.h"
#include "db_errno.h"
#include "isyncer.h"
#include "log_print.h"
#include "single_ver_sync_state_machine.h"
#include "single_ver_sync_target.h"
#include "sync_types.h"

namespace DistributedDB {
SingleVerSyncTaskContext::SingleVerSyncTaskContext()
    : SyncTaskContext(),
      token_(nullptr),
      endMark_(0),
      needClearRemoteStaleData_(false)
{}

SingleVerSyncTaskContext::~SingleVerSyncTaskContext()
{
    token_ = nullptr;
    subManager_ = nullptr;
}

int SingleVerSyncTaskContext::Initialize(const std::string &deviceId, ISyncInterface *syncInterface,
    const std::shared_ptr<Metadata> &metadata, ICommunicator *communicator)
{
    if (deviceId.empty() || syncInterface == nullptr || metadata == nullptr ||
        communicator == nullptr) {
        return -E_INVALID_ARGS;
    }
    stateMachine_ = new (std::nothrow) SingleVerSyncStateMachine;
    if (stateMachine_ == nullptr) {
        return -E_OUT_OF_MEMORY;
    }
    deviceId_ = deviceId;
    std::vector<uint8_t> dbIdentifier = syncInterface->GetIdentifier();
    dbIdentifier.resize(3); // only show 3 bytes
    syncActionName_ = DBDfxAdapter::SYNC_ACTION + "_" +
        DBCommon::VectorToHexString(dbIdentifier) + "_" + deviceId_.c_str();
    TimerAction timeOutCallback;
    int errCode = stateMachine_->Initialize(this, syncInterface, metadata, communicator);
    if (errCode != E_OK) {
        LOGE("[SingleVerSyncTaskContext] stateMachine Initialize failed, err %d.", errCode);
        goto ERROR_OUT;
    }

    timeHelper_ = std::make_unique<TimeHelper>();
    errCode = timeHelper_->Initialize(syncInterface, metadata);
    if (errCode != E_OK) {
        LOGE("[SingleVerSyncTaskContext] timeHelper Initialize failed, err %d.", errCode);
        goto ERROR_OUT;
    }
    timeOutCallback = std::bind(&SyncStateMachine::TimeoutCallback,
        static_cast<SingleVerSyncStateMachine *>(stateMachine_),
        std::placeholders::_1);
    SetTimeoutCallback(timeOutCallback);

    syncInterface_ = syncInterface;
    communicator_ = communicator;
    taskExecStatus_ = INIT;
    OnKill([this]() { this->KillWait(); });
    {
        std::lock_guard<std::mutex> lock(synTaskContextSetLock_);
        synTaskContextSet_.insert(this);
    }
    return errCode;

ERROR_OUT:
    delete stateMachine_;
    stateMachine_ = nullptr;
    return errCode;
}

int SingleVerSyncTaskContext::AddSyncOperation(SyncOperation *operation)
{
    if (operation == nullptr) {
        return -E_INVALID_ARGS;
    }

    // If auto sync, just update the end watermark
    if (operation->IsAutoSync()) {
        std::lock_guard<std::mutex> lock(targetQueueLock_);
        bool isQuerySync = operation->IsQuerySync();
        std::string queryId = operation->GetQueryId();
        auto iter = std::find_if(requestTargetQueue_.begin(), requestTargetQueue_.end(),
            [isQuerySync, queryId](const ISyncTarget *target) {
            if (target == nullptr) {
                return false;
            }
            if (isQuerySync) {
                SyncOperation *tmpOperation = nullptr;
                target->GetSyncOperation(tmpOperation);
                return (tmpOperation != nullptr && tmpOperation->GetQueryId() == queryId) && target->IsAutoSync();
            }
            return target->IsAutoSync();
        });
        if (iter != requestTargetQueue_.end()) {
            static_cast<SingleVerSyncTarget *>(*iter)->SetEndWaterMark(timeHelper_->GetTime());
            operation->SetStatus(deviceId_, SyncOperation::OP_FINISHED_ALL);
            return E_OK;
        }
    }

    auto *newTarget = new (std::nothrow) SingleVerSyncTarget;
    if (newTarget == nullptr) {
        return -E_OUT_OF_MEMORY;
    }
    newTarget->SetSyncOperation(operation);
    Timestamp timstamp = timeHelper_->GetTime();
    newTarget->SetEndWaterMark(timstamp);
    newTarget->SetTaskType(ISyncTarget::REQUEST);
    AddSyncTarget(newTarget);
    return E_OK;
}

void SingleVerSyncTaskContext::SetEndMark(WaterMark endMark)
{
    endMark_ = endMark;
}

WaterMark SingleVerSyncTaskContext::GetEndMark() const
{
    return endMark_;
}

void SingleVerSyncTaskContext::GetContinueToken(ContinueToken &outToken) const
{
    outToken = token_;
}

void SingleVerSyncTaskContext::SetContinueToken(ContinueToken token)
{
    token_ = token;
    return;
}

void SingleVerSyncTaskContext::ReleaseContinueToken()
{
    if (token_ != nullptr) {
        static_cast<SyncGenericInterface *>(syncInterface_)->ReleaseContinueToken(token_);
        token_ = nullptr;
    }
}

int SingleVerSyncTaskContext::PopResponseTarget(SingleVerSyncTarget &target)
{
    std::lock_guard<std::mutex> lock(targetQueueLock_);
    LOGD("[SingleVerSyncTaskContext] GetFrontExtWaterMark size = %zu", responseTargetQueue_.size());
    if (!responseTargetQueue_.empty()) {
        ISyncTarget *tmpTarget = responseTargetQueue_.front();
        responseTargetQueue_.pop_front();
        target = *(static_cast<SingleVerSyncTarget *>(tmpTarget));
        delete tmpTarget;
        tmpTarget = nullptr;
        return E_OK;
    }
    return -E_LENGTH_ERROR;
}

int SingleVerSyncTaskContext::GetRspTargetQueueSize() const
{
    std::lock_guard<std::mutex> lock(targetQueueLock_);
    return responseTargetQueue_.size();
}

void SingleVerSyncTaskContext::SetResponseSessionId(uint32_t responseSessionId)
{
    responseSessionId_ = responseSessionId;
}

uint32_t SingleVerSyncTaskContext::GetResponseSessionId() const
{
    return responseSessionId_;
}

void SingleVerSyncTaskContext::CopyTargetData(const ISyncTarget *target, const TaskParam &taskParam)
{
    const SingleVerSyncTarget *targetTmp = static_cast<const SingleVerSyncTarget *>(target);
    SyncTaskContext::CopyTargetData(target, taskParam);
    mode_ = targetTmp->GetMode();
    endMark_ = targetTmp->GetEndWaterMark();
    if (mode_ == SyncModeType::RESPONSE_PULL) {
        responseSessionId_ = targetTmp->GetResponseSessionId();
    }
    {
        std::lock_guard<std::mutex> autoLock(queryMutex_);
        query_ = targetTmp->GetQuery();
    }
    isQuerySync_ = targetTmp->IsQuerySync();
}

void SingleVerSyncTaskContext::Clear()
{
    retryTime_ = 0;
    ClearSyncOperation();
    SyncTaskContext::Clear();
    SetMode(SyncModeType::INVALID_MODE);
    syncId_ = 0;
    isAutoSync_ = false;
    SetOperationStatus(SyncOperation::OP_WAITING);
    SetEndMark(0);
    SetResponseSessionId(0);
    {
        std::lock_guard<std::mutex> autoLock(queryMutex_);
        query_ = QuerySyncObject();
    }
    isQuerySync_ = false;
}

void SingleVerSyncTaskContext::Abort(int status)
{
    {
        std::lock_guard<std::mutex> lock(operationLock_);
        if (syncOperation_ != nullptr) {
            syncOperation_->SetStatus(deviceId_, status);
            if ((status >= SyncOperation::OP_FINISHED_ALL)) {
                UnlockObj();
                if (syncOperation_->CheckIsAllFinished()) {
                    syncOperation_->Finished();
                }
                LockObj();
            }
        }
    }
    StopFeedDogForSync(SyncDirectionFlag::SEND);
    StopFeedDogForSync(SyncDirectionFlag::RECEIVE);
    Clear();
}

void SingleVerSyncTaskContext::ClearAllSyncTask()
{
    // clear request queue sync task and responsequeue first.
    std::list<ISyncTarget *> targetQueue;
    {
        std::lock_guard<std::mutex> lock(targetQueueLock_);
        LOGI("[SingleVerSyncTaskContext] request taskcount=%zu, responsecount=%zu", requestTargetQueue_.size(),
            responseTargetQueue_.size());
        while (!requestTargetQueue_.empty()) {
            ISyncTarget *tmpTarget = requestTargetQueue_.front();
            requestTargetQueue_.pop_front();
            SyncOperation *tmpInfOperation = nullptr;
            tmpTarget->GetSyncOperation(tmpInfOperation);
            RefObject::IncObjRef(tmpInfOperation);
            targetQueue.push_back(tmpTarget);
        }
        while (!responseTargetQueue_.empty()) {
            ISyncTarget *tmpTarget = responseTargetQueue_.front();
            responseTargetQueue_.pop_front();
            delete tmpTarget;
            tmpTarget = nullptr;
        }
    }
    while (!targetQueue.empty()) {
        ISyncTarget *target = targetQueue.front();
        targetQueue.pop_front();
        SyncOperation *tmpOperation = nullptr;
        target->GetSyncOperation(tmpOperation);
        if (tmpOperation == nullptr) {
            LOGE("[ClearAllSyncTask] tmpOperation is nullptr");
            continue; // not exit this scene
        }
        LOGI("[SingleVerSyncTaskContext] killing syncId=%d,dev=%s", tmpOperation->GetSyncId(), STR_MASK(deviceId_));
        if (target->IsAutoSync()) {
            tmpOperation->SetStatus(deviceId_, SyncOperation::OP_FINISHED_ALL);
        } else {
            tmpOperation->SetStatus(deviceId_, SyncOperation::OP_COMM_ABNORMAL);
        }
        if (tmpOperation->CheckIsAllFinished()) {
            tmpOperation->Finished();
        }
        delete target;
        target = nullptr;
        RefObject::DecObjRef(tmpOperation);
    }
    if (GetTaskExecStatus() == SyncTaskContext::RUNNING) {
        // clear syncing task.
        stateMachine_->CommErrAbort();
    }
    // reset last push status for sync merge
    ResetLastPushTaskStatus();
}

void SingleVerSyncTaskContext::EnableClearRemoteStaleData(bool enable)
{
    needClearRemoteStaleData_ = enable;
}

bool SingleVerSyncTaskContext::IsNeedClearRemoteStaleData() const
{
    return needClearRemoteStaleData_;
}

bool SingleVerSyncTaskContext::StartFeedDogForSync(uint32_t time, SyncDirectionFlag flag)
{
    return stateMachine_->StartFeedDogForSync(time, flag);
}

void SingleVerSyncTaskContext::StopFeedDogForSync(SyncDirectionFlag flag)
{
    stateMachine_->StopFeedDogForSync(flag);
}

bool SingleVerSyncTaskContext::IsReceiveWaterMarkErr() const
{
    return isReceiveWaterMarkErr_;
}

void SingleVerSyncTaskContext::SetReceiveWaterMarkErr(bool isErr)
{
    isReceiveWaterMarkErr_ = isErr;
}

void SingleVerSyncTaskContext::SetRemoteSeccurityOption(SecurityOption secOption)
{
    std::lock_guard<std::mutex> autoLock(securityOptionMutex_);
    remoteSecOption_ = secOption;
}

SecurityOption SingleVerSyncTaskContext::GetRemoteSeccurityOption() const
{
    std::lock_guard<std::mutex> autoLock(securityOptionMutex_);
    return remoteSecOption_;
}

void SingleVerSyncTaskContext::SetReceivcPermitCheck(bool isChecked)
{
    isReceivcPermitChecked_ = isChecked;
}

bool SingleVerSyncTaskContext::GetReceivcPermitCheck() const
{
    return isReceivcPermitChecked_;
}

void SingleVerSyncTaskContext::SetSendPermitCheck(bool isChecked)
{
    isSendPermitChecked_ = isChecked;
}

bool SingleVerSyncTaskContext::GetSendPermitCheck() const
{
    return isSendPermitChecked_;
}

bool SingleVerSyncTaskContext::IsSkipTimeoutError(int errCode) const
{
    if (errCode == -E_TIMEOUT && IsSyncTaskNeedRetry() && (GetRetryTime() < GetSyncRetryTimes())) {
        LOGE("[SingleVerSyncTaskContext] send message timeout error occurred");
        return true;
    } else {
        return false;
    }
}

bool SingleVerSyncTaskContext::FindResponseSyncTarget(uint32_t responseSessionId) const
{
    std::lock_guard<std::mutex> lock(targetQueueLock_);
    auto iter = std::find_if(responseTargetQueue_.begin(), responseTargetQueue_.end(),
        [responseSessionId](const ISyncTarget *target) {
            return target->GetResponseSessionId() == responseSessionId;
        });
    if (iter == responseTargetQueue_.end()) {
        return false;
    }
    return true;
}

void SingleVerSyncTaskContext::SetQuery(const QuerySyncObject &query)
{
    std::lock_guard<std::mutex> autoLock(queryMutex_);
    query_ = query;
}

QuerySyncObject SingleVerSyncTaskContext::GetQuery() const
{
    std::lock_guard<std::mutex> autoLock(queryMutex_);
    return query_;
}

void SingleVerSyncTaskContext::SetQuerySync(bool isQuerySync)
{
    isQuerySync_ = isQuerySync;
}

bool SingleVerSyncTaskContext::IsQuerySync() const
{
    return isQuerySync_;
}

std::set<CompressAlgorithm> SingleVerSyncTaskContext::GetRemoteCompressAlgo() const
{
    std::lock_guard<std::mutex> autoLock(remoteDbAbilityLock_);
    std::set<CompressAlgorithm> compressAlgoSet;
    for (const auto &algo : SyncConfig::COMPRESSALGOMAP) {
        if (remoteDbAbility_.GetAbilityItem(algo.second) == SUPPORT_MARK) {
            compressAlgoSet.insert(static_cast<CompressAlgorithm>(algo.first));
        }
    }
    return compressAlgoSet;
}

std::string SingleVerSyncTaskContext::GetRemoteCompressAlgoStr() const
{
    static std::map<CompressAlgorithm, std::string> algoMap = {{CompressAlgorithm::ZLIB, "zlib"}};
    std::set<CompressAlgorithm> remoteCompressAlgoSet = GetRemoteCompressAlgo();
    if (remoteCompressAlgoSet.empty()) {
        return "none";
    }
    std::string currentAlgoStr;
    for (const auto &algo : remoteCompressAlgoSet) {
        auto iter = algoMap.find(algo);
        if (iter != algoMap.end()) {
            currentAlgoStr += algoMap[algo] + ",";
        }
    }
    return currentAlgoStr.substr(0, currentAlgoStr.length() - 1);
}

void SingleVerSyncTaskContext::SetDbAbility(DbAbility &remoteDbAbility)
{
    {
        std::lock_guard<std::mutex> autoLock(remoteDbAbilityLock_);
        remoteDbAbility_ = remoteDbAbility;
    }
    LOGI("[SingleVerSyncTaskContext] set dev=%s compressAlgo=%s, IsSupAllPredicateQuery=%u,"
        "IsSupSubscribeQuery=%u, inKeys=%u",
        STR_MASK(GetDeviceId()), GetRemoteCompressAlgoStr().c_str(),
        remoteDbAbility.GetAbilityItem(SyncConfig::ALLPREDICATEQUERY),
        remoteDbAbility.GetAbilityItem(SyncConfig::SUBSCRIBEQUERY),
        remoteDbAbility.GetAbilityItem(SyncConfig::INKEYS_QUERY));
}

CompressAlgorithm SingleVerSyncTaskContext::ChooseCompressAlgo() const
{
    std::set<CompressAlgorithm> remoteAlgo = GetRemoteCompressAlgo();
    if (remoteAlgo.empty()) {
        return CompressAlgorithm::NONE;
    }
    std::set<CompressAlgorithm> localAlgorithmSet;
    (void)(static_cast<SyncGenericInterface *>(syncInterface_))->GetCompressionAlgo(localAlgorithmSet);
    std::set<CompressAlgorithm> algoIntersection;
    set_intersection(remoteAlgo.begin(), remoteAlgo.end(), localAlgorithmSet.begin(), localAlgorithmSet.end(),
        inserter(algoIntersection, algoIntersection.begin()));
    if (algoIntersection.empty()) {
        return CompressAlgorithm::NONE;
    }
    return *(algoIntersection.begin());
}

bool SingleVerSyncTaskContext::IsNotSupportAbility(const AbilityItem &abilityItem) const
{
    std::lock_guard<std::mutex> autoLock(remoteDbAbilityLock_);
    return remoteDbAbility_.GetAbilityItem(abilityItem) != SUPPORT_MARK;
}

void SingleVerSyncTaskContext::SetSubscribeManager(std::shared_ptr<SubscribeManager> &subManager)
{
    subManager_ = subManager;
}

std::shared_ptr<SubscribeManager> SingleVerSyncTaskContext::GetSubscribeManager() const
{
    return subManager_;
}
DEFINE_OBJECT_TAG_FACILITIES(SingleVerSyncTaskContext)

bool SingleVerSyncTaskContext::IsCurrentSyncTaskCanBeSkipped() const
{
    SyncOperation *operation = GetAndIncSyncOperation();
    bool res = IsCurrentSyncTaskCanBeSkippedInner(operation);
    RefObject::DecObjRef(operation);
    return res;
}

void SingleVerSyncTaskContext::SaveLastPushTaskExecStatus(int finalStatus)
{
    if (IsTargetQueueEmpty()) {
        LOGD("sync que is empty, reset last push status");
        ResetLastPushTaskStatus();
        return;
    }
    if (mode_ == SyncModeType::PUSH || mode_ == SyncModeType::PUSH_AND_PULL || mode_ == SyncModeType::RESPONSE_PULL) {
        lastFullSyncTaskStatus_ = finalStatus;
    } else if (mode_ == SyncModeType::QUERY_PUSH || mode_ == SyncModeType::QUERY_PUSH_PULL) {
        std::lock_guard<std::mutex> autoLock(queryTaskStatusMutex_);
        lastQuerySyncTaskStatusMap_[syncOperation_->GetQueryId()] = finalStatus;
    }
}

int SingleVerSyncTaskContext::GetCorrectedSendWaterMarkForCurrentTask(const SyncOperation *operation,
    uint64_t &waterMark) const
{
    if (operation != nullptr && operation->IsQuerySync()) {
        LOGD("Is QuerySync");
        int errCode = static_cast<SingleVerSyncStateMachine *>(stateMachine_)->GetSendQueryWaterMark(
            operation->GetQueryId(), deviceId_,
            lastFullSyncTaskStatus_ == SyncOperation::OP_FINISHED_ALL, waterMark);
        if (errCode != E_OK) {
            return errCode;
        }
    } else {
        LOGD("Not QuerySync");
        static_cast<SingleVerSyncStateMachine *>(stateMachine_)->GetLocalWaterMark(deviceId_, waterMark);
    }
    return E_OK;
}

void SingleVerSyncTaskContext::ResetLastPushTaskStatus()
{
    lastFullSyncTaskStatus_ = SyncOperation::OP_WAITING;
    std::lock_guard<std::mutex> autoLock(queryTaskStatusMutex_);
    lastQuerySyncTaskStatusMap_.clear();
}

void SingleVerSyncTaskContext::SetCommNormal(bool isCommNormal)
{
    isCommNormal_ = isCommNormal;
}

bool SingleVerSyncTaskContext::IsCurrentSyncTaskCanBeSkippedInner(const SyncOperation *operation) const
{
    if (mode_ == SyncModeType::PUSH) {
        if (lastFullSyncTaskStatus_ != SyncOperation::OP_FINISHED_ALL) {
            return false;
        }
        if (operation == nullptr) {
            return true;
        }
    } else if (mode_ == SyncModeType::QUERY_PUSH) {
        if (operation == nullptr) {
            return true;
        }
        std::lock_guard<std::mutex> autoLock(queryTaskStatusMutex_);
        auto it = lastQuerySyncTaskStatusMap_.find(operation->GetQueryId());
        if (it == lastQuerySyncTaskStatusMap_.end()) {
            // no last query_push and push
            if (lastFullSyncTaskStatus_ != SyncOperation::OP_FINISHED_ALL) {
                LOGD("no prev query push or successful prev push");
                return false;
            }
        } else {
            if (it->second != SyncOperation::OP_FINISHED_ALL) {
                LOGD("last query push status = %d.", it->second);
                return false;
            }
        }
    } else {
        return false;
    }

    Timestamp maxTimestampInDb;
    syncInterface_->GetMaxTimestamp(maxTimestampInDb);
    uint64_t localWaterMark = 0;
    int errCode = GetCorrectedSendWaterMarkForCurrentTask(operation, localWaterMark);
    if (errCode != E_OK) {
        LOGE("GetLocalWaterMark in state machine failed: %d", errCode);
        return false;
    }
    if (localWaterMark > maxTimestampInDb) {
        LOGI("skip current push task, deviceId_ = %s, localWaterMark = %" PRIu64 ", maxTimestampInDb = %" PRIu64,
            STR_MASK(deviceId_), localWaterMark, maxTimestampInDb);
        return true;
    }
    return false;
}

void SingleVerSyncTaskContext::StartFeedDogForGetData(uint32_t sessionId)
{
    stateMachine_->StartFeedDogForGetData(sessionId);
}

void SingleVerSyncTaskContext::StopFeedDogForGetData()
{
    stateMachine_->StopFeedDogForGetData();
}
} // namespace DistributedDB