fbus支持类似rpc的远程进程调用方法,是利用invoke方法实现的,invoke方法提供了多个重载版本,参数不同,但最终还是调用submit来将消息或者说CBasejob对象提交到事件循环的队列中。

submit是CFdbMessage提供的一个方法,该方法具体实现了同步调用和异步调用。

下面把CFdbBaseObject类下invoke的相关声明列出来:

    bool invoke(FdbSessionId_t receiver
                , FdbMsgCode_t code
                , IFdbMsgBuilder &data
                , int32_t timeout = 0
                , EFdbQOS qos = FDB_QOS_DEFAULT);
    bool invoke(FdbSessionId_t receiver
                , CFdbMessage *msg
                , IFdbMsgBuilder &data
                , int32_t timeout = 0);
    bool invoke(FdbMsgCode_t code
                , IFdbMsgBuilder &data
                , int32_t timeout = 0
                , EFdbQOS qos = FDB_QOS_DEFAULT);
    bool invoke(CFdbMessage *msg
                , IFdbMsgBuilder &data
                , int32_t timeout = 0);
    bool invoke(FdbSessionId_t receiver
                , CBaseJob::Ptr &msg_ref
                , IFdbMsgBuilder &data
                , int32_t timeout = 0);
    bool invoke(CBaseJob::Ptr &msg_ref
                , IFdbMsgBuilder &data
                , int32_t timeout = 0);
    bool invoke(FdbSessionId_t receiver
                , FdbMsgCode_t code
                , const void *buffer = 0
                , int32_t size = 0
                , int32_t timeout = 0
                , EFdbQOS qos = FDB_QOS_DEFAULT
                , const char *log_info = 0);
    bool invoke(FdbSessionId_t receiver
                , CFdbMessage *msg
                , const void *buffer = 0
                , int32_t size = 0
                , int32_t timeout = 0);
    bool invoke(FdbMsgCode_t code
                , const void *buffer = 0
                , int32_t size = 0
                , int32_t timeout = 0
                , EFdbQOS qos = FDB_QOS_DEFAULT
                , const char *log_data = 0);
    bool invoke(CFdbMessage *msg
                , const void *buffer = 0
                , int32_t size = 0
                , int32_t timeout = 0);
    bool invoke(FdbSessionId_t receiver
                , CBaseJob::Ptr &msg_ref
                , const void *buffer = 0
                , int32_t size = 0
                , int32_t timeout = 0);
    bool invoke(CBaseJob::Ptr &msg_ref
                , const void *buffer = 0
                , int32_t size = 0
                , int32_t timeout = 0);
    bool invoke(FdbMsgCode_t code
                , IFdbMsgBuilder &data
                , tInvokeCallbackFn callback
                , CBaseWorker *worker = 0
                , int32_t timeout = 0
                , EFdbQOS qos = FDB_QOS_DEFAULT);
    bool invoke(FdbMsgCode_t code
                , tInvokeCallbackFn callback
                , const void *buffer = 0
                , int32_t size = 0
                , CBaseWorker *worker = 0
                , int32_t timeout = 0
                , EFdbQOS qos = FDB_QOS_DEFAULT
                , const char *log_info = 0);

上面有很多invoke方法,但是从功能上分为两类:一类是异步,一类是同步。若要知道那个是同步还是异步,查看头文件即可。上面列出了多种invoke接口,这些个方法都是由端点调用的,是用户调用的入口。

上面CFdbBaseObject提供了很多invoke方法,然后CFdbMessage也提供了三种invoke方法,他们之间的关系,CFdbBaseObject提供的这些invoke方法,无论如何变形最终调用CFdbMessage提供的3种invoke方法。CFdbMessage提供的3中invoke方法声明如下:

    bool invoke(CBaseJob::Ptr &msg_ref
                , uint32_t tx_flag
                , int32_t timeout);
bool invoke(int32_t timeout = 0);
static bool invoke(CBaseJob::Ptr &msg_ref, int32_t timeout = 0);

前文对invoke之间的关系进行了详细的总结,通过上面我们可以得出结论调用invoke的两种方法:

  1. 网络端点无论是客户端还是服务端可以调用invoke方法消息
  2. 通过定义一个CFdbMessage实例,也可以直接调用相关的方法,如下
CFdbMessage msg;
msg->invoke(...);

 好了,下面开始进入正题,我看fdbus源码时产生了这样一个疑问,invoke同步是如何实现的?如果让我去实现同步,应该如何做?带着这个疑问深入了去看了fdbus源码,终于让我搞明白了在fdbus框架中同步是如何实现的,下面我尽我所能将它呈现出来。

