modularization/src/driver/driver_cloud_grpc_client_h264/grpcclient.cpp

#include "grpcclient.h"

grpcclient * ggrpcclient;

void ListenData(const char * strdata,const unsigned int nSize,const unsigned int index,const QDateTime * dt,const char * strmemname)
{
    ggrpcclient->UpdateData(strdata,nSize,strmemname);

}

//#define TESTUP
void ListenPicData(const char * strdata,const unsigned int nSize,const unsigned int index,const QDateTime * dt,const char * strmemname)
{
#ifndef TESTUP
    ggrpcclient->UpdatePicData(strdata,nSize,strmemname);
#endif

#ifdef TESTUP
    ggrpcclient->UpdatePicData(strdata,nSize,"h264front");
    ggrpcclient->UpdatePicData(strdata,nSize,"h264rear");
    ggrpcclient->UpdatePicData(strdata,nSize,"h264left");
    ggrpcclient->UpdatePicData(strdata,nSize,"h264right");
#endif

}


grpcclient::grpcclient(std::string stryamlpath)
{
    ggrpcclient = this;
    dec_yaml(stryamlpath.data());


    mstrpicmsgname[0] = "h264front";
    mstrpicmsgname[1] = "h264rear";
    mstrpicmsgname[2] = "h264left";
    mstrpicmsgname[3] = "h264right";

    unsigned int i;
    if(mbFrameUpdate)
    {
        for(i=0;i<mvectormsgunit.size();i++)
        {
            mvectormsgunit[i].mpa = iv::modulecomm::RegisterRecv(mvectormsgunit[i].mstrmsgname,ListenData);
        }
    }

    for(i=0;i<mvectorctrlmsgunit.size();i++)
    {
        mvectorctrlmsgunit[i].mpa = iv::modulecomm::RegisterSend(mvectorctrlmsgunit[i].mstrmsgname,mvectorctrlmsgunit[i].mnBufferSize,
                                                                 mvectorctrlmsgunit[i].mnBufferCount);
    }

    for(i=0;i<NUM_CAM;i++)
    {
        mpaPic[i] = iv::modulecomm::RegisterRecv(mstrpicmsgname[i].data(),ListenPicData);
    }


    if(mbFrameUpdate)
    {
        for(i=0;i<NUM_CAM;i++)
        {
            unsigned int j;
            for(j=0;j<NUM_THREAD_PERCAM;j++)
            {
                mpThread[i*NUM_THREAD_PERCAM + j] = new std::thread(&grpcclient::threadpicupload,this,i);
            }
        }
    }

    for(i=0;i<NUM_CAM;i++)
    {
        mpicbuf[i].mnSkipBase = mnskip;
    }

}

grpcclient::~grpcclient()
{
    std::cout<<" enter ~grpcclient"<<std::endl;
    mbPicUpload = false;

    requestInterruption();
    while(this->isFinished() == false)
    {

    }

    std::cout<<"now join grpcclient thread"<<std::endl;

    unsigned int i;
    for(i=0;i<NUM_CAM;i++)
    {
        unsigned int j;
        for(j=0;j<NUM_THREAD_PERCAM;j++)
        {
            mpThread[i*NUM_THREAD_PERCAM + j]->join();
        }
    }

    for(i=0;i<mvectorctrlmsgunit.size();i++)
    {
        iv::modulecomm::Unregister(mvectorctrlmsgunit[i].mpa);
    }

    for(i=0;i<mvectormsgunit.size();i++)
    {
        iv::modulecomm::Unregister(mvectormsgunit[i].mpa);
    }

    std::cout<<"complete ~grpcclient"<<std::endl;

}

