eUi/rpcModule/rpc_finsTcp_boost_V2/rpc_finsTcp.cpp

#include "rpc_finsTcp.h"
#include "../RpcExeProcessThread.h"

#include <QDebug>
#include <QElapsedTimer>
#include <QMutexLocker>

rpcModuleBase* create()
{
    qDebug() << "rpc_finsTcp create success!";
    return new rpc_finsTcp;
}

void destory(rpcModuleBase* ptr)
{
    delete ptr;
    qDebug() << "rpc_finsTcp destory success!";
}

rpc_finsTcp::rpc_finsTcp() = default;

int rpc_finsTcp::assignThreadIndex()
{
    QMutexLocker locker(&threadUseCacheMutex);
    const std::thread::id currentThreadId = std::this_thread::get_id();
    const auto cached = threadUseCache.constFind(currentThreadId);
    if (cached != threadUseCache.constEnd()) {
        return cached.value();
    }

    const int assigned = threadUseIndex.fetch_add(1);
    threadUseCache.insert(currentThreadId, assigned % qMax(1, threadCount));
    return assigned % qMax(1, threadCount);
}

QString rpc_finsTcp::moduleNameForType(int type, int threadId) const
{
    switch (type) {
    case finsRead:
        return "finstcp_instance_0";
    case finsWrite:
        return "finstcp_instance_" + QString::number(threadId + 1);
    case finsWrite_1:
        return "finstcp_instance_1";
    case finsWrite_2:
        return "finstcp_instance_2";
    case finsWrite_3:
        return "finstcp_instance_3";
    case finsWrite_4:
        return "finstcp_instance_4";
    default:
        return "finstcp_instance_0";
    }
}

rpcRequest rpc_finsTcp::packageRequest(quint16 command, const QByteArray& text,
                                       int type)
{
    rpcRequest request;
    request.command = command == RpcFrameCodec::CmdInit
        ? ("init_" + m_IP + "_" + QString::number(m_port)).toUtf8()
        : text;
    request.id = idCounter.fetch_add(1);
    request.threadId = assignThreadIndex();
    const QString module = moduleNameForType(type, request.threadId);
    request.requestByte = RpcFrameCodec::build(
        request.id,
        command,
        RpcFrameCodec::requestPayload(module, QString::fromUtf8(text)));
    return request;
}

rpcRequest rpc_finsTcp::packageExecuteRequest(const QByteArray& command, int type)
{
    return packageRequest(RpcFrameCodec::CmdExecute, command, type);
}

rpcRequest rpc_finsTcp::packageInitRequest(int type)
{
    return packageRequest(
        RpcFrameCodec::CmdInit,
        (m_IP + "\n" + QString::number(m_port)).toUtf8(),
        type);
}

QByteArray rpc_finsTcp::executeReadCommand(QByteArray& input)
{
    RpcClient* client = nullptr;
    {
        QMutexLocker locker(&mutex);
        client = readClients.value(input);
        if (!client) {
            client = new RpcClient("127.0.0.1", 9000);
            if (client->_connectionStatus != Connected) {
                delete client;
                return QByteArray();
            }
            if (!client->initialize("finstcp_instance_0", m_IP, m_port)) {
                qWarning() << "[rpc_finsTcp_boost_V2] read client initialization failed"
                           << m_IP << m_port;
                delete client;
                return QByteArray();
            }
            readClients.insert(input, client);
        }
    }
    return client->execute("finstcp_instance_0", input);
}

bool rpc_finsTcp::executeWriteCommand(QByteArray& input)
{
    if (input == "null") {
        return true;
    }
    if (input.startsWith("init_")) {
        return parseInitCommand(input);
    }

    const QList<QByteArray> parts = input.split('_');
    if (parts.size() < 4 || parts.at(0) == "read") {
        qWarning() << "[rpc_finsTcp_boost_V2] asynchronous write rejected: invalid command" << input;
        return false;
    }
    const rpcRequest request = packageExecuteRequest(input, finsWrite);
    return enqueueWrite(request);
}

