master

分支 (1)

管理

管理

master

Tars
/
cpp
/
framework
/
StatServer
/
ReapSSDThread.cpp

/**
 * Tencent is pleased to support the open source community by making Tars available.
 *
 * Copyright (C) 2016THL A29 Limited, a Tencent company. All rights reserved.
 *
 * Licensed under the BSD 3-Clause License (the "License"); you may not use this file except
 * in compliance with the License. You may obtain a copy of the License at
 *
 * https://opensource.org/licenses/BSD-3-Clause
 *
 * 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 "ReapSSDThread.h"
#include "util/tc_config.h"
#include "StatServer.h"


ReapSSDProcThread::ReapSSDProcThread(ReapSSDThread * proc)
: _terminate(false)
, _proc(proc)
{
}

ReapSSDProcThread::~ReapSSDProcThread()
{
    if(isAlive())
    {
        terminate();
        getThreadControl().join();
    }
    _queue.clear();
}

void ReapSSDProcThread::terminate()
{
    _terminate = true;

    _queue.notifyT();
}

void ReapSSDProcThread::put(QueueItem data)
{
    if(!_terminate)
    {
        _queue.push_back(data);
    }
}

bool ReapSSDProcThread::pop(QueueItem & data)
{
    return _queue.pop_front(data, 1000);
}

int ReapSSDProcThread::sendAlarmSMS(const string &sMsg)
{
    string errInfo = " ERROR:" + ServerConfig::LocalIp + "_" + sMsg;
    TARS_NOTIFY_ERROR(errInfo);

    return 0;
}

void ReapSSDProcThread::run()
{
    string sDate1("");
    string sFlag1("");
    string sDate2("");
    string sFlag2("");

    while (!_terminate)
    {
        try
        {
            sDate1 = "";
            sFlag1 = "";
            sDate2 = "";
            sFlag2 = "";

            QueueItem item;

            if(pop(item))
            {
                if(item._statmsg != NULL)
                {
                    int64_t iBegin = TNOWMS;
                    int64_t iEnd = 0;

                    StatDbManager::getInstance()->insert2MultiDbs(item._index, *item._statmsg, item._date, item._tflag);

                    iEnd = TNOWMS;

                    TLOGDEBUG("ReapSSDProcThread::run stat ip:" << ServerConfig::LocalIp  << "|dbIndex:" << item._index << "|" << StatDbManager::getInstance()->getIpAndPort(item._index)
                        << "|date:" << item._date << "|tflag:" << item._tflag << "|timecost(ms):" << (iEnd - iBegin) << "|iBegin:" << iBegin << "|iEnd:" << iEnd << endl);

                    FDLOG("CountStat") << "ReapSSDProcThread::run stat ip:" << ServerConfig::LocalIp  << "|dbIndex:" << item._index << "|" << StatDbManager::getInstance()->getIpAndPort(item._index)
                        << "|date:" << item._date << "|tflag:" << item._tflag << "|timecost(ms):" << (iEnd - iBegin) << "|iBegin:" << iBegin << "|iEnd:" << iEnd << endl;

                    if((iEnd - iBegin)/1000 > (g_app.getInserInterv() - 2) * 60)
                    {
                        FDLOG("CountStat") << "stat ip:" << ServerConfig::LocalIp << "|ReapSSDProcThread::run timeout 8 minute." << endl;
                        string sMsg("stat ip:");
                        sMsg += ServerConfig::LocalIp;
                        sMsg += " ReapSSDProcThread::run write db:";
                        sMsg += StatDbManager::getInstance()->getIpAndPort(item._index);
                        sMsg += "|timeout ";
                        sMsg += TC_Common::tostr(g_app.getInserInterv() - 2);
                        sMsg += " Minute.";

                        sendAlarmSMS(sMsg);
                    }

                    delete item._statmsg;
                    item._statmsg = NULL;
                }
                else
                {
                    TLOGERROR("ReapSSDProcThread::run item._statmsg == NULL." << endl);
                }
            }
        }
        catch(exception& e)
        {
            TLOGERROR("ReapSSDProcThread::run exception:" << e.what() << endl);
            FDLOG("CountStat") << "ReapSSDProcThread::run exception:" << e.what() << endl;
        }
    }
}
//////////////////////////////////////////////////////////////
ReapSSDThread::ReapSSDThread()
:  _terminate(false)
, _curWeight(0)
, _lastSq(-1)
{
    TLOGDEBUG("ReapSSDThread begin ok" << endl);
}

ReapSSDThread::~ReapSSDThread()
{
    if (isAlive())
    {
        terminate();

        getThreadControl().join();
    }
}

void ReapSSDThread::terminate()
{
    TLOGDEBUG("ReapSSDThread terminate." << endl);

    _terminate = true;

    TC_ThreadLock::Lock lock(*this);

    notifyAll();
}

void ReapSSDThread::run()
{
    int iInsertDataNum = StatDbManager::getInstance()->getDbIpNum();

    for(int i = 0; i < iInsertDataNum; ++i)
    {
        ReapSSDProcThread *r = new ReapSSDProcThread(this);

        r->start();

        _runners.push_back(r);
    }

    string sDate,sTime;

    int dbNumber = StatDbManager::getInstance()->getDbNumber();

    string sRandOrder;

    uint64_t iTotalNum = 0;

    int iLastIndex = -1;

    FDLOG("CountStat") << "stat ip:" << ServerConfig::LocalIp << "|ReapSSDThread::run iInsertDataThread:" << iInsertDataNum << "|dbNumber:" << dbNumber << endl;

    while (!_terminate)
    {
        try
        {
            //双buffer中一个buffer入库
            int iBufferIndex = !(g_app.getSelectBufferIndex());
            int64_t iInterval = 3;
            if(iBufferIndex != iLastIndex && g_app.getSelectBuffer(iBufferIndex, iInterval))
            {
                iLastIndex = iBufferIndex;

                iTotalNum = 0;

                vector<StatMsg*> vAllStatMsg;
                for(int iStatIndex = 0; iStatIndex < dbNumber; ++iStatIndex)
                {
                    vAllStatMsg.push_back(new StatMsg());
                }

                int64_t tBegin = TNOWMS;

                getDataFromBuffer(iBufferIndex, vAllStatMsg, iTotalNum);

                int64_t tEnd = TNOWMS;

                TLOGDEBUG("stat ip:" << ServerConfig::LocalIp << "|Buffer Index:" << iBufferIndex << "|ReapSSDThread::run getDataFromBuffer timecost(ms):" << (tEnd - tBegin) << endl);
                FDLOG("CountStat") << "stat ip:" << ServerConfig::LocalIp << "|Buffer Index:" << iBufferIndex << "|ReapSSDThread::run getDataFromBuffer timecost(ms):" << (tEnd - tBegin) << endl;

                TLOGDEBUG("stat ip:" << ServerConfig::LocalIp << "|Buffer Index:" << iBufferIndex << "|ReapSSDThread::run insert begin _vAllStatMsg.size:" << vAllStatMsg.size() << "|record num:" << iTotalNum << endl);
                FDLOG("CountStat") << "stat ip:" << ServerConfig::LocalIp << "|Buffer Index:" << iBufferIndex << "|ReapSSDThread::run insert begin _vAllStatMsg.size:" << vAllStatMsg.size() << "|record num:" << iTotalNum << endl;

                if(iTotalNum <= 0)
                {
                    for(int iStatIndex = 0; iStatIndex < dbNumber; ++iStatIndex)
                    {
                        delete vAllStatMsg[iStatIndex];
                    }

                    vAllStatMsg.clear();
                }
                else
                {

                    string sFile="";
                    string sDate="";
                    string sFlag="";
                    time_t time=0;
                    g_app.getTimeInfo(time,sDate,sFlag);

                    //size_t iSize = vAllStatMsg.size();

                    QueueItem item;
                    int iInsertThreadIndex = 0;
                    sRandOrder = g_app.getRandOrder();

                    if (sRandOrder == "")
                    {
                        sRandOrder = "0";
                    }

                    map<string, vector<size_t> >& mIpHasDbInfo = StatDbManager::getInstance()->getIpHasDbInfo();
                    map<string, vector<size_t> >::iterator m_iter = mIpHasDbInfo.begin();

                    while(m_iter != mIpHasDbInfo.end())
                    {
                        vector<size_t> &vDb = m_iter->second;

                        for(size_t i = 0; i < vDb.size(); ++i)
                        {
                            int k = (i + TC_Common::strto<int>(sRandOrder)) % vDb.size();

                            item._index = vDb[k];
                            item._date  = sDate;
                            item._tflag = sFlag;
                            item._statmsg = vAllStatMsg[item._index];

                            iInsertThreadIndex = StatDbManager::getInstance()->getDbToIpIndex(vDb[k]);

                            assert(iInsertThreadIndex >= 0);

                            _runners[iInsertThreadIndex]->put(item);
                        }

                        ++m_iter;
                    }

                    if(_terminate)
                    {
                        break;
                    }
                }

                for(int k = 0; k < g_app.getBuffNum(); ++k)
                {
                    StatHashMap *pHashMap = g_app.getHashMapBuff(iBufferIndex, k);
                    pHashMap->clear();
                }

                TLOGDEBUG("stat ip:" << ServerConfig::LocalIp << "|Buffer Index:" << iBufferIndex << "|ReapSSDThread::run insert record num:" << iTotalNum << "|tast patch finished." << endl);
                FDLOG("CountStat") << "stat ip:" << ServerConfig::LocalIp << "|Buffer Index:" << iBufferIndex << "|ReapSSDThread::run insert record num:" << iTotalNum << "|tast patch finished." << endl;
            }

        }
        catch(exception& ex)
        {
            TLOGERROR("ReapSSDThread::run exception:"<< ex.what() << endl);
        }
        catch(...)
        {
            TLOGERROR("ReapSSDThread::run ReapSSDThread unkonw exception catched" << endl);
        }

        TC_ThreadLock::Lock lock(*this);
        timedWait(REAP_INTERVAL);
    }

    TLOGDEBUG("ReapSSDThread run setITerminateFlag true." << endl);

    StatDbManager::getInstance()->setITerminateFlag(true);

    for(size_t i = 0; i < _runners.size(); ++i)
    {
        if(_runners[i]->isAlive())
        {
            _runners[i]->terminate();

            _runners[i]->getThreadControl().join();
        }
    }

    for(size_t i = 0; i < _runners.size(); ++i)
    {
        if(_runners[i])
        {
            delete _runners[i];
            _runners[i] = NULL;
        }
    }

    TLOGDEBUG("ReapSSDThread run terminate." << endl);
}

int ReapSSDThread::getIndexWithWeighted(int iMaxDb,int iGcd,int iMaxW,const vector<int>& vDbWeight)
{
    while (true){

        _lastSq = (_lastSq + 1) % iMaxDb;

        if (_lastSq == 0)
        {
            _curWeight = _curWeight - iGcd;
            if (_curWeight <= 0)
            {
                _curWeight = iMaxW;
                if(_curWeight == 0)
                {
                    return 0;
                }
            }
        }

        if (vDbWeight[_lastSq] >= _curWeight)
        {
            return _lastSq;
        }
    }
}
void ReapSSDThread::getDataFromBuffer(int iIndex, vector<StatMsg*> &vAllStatMsg, uint64_t &iTotalNum)
{
    TLOGDEBUG("ReapSSDThread::getDataFromBuffer iIndex:" << iIndex << "|begin..." << endl);

    try
    {
        int iCount = 0,dbSeq=0;

        //获取db个数
        int dbNumber = StatDbManager::getInstance()->getDbNumber();

        vector<int> vDbWeight;
        int iGcd = 0,iMaxW = 0;

        StatDbManager::getInstance()->getDbWeighted(iGcd,iMaxW,vDbWeight);

        bool bEnable = StatDbManager::getInstance()->IsEnableWeighted();

        for(int k = 0; k < g_app.getBuffNum(); ++k)
        {
            if(_terminate)
            {
                break;
            }

            StatHashMap *pHashMap = g_app.getHashMapBuff(iIndex, k);

            if(pHashMap->size() == 0)
            {
                continue ;
            }

            FDLOG("CountStat") << "stat ip:" << ServerConfig::LocalIp << "|Buffer Index:" << iIndex << "|ReapSSDThread::getData load hashmap k:" << k << endl;

            HashMap::lock_iterator it = pHashMap->beginSetTime();
            while ( it != pHashMap->end() )
            {
                if(_terminate)
                {
                    break;
                }

                StatMicMsgHead head;
                StatMicMsgBody body;
                int ret = it->get( head, body );
                if ( ret < 0 )
                {
                    ++it;
                    continue;
                }

                if (dbNumber > 0)
                {
                    if(bEnable)//按权重入库
                    {
                        dbSeq = getIndexWithWeighted(dbNumber,iGcd,iMaxW,vDbWeight);
                        TLOGINFO("ReapSSDThread::getIndexWithWeighted |" << dbSeq << endl);
                    }
                    else
                    {
                        dbSeq = iCount % dbNumber;
                    }

                    (*(vAllStatMsg[dbSeq]))[head] = body;
                }

                iCount++;

                ++it;
            }

        }

        iTotalNum = iCount;

        TLOGDEBUG("ReapSSDThread::getDataFromBuffer Buffer Index:" << iIndex << "|get total size:" << iCount << endl);
        FDLOG("CountStat") << "stat ip:" << ServerConfig::LocalIp << "|Buffer Index:" << iIndex << "|ReapSSDThread::getData get total size:" << iCount << "|end..." << endl;
    }
    catch (exception& ex)
    {
        TLOGERROR("ReapSSDThread::getDataFromBuffer exception:" << ex.what() << endl);
        FDLOG("CountStat") << "stat ip:" << ServerConfig::LocalIp << "|Buffer Index:" << iIndex << "|ReapSSDThread::getData exception:" << ex.what() << endl;

        string sMsg("ReapSSDThread::getDataFromBuffer Buffer Index:");
        sMsg += TC_Common::tostr(iIndex);
        sMsg += " exception:";
        sMsg += ex.what();
        sendAlarmSMS(sMsg);
    }
}

int ReapSSDThread::sendAlarmSMS(const string &sMsg)
{
    string errInfo = " ERROR:" + ServerConfig::LocalIp + "_" + sMsg;
    TARS_NOTIFY_ERROR(errInfo);

    return 0;
}