void grpcclient::run()
{
    int nsize = mvectormsgunit.size();
    int i;

    int ninterval = atoi(gstruploadinterval.data());
    if(ninterval<=0)ninterval = 100;

    QTime xTime;
    xTime.start();
    int nlastsend = xTime.elapsed();

    std::string target_str = gstrserverip+":";
    target_str = target_str + gstrserverport ;//std::to_string()
    std::cout<<" server : "<<target_str<<std::endl;
    auto cargs = grpc::ChannelArguments();
    cargs.SetMaxReceiveMessageSize(1024 * 1024 * 1024); // 1 GB
    cargs.SetMaxSendMessageSize(1024 * 1024 * 1024);

    std::shared_ptr<Channel> channel = grpc::CreateCustomChannel(
             target_str, grpc::InsecureChannelCredentials(),cargs);

    std::unique_ptr<iv::UploadThread::Stub> stub_ = iv::UploadThread::NewStub(channel);


    iv::UploadRequestThread request;

    int nid = 0;

    // Container for the data we expect from the server.
    iv::UploadReplyThread reply;

    gpr_timespec timespec;
      timespec.tv_sec = 30;//设置阻塞时间为2秒
      timespec.tv_nsec = 0;
      timespec.clock_type = GPR_TIMESPAN;

 //   ClientContext context;

    std::vector<qint64> xvectorlatency;


    while(!QThread::isInterruptionRequested())
    {
        std::this_thread::sleep_for(std::chrono::milliseconds(1));
        if((xTime.elapsed()-nlastsend)<ninterval)
        {
            continue;
        }

            bool bImportant = false;
            int nkeeptime = 0;
            iv::cloud::cloudmsg xmsg;
            xmsg.set_xtime(QDateTime::currentMSecsSinceEpoch());
            gMutexMsg.lock();
            for(i=0;i<nsize;i++)
            {
                if(mvectormsgunit[i].mbRefresh)
                {
                    mvectormsgunit[i].mbRefresh = false;
                    if(mvectormsgunit[i].mbImportant)
                    {
                        bImportant = true;
                    }
                    if(mvectormsgunit[i].mnkeeptime > nkeeptime)
                    {
                        nkeeptime = mvectormsgunit[i].mnkeeptime;
                    }
                    iv::cloud::cloudunit xcloudunit;
                    xcloudunit.set_msgname(mvectormsgunit[i].mstrmsgname);
                    xcloudunit.set_data(mvectormsgunit[i].mpstrmsgdata.get(),mvectormsgunit[i].mndatasize);
                    iv::cloud::cloudunit * pcu = xmsg.add_xclouddata();
                    pcu->CopyFrom(xcloudunit);
                }

            }
            gMutexMsg.unlock();

            int nbytesize = xmsg.ByteSize();
            char * strbuf = new char[nbytesize];
            std::shared_ptr<char> pstrbuf;
            pstrbuf.reset(strbuf);
            if(xmsg.SerializeToArray(strbuf,nbytesize))
            {

                ClientContext context ;
                context.set_deadline(timespec);
                qint64 time1 = QDateTime::currentMSecsSinceEpoch();



                request.set_id(nid);
                request.set_ntime(time1);
                request.set_strquerymd5(gstrqueryMD5);
                request.set_strctrlmd5(gstrctrlMD5);
                request.set_strvin(gstrVIN);
                request.set_xdata(strbuf,nbytesize);
                request.set_kepptime(nkeeptime);
                request.set_bimportant(bImportant);
                request.set_nsendtime(QDateTime::currentMSecsSinceEpoch());
                request.set_nlatency(CalcLateny(xvectorlatency));
                nid++;

                nlastsend = xTime.elapsed();
                // The actual RPC.
                Status status = stub_->uploaddata(&context, request, &reply);
                if (status.ok()) {
//                    std::cout<<"  data size is "<<nbytesize<<std::endl;
//                    std::cout<<nid<<" upload successfully"<<std::endl;
                    xvectorlatency.push_back((QDateTime::currentMSecsSinceEpoch() - reply.nreqsendtime()));
                    while(xvectorlatency.size()>10)xvectorlatency.erase(xvectorlatency.begin());
                    if(reply.nres() == 1)
                    {
                        iv::cloud::cloudmsg xmsg;
                        if(xmsg.ParseFromArray(reply.xdata().data(),reply.xdata().size()))
                        {
                            sharectrlmsg(&xmsg);
                        }
                    }
                } else {
                  std::cout << status.error_code() << ": " << status.error_message()
                            << std::endl;
                  std::cout<<"RPC failed"<<std::endl;
                  if(status.error_code() == 4)
                  {
                      std::cout<<" RPC Exceed Time, Create New stub_"<<std::endl;
                      channel = grpc::CreateCustomChannel(
                               target_str, grpc::InsecureChannelCredentials(),cargs);

                      stub_ = iv::UploadThread::NewStub(channel);
                  }
                  std::this_thread::sleep_for(std::chrono::milliseconds(300));

                }

            }


    }

    std::cout<<" grpcclient:run complete."<<std::endl;
}


