/* Copyright (C) 2014 InfiniDB, Inc.
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$
*
*******************************************************************************/
/*
* we_dataloader.cpp
*
* Created on: Oct 4, 2011
* Author: Boby Paul: bpaul@calpont.com
*/
#include "mcsconfig.h" // Used to pickup STRERROR_R_CHAR_P definition
#include <cstdlib>
#include <csignal>
#include <cstring>
#include <cerrno>
#include <unistd.h> //pipe() && fork()
#include <wait.h> //wait()
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include "bytestream.h"
#include "rwlock_local.h"
#include <iostream>
#include <fstream>
#include <vector>
#include <queue>
#include <string>
#include <map>
using namespace std;
#include <boost/thread/condition.hpp>
#include <boost/scoped_array.hpp>
#include <boost/thread.hpp>
#include <boost/filesystem.hpp>
using namespace boost;
#include "bytestream.h"
#include "messagequeue.h"
using namespace messageqcpp;
#include "we_messages.h"
#include "we_brmrprtparser.h"
#include "we_cleartablelockcmd.h"
#include "we_dataloader.h"
#include "we_readthread.h"
#include "installdir.h"
namespace WriteEngine
{
// bool WEDataLoader::fTearDownCpimport=false; // @bug 4267
//-----------------------------------------------------------------------------
/**
*
* @brief WEDataLoader::Constructor
*
**/
WEDataLoader::WEDataLoader(SplitterReadThread& Srt)
: fRef(Srt)
, fMode(-1)
, fDataDumpFile()
, fTxBytes(0)
, fRxBytes(0)
, fPmId(0)
, fCh_pid(0)
, fThis_pid(0)
, fP_pid(0)
, fpCfThread(0)
, fTearDownCpimport(false)
, // @bug 4267
fWaitPidRc(0)
, // @bug 4267
fWaitPidStatus(0)
, // @bug 4267
fForceKill(false)
, fPipeErr(false)
, fpSysLog(0)
{
Config weConfig;
uint16_t localModuleId = weConfig.getLocalModuleID();
fPmId = static_cast<char>(localModuleId);
srand(time(NULL)); // initialize random seed
int aObjId = rand() % 10000 + 1; // generate a random number
setObjId(aObjId);
setupSignalHandlers();
if (!fpSysLog)
{
fpSysLog = SimpleSysLog::instance();
fpSysLog->setLoggingID(logging::LoggingID(SUBSYSTEM_ID_WE_SRV));
}
}
//-----------------------------------------------------------------------------
/**
*
* @brief WEDataLoader::Destructor
*
**/
WEDataLoader::~WEDataLoader()
{
try
{
if (fDataDumpFile.is_open())
fDataDumpFile.close();
cout << "\tRx Bytes " << getRxBytes() << endl;
cout << "\tTX Bytes " << getTxBytes() << endl;
cout << "\tChild PID " << getChPid() << endl;
if (getChPid())
{
if (2 == getMode()) //@bug 5012
{
kill(getChPid(), SIGINT);
teardownCpimport(fTearDownCpimport);
}
else
{
teardownCpimport(false); // @bug 4267
}
}
}
catch (std::exception&) // @bug 4164: exception causing thread to exit
{
cout << "Error tearing down cpimport in WEDataLoader destructor" << endl;
}
// cout << "Leaving WEDataLoader destructor" << endl;
}
//------------------------------------------------------------------------------
// Initialize signal handling
//------------------------------------------------------------------------------
void WEDataLoader::setupSignalHandlers()
{
signal(SIGPIPE, SIG_IGN);
signal(SIGCHLD, WEDataLoader::onSigChild);
}
//------------------------------------------------------------------------------
// handles on signal Terminate
//------------------------------------------------------------------------------
void WEDataLoader::onSigChild(int aInt)
{
std::string aStr = "Received SIGCHLD of terminated process..";
cout << aStr << endl;
// fTearDownCpimport = true; // @bug 4267
// commented out for non-static variables
// ostringstream oss;
// oss << getObjId() <<" : " <<aStr;
// logging::Message::Args errMsgArgs;
// errMsgArgs.add(oss.str());
// fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_ERROR, logging::M0000);
}
//-----------------------------------------------------------------------------
/**
*
* @brief WEDataLoader::update
*
**/
bool WEDataLoader::update(Subject* pSub)
{
return true;
}
//-----------------------------------------------------------------------------
/**
*
* @brief setup cpimport as a seperate process
*
**/
bool WEDataLoader::setupCpimport() // fork the cpimport
{
pid_t aChPid;
errno = 0;
if (pipe(fFIFO) == -1)
{
int errnum = errno;
ostringstream oss;
oss << getObjId() << " : Error in creating pipe (errno-" << errnum << "); " << strerror(errnum);
throw runtime_error(oss.str());
}
setPid(getpid());
setPPid(getppid());
errno = 0;
aChPid = fork();
if (aChPid == -1) // an error caused
{
int errnum = errno;
ostringstream oss;
oss << getObjId() << " : Error in forking cpimport.bin (errno-" << errnum << "); " << strerror(errnum);
throw runtime_error(oss.str());
}
else if (aChPid == 0) // we are in child
{
int aStartFD = 3;
int aEndFD = fFIFO[1] + 256;
close(fFIFO[1]); // close the WRITER of CHILD
cout << "Child Process Info: PID = " << getpid() << " (fFIFO[0], fFIFO[1]) = (" << fFIFO[0] << ","
<< fFIFO[1] << ")"
<< " (StartFD, EndFD) = (" << aStartFD << "," << aEndFD << ")" << endl;
std::vector<char*> Cmds;
// str2Argv(fCmdLineStr, Cmds); // to avoid out-of-scope problem
std::string aCmdLine = fCmdLineStr;
std::istringstream ss(aCmdLine);
std::string arg;
std::vector<std::string> v2(20, "");
unsigned int i = 0;
while (ss >> arg)
{
v2[i++] = arg;
}
for (unsigned int j = 0; j < i; ++j)
{
Cmds.push_back(const_cast<char*>(v2[j].c_str()));
}
Cmds.push_back(0); // null terminate
// updatePrgmPath(Cmds);
// NOTE: for debugging
int aSize = Cmds.size();
for (int aIdx = 0; aIdx < aSize; ++aIdx)
{
cout << "Args " << Cmds[aIdx] << endl;
}
cout.flush();
close(0); // close stdin for the child
dup2(fFIFO[0], 0); // make stdin be the reading end of the pipe
// BUG 4410 : hacky solution so that CHLD process get EOF on close of pipe
for (int i = aStartFD; i < aEndFD; i++)
close(i);
errno = 0;
int aRet = execvp(Cmds[0], &Cmds[0]); // NOTE - works with full Path
// int aRet = execvp(Cmds[0], &Cmds[0]); //NOTE - works if $PATH has cpimport
int execvErrno = errno;
cout << "Return status of cpimport is " << aRet << endl;
cout.flush();
close(fFIFO[0]); // will trigger an EOF on stdin
ostringstream oss;
oss << getObjId() << " : execv error: cpimport.bin invocation failed; "
<< "(errno-" << errno << "); " << strerror(execvErrno) << "; Check file and try invoking locally.";
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_ERROR, logging::M0000);
if (aRet == -1)
exit(-1);
}
else // parent
{
setChPid(aChPid); // This is the child PID
cout << "Child PID is " << this->getChPid() << endl;
close(fFIFO[0]); // close the READER of PARENT
// now we can send all the data thru FIFO[1], writer of PARENT
}
if (aChPid == 0)
cout << "******** Child finished its work ********" << endl;
return true;
}
//-----------------------------------------------------------------------------
/**
*
* @brief close all file handles opened for cpimport
* @brief wait for the cpimport process to finish work
*
**/
void WEDataLoader::teardownCpimport(bool useStoredWaitPidStatus) // @bug 4267
{
fTearDownCpimport = false; // Reset it
// cout << "Tearing down Cpimport" << endl;
int aStatus;
// cout << "checking fpCfThread value" << endl;
if (fpCfThread)
{
// cout << "checking fpCfThread has a valid value" << endl;
// wait until we are done with the queued messages
while ((!fpCfThread->isMsgQueueEmpty()) && (!fpCfThread->isStopped()))
{
// cout << "DEBUG : MsgQueue not empty" << endl;
// cannot be too low, since there is a lock in isMsgQueueEmpty()
usleep(2000000);
}
// while(fpCfThread->isPushing())
// {
// cout << "DEBUG : still pushing" << endl;
// usleep(100000);
// }
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << " : Message Queue is empty; Stopping CF Thread";
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_DEBUG, logging::M0000);
}
fpCfThread->stopThread();
while (!fpCfThread->isStopped())
{
cout << "DEBUG : still not stopped" << endl;
usleep(100000);
}
delete fpCfThread;
fpCfThread = 0;
}
closeWritePipe();
pid_t aPid;
// @bug 4267 begin: call waitpid() to get job status or use stored job status
// aPid = waitpid(getChPid(), &aStatus, 0); // wait until cpimport finishs
if (useStoredWaitPidStatus)
{
aPid = fWaitPidRc;
aStatus = fWaitPidStatus;
}
else
{
// aPid = waitpid(getChPid(), &aStatus, 0); // wait until cpimport finishs
aPid = waitpid(getChPid(), &aStatus, WNOHANG); // wait until cpimport finishs
int aIdx = 0;
while ((aPid == 0) && (aIdx < 25 * MAX_QSIZE)) // Do not loop infinitly
{
usleep(2000000);
aPid = waitpid(getChPid(), &aStatus, WNOHANG);
cout << "Inside tearDown waitpid rc[" << aIdx << "] = " << aPid << endl;
++aIdx;
}
}
// @bug 4267 end // BP - added -1 as per DMC comment below
if ((aPid == getChPid()) || (aPid == -1)) // @bug 4267 (DMC-shouldn't we check for aPid of -1?)
{
setChPid(0);
if ((WIFEXITED(aStatus)) && (WEXITSTATUS(aStatus) == 0))
{
cout << "\tCpimport exit on success" << endl;
ostringstream oss;
oss << getObjId() << " : cpimport exit on success";
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_INFO, logging::M0000);
onCpimportSuccess();
}
else
{
int termsig = (WIFSIGNALED(aStatus) ? WTERMSIG(aStatus) : -1);
if (!fForceKill)
{
cout << "\tCpimport exit on failure (signal " << termsig << ')' << endl;
ostringstream oss;
oss << getObjId() << " : cpimport exit on failure (signal " << termsig << ')';
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_ERROR, logging::M0000);
onCpimportFailure();
}
else
{
cout << "\tCpimport exit on Force Kill!!" << endl;
ostringstream oss;
oss << getObjId() << " : cpimport exit on Force kill!!";
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_INFO, logging::M0000);
onCpimportSuccess();
}
}
}
}
//-----------------------------------------------------------------------------
/**
* @brief Push the data to cpimport from the incoming ByteStream
* @param Incoming ByteStream
*
*/
//
void WEDataLoader::pushData2Cpimport(ByteStream& Ibs)
{
if (Ibs.length() > 0)
{
int aLen = Ibs.length();
char* pStart = reinterpret_cast<char*>(Ibs.buf());
char* pEnd = pStart + aLen;
char* pPtr = pStart;
while (pPtr < pEnd)
{
// if(pEnd > (pPtr + MAX_LEN))
// {
// int aRet = write(fFIFO[1], pPtr, MAX_LEN);
// if(aRet == -1) throw runtime_error("Pipe write error");
// //write(fFIFO[1], Ibs.buf(), Ibs.length());
// pPtr += MAX_LEN;
// }
// else
// {
// int aStrLen = pEnd - pPtr;
// int aRet = write(fFIFO[1], pPtr, aStrLen);
// if(aRet == -1) throw runtime_error("Pipe write error");
// pPtr += aStrLen;
// }
try
{
int aRet = write(fFIFO[1], pPtr, pEnd - pPtr);
if (aRet < 0)
{
if (!fPipeErr)
{
int e = errno;
fPipeErr = true;
std::string aStr = "pushing data : PIPE error .........";
char errMsgBuf[160];
#if STRERROR_R_CHAR_P
const char* pErrMsg = strerror_r(e, errMsgBuf, sizeof(errMsgBuf));
if (pErrMsg)
aStr += pErrMsg;
#else
int errMsgRc = strerror_r(e, errMsgBuf, sizeof(errMsgBuf));
if (errMsgRc == 0)
aStr += errMsgBuf;
#endif
logging::Message::Args errMsgArgs;
errMsgArgs.add(aStr);
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_ERROR, logging::M0000);
}
throw runtime_error("Pipe Error - cpimport.bin exited already!!");
}
pPtr += aRet;
}
catch (...)
{
// std::string aStr = "pushing data PIPE error .........";
// logging::Message::Args errMsgArgs;
// errMsgArgs.add(aStr);
// fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_ERROR, logging::M0000);
throw runtime_error("Pipe Error - cpimport.bin exited already!!");
}
}
}
}
//-----------------------------------------------------------------------------
/**
*
* @brief Close the pipe through which data was written to cpimport
* @brief This will also signal a EOF to the reading pipe.
*
**/
void WEDataLoader::closeWritePipe()
{
cout << "Going to close call = " << fFIFO[1] << endl;
// NOTE this will flush the file buffer and close it.
int aRet = close(fFIFO[1]); // will trigger EOD
cout << "----- closed both pipes -------- aRet = " << aRet << endl;
}
//-----------------------------------------------------------------------------
/**
*
* @brief Tokenize a string into char** argv format and store in a vector
* @brief we pass the V as arguments to exec cpimport
* @param CmdLine is the string form of arguments demlimited by space
* @param V vector which contains each element of argv
*
**/
void WEDataLoader::str2Argv(std::string CmdLine, std::vector<char*>& V)
{
std::istringstream ss(CmdLine);
std::string arg;
std::vector<std::string> v2;
while (ss >> arg)
{
v2.push_back(arg);
V.push_back(const_cast<char*>(v2.back().c_str()));
}
V.push_back(0); // null terminate
}
//-----------------------------------------------------------------------------
/**
*
* @brief Event to trigger when cpimport is successful.
*
**/
void WEDataLoader::onCpimportSuccess()
{
ByteStream obs;
cout << "Sending BRMRPT" << endl;
obs << (ByteStream::byte)WE_CLT_SRV_BRMRPT;
obs << (ByteStream::byte)fPmId; // PM id
// for testing
// std::string fRptFileName("ReportFile.txt");
BrmReportParser aBrmRptParser;
bool aRet = aBrmRptParser.serialize(fBrmRptFileName, obs);
if (aRet)
{
boost::mutex::scoped_lock aLock(fClntMsgMutex);
// aBrmRptParser.unserialize(obs); - was for testing
updateTxBytes(obs.length());
try
{
fRef.fIos.write(obs);
}
catch (...)
{
cout << "Broken Pipe .." << endl;
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << " : Broken Pipe : socket write failed ";
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_INFO, logging::M0000);
}
}
aLock.unlock();
cout << "Finished Sending BRMRPT" << endl;
}
else
{
cout << "Failed to serialize BRMRpt " << endl;
}
// if(remove(fBrmRptFileName.c_str()) != 0)
// cout <<"Failed to delete BRMRpt File "<< fBrmRptFileName << endl;
// usleep(1000000); //sleep 1 second.
obs.reset();
obs << (ByteStream::byte)WE_CLT_SRV_CPIPASS;
obs << (ByteStream::byte)fPmId; // PM id
boost::mutex::scoped_lock aLock(fClntMsgMutex);
updateTxBytes(obs.length());
try
{
fRef.fIos.write(obs);
}
catch (...)
{
cout << "Broken Pipe .." << endl;
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << " : Broken Pipe : socket write failed ";
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_INFO, logging::M0000);
}
}
aLock.unlock();
cout << "Sent CPIPASS info" << endl;
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << " : onCpimportSuccess BrmReport Send";
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_DEBUG, logging::M0000);
}
}
//-----------------------------------------------------------------------------
/**
*
* @brief Event to trigger if a cpimport failure occurs.
*
**/
void WEDataLoader::onCpimportFailure()
{
// Send failure notice back to the parent splitter job
sendCpimportFailureNotice();
// Even if we failed, we have failure info in BRMRPT
ByteStream obs;
obs << (ByteStream::byte)WE_CLT_SRV_BRMRPT;
obs << (ByteStream::byte)fPmId; // PM id
BrmReportParser aBrmRptParser;
bool aRet = aBrmRptParser.serialize(fBrmRptFileName, obs);
if (aRet)
{
boost::mutex::scoped_lock aLock(fClntMsgMutex);
updateTxBytes(obs.length());
try
{
fRef.fIos.write(obs);
}
catch (std::exception& ex)
{
cout << "Broken Pipe .." << ex.what() << endl;
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << " : Broken Pipe : socket write failed; " << ex.what();
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_ERROR, logging::M0000);
}
}
aLock.unlock();
cout << "Finished Sending BRMRPT" << endl;
}
if (remove(fBrmRptFileName.c_str()) != 0)
cout << "Failed to delete BRMRpt File " << fBrmRptFileName << endl;
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << " : onCpimportFailure BrmReport Send";
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_DEBUG, logging::M0000);
}
}
//-----------------------------------------------------------------------------
// Send msg to front-end splitter to notify it that a cpimport.bin pgm failed.
//-----------------------------------------------------------------------------
void WEDataLoader::sendCpimportFailureNotice()
{
ByteStream obs;
obs << (ByteStream::byte)WE_CLT_SRV_CPIFAIL;
obs << (ByteStream::byte)fPmId; // PM id
boost::mutex::scoped_lock aLock(fClntMsgMutex);
updateTxBytes(obs.length());
try
{
fRef.fIos.write(obs);
}
catch (...)
{
cout << "Broken Pipe .." << endl;
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << " : Broken Pipe : socket write failed ";
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_ERROR, logging::M0000);
}
}
}
//-----------------------------------------------------------------------------
/**
* @brief Event when a KEEPALIVE arrives.
* @param Incoming ByteStream, not used currently
*/
void WEDataLoader::onReceiveKeepAlive(ByteStream& Ibs)
{
/*
// TODO comment out when we done with debug
if(fpSysLog)
{
ostringstream oss;
oss << getObjId() <<" : Received KEEPALIVE";
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_INFO, logging::M0000);
}
*/
// cout << "Received KEEPALIVE" << endl;
// NOTE only seldom a KEEPALIVE receives,
// so nothing wrong in responding with a KEEPALIVE.
ByteStream obs;
obs << (ByteStream::byte)WE_CLT_SRV_KEEPALIVE;
obs << (ByteStream::byte)fPmId; // PM id
boost::mutex::scoped_lock aLock(fClntMsgMutex);
updateTxBytes(obs.length());
try
{
fRef.fIos.write(obs);
}
catch (...)
{
cout << "Broken Pipe .." << endl;
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << ": Broken Pipe : socket write failed ";
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_INFO, logging::M0000);
}
}
aLock.unlock();
// @bug 4267 begin
int aStatus;
pid_t aPid;
if (getChPid() > 0)
{
aPid = waitpid(getChPid(), &aStatus, WNOHANG); // wait until cpimport finishs
if (aPid != 0)
{
cout << "waitpid(" << getChPid() << "): rc-" << aPid << "; status-" << aStatus << "; exited-"
<< (WIFEXITED(aStatus)) << endl;
}
if ((aPid == getChPid()) || (aPid == -1))
{
fTearDownCpimport = true;
fWaitPidRc = aPid;
fWaitPidStatus = aStatus;
}
}
// @bug 4267 end
if (fTearDownCpimport)
{
// fTearDownCpimport = false; //Reset it // commented out to use the flag in EOD
cout << "Cpimport terminated " << endl;
if (0 == getMode())
onReceiveEod(Ibs);
else if (1 == getMode()) // mode 1 has to drive from UM
{
ByteStream obs;
obs << (ByteStream::byte)WE_CLT_SRV_EOD;
obs << (ByteStream::byte)fPmId; // PM id
boost::mutex::scoped_lock aLock(fClntMsgMutex);
updateTxBytes(obs.length());
try
{
fRef.fIos.write(obs);
}
catch (...)
{
cout << "Broken Pipe .." << endl;
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << ": Broken Pipe : socket write failed ";
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_INFO, logging::M0000);
}
}
aLock.unlock();
}
else if (2 == getMode())
{
// if(getChPid()) teardownCpimport(true); // @bug 4267
if (getChPid())
teardownCpimport(fTearDownCpimport); // @bug 4267 BP
}
}
else
{
// if(1 == getMode())
// if(fpCfThread) cout << "Queue Size = " << fpCfThread->getQueueSize() << endl;
}
}
//-----------------------------------------------------------------------------
/**
* @brief trigger when a DATA arrives.
* @param Incoming ByteStream which contains data
*/
void WEDataLoader::onReceiveData(ByteStream& Ibs)
{
if ((0 == getMode()) && (fDataDumpFile.is_open()))
{
// Will write to the output file.
fDataDumpFile << Ibs;
sendDataRequest();
}
else if (1 == getMode())
{
// commented out since we are going to use seperate thread
// pushData2Cpimport(Ibs);
if (fpCfThread)
{
fpCfThread->add2MsgQueue(Ibs);
// sendDataRequest(); // Need to control Queue Size
// Bug 5031 : Will only send 1 rqst for a batch to cpimport.bin
// if(fpCfThread->getQueueSize()<MIN_QSIZE) sendDataRequest();
// if(fpCfThread->getQueueSize()<MAX_QSIZE) sendDataRequest();
int aQsz = (fpCfThread) ? fpCfThread->getQueueSize() : 0;
// Bug 5031 : If Q size goes above 100 (2*250);
if (aQsz < MAX_QSIZE)
sendDataRequest();
if (aQsz > 1.5 * static_cast<int>(MAX_QSIZE)) // > 2*250
{
cout << "WARNING : Data Queuing up : QSize = " << aQsz << endl;
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << "WARNING : Data Queuing up : QSize = " << aQsz;
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_INFO, logging::M0000);
}
}
}
}
else if (2 == getMode())
{
cout << "onReceiveData : In Mode 2 NO data suppose to receive" << endl;
}
// ByteStream obs;
// obs << (ByteStream::byte)WE_CLT_SRV_DATARQST;
// obs << (ByteStream::byte)fPmId; // PM id
// updateTxBytes(obs.length());
// fRef.fIos.write(obs);
}
//-----------------------------------------------------------------------------
/**
* @brief trigger when a EOD arrives.
* @param Incoming ByteStream; not relevent for now
*/
void WEDataLoader::onReceiveEod(ByteStream& Ibs)
{
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << " : onReceiveEOD : child ID = " << getChPid();
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_DEBUG, logging::M0000);
}
cout << "Received EOD " << endl;
if (0 == getMode())
{
fDataDumpFile.close();
}
ByteStream obs;
obs << (ByteStream::byte)WE_CLT_SRV_EOD;
obs << (ByteStream::byte)fPmId; // PM id
boost::mutex::scoped_lock aLock(fClntMsgMutex);
updateTxBytes(obs.length());
try
{
fRef.fIos.write(obs);
}
catch (...)
{
cout << "Broken Pipe .." << endl;
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << " : Broken Pipe : socket write failed ";
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_INFO, logging::M0000);
}
}
aLock.unlock();
// if(( 1 == getMode())||( 2 == getMode()))
if (1 == getMode()) // BUG 4370 - seperated mode 1 & 2
{
// if(getChPid()) teardownCpimport(false); // @bug 4267
if (getChPid())
teardownCpimport(fTearDownCpimport); // @bug 4267 //BP changed to send the correct flag
}
else if (2 == getMode()) // BUG 4370
{
if (getChPid())
{
kill(getChPid(), SIGINT); // BUG 4370
fForceKill = true;
teardownCpimport(fTearDownCpimport); // BUG 4370
}
}
}
//-----------------------------------------------------------------------------
/**
* @brief Event on Command Received. It should contain sub commands
* @param Incoming ByteStream, will have sub commands
*/
void WEDataLoader::onReceiveCmd(ByteStream& bs)
{
// TODO - can be cpimport cmd or server cmd, for now write to a file
ByteStream::byte aCmdId;
bs >> aCmdId;
}
//-----------------------------------------------------------------------------
/**
* @brief The mode in which WES running.
* @param Incoming ByteStream, not relevent
*/
void WEDataLoader::onReceiveMode(ByteStream& Ibs)
{
// Assigning it here since WEDataLoader constructor is called multiple times
// while coping in readthread class.
if (!fpSysLog)
{
fpSysLog = SimpleSysLog::instance();
fpSysLog->setLoggingID(logging::LoggingID(SUBSYSTEM_ID_WE_SRV));
}
Ibs >> (ByteStream::quadbyte&)fMode;
cout << "Setting fMode = " << fMode << endl;
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << " : onReceiveMode() Setting fMode = " << fMode;
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_DEBUG, logging::M0000);
}
char aName[64];
snprintf(aName, sizeof(aName), "ModuleDBRootCount%d-3", fPmId);
string aStrDbRootCnt = config::Config::makeConfig()->getConfig("SystemModuleConfig", aName);
cout << "DbRootCnt = " << aStrDbRootCnt << endl;
ByteStream::byte aDbCnt = (ByteStream::byte)atoi(aStrDbRootCnt.c_str());
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << " : onReceiveMode() DbRoot Count = " + aStrDbRootCnt;
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_DEBUG, logging::M0000);
}
// Send No. of DBRoots to Client
ByteStream aObs;
aObs << (ByteStream::byte)WE_CLT_SRV_DBRCNT;
aObs << (ByteStream::byte)fPmId;
aObs << (ByteStream::byte)aDbCnt;
boost::mutex::scoped_lock aLock(fClntMsgMutex);
updateTxBytes(aObs.length());
try
{
fRef.fIos.write(aObs);
}
catch (...)
{
cout << "Broken Pipe .." << endl;
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << " : Broken Pipe : socket write failed ";
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_INFO, logging::M0000);
}
}
aLock.unlock();
}
//-----------------------------------------------------------------------------
/**
* @brief Acknowledgment. Not relevant at this point of time.
* @brief Can make use to update the BRM
* @param Incoming ByteStream, not relevant for now
*/
void WEDataLoader::onReceiveAck(ByteStream& Ibs)
{
// All is good
// update the status
}
//-----------------------------------------------------------------------------
/**
* @brief NAK. A Failure, Rollback should be initiated.
* @brief Can make use to update the BRM
* @param Incoming ByteStream, not relevant for now
*/
void WEDataLoader::onReceiveNak(ByteStream& Ibs)
{
// TODO - handle the problem
}
//-----------------------------------------------------------------------------
/**
* @brief ERROR. A Failure, Rollback should be initiated.
* @brief Can make use to update the BRM
* @param Incoming ByteStream, not relevant for now
*/
void WEDataLoader::onReceiveError(ByteStream& Ibs)
{
// TODO - handle the failure situation.
}
//------------------------------------------------------------------------------
// onReceiveCmdLineArgs - do what ever need to do with command line args
//------------------------------------------------------------------------------
/**
* @brief Command line args received.
* @brief Can make use to update the BRM
* @param Incoming ByteStream, not relevant for now
*/
void WEDataLoader::onReceiveCmdLineArgs(ByteStream& Ibs)
{
Ibs >> fCmdLineStr;
cout << "CMD LINE ARGS came in " << fCmdLineStr << endl;
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << " : CMD LINE ARGS came in " << fCmdLineStr;
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_DEBUG, logging::M0000);
}
ByteStream obs;
// TODO - Need to check all clear for starting CPI
if (0 != getMode())
{
obs << (ByteStream::byte)WE_CLT_SRV_STARTCPI;
}
else
{
obs << (ByteStream::byte)WE_CLT_SRV_DATARQST;
}
obs << (ByteStream::byte)fPmId; // PM id
boost::mutex::scoped_lock aLock(fClntMsgMutex);
updateTxBytes(obs.length());
try
{
fRef.fIos.write(obs);
}
catch (...)
{
cout << "Broken Pipe .." << endl;
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << " : Broken Pipe : socket write failed ";
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_INFO, logging::M0000);
}
}
aLock.unlock();
}
//-----------------------------------------------------------------------------
void WEDataLoader::onReceiveStartCpimport()
{
cout << "Start Cpimport command reached!!" << endl;
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << " : Start Cpimport command reached!!";
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_DEBUG, logging::M0000);
}
try
{
setupCpimport();
if (1 == getMode()) // create a thread to handle the data feeding part
{
fpCfThread = new WECpiFeederThread(*this);
fpCfThread->startFeederThread();
}
}
catch (std::exception& ex)
{
// send an CPI FAIL command back to splitter
if (fpSysLog)
{
logging::Message::Args errMsgArgs;
errMsgArgs.add(ex.what());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_ERROR, logging::M0000);
sendCpimportFailureNotice();
return;
}
}
if (1 == getMode()) // In mode 2/0 we do not rqst data.
{
sendDataRequest();
}
// We need to respond to KEEP ALIVES
// else if(2 == getMode()) // Now we wait till cpimport comes back
//{
// if(getChPid())
// teardownCpimport();
//}
}
//-----------------------------------------------------------------------------
void WEDataLoader::onReceiveBrmRptFileName(ByteStream& Ibs)
{
Ibs >> fBrmRptFileName;
cout << "Brm Rpt Filename Arrived " << fBrmRptFileName << endl;
// BUG 4645
string::size_type idx = fBrmRptFileName.find_last_of('/');
if (idx > 0 && idx < string::npos)
{
string dirname(fBrmRptFileName, 0, idx);
struct stat st;
if (stat(dirname.c_str(), &st) != 0)
{
cout << "Creating directory : " << dirname << endl;
boost::filesystem::create_directories(dirname.c_str());
}
}
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << " : Brm Rpt Filename Arrived " << fBrmRptFileName;
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_DEBUG, logging::M0000);
}
}
//-----------------------------------------------------------------------------
void WEDataLoader::onReceiveCleanup(ByteStream& Ibs)
{
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << " : OnReceiveCleanup arrived";
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_DEBUG, logging::M0000);
}
std::string aErrMsg;
WE_ClearTableLockCmd aClrTblLockCmd("DataLoader");
int aRet = aClrTblLockCmd.processCleanup(Ibs, aErrMsg);
ByteStream obs;
obs << (ByteStream::byte)WE_CLT_SRV_CLEANUP;
obs << (ByteStream::byte)fPmId; // PM id
if (aRet == 0)
obs << (ByteStream::byte)1; // cleanup success
else
obs << (ByteStream::byte)0; // cleanup failed
boost::mutex::scoped_lock aLock(fClntMsgMutex);
updateTxBytes(obs.length());
try
{
fRef.fIos.write(obs);
}
catch (...)
{
cout << "Broken Pipe .." << endl;
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << " : Broken Pipe : socket write failed ";
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_INFO, logging::M0000);
}
}
aLock.unlock();
}
//-----------------------------------------------------------------------------
void WEDataLoader::onReceiveRollback(ByteStream& Ibs)
{
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << " : OnReceiveRollback arrived";
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_DEBUG, logging::M0000);
}
std::string aErrMsg;
WE_ClearTableLockCmd aClrTblLockCmd("DataLoader");
int aRet = aClrTblLockCmd.processRollback(Ibs, aErrMsg);
ByteStream obs;
obs << (ByteStream::byte)WE_CLT_SRV_ROLLBACK;
obs << (ByteStream::byte)fPmId; // PM id
if (aRet == 0)
obs << (ByteStream::byte)1; // Rollback success
else
obs << (ByteStream::byte)0; // Rollback failed
boost::mutex::scoped_lock aLock(fClntMsgMutex);
updateTxBytes(obs.length());
try
{
fRef.fIos.write(obs);
}
catch (...)
{
cout << "Broken Pipe .." << endl;
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << " : Broken Pipe : socket write failed ";
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_INFO, logging::M0000);
}
}
aLock.unlock();
}
//-----------------------------------------------------------------------------
void WEDataLoader::onReceiveImportFileName(ByteStream& Ibs)
{
bool aGoodFile = true;
std::string aFileName;
Ibs >> aFileName;
// BUG 4245 : Need to check the file or path exists
{
std::fstream aFin;
aFin.open(aFileName.c_str(), std::ios::in);
if (aFin.is_open()) // File exist, send an ERROR immediately
{
// file exists
ByteStream obs;
obs << (ByteStream::byte)WE_CLT_SRV_IMPFILEERROR;
obs << (ByteStream::byte)fPmId;
updateTxBytes(obs.length());
boost::mutex::scoped_lock aLock(fClntMsgMutex);
try
{
fRef.fIos.write(obs);
}
catch (...)
{
cout << "Broken Pipe .." << endl;
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << " : Broken Pipe : socket write failed ";
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_INFO, logging::M0000);
}
}
aGoodFile = false;
aLock.unlock();
}
aFin.close();
}
if (aGoodFile)
{
fDataDumpFile.open(aFileName.c_str(), std::ios::app);
// BUG 4245 : If file dir is not existing, we need to fail this import
if (!fDataDumpFile.good())
{
ByteStream obs;
obs << (ByteStream::byte)WE_CLT_SRV_IMPFILEERROR;
obs << (ByteStream::byte)fPmId; // PM id
boost::mutex::scoped_lock aLock(fClntMsgMutex);
updateTxBytes(obs.length());
try
{
fRef.fIos.write(obs);
}
catch (...)
{
cout << "Broken Pipe .." << endl;
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << " : Broken Pipe : socket write failed ";
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_INFO, logging::M0000);
}
}
aLock.unlock();
}
}
}
//------------------------------------------------------------------------------
void WEDataLoader::onReceiveJobId(ByteStream& Ibs)
{
std::string aJobFileName;
Ibs >> aJobFileName;
cout << "Incoming JobFileName : " << aJobFileName << endl;
// BUG 4645
string::size_type idx = aJobFileName.find_last_of('/');
if (idx > 0 && idx < string::npos)
{
string dirname(aJobFileName, 0, idx);
struct stat st;
if (stat(dirname.c_str(), &st) != 0)
{
cout << "Creating directory : " << dirname << endl;
boost::filesystem::create_directories(dirname.c_str());
}
}
fJobFile.open(aJobFileName.c_str());
}
//------------------------------------------------------------------------------
void WEDataLoader::onReceiveJobData(ByteStream& Ibs)
{
// Will write to the output file.
std::string aData;
Ibs >> aData;
fJobFile << aData;
fJobFile.close();
}
//------------------------------------------------------------------------------
void WEDataLoader::onReceiveErrFileRqst(ByteStream& Ibs)
{
std::string aErrFileName;
Ibs >> aErrFileName;
cout << "Error Filename Arrived " << aErrFileName << endl;
ByteStream obs;
obs << (ByteStream::byte)WE_CLT_SRV_ERRLOG;
obs << (ByteStream::byte)fPmId; // PM id
obs << aErrFileName;
BrmReportParser aErrFileParser;
bool aRet = aErrFileParser.serialize(aErrFileName, obs);
if (aRet)
{
boost::mutex::scoped_lock aLock(fClntMsgMutex);
updateTxBytes(obs.length());
try
{
fRef.fIos.write(obs);
}
catch (...)
{
cout << "Broken Pipe .." << endl;
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << " : Broken Pipe : socket write failed ";
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_INFO, logging::M0000);
}
}
aLock.unlock();
}
// delete the temp files
if (remove(aErrFileName.c_str()) != 0)
cout << "Failed in removing Error file: " << aErrFileName << endl;
}
//------------------------------------------------------------------------------
// Process the receipt of a msg containing the contents of a *.bad file.
//------------------------------------------------------------------------------
void WEDataLoader::onReceiveBadFileRqst(ByteStream& Ibs)
{
std::string aBadFileName;
Ibs >> aBadFileName;
cout << "Error Filename Arrived " << aBadFileName << endl;
ByteStream obs;
obs << (ByteStream::byte)WE_CLT_SRV_BADLOG;
obs << (ByteStream::byte)fPmId; // PM id
obs << aBadFileName;
BrmReportParser aBadFileParser;
bool aRet = aBadFileParser.serializeBlocks(aBadFileName, obs);
if (aRet)
{
boost::mutex::scoped_lock aLock(fClntMsgMutex);
updateTxBytes(obs.length());
try
{
fRef.fIos.write(obs);
}
catch (...)
{
cout << "Broken Pipe .." << endl;
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << " : Broken Pipe : socket write failed ";
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_INFO, logging::M0000);
}
}
aLock.unlock();
}
// delete the temp files
if (remove(aBadFileName.c_str()) != 0)
cout << "Failed in removing Error file: " << aBadFileName << endl;
}
//------------------------------------------------------------------------------
void WEDataLoader::sendDataRequest()
{
int aQsz = (fpCfThread) ? fpCfThread->getQueueSize() : 0;
// if(aQsz>MIN_QSIZE)
// Bug 5031 : If Q size goes above 100 (2*50); there is some thing wrong
// will put a warning in info log. Controlled in Cpimport init data rqst cnt
if (aQsz > MAX_QSIZE) // >250
{
cout << "WARNING : Data Queuing up : QSize = " << aQsz << endl;
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << "WARNING : Data Queuing up : QSize = " << aQsz;
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_INFO, logging::M0000);
}
}
boost::mutex::scoped_lock aLock(fClntMsgMutex);
ByteStream obs;
obs << (ByteStream::byte)WE_CLT_SRV_DATARQST;
obs << (ByteStream::byte)fPmId; // PM id
updateTxBytes(obs.length());
try
{
fRef.fIos.write(obs);
}
catch (...)
{
cout << "Broken Pipe .." << endl;
if (fpSysLog)
{
ostringstream oss;
oss << getObjId() << " : Broken Pipe : socket write failed ";
logging::Message::Args errMsgArgs;
errMsgArgs.add(oss.str());
fpSysLog->logMsg(errMsgArgs, logging::LOG_TYPE_INFO, logging::M0000);
}
}
aLock.unlock();
}
//------------------------------------------------------------------------------
void WEDataLoader::serialize(messageqcpp::ByteStream& b) const
{
// TODO to be changed. left it here to understand how to implement
/*
b << (ObjectReader::id_t) ObjectReader::SIMPLECOLUMN;
ReturnedColumn::serialize(b); // parent class serialize
b << (uint32_t) fOid;
b << fData;
b << static_cast<const ByteStream::doublebyte>(fReturnAll);
b << (uint32_t) fSequence;
*/
}
//-----------------------------------------------------------------------------
void WEDataLoader::unserialize(messageqcpp::ByteStream& b)
{
// TODO to be changed. left it here to understand how to implement
/*
ObjectReader::checkType(b, ObjectReader::SIMPLECOLUMN);
ReturnedColumn::unserialize(b); // parent class unserialize
b >> (uint32_t&) fOid;
b >> fData;
b >> reinterpret_cast<ByteStream::doublebyte&>(fReturnAll);
b >> (uint32_t&) fSequence;
*/
}
//-----------------------------------------------------------------------------
} // namespace WriteEngine