src/tools/csqlcacheserver.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 <AbsSqlConnection.h>
  17 #include <AbsSqlStatement.h>
  18 #include <SqlOdbcStatement.h>
  19 #include <SqlFactory.h>
  20 #include <CSql.h>
  21 #include <CacheTableLoader.h>
  22
  23 int insert(Table *table, int pkid);
  24 int remove(Table *table, int pkid);
  25 int getRecordsFromTargetDb(int mode);
  26 void createCacheTableList();
  27 DbRetVal getCacheField(char *tblName,char *fldName);
  28 DbRetVal getCacheProjField(char *tblName,char *fielflist);
  29 DbRetVal getCacheCondition(char *tblName,char *condition);
  30 List cacheTableList;
  31 int srvStop =0;
  32 static void sigTermHandler(int sig)
  33 {
  34     printf("Received signal %d\nStopping the server\n", sig);
  35     srvStop = 1;
  36 }
  37
  38 void printUsage()
  39 {
  40    printf("Usage: csqlcacheserver \n");
  41    printf("Description: Start the csql caching server.\n");
  42    return;
  43 }
  44 AbsSqlConnection *targetconn;
  45 Connection conn;
  46 int main(int argc, char **argv)
  47 {
  48     int c = 0, opt = 0;
  49     while ((c = getopt(argc, argv, "?")) != EOF)
  50     {
  51         switch (c)
  52         {
  53             case '?' : { opt = 10; break; } //print help
  54             default: opt=10;
  55
  56         }
  57     }//while options
  58
  59     if (opt == 10) {
  60         printUsage();
  61         return 0;
  62     }
  63
  64     os::signal(SIGINT, sigTermHandler);
  65     os::signal(SIGTERM, sigTermHandler);
  66     DbRetVal rv = conn.open("root", "manager");
  67     if (rv != OK) return 1;
  68     targetconn = SqlFactory::createConnection(CSqlAdapter);
  69     rv = targetconn->connect("root", "manager");
  70     if (rv != OK) return 1;
  71     if (!Conf::config.useCache())
  72     {
  73         printf("Cache is set to OFF in csql.conf file\n");
  74         return 1;
  75     }
  76     AbsSqlStatement *stmt = SqlFactory::createStatement(CSqlAdapter);
  77     stmt->setConnection(targetconn);
  78     /*rv = stmt->prepare("create table csql_log_int(tablename char(64), pkid int, operation int, id int not null unique auto_increment)engine='innodb';");
  79     targetconn->beginTrans();
  80     int rows=0;
  81     stmt->execute(rows);
  82     targetconn->commit();
  83     stmt->free();
  84     delete stmt;*/
  85
  86     printf("Cache server started\n");
  87     int ret = 0;
  88     struct stat ofstatus,nfstatus;
  89     ret=stat(Conf::config.getTableConfigFile(),&ofstatus);
  90     struct timeval timeout, tval;
  91     timeout.tv_sec = Conf::config.getCacheWaitSecs();
  92     timeout.tv_usec = 0;
  93     createCacheTableList();
  94     while(!srvStop)
  95     {
  96         tval.tv_sec = timeout.tv_sec;
  97         tval.tv_usec = timeout.tv_usec;
  98         ret = os::select(0, NULL, 0, 0, &tval);
  99         printf("Checking for cache updates\n");
 100         ret=stat(Conf::config.getTableConfigFile(),&nfstatus);
 101         if(ofstatus.st_mtime != nfstatus.st_mtime)
 102         {
 103             cacheTableList.reset();
 104             createCacheTableList();
 105             ofstatus.st_mtime = nfstatus.st_mtime;
 106         }
 107         ret = getRecordsFromTargetDb(1);
 108         if (ret !=0) srvStop = 1;
 109         //ret = getRecordsFromTargetDb(2);
 110         if (ret !=0) srvStop = 1;
 111     }
 112     printf("Cache Server Exiting\n");
 113     cacheTableList.reset();
 114     conn.close();
 115     targetconn->disconnect();
 116     return 0;
 117 }
 118 int getRecordsFromTargetDb(int mode)
 119 {
 120     int pkid;
 121     char tablename[64];
 122     int op, id,caId;
 123     int rows =0;
 124     DbRetVal rv = OK;
 125     char StmtStr[1024];
 126     caId =Conf::config.getCacheID();
 127     AbsSqlStatement *stmt = SqlFactory::createStatement(CSqlAdapter);
 128     stmt->setConnection(targetconn);
 129     AbsSqlStatement *delstmt = SqlFactory::createStatement(CSqlAdapter);
 130     delstmt->setConnection(targetconn);
 131     if (mode == 1 ) {
 132         //rv = delstmt->prepare("DELETE from csql_log_int where id=?;");
 133         sprintf(StmtStr, "SELECT * FROM csql_log_int where cacheid = %d;", caId);
 134         rv = stmt->prepare(StmtStr);
 135         if (rv != OK) {printf("Stmt prepare failed\n"); return 1; }
 136     }
 137     else {
 138         rv = stmt->prepare("SELECT * FROM csql_log_char;");
 139         if (rv != OK) {printf("Stmt prepare failed\n"); return 1; }
 140         //rv = delstmt->prepare("DELETE from csql_log_char where id=?;");
 141     }
 142     if (rv != OK) {printf("Stmt prepare failed\n"); return 1; }
 143     stmt->bindField(1, tablename);
 144     stmt->bindField(2, &pkid);
 145     stmt->bindField(3, &op);
 146     stmt->bindField(4, &caId);
 147     stmt->bindField(5, &id);
 148
 149     DatabaseManager *dbMgr = conn.getDatabaseManager();
 150     while(true) {
 151       rv = targetconn->beginTrans();
 152       rv = stmt->execute(rows);
 153       if (rv != OK)
 154       {
 155           printError(ErrSysInit, "Unable to execute stmt in target db");
 156           targetconn->rollback();
 157           stmt->free();
 158           delstmt->free();
 159           delete stmt;
 160           delete delstmt;
 161           return 1;
 162       }
 163       if (stmt->fetch() != NULL) {
 164           printf("Row value is %s %d %d %d\n", tablename, pkid, op,caId);
 165
 166                 Table *table = dbMgr->openTable(tablename);
 167                 int ret = 0;
 168                 if (table == NULL)
 169                 {
 170                         printError(ErrSysInit, "Table %s not exist in csql", tablename);
 171                         targetconn->rollback();
 172                         stmt->free();
 173                         delstmt->free();
 174                         delete stmt;
 175                         delete delstmt;
 176                         break;
 177                 }
 178                 if (op == 2)//DELETE
 179                 {
 180                         ret = remove(table,pkid);
 181                 }
 182                 else //INSERT
 183                 {
 184                         ret = insert(table, pkid);
 185                 }
 186                 dbMgr->closeTable(table);
 187                 rv = targetconn->commit();
 188                 rv = targetconn->beginTrans();
 189           //Remove record from csql_log_XXX table
 190                 sprintf(StmtStr, "DELETE from csql_log_int where id=%d ;", id);
 191                 rv = delstmt->prepare(StmtStr);
 192                 if (rv != OK) {printf("FAILED\n"); return 1; }
 193                 // delstmt->setIntParam(1, id);
 194                 rv = delstmt->execute(rows);
 195                 if (rv != OK)
 196                 {
 197                         printf("log record not deleted from the target db %d\n", rv);
 198                         targetconn->rollback();
 199                         stmt->free();
 200                         delstmt->free();
 201                         delete stmt;
 202                         delete delstmt;
 203                 }
 204                 delstmt->free();
 205
 206                 rv = targetconn->commit();
 207
 208      }
 209      else {
 210          stmt->close();
 211          break;
 212      }
 213      stmt->close();
 214    }
 215    stmt->free();
 216    delstmt->free();
 217    delete stmt;
 218    delete delstmt;
 219    return 0;
 220 }
 221 int insert(Table *table, int pkid)
 222 {
 223     AbsSqlStatement *stmt = SqlFactory::createStatement(CSqlAdapter);
 224     stmt->setConnection(targetconn);
 225     SqlOdbcStatement *ostmt = (SqlOdbcStatement*) stmt;
 226     char pkfieldname[128];
 227     DbRetVal rv=getCacheField(table->getName(), pkfieldname);
 228     if(rv!=OK){
 229         ostmt->getPrimaryKeyFieldName(table->getName(), pkfieldname);
 230     }
 231     char fieldlist[IDENTIFIER_LENGTH];
 232     char condition[IDENTIFIER_LENGTH];
 233     char sbuf[1024];
 234     rv=getCacheProjField(table->getName(),fieldlist);
 235     if(rv!=OK){
 236         rv=getCacheCondition(table->getName(),condition);
 237         if(rv!=OK){
 238             sprintf(sbuf, "SELECT * FROM %s where %s = %d;", table->getName(), pkfieldname, pkid);
 239         } else {
 240             sprintf(sbuf, "SELECT * FROM %s where %s = %d and %s ;", table->getName(), pkfieldname, pkid,condition);
 241         }
 242     } else {
 243         rv=getCacheCondition(table->getName(),condition);
 244         if(rv!=OK){
 245             sprintf(sbuf, "SELECT %s FROM %s where %s = %d;",fieldlist,table->getName(), pkfieldname, pkid);
 246         } else {
 247             sprintf(sbuf, "SELECT %s FROM %s where %s = %d and %s;",fieldlist,table->getName(), pkfieldname, pkid,condition);
 248         }
 249     }
 250     //TODO::get the primary key field name from the table interface. need to implement it
 251     rv = stmt->prepare(sbuf);
 252     if (rv != OK) return 1;
 253
 254     List fNameList = table->getFieldNameList();
 255     ListIterator fNameIter = fNameList.getIterator();
 256     FieldInfo *info = new FieldInfo();
 257     int fcount =1; void *valBuf; int fieldsize=0;
 258     void *buf[128];//TODO:resticts to support only 128 fields in table
 259     Identifier *elem = NULL;
 260     while (fNameIter.hasElement()) {
 261         elem = (Identifier*) fNameIter.nextElement();
 262         table->getFieldInfo((const char*)elem->name, info);
 263         valBuf = AllDataType::alloc(info->type, info->length);
 264         buf[fcount] = valBuf;
 265         table->bindFld(elem->name, valBuf);
 266         stmt->bindField(fcount++, valBuf);
 267
 268     }
 269     delete info;
 270     int rows=0;
 271     int retValue = stmt->execute(rows);
 272     if (retValue && rows != 1) {printError(ErrSysInit, "Unable to execute statement at target db\n"); return ErrSysInit; }
 273     conn.startTransaction();
 274     if (stmt->fetch() != NULL) {
 275         ostmt->setNullInfo(table);
 276         table->insertTuple();
 277         //Note:insert may fail if the record is inserted from this cache
 278     }
 279     for (int i=1; i < fcount; i++) {
 280         free(buf[i]);
 281     }
 282     stmt->free();
 283     delete stmt;
 284     conn.commit();
 285     return 0;
 286 }
 287 int remove(Table *table, int pkid)
 288 {
 289     DbRetVal rv = OK;
 290     AbsSqlStatement *stmt = SqlFactory::createStatement(CSqlAdapter);
 291     stmt->setConnection(targetconn);
 292     SqlOdbcStatement *ostmt = (SqlOdbcStatement*) stmt;
 293     char pkfieldname[128];
 294     rv=getCacheField(table->getName(), pkfieldname);
 295     if(rv!=OK){
 296         ostmt->getPrimaryKeyFieldName(table->getName(), pkfieldname);
 297     }
 298     delete stmt;
 299     Condition p1;
 300     p1.setTerm(pkfieldname, OpEquals, &pkid);
 301     table->setCondition(&p1);
 302     rv = conn.startTransaction();
 303     if (rv != OK) return 1;
 304     rv = table->execute();
 305     if (rv != OK)
 306     {
 307         table->setCondition(NULL);
 308         conn.rollback();
 309         return 1;
 310     }
 311     if (table->fetch() != NULL)
 312         rv = table->deleteTuple();
 313         //Note:Delete may fail if the record is deleted from this cache
 314     table->setCondition(NULL);
 315     rv = conn.commit();
 316     if (rv != OK) return 1;
 317     return 0;
 318 }
 319 void createCacheTableList()
 320 {
 321     FILE *fp;
 322     fp = fopen(Conf::config.getTableConfigFile(),"r");
 323     if( fp == NULL ) {
 324         printError(ErrSysInit, "cachetable.conf file does not exist");
 325         fclose(fp);
 326     }
 327     char tablename[IDENTIFIER_LENGTH];
 328     char fieldname[IDENTIFIER_LENGTH];
 329     char condition[IDENTIFIER_LENGTH];
 330     char field[IDENTIFIER_LENGTH];
 331     int mode;
 332     while(!feof(fp))
 333     {
 334         fscanf(fp,"%d:%s %s %s %s\n",&mode,tablename,fieldname,condition,field);
 335         CacheTableInfo  *cacheTable=new CacheTableInfo();
 336         cacheTable->setTableName(tablename);
 337         cacheTable->setFieldName(fieldname);
 338         cacheTable->setProjFieldList(field);
 339         cacheTable->setCondition(condition);
 340         cacheTableList.append(cacheTable);
 341     }
 342    // printf("Table %s is not cached\n",tabname);
 343     fclose(fp);
 344
 345 }
 346
 347 DbRetVal getCacheCondition(char *tblName,char *condition)
 348 {
 349     ListIterator iter=cacheTableList.getIterator();
 350     CacheTableInfo  *cacheTable;
 351     while(iter.hasElement())
 352     {
 353         cacheTable=(CacheTableInfo*)iter.nextElement();
 354         if(strcmp(cacheTable->getTableName(),tblName)==0){
 355             if(strcmp(cacheTable->getCondition(),"NULL")!=0)
 356             {
 357                 strcpy(condition,cacheTable->getCondition());
 358                 return OK;
 359             }
 360         }
 361     }
 362     return ErrNotExists;
 363 }
 364
 365 DbRetVal getCacheProjField(char *tblName,char *fieldlist)
 366 {
 367     ListIterator iter=cacheTableList.getIterator();
 368     CacheTableInfo  *cacheTable;
 369     while(iter.hasElement())
 370     {
 371         cacheTable=(CacheTableInfo*)iter.nextElement();
 372         if(strcmp(cacheTable->getTableName(),tblName)==0){
 373             if(strcmp(cacheTable->getProjFieldList(),"NULL")!=0)
 374             {
 375                 strcpy(fieldlist,cacheTable->getProjFieldList());
 376                 return OK;
 377             }
 378         }
 379     }
 380     return ErrNotExists;
 381 }
 382 DbRetVal getCacheField(char *tblName,char *fldName)
 383 {
 384     ListIterator iter=cacheTableList.getIterator();
 385     CacheTableInfo  *cacheTable;
 386     while(iter.hasElement())
 387     {
 388         cacheTable=(CacheTableInfo*)iter.nextElement();
 389         if(strcmp(cacheTable->getTableName(),tblName)==0){
 390             if(strcmp(cacheTable->getFieldName(),"NULL")!=0)
 391             {
 392                 strcpy(fldName,cacheTable->getFieldName());
 393                 return OK;
 394             }
 395         }
 396
 397     }
 398     return ErrNotExists;
 399 }