void grpcclient::dec_yaml(const char * stryamlpath)
{

    YAML::Node config;
    try
    {
        config = YAML::LoadFile(stryamlpath);
    }
    catch(YAML::BadFile e)
    {
        qDebug("load error.");
        return;
    }

    std::vector<std::string> vecmodulename;


    if(config["server"])
    {
        gstrserverip = config["server"].as<std::string>();
    }

    std::cout<<" server ip: "<<gstrserverip<<std::endl;

    if(config["port"])
    {
        gstrserverport = config["port"].as<std::string>();
    }
    if(config["uploadinterval"])
    {
        gstruploadinterval = config["uploadinterval"].as<std::string>();
    }
    if(config["skip"])
    {
        std::string strskip = config["skip"].as<std::string>();
        mnskip = atoi(strskip.data());
        if(mnskip<1)mnskip = 1;
    }

    if(config["VIN"])
    {
        gstrVIN = config["VIN"].as<std::string>();
    }

    if(config["queryMD5"])
    {
        gstrqueryMD5 = config["queryMD5"].as<std::string>();
    }
    else
    {
        return;
    }

    if(config["ctrlMD5"])
    {
        gstrctrlMD5 = config["ctrlMD5"].as<std::string>();
    }


    std::string strmsgname;

    if(config["uploadmessage"])
    {

        for(YAML::const_iterator it= config["uploadmessage"].begin(); it != config["uploadmessage"].end();++it)
        {
            std::string strtitle = it->first.as<std::string>();
            std::cout<<strtitle<<std::endl;

            if(config["uploadmessage"][strtitle]["msgname"]&&config["uploadmessage"][strtitle]["buffersize"]&&config["uploadmessage"][strtitle]["buffercount"])
            {
                iv::msgunit xmu;
                strmsgname = config["uploadmessage"][strtitle]["msgname"].as<std::string>();
                strncpy(xmu.mstrmsgname,strmsgname.data(),255);
                xmu.mnBufferSize = config["uploadmessage"][strtitle]["buffersize"].as<int>();
                xmu.mnBufferCount = config["uploadmessage"][strtitle]["buffercount"].as<int>();
                if(config["uploadmessage"][strtitle]["bimportant"])
                {
                   std::string strimportant =    config["uploadmessage"][strtitle]["bimportant"].as<std::string>();
                   if(strimportant == "true")
                   {
                       xmu.mbImportant = true;
                   }
                }
                if(config["uploadmessage"][strtitle]["keeptime"])
                {
                   std::string strkeep =    config["uploadmessage"][strtitle]["keeptime"].as<std::string>();
                   xmu.mnkeeptime = atoi(strkeep.data());
                }
                mvectormsgunit.push_back(xmu);
            }
        }
    }
    else
    {


    }

    if(!config["ctrlMD5"])
    {
       return;
    }

    if(config["ctrlmessage"])
    {
        std::string strnodename = "ctrlmessage";
        for(YAML::const_iterator it= config[strnodename].begin(); it != config[strnodename].end();++it)
        {
            std::string strtitle = it->first.as<std::string>();
            std::cout<<strtitle<<std::endl;

            if(config[strnodename][strtitle]["msgname"]&&config[strnodename][strtitle]["buffersize"]&&config[strnodename][strtitle]["buffercount"])
            {
                iv::msgunit xmu;
                strmsgname = config[strnodename][strtitle]["msgname"].as<std::string>();
                strncpy(xmu.mstrmsgname,strmsgname.data(),255);
                xmu.mnBufferSize = config[strnodename][strtitle]["buffersize"].as<int>();
                xmu.mnBufferCount = config[strnodename][strtitle]["buffercount"].as<int>();
                mvectorctrlmsgunit.push_back(xmu);
            }
        }
    }
    else
    {

    }

    return;

}