首先搞明白同步调用和异步调用的含义:

  • 同步调用:当调用一个函数的时候,调用函数需要等待被调用函数执行完成,并返回结果后,才会执行调用函数下面的语句。在被调用函数的执行逻辑没有完成之前,是不会返回的,从而造成调用函数会阻塞等待,举一个极端的例子,如果一个函数执行完需要2s,那么调用函数则会这个期间等待2s才会执行其后面的语句。举个最简单的例子,我们在同一个线程中发生的函数调用都是同步调用。这里还是要补充一点,一般情况下同步调用会包含一个阻塞时间,这个时间是防止一直阻塞,当然也可以选择一直阻塞,是否设置这个阻塞时间由用户根据自己的业务需求决定。
  • 异步调用:当调用一个函数的时候,调用函数在调用一个函数请求某个结果时,假设该被调用结果执行需要2s,在异步调用的情况下,该被调用函数会立即返回,让调用函数继续执行其后面的语句。然后在2s后执行完成后,再将结果通过某种方式返回给调用方。当然异步调用一般是与队列,事件循环、多线程相关。即多线程或者多进程情况下调用线程将请求压入到事件队列,然后事件循环依次依次遍历该事件队列,按照事件的类型推送到被调用线程或者进程进行处理,当被调用的线程或者进程收到事件数据后,将结果通过某种方式返回给调用线程或者进程,这样一个过程统称为异步调用。

本文仅对fdbus的同步调用进行剖析。

下面以msg->invoke()为例进行介绍,

bool CFdbMessage::invoke(CBaseJob::Ptr &msg_ref , int32_t timeout)
{
    auto msg = castToMessage<CFdbMessage *>(msg_ref);
    return msg ? msg->invoke(msg_ref, FDB_MSG_TX_SYNC, timeout) : false;
}

通过上面代码可以看到CFdbMessage下invoke的同步调用接口包含两个参数,一个是指向消息对象的共享指针,一个是超时时间。可以看到在上面代码中调用了CFdbMessage中的另外一个invoke方法,其中第二个参数不知大家有没有注意到,没错就是FDB_MSG_TX_SYNC这个参数,通过字面意思大家都明白这是设置了同步标识,源码如下所示:

bool CFdbMessage::invoke(CBaseJob::Ptr &msg_ref
                         , uint32_t tx_flag
                         , int32_t timeout)
{
    mType = FDB_MT_REQUEST;
    return submit(msg_ref, tx_flag, timeout);
}

上面的代码中tx_flag是一个消息标志。之前所有提到invoke方法最终都是需要调用这个invoke方法进行逻辑处理。

bool CFdbMessage::submit(CBaseJob::Ptr &msg_ref
                         , uint32_t tx_flag
                         , int32_t timeout)
{
    ....

    bool sync = !!(tx_flag & FDB_MSG_TX_SYNC);
    if (sync && mContext->isSelf())
    {
        setStatusMsg(FDB_ST_UNABLE_TO_SEND, "Cannot send sychronously from context");
        return false;
    }

    if (tx_flag & FDB_MSG_TX_NO_REPLY)
    {
        mFlag |= MSG_FLAG_NOREPLY_EXPECTED;
    }
    else
    {
        mFlag &= ~MSG_FLAG_NOREPLY_EXPECTED;
        if (sync)
        {
            mFlag |= MSG_FLAG_SYNC_REPLY;
        }
        if (timeout > 0)
        {
            mTimer = new CMessageTimer(timeout);
        }
    }

    bool ret = true;
    if (tx_flag & FDB_MSG_TX_NO_QUEUE)
    {
        dispatchMsg(msg_ref);
    }
    else
    {
        setCallable(std::bind(&CFdbMessage::dispatchMsg, this, _1));
        if (sync)
        {
            ret = mContext->sendSync(msg_ref);
        }
        else
        {
            ret = mContext->sendAsync(msg_ref);
        }
    }

    ...
}
bool CBaseWorker::sendSync(CBaseJob::Ptr &job, int32_t milliseconds, bool urgent)
{
    return send(job, milliseconds, urgent);
}
bool CBaseWorker::send(CBaseJob::Ptr &job, int32_t milliseconds, bool urgent)
{
    if (job->mSyncReq)
    {
        return false;
    }

    // now we can assure the job is not in any worker queue
    CBaseJob::CSyncRequest sync_req(job.use_count());
    job->mSyncReq = &sync_req;
    if (!send(job, urgent))
    {
        return false;
    }

    if (job->mSyncReq)
    {
        job->mSyncLock.lock();
        if (job->mSyncReq)
        {
            if (milliseconds <= 0)
            {    // job->mSyncLock will be released
                sync_req.mWakeupSignal.wait(job->mSyncLock);
            }
            else
            {    // job->mSyncLock will be released
                std::cv_status status = sync_req.mWakeupSignal.wait_for(job->mSyncLock,
                                            std::chrono::milliseconds(milliseconds));
                if (status == std::cv_status::timeout)
                { // timeout! nothing to do.
                }
            }
            job->mSyncReq = 0;
        }
        job->mSyncLock.unlock();
    }


    return true;
}

 条件变量mWakeupSignal通过wait方法进行阻塞,

