include/SqlLogConnection.h

   1 /***************************************************************************
   2  *   Copyright (C) 2007 by Prabakaran Thirumalai   *
   3  *   praba_tuty@yahoo.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  *   You should have received a copy of the GNU General Public License     *
  16  *   along with this program; if not, write to the                         *
  17  *   Free Software Foundation, Inc.,                                       *
  18  *   59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.             *
  19  ***************************************************************************/
  20 #ifndef SQLLOGCONNECTION_H
  21 #define SQLLOGCONNECTION_H
  22 #include<CSql.h>
  23 #include<SqlFactory.h>
  24 #include<Util.h>
  25 #include<Network.h>
  26
  27 /**
  28 * @class SqlLogConnection
  29 *
  30 */
  31
  32 typedef struct my_msgbuffer {
  33     long mtype;
  34     char data[1];
  35 } Message;
  36
  37 class AbsSqlLogSend
  38 {
  39     public:
  40     virtual DbRetVal prepare(int tId, int sId, int len, char *st,
  41                                                               char *tn)=0;
  42     virtual DbRetVal commit(int len, void *data)=0;
  43     virtual DbRetVal free(int txnId, int stmtId)=0;
  44 };
  45
  46 class MsgQueueSend : public AbsSqlLogSend
  47 {
  48     int msgQId;
  49     public:
  50     MsgQueueSend() { msgQId = os::msgget(Conf::config.getMsgKey(), 0666); }
  51     DbRetVal prepare(int tId, int sId, int len, char *stmt, char *tn);
  52     DbRetVal commit(int len, void *data);
  53     DbRetVal free(int txnId, int stmtId);
  54 };
  55
  56 class FileSend : public AbsSqlLogSend
  57 {
  58     int fdRedoLog;
  59     public:
  60     FileSend();
  61     ~FileSend();
  62     DbRetVal openRedoFile();
  63     DbRetVal prepare(int txnId, int stmtId, int len, char *stmt, char*tn);
  64     DbRetVal commit(int len, void *data);
  65     DbRetVal free(int txnId, int stmtId);
  66 };
  67
  68 enum ExecType
  69 {
  70     EXECONLY = 0,
  71     SETPARAM
  72 };
  73
  74 class ExecLogInfo
  75 {
  76     public:
  77     ExecLogInfo() : pos(0), len(0) {}
  78     int stmtId;
  79     ExecType type;
  80     int pos;
  81     DataType dataType;
  82     int len;
  83     int value; //Extendible value as per parameter type size
  84 };
  85
  86 class SqlLogConnection : public AbsSqlConnection
  87 {
  88     Connection dummyConn;
  89
  90     //stores all the sql log packets to be shipped to peers
  91     List logStore;
  92
  93     List execLogStore;
  94     int execLogStoreSize;
  95
  96     //stores all the prepare log packets to be shipped to peers
  97     //as soon as connection is reestablished to cache server
  98     List prepareStore;
  99
 100     //stores all the prepare log packets to be shipped between two
 101     //consecutive commits. Commit() call sends first all the stmts
 102     //prepared during the course and then sends the exec pkts
 103     List curPrepareStore;
 104
 105     //sync mode of the current transaction
 106     TransSyncMode syncMode;
 107
 108     //stores client objects in it for peer
 109     NetworkTable nwTable;
 110     AbsSqlLogSend *msgQSend;
 111     AbsSqlLogSend *fileSend;
 112
 113     GlobalUniqueID txnUID;
 114     static List cacheList;
 115     int txnID;
 116     DbRetVal populateCachedTableList();
 117     public:
 118     SqlLogConnection() {
 119         innerConn = NULL; syncMode = ASYNC;
 120         if (Conf::config.useCache() &&
 121             Conf::config.getCacheMode()==ASYNC_MODE)
 122                msgQSend = new MsgQueueSend();
 123         else msgQSend = NULL;
 124         if (Conf::config.useDurability()) { fileSend = new FileSend(); }
 125         else fileSend = NULL;
 126         txnUID.open();
 127         execLogStoreSize =0;
 128         noMsgLog = false;
 129     }
 130     ~SqlLogConnection();
 131     bool isTableCached(char *name);
 132     bool noMsgLog;
 133     //Note::forced to implement this as it is pure virtual in base class
 134     Connection& getConnObject(){  return dummyConn; }
 135
 136     DbRetVal connect (char *user, char * pass);
 137
 138     DbRetVal disconnect();
 139
 140     DbRetVal commit();
 141
 142     DbRetVal rollback();
 143
 144     DbRetVal beginTrans (IsolationLevel isoLevel, TransSyncMode mode);
 145
 146     DbRetVal msgPrepare(int tId, int sId, int len, char *stmt, char *tname)
 147     {
 148         return msgQSend->prepare(tId, sId, len, stmt, tname);
 149     }
 150     DbRetVal fileLogPrepare(int tId, int sId, int len, char *stmt, char *tname)
 151     {
 152         return fileSend->prepare(tId, sId, len, stmt, tname);
 153     }
 154     DbRetVal commitLogs(int logSize, void *data)
 155     {
 156         int txnId = getTxnID();
 157         if (((Conf::config.useCache() &&
 158              Conf::config.getCacheMode() == ASYNC_MODE)) && !noMsgLog)
 159             msgQSend->commit(logSize, data);
 160         if (Conf::config.useDurability()) fileSend->commit(logSize, data);
 161         return OK;
 162     }
 163     DbRetVal freeLogs(int stmtId)
 164     {
 165         int txnId = getTxnID();
 166         if ( ((Conf::config.useCache() &&
 167              Conf::config.getCacheMode() == ASYNC_MODE)) && !noMsgLog)
 168             msgQSend->free(txnId, stmtId);
 169         if (Conf::config.useDurability()) fileSend->free(txnId, stmtId);
 170         return OK;
 171     }
 172     void addExecLog(ExecLogInfo *info) { execLogStore.append(info); }
 173     void addToExecLogSize(int size){ execLogStoreSize += size; }
 174     int getExecLogStoreSize() { return execLogStoreSize; }
 175     List getExecLogList() { return execLogStore; }
 176     DbRetVal addPacket(BasePacket *pkt);
 177     DbRetVal addPreparePacket(PacketPrepare *pkt);
 178     DbRetVal removePreparePacket(int stmtid);
 179
 180     DbRetVal setSyncMode(TransSyncMode mode);
 181     void setNoMsgLog(bool nmlog) { noMsgLog = nmlog; }
 182     TransSyncMode getSyncMode() { return syncMode; }
 183     int getTxnID() { return txnID; }
 184     DbRetVal connectIfNotConnected() { return nwTable.connectIfNotConnected(); }
 185     DbRetVal sendAndReceive(NetworkPacketType type, char *packet, int length);
 186     friend class SqlFactory;
 187 };
 188
 189 #endif