void grpcclient::sharectrlmsg(iv::cloud::cloudmsg * pxmsg)
{
    int i;
    int nsize = pxmsg->xclouddata_size();
    for(i=0;i<nsize;i++)
    {
        int j;
        int nquerysize = mvectorctrlmsgunit.size();
        for(j=0;j<nquerysize;j++)
        {
            if(strncmp(pxmsg->xclouddata(i).msgname().data(), mvectorctrlmsgunit[j].mstrmsgname,255) == 0)
            {
 //               qDebug("size is %d ",pxmsg->xclouddata(i).data().size());
                iv::modulecomm::ModuleSendMsg(mvectorctrlmsgunit[j].mpa,pxmsg->xclouddata(i).data().data(),pxmsg->xclouddata(i).data().size());
                break;
            }
        }
    }
}



void grpcclient::UpdateData(const char *strdata, const unsigned int nSize, const char *strmemname)
{
    int nsize = mvectormsgunit.size();
    int i;
    for(i=0;i<nsize;i++)
    {

        if(strncmp(strmemname,mvectormsgunit[i].mstrmsgname,255) == 0)
        {
            gMutexMsg.lock();
            char * strtem = new char[nSize];
            memcpy(strtem,strdata,nSize);
            mvectormsgunit[i].mpstrmsgdata.reset(strtem);
            mvectormsgunit[i].mndatasize = nSize;
            mvectormsgunit[i].mbRefresh = true;
            gMutexMsg.unlock();
            break;
        }
    }
}

void grpcclient::UpdatePicData(const char *strdata, const unsigned int nSize, const char *strmemname)
{
    int npos = -1;

    unsigned int i;
    for(i=0;i<NUM_CAM;i++)
    {
        if(strncmp(strmemname,mstrpicmsgname[i].data(),255) == 0)
        {
            npos = i;
            break;
        }
    }

    if(npos<0)
    {
        std::cout<<"grpcclient::UpdatePicData not found pic. msg name is "<<strmemname<<std::endl;
        return;
    }

    if(npos>= NUM_CAM)
    {
        std::cout<<"Camera count is "<<NUM_CAM<<" NOW camear is "<<npos<<std::endl;
        return;
    }

    if(nSize<5)return;
    iv::h264frame xframe;
    if((strdata[4] == 0x27)||(strdata[4] == 0x47)||(strdata[4] == 0x67))
    {
        xframe.mbIframe = true;
    }
    else
    {
        xframe.mbIframe = false;
    }

    if(mpicbuf[npos].mbRecvIFrame == false)
    {
      if(xframe.mbIframe == false)
      {
          return;
      }
      else
      {
          mpicbuf[npos].mbRecvIFrame = true;
      }
    }

    xframe.mpstrframedata = std::shared_ptr<char>(new char[nSize]);
    xframe.mDataSize = nSize;
    memcpy(xframe.mpstrframedata.get(),strdata,nSize);

    iv::threadpicunit * ppicbuf = &mpicbuf[npos];
    ppicbuf->mMutex.lock();
    ppicbuf->mnMsgTime = QDateTime::currentMSecsSinceEpoch();
    ppicbuf->mbRefresh = true;

    if(ppicbuf->mvectorframe.size()>=NUM_FRAMEBUFFSIZE)
    {
        while((ppicbuf->mvectorframe.size()>=NUM_FRAMEBUFFSIZE)||((ppicbuf->mvectorframe.size()>0)&&(ppicbuf->mvectorframe.size()<NUM_FRAMEBUFFSIZE)&&(ppicbuf->mvectorframe[0].mbIframe == false)))
        {
            ppicbuf->mvectorframe.erase(ppicbuf->mvectorframe.begin()+0);
        }
        if(ppicbuf->mvectorframe.size() == 0)
        {
            std::cout<<" Reset SPS Iframe Mark."<<std::endl;
            ppicbuf->mbRecvIFrame = false;
        }
    }
    ppicbuf->mvectorframe.push_back(xframe);

//    mpicbuf[npos].mpstrmsgdata = std::shared_ptr<char>(new char[nSize]);
//    mpicbuf[npos].mDataSize = nSize;
//    memcpy(mpicbuf[npos].mpstrmsgdata.get(),strdata,nSize);
    ppicbuf->mMutex.unlock();
    ppicbuf->mwc.wakeAll();
}


