xref: /ohos5.0/foundation/distributeddatamgr/kv_store/frameworks/libs/distributeddb/communicator/src/network_adapter.cpp

/*
 * 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 "network_adapter.h"
#include "db_constant.h"
#include "db_common.h"
#include "db_errno.h"
#include "log_print.h"
#include "runtime_context.h"

namespace DistributedDB {
namespace {
const std::string DEFAULT_PROCESS_LABEL = "Distributeddb_Anonymous_Process";
const std::string SCHEDULE_QUEUE_TAG = "NetworkAdapter";
}

NetworkAdapter::NetworkAdapter()
    : processLabel_(DEFAULT_PROCESS_LABEL), processCommunicator_(nullptr)
{
}

NetworkAdapter::NetworkAdapter(const std::string &inProcessLabel)
    : processLabel_(inProcessLabel), processCommunicator_(nullptr)
{
}

NetworkAdapter::NetworkAdapter(const std::string &inProcessLabel,
    const std::shared_ptr<IProcessCommunicator> &inCommunicator)
    : processLabel_(inProcessLabel), processCommunicator_(inCommunicator)
{
}

NetworkAdapter::~NetworkAdapter()
{
}

int NetworkAdapter::StartAdapter()
{
    LOGI("[NAdapt][Start] Enter, ProcessLabel=%s.", STR_MASK(DBCommon::TransferStringToHex(processLabel_)));
    if (processLabel_.empty()) {
        return -E_INVALID_ARGS;
    }
    if (!processCommunicator_) {
        LOGE("[NAdapt][Start] ProcessCommunicator not be designated yet.");
        return -E_INVALID_ARGS;
    }
    DBStatus errCode = processCommunicator_->Start(processLabel_);
    if (errCode != DBStatus::OK) {
        LOGE("[NAdapt][Start] Start Fail, errCode=%d.", static_cast<int>(errCode));
        return -E_PERIPHERAL_INTERFACE_FAIL;
    }
    errCode = processCommunicator_->RegOnDataReceive(
        [this](const DeviceInfos &srcDevInfo, const uint8_t *data, uint32_t length) {
            OnDataReceiveHandler(srcDevInfo, data, length);
        });
    if (errCode != DBStatus::OK) {
        LOGE("[NAdapt][Start] RegOnDataReceive Fail, errCode=%d.", static_cast<int>(errCode));
        // DO ROLLBACK
        errCode = processCommunicator_->Stop();
        LOGI("[NAdapt][Start] ROLLBACK: Stop errCode=%d.", static_cast<int>(errCode));
        return -E_PERIPHERAL_INTERFACE_FAIL;
    }
    errCode = processCommunicator_->RegOnDeviceChange([this](const DeviceInfos &devInfo, bool isOnline) {
        OnDeviceChangeHandler(devInfo, isOnline);
    });
    if (errCode != DBStatus::OK) {
        LOGE("[NAdapt][Start] RegOnDeviceChange Fail, errCode=%d.", static_cast<int>(errCode));
        // DO ROLLBACK
        errCode = processCommunicator_->RegOnDataReceive(nullptr);
        LOGI("[NAdapt][Start] ROLLBACK: UnRegOnDataReceive errCode=%d.", static_cast<int>(errCode));
        errCode = processCommunicator_->Stop();
        LOGI("[NAdapt][Start] ROLLBACK: Stop errCode=%d.", static_cast<int>(errCode));
        return -E_PERIPHERAL_INTERFACE_FAIL;
    }
    processCommunicator_->RegOnSendAble([this](const DeviceInfos &devInfo, int softBusErrCode) {
        if (onSendableHandle_ != nullptr) {
            onSendableHandle_(devInfo.identifier, softBusErrCode);
        }
    });
    // These code is compensation for the probable defect of IProcessCommunicator implementation.
    // As described in the agreement, for the missed online situation, we search for the online devices at beginning.
    // OnDeviceChangeHandler is reused to check the existence of peer process.
    // Since at this point, the CommunicatorAggregator had not been fully initialized,
    // We need an async task which bring about dependency on the lifecycle of this NetworkAdapter Object.
    SearchOnlineRemoteDeviceAtStartup();
    LOGI("[NAdapt][Start] Exit.");
    return E_OK;
}

// StartAdapter and StopAdapter are all innerly called by ICommunicatorAggregator
// If StopAdapter is called, the StartAdapter must have been called successfully before,
// so processCommunicator_ won't be null
void NetworkAdapter::StopAdapter()
{
    LOGI("[NAdapt][Stop] Enter, ProcessLabel=%s.", processLabel_.c_str());
    processCommunicator_->RegOnSendAble(nullptr);
    DBStatus errCode = processCommunicator_->RegOnDeviceChange(nullptr);
    if (errCode != DBStatus::OK) {
        LOGE("[NAdapt][Stop] UnRegOnDeviceChange Fail, errCode=%d.", static_cast<int>(errCode));
    }
    errCode = processCommunicator_->RegOnDataReceive(nullptr);
    if (errCode != DBStatus::OK) {
        LOGE("[NAdapt][Stop] UnRegOnDataReceive Fail, errCode=%d.", static_cast<int>(errCode));
    }
    errCode = processCommunicator_->Stop();
    if (errCode != DBStatus::OK) {
        LOGE("[NAdapt][Stop] Stop Fail, errCode=%d.", static_cast<int>(errCode));
    }
    // We don't reset the shared_ptr of commProvider here, the release of commProvider is done by deconstruct of adapter
    // In this way, the adapter can be start again after stop it, since it still hold the an valid commProvider
    // The async task is dependent on this Object. we have to wait until all async task finished.
    LOGI("[NAdapt][Stop] Wait all async task done.");
    std::unique_lock<std::mutex> asyncTaskDoneLock(asyncTaskDoneMutex_);
    asyncTaskDoneCv_.wait(asyncTaskDoneLock, [this] { return pendingAsyncTaskCount_ <= 0; });
    LOGI("[NAdapt][Stop] Exit.");
}

namespace {
uint32_t CheckAndAdjustMtuSize(uint32_t inMtuSize)
{
    if (inMtuSize < DBConstant::MIN_MTU_SIZE) {
        return DBConstant::MIN_MTU_SIZE;
    } else if (inMtuSize > DBConstant::MAX_MTU_SIZE) {
        return DBConstant::MAX_MTU_SIZE;
    } else {
        return (inMtuSize - (inMtuSize % sizeof(uint64_t))); // Octet alignment
    }
}

uint32_t CheckAndAdjustTimeout(uint32_t inTimeout)
{
    if (inTimeout < DBConstant::MIN_TIMEOUT) {
        return DBConstant::MIN_TIMEOUT;
    } else if (inTimeout > DBConstant::MAX_TIMEOUT) {
        return DBConstant::MAX_TIMEOUT;
    } else {
        return inTimeout;
    }
}
}

uint32_t NetworkAdapter::GetMtuSize()
{
    std::lock_guard<std::mutex> mtuSizeLockGuard(mtuSizeMutex_);
    if (!isMtuSizeValid_) {
        mtuSize_ = processCommunicator_->GetMtuSize();
        LOGD("[NAdapt][GetMtu] mtuSize=%" PRIu32 ".", mtuSize_);
        mtuSize_ = CheckAndAdjustMtuSize(mtuSize_);
        isMtuSizeValid_ = true;
    }
    return mtuSize_;
}

uint32_t NetworkAdapter::GetMtuSize(const std::string &target)
{
#ifndef OMIT_MTU_CACHE
    DeviceInfos devInfo;
    devInfo.identifier = target;
    uint32_t oriMtuSize = processCommunicator_->GetMtuSize(devInfo);
    return CheckAndAdjustMtuSize(oriMtuSize);
#else
    std::lock_guard<std::mutex> mtuSizeLockGuard(mtuSizeMutex_);
    if (devMapMtuSize_.count(target) == 0) {
        DeviceInfos devInfo;
        devInfo.identifier = target;
        uint32_t oriMtuSize = processCommunicator_->GetMtuSize(devInfo);
        LOGD("[NAdapt][GetMtu] mtuSize=%" PRIu32 " of target=%s{private}.", oriMtuSize, target.c_str());
        devMapMtuSize_[target] = CheckAndAdjustMtuSize(oriMtuSize);
    }
    return devMapMtuSize_[target];
#endif
}

uint32_t NetworkAdapter::GetTimeout()
{
    uint32_t timeout = processCommunicator_->GetTimeout();
    LOGD("[NAdapt][GetTimeout] timeout_=%" PRIu32 " ms.", timeout);
    return CheckAndAdjustTimeout(timeout);
}

uint32_t NetworkAdapter::GetTimeout(const std::string &target)
{
    DeviceInfos devInfos;
    devInfos.identifier = target;
    uint32_t timeout = processCommunicator_->GetTimeout(devInfos);
    LOGD("[NAdapt][GetTimeout] timeout=%" PRIu32 " ms of target=%s{private}.", timeout, target.c_str());
    return CheckAndAdjustTimeout(timeout);
}

int NetworkAdapter::GetLocalIdentity(std::string &outTarget)
{
    std::lock_guard<std::mutex> identityLockGuard(identityMutex_);
    DeviceInfos devInfo = processCommunicator_->GetLocalDeviceInfos();
    if (devInfo.identifier.empty()) {
        LOGE("[NetworkAdapter] Get empty local id");
        return -E_PERIPHERAL_INTERFACE_FAIL;
    }
    if (devInfo.identifier != localIdentity_) {
        LOGI("[NAdapt][GetLocal] localIdentity=%s{private}.", devInfo.identifier.c_str());
    }
    localIdentity_ = devInfo.identifier;
    outTarget = localIdentity_;
    return E_OK;
}

int NetworkAdapter::SendBytes(const std::string &dstTarget, const uint8_t *bytes, uint32_t length, uint32_t totalLength)
{
    if (bytes == nullptr || length == 0) {
        return -E_INVALID_ARGS;
    }
    LOGI("[NAdapt][SendBytes] Enter, to=%s{private}, length=%u, totalLength=%u", dstTarget.c_str(), length,
        totalLength);
    DeviceInfos dstDevInfo;
    dstDevInfo.identifier = dstTarget;
    DBStatus errCode = processCommunicator_->SendData(dstDevInfo, bytes, length,
        totalLength > length ? totalLength : length);
    if (errCode == DBStatus::RATE_LIMIT) {
        LOGD("[NAdapt][SendBytes] rate limit!");
        return -E_WAIT_RETRY;
    }
    if (errCode != DBStatus::OK) {
        LOGE("[NAdapt][SendBytes] SendData Fail, errCode=%d.", static_cast<int>(errCode));
        // These code is compensation for the probable defect of IProcessCommunicator implementation.
        // As described in the agreement, for the missed offline situation, we check if still online at send fail.
        // OnDeviceChangeHandler is reused but check the existence of peer process is done outerly.
        // Since this thread is the sending_thread of the CommunicatorAggregator,
        // We need an async task which bring about dependency on the lifecycle of this NetworkAdapter Object.
        CheckDeviceOfflineAfterSendFail(dstDevInfo);
        return static_cast<int>(errCode);
    }
    return E_OK;
}

int NetworkAdapter::RegBytesReceiveCallback(const BytesReceiveCallback &onReceive, const Finalizer &inOper)
{
    std::lock_guard<std::mutex> onReceiveLockGard(onReceiveMutex_);
    return RegCallBack(onReceive, onReceiveHandle_, inOper, onReceiveFinalizer_);
}

int NetworkAdapter::RegTargetChangeCallback(const TargetChangeCallback &onChange, const Finalizer &inOper)
{
    std::lock_guard<std::mutex> onChangeLockGard(onChangeMutex_);
    return RegCallBack(onChange, onChangeHandle_, inOper, onChangeFinalizer_);
}

int NetworkAdapter::RegSendableCallback(const SendableCallback &onSendable, const Finalizer &inOper)
{
    std::lock_guard<std::mutex> onSendableLockGard(onSendableMutex_);
    return RegCallBack(onSendable, onSendableHandle_, inOper, onSendableFinalizer_);
}

void NetworkAdapter::OnDataReceiveHandler(const DeviceInfos &srcDevInfo, const uint8_t *data, uint32_t length)
{
    if (data == nullptr || length == 0) {
        LOGE("[NAdapt][OnDataRecv] data nullptr or length = %u.", length);
        return;
    }
    uint32_t headLength = 0;
    std::vector<std::string> userId;
    DBStatus errCode = processCommunicator_->CheckAndGetDataHeadInfo(data, length, headLength, userId);
    LOGI("[NAdapt][OnDataRecv] Enter, from=%s{private}, extendHeadLength=%u, totalLength=%u",
        srcDevInfo.identifier.c_str(), headLength, length);
    if (errCode == NO_PERMISSION) {
        LOGI("[NAdapt][OnDataRecv] userId dismatched, drop packet");
        return;
    }
    if (headLength >= length) {
        LOGW("[NAdapt][OnDataRecv] head len is too big, drop packet");
        return;
    }
    {
        std::lock_guard<std::mutex> onReceiveLockGard(onReceiveMutex_);
        if (!onReceiveHandle_) {
            LOGE("[NAdapt][OnDataRecv] onReceiveHandle invalid.");
            return;
        }
        std::string currentUserId;
        if (userId.size() >= 1) {
            currentUserId = userId[0];
        }
        onReceiveHandle_(srcDevInfo.identifier, data + headLength, length - headLength, currentUserId);
    }
    // These code is compensation for the probable defect of IProcessCommunicator implementation.
    // As described in the agreement, for the missed online situation, we check the source dev when received.
    // OnDeviceChangeHandler is reused to check the existence of peer process.
    // Since this thread is the callback_thread of IProcessCommunicator, we do this check task directly in this thread.
    CheckDeviceOnlineAfterReception(srcDevInfo);
}

void NetworkAdapter::OnDeviceChangeHandler(const DeviceInfos &devInfo, bool isOnline)
{
    LOGI("[NAdapt][OnDeviceChange] Enter, dev=%s{private}, isOnline=%d", devInfo.identifier.c_str(), isOnline);
    // These code is compensation for the probable defect of IProcessCommunicator implementation.
    // As described in the agreement, for the mistake online situation, we check the existence of peer process.
    // The IProcessCommunicator implementation guarantee that no mistake offline will happen.
    if (isOnline) {
        if (!processCommunicator_->IsSameProcessLabelStartedOnPeerDevice(devInfo)) {
            LOGI("[NAdapt][OnDeviceChange] Detect Not Really Online.");
            std::lock_guard<std::mutex> onlineRemoteDevLockGuard(onlineRemoteDevMutex_);
            onlineRemoteDev_.erase(devInfo.identifier);
            return;
        }
        std::lock_guard<std::mutex> onlineRemoteDevLockGuard(onlineRemoteDevMutex_);
        onlineRemoteDev_.insert(devInfo.identifier);
    } else {
        std::lock_guard<std::mutex> onlineRemoteDevLockGuard(onlineRemoteDevMutex_);
        onlineRemoteDev_.erase(devInfo.identifier);
    }
    // End compensation, do callback.
    std::lock_guard<std::mutex> onChangeLockGard(onChangeMutex_);
    if (!onChangeHandle_) {
        LOGE("[NAdapt][OnDeviceChange] onChangeHandle_ invalid.");
        return;
    }
    onChangeHandle_(devInfo.identifier, isOnline);
}

void NetworkAdapter::SearchOnlineRemoteDeviceAtStartup()
{
    std::vector<DeviceInfos> onlineDev = processCommunicator_->GetRemoteOnlineDeviceInfosList();
    LOGI("[NAdapt][SearchOnline] onlineDev count = %zu.", onlineDev.size());
    if (onlineDev.empty()) {
        return;
    }
    pendingAsyncTaskCount_.fetch_add(1);
    // Note: onlineDev should be captured by value (must not by reference)
    TaskAction callbackTask = [onlineDev, this]() {
        LOGI("[NAdapt][SearchOnline] Begin Callback In Async Task.");
        std::string localIdentity;
        GetLocalIdentity(localIdentity); // It doesn't matter if getlocal fail and localIdentity be an empty string
        for (auto &entry : onlineDev) {
            if (entry.identifier == localIdentity) {
                LOGW("[NAdapt][SearchOnline] Detect Local Device in Remote Device List.");
                continue;
            }
            OnDeviceChangeHandler(entry, true);
        }
        pendingAsyncTaskCount_.fetch_sub(1);
        asyncTaskDoneCv_.notify_all();
        LOGI("[NAdapt][SearchOnline] End Callback In Async Task.");
    };
    // Use ScheduleQueuedTask to keep order
    int errCode = RuntimeContext::GetInstance()->ScheduleQueuedTask(SCHEDULE_QUEUE_TAG, callbackTask);
    if (errCode != E_OK) {
        LOGE("[NAdapt][SearchOnline] ScheduleQueuedTask failed, errCode = %d.", errCode);
        pendingAsyncTaskCount_.fetch_sub(1);
        asyncTaskDoneCv_.notify_all();
    }
}

void NetworkAdapter::CheckDeviceOnlineAfterReception(const DeviceInfos &devInfo)
{
    bool isAlreadyOnline = true;
    {
        std::lock_guard<std::mutex> onlineRemoteDevLockGuard(onlineRemoteDevMutex_);
        if (onlineRemoteDev_.count(devInfo.identifier) == 0) {
            isAlreadyOnline = false;
        }
    }

    // Seem offline but receive data from it, let OnDeviceChangeHandler check whether it is really online
    if (!isAlreadyOnline) {
        OnDeviceChangeHandler(devInfo, true);
    }
}

void NetworkAdapter::CheckDeviceOfflineAfterSendFail(const DeviceInfos &devInfo)
{
    // Note: only the identifier field of devInfo is valid, enough to call IsSameProcessLabelStartedOnPeerDevice
    bool isAlreadyOffline = true;
    {
        std::lock_guard<std::mutex> onlineRemoteDevLockGuard(onlineRemoteDevMutex_);
        if (onlineRemoteDev_.count(devInfo.identifier) != 0) {
            isAlreadyOffline = false;
        }
    }

    // Seem online but send fail, we have to check whether still online
    if (!isAlreadyOffline) {
        if (!processCommunicator_->IsSameProcessLabelStartedOnPeerDevice(devInfo)) {
            LOGW("[NAdapt][CheckAfterSend] Missed Offline Detected.");
            {
                // Mark this device not online immediately to avoid repeatedly miss-offline detect when send continually
                std::lock_guard<std::mutex> onlineRemoteDevLockGuard(onlineRemoteDevMutex_);
                onlineRemoteDev_.erase(devInfo.identifier);
            }
            pendingAsyncTaskCount_.fetch_add(1);
            // Note: devInfo should be captured by value (must not by reference)
            TaskAction callbackTask = [devInfo, this]() {
                LOGI("[NAdapt][CheckAfterSend] In Async Task, devInfo=%s{private}.", devInfo.identifier.c_str());
                OnDeviceChangeHandler(devInfo, false);
                pendingAsyncTaskCount_.fetch_sub(1);
                asyncTaskDoneCv_.notify_all();
            };
            // Use ScheduleQueuedTask to keep order
            int errCode = RuntimeContext::GetInstance()->ScheduleQueuedTask(SCHEDULE_QUEUE_TAG, callbackTask);
            if (errCode != E_OK) {
                LOGE("[NAdapt][CheckAfterSend] ScheduleQueuedTask failed, errCode = %d.", errCode);
                pendingAsyncTaskCount_.fetch_sub(1);
                asyncTaskDoneCv_.notify_all();
            }
        }
    }
}

bool NetworkAdapter::IsDeviceOnline(const std::string &device)
{
    std::lock_guard<std::mutex> onlineRemoteDevLockGuard(onlineRemoteDevMutex_);
    return (onlineRemoteDev_.find(device) != onlineRemoteDev_.end());
}

std::shared_ptr<ExtendHeaderHandle> NetworkAdapter::GetExtendHeaderHandle(const ExtendInfo &paramInfo)
{
    return processCommunicator_->GetExtendHeaderHandle(paramInfo);
}
} // namespace DistributedDB