database/mariadb-columnstore-engine
1
0
Fork 0
mirror of https://github.com/mariadb-corporation/mariadb-columnstore-engine.git synced 2025-04-18 21:44:02 +03:00
Code
mariadb-columnstore-engine/dmlproc/dmlprocessor.cpp
Denis Khalikov 254ced25cf Review fixes
2024-05-23 00:01:34 +04:00

2025 lines
72 KiB
C++
Raw Permalink Blame History

/* 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: dmlprocessor.cpp 1024 2013-07-26 16:23:59Z chao $
*
*
***********************************************************************/
/** @file */
#include "configcpp.h"
#include <signal.h>
#include <ctime>
//#define SERIALIZE_DDL_DML_CPIMPORT 1
#include <boost/thread/mutex.hpp>
#include <boost/scoped_ptr.hpp>
#include <boost/scoped_array.hpp>
#include <boost/shared_ptr.hpp>
using namespace boost;
#include "cacheutils.h"
#include "vss.h"
#include "dbrm.h"
#include "brmtypes.h"
#include "idberrorinfo.h"
#include "errorids.h"
#include "batchinsertprocessor.h"
#include "tablelockdata.h"
#include "oamcache.h"
#include "messagelog.h"
#include "sqllogger.h"
#include "we_messages.h"
#include "dmlprocessor.h"
using namespace BRM;
using namespace config;
using namespace execplan;
using namespace std;
using namespace messageqcpp;
using namespace dmlpackage;
using namespace dmlpackageprocessor;
using namespace joblist;
using namespace logging;
using namespace oam;
using namespace WriteEngine;
#include "querytele.h"
using namespace querytele;
extern boost::mutex mute;
extern boost::condition_variable cond;
#define MCOL_140 // Undefine to test VSS for out of order transactions
namespace
{
[[maybe_unused]] const std::string myname = "DMLProc";
}
namespace dmlprocessor
{
// Map to store the package handler objects so we can set flags during execution
// for things like ctrl+c
DMLProcessor::PackageHandlerMap_t DMLProcessor::packageHandlerMap;
boost::mutex DMLProcessor::packageHandlerMapLock;
// Map to store the BatchInsertProc object
std::map<uint32_t, BatchInsertProc*> DMLProcessor::batchinsertProcessorMap;
boost::mutex DMLProcessor::batchinsertProcessorMapLock;
// MCOL-140 Map to hold table oids for tables being changed.
std::map<uint32_t, PackageHandler::tableAccessQueue_t> PackageHandler::tableOidMap;
boost::condition_variable PackageHandler::tableOidCond;
boost::mutex PackageHandler::tableOidMutex;
//------------------------------------------------------------------------------
// A thread to periodically call dbrm to see if a user is
// shutting down the system or has put the system into write
// suspend mode. DBRM has 2 flags to check in this case, the
// ROLLBACK flag, and the FORCE flag. These flags will be
// reported when we ask for the Shutdown Pending flag (which we
// ignore at this point). Even if the user is putting the system
// into write suspend mode, this call will return the flags we
// are interested in. If ROLLBACK is set, we cancel normally.
// If FORCE is set, we can't rollback.
struct CancellationThread
{
CancellationThread(DBRM* aDbrm, DMLServer& aServer) : fDbrm(aDbrm), fServer(aServer)
{
}
void operator()()
{
bool bDoingRollback = false;
bool bRollback = false;
bool bForce = false;
ostringstream oss;
std::vector<BRM::TableLockInfo> tableLocks;
BRM::TxnID txnId;
DMLProcessor::PackageHandlerMap_t::iterator phIter;
uint32_t sessionID;
int rc = 0;
while (true)
{
usleep(1000000); // 1 seconds
// Check to see if someone has ordered a shutdown or suspend with rollback.
(void)fDbrm->getSystemShutdownPending(bRollback, bForce);
if (bForce)
break;
if (bDoingRollback && bRollback)
{
continue;
// We've already started the rollbacks. Don't start again.
}
bDoingRollback = false;
if (bRollback)
{
RollbackTransactionProcessor rollbackProcessor(fDbrm);
SessionManager sessionManager;
uint64_t uniqueId = fDbrm->getUnique64();
std::string errorMsg;
int activeTransCount = 0;
int idleTransCount = 0;
bDoingRollback = true;
ostringstream oss;
oss << "DMLProc has been told to rollback all DML transactions.";
DMLProcessor::log(oss.str(), logging::LOG_TYPE_INFO);
// Tell any active processors to stop working and return an error
// The front end will respond with a ROLLBACK command.
// Mark all active processors to rollback
boost::mutex::scoped_lock lk2(DMLProcessor::packageHandlerMapLock);
for (phIter = DMLProcessor::packageHandlerMap.begin();
phIter != DMLProcessor::packageHandlerMap.end(); ++phIter)
{
ostringstream oss;
oss << "DMLProc will rollback active session " << phIter->second->getSessionID() << " Transaction "
<< phIter->second->getTxnid();
DMLProcessor::log(oss.str(), logging::LOG_TYPE_INFO);
++activeTransCount;
phIter->second->rollbackPending();
}
if (activeTransCount > 0)
{
ostringstream oss1;
oss1 << "DMLProc is rolling back back " << activeTransCount << " active transactions.";
DMLProcessor::log(oss1.str(), logging::LOG_TYPE_INFO);
}
// WIP Need to set cluster to read-only via CMAPI before shutting the cluster down.
if (fDbrm->isReadWrite())
{
continue;
}
// Check for any open DML transactions that don't currently have
// a processor
tableLocks = fDbrm->getAllTableLocks();
if (tableLocks.size() > 0)
{
for (uint32_t i = 0; i < tableLocks.size(); ++i)
{
sessionID = tableLocks[i].ownerSessionID;
phIter = DMLProcessor::packageHandlerMap.find(sessionID);
if (phIter == DMLProcessor::packageHandlerMap.end())
{
// We have found an active transaction without a packagehandler.
// This means that a transaction is open with autocommit turned
// off, but there's no current activity on the transaction. We
// need to roll it back if it's a DML transaction.
// If ownerName == "DMLProc" then it's a DML transaction.
if (tableLocks[i].ownerName == "DMLProc")
{
// OK, we know this is an idle DML transaction, so roll it back.
++idleTransCount;
txnId.id = tableLocks[i].ownerTxnID;
txnId.valid = true;
rc = rollbackProcessor.rollBackTransaction(uniqueId, txnId, sessionID, errorMsg);
if (rc == 0)
{
fDbrm->invalidateUncommittedExtentLBIDs(txnId.id, false);
//@Bug 4524. In case it is batchinsert, call bulkrollback.
rc = rollbackProcessor.rollBackBatchAutoOnTransaction(uniqueId, txnId, sessionID,
tableLocks[i].tableOID, errorMsg);
if (rc == 0)
{
logging::logCommand(0, tableLocks[i].ownerTxnID, "ROLLBACK;");
bool lockReleased = true;
try
{
lockReleased = fDbrm->releaseTableLock(tableLocks[i].id);
TablelockData::removeTablelockData(sessionID);
}
catch (std::exception&)
{
throw std::runtime_error(IDBErrorInfo::instance()->errorMsg(ERR_HARD_FAILURE));
}
if (lockReleased)
{
sessionManager.rolledback(txnId);
ostringstream oss;
oss << "DMLProc rolled back idle transaction " << tableLocks[i].ownerTxnID
<< " and table lock id " << tableLocks[i].id << " is released.";
DMLProcessor::log(oss.str(), logging::LOG_TYPE_INFO);
}
else
{
ostringstream oss;
oss << "DMLProc rolled back idle transaction " << tableLocks[i].ownerTxnID
<< " and tble lock id " << tableLocks[i].id << " is not released.";
DMLProcessor::log(oss.str(), logging::LOG_TYPE_INFO);
}
}
else
{
ostringstream oss;
oss << " problem with bulk rollback of idle transaction " << tableLocks[i].ownerTxnID
<< "and DBRM is setting to readonly and table lock is not released: " << errorMsg;
DMLProcessor::log(oss.str(), logging::LOG_TYPE_CRITICAL);
rc = fDbrm->setReadOnly(true);
}
}
else
{
ostringstream oss;
oss << " problem with rollback of idle transaction " << tableLocks[i].ownerTxnID
<< "and DBRM is setting to readonly and table lock is not released: " << errorMsg;
DMLProcessor::log(oss.str(), logging::LOG_TYPE_CRITICAL);
rc = fDbrm->setReadOnly(true);
}
}
}
}
}
// If there are any abandonded transactions without locks
// release them.
int len;
std::shared_ptr<BRM::SIDTIDEntry[]> activeTxns = sessionManager.SIDTIDMap(len);
for (int i = 0; i < len; i++)
{
// If there isn't a table lock for this transaction, roll it back. Otherwise, assume
// it has an active processor or is not DML initiated and leave it alone. It's someone
// else's concern.
bool bFoundit = false;
for (uint32_t j = 0; j < tableLocks.size(); ++j)
{
if (tableLocks[j].ownerTxnID == activeTxns[i].txnid.id)
{
bFoundit = true;
break;
}
}
if (!bFoundit && activeTxns[i].txnid.valid)
{
rollbackProcessor.rollBackTransaction(uniqueId, activeTxns[i].txnid, activeTxns[i].sessionid,
errorMsg);
sessionManager.rolledback(activeTxns[i].txnid);
++idleTransCount;
ostringstream oss;
oss << "DMLProc rolled back idle transaction with no tablelock" << tableLocks[i].ownerTxnID;
DMLProcessor::log(oss.str(), logging::LOG_TYPE_INFO);
}
}
if (idleTransCount > 0)
{
ostringstream oss2;
oss2 << "DMLProc has rolled back " << idleTransCount << " idle transactions.";
DMLProcessor::log(oss2.str(), logging::LOG_TYPE_INFO);
}
// Here is the end of the rollback if so DMLProc rollbacks what it can.
break;
}
}
// Setting the flag to tell DMLServer to exit.
fServer.startShutdown();
}
DBRM* fDbrm;
DMLServer& fServer;
};
PackageHandler::PackageHandler(const messageqcpp::IOSocket& ios,
boost::shared_ptr<messageqcpp::ByteStream> bs, uint8_t packageType,
joblist::DistributedEngineComm* ec, bool concurrentSupport,
uint64_t maxDeleteRows, uint32_t sessionID,
execplan::CalpontSystemCatalog::SCN txnId, DBRM* aDbrm,
const QueryTeleClient& qtc,
boost::shared_ptr<execplan::CalpontSystemCatalog> csc)
: fIos(ios)
, fByteStream(bs)
, fPackageType(packageType)
, fEC(ec)
, fConcurrentSupport(concurrentSupport)
, fMaxDeleteRows(maxDeleteRows)
, fSessionID(sessionID)
, fTableOid(0)
, fTxnid(txnId)
, fDbrm(aDbrm)
, fQtc(qtc)
, fcsc(csc)
{
}
PackageHandler::~PackageHandler()
{
// cout << "In destructor" << endl;
}
// MCOL-140
// Blocks a thread if there is another trx working on the same fTableOid
// return 1 when thread should continue.
// return 0 if error. Right now, no error detection is implemented.
//
// txnid was being created before the call to this function. This caused race conditions
// so creation is delayed until we're inside the lock here. Nothing needs it before
// this point in the execution.
//
// The algorithm is this. When the first txn for a given fTableOid arrives, start a queue
// containing a list of waiting or working txnId. Put this txnId into the queue (working)
// Put the queue into a map keyed on fTableOid.
//
// When the next txn for this fTableOid arrives, it finds the queue in the map and adds itself,
// then waits for condition.
// When a thread finishes, it removes its txnId from the queue and notifies all. If the queue is
// empty, it removes the entry from the map.
// Upon wakeup from wait(), a thread checks to see if it's next in the queue. If so, it is released
// to do work. Otherwise it goes back to wait.
//
// There's a chance (CTRL+C) for instance, that the txn is no longer in the queue. Release it to work.
// Rollback will most likely be next.
//
// A tranasaction for one fTableOid is not blocked by a txn for a different fTableOid.
int PackageHandler::synchTableAccess(dmlpackage::CalpontDMLPackage* dmlPackage)
{
// MCOL-140 Wait for any other DML using this table.
std::map<uint32_t, PackageHandler::tableAccessQueue_t>::iterator it;
boost::unique_lock<boost::mutex> lock(tableOidMutex);
BRM::TxnID txnid;
if (fPackageType != dmlpackage::DML_COMMAND)
{
txnid = sessionManager.getTxnID(fSessionID);
if (!txnid.valid)
{
txnid = sessionManager.newTxnID(fSessionID, true);
if (!txnid.valid)
{
throw std::runtime_error(std::string("Unable to start a transaction. Check critical log."));
}
}
}
else
{
txnid = sessionManager.getTxnID(fSessionID);
}
fTxnid = txnid.id;
if ((it = tableOidMap.find(fTableOid)) != tableOidMap.end())
{
PackageHandler::tableAccessQueue_t& tableOidQueue = it->second;
// There's at least one working txn on this table. We may be the same txn.
if (fTxnid == tableOidQueue.front())
{
return 1; // We're next in line or the same as the last. Keep working
}
tableOidQueue.push(fTxnid); // Get on the waiting list.
// We need to wait
// tableOidQueue here is the queue holding the waitng transactions for this fTableOid
while (true)
{
// Log something that we're waiting
LoggingID logid(21, fSessionID, fTxnid);
logging::Message::Args args1;
logging::Message msg(1);
ostringstream oss;
oss << "Txn is waiting for" << tableOidQueue.front() << " " << dmlPackage->get_SQLStatement() << "; |"
<< dmlPackage->get_SchemaName() << "|";
args1.add(oss.str());
args1.add((uint64_t)fTableOid);
msg.format(args1);
logging::Logger logger(logid.fSubsysID);
logger.logMessage(LOG_TYPE_DEBUG, msg, logid);
tableOidCond.wait(lock);
// In case of CTRL+C, the tableOidQueue could be invalidated
if ((tableOidMap.find(fTableOid))->second != tableOidQueue)
{
break;
}
if (tableOidQueue.front() == fTxnid)
{
// We're up next. Let's go do stuff.
break;
}
if (tableOidQueue.empty())
{
// If we had been the last txn waiting and CTRL+C was hit, then the queue is empty now.
// Empty queues must be erased from the map.
tableOidMap.erase(fTableOid);
break;
}
// If we're not in the queue at all, then continue. CTRL+C was probably hit.
PackageHandler::tableAccessQueue_t::container_type::iterator c_it = tableOidQueue.find(fTxnid);
if (c_it == tableOidQueue.end())
{
break;
}
// We're still in the queue and not on top. Go back and wait some more.
}
}
else
{
// We're the first for this tableoid. Start a new queue.
tableAccessQueue_t tableOidQueue;
tableOidQueue.push(fTxnid);
tableOidMap[fTableOid] = tableOidQueue;
}
return 1;
}
// MCOL-140 Called when it's time to release the next thread for this tablOid
int PackageHandler::releaseTableAccess()
{
// take us out of the queue
std::map<uint32_t, PackageHandler::tableAccessQueue_t>::iterator it;
boost::lock_guard<boost::mutex> lock(tableOidMutex);
if (fTableOid == 0 || (it = tableOidMap.find(fTableOid)) == tableOidMap.end())
{
return 2; // For now, return codes are not used
}
PackageHandler::tableAccessQueue_t& tableOidQueue = it->second;
if (tableOidQueue.front() != fTxnid)
{
// This is a severe error. The front should be the working thread. If we're here,
// we're the working thread and should be front().
cout << fTxnid << " " << fTableOid << " We got to release and we're not on top " << tableOidQueue.front()
<< endl;
LoggingID logid(21, fSessionID, fTxnid);
logging::Message::Args args1;
logging::Message msg(1);
args1.add(
"ReleaseTableAccess: Txn being released is not the current txn in the tablOidQueue for tableid");
args1.add((uint64_t)fTableOid);
msg.format(args1);
logging::Logger logger(logid.fSubsysID);
logger.logMessage(LOG_TYPE_ERROR, msg, logid);
}
else
{
if (!tableOidQueue.empty())
tableOidQueue.pop(); // Get off the waiting list.
if (tableOidQueue.empty())
{
// remove the queue from the map.
tableOidMap.erase(fTableOid);
}
}
// release the condition
tableOidCond.notify_all();
return 1;
}
int PackageHandler::forceReleaseTableAccess()
{
// By removing the txnid from the queue, the logic after the wait in
// synchTableAccess() will release the thread and clean up if needed.
std::map<uint32_t, PackageHandler::tableAccessQueue_t>::iterator it;
boost::lock_guard<boost::mutex> lock(tableOidMutex);
if (fTableOid == 0 || (it = tableOidMap.find(fTableOid)) == tableOidMap.end())
{
// This will happen for DML_COMMAND, as we never got the tableoid or called synchTableAccess
return 2;
}
PackageHandler::tableAccessQueue_t& tableOidQueue = it->second;
tableOidQueue.erase(fTxnid);
if (tableOidQueue.empty())
{
// remove the queue from the map.
tableOidMap.erase(fTableOid);
}
// release the condition
tableOidCond.notify_all();
return 1;
}
// static
// Called upon sighup, often because PrimProc crashed. We don't want to leave all the transactions hung,
// though some may be because they never returned from PrimProc and will leave the table lock on.
int PackageHandler::clearTableAccess()
{
tableOidMap.clear();
return 1;
}
CalpontSystemCatalog::ROPair PackageHandler::getTableRID(
boost::shared_ptr<execplan::CalpontSystemCatalog> fcsc,
execplan::CalpontSystemCatalog::TableName& tableName)
{
execplan::CalpontSystemCatalog::ROPair roPair;
try
{
roPair = fcsc->tableRID(tableName);
}
catch (...)
{
if (setupDec())
throw;
roPair = fcsc->tableRID(tableName);
}
return roPair;
}
void PackageHandler::run()
{
ResourceManager* frm = ResourceManager::instance();
dmlpackageprocessor::DMLPackageProcessor::DMLResult result;
result.result = dmlpackageprocessor::DMLPackageProcessor::NO_ERROR;
// cout << "PackageHandler handling ";
std::string stmt;
unsigned DMLLoggingId = 21;
oam::OamCache* oamCache = oam::OamCache::makeOamCache();
SynchTable synchTable;
try
{
switch (fPackageType)
{
case dmlpackage::DML_INSERT:
{
// build an InsertDMLPackage from the bytestream
// cout << "an INSERT package" << endl;
dmlpackage::InsertDMLPackage insertPkg;
// boost::shared_ptr<messageqcpp::ByteStream> insertBs (new messageqcpp::ByteStream);
messageqcpp::ByteStream bsSave = *(fByteStream.get());
insertPkg.readMetaData(*(fByteStream.get()));
#ifdef MCOL_140
if (fConcurrentSupport)
{
fTableOid = insertPkg.getTableOid();
// Single Insert has no start like bulk does, so insertPkg.getTableOid()
// isn't set. Go get it now.
if (fTableOid == 0)
{
CalpontSystemCatalog::TableName tableName;
tableName.schema = insertPkg.get_Table()->get_SchemaName();
tableName.table = insertPkg.get_Table()->get_TableName();
CalpontSystemCatalog::ROPair roPair = getTableRID(fcsc, tableName);
fTableOid = roPair.objnum;
}
synchTable.setPackage(this, &insertPkg); // Blocks if another DML thread is using this fTableOid
}
#endif
QueryTeleStats qts;
qts.query_uuid = QueryTeleClient::genUUID();
qts.msg_type = QueryTeleStats::QT_START;
qts.start_time = QueryTeleClient::timeNowms();
qts.session_id = fSessionID;
qts.query_type = "INSERT";
qts.query = insertPkg.get_SQLStatement();
qts.system_name = oamCache->getSystemName();
qts.module_name = oamCache->getModuleName();
qts.schema_name = insertPkg.get_SchemaName();
fQtc.postQueryTele(qts);
// cout << "This is batch insert " << insertPkg->get_isBatchInsert() << endl;
if (insertPkg.get_isBatchInsert())
{
fByteStream->reset();
// cout << "This is batch insert " << endl;
BatchInsertProc* batchProcessor = NULL;
{
boost::mutex::scoped_lock lk(DMLProcessor::batchinsertProcessorMapLock);
std::map<uint32_t, BatchInsertProc*>::iterator batchIter =
DMLProcessor::batchinsertProcessorMap.find(fSessionID);
if (batchIter == DMLProcessor::batchinsertProcessorMap.end())
{
batchProcessor =
new BatchInsertProc(insertPkg.get_isAutocommitOn(), insertPkg.getTableOid(), fTxnid, fDbrm);
DMLProcessor::batchinsertProcessorMap[fSessionID] = batchProcessor;
// cout << "batchProcessor is created " << batchProcessor << endl;
}
else
{
batchProcessor = batchIter->second;
// cout << "Found batchProcessor " << batchProcessor << endl;
}
}
if (insertPkg.get_Logging())
{
LoggingID logid(DMLLoggingId, insertPkg.get_SessionID(), fTxnid);
logging::Message::Args args1;
logging::Message msg(1);
args1.add("Start SQL statement: ");
if (!insertPkg.get_isCacheInsert())
{
ostringstream oss;
oss << insertPkg.get_SQLStatement() << "; |" << insertPkg.get_SchemaName() << "|";
args1.add(oss.str());
}
msg.format(args1);
logging::Logger logger(logid.fSubsysID);
logger.logMessage(LOG_TYPE_DEBUG, msg, logid);
TablelockData* tablelockData = TablelockData::makeTablelockData(insertPkg.get_SessionID());
uint64_t tableLockId = tablelockData->getTablelockId(insertPkg.getTableOid());
// cout << "Processing table oid " << insertPkg.getTableOid() << " for transaction "<< (int)fTxnid
// << endl;
if (tableLockId == 0)
{
// cout << "Grabing tablelock for batchProcessor " << batchProcessor << endl;
tableLockId = batchProcessor->grabTableLock(insertPkg.get_SessionID());
if (tableLockId == 0)
{
BRM::TxnID brmTxnID;
brmTxnID.id = fTxnid;
brmTxnID.valid = true;
sessionManager.rolledback(brmTxnID);
string errMsg;
int rc = 0;
batchProcessor->getError(rc, errMsg);
result.result = DMLPackageProcessor::TABLE_LOCK_ERROR;
logging::Message::Args args;
logging::Message message(1);
args.add("Insert Failed: ");
args.add(errMsg);
args.add("");
args.add("");
message.format(args);
result.message = message;
break;
}
if (tableLockId > 0)
tablelockData->setTablelock(insertPkg.getTableOid(), tableLockId);
}
}
if (insertPkg.get_Logending() && insertPkg.get_Logging()) // only one batch need to be processed.
{
// cout << "dmlprocessor add last pkg" << endl;
// need to add error handling.
batchProcessor->addPkg(bsSave);
batchProcessor->sendFirstBatch();
batchProcessor->receiveOutstandingMsg();
//@Bug 5162. Get the correct error message before the last message.
string errMsg;
int rc = 0;
batchProcessor->getError(rc, errMsg);
batchProcessor->sendlastBatch();
batchProcessor->receiveAllMsg();
if (rc == DMLPackageProcessor::IDBRANGE_WARNING)
{
result.result = dmlpackageprocessor::DMLPackageProcessor::IDBRANGE_WARNING;
LoggingID logid(DMLLoggingId, insertPkg.get_SessionID(), fTxnid);
logging::Message::Args args1;
logging::Message msg(1);
args1.add("End SQL statement with warnings");
msg.format(args1);
logging::Logger logger(logid.fSubsysID);
logger.logMessage(LOG_TYPE_DEBUG, msg, logid);
logging::logDML(insertPkg.get_SessionID(), fTxnid, insertPkg.get_SQLStatement() + ";",
insertPkg.get_SchemaName());
logging::Message::Args args;
logging::Message message(1);
args.add(errMsg);
args.add("");
args.add("");
message.format(args);
result.message = message;
}
else if (rc != 0)
{
result.result = dmlpackageprocessor::DMLPackageProcessor::INSERT_ERROR;
logging::Message::Args args;
logging::Message message(1);
cout << "Got error in the end of one batchinsert." << endl;
args.add("Insert Failed: ");
args.add(errMsg);
args.add("");
args.add("");
message.format(args);
result.message = message;
LoggingID logid(DMLLoggingId, insertPkg.get_SessionID(), fTxnid);
logging::Message::Args args1;
logging::Message msg(1);
args1.add("End SQL statement with error");
msg.format(args1);
logging::Logger logger(logid.fSubsysID);
logger.logMessage(LOG_TYPE_DEBUG, msg, logid);
logging::logDML(insertPkg.get_SessionID(), fTxnid, insertPkg.get_SQLStatement() + ";",
insertPkg.get_SchemaName());
}
else
{
// if (!insertPkg.get_isAutocommitOn())
// {
// batchProcessor->setHwm();
// }
LoggingID logid(DMLLoggingId, insertPkg.get_SessionID(), fTxnid);
logging::Message::Args args1;
logging::Message msg(1);
args1.add("End SQL statement");
msg.format(args1);
logging::Logger logger(logid.fSubsysID);
logger.logMessage(LOG_TYPE_DEBUG, msg, logid);
logging::logDML(insertPkg.get_SessionID(), fTxnid, insertPkg.get_SQLStatement() + ";",
insertPkg.get_SchemaName());
}
// remove the batch insert object
{
boost::mutex::scoped_lock lk(DMLProcessor::batchinsertProcessorMapLock);
std::map<uint32_t, BatchInsertProc*>::iterator batchIter =
DMLProcessor::batchinsertProcessorMap.find(fSessionID);
if (batchIter != DMLProcessor::batchinsertProcessorMap.end())
{
delete batchIter->second;
DMLProcessor::batchinsertProcessorMap.erase(fSessionID);
}
}
}
else if (insertPkg.get_Logending()) // Last batch
{
int rc = 0;
string errMsg;
batchProcessor->getError(rc, errMsg);
// cout <<"dmlprocessor received last pkg from mysql rc == " << rc << endl;
if ((rc == 0) || (rc == DMLPackageProcessor::IDBRANGE_WARNING))
{
// cout << " rc = " << rc << endl;
batchProcessor->addPkg(bsSave);
batchProcessor->sendNextBatch();
batchProcessor->receiveOutstandingMsg();
//@Bug 5162. Get the correct error message before the last message.
batchProcessor->getError(rc, errMsg);
batchProcessor->sendlastBatch();
batchProcessor->receiveAllMsg();
if (rc == DMLPackageProcessor::IDBRANGE_WARNING)
{
result.result = dmlpackageprocessor::DMLPackageProcessor::IDBRANGE_WARNING;
LoggingID logid(DMLLoggingId, insertPkg.get_SessionID(), fTxnid);
logging::Message::Args args1;
logging::Message msg(1);
args1.add("End SQL statement with warnings");
msg.format(args1);
logging::Logger logger(logid.fSubsysID);
logger.logMessage(LOG_TYPE_DEBUG, msg, logid);
logging::logDML(insertPkg.get_SessionID(), fTxnid, insertPkg.get_SQLStatement() + ";",
insertPkg.get_SchemaName());
logging::Message::Args args;
logging::Message message(1);
args.add(errMsg);
args.add("");
args.add("");
message.format(args);
result.message = message;
}
else if (rc != 0)
{
// cout << "Got error in the end of last batchinsert. error message is " << errMsg << endl;
result.result = dmlpackageprocessor::DMLPackageProcessor::INSERT_ERROR;
logging::Message::Args args;
logging::Message message(1);
args.add("Insert Failed: ");
args.add(errMsg);
args.add("");
args.add("");
message.format(args);
result.message = message;
LoggingID logid(DMLLoggingId, insertPkg.get_SessionID(), fTxnid);
logging::Message::Args args1;
logging::Message msg(1);
args1.add("End SQL statement with error");
msg.format(args1);
logging::Logger logger(logid.fSubsysID);
logger.logMessage(LOG_TYPE_DEBUG, msg, logid);
logging::logDML(insertPkg.get_SessionID(), fTxnid, insertPkg.get_SQLStatement() + ";",
insertPkg.get_SchemaName());
}
else
{
LoggingID logid(DMLLoggingId, insertPkg.get_SessionID(), fTxnid);
logging::Message::Args args1;
logging::Message msg(1);
args1.add("End SQL statement");
msg.format(args1);
logging::Logger logger(logid.fSubsysID);
logger.logMessage(LOG_TYPE_DEBUG, msg, logid);
logging::logDML(insertPkg.get_SessionID(), fTxnid, insertPkg.get_SQLStatement() + ";",
insertPkg.get_SchemaName());
}
// cout << "finished batch insert" << endl;
}
else
{
// error occurred. Receive all outstanding messages before erroring out.
batchProcessor->receiveOutstandingMsg();
batchProcessor->sendlastBatch(); // needs to flush files
batchProcessor->receiveAllMsg();
result.result = dmlpackageprocessor::DMLPackageProcessor::INSERT_ERROR;
// cout << "Got error in the end of batchinsert2. error msg is " << errMsg<< endl;
logging::Message::Args args;
logging::Message message(1);
args.add("Insert Failed: ");
args.add(errMsg);
args.add("");
args.add("");
message.format(args);
result.message = message;
LoggingID logid(DMLLoggingId, insertPkg.get_SessionID(), fTxnid);
logging::Message::Args args1;
logging::Message msg(1);
args1.add("End SQL statement with error");
msg.format(args1);
logging::Logger logger(logid.fSubsysID);
logger.logMessage(LOG_TYPE_DEBUG, msg, logid);
logging::logDML(insertPkg.get_SessionID(), fTxnid, insertPkg.get_SQLStatement() + ";",
insertPkg.get_SchemaName());
}
// remove from map
{
boost::mutex::scoped_lock lk(DMLProcessor::batchinsertProcessorMapLock);
std::map<uint32_t, BatchInsertProc*>::iterator batchIter =
DMLProcessor::batchinsertProcessorMap.find(fSessionID);
if (batchIter != DMLProcessor::batchinsertProcessorMap.end())
{
// cout << "Batchinsertprcessor is deleted. " << batchIter->second << endl;
delete batchIter->second;
DMLProcessor::batchinsertProcessorMap.erase(fSessionID);
}
}
}
else
{
int rc = 0;
string errMsg;
batchProcessor->getError(rc, errMsg);
if (rc == DMLPackageProcessor::IDBRANGE_WARNING)
{
result.result = dmlpackageprocessor::DMLPackageProcessor::IDBRANGE_WARNING;
}
else if (rc != 0)
{
result.result = dmlpackageprocessor::DMLPackageProcessor::INSERT_ERROR;
//@Bug
// cout << "Got error during batchinsert. with message " << errMsg << endl;
logging::Message::Args args;
logging::Message message(6);
args.add(errMsg);
message.format(args);
result.message = message;
batchProcessor->receiveOutstandingMsg();
batchProcessor->sendlastBatch(); // needs to flush files
// cout << "Last batch is sent to WES." << endl;
batchProcessor->receiveAllMsg();
LoggingID logid(DMLLoggingId, insertPkg.get_SessionID(), fTxnid);
logging::Message::Args args1;
logging::Message msg(1);
args1.add("End SQL statement with error");
msg.format(args1);
logging::Logger logger(logid.fSubsysID);
logger.logMessage(LOG_TYPE_DEBUG, msg, logid);
// remove from map
{
boost::mutex::scoped_lock lk(DMLProcessor::batchinsertProcessorMapLock);
std::map<uint32_t, BatchInsertProc*>::iterator batchIter =
DMLProcessor::batchinsertProcessorMap.find(fSessionID);
if (batchIter != DMLProcessor::batchinsertProcessorMap.end())
{
// cout << "Batchinsertprcessor is deleted. " << batchIter->second << endl;
delete batchIter->second;
DMLProcessor::batchinsertProcessorMap.erase(fSessionID);
}
}
break;
}
batchProcessor->addPkg(bsSave);
batchProcessor->sendNextBatch();
break;
}
}
else // Single Insert
{
// make sure insertPkg.readMetaData() is called before
// this on fByteStream!
// TODO: Similar to batch inserts, don't
// deserialize the row data here for single inserts.
insertPkg.readRowData(*(fByteStream.get()));
insertPkg.set_TxnID(fTxnid);
fProcessor.reset(new dmlpackageprocessor::InsertPackageProcessor(fDbrm, insertPkg.get_SessionID()));
result = fProcessor->processPackage(insertPkg);
}
qts.msg_type = QueryTeleStats::QT_SUMMARY;
qts.max_mem_pct = result.stats.fMaxMemPct;
qts.num_files = result.stats.fNumFiles;
qts.phy_io = result.stats.fPhyIO;
qts.cache_io = result.stats.fCacheIO;
qts.msg_rcv_cnt = result.stats.fMsgRcvCnt;
qts.cp_blocks_skipped = result.stats.fCPBlocksSkipped;
qts.msg_bytes_in = result.stats.fMsgBytesIn;
qts.msg_bytes_out = result.stats.fMsgBytesOut;
qts.rows = result.stats.fRows;
qts.end_time = QueryTeleClient::timeNowms();
qts.blocks_changed = result.stats.fBlocksChanged;
fQtc.postQueryTele(qts);
}
break;
case dmlpackage::DML_UPDATE:
{
// build an UpdateDMLPackage from the bytestream
// cout << "an UPDATE package" << endl;
boost::scoped_ptr<dmlpackage::UpdateDMLPackage> updatePkg(new dmlpackage::UpdateDMLPackage());
updatePkg->read(*(fByteStream.get()));
#ifdef MCOL_140
if (fConcurrentSupport)
{
fTableOid = updatePkg->getTableOid();
// Update generally doesn't set fTableOid in updatePkg. Go get it now.
if (fTableOid == 0)
{
CalpontSystemCatalog::TableName tableName;
tableName.schema = updatePkg->get_Table()->get_SchemaName();
tableName.table = updatePkg->get_Table()->get_TableName();
CalpontSystemCatalog::ROPair roPair = getTableRID(fcsc, tableName);
fTableOid = roPair.objnum;
}
synchTable.setPackage(this,
updatePkg.get()); // Blocks if another DML thread is using this fTableOid
}
#endif
updatePkg->set_TxnID(fTxnid);
QueryTeleStats qts;
qts.query_uuid = updatePkg->uuid();
qts.msg_type = QueryTeleStats::QT_START;
qts.start_time = QueryTeleClient::timeNowms();
qts.session_id = fSessionID;
qts.query_type = "UPDATE";
qts.query = updatePkg->get_SQLStatement();
qts.system_name = oamCache->getSystemName();
qts.module_name = oamCache->getModuleName();
qts.schema_name = updatePkg->get_SchemaName();
fQtc.postQueryTele(qts);
// process it
//@Bug 1341. Don't remove calpontsystemcatalog from this
// session to take advantage of cache.
fProcessor.reset(new dmlpackageprocessor::UpdatePackageProcessor(fDbrm, updatePkg->get_SessionID()));
fProcessor->setEngineComm(fEC);
fProcessor->setRM(frm);
idbassert(fTxnid != 0);
result = fProcessor->processPackage(*(updatePkg.get()));
qts.msg_type = QueryTeleStats::QT_SUMMARY;
qts.max_mem_pct = result.stats.fMaxMemPct;
qts.num_files = result.stats.fNumFiles;
qts.phy_io = result.stats.fPhyIO;
qts.cache_io = result.stats.fCacheIO;
qts.msg_rcv_cnt = result.stats.fMsgRcvCnt;
qts.cp_blocks_skipped = result.stats.fCPBlocksSkipped;
qts.msg_bytes_in = result.stats.fMsgBytesIn;
qts.msg_bytes_out = result.stats.fMsgBytesOut;
qts.rows = result.stats.fRows;
qts.end_time = QueryTeleClient::timeNowms();
qts.blocks_changed = result.stats.fBlocksChanged;
fQtc.postQueryTele(qts);
}
break;
case dmlpackage::DML_DELETE:
{
boost::scoped_ptr<dmlpackage::DeleteDMLPackage> deletePkg(new dmlpackage::DeleteDMLPackage());
deletePkg->read(*(fByteStream.get()));
#ifdef MCOL_140
if (fConcurrentSupport)
{
fTableOid = deletePkg->getTableOid();
// Delete generally doesn't set fTableOid in updatePkg. Go get it now.
if (fTableOid == 0)
{
CalpontSystemCatalog::TableName tableName;
tableName.schema = deletePkg->get_Table()->get_SchemaName();
tableName.table = deletePkg->get_Table()->get_TableName();
CalpontSystemCatalog::ROPair roPair = getTableRID(fcsc, tableName);
fTableOid = roPair.objnum;
}
synchTable.setPackage(this,
deletePkg.get()); // Blocks if another DML thread is using this fTableOid
}
#endif
deletePkg->set_TxnID(fTxnid);
QueryTeleStats qts;
qts.query_uuid = deletePkg->uuid();
qts.msg_type = QueryTeleStats::QT_START;
qts.start_time = QueryTeleClient::timeNowms();
qts.session_id = fSessionID;
qts.query_type = "DELETE";
qts.query = deletePkg->get_SQLStatement();
qts.system_name = oamCache->getSystemName();
qts.module_name = oamCache->getModuleName();
qts.schema_name = deletePkg->get_SchemaName();
fQtc.postQueryTele(qts);
// process it
//@Bug 1341. Don't remove calpontsystemcatalog from this session to take advantage of cache.
fProcessor.reset(new dmlpackageprocessor::DeletePackageProcessor(fDbrm, deletePkg->get_SessionID()));
fProcessor->setEngineComm(fEC);
fProcessor->setRM(frm);
idbassert(fTxnid != 0);
result = fProcessor->processPackage(*(deletePkg.get()));
qts.msg_type = QueryTeleStats::QT_SUMMARY;
qts.max_mem_pct = result.stats.fMaxMemPct;
qts.num_files = result.stats.fNumFiles;
qts.phy_io = result.stats.fPhyIO;
qts.cache_io = result.stats.fCacheIO;
qts.msg_rcv_cnt = result.stats.fMsgRcvCnt;
qts.cp_blocks_skipped = result.stats.fCPBlocksSkipped;
qts.msg_bytes_in = result.stats.fMsgBytesIn;
qts.msg_bytes_out = result.stats.fMsgBytesOut;
qts.rows = result.stats.fRows;
qts.end_time = QueryTeleClient::timeNowms();
qts.blocks_changed = result.stats.fBlocksChanged;
fQtc.postQueryTele(qts);
}
break;
case dmlpackage::DML_COMMAND:
{
// build a CommandDMLPackage from the bytestream
// cout << "a COMMAND package" << endl;
dmlpackage::CommandDMLPackage commandPkg;
commandPkg.read(*(fByteStream.get()));
stmt = commandPkg.get_DMLStatement();
boost::algorithm::to_upper(stmt);
trim(stmt);
if (stmt == "CLEANUP")
{
execplan::CalpontSystemCatalog::removeCalpontSystemCatalog(commandPkg.get_SessionID());
execplan::CalpontSystemCatalog::removeCalpontSystemCatalog(commandPkg.get_SessionID() | 0x80000000);
}
else
{
// process it
//@Bug 1341. Don't remove calpontsystemcatalog from this session to take advantage of cache.
fProcessor.reset(
new dmlpackageprocessor::CommandPackageProcessor(fDbrm, commandPkg.get_SessionID()));
// cout << "got command " << stmt << " for session " << commandPkg.get_SessionID() << endl;
result = fProcessor->processPackage(commandPkg);
}
}
break;
}
// Log errors
if ((result.result != dmlpackageprocessor::DMLPackageProcessor::NO_ERROR) &&
(result.result != dmlpackageprocessor::DMLPackageProcessor::IDBRANGE_WARNING) &&
(result.result != dmlpackageprocessor::DMLPackageProcessor::ACTIVE_TRANSACTION_ERROR) &&
(result.result != dmlpackageprocessor::DMLPackageProcessor::VB_OVERFLOW_ERROR))
{
logging::LoggingID lid(21);
logging::MessageLog ml(lid);
ml.logErrorMessage(result.message);
}
else if (result.result == dmlpackageprocessor::DMLPackageProcessor::IDBRANGE_WARNING)
{
logging::LoggingID lid(21);
logging::MessageLog ml(lid);
ml.logWarningMessage(result.message);
}
}
catch (std::exception& e)
{
cout << "dmlprocessor.cpp PackageHandler::run() package type(" << fPackageType
<< ") exception: " << e.what() << endl;
logging::LoggingID lid(21);
logging::MessageLog ml(lid);
logging::Message::Args args;
logging::Message message(1);
args.add("dmlprocessor.cpp PackageHandler::run() package type");
args.add((uint64_t)fPackageType);
args.add(e.what());
message.format(args);
ml.logErrorMessage(message);
result.result = DMLPackageProcessor::COMMAND_ERROR;
result.message = message;
}
catch (...)
{
logging::LoggingID lid(21);
logging::MessageLog ml(lid);
logging::Message::Args args;
logging::Message message(1);
args.add("dmlprocessor.cpp PackageHandler::run() ... exception package type");
args.add((uint64_t)fPackageType);
message.format(args);
ml.logErrorMessage(message);
result.result = DMLPackageProcessor::COMMAND_ERROR;
result.message = message;
}
// We put the packageHandler into a map so that if we receive a
// message to affect the previous command, we can find it.
// We need to remove it from the list before sending the response back.
// If we remove it after sending the results, it's possible for a commit
// or rollback be sent and get processed before it is removed, and that
// will fail.
boost::mutex::scoped_lock lk2(DMLProcessor::packageHandlerMapLock);
DMLProcessor::packageHandlerMap.erase(getSessionID());
lk2.unlock();
// send back the results
messageqcpp::ByteStream results;
messageqcpp::ByteStream::octbyte rowCount = result.rowCount;
messageqcpp::ByteStream::byte retval = result.result;
results << retval;
results << rowCount;
results << result.message.msg();
results << result.tableLockInfo; // ? connector does not get
// query stats
results << result.queryStats;
results << result.extendedStats;
results << result.miniStats;
result.stats.serialize(results);
fIos.write(results);
// Bug 5226. dmlprocessor thread will close the socket to mysqld.
// if (stmt == "CLEANUP")
// fIos.close();
}
void PackageHandler::rollbackPending()
{
if (fProcessor.get() == NULL)
{
// This happens when batch insert
return;
}
fProcessor->setRollbackPending(true);
// Force a release of the processing from MCOL-140
#ifdef MCOL_140
if (fConcurrentSupport)
{
// MCOL-140 We're not necessarily the next in line.
// This forces this thread to be released anyway.
forceReleaseTableAccess();
}
#endif
ostringstream oss;
oss << "PackageHandler::rollbackPending: Processing DMLPackage.";
DMLProcessor::log(oss.str(), logging::LOG_TYPE_DEBUG);
}
void added_a_pm(int)
{
DistributedEngineComm* dec;
ResourceManager* rm = ResourceManager::instance();
dec = DistributedEngineComm::instance(rm);
dec->Setup();
// MCOL-140 clear the waiting queue as all transactions are probably going to fail
PackageHandler::clearTableAccess();
}
DMLServer::DMLServer(int packageMaxThreads, int packageWorkQueueSize, DBRM* dbrm)
: fPackageMaxThreads(packageMaxThreads)
, fPackageWorkQueueSize(packageWorkQueueSize)
, fDbrm(dbrm)
, fShutdownFlag(false)
{
fMqServer.reset(new MessageQueueServer("DMLProc"));
fDmlPackagepool.setMaxThreads(fPackageMaxThreads);
fDmlPackagepool.setQueueSize(fPackageWorkQueueSize);
fDmlPackagepool.setName("DmlPackagepool");
}
int DMLServer::start()
{
messageqcpp::IOSocket ios;
uint32_t nextID = 1;
try
{
// CancellationThread is for telling all active transactions
// to quit working because the system is either going down
// or going into write suspend mode
CancellationThread cancelObject(fDbrm, *this);
boost::thread cancelThread(cancelObject);
cout << "DMLProc is ready..." << endl;
const static struct timespec timeout = {1, 100}; // roughly 1 second TO
for (;;)
{
ios = fMqServer->accept(&timeout);
// MCS polls in a loop watching for a pending shutdown
// that is signalled via fShutdownFlag set in a
// CancellationThread. CT sets the flag if a cluster state
// has SS_SHUTDOWNPENDING value set.
while (!ios.hasSocketDescriptor() && !pendingShutdown())
ios = fMqServer->accept(&timeout);
if (pendingShutdown())
break;
ios.setSockID(nextID++);
fDmlPackagepool.invoke(DMLProcessor(ios, fDbrm));
}
cancelThread.join();
return EXIT_SUCCESS;
}
catch (std::exception& ex)
{
cerr << ex.what() << endl;
logging::LoggingID lid(21);
Message::Args args;
Message message(8);
args.add("DMLProc init caught exception: ");
args.add(ex.what());
message.format(args);
logging::Logger logger(lid.fSubsysID);
logger.logMessage(logging::LOG_TYPE_CRITICAL, message, lid);
return EXIT_FAILURE;
}
catch (...)
{
cerr << "Caught unknown exception!" << endl;
logging::LoggingID lid(21);
Message::Args args;
Message message(8);
args.add("DMLProc init caught unknown exception");
message.format(args);
logging::Logger logger(lid.fSubsysID);
logger.logMessage(logging::LOG_TYPE_CRITICAL, message, lid);
return EXIT_FAILURE;
}
}
DMLProcessor::DMLProcessor(messageqcpp::IOSocket ios, BRM::DBRM* aDbrm) : fIos(ios), fDbrm(aDbrm)
{
csc = CalpontSystemCatalog::makeCalpontSystemCatalog();
csc->identity(CalpontSystemCatalog::EC);
string teleServerHost(config::Config::makeConfig()->getConfig("QueryTele", "Host"));
if (!teleServerHost.empty())
{
int teleServerPort =
config::Config::fromText(config::Config::makeConfig()->getConfig("QueryTele", "Port"));
if (teleServerPort > 0)
{
fQtc.serverParms(QueryTeleServerParms(teleServerHost, teleServerPort));
}
}
}
void DMLProcessor::operator()()
{
bool bIsDbrmUp = true;
try
{
boost::shared_ptr<messageqcpp::ByteStream> bs1(new messageqcpp::ByteStream());
// messageqcpp::ByteStream bs;
uint8_t packageType;
ResourceManager* rm = ResourceManager::instance();
DistributedEngineComm* fEC = DistributedEngineComm::instance(rm);
uint64_t maxDeleteRows = rm->getDMLMaxDeleteRows();
fConcurrentSupport = true;
string concurrentTranStr =
config::Config::makeConfig()->getConfig("SystemConfig", "ConcurrentTransactions");
if (concurrentTranStr.length() != 0)
{
if ((concurrentTranStr.compare("N") == 0) || (concurrentTranStr.compare("n") == 0))
fConcurrentSupport = false;
}
struct sigaction ign;
memset(&ign, 0, sizeof(ign));
ign.sa_handler = added_a_pm;
sigaction(SIGHUP, &ign, 0);
fEC->Open();
for (;;)
{
// cout << "DMLProc is waiting for a Calpont DML Package on " << fIos.getSockID() << endl;
try
{
bs1.reset(new messageqcpp::ByteStream(fIos.read()));
// cout << "received from mysql socket " << fIos.getSockID() << endl;
}
catch (std::exception& ex)
{
// This is an I/O error from InetStreamSocket::read(), just close and move on...
cout << "runtime error during read on " << fIos.getSockID() << " " << ex.what() << endl;
bs1->reset();
}
catch (...)
{
cout << "... error during read " << fIos.getSockID() << endl;
// all this throw does is cause this thread to silently go away. I doubt this is the right
// thing to do...
throw;
}
if (!bs1 || bs1->length() == 0)
{
cout << "Read 0 bytes. Closing connection " << fIos.getSockID() << endl;
fIos.close();
break;
}
uint32_t sessionID;
*bs1 >> sessionID;
*bs1 >> packageType;
// cout << "DMLProc received pkg. sessionid:type = " << sessionID <<":"<<(int)packageType << endl;
uint32_t stateFlags;
messageqcpp::ByteStream::byte status = 255;
messageqcpp::ByteStream::octbyte rowCount = 0;
if (fDbrm->getSystemState(stateFlags) >
0) // > 0 implies succesful retrieval. It doesn't imply anything about the contents
{
messageqcpp::ByteStream results;
std::string responseMsg;
bool bReject = false;
// Check to see if we're in write suspended mode
// If so, we can't process the request.
if (stateFlags & SessionManagerServer::SS_SUSPENDED)
{
status = DMLPackageProcessor::NOT_ACCEPTING_PACKAGES;
responseMsg = "Writing to the database is disabled.";
bReject = true;
}
// Check to see if we're in write suspend or shutdown pending mode
if (packageType != dmlpackage::DML_COMMAND) // Not a commit or rollback
{
if (stateFlags & SessionManagerServer::SS_SUSPEND_PENDING ||
stateFlags & SessionManagerServer::SS_SHUTDOWN_PENDING)
{
if (stateFlags & SessionManagerServer::SS_SUSPEND_PENDING)
{
responseMsg = "Writing to the database is disabled.";
}
else
{
responseMsg = "The database is being shut down.";
}
// Refuse all non active tranasactions
// Check the rollback flag
// -- Set: Rollback active transactions.
// -- Not set: Allow active transactions.
if (sessionManager.isTransactionActive(sessionID, bIsDbrmUp))
{
if (stateFlags & SessionManagerServer::SS_ROLLBACK)
{
status = DMLPackageProcessor::JOB_CANCELED;
bReject = true;
}
}
else
{
status = DMLPackageProcessor::NOT_ACCEPTING_PACKAGES;
bReject = true;
}
}
// MCOL-4988 Check if DBRM is in READ ONLY mode
if (fDbrm->isReadWrite() == BRM::ERR_READONLY)
{
BRM::errString(BRM::ERR_READONLY, responseMsg);
status = DMLPackageProcessor::DBRM_READ_ONLY;
bReject = true;
}
if (bReject)
{
// For batch insert, we need to send a lastpkg message
// to batchInsertProcessor so it can clean things up.
if (packageType == dmlpackage::DML_INSERT)
{
// build an InsertDMLPackage from the bytestream
// We need the flags from the package to know what
// type of package we're dealing with before we can
// take special action for the last package of a
// batch insert.
dmlpackage::InsertDMLPackage insertPkg;
messageqcpp::ByteStream bsSave = *(bs1.get());
insertPkg.read(*(bs1.get()));
BatchInsertProc* batchInsertProcessor = NULL;
if (insertPkg.get_isBatchInsert() && insertPkg.get_Logending())
{
{
boost::mutex::scoped_lock lk(DMLProcessor::batchinsertProcessorMapLock);
std::map<uint32_t, BatchInsertProc*>::iterator batchIter =
DMLProcessor::batchinsertProcessorMap.find(sessionID);
if (batchIter !=
DMLProcessor::batchinsertProcessorMap.end()) // The first batch, no need to do anything
{
batchInsertProcessor = batchIter->second;
batchInsertProcessor->addPkg(bsSave);
batchInsertProcessor->sendlastBatch();
batchInsertProcessor->receiveAllMsg();
if (!insertPkg.get_isAutocommitOn())
{
batchInsertProcessor->setHwm();
}
batchIter = DMLProcessor::batchinsertProcessorMap.find(sessionID);
if (batchIter != DMLProcessor::batchinsertProcessorMap.end())
{
DMLProcessor::batchinsertProcessorMap.erase(sessionID);
}
}
}
}
}
results << status;
results << rowCount;
logging::Message::Args args;
logging::Message message(2);
args.add(responseMsg);
message.format(args);
results << message.msg();
fIos.write(results);
continue;
}
}
}
// This section is to check to see if the user hit CTRL+C while the
// DML was processing If so, the sessionID will be found in
// packageHandlerMap and we can set rollbackPending in the
// associated packageHandler. Other than CTRL+C, we should never
// find our own sessionID in the map.
// This mechanism may prove useful for other things, so the above
// comment may change.
{
boost::mutex::scoped_lock lk2(DMLProcessor::packageHandlerMapLock);
DMLProcessor::PackageHandlerMap_t::iterator phIter = packageHandlerMap.find(sessionID);
if (phIter != packageHandlerMap.end())
{
if (packageType == dmlpackage::DML_COMMAND)
{
// MCOL-66 It's possible for a commit or rollback to get here if
// the timing is just right. Don't destroy its data
messageqcpp::ByteStream bsctrlc(bs1);
dmlpackage::CommandDMLPackage commandPkg;
commandPkg.read(bsctrlc);
std::string stmt = commandPkg.get_DMLStatement();
boost::algorithm::to_upper(stmt);
trim(stmt);
if (stmt == "CTRL+C")
{
phIter->second->rollbackPending();
fIos.close();
break;
}
}
else
{
// If there's a PackageHandler already working for this
// sessionID, we have a problem. Reject this package
messageqcpp::ByteStream results;
ostringstream oss;
oss << "Received a DML command for session " << sessionID
<< " while still processing a command for the same sessionID";
results << static_cast<messageqcpp::ByteStream::byte>(DMLPackageProcessor::DEAD_LOCK_ERROR);
results << static_cast<messageqcpp::ByteStream::octbyte>(0); // rowcount
logging::Message::Args args;
logging::Message message(2);
args.add(oss.str());
message.format(args);
logging::LoggingID lid(20);
logging::MessageLog ml(lid);
ml.logErrorMessage(message);
results << message.msg();
fIos.write(results);
continue;
}
}
}
// cout << " got a ";
switch (packageType)
{
case dmlpackage::DML_INSERT:
// cout << "DML_INSERT";
break;
case dmlpackage::DML_UPDATE:
// cout << "DML_UPDATE";
break;
case dmlpackage::DML_DELETE:
// cout << "DML_DELETE";
break;
case dmlpackage::DML_COMMAND:
// cout << "DML_COMMAND";
break;
case dmlpackage::DML_INVALID_TYPE:
// cout << "DML_INVALID_TYPE";
break;
default:
// cout << "UNKNOWN";
break;
}
// cout << " package" << endl;
BRM::TxnID txnid;
if (!fConcurrentSupport)
{
// Check if any other active transaction
bool anyOtherActiveTransaction = true;
BRM::SIDTIDEntry blockingsid;
// For logout commit trigger
if (packageType == dmlpackage::DML_COMMAND)
{
anyOtherActiveTransaction = false;
}
int i = 0;
int waitPeriod = 10;
//@Bug 2487 Check transaction map every 1/10 second
int sleepTime = 100; // sleep 100 milliseconds between checks
int numTries = 10; // try 10 times per second
string waitPeriodStr = config::Config::makeConfig()->getConfig("SystemConfig", "WaitPeriod");
if (waitPeriodStr.length() != 0)
waitPeriod = static_cast<int>(config::Config::fromText(waitPeriodStr));
numTries = waitPeriod * 10;
struct timespec rm_ts;
rm_ts.tv_sec = sleepTime / 1000;
rm_ts.tv_nsec = sleepTime % 1000 * 1000000;
// cout << "starting i = " << i << endl;
// txnid = sessionManager.getTxnID(sessionID);
while (anyOtherActiveTransaction)
{
anyOtherActiveTransaction =
sessionManager.checkActiveTransaction(sessionID, bIsDbrmUp, blockingsid);
// cout << "session " << sessionID << " with package type " << (int)packageType << " got
// anyOtherActiveTransaction " << anyOtherActiveTransaction << endl;
if (anyOtherActiveTransaction)
{
for (; i < numTries; i++)
{
struct timespec abs_ts;
// cout << "session " << sessionID << " nanosleep on package type " << (int)packageType << endl;
do
{
abs_ts.tv_sec = rm_ts.tv_sec;
abs_ts.tv_nsec = rm_ts.tv_nsec;
} while (nanosleep(&abs_ts, &rm_ts) < 0);
anyOtherActiveTransaction =
sessionManager.checkActiveTransaction(sessionID, bIsDbrmUp, blockingsid);
if (!anyOtherActiveTransaction)
{
txnid = sessionManager.getTxnID(sessionID);
// cout << "Ready to process type " << (int)packageType << " with txd " << txnid << endl;
if (!txnid.valid)
{
txnid = sessionManager.newTxnID(sessionID, true);
if (txnid.valid)
{
// cout << "Ready to process type " << (int)packageType << " for session "<< sessionID <<
// " with new txnid " << txnid.id << endl;
anyOtherActiveTransaction = false;
break;
}
else
{
anyOtherActiveTransaction = true;
}
}
else
{
anyOtherActiveTransaction = false;
// cout << "already have transaction to process type " << (int)packageType << " for session
// "<< sessionID <<" with existing txnid " << txnid.id << endl;
break;
}
}
}
// cout << "ending i = " << i << endl;
}
else
{
// cout << "Ready to process type " << (int)packageType << endl;
txnid = sessionManager.getTxnID(sessionID);
if (!txnid.valid)
{
txnid = sessionManager.newTxnID(sessionID, true);
if (txnid.valid)
{
// cout << "later Ready to process type " << (int)packageType << " for session "<< sessionID
// << " with new txnid " << txnid.id << endl;
anyOtherActiveTransaction = false;
}
else
{
anyOtherActiveTransaction = true;
// cout << "Cannot get txnid for process type " << (int)packageType << " for session "<<
// sessionID << endl;
}
}
else
{
anyOtherActiveTransaction = false;
// cout << "already have transaction to process type " << (int)packageType << " for session "<<
// sessionID <<" with txnid " << txnid.id << endl;
break;
}
}
if ((anyOtherActiveTransaction) && (i >= numTries))
{
// cout << " Erroring out on package type " << (int)packageType << " for session " << sessionID <<
// endl;
break;
}
}
if (anyOtherActiveTransaction && (i >= numTries))
{
// cout << " again Erroring out on package type " << (int)packageType << endl;
messageqcpp::ByteStream results;
//@Bug 2681 set error code for active transaction
status = DMLPackageProcessor::ACTIVE_TRANSACTION_ERROR;
rowCount = 0;
results << status;
results << rowCount;
Message::Args args;
args.add(static_cast<uint64_t>(blockingsid.sessionid));
results << IDBErrorInfo::instance()->errorMsg(ERR_ACTIVE_TRANSACTION, args);
//@Bug 3854 Log to debug.log
LoggingID logid(20, 0, 0);
logging::Message::Args args1;
logging::Message msg(1);
args1.add(IDBErrorInfo::instance()->errorMsg(ERR_ACTIVE_TRANSACTION, args));
msg.format(args1);
logging::Logger logger(logid.fSubsysID);
logger.logMessage(LOG_TYPE_DEBUG, msg, logid);
fIos.write(results);
}
else
{
// cout << "starting processing package type " << (int) packageType << " for session " << sessionID
// << " with id " << txnid.id << endl;
boost::shared_ptr<PackageHandler> php(new PackageHandler(fIos, bs1, packageType, fEC,
fConcurrentSupport, maxDeleteRows,
sessionID, txnid.id, fDbrm, fQtc, csc));
// We put the packageHandler into a map so that if we receive a
// message to affect the previous command, we can find it.
boost::mutex::scoped_lock lk2(DMLProcessor::packageHandlerMapLock, boost::defer_lock);
lk2.lock();
packageHandlerMap[sessionID] = php;
lk2.unlock();
php->run(); // Operates in this thread.
// Move this to the end of PackageHandler so it is removed from the map before the response is sent
// lk2.lock();
// packageHandlerMap.erase(sessionID);
// lk2.unlock();
}
}
else
{
#if 0
if (packageType != dmlpackage::DML_COMMAND)
{
txnid = sessionManager.getTxnID(sessionID);
if ( !txnid.valid )
{
txnid = sessionManager.newTxnID(sessionID, true);
if (!txnid.valid)
{
throw std::runtime_error( std::string("Unable to start a transaction. Check critical log.") );
}
}
}
else
{
txnid = sessionManager.getTxnID(sessionID);
}
#endif
boost::shared_ptr<PackageHandler> php(new PackageHandler(
fIos, bs1, packageType, fEC, fConcurrentSupport, maxDeleteRows, sessionID, 0, fDbrm, fQtc, csc));
// We put the packageHandler into a map so that if we receive a
// message to affect the previous command, we can find it.
boost::mutex::scoped_lock lk2(DMLProcessor::packageHandlerMapLock, boost::defer_lock);
lk2.lock();
packageHandlerMap[sessionID] = php;
lk2.unlock();
php->run(); // Operates in this thread.
// Move this to the end of PackageHandler so it is removed from the map before the response is sent
// lk2.lock();
// packageHandlerMap.erase(sessionID);
// lk2.unlock();
}
}
}
catch (std::exception& ex)
{
ostringstream oss;
oss << "DMLProcessor failed on: " << ex.what();
DMLProcessor::log(oss.str(), logging::LOG_TYPE_DEBUG);
fIos.close();
}
catch (...)
{
ostringstream oss;
oss << "DMLProcessor failed on: processing DMLPackage.";
DMLProcessor::log(oss.str(), logging::LOG_TYPE_DEBUG);
cerr << "Caught unknown exception! " << oss.str();
fIos.close();
}
}
void RollbackTransactionProcessor::processBulkRollback(BRM::TableLockInfo lockInfo, BRM::DBRM* dbrm,
uint64_t uniqueId,
OamCache::dbRootPMMap_t& dbRootPMMap,
bool& lockReleased)
{
// Take over ownership of stale lock.
// Use "DMLProc" as process name, session id and transaction id -1 to distinguish from real DMLProc rollback
int32_t sessionID = -1;
int32_t txnid = -1;
std::string processName("DMLProc");
uint32_t processID = ::getpid();
bool ownerChanged = true;
lockReleased = true;
try
{
ownerChanged = dbrm->changeOwner(lockInfo.id, processName, processID, sessionID, txnid);
}
catch (std::exception&)
{
lockReleased = false;
throw std::runtime_error(IDBErrorInfo::instance()->errorMsg(ERR_HARD_FAILURE));
}
if (!ownerChanged)
{
lockReleased = false;
throw std::runtime_error(std::string("Unable to grab lock; lock not found or still in use."));
}
// send to all PMs
boost::shared_ptr<messageqcpp::ByteStream> bsIn;
messageqcpp::ByteStream bsOut;
string tableName("");
fWEClient->addQueue(uniqueId);
// find the PMs need to send the message to
std::set<int> pmSet;
int pmId;
for (uint32_t i = 0; i < lockInfo.dbrootList.size(); i++)
{
pmId = (*dbRootPMMap)[lockInfo.dbrootList[i]];
pmSet.insert(pmId);
}
if (lockInfo.state == BRM::LOADING)
{
bsOut << (messageqcpp::ByteStream::byte)WE_SVR_DML_BULKROLLBACK;
bsOut << uniqueId;
bsOut << lockInfo.id;
bsOut << lockInfo.tableOID;
bsOut << tableName;
bsOut << processName;
std::set<int>::const_iterator iter = pmSet.begin();
while (iter != pmSet.end())
{
fWEClient->write(bsOut, *iter);
iter++;
}
// Wait for "all" the responses, and accumulate any/all errors
unsigned int pmMsgCnt = 0;
while (pmMsgCnt < pmSet.size())
{
std::string rollbackErrMsg;
bsIn.reset(new messageqcpp::ByteStream());
fWEClient->read(uniqueId, bsIn);
if (bsIn->length() == 0)
{
fWEClient->removeQueue(uniqueId);
lockReleased = false;
throw std::runtime_error("Network error, PM rollback; ");
}
else
{
messageqcpp::ByteStream::byte rc;
uint16_t pmNum;
*bsIn >> rc;
*bsIn >> rollbackErrMsg;
*bsIn >> pmNum;
if (rc != 0)
{
fWEClient->removeQueue(uniqueId);
lockReleased = false;
throw std::runtime_error(rollbackErrMsg);
}
}
pmMsgCnt++;
} // end of while loop to process all responses to bulk rollback
// If no errors so far, then change state to CLEANUP state.
// We ignore the return stateChange flag.
dbrm->changeState(lockInfo.id, BRM::CLEANUP);
} // end of (lockInfo.state == BRM::LOADING)
// delete meta data backup rollback files
bsOut.reset();
bsOut << (messageqcpp::ByteStream::byte)WE_SVR_DML_BULKROLLBACK_CLEANUP;
bsOut << uniqueId;
bsOut << lockInfo.tableOID;
std::set<int>::const_iterator iter = pmSet.begin();
while (iter != pmSet.end())
{
fWEClient->write(bsOut, *iter);
iter++;
}
// Wait for "all" the responses, and accumulate any/all errors
unsigned int pmMsgCnt = 0;
//@Bug 4517 Release tablelock when it is in CLEANUP state
uint32_t rcCleanup = 0;
std::string fileDeleteErrMsg;
while (pmMsgCnt < pmSet.size())
{
bsIn.reset(new messageqcpp::ByteStream());
fWEClient->read(uniqueId, bsIn);
if (bsIn->length() == 0)
{
fWEClient->removeQueue(uniqueId);
rcCleanup = 1;
fileDeleteErrMsg = "Network error, PM clean up; ";
}
else
{
messageqcpp::ByteStream::byte rc;
uint16_t pmNum;
*bsIn >> rc;
*bsIn >> fileDeleteErrMsg;
*bsIn >> pmNum;
if ((rc != 0) && (rcCleanup == 0))
{
fWEClient->removeQueue(uniqueId);
rcCleanup = rc;
}
}
pmMsgCnt++;
} // end of while loop to process all responses to rollback cleanup
fWEClient->removeQueue(uniqueId);
// We ignore return release flag from releaseTableLock().
dbrm->releaseTableLock(lockInfo.id);
if (rcCleanup != 0)
throw std::runtime_error(fileDeleteErrMsg);
}
void DMLProcessor::log(const std::string& msg, logging::LOG_TYPE level)
{
logging::Message::Args args;
logging::Message message(2);
args.add(msg);
message.format(args);
logging::LoggingID lid(20);
logging::MessageLog ml(lid);
switch (level)
{
case LOG_TYPE_DEBUG: ml.logDebugMessage(message); break;
case LOG_TYPE_INFO: ml.logInfoMessage(message); break;
case LOG_TYPE_WARNING: ml.logWarningMessage(message); break;
case LOG_TYPE_ERROR: ml.logErrorMessage(message); break;
case LOG_TYPE_CRITICAL: ml.logCriticalMessage(message); break;
}
}
} // namespace dmlprocessor
View Git Blame Copy Permalink