void grpcclient::threadpicupload(int nCamPos)
{
    std::cout<<"thread cam "<<nCamPos<<"run"<<std::endl;
    int nsize = mvectormsgunit.size();
    int i;

    int ninterval = atoi(gstruploadinterval.data());
    if(ninterval<=0)ninterval = 100;

    QTime xTime;
    xTime.start();
    int nlastsend = xTime.elapsed();

    std::string target_str = gstrserverip+":";
    target_str = target_str + gstrserverport ;//std::to_string()
    auto cargs = grpc::ChannelArguments();
    cargs.SetMaxReceiveMessageSize(1024 * 1024 * 1024); // 1 GB
    cargs.SetMaxSendMessageSize(1024 * 1024 * 1024);

    std::shared_ptr<Channel> channel = grpc::CreateCustomChannel(
             target_str, grpc::InsecureChannelCredentials(),cargs);

    std::unique_ptr<iv::UploadThread::Stub> stub_ = iv::UploadThread::NewStub(channel);


    iv::PicUpRequestThread request;

    int nid = 0;

    // Container for the data we expect from the server.
    iv::PicUpReplyThread reply;

    gpr_timespec timespec;
      timespec.tv_sec = 30;//设置阻塞时间为2秒
      timespec.tv_nsec = 0;
      timespec.clock_type = GPR_TIMESPAN;

 //   ClientContext context;

    while(mbPicUpload)
    {
        std::shared_ptr<char> pstr_ptr;
        if((nCamPos<0)||(nCamPos >= NUM_CAM))
        {
            std::cout<<"Cam Pos Error. "<<"Pos: "<<nCamPos<<" TOTAL:"<<NUM_CAM<<std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
            continue;
        }

        bool bUpdate = false;
        qint64 nMsgTime = 0;
        int nSize = 0;
        qint64 npiclatency;
        int nSkipBase = 1;
        int nCount;

        if(mpicbuf[nCamPos].mvectorframe.size() == 0)
        {
            mpicbuf[nCamPos].mWaitMutex.lock();
            mpicbuf[nCamPos].mwc.wait(&mpicbuf[nCamPos].mWaitMutex,100);
            mpicbuf[nCamPos].mWaitMutex.unlock();

        }
        mpicbuf[nCamPos].mMutex.lock();
//        bUpdate = mpicbuf[nCamPos].mbRefresh;
        if(mpicbuf[nCamPos].mvectorframe.size() > 0)bUpdate = true;
        if(bUpdate == true)
        {
            nMsgTime = mpicbuf[nCamPos].mnMsgTime;

            mpicbuf[nCamPos].mbRefresh = false;
            const int npacmax = 10;
            int npaccount = mpicbuf[nCamPos].mvectorframe.size();
            if(npaccount > npacmax)npaccount = npacmax;
            int nalldatasize = 0;
            int j;
            for(j=0;j<npaccount;j++)
            {
                nalldatasize = nalldatasize + mpicbuf[nCamPos].mvectorframe[j].mDataSize;
            }
            int nsendpacsize = (npaccount)*sizeof(int) + nalldatasize;
            pstr_ptr = std::shared_ptr<char>(new char[nsendpacsize]);
            char * pstrvalue = (char * )pstr_ptr.get();
            int npos = 0;
            for(j=0;j<npaccount;j++)
            {
                int * psize = (int * )(pstrvalue + npos);
                *psize = mpicbuf[nCamPos].mvectorframe[j].mDataSize;
                npos = npos + sizeof(int);
                memcpy(pstrvalue+ npos,mpicbuf[nCamPos].mvectorframe[j].mpstrframedata.get(),mpicbuf[nCamPos].mvectorframe[j].mDataSize);
                npos = npos + mpicbuf[nCamPos].mvectorframe[j].mDataSize;
            }
            std::cout<<"pac count "<<npaccount<<std::endl;

            for(j=0;j<npaccount;j++)mpicbuf[nCamPos].mvectorframe.erase(mpicbuf[nCamPos].mvectorframe.begin());
            nSize = nsendpacsize;
//            pstr_ptr = mpicbuf[nCamPos].mvectorframe[0].mpstrframedata;
//            nSize = mpicbuf[nCamPos].mvectorframe[0].mDataSize;
//            mpicbuf[nCamPos].mvectorframe.erase(mpicbuf[nCamPos].mvectorframe.begin());
//            pstr_ptr = mpicbuf[nCamPos].mpstrmsgdata;
//            nSize = mpicbuf[nCamPos].mDataSize;
            npiclatency = CalcLateny(mpicbuf[nCamPos].mvectorlatency);
            nSkipBase = mpicbuf[nCamPos].mnSkipBase;
            nCount = mpicbuf[nCamPos].mnCount;
            mpicbuf[nCamPos].mnCount++;
//            if(npiclatency >  500)
//            {
//                if(mpicbuf[nCamPos].mnSkipBase<30)mpicbuf[nCamPos].mnSkipBase++;
//            }
//            else
//            {
//                if(npiclatency<300)
//                    if(mpicbuf[nCamPos].mnSkipBase > mpicbuf[nCamPos].mnDefSkipBase)mpicbuf[nCamPos].mnSkipBase--;
//            }
            std::cout<<"upload "<<nMsgTime<<"  latency:  "<<npiclatency<<" skip param: "<<nSkipBase<<std::endl;
        }
        mpicbuf[nCamPos].mMutex.unlock();
//        if(bUpdate == false)
//        {
// //           std::this_thread::sleep_for(std::chrono::milliseconds(10));
//            continue;
//        }

//        if(nCount%nSkipBase != 0)
//        {
//            continue;
//        }



        ClientContext context ;
        context.set_deadline(timespec);
        qint64 time1 = QDateTime::currentMSecsSinceEpoch();

        request.set_npictime(nMsgTime);
        request.set_ncampos(nCamPos);
        request.set_strvin(gstrVIN);
        request.set_xdata(pstr_ptr.get(),nSize);
        request.set_nlatency(npiclatency);
        nid++;

        nlastsend = xTime.elapsed();
        // The actual RPC.
        Status status = stub_->uploadpic(&context, request, &reply);
        if (status.ok()) {

            qint64 nlaten = QDateTime::currentMSecsSinceEpoch() - time1;
            mpicbuf[nCamPos].mMutex.lock();
            mpicbuf[nCamPos].mvectorlatency.push_back(nlaten);
            while(mpicbuf[nCamPos].mvectorlatency.size()>10)mpicbuf[nCamPos].mvectorlatency.erase(mpicbuf[nCamPos].mvectorlatency.begin());
            mpicbuf[nCamPos].mMutex.unlock();
            if(reply.nres() == 1)
            {
//                iv::cloud::cloudmsg xmsg;
//                if(xmsg.ParseFromArray(reply.xdata().data(),reply.xdata().size()))
//                {
//                    sharectrlmsg(&xmsg);
//                }
            }
        } else {
          std::cout << status.error_code() << ": " << status.error_message()
                    << std::endl;
          std::cout<<"RPC failed"<<std::endl;
          if(status.error_code() == 4)
          {
              std::cout<<nCamPos<<" RPC Exceed Time, Create New stub_"<<std::endl;
              channel = grpc::CreateCustomChannel(
                       target_str, grpc::InsecureChannelCredentials(),cargs);

              stub_ = iv::UploadThread::NewStub(channel);
          }
          std::this_thread::sleep_for(std::chrono::milliseconds(900));

        }


    }

    std::cout<<"threadpicupload cam pos: "<<nCamPos<<" exit."<<std::endl;
}

qint64 grpcclient::CalcLateny(std::vector<qint64> &xvectorlatency)
{
    if(xvectorlatency.size() == 0)return 1000;
    unsigned int i;
    qint64 nLatencyTotal = 0;
    for(i=0;i<xvectorlatency.size();i++)
    {
        nLatencyTotal = nLatencyTotal + xvectorlatency[i];
    }
    qint64 nLatencyAvg = nLatencyTotal/xvectorlatency.size();
    return nLatencyAvg;
}