QDb.cc

Go to the documentation of this file.

/*
* APOLLO: A complete DAQ and Online Data Analysis Framework for Diana 
* $Id: QDb.cc 631 2006-12-19 11:54:27Z giacher $
* M.P. created 21/9/4
* 
* Class QDb: base class for DB interfaces (Apollo & Diana)
*
*/
 
#include <stdexcept>
#include <iostream>
#include <sstream>
#include <cstdio>
#include <unistd.h>
 
#include "QDb.hh"
#include "QError.hh"
 
#include "QDbWatchdogThreadHandler.hh"
#include "QWatchdogTimer.hh"
 
#include <boost/thread/mutex.hpp>
#include <boost/thread/locks.hpp>
 
 
//_____________________________________________________________________________
QDb::QDb(const std::string& dbHost, 
         const std::string& dbPort, 
         const std::string& dbUsr, 
         const std::string& dbPwd, 
         const std::string& dbName):
   fConnection(NULL),
   fDbHost(dbHost),
   fDbPort(dbPort),
   fDbUser(dbUsr),
   fDbPwd(dbPwd),
   fDbName(dbName),
   fConnectAttemptTimeoutSec(0),
   fMinConnectDelaySec(-1),
   fMaxConnectDelaySec(-1),
   fDisconnectTimeoutSec(-1),
   fThreadHandler(NULL),
   fWatchdogTimer(NULL),
   fMutex(NULL)
{
   fMutex = new boost::mutex();
}
 
 
//_____________________________________________________________________________
QDb::~QDb() 
{
  if (IsConnected())
     Disconnect();
 
  if(fThreadHandler != NULL)
     delete fThreadHandler;
  if(fWatchdogTimer != NULL)
     delete fWatchdogTimer;
 
  delete fMutex;
}
 
 
//_____________________________________________________________________________
bool QDb::IsConnected() const
{
   return (CONNECTION_OK == PQstatus(fConnection));
}
 
 
//_____________________________________________________________________________
QError QDb::Connect()
{
   QError err = QERR_SUCCESS;
 
   // variables for handling multiple connect attempts
   int maxNumAttempts = 0;
   double sleepTimeSec = 1;
   if(fConnectAttemptTimeoutSec > 0)
      maxNumAttempts = fConnectAttemptTimeoutSec / sleepTimeSec;
 
   err = InitWatchdogThread();
   if(err != QERR_SUCCESS)
      return err;
   
   int attempt = 0;
   while(!IsConnected())
   {
      if(0 == attempt)
         RandomWaitBeforeConnect();
      
      err = ConnectBase();
      if(err == QERR_SUCCESS)
         break;
      
      if(attempt++ < maxNumAttempts)
         sleep(sleepTimeSec);
      else
         break;
   }
 
   if(err == QERR_SUCCESS) // i.e. connection was successful
   {
      err = UpdateWatchdogTimer();
      if(err != QERR_SUCCESS)
         return err;
   }
   
   return err;
}
 
 
//_____________________________________________________________________________
QError QDb::ConnectBase() 
{
   if (IsConnected())
      return QERR_SUCCESS;
 
   const char* pgoptions = NULL;
   const char* pgtty = NULL;
   
   fConnection = PQsetdbLogin(fDbHost.c_str(),
                              fDbPort.c_str(),
                              pgoptions,
                              pgtty, 
                              fDbName.c_str(),
                              fDbUser.c_str(),
                              fDbPwd.c_str());
   
   if( !IsConnected() )
   {
     std::string msg = "Cannot connect to DB. Srv="
        + fDbHost
        + " Port="
        + fDbPort
        + " DataBase="
        + fDbName
        + " User="
        + fDbUser
        + ": "
        + PQerrorMessage(fConnection);
     //std::cerr << msg << std::endl;
     return QError(QERR_DB_CONN_FAILED, __FILE__, __LINE__, msg);
  }
  return QERR_SUCCESS;
}
 
 
//_____________________________________________________________________________
bool QDb::Disconnect() 
{
  boost::lock_guard<boost::mutex> guard(*fMutex);
  return UnlockedDisconnect();  
}
 