bool CBaseWorker::send(CBaseJob::Ptr &job, bool urgent, bool swap)
{
    bool ret;
    
    if (mExitCode || !mEventLoop)
    {
        return false;
    }

    if (urgent)
    {
        ret = mUrgentJobQueue.enqueue(job, swap);
    }
    else
    {
        ret = mNormalJobQueue.enqueue(job, swap);
    }

    return ret;
}

 虽然是同步调用,但是在条件变量mWakeupSignal阻塞之前调用了send方法,通过send方法源码可以看到虽然是同步调用,但是还是将消息数据或者说消息对象压入到事件队列,然后通过事件循环依次进行处理,不同的是通过条件变量mWakeupSignal调用wait方法对调用线程进行了阻塞,然后等待接收回复消息,并唤醒调用线程。

上面的代码是发送环节的逻辑,下面将解除调用线程阻塞的关键源码列出:

void CBaseSession::doResponse(CFdbMessageHeader &head)
{
    bool found;
    PendingMsgTable_t::EntryContainer_t::iterator it;
    CBaseJob::Ptr &msg_ref = mPendingMsgTable.retrieveEntry(head.serial_number(), it, found);
    if (found)
    {
        auto msg = castToMessage<CFdbMessage *>(msg_ref);
        auto object_id = head.object_id();
        if (msg->objectId() != object_id)
        {
            LOG_E("CFdbSession: object id of response %d does not match that in request: %d\n",
                    object_id, msg->objectId());
            terminateMessage(msg_ref, FDB_ST_OBJECT_NOT_FOUND, "Object ID does not match.");
            mPendingMsgTable.deleteEntry(it);
            delete[] mPayloadBuffer;
            mPayloadBuffer = 0;
            return;
        }

        auto object = mContainer->owner()->getObject(msg, false);
        if (object)
        {
            msg->update(head, mMsgPrefix);
            msg->decodeDebugInfo(head);
            msg->replaceBuffer(mPayloadBuffer, head.payload_size(), mMsgPrefix.mHeadLength);
            mPayloadBuffer = 0;
            auto type = msg->type();
            doStatistics(type, head.flag(), mStatistics.mRx);
            if (!msg->sync())
            {
                switch (type)
                {
                    case FDB_MT_REQUEST:
                        object->doReply(msg_ref);
                    break;
                    case FDB_MT_SIDEBAND_REQUEST:
                        object->onSidebandReply(msg_ref);
                    break;
                    case FDB_MT_GET_EVENT:
                        object->doReturnEvent(msg_ref);
                    break;
                    case FDB_MT_SET_EVENT:
                    default:
                        if (head.type() == FDB_MT_STATUS)
                        {
                            object->doStatus(msg_ref);
                        }
                        else
                        {
                            LOG_E("CFdbSession: request type %d doesn't match response type %d!\n",
                                  type, head.type());
                        }
                    break;
                }
            }
        }

        msg_ref->terminate(msg_ref);
        mPendingMsgTable.deleteEntry(it);
    }
}

收到响应的入口处理函数。

void CBaseJob::terminate(Ptr &ref)
{
    if (!ref)
    {
        return;
    }

    if (mSyncReq)
    {
        mSyncLock.lock();
        if (mSyncReq)
        {
            // Why +1? because the job must be referred locally.
            // Warning: ref count of the job can not be changed
            // during the sync waiting!!!
            if (ref.use_count() == (mSyncReq->mInitSharedCnt + 1))
            {
                mSyncReq->mWakeupSignal.notify_one();
                mSyncReq = 0;
            }
        }
        mSyncLock.unlock();
    }
}

 看到了吧,这里条件变量通过notify_one被唤醒。

上面的代码应该能够很清晰的看出fdbus同步调用的实现。其实就3步:

  1. 调用线程将消息压入消息队列
  2. 压入消息队列后,通过条件变量阻塞调用线程,等待唤醒
  3. 收到回复消息后,发送notify_one唤醒阻塞线程

这里再啰嗦一点个人认为比较重要的,我们都知道函数有输入参数,输出参数,也有输入输出参数,即在一个函数中的参数既可以是输入参数又可以输出参数,那么在fdbus中发生同步调用的参数CBaseJob::Ptr &msg_ref是如何做到编成收到的消息内容的呢?

请看下面的文章

总结

通过分析fdbus同步实现,了解了一种同步实现方式,对同步调用了有了更深的理解,当然fdbus是基于网络的,所以即使是同步调用,本质上也是需要压入消息队列或者事件队列,按照的异步的方式处理。

09-28 11:35