rpc_finsTcp::~rpc_finsTcp()
{
    stopWriteWorkers();
    qDeleteAll(readClients);
    readClients.clear();
    stopRpcServer();
    qDebug() << "~rpc_finsTcp success!";
}

bool rpc_finsTcp::readData(std::vector<cv::Mat>& vec_mat, QByteArray& data)
{
    Q_UNUSED(vec_mat);
    RpcClient* client = nullptr;
    {
        QMutexLocker locker(&mutex);
        client = readClients.value(data);
        if (!client) {
            client = new RpcClient("127.0.0.1", 9000);
            if (client->_connectionStatus != Connected) {
                delete client;
                return false;
            }
            client->initialize("finstcp_instance_0", m_IP, m_port);
            readClients.insert(data, client);
        }
    }

    const QByteArray response = client->execute("finstcp_instance_0", data);
    if (response.isEmpty()) {
        return false;
    }
    data = response;
    return true;
}

bool rpc_finsTcp::writeData(std::vector<cv::Mat>& vec_mat, QByteArray& data)
{
    Q_UNUSED(vec_mat);
    const QList<QByteArray> parts = data.split('_');
    if (parts.size() < 3) {
        qWarning() << "[rpc_finsTcp_boost_V2] asynchronous write rejected: invalid command" << data;
        return false;
    }

    if (parts.at(0) == "init") {
        if (!parseInitCommand(data)) {
            qWarning() << "[rpc_finsTcp_boost_V2] asynchronous write rejected: invalid init command" << data;
            return false;
        }
    } else if (data.contains("read")) {
        readDataQueue.append(data);
    } else {
        if (data.contains("null")) {
            return true;
        }
        if (data.contains("Heart")) {
            const rpcRequest request = packageExecuteRequest(data, finsWrite);
            return enqueueWrite(request);
        }
        const rpcRequest request = packageExecuteRequest(data, finsWrite);
        if (!enqueueWrite(request)) {
            return false;
        }
    }
    return true;
}

bool rpc_finsTcp::initParas(QByteArray& data)
{
    // The initialization text is supplied by the subModuleBase caller.  The
    // thread count is optional; init_<ip>_<port> is the canonical form.
    if (!parseInitCommand(data)) {
        return false;
    }

    if (!startRpcServer()) {
        return false;
    }

    // Each configured write instance has its own FIFO and dedicated worker.
    writeClients.resize(threadCount);
    for (int i = 0; i < threadCount; ++i) {
        writeClients[i] = new RpcClientWorker();
        if (!writeClients[i]->startAndWait()) {
            qWarning() << "[rpc_finsTcp_boost_V2] write worker failed to connect"
                       << "worker=" << i;
            stopWriteWorkers();
            stopRpcServer();
            return false;
        }
    }
    for (int i = 0; i < threadCount; ++i) {
        const rpcRequest request = packageInitRequest(finsWrite_1 + i);
        if (!enqueueWriteToWorker(request, i)) {
            stopWriteWorkers();
            stopRpcServer();
            return false;
        }
    }
    return true;
}

bool rpc_finsTcp::startRpcServer()
{
    if (!processThread) {
        processThread = new RpcExeProcessThread();
    }
    if (processThread->startExe("post_rpc_server_1_2_3.exe")) {
        return true;
    }
    delete processThread;
    processThread = nullptr;
    return false;
}

void rpc_finsTcp::stopRpcServer()
{
    if (!processThread) {
        return;
    }
    processThread->stopExe();
    delete processThread;
    processThread = nullptr;
}

bool rpc_finsTcp::parseInitCommand(const QByteArray& input)
{
    const QList<QByteArray> parts = input.split('_');
    if (parts.size() < 3 || parts.at(0) != "init") {
        qCritical() << input << "expected init_ip_port[_threadCount]";
        return false;
    }

    bool portOk = false;
    const int port = parts.at(2).toInt(&portOk);
    if (parts.at(1).isEmpty() || !portOk || port <= 0 || port > 65535) {
        qCritical() << input << "invalid init parameters";
        return false;
    }

    int parsedThreadCount = 1;
    if (parts.size() >= 4 && !parts.at(3).isEmpty()) {
        bool countOk = false;
        const int count = parts.at(3).toInt(&countOk);
        if (!countOk || count <= 0) {
            qCritical() << input << "invalid init thread count";
            return false;
        }
        parsedThreadCount = qBound(1, count, 4);
    }

    m_IP = QString::fromUtf8(parts.at(1));
    m_port = port;
    threadCount = parsedThreadCount;
    return true;
}