//_____________________________________________________________________________
bool QDb::UnlockedDisconnect() 
{
  if(NULL == fConnection)
     return true;
  PQfinish(fConnection);
  fConnection = NULL;
  return true;
}
 
 
//_____________________________________________________________________________
QDb::QDbTable QDb::DoQuery(const std::string& query) 
{
   QDbTable table;
   QError err = DoQuery(query, table);
   if(err != QERR_SUCCESS)
      throw err;
   return table;
   
}
 
 
//_____________________________________________________________________________
double QDb::DoQueryDouble( const std::string& query)
{
   QDbTable table;
   QError err = DoQuery(query, table);
   if(err != QERR_SUCCESS)
      throw err;
 
   err = CheckSingleResult(table, query, __FUNCTION__, __FILE__, __LINE__);
   if(err != QERR_SUCCESS)
      throw err;
   
   return table.begin()->second.front().GetDouble();
}
 
 
//_____________________________________________________________________________
int QDb::DoQueryInt( const std::string& query)
{
   QDbTable table;
   QError err = DoQuery(query, table);
   if(err != QERR_SUCCESS)
      throw err;
 
   err = CheckSingleResult(table, query, __FUNCTION__, __FILE__, __LINE__);
   if(err != QERR_SUCCESS)
      throw err;
   
   return table.begin()->second.front().GetInt();
}
 
 
//_____________________________________________________________________________
bool QDb::DoQueryBool( const std::string& query)
{
   QDbTable table;
   QError err = DoQuery(query, table);
   if(err != QERR_SUCCESS)
      throw err;
 
   err = CheckSingleResult(table, query, __FUNCTION__, __FILE__, __LINE__);
   if(err != QERR_SUCCESS)
      throw err;
   
   return table.begin()->second.front().GetBool();
}
 
 
//_____________________________________________________________________________
std::string QDb::DoQueryString( const std::string& query)
{
   QDbTable table;
   QError err = DoQuery(query, table);
   if(err != QERR_SUCCESS)
      throw err;
 
   err = CheckSingleResult(table, query, __FUNCTION__, __FILE__, __LINE__);
   if(err != QERR_SUCCESS)
      throw err;
   
   return table.begin()->second.front().GetString();
}
 
 
//_____________________________________________________________________________
std::vector<int> QDb::DoQueryVectorInt( const std::string& query)
{
   QDbTable table;
   QError err = DoQuery(query, table);
   if(err != QERR_SUCCESS)
      throw err;
 
   err = CheckSingleResult(table, query, __FUNCTION__, __FILE__, __LINE__);
   if(err != QERR_SUCCESS)
      throw err;
 
   return table.begin()->second[0].GetVectorInt();
}
 
//_____________________________________________________________________________
std::vector<double> QDb::DoQueryVectorDouble( const std::string& query)
{
   QDbTable table;
   QError err = DoQuery(query, table);
   if(err != QERR_SUCCESS)
      throw err;
 
   err = CheckSingleResult(table, query, __FUNCTION__, __FILE__, __LINE__);
   if(err != QERR_SUCCESS)
      throw err;
 
   return table.begin()->second[0].GetVectorDouble();
}
 
 
//_____________________________________________________________________________
std::vector<std::string> QDb::DoQueryVectorString( const std::string& query)
{
   QDbTable table;
   QError err = DoQuery(query, table);
   if(err != QERR_SUCCESS)
      throw err;
 
   err = CheckSingleResult(table, query, __FUNCTION__, __FILE__, __LINE__);
   if(err != QERR_SUCCESS)
      throw err;
 
   return table.begin()->second[0].GetVectorString();
}
 
 
//_____________________________________________________________________________
QError QDb::DoSQLNoReturn(const std::string& query) 
{
   PGresult* result = NULL;
   QError err = Query(query, &result);
   if(err != QERR_SUCCESS)
      return err;
 
   PQclear(result);
   return QERR_SUCCESS;
}
 
 
//_____________________________________________________________________________
int QDb::DoExec(const std::string& query)
{
   PGresult* result = NULL;
   QError err = Query(query, &result);
   if(err != QERR_SUCCESS)
      throw err;
   
    int affected = -1;
    sscanf(PQcmdTuples(result), "%d", &affected);
    PQclear(result);
    return affected;
}
 
 
//_____________________________________________________________________________
int QDb::Insert(const std::string& tableName, const column& fields, const column & values)
{
    if(fields.size() != values.size()) {
        QError err = QERR_SIZE_NOT_MATCH;
        std::string errMsg = "QDb::Insert: size of fields differs from size of values";
        err.SetDescription(__FILE__,__LINE__,errMsg);
        throw err;
    }
 
    std::string fieldlist;
    std::string valuelist;
    for(size_t i = 0; i < fields.size(); i++) {
        std::string valuestr = values[i].GetString();
        if(valuestr == "NULL" || valuestr == "Null" || valuestr == "null") { continue; }
        if(i >  0) {
            fieldlist+=",";
            valuelist+=",";
        }
        valuelist += valuestr;
        fieldlist += fields[i].GetString();
    }
 
    std::stringstream query;
    query<<"INSERT into "<<tableName<<" ("<<fieldlist<<")"
         <<" SELECT "<<valuelist<<" WHERE NOT EXISTS ("
         <<" SELECT * from "<<tableName<<" WHERE "
         <<"("<<fieldlist<<") IN (("<<valuelist<<")))";
    return DoExec(query.str());
}
 
 
//_____________________________________________________________________________
QError QDb::Query(const std::string& query, PGresult** result)
{
   boost::lock_guard<boost::mutex> guard(*fMutex);
   QError err = Connect();
   if(err != QERR_SUCCESS)
      return err;
 
   err = BasicQuery(query, result);
   if(err != QERR_SUCCESS) // try again once
   {
      UnlockedDisconnect();
 
      err = Connect();
      if(err != QERR_SUCCESS)
         return err;
 
      err = BasicQuery(query, result);
      return err;
   }
   return QERR_SUCCESS;
}
 
 
//_____________________________________________________________________________
QError QDb::BasicQuery(const std::string& query, PGresult** result) const
{
   *result = PQexec(fConnection, query.c_str());
   ExecStatusType status = PQresultStatus(*result);
   switch(status)
   {
   case PGRES_TUPLES_OK:
   case PGRES_COMMAND_OK:
      return QERR_SUCCESS;
 
   case PGRES_BAD_RESPONSE:
   case PGRES_NONFATAL_ERROR:
   case PGRES_FATAL_ERROR:
   case PGRES_EMPTY_QUERY:
   case PGRES_COPY_OUT:
   case PGRES_COPY_IN:
   case PGRES_COPY_BOTH:
   default:
      {
         std::string msg = "DoQuery: Error while executing  query \""
            + query
            + "\" - "
            + PQresStatus(status)
            + " - "
            + PQresultErrorMessage(*result);
         PQclear(*result);
         return QError(QERR_DB_QUERY_FAILED, __FILE__, __LINE__, msg);
      }
   }
}
 
 
//_____________________________________________________________________________
QError QDb::DoQuery(const std::string& query, QDbTable& table)
{
   table.clear();
   
   PGresult *result = NULL;
   QError err = Query(query, &result);
   if(err != QERR_SUCCESS)
      return err;
   
   if (PQresultStatus (result) != PGRES_TUPLES_OK)
    {
       std::string errMsg = "DoQuery: Error while executing  query \"";
      errMsg += query;
      errMsg += "\" - ";
      errMsg += PQresultErrorMessage(result);
      err.Set(QERR_DB_QUERY_FAILED, __FILE__, __LINE__, errMsg);
      PQclear(result);
      return err;
    }
  
  int nRows = PQntuples(result);
 
  if (0 == nRows)
  {
     PQclear(result);
     return QERR_SUCCESS;
  }
 
  
  int numFields = PQnfields(result);
  for (int field = 0; field < numFields; field++) 
  {
     const char *fieldName = PQfname(result, field);
     table[fieldName].reserve(nRows);
     for (int row = 0; row < nRows; row++) 
     {
        const char *value = PQgetvalue(result, row, field);
        table[fieldName].push_back(QVdt(value, fieldName));
     }
  }
  
  PQclear(result);
  return QERR_SUCCESS;
}
 
 
//_____________________________________________________________________________
QError QDb::CheckSingleResult(const QDbTable& table,
                              const std::string& query,
                              const std::string& funcName,
                              const std::string& file,
                              int line) const
{
   // a single result is a single field and a single row
 
   // check single field
   if(table.empty())
      return QError(QERR_DB_NULL_RESULT, __FILE__, __LINE__,
                    "No fields returned in "
                    + funcName
                    + ". Query was: \""
                    + query + "\"");
   if(table.size() > 1)
      return QError(QERR_UNDEFINED_STATUS, __FILE__, __LINE__,
                    "Multiple fields returned in "
                    + funcName
                    + ". Query was: \""
                    + query + "\"");
 
   // check single row
   const column& col = table.begin()->second;
   if(col.empty())
      return QError(QERR_DB_NULL_RESULT, __FILE__, __LINE__,
                    "No rows returned in "
                    + funcName
                    + ". Query was: \""
                    + query + "\"");
   if(col.size() > 1)
      return QError(QERR_DB_MAXLINES_EXCEEDED, file, line,
                    "Multiple rows returned in "
                    + funcName
                    + ". Query was: \""
                    + query + "\"");
   
   return QERR_SUCCESS;
}
 
 
//_____________________________________________________________________________
void QDb::RandomWaitBeforeConnect() const {
   if(fMinConnectDelaySec < 0
      || fMaxConnectDelaySec < 0
      || fMinConnectDelaySec > fMaxConnectDelaySec)
      return;
 
   // the if check allows fMaxConnectDelaySec == fMinConnectDelaySec,
   // however the algorithm below handles this limit case properly, as
   // randSleep is zero and the sleep time becomes fMinConnectDelaySec.
   
   double intervalSec = fMaxConnectDelaySec - fMinConnectDelaySec;
   int randGranularity = 5000;
   double randSleep
      = intervalSec * (rand() % randGranularity) / randGranularity;
   double sleepTimeSec = fMinConnectDelaySec + randSleep;
 
   int numSecs = (int)sleepTimeSec;
   int numMicroSecs = 1e6*(sleepTimeSec - numSecs);
   sleep(numSecs);
   usleep(numMicroSecs);
   return;
}
 
 
//_____________________________________________________________________________
QError QDb::InitWatchdogThread()
{
   if(fDisconnectTimeoutSec <= 0)
      return QERR_SUCCESS;
   
   if(NULL == fThreadHandler)
   {
      fThreadHandler = new QDbWatchdogThreadHandler();
      fWatchdogTimer = new QWatchdogTimer();
 
      fThreadHandler->SetDatabase(this);
      fWatchdogTimer->SetTimeoutSec(fDisconnectTimeoutSec);
      fThreadHandler->SetWatchdogTimer(fWatchdogTimer);
      QError err = fThreadHandler->Start();
      if(err != QERR_SUCCESS)
      {
         delete fThreadHandler;
         delete fWatchdogTimer;
         fThreadHandler = NULL;
         fWatchdogTimer = NULL;
         return err;
      }
   }
   return QERR_SUCCESS;
}
 
 
//_____________________________________________________________________________
QError QDb::UpdateWatchdogTimer()
{
   if(NULL == fThreadHandler)
      return QERR_SUCCESS;
 
   QError err = QERR_SUCCESS;
   err = fWatchdogTimer->Start();
   if(err != QERR_SUCCESS)
      return err;
 
   bool isRunning = false;
   err = fThreadHandler->Check(isRunning, 0);
   if(err != QERR_SUCCESS)
      return err;
 
   if( ! isRunning )
      return QError(QERR_UNDEFINED_STATUS, __FILE__, __LINE__,
                    "Watchdog timer thread died unexpectedly");
 
   return QERR_SUCCESS;
}