src/tools/csqlserver.cxx

   1 /***************************************************************************
   2  *   Copyright (C) 2007 by www.databasecache.com                           *
   3  *   Contact: praba_tuty@databasecache.com                                 *
   4  *                                                                         *
   5  *   This program is free software; you can redistribute it and/or modify  *
   6  *   it under the terms of the GNU General Public License as published by  *
   7  *   the Free Software Foundation; either version 2 of the License, or     *
   8  *   (at your option) any later version.                                   *
   9  *                                                                         *
  10  *   This program is distributed in the hope that it will be useful,       *
  11  *   but WITHOUT ANY WARRANTY; without even the implied warranty of        *
  12  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the         *
  13  *   GNU General Public License for more details.                          *
  14  *                                                                         *
  15   ***************************************************************************/
  16 #include<os.h>
  17 #include<CSql.h>
  18 #include<SessionImpl.h>
  19 #include<Debug.h>
  20 #include<Process.h>
  21 #include<Database.h>
  22 #include<Transaction.h>
  23 #include<Lock.h>
  24 #include<CacheTableLoader.h>
  25 char* version = "csql-linux-i686-3.0GA";
  26 int srvStop =0;
  27 pid_t asyncpid=0;
  28 pid_t sqlserverpid=0;
  29 pid_t cachepid=0;
  30 pid_t chkptpid=0;
  31 bool recoverFlag=false;
  32 bool monitorServer= false;
  33 SessionImpl *session = NULL;
  34 static void sigTermHandler(int sig)
  35 {
  36     printf("Received signal %d\nStopping the server\n", sig);
  37     srvStop = 1;
  38     monitorServer=false;
  39 }
  40 static void sigChildHandler(int sig)
  41 {
  42     os::signal(SIGCHLD, sigChildHandler);
  43     int stat;
  44     waitpid(-1, &stat, WNOHANG);
  45     //TODO::move waitpid to os wrapper
  46 }
  47
  48 bool checkDead(pid_t  pid)
  49 {
  50     int ret = os::kill(pid, 0);
  51     if (ret == -1) {
  52        if (errno == EPERM)
  53           printError(ErrWarning, "No permission to check process %d is alive.");
  54        else
  55           return true;
  56     }
  57     return false;
  58 }
  59
  60 DbRetVal releaseAllResources(Database *sysdb, ThreadInfo *info )
  61 {
  62     printf("Releasing all the resources for process %d %lu\n", info->pid_, info->thrid_);
  63     //recover for all the mutexes in has_
  64     for (int i =0;  i < MAX_MUTEX_PER_THREAD; i++)
  65     {
  66         if (info->has_[i] != NULL)
  67         {
  68             printf("Dead Procs: %d %lu holding mutex %x %s \n", info->pid_, info->thrid_, info->has_[i], info->has_[i]->name);
  69             logFine(Conf::logger, "Dead Procs: %d %lu holding mutex %x %s \n", info->pid_, info->thrid_, info->has_[i], info->has_[i]->name);
  70             //TODO::recovery of mutexes
  71             sysdb->recoverMutex(info->has_[i]);
  72             //srvStop = 1;
  73             //return ErrSysFatal;
  74         }
  75     }
  76     TransactionManager *tm = new TransactionManager();
  77     LockManager *lm = new LockManager(sysdb);
  78     if (info->thrTrans_.trans_ != NULL && info->thrTrans_.trans_->status_ == TransRunning)
  79     {
  80         printf("Rollback Transaction %x\n", info->thrTrans_.trans_);
  81         tm->rollback(lm, info->thrTrans_.trans_);
  82         info->thrTrans_.trans_->status_ = TransNotUsed;
  83     }
  84     info->init();
  85     delete tm;
  86     delete lm;
  87     return OK;
  88 }
  89
  90 DbRetVal cleanupDeadProcs(Database *sysdb)
  91 {
  92     DbRetVal rv = sysdb->getProcessTableMutex(false);
  93     if (OK != rv)
  94     {
  95         printError(rv,"Unable to get process table mutex");
  96         return rv;
  97     }
  98     pid_t pid;
  99     pid = os::getpid();
 100     pthread_t thrid = os::getthrid();
 101
 102
 103     ThreadInfo* tInfo = sysdb->getThreadInfo(0);
 104     int i=0;
 105     ThreadInfo* freeSlot = NULL;
 106     for (; i < Conf::config.getMaxProcs(); i++)
 107     {
 108         //check whether it is alive
 109         if (tInfo->pid_ !=0 && checkDead(tInfo->pid_)) releaseAllResources(sysdb, tInfo);
 110         tInfo++;
 111     }
 112     sysdb->releaseProcessTableMutex(false);
 113     return OK;
 114 }
 115
 116
 117 DbRetVal logActiveProcs(Database *sysdb)
 118 {
 119     DbRetVal rv = sysdb->getProcessTableMutex(false);
 120     if (OK != rv)
 121     {
 122         printError(rv,"Unable to get process table mutex");
 123         return rv;
 124     }
 125     ThreadInfo* tInfo = sysdb->getThreadInfo(0);
 126     int i=0, count =0;
 127     ThreadInfo* freeSlot = NULL;
 128     for (; i < Conf::config.getMaxProcs(); i++)
 129     {
 130         if (tInfo->pid_ !=0 ) {
 131            logFine(Conf::logger, "Registered Procs: %d %lu\n", tInfo->pid_, tInfo->thrid_);
 132            printf("Client process with pid %d is still registered\n", tInfo->pid_);
 133            if( tInfo->pid_ != asyncpid && tInfo->pid_ != cachepid &&
 134                tInfo->pid_ != sqlserverpid)
 135                count++;
 136         }
 137         tInfo++;
 138     }
 139     sysdb->releaseProcessTableMutex(false);
 140     if (count) return ErrSysInternal; else return OK;
 141 }
 142 void startCacheServer()
 143 {
 144      char execName[1024];
 145      sprintf(execName, "%s/bin/csqlcacheserver", os::getenv("CSQL_INSTALL_ROOT"));
 146      if (srvStop) return;
 147      //printf("filename is %s\n", execName);
 148      cachepid = os::createProcess(execName, "csqlcacheserver");
 149      if (cachepid != -1)
 150      printf("Cache Receiver Started\t   [PID=%d]\n",cachepid);
 151      return;
 152 }
 153
 154 void startServiceClient()
 155 {
 156      char execName[1024];
 157      sprintf(execName, "%s/bin/csqlsqlserver", os::getenv("CSQL_INSTALL_ROOT"));
 158      //printf("filename is %s\n", execName);
 159      if (srvStop) return;
 160      sqlserverpid = os::createProcess(execName, "csqlsqlserver");
 161      if (sqlserverpid != -1)
 162      printf("Network Server Started\t   [PID=%d] [PORT=%d]\n", sqlserverpid,Conf::config.getPort());
 163
 164      return;
 165 }
 166
 167 void startAsyncServer()
 168 {
 169      char execName[1024];
 170      sprintf(execName, "%s/bin/csqlasyncserver", os::getenv("CSQL_INSTALL_ROOT"));
 171      //printf("filename is %s\n", execName);
 172      if (srvStop) return;
 173      asyncpid = os::createProcess(execName, "csqlasyncserver");
 174      if (asyncpid != -1)
 175          printf("Async Cache Server Started [PID=%d]\n", asyncpid);
 176      return;
 177 }
 178 void startCheckpointServer()
 179 {
 180     char execName[1024];
 181     sprintf(execName, "%s/bin/csqlcheckpointserver", os::getenv("CSQL_INSTALL_ROOT"));
 182     if (srvStop) return;
 183     chkptpid = os::createProcess(execName, "csqlcheckpointserver");
 184     if (chkptpid != -1) {
 185         printf("Checkpoint Server Started [PID=%d]\n", chkptpid);
 186         logFine(Conf::logger, "Checkpoint Server Started pid:%d", chkptpid);
 187     }
 188     return;
 189 }
 190
 191
 192
 193 void printUsage()
 194 {
 195    printf("Usage: csqlserver [-c] [-v]\n");
 196    printf("       v -> print the version.\n");
 197    printf("       c -> recover all cached tables from the target database.\n");
 198    printf("Description: Start the csql server and initialize the database.\n");
 199    return;
 200 }
 201 int main(int argc, char **argv)
 202 {
 203     int c = 0,opt = 0;
 204     char cmd[1024];
 205     while ((c = getopt(argc, argv, "cv?")) != EOF)
 206     {
 207         switch (c)
 208         {
 209             case '?' : { opt = 10; break; } //print help
 210             case 'c' : { opt = 1; break; } //recover all the tables from cache
 211             case 'v' : { opt = 2; break; } //print version
 212             default: opt=10;
 213
 214         }
 215     }//while options
 216
 217     if (opt == 10) {
 218         printUsage();
 219         return 0;
 220     }else if (opt ==2) {
 221         printf("%s\n",version);
 222         return 0;
 223     }
 224     session = new SessionImpl();
 225     DbRetVal rv = session->readConfigFile();
 226     if (rv != OK)
 227     {
 228         printf("Unable to read the configuration file \n");
 229         return 1;
 230     }
 231     os::signal(SIGINT, sigTermHandler);
 232     os::signal(SIGTERM, sigTermHandler);
 233     os::signal(SIGCHLD, sigChildHandler);
 234     rv = Conf::logger.startLogger(Conf::config.getLogFile(), true);
 235     if (rv != OK)
 236     {
 237         printf("Unable to start the Conf::logger\n");
 238         return 2;
 239     }
 240     bool isInit = true;
 241     logFine(Conf::logger, "Server Started");
 242     int ret  = session->initSystemDatabase();
 243     if (0  != ret)
 244     {
 245         //printf(" System Database Initialization failed\n");
 246         printf("Attaching to exising database\n");
 247         isInit = false;
 248         delete session;
 249         session = new SessionImpl();
 250         ret = session->open(DBAUSER, DBAPASS);
 251         if (ret !=0) {
 252             printf("Unable to attach to existing database\n");
 253             return 3;
 254         }
 255     }
 256     bool end = false;
 257     struct timeval timeout, tval;
 258     timeout.tv_sec = 5;
 259     timeout.tv_usec = 0;
 260     Database* sysdb = session->getSystemDatabase();
 261     recoverFlag = false;
 262
 263     GlobalUniqueID UID;
 264     if (isInit) UID.create();
 265
 266     if(isInit && Conf::config.useDurability())
 267     {
 268         char dbRedoFileName[MAX_FILE_LEN];
 269         char dbChkptSchema[MAX_FILE_LEN];
 270         char dbChkptMap[MAX_FILE_LEN];
 271         char dbChkptData[MAX_FILE_LEN];
 272         char dbBackupFile[MAX_FILE_LEN];
 273
 274         //check for check point file if present recover
 275         sprintf(dbChkptSchema, "%s/db.chkpt.schema1", Conf::config.getDbFile());
 276         if (FILE *file = fopen(dbChkptSchema, "r")) {
 277             fclose(file);
 278             sprintf(cmd, "cp -f %s %s/db.chkpt.schema", dbChkptSchema, Conf::config.getDbFile());
 279             int ret = system(cmd);
 280             if (ret != 0) {
 281                 Conf::logger.stopLogger();
 282                 session->destroySystemDatabase();
 283                 delete session;
 284                 return 20;
 285             }
 286         }
 287         sprintf(dbChkptMap, "%s/db.chkpt.map1", Conf::config.getDbFile());
 288         if (FILE *file = fopen(dbChkptMap, "r")) {
 289             fclose(file);
 290             sprintf(cmd, "cp -f %s %s/db.chkpt.map", dbChkptMap, Conf::config.getDbFile());
 291             int ret = system(cmd);
 292             if (ret != 0) {
 293                 Conf::logger.stopLogger();
 294                 session->destroySystemDatabase();
 295                 delete session;
 296                 return 30;
 297             }
 298         }
 299         int chkptID= Database::getCheckpointID();
 300         sprintf(dbChkptData, "%s/db.chkpt.data%d", Conf::config.getDbFile(),
 301                                                    chkptID);
 302         sprintf(dbBackupFile, "%s/db.chkpt.data1", Conf::config.getDbFile());
 303         FILE *fl = NULL;
 304         if (!Conf::config.useMmap() && (fl = fopen(dbBackupFile, "r"))) {
 305             fclose(fl);
 306             sprintf(cmd, "cp %s/db.chkpt.data1 %s", Conf::config.getDbFile(), dbChkptData);
 307             int ret = system(cmd);
 308             if (ret != 0) {
 309                 printError(ErrOS, "Unable to take backup for chkpt data file");
 310                 return 40;
 311             }
 312         }
 313         if (FILE *file = fopen(dbChkptData, "r")) {
 314             fclose(file);
 315             int ret = system("recover");
 316             if (ret != 0) {
 317                 printf("Recovery failed\n");
 318                 Conf::logger.stopLogger();
 319                 session->destroySystemDatabase();
 320                 delete session;
 321                 return 50;
 322             }
 323         }
 324
 325         //check for redo log file if present apply redo logs
 326         sprintf(dbRedoFileName, "%s/csql.db.cur", Conf::config.getDbFile());
 327         if (FILE *file = fopen(dbRedoFileName, "r"))
 328         {
 329             fclose(file);
 330             int ret = system("redo -a");
 331             if (ret != 0) {
 332                 printf("Recovery failed. Redo log file corrupted\n");
 333                 Conf::logger.stopLogger();
 334                 session->destroySystemDatabase();
 335                 delete session;
 336                 return 60;
 337             }
 338
 339             // take check point at this moment
 340             DatabaseManager *dbMgr = session->getDatabaseManager();
 341             rv = dbMgr->checkPoint();
 342             if (rv != OK)
 343             {
 344                 printError(ErrSysInternal, "checkpoint failed after redo log apply");
 345                 Conf::logger.stopLogger();
 346                 session->destroySystemDatabase();
 347                 delete session;
 348                 return 70;
 349             }
 350         }
 351     }
 352     bool isCacheReq = false, isSQLReq= false, isAsyncReq=false, isChkptReq=false;
 353     recoverFlag = true;
 354     if (opt == 1 && isInit && ! Conf::config.useDurability()) {
 355         if (Conf::config.useCache()) {
 356             printf("Database server recovering cached tables...\n");
 357             int ret = system("cachetable -R");
 358             if (ret != 0) {
 359                 printf("Cached Tables recovery failed %d\n", ret);
 360                 Conf::logger.stopLogger();
 361                 session->destroySystemDatabase();
 362                 delete session;
 363                 return 2;
 364             }
 365             printf("Cached Tables recovered\n");
 366         } else {
 367             printf("Cache mode is not set in csql.conf. Cannot recover\n");
 368             Conf::logger.stopLogger();
 369             session->destroySystemDatabase();
 370             delete session;
 371             return 1;
 372         }
 373     }
 374     //TODO:: kill all the child servers and restart if !isInit
 375
 376     if(Conf::config.useCsqlSqlServer()) {
 377         isSQLReq = true;
 378         startServiceClient();
 379     }
 380     if ( (Conf::config.useCache() &&
 381            Conf::config.getCacheMode()==ASYNC_MODE)) {
 382         int msgid = os::msgget(Conf::config.getMsgKey(), 0666);
 383         if (msgid != -1) os::msgctl(msgid, IPC_RMID, NULL);
 384         isAsyncReq = true;
 385         startAsyncServer();
 386     }
 387     if (Conf::config.useCache() && Conf::config.useTwoWayCache()) {
 388         isCacheReq = true;
 389         startCacheServer();
 390     }
 391     if(Conf::config.useDurability())
 392     {
 393         isChkptReq = true;
 394         startCheckpointServer();
 395     }
 396
 397     printf("Database Server Started...\n");
 398     logFine(Conf::logger, "Database Server Started");
 399     monitorServer= Conf::config.useMonitorServers();
 400
 401 reloop:
 402     while(!srvStop)
 403     {
 404         tval.tv_sec = timeout.tv_sec;
 405         tval.tv_usec = timeout.tv_usec;
 406         os::select(0, 0, 0, 0, &tval);
 407
 408         //send signal to all the registered process..check they are alive
 409         cleanupDeadProcs(sysdb);
 410         if (srvStop) break;
 411         if (monitorServer) {
 412           if (isCacheReq && cachepid !=0  && checkDead(cachepid)) {
 413             logFine(Conf::logger, "Cache Receiver Died pid:%d", cachepid);
 414             startCacheServer();
 415           }
 416           if (isAsyncReq && asyncpid !=0  && checkDead(asyncpid)) {
 417             logFine(Conf::logger, "Async Server Died pid:%d", asyncpid);
 418             int msgid = os::msgget(Conf::config.getMsgKey(), 0666);
 419             if (msgid != -1) os::msgctl(msgid, IPC_RMID, NULL);
 420             startAsyncServer();
 421           }
 422           if (isSQLReq && sqlserverpid !=0  && checkDead(sqlserverpid)) {
 423             logFine(Conf::logger, "Network Server Died pid:%d", sqlserverpid);
 424             startServiceClient();
 425           }
 426           if (isChkptReq && chkptpid !=0  && checkDead(chkptpid)) {
 427             logFine(Conf::logger, "Checkpoint Server Died pid:%d", chkptpid);
 428             startCheckpointServer();
 429           }
 430        }
 431     }
 432     if (logActiveProcs(sysdb) != OK) {srvStop = 0;
 433         monitorServer= Conf::config.useMonitorServers();
 434         goto reloop;
 435     }
 436     if (cachepid) os::kill(cachepid, SIGTERM);
 437     if(asyncpid) os::kill(asyncpid, SIGTERM);
 438     if (sqlserverpid) os::kill(sqlserverpid, SIGTERM);
 439     if (chkptpid) os::kill(chkptpid, SIGTERM);
 440     if (Conf::config.useDurability() && Conf::config.useMmap()) {
 441         //ummap the memory
 442         char *startAddr = (char *) sysdb->getMetaDataPtr();
 443         msync(startAddr + Conf::config.getMaxSysDbSize(),Conf::config.getMaxDbSize(), MS_SYNC);
 444         munmap(startAddr + Conf::config.getMaxSysDbSize(), Conf::config.getMaxDbSize());
 445     }
 446     logFine(Conf::logger, "Server Exiting");
 447     printf("Server Exiting\n");
 448     logFine(Conf::logger, "Server Ended");
 449     UID.destroy();
 450     session->destroySystemDatabase();
 451     Conf::logger.stopLogger();
 452     delete session;
 453     return 0;
 454 }