bool rpc_finsTcp::run()
{
    return true;
}

RpcClient::RpcClient(const QString& host, quint16 port)
    : host_(host), port_(port)
{
    socket_.connectToHost(host_, port_);
    if (!socket_.waitForConnected(3000)) {
        qWarning() << "[RpcClient] Initial connection failed:" << socket_.errorString();
        _connectionStatus = Disconnected;
    } else {
        qDebug() << "[RpcClient] Connected to" << host_ << ":" << port_;
        _connectionStatus = Connected;
    }
}

bool RpcClient::reconnect()
{
    if (socket_.state() == QAbstractSocket::ConnectedState) {
        return true;
    }
    socket_.abort();
    codec_.clear();
    socket_.connectToHost(host_, port_);
    return socket_.waitForConnected(3000);
}

bool RpcClient::initialize(const QString& module, const QString& ip, int port)
{
    QMutexLocker locker(&mutex_);
    if (!reconnect()) {
        return false;
    }
    const quint32 id = idCounter_++;
    const QByteArray frame = RpcFrameCodec::build(
        id, RpcFrameCodec::CmdInit,
        RpcFrameCodec::requestPayload(module, ip + "\n" + QString::number(port)));
    RpcFrameCodec::Frame response;
    return sendFrameAndAwait(frame, id, response) &&
           response.cmd == RpcFrameCodec::RespOk;
}

bool RpcClient::isConnected(const QString& module)
{
    QMutexLocker locker(&mutex_);
    if (!reconnect()) {
        return false;
    }
    const quint32 id = idCounter_++;
    const QByteArray frame = RpcFrameCodec::build(
        id, RpcFrameCodec::CmdIsConnected,
        RpcFrameCodec::requestPayload(module));
    RpcFrameCodec::Frame response;
    if (!sendFrameAndAwait(frame, id, response) || response.cmd != RpcFrameCodec::RespOk) {
        return false;
    }
    return !response.payload.isEmpty() && response.payload.at(0) != '\0';
}

QByteArray RpcClient::execute(const QString& module, const QByteArray& command)
{
    QMutexLocker locker(&mutex_);
    if (!reconnect()) {
        return QByteArray();
    }
    const quint32 id = idCounter_++;
    const QByteArray frame = RpcFrameCodec::build(
        id, RpcFrameCodec::CmdExecute,
        RpcFrameCodec::requestPayload(module, QString::fromUtf8(command)));
    RpcFrameCodec::Frame response;
    if (!sendFrameAndAwait(frame, id, response) || response.cmd != RpcFrameCodec::RespOk) {
        qWarning() << "[RpcClient] execute request failed";
        return QByteArray();
    }
    return response.payload;
}

bool RpcClient::sendFrame(const QByteArray& request, quint32 id)
{
    QMutexLocker locker(&mutex_);
    if (!reconnect()) {
        qWarning() << "[RpcClient] asynchronous write failed: RPC server is not connected";
        return false;
    }
    RpcFrameCodec::Frame response;
    if (!sendFrameAndAwait(request, id, response)) {
        qWarning() << "[RpcClient] asynchronous write failed: response timeout or socket error"
                   << "requestId=" << id;
        return false;
    }
    if (response.cmd != RpcFrameCodec::RespOk) {
        qWarning() << "[RpcClient] asynchronous write rejected by server"
                   << "requestId=" << id
                   << "response=" << response.payload;
        return false;
    }
    return true;
}

bool RpcClient::waitAndReadResponse(quint32 expectId,
                                    RpcFrameCodec::Frame& response,
                                    int timeoutMs)
{
    QElapsedTimer timer;
    timer.start();
    while (timer.elapsed() < timeoutMs) {
        if (socket_.bytesAvailable() == 0) {
            const int remain = timeoutMs - static_cast<int>(timer.elapsed());
            if (!socket_.waitForReadyRead(qMin(remain, 1000))) {
                continue;
            }
        }
        const QByteArray chunk = socket_.readAll();
        if (!chunk.isEmpty()) {
            codec_.append(chunk);
        }
        const QList<RpcFrameCodec::Frame> frames = codec_.takeFrames();
        for (const RpcFrameCodec::Frame& frame : frames) {
            if (frame.id == expectId) {
                response = frame;
                return true;
            }
        }
    }
    return false;
}

bool RpcClient::sendFrameAndAwait(const QByteArray& frame, quint32 id,
                                  RpcFrameCodec::Frame& response, int timeoutMs)
{
    if (socket_.write(frame) == -1 || !socket_.waitForBytesWritten(1000)) {
        qWarning() << "[RpcClient] frame write failed";
        return false;
    }
    return waitAndReadResponse(id, response, timeoutMs);
}

RpcClientWorker::~RpcClientWorker()
{
    stopWorker();
}

bool RpcClientWorker::startAndWait(int timeoutMs)
{
    start();

    QMutexLocker locker(&stateMutex_);
    if (!initialized_) {
        initializedCondition_.wait(&stateMutex_, timeoutMs);
    }
    return initialized_ && connection_ == Connected;
}

bool RpcClientWorker::enqueue(const QByteArray& request, quint32 id,
                              const QByteArray& command)
{
    QMutexLocker locker(&queueMutex_);
    if (stopping_) {
        qWarning() << "write rejected: worker stopping"
                   << "requestId=" << id;
        return false;
    }
    if (!isReady()) {
        qWarning() << "write rejected: worker not ready"
                   << "requestId=" << id;
        return false;
    }

    queue_.enqueue({request, command, id});
    queueCondition_.wakeOne();
    return true;
}

bool RpcClientWorker::isReady() const
{
    QMutexLocker locker(&stateMutex_);
    return initialized_ && connection_ == Connected;
}

void RpcClientWorker::stopWorker()
{
    {
        QMutexLocker locker(&queueMutex_);
        if (stopping_) {
            return;
        }
        stopping_ = true;
        queue_.clear();
        queueCondition_.wakeOne();
    }

    if (QThread::currentThread() != this && isRunning() && !wait(10000)) {
        qWarning() << "writer stop timeout";
    }
}

void RpcClientWorker::run()
{
    RpcClient client;
    {
        QMutexLocker locker(&stateMutex_);
        initialized_ = true;
        connection_ = client._connectionStatus;
        initializedCondition_.wakeAll();
    }

    for (;;) {
        PendingWrite pending;
        {
            QMutexLocker locker(&queueMutex_);
            while (queue_.isEmpty() && !stopping_) {
                queueCondition_.wait(&queueMutex_);
            }
            if (stopping_ && queue_.isEmpty()) {
                break;
            }
            pending = queue_.dequeue();
        }

        if (!client.sendFrame(pending.request, pending.id)) {
            qWarning() << "write failed" << pending.command << pending.id;
        }
        else  if (!pending.command.contains("Heart")) {
            qDebug() << pending.command;
        }
    }
}

bool rpc_finsTcp::enqueueWrite(const rpcRequest& request)
{
    return enqueueWriteToWorker(request, request.threadId);
}

bool rpc_finsTcp::enqueueWriteToWorker(const rpcRequest& request, int workerIndex)
{
    if (workerIndex < 0 || workerIndex >= writeClients.size()
        || !writeClients.at(workerIndex)
        || !writeClients.at(workerIndex)->enqueue(request.requestByte, request.id,
                                                  request.command)) {
        qWarning() << "[rpc_finsTcp_boost_V2] asynchronous write enqueue failed"
                   << "worker=" << workerIndex
                   << "requestId=" << request.id;
        return false;
    }
    return true;
}

void rpc_finsTcp::stopWriteWorkers()
{
    for (RpcClientWorker*& worker : writeClients) {
        if (worker) {
            worker->stopWorker();
            delete worker;
            worker = nullptr;
        }
    }
    writeClients.clear();
}