mariadb-columnstore-engine/writeengine/client/we_clients.cpp

/* Copyright (C) 2014 InfiniDB, Inc.
 * Copyright (C) 2016 MariaDB Corporation.

   This program is free software; you can redistribute it and/or
   modify it under the terms of the GNU General Public License
   as published by the Free Software Foundation; version 2 of
   the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
   MA 02110-1301, USA. */

// $Id: weclients.h 525 2010-01-19 23:18:05Z xlou $
//
/** @file */

#include <sstream>
#include <stdexcept>
#include <cassert>
#include <ctime>
#include <algorithm>
#include <unistd.h>
#ifndef _MSC_VER
#include <arpa/inet.h>
#if __FreeBSD__
#include <sys/socket.h>
#endif
#endif
using namespace std;

#include <boost/thread/mutex.hpp>
using namespace boost;

#include "messagequeue.h"
#include "bytestream.h"
using namespace messageqcpp;

#include "configcpp.h"
using namespace config;

#include "errorids.h"
#include "exceptclasses.h"
#include "messagelog.h"
#include "messageobj.h"
#include "loggingid.h"
using namespace logging;

#include "liboamcpp.h"
#include "alarmmanager.h"
using namespace alarmmanager;
using namespace oam;

#include "we_clients.h"
#include "we_messages.h"
using namespace WriteEngine;

#include "atomicops.h"

namespace
{

void  writeToLog(const char* file, int line, const string& msg, LOG_TYPE logto = LOG_TYPE_INFO)
{
    LoggingID lid(05);
    MessageLog ml(lid);
    Message::Args args;
    Message m(0);
    args.add(file);
    args.add("@");
    args.add(line);
    args.add(msg);
    m.format(args);

    switch (logto)
    {
        case LOG_TYPE_DEBUG:
            ml.logDebugMessage(m);
            break;

        case LOG_TYPE_INFO:
            ml.logInfoMessage(m);
            break;

        case LOG_TYPE_WARNING:
            ml.logWarningMessage(m);
            break;

        case LOG_TYPE_ERROR:
            ml.logErrorMessage(m);
            break;

        case LOG_TYPE_CRITICAL:
            ml.logCriticalMessage(m);
            break;
    }
}

string getModuleNameByIPAddr(oam::ModuleTypeConfig moduletypeconfig,
                             string ipAddress)
{
    string modulename = "pm1";
    DeviceNetworkList::iterator pt = moduletypeconfig.ModuleNetworkList.begin();

    for ( ; pt != moduletypeconfig.ModuleNetworkList.end() ; pt++)
    {
        modulename = (*pt).DeviceName;
        HostConfigList::iterator pt1 = (*pt).hostConfigList.begin();

        for ( ; pt1 != (*pt).hostConfigList.end() ; pt1++)
        {
            if (ipAddress == (*pt1).IPAddr)
                return modulename;
        }
    }

    return modulename;
}

struct WEClientRunner
{
    WEClientRunner(WriteEngine::WEClients* jl,
                   boost::shared_ptr<MessageQueueClient> cl, uint32_t connectionIndex) : jbl(jl), client(cl),
        connIndex(connectionIndex) {}
    WriteEngine::WEClients* jbl;
    boost::shared_ptr<MessageQueueClient> client;
    uint32_t connIndex;
    void operator()()
    {
        //cout << "Listening on client at 0x" << hex << (ptrdiff_t)client << dec << endl;
        try
        {
            jbl->Listen(client, connIndex);
            //cout << "Listening connIndex " << connIndex << endl;
        }
        catch (std::exception& ex)
        {
            string what(ex.what());
            cerr << "exception caught in WEClient: " << what << endl;

            if (what.find("St9bad_alloc") != string::npos)
            {
                writeToLog(__FILE__, __LINE__, what, LOG_TYPE_CRITICAL);
//           abort();
            }
            else  writeToLog(__FILE__, __LINE__, what);
        }
        catch (...)
        {
            string msg("exception caught in WEClientRunner.");
            writeToLog(__FILE__, __LINE__, msg);
            cerr << msg << endl;
        }
    }
};

template <typename T>
struct QueueShutdown : public unary_function<T&, void>
{
    void operator()(T& x)
    {
        x.shutdown();
    }
};

inline const string sin_addr2String(const in_addr src)
{
    string s;
#ifdef _MSC_VER
    s = inet_ntoa(src);
#else
    char dst[INET_ADDRSTRLEN];
    s = inet_ntop(AF_INET, &src, dst, INET_ADDRSTRLEN);
#endif
    return s;
}
}

namespace WriteEngine
{
WEClients::WEClients(int PrgmID) :
    fPrgmID(PrgmID),
    pmCount(0)
{
    closingConnection = 0;
    Setup();
}

WEClients::~WEClients()
{

    Close();
}

void WEClients::Setup()
{
    makeBusy(true);
    joblist::ResourceManager* rm = joblist::ResourceManager::instance();
    oam::Oam oam;
    string ipAddress;
    ModuleTypeConfig moduletypeconfig;

    try
    {
        oam.getSystemConfig("pm", moduletypeconfig);
    }
    catch (...)
    {
        writeToLog(__FILE__, __LINE__, "oam.getSystemConfig error, unknown exception", LOG_TYPE_ERROR);
        throw runtime_error("Setup failed");
    }

    uint32_t pmCountConfig =  moduletypeconfig.ModuleCount;
    pmCount = 0;
    int moduleID = 1;

    char buff[32];
    ByteStream bs, bsRead;

    if (fPrgmID == DDLPROC)
    {
        bs << (ByteStream::byte) WE_SVR_DDL_KEEPALIVE;
        bs << (ByteStream::octbyte) moduleID;
    }
    else if (fPrgmID == DMLPROC)
    {
        bs << (ByteStream::byte) WE_SVR_DML_KEEPALIVE;
        bs << (ByteStream::octbyte) moduleID;
    }
    else if (fPrgmID == SPLITTER)
    {
        bs << (ByteStream::byte) WE_CLT_SRV_KEEPALIVE;
    }
    else if (fPrgmID == BATCHINSERTPROC)
    {
        bs << (ByteStream::byte) WE_SVR_BATCH_KEEPALIVE;
        bs << (ByteStream::octbyte) moduleID;
    }

    for (unsigned i = 0; i < pmCountConfig; i++)
    {
        //Find the module id
        moduleID = atoi((moduletypeconfig.ModuleNetworkList[i]).DeviceName.substr(MAX_MODULE_TYPE_SIZE, MAX_MODULE_ID_SIZE).c_str());
        //cout << "setting connection to moduleid " << moduleID << endl;
        snprintf(buff, sizeof(buff), "pm%u_WriteEngineServer", moduleID);
        string fServer (buff);

        boost::shared_ptr<MessageQueueClient>
        cl(new MessageQueueClient(fServer, rm->getConfig()));
        boost::shared_ptr<boost::mutex> nl(new boost::mutex());

        //Bug 5224. Take out the retrys. If connection fails, we assume the server is down.
        try
        {
            if (cl->connect())
            {
                try
                {
                    cl->write(bs);
                }
                catch (std::exception& ex1)
                {
                    ostringstream oss;
                    oss << "Write to WES during connect failed due to " << ex1.what();
                    throw runtime_error(oss.str());
                }

                try
                {
                    bsRead = cl->read();

                    if (bsRead.length() == 0)
                        throw runtime_error("Got byte 0 during reading " );
                }
                catch (std::exception& ex2)
                {
                    ostringstream oss;
                    oss << "Read from WES during connect failed due to " << ex2.what() << " and this = " << this;
                    throw runtime_error(oss.str());
                }

                fPmConnections[moduleID] = cl;
                //cout << "connection is open. this = " << this << endl;
                //cout << "set up connection to mudule " << moduleID << endl;
                // assign the module name
                //ipAddress = sin_addr2String(cl->serv_addr().sin_addr);
                ipAddress = cl->addr2String();
                cl->moduleName(getModuleNameByIPAddr(moduletypeconfig, ipAddress));
                StartClientListener(cl, i);
                pmCount++;
                //ostringstream oss;
                //oss << "WECLIENT: connected to " << fServer + " and this = " << this << " and pmcount is now " << pmCount;
                //writeToLog(__FILE__, __LINE__, oss.str() , LOG_TYPE_DEBUG);
            }
            else
            {
                throw runtime_error("Connection refused");
            }
        }
        catch (std::exception& ex)
        {
            writeToLog(__FILE__, __LINE__, "Could not connect to " + fServer + ": " + ex.what(), LOG_TYPE_ERROR);
            cerr << "Could not connect to " << fServer << ": " << ex.what() << endl;
        }
        catch (...)
        {
            writeToLog(__FILE__, __LINE__, "Could not connect to " + fServer, LOG_TYPE_ERROR);
        }
    }

}

int WEClients::Close()
{
    makeBusy(false);
    closingConnection = 1;
    ByteStream bs;
    bs << (ByteStream::byte) WE_SVR_CLOSE_CONNECTION;
    write_to_all(bs);

//cout << "connection is closed. this = " << this << " and closingConnection = " << closingConnection << endl;
    for (uint32_t i = 0; i < fWESReader.size(); i++)
    {
        fWESReader[i]->join();
    }

    fWESReader.clear();
    fPmConnections.clear();
    pmCount = 0;
    //ostringstream oss;
    //oss << "WECLIENT: closed connection to wes and this = " << this << " and pmcount is now " << pmCount;
    //writeToLog(__FILE__, __LINE__, oss.str() , LOG_TYPE_DEBUG);
    return 0;
}

void WEClients::Listen ( boost::shared_ptr<MessageQueueClient> client, uint32_t connIndex)
{
    SBS sbs;

    try
    {
        while ( Busy() )
        {
            //TODO: This call blocks so setting Busy() in another thread doesn't work here...
            sbs = client->read();

            if ( sbs->length() != 0 )
            {
                //cout << "adding data to connIndex " << endl;
                addDataToOutput(sbs, connIndex);
            }
            else // got zero bytes on read, nothing more will come
            {
                if (closingConnection > 0)
                {
                    return;
                }

                cerr << "WEC got 0 byte message for object " << this << endl;
                goto Error;
            }
        }

        return;
    }
    catch (std::exception& e)
    {
        cerr << "WEC Caught EXCEPTION: " << e.what() << endl;
        goto Error;
    }
    catch (...)
    {
        cerr << "WEC Caught UNKNOWN EXCEPT" << endl;
        goto Error;
    }

Error:
    // error condition! push 0 length bs to messagequeuemap and
    // eventually let jobstep error out.
    mutex::scoped_lock lk(fMlock);

    MessageQueueMap::iterator map_tok;
    sbs.reset(new ByteStream(0));

    for (map_tok = fSessionMessages.begin(); map_tok != fSessionMessages.end(); ++map_tok)
    {
        map_tok->second->queue.clear();
        (void)atomicops::atomicInc(&map_tok->second->unackedWork[0]);
        map_tok->second->queue.push(sbs);
    }

    lk.unlock();

    // reset the pmconnection map
    {
        mutex::scoped_lock onErrLock(fOnErrMutex);
        string moduleName = client->moduleName();
        ClientList::iterator itor = fPmConnections.begin();

        while (itor != fPmConnections.end())
        {
            if (moduleName == (itor->second)->moduleName())
            {
                (fPmConnections[itor->first]).reset();
                pmCount--;
                ostringstream oss;
                //oss << "WECLIENT: connection to is reset and this = " << this << " and pmcount is decremented.";
                //writeToLog(__FILE__, __LINE__, oss.str() , LOG_TYPE_DEBUG);
            }

            itor++;
        }

        // send alarm
        ALARMManager alarmMgr;
//		string alarmItem = sin_addr2String(client->serv_addr().sin_addr);
        string alarmItem = client->addr2String();
        alarmItem.append(" WriteEngineServer");
        alarmMgr.sendAlarmReport(alarmItem.c_str(), oam::CONN_FAILURE, SET);
    }
    return;
}

void WEClients::addQueue(uint32_t key)
{
    bool b;

    mutex* lock = new mutex();
    condition* cond = new condition();
    boost::shared_ptr<MQE> mqe(new MQE(pmCount));

    mqe->queue = WESMsgQueue(lock, cond);

    mutex::scoped_lock lk ( fMlock );
    b = fSessionMessages.insert(pair<uint32_t, boost::shared_ptr<MQE> >(key, mqe)).second;

    if (!b)
    {
        ostringstream os;
        os << "WEClient: attempt to add a queue with a duplicate ID " << key << endl;
        throw runtime_error(os.str());
    }
}

void WEClients::removeQueue(uint32_t key)
{
    mutex::scoped_lock lk(fMlock);
    MessageQueueMap::iterator map_tok = fSessionMessages.find(key);

    if (map_tok == fSessionMessages.end())
        return;

    map_tok->second->queue.shutdown();
    map_tok->second->queue.clear();
    fSessionMessages.erase(map_tok);
}

void WEClients::shutdownQueue(uint32_t key)
{
    mutex::scoped_lock lk(fMlock);
    MessageQueueMap::iterator map_tok = fSessionMessages.find(key);

    if (map_tok == fSessionMessages.end())
        return;

    map_tok->second->queue.shutdown();
    map_tok->second->queue.clear();
}

void WEClients::read(uint32_t key, SBS& bs)
{
    boost::shared_ptr<MQE> mqe;

    //Find the StepMsgQueueList for this session
    mutex::scoped_lock lk(fMlock);
    MessageQueueMap::iterator map_tok = fSessionMessages.find(key);

    if (map_tok == fSessionMessages.end())
    {
        ostringstream os;
        //cout << " reading for key " << key << " not found" << endl;
        os << "WEClient: attempt to read(bs) from a nonexistent queue\n";
        throw runtime_error(os.str());
    }

    mqe = map_tok->second;
    lk.unlock();

    //this method can block: you can't hold any locks here...
    (void)mqe->queue.pop(&bs);

    if (!bs)
        bs.reset(new ByteStream());
}

void WEClients::write(const messageqcpp::ByteStream& msg, uint32_t connection)
{
    if (pmCount == 0)
    {
        ostringstream oss;
        oss << "WECLIENT: There is no connection to WES and this = " << this ;
        writeToLog(__FILE__, __LINE__, oss.str(), LOG_TYPE_DEBUG);
        throw runtime_error("There is no WriteEngineServer to send message to.");
    }

    if (fPmConnections[connection] != 0)
        fPmConnections[connection]->write(msg);
    else
    {
        ostringstream os;
        os << "Lost connection to WriteEngineServer on pm" << connection;
        throw runtime_error(os.str());
    }
}

void WEClients::write_to_all(const messageqcpp::ByteStream& msg)
{
    if (pmCount == 0)
    {
        ostringstream oss;
        oss << "WECLIENT:  There is no connection to WES and this = " << this ;
        writeToLog(__FILE__, __LINE__, oss.str(), LOG_TYPE_DEBUG);
        throw runtime_error("There is no WriteEngineServer to send message to.");
    }

    ClientList::iterator itor = fPmConnections.begin();

    while (itor != fPmConnections.end())
    {
        if (itor->second != NULL)
        {
            itor->second->write(msg);
        }

        itor++;
    }
}

void WEClients::StartClientListener(boost::shared_ptr<MessageQueueClient> cl, uint32_t connIndex)
{
    boost::thread* thrd = new boost::thread(WEClientRunner(this, cl, connIndex));
    fWESReader.push_back(thrd);
}


void WEClients::addDataToOutput(SBS sbs, uint32_t connIndex)
{
    // ISMPacketHeader *hdr = (ISMPacketHeader*)(sbs->buf());
    // PrimitiveHeader *p = (PrimitiveHeader *)(hdr+1);
    //uint32_t uniqueId = p->UniqueID;
    uint64_t uniqueId = 0;
    *sbs >> uniqueId;
    boost::shared_ptr<MQE> mqe;

    mutex::scoped_lock lk(fMlock);
    MessageQueueMap::iterator map_tok = fSessionMessages.find(uniqueId);

    if (map_tok == fSessionMessages.end())
    {
        return;
    }

    mqe = map_tok->second;
    lk.unlock();

    if (pmCount > 0)
    {
        atomicops::atomicInc(&mqe->unackedWork[connIndex % pmCount]);
    }

    (void)mqe->queue.push(sbs);

}

}

// vim:ts=4 sw=4: