database/mariadb-columnstore-engine
1
0
Fork 0
mirror of https://github.com/mariadb-corporation/mariadb-columnstore-engine.git synced 2025-07-29 08:21:15 +03:00
Code Activity

clang format apply

Browse Source
This commit is contained in:
Leonid Fedorov
2022-01-21 16:43:49 +00:00
parent 6b6411229f
commit 04752ec546
1376 changed files with 393460 additions and 412662 deletions
Download Patch File Download Diff File Expand all files Collapse all files

801
writeengine/client/we_clients.cpp
View File

@ -62,500 +62,487 @@ using namespace WriteEngine;
namespace
{
void writeToLog(const char* file, int line, const string& msg, LOG_TYPE logto = LOG_TYPE_INFO)
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);
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;
switch (logto)
{
case LOG_TYPE_DEBUG: ml.logDebugMessage(m); break;
case LOG_TYPE_INFO:
ml.logInfoMessage(m);
break;
case LOG_TYPE_INFO: ml.logInfoMessage(m); break;
case LOG_TYPE_WARNING:
ml.logWarningMessage(m);
break;
case LOG_TYPE_WARNING: ml.logWarningMessage(m); break;
case LOG_TYPE_ERROR:
ml.logErrorMessage(m);
break;
case LOG_TYPE_ERROR: ml.logErrorMessage(m); break;
case LOG_TYPE_CRITICAL:
ml.logCriticalMessage(m);
break;
}
case LOG_TYPE_CRITICAL: ml.logCriticalMessage(m); break;
}
}
string getModuleNameByIPAddr(oam::ModuleTypeConfig moduletypeconfig,
string ipAddress)
string getModuleNameByIPAddr(oam::ModuleTypeConfig moduletypeconfig, string ipAddress)
{
string modulename = "pm1";
DeviceNetworkList::iterator pt = moduletypeconfig.ModuleNetworkList.begin();
string modulename = "pm1";
DeviceNetworkList::iterator pt = moduletypeconfig.ModuleNetworkList.begin();
for ( ; pt != moduletypeconfig.ModuleNetworkList.end() ; pt++)
for (; pt != moduletypeconfig.ModuleNetworkList.end(); pt++)
{
modulename = (*pt).DeviceName;
HostConfigList::iterator pt1 = (*pt).hostConfigList.begin();
for (; pt1 != (*pt).hostConfigList.end(); pt1++)
{
modulename = (*pt).DeviceName;
HostConfigList::iterator pt1 = (*pt).hostConfigList.begin();
for ( ; pt1 != (*pt).hostConfigList.end() ; pt1++)
{
if (ipAddress == (*pt1).IPAddr)
return modulename;
}
if (ipAddress == (*pt1).IPAddr)
return modulename;
}
}
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()()
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
{
//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;
}
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();
}
void operator()(T& x)
{
x.shutdown();
}
};
}
} // namespace
namespace WriteEngine
{
WEClients::WEClients(int PrgmID) :
fPrgmID(PrgmID),
pmCount(0)
WEClients::WEClients(int PrgmID) : fPrgmID(PrgmID), pmCount(0)
{
closingConnection = 0;
Setup();
closingConnection = 0;
Setup();
}
WEClients::~WEClients()
{
Close();
Close();
}
void WEClients::Setup()
{
makeBusy(true);
joblist::ResourceManager* rm = joblist::ResourceManager::instance();
oam::Oam oam;
string ipAddress;
ModuleTypeConfig moduletypeconfig;
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
{
oam.getSystemConfig("pm", moduletypeconfig);
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__, "oam.getSystemConfig error, unknown exception", LOG_TYPE_ERROR);
throw runtime_error("Setup failed");
writeToLog(__FILE__, __LINE__, "Could not connect to " + fServer, LOG_TYPE_ERROR);
}
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);
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();
}
// 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;
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)
void WEClients::Listen(boost::shared_ptr<MessageQueueClient> client, uint32_t connIndex)
{
SBS sbs;
SBS sbs;
try
try
{
while (Busy())
{
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)
{
//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;
}
return;
}
catch (std::exception& e)
{
cerr << "WEC Caught EXCEPTION: " << e.what() << endl;
goto Error;
}
catch (...)
{
cerr << "WEC Caught UNKNOWN EXCEPT" << endl;
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.
boost::mutex::scoped_lock lk(fMlock);
// error condition! push 0 length bs to messagequeuemap and
// eventually let jobstep error out.
boost::mutex::scoped_lock lk(fMlock);
MessageQueueMap::iterator map_tok;
sbs.reset(new ByteStream(0));
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
{
boost::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++;
}
}
return;
}
void WEClients::addQueue(uint32_t key)
{
bool b;
boost::mutex* lock = new boost::mutex();
condition* cond = new condition();
boost::shared_ptr<MQE> mqe(new MQE(pmCount));
mqe->queue = WESMsgQueue(lock, cond);
boost::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)
{
boost::mutex::scoped_lock lk(fMlock);
MessageQueueMap::iterator map_tok = fSessionMessages.find(key);
if (map_tok == fSessionMessages.end())
return;
map_tok->second->queue.shutdown();
for (map_tok = fSessionMessages.begin(); map_tok != fSessionMessages.end(); ++map_tok)
{
map_tok->second->queue.clear();
fSessionMessages.erase(map_tok);
}
(void)atomicops::atomicInc(&map_tok->second->unackedWork[0]);
map_tok->second->queue.push(sbs);
}
void WEClients::shutdownQueue(uint32_t key)
{
boost::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
boost::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.");
}
lk.unlock();
// reset the pmconnection map
{
boost::mutex::scoped_lock onErrLock(fOnErrMutex);
string moduleName = client->moduleName();
ClientList::iterator itor = fPmConnections.begin();
while (itor != fPmConnections.end())
{
if (itor->second != NULL)
{
itor->second->write(msg);
}
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++;
itor++;
}
}
return;
}
void WEClients::addQueue(uint32_t key)
{
bool b;
boost::mutex* lock = new boost::mutex();
condition* cond = new condition();
boost::shared_ptr<MQE> mqe(new MQE(pmCount));
mqe->queue = WESMsgQueue(lock, cond);
boost::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)
{
boost::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)
{
boost::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
boost::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);
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;
// 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;
boost::mutex::scoped_lock lk(fMlock);
MessageQueueMap::iterator map_tok = fSessionMessages.find(uniqueId);
boost::mutex::scoped_lock lk(fMlock);
MessageQueueMap::iterator map_tok = fSessionMessages.find(uniqueId);
if (map_tok == fSessionMessages.end())
{
return;
}
if (map_tok == fSessionMessages.end())
{
return;
}
mqe = map_tok->second;
lk.unlock();
mqe = map_tok->second;
lk.unlock();
if (pmCount > 0)
{
atomicops::atomicInc(&mqe->unackedWork[connIndex % pmCount]);
}
(void)mqe->queue.push(sbs);
if (pmCount > 0)
{
atomicops::atomicInc(&mqe->unackedWork[connIndex % pmCount]);
}
(void)mqe->queue.push(sbs);
}
}
} // namespace WriteEngine
// vim:ts=4 sw=4:

227
writeengine/client/we_clients.h
View File

@ -46,128 +46,137 @@ namespace WriteEngine
{
class WEClients
{
public:
/**
* Constructors
*/
EXPORT WEClients(int PrgmID);
EXPORT ~WEClients();
public:
/**
* Constructors
*/
EXPORT WEClients(int PrgmID);
EXPORT ~WEClients();
//static boost::mutex map_mutex;
EXPORT void addQueue(uint32_t key);
EXPORT void removeQueue(uint32_t key);
EXPORT void shutdownQueue(uint32_t key);
// static boost::mutex map_mutex;
EXPORT void addQueue(uint32_t key);
EXPORT void removeQueue(uint32_t key);
EXPORT void shutdownQueue(uint32_t key);
/** @brief read a Write Engine Server response
*
* Returns the next message in the inbound queue for unique ids.
* @param bs A pointer to the ByteStream to fill in.
* @note: saves a copy vs read(uint32_t, uint32_t).
*/
EXPORT void read(uint32_t key, messageqcpp::SBS&);
/** @brief read a Write Engine Server response
*
* Returns the next message in the inbound queue for unique ids.
* @param bs A pointer to the ByteStream to fill in.
* @note: saves a copy vs read(uint32_t, uint32_t).
*/
EXPORT void read(uint32_t key, messageqcpp::SBS&);
/** @brief write function to write to specified PM
*/
EXPORT void write(const messageqcpp::ByteStream& msg, uint32_t connection);
/** @brief write function to write to specified PM
*/
EXPORT void write(const messageqcpp::ByteStream& msg, uint32_t connection);
/** @brief write function to write to all PMs
*/
EXPORT void write_to_all(const messageqcpp::ByteStream& msg);
/** @brief write function to write to all PMs
*/
EXPORT void write_to_all(const messageqcpp::ByteStream& msg);
/** @brief Shutdown this object
*
* Closes all the connections created during Setup() and cleans up other stuff.
*/
EXPORT int Close();
/** @brief Shutdown this object
*
* Closes all the connections created during Setup() and cleans up other stuff.
*/
EXPORT int Close();
/** @brief Start listening for Write Engine Server responses
*
* Starts the current thread listening on the client socket for Write Engine Server response messages. Will not return
* until busy() returns false or a zero-length response is received.
*/
EXPORT void Listen(boost::shared_ptr<messageqcpp::MessageQueueClient> client, uint32_t connIndex);
/** @brief Start listening for Write Engine Server responses
*
* Starts the current thread listening on the client socket for Write Engine Server response messages. Will
* not return until busy() returns false or a zero-length response is received.
*/
EXPORT void Listen(boost::shared_ptr<messageqcpp::MessageQueueClient> client, uint32_t connIndex);
/** @brief set/unset busy flag
*
* Set or unset the busy flag so Listen() can return.
*/
EXPORT void makeBusy(bool b)
/** @brief set/unset busy flag
*
* Set or unset the busy flag so Listen() can return.
*/
EXPORT void makeBusy(bool b)
{
fBusy = b;
}
/** @brief fBusy accessor
*
*/
EXPORT bool Busy() const
{
return fBusy;
}
EXPORT void Setup();
uint64_t connectedWEServers() const
{
return fPmConnections.size();
}
/** @brief accessor
*/
uint32_t getPmCount()
{
return pmCount;
}
private:
WEClients(const WEClients& weClient);
WEClients& operator=(const WEClients& weClient);
typedef std::vector<boost::thread*> ReaderList;
typedef std::map<unsigned, boost::shared_ptr<messageqcpp::MessageQueueClient> > ClientList;
// A queue of ByteStreams coming in from Write Engine Server
typedef joblist::ThreadSafeQueue<messageqcpp::SBS> WESMsgQueue;
/* To keep some state associated with the connection */
struct MQE
{
MQE(uint32_t pCount) : ackSocketIndex(0), pmCount(pCount)
{
fBusy = b;
unackedWork.reset(new volatile uint32_t[pmCount]);
memset((void*)unackedWork.get(), 0, pmCount * sizeof(uint32_t));
}
/** @brief fBusy accessor
*
*/
EXPORT bool Busy() const
{
return fBusy;
}
EXPORT void Setup();
uint64_t connectedWEServers() const
{
return fPmConnections.size();
}
/** @brief accessor
*/
uint32_t getPmCount()
{
return pmCount;
}
private:
WEClients(const WEClients& weClient);
WEClients& operator=(const WEClients& weClient);
typedef std::vector<boost::thread*> ReaderList;
typedef std::map<unsigned, boost::shared_ptr<messageqcpp::MessageQueueClient> > ClientList;
//A queue of ByteStreams coming in from Write Engine Server
typedef joblist::ThreadSafeQueue<messageqcpp::SBS> WESMsgQueue;
/* To keep some state associated with the connection */
struct MQE
{
MQE(uint32_t pCount) : ackSocketIndex(0), pmCount(pCount)
{
unackedWork.reset(new volatile uint32_t[pmCount]);
memset((void*) unackedWork.get(), 0, pmCount * sizeof(uint32_t));
}
WESMsgQueue queue;
uint32_t ackSocketIndex;
boost::scoped_array<volatile uint32_t> unackedWork;
uint32_t pmCount;
};
//The mapping of session ids to StepMsgQueueLists
typedef std::map<unsigned, boost::shared_ptr<MQE> > MessageQueueMap;
void StartClientListener(boost::shared_ptr<messageqcpp::MessageQueueClient> cl, uint32_t connIndex);
/** @brief Add a message to the queue
*
*/
void addDataToOutput(messageqcpp::SBS, uint32_t connIndex);
int fPrgmID;
ClientList fPmConnections; // all the Write Engine servers
ReaderList fWESReader; // all the reader threads for the pm servers
MessageQueueMap fSessionMessages; // place to put messages from the pm server to be returned by the Read method
boost::mutex fMlock; //sessionMessages mutex
std::vector<boost::shared_ptr<boost::mutex> > fWlock; //WES socket write mutexes
bool fBusy;
volatile uint32_t closingConnection;
WESMsgQueue queue;
uint32_t ackSocketIndex;
boost::scoped_array<volatile uint32_t> unackedWork;
uint32_t pmCount;
boost::mutex fOnErrMutex; // to lock function scope to reset pmconnections under error condition
};
boost::mutex ackLock;
public:
enum {DDLPROC = 0, SPLITTER, DMLPROC, BATCHINSERTPROC};
// The mapping of session ids to StepMsgQueueLists
typedef std::map<unsigned, boost::shared_ptr<MQE> > MessageQueueMap;
void StartClientListener(boost::shared_ptr<messageqcpp::MessageQueueClient> cl, uint32_t connIndex);
/** @brief Add a message to the queue
*
*/
void addDataToOutput(messageqcpp::SBS, uint32_t connIndex);
int fPrgmID;
ClientList fPmConnections; // all the Write Engine servers
ReaderList fWESReader; // all the reader threads for the pm servers
MessageQueueMap
fSessionMessages; // place to put messages from the pm server to be returned by the Read method
boost::mutex fMlock; // sessionMessages mutex
std::vector<boost::shared_ptr<boost::mutex> > fWlock; // WES socket write mutexes
bool fBusy;
volatile uint32_t closingConnection;
uint32_t pmCount;
boost::mutex fOnErrMutex; // to lock function scope to reset pmconnections under error condition
boost::mutex ackLock;
public:
enum
{
DDLPROC = 0,
SPLITTER,
DMLPROC,
BATCHINSERTPROC
};
};
}
} // namespace WriteEngine
#undef EXPORT

87
writeengine/client/we_ddlcommandclient.cpp
View File

@ -36,63 +36,62 @@ namespace WriteEngine
{
WE_DDLCommandClient::WE_DDLCommandClient()
{
fWEClient = new WEClients(WEClients::DDLPROC);
fWEClient = new WEClients(WEClients::DDLPROC);
}
WE_DDLCommandClient::~WE_DDLCommandClient()
{
delete fWEClient;
fWEClient = NULL;
delete fWEClient;
fWEClient = NULL;
}
uint8_t WE_DDLCommandClient::UpdateSyscolumnNextval(uint32_t columnOid, uint64_t nextVal, uint32_t sessionID)
{
ByteStream command, response;
uint8_t err = 0;
uint64_t uniqueId = fDbrm.getUnique64();
fWEClient->addQueue(uniqueId);
command << (ByteStream::byte)WE_UPDATE_NEXTVAL;
command << uniqueId;
command << columnOid;
command << nextVal;
command << sessionID;
uint16_t dbRoot;
BRM::OID_t oid = 1021;
fDbrm.getSysCatDBRoot(oid, dbRoot);
int pmNum = 1;
boost::shared_ptr<messageqcpp::ByteStream> bsIn;
ByteStream command, response;
uint8_t err = 0;
uint64_t uniqueId = fDbrm.getUnique64();
fWEClient->addQueue(uniqueId);
command << (ByteStream::byte)WE_UPDATE_NEXTVAL;
command << uniqueId;
command << columnOid;
command << nextVal;
command << sessionID;
uint16_t dbRoot;
BRM::OID_t oid = 1021;
fDbrm.getSysCatDBRoot(oid, dbRoot);
int pmNum = 1;
boost::shared_ptr<messageqcpp::ByteStream> bsIn;
try
try
{
fOam.getDbrootPmConfig(dbRoot, pmNum);
fWEClient->write(command, pmNum);
while (1)
{
fOam.getDbrootPmConfig (dbRoot, pmNum);
fWEClient->write(command, pmNum);
bsIn.reset(new ByteStream());
fWEClient->read(uniqueId, bsIn);
while (1)
{
bsIn.reset(new ByteStream());
fWEClient->read(uniqueId, bsIn);
if ( bsIn->length() == 0 ) //read error
{
err = 1;
break;
}
else
{
*bsIn >> err;
break;
}
}
}
catch (...)
{
if (bsIn->length() == 0) // read error
{
err = 1;
break;
}
else
{
*bsIn >> err;
break;
}
}
}
catch (...)
{
err = 1;
}
fWEClient->removeQueue(uniqueId);
return err;
}
fWEClient->removeQueue(uniqueId);
return err;
}
} // namespace WriteEngine

34
writeengine/client/we_ddlcommandclient.h
View File

@ -39,26 +39,24 @@ namespace WriteEngine
{
class WE_DDLCommandClient
{
public:
EXPORT WE_DDLCommandClient();
EXPORT ~WE_DDLCommandClient();
/** @brief Update SYSCOLUMN nextval column for the columnoid with nextVal.
*
* Update SYSCOLUMN nextval column for the columnoid with nexValue.
* @param columnOid (in) The column OID
* @param nextVal (in) The partition number
* @return 0 on success, non-0 on error.
*/
EXPORT uint8_t UpdateSyscolumnNextval(uint32_t columnOid, uint64_t nextVal, uint32_t sessionID = 0);
private:
BRM::DBRM fDbrm;
WEClients* fWEClient;
oam::Oam fOam;
public:
EXPORT WE_DDLCommandClient();
EXPORT ~WE_DDLCommandClient();
/** @brief Update SYSCOLUMN nextval column for the columnoid with nextVal.
*
* Update SYSCOLUMN nextval column for the columnoid with nexValue.
* @param columnOid (in) The column OID
* @param nextVal (in) The partition number
* @return 0 on success, non-0 on error.
*/
EXPORT uint8_t UpdateSyscolumnNextval(uint32_t columnOid, uint64_t nextVal, uint32_t sessionID = 0);
private:
BRM::DBRM fDbrm;
WEClients* fWEClient;
oam::Oam fOam;
};
}
} // namespace WriteEngine
#undef EXPORT

10
writeengine/client/we_dmlcommandclient.cpp
View File

@ -29,21 +29,19 @@ using namespace messageqcpp;
using namespace dataconvert;
using namespace dmlpackage;
#include "we_dmlcommandclient.h"
namespace WriteEngine
{
WE_DMLCommandClient::WE_DMLCommandClient()
{
fWEClient = new WEClients(WEClients::DDLPROC);
fWEClient = new WEClients(WEClients::DDLPROC);
}
WE_DMLCommandClient::~WE_DMLCommandClient()
{
delete fWEClient;
fWEClient = NULL;
}
delete fWEClient;
fWEClient = NULL;
}
} // namespace WriteEngine

18
writeengine/client/we_dmlcommandclient.h
View File

@ -39,18 +39,16 @@ namespace WriteEngine
{
class WE_DMLCommandClient
{
public:
EXPORT WE_DMLCommandClient();
EXPORT ~WE_DMLCommandClient();
private:
BRM::DBRM fDbrm;
WEClients* fWEClient;
oam::Oam fOam;
public:
EXPORT WE_DMLCommandClient();
EXPORT ~WE_DMLCommandClient();
private:
BRM::DBRM fDbrm;
WEClients* fWEClient;
oam::Oam fOam;
};
}
} // namespace WriteEngine
#undef EXPORT