test/dbapi/Threads/DMLThreadTest.c

   1 //#include <NanoTimer.h>
   2 //TestCase: It creates 5 threads and all threads insert ITERATION records simultanesouly.
   3 //after insertion, it checks whether 50000 records are there in the table
   4 //after that, 5 threads select 10K records each simultanesouly.
   5 //verify that 10K records are selected in each thread.
   6 //The test is repeated with no index, hash index and tree index
   7 #include<CSql.h>
   8 #define THREADS 5
   9 #define ITERATION 25000
  10 void* runInsTest(void *p);
  11 void* runSelTest(void *p);
  12 void* runDelTest(void *p);
  13 int main()
  14 {
  15
  16     Connection conn;
  17     DbRetVal rv = conn.open("root", "manager");
  18     if (rv != OK)
  19     {
  20        printf("Error during connection %d\n", rv);
  21        return -1;
  22     }
  23     DatabaseManager *dbMgr = conn.getDatabaseManager();
  24     if (dbMgr == NULL) { printf("Auth failed\n"); return -1;}
  25     TableDef tabDef;
  26     tabDef.addField("f1", typeInt, 0, NULL, true);
  27 #ifdef VARCHAR
  28     tabDef.addField("f2", typeVarchar, 12);
  29 #else
  30     tabDef.addField("f2", typeString, 12);
  31 #endif
  32     rv = dbMgr->createTable("t1", tabDef);
  33     if (rv != OK) { printf("Table creation failed\n"); return -1; }
  34     HashIndexInitInfo *idxInfo = new HashIndexInitInfo();
  35     strcpy(idxInfo->tableName, "t1");
  36     idxInfo->list.append("f1");
  37 #ifdef HASH
  38     idxInfo->indType = hashIndex;
  39     rv = dbMgr->createIndex("indx1", idxInfo);
  40     if (rv != OK) { printf("Index creation failed\n"); return -1; }
  41 #endif
  42 #ifdef TREE
  43     idxInfo->indType = treeIndex;
  44     //idxInfo->bucketSize = 25;
  45     rv = dbMgr->createIndex("indx1", idxInfo);
  46     if (rv != OK) { printf("Index creation failed\n"); return -1; }
  47 #endif
  48     delete idxInfo;
  49     pthread_t thr[THREADS];
  50     int message[THREADS];
  51     int status;
  52     for (int i=0; i <THREADS; i++) {
  53         message[i] = i;
  54         pthread_create (&thr[i], NULL,
  55                   &runInsTest, (void *) &message[i]);
  56     }
  57     for (int i=0; i <THREADS; i++) {
  58         pthread_join(thr[i], (void**)&status);
  59     }
  60
  61     Table *table = dbMgr->openTable("t1");
  62     if (table == NULL) { printf("Unable to open table\n"); return -1; }
  63     table->setCondition(NULL);
  64     rv = conn.startTransaction();
  65     if (rv != OK) while (rv !=OK) rv = conn.startTransaction();
  66     table->execute();
  67     int count=0;
  68     void *tuple = NULL;
  69     while ((tuple = (char*) table->fetch())) {
  70          count++;
  71     }
  72     printf("Tuples found: %d\n", count);
  73     table->closeScan();
  74     conn.commit();
  75
  76 #ifdef HASH
  77     Condition p1;
  78     int valTerm = 0;
  79     p1.setTerm("f1", OpEquals, &valTerm);
  80     table->setCondition(&p1);
  81     int icount=0;
  82     for(int i = 0; i< THREADS * ITERATION; i++)
  83     {
  84         rv = conn.startTransaction();
  85         if (rv != OK) while (rv !=OK) rv = conn.startTransaction();
  86         valTerm = i;
  87         rv = table->execute();
  88         tuple = (char*)table->fetch() ;
  89         if (tuple == NULL)  break;
  90         icount++;
  91         table->closeScan();
  92         conn.commit();
  93         //if (i%ITERATION == 0) printf("%d\n", i);
  94     }
  95     printf("Index Tuples found: %d\n", icount);
  96 #endif
  97
  98
  99
 100     dbMgr->closeTable(table);
 101
 102     for (int i=0; i <THREADS; i++) {
 103         message[i] = i;
 104         pthread_create (&thr[i], NULL,
 105                   &runSelTest, (void *) &message[i]);
 106     }
 107     for (int i=0; i <THREADS; i++) {
 108         pthread_join(thr[i], (void**)&status);
 109     }
 110
 111     for (int i=0; i <THREADS; i++) {
 112         message[i] = i;
 113         pthread_create (&thr[i], NULL,
 114                   &runDelTest, (void *) &message[i]);
 115     }
 116     for (int i=0; i <THREADS; i++) {
 117         pthread_join(thr[i], (void**)&status);
 118     }
 119
 120
 121     dbMgr->dropTable("t1");
 122
 123     conn.close();
 124     return 0;
 125 }
 126 void* runInsTest(void *message)
 127 {
 128     Connection conn;
 129     DbRetVal rv = conn.open("root", "manager");
 130     if (rv != OK)
 131     {
 132        printf("Error during connection %d\n", rv);
 133        return NULL;
 134     }
 135     DatabaseManager *dbMgr = conn.getDatabaseManager();
 136     if (dbMgr == NULL) { printf("Auth failed\n"); return NULL;}
 137     Table *table = dbMgr->openTable("t1");
 138     if (table == NULL) { printf("Unable to open table\n"); return NULL; }
 139     int id = 0;
 140     char name[20] = "PRABAKARAN";
 141     table->bindFld("f1", &id);
 142     table->bindFld("f2", name);
 143     char *tuple;
 144     int ret;
 145     int i;
 146     int icount =0;
 147     int val = *(int*)message;
 148     int retrycnt=0;
 149     for(i = val * ITERATION; i< (val *ITERATION) +ITERATION; i++)
 150     {
 151         rv = conn.startTransaction();
 152         if (rv != OK) while (rv !=OK) rv = conn.startTransaction();
 153         id= i;
 154         strcpy(name, "KARAN123");
 155         ret = table->insertTuple();
 156         retrycnt=0;
 157         while (ret == ErrLockTimeOut)
 158         {
 159             rv = conn.rollback();
 160             if (rv != OK)  {
 161                 while (rv == ErrLockTimeOut)
 162                 {
 163                     printf("retrying abort\n");
 164                     rv = conn.rollback();
 165                 }
 166             }
 167             if (retrycnt == 3) { ret = 1; break;}
 168             rv = conn.startTransaction();
 169             if (rv != OK) {
 170                 while (rv == ErrLockTimeOut)
 171                 {
 172                     printf("retrying startTransaction\n");
 173                     rv = conn.startTransaction();
 174                 }
 175             }
 176             ret = table->insertTuple();
 177             retrycnt++;
 178             //ret =1; break; //temp to analyse issue
 179         }
 180         if (ret != 0) { printf("RETURNING EARLY: %d with i:%d\n", ret, i); break;}
 181         icount++;
 182         rv = conn.commit();
 183         if (rv != OK) {
 184             printf("COMMIT returned %d\n", rv);
 185             os::usleep(500);
 186             while (rv == ErrLockTimeOut) {
 187                 printf("retrying commit\n");
 188                 rv = conn.commit();
 189             }
 190         }
 191         //if (icount %ITERATION ==0) printf("%d\n", i);
 192     }
 193    char msgBuf[1024];
 194    sprintf(msgBuf,"Insert: Thread %lu %d: Total rows :%d\n",os::getthrid(), val, icount);
 195    os::write(1,msgBuf,strlen(msgBuf));
 196    dbMgr->closeTable(table);
 197    rv  = conn.close();
 198    return NULL;
 199 }
 200 void* runSelTest(void *message)
 201 {
 202     Connection conn;
 203     DbRetVal rv = conn.open("root", "manager");
 204     if (rv != OK)
 205     {
 206        printf("Error during connection %d\n", rv);
 207        return NULL;
 208     }
 209     DatabaseManager *dbMgr = conn.getDatabaseManager();
 210     if (dbMgr == NULL) { printf("Auth failed\n"); return NULL;}
 211     Table *table = dbMgr->openTable("t1");
 212     if (table == NULL) { printf("Unable to open table\n"); return NULL; }
 213     int id = 0;
 214     char name[1020] = "PRABAKARAN";
 215     table->bindFld("f1", &id);
 216     table->bindFld("f2", name);
 217     char *tuple;
 218     int i;
 219     int icount =0;
 220     int val = *(int*)message;
 221
 222     Condition p1;
 223     int valTerm = 0;
 224     p1.setTerm("f1", OpEquals, &valTerm);
 225     table->setCondition(&p1);
 226
 227     for(i = val * ITERATION; i< (val *ITERATION) +ITERATION; i++)
 228     {
 229         rv = conn.startTransaction();
 230         if (rv != OK) while (rv !=OK) rv = conn.startTransaction();
 231         valTerm = i;
 232         rv = table->execute();
 233         if (rv != OK) {
 234             printf("Execute returned %d\n", rv);
 235             os::usleep(500);
 236             int retry=5;
 237             while (rv == ErrLockTimeOut) {
 238                 if (retry == 0) break;
 239                 printf("retrying execute\n");
 240                 rv = table->execute();
 241                 retry--;
 242             }
 243         }
 244         tuple = (char*)table->fetch() ;
 245         if (tuple == NULL)  break;
 246         icount++;
 247         table->closeScan();
 248         rv = conn.commit();
 249         if (rv != OK) {
 250             while (rv == ErrLockTimeOut) {
 251                 printf("retrying commit\n");
 252                 rv = conn.commit();
 253             }
 254         }
 255     }
 256
 257    char msgBuf[1024];
 258    sprintf(msgBuf,"Select: Thread %lu %d: Total rows :%d\n",os::getthrid(), val, icount);
 259    os::write(1,msgBuf,strlen(msgBuf));
 260    dbMgr->closeTable(table);
 261    rv  = conn.close();
 262    return NULL;
 263 }
 264 void* runDelTest(void *message)
 265 {
 266     Connection conn;
 267     DbRetVal rv = conn.open("root", "manager");
 268     if (rv != OK)
 269     {
 270        printf("Error during connection %d\n", rv);
 271        return NULL;
 272     }
 273     DatabaseManager *dbMgr = conn.getDatabaseManager();
 274     if (dbMgr == NULL) { printf("Auth failed\n"); return NULL;}
 275     Table *table = dbMgr->openTable("t1");
 276     if (table == NULL) { printf("Unable to open table\n"); return NULL; }
 277     int id = 0;
 278     char name[20] = "PRABAKARAN";
 279     table->bindFld("f1", &id);
 280     table->bindFld("f2", name);
 281     char *tuple;
 282     int i;
 283     int ret;
 284     int retrycnt=0;
 285     int icount =0;
 286     int val = *(int*)message;
 287
 288     Condition p1;
 289     int valTerm = 0;
 290     p1.setTerm("f1", OpEquals, &valTerm);
 291     table->setCondition(&p1);
 292
 293     for(i = val * ITERATION; i< (val *ITERATION) +ITERATION; i++)
 294     {
 295         rv = conn.startTransaction();
 296         if (rv != OK) while (rv !=OK) rv = conn.startTransaction();
 297         id= i;
 298         valTerm = i;
 299         rv = table->execute();
 300         if (rv != OK) {
 301             printf("Execute returned %d\n", rv);
 302             os::usleep(500);
 303             int retry=5;
 304             while (rv == ErrLockTimeOut) {
 305                 if (retry == 0) { printf("DELETE:execute failed\n"); break;}
 306                 printf("retrying execute\n");
 307                 rv = table->execute();
 308                 retry--;
 309             }
 310         }
 311         tuple = (char*)table->fetch() ;
 312         if (tuple == NULL)  { printf("DELETE:fetch failed\n");break;}
 313         ret = table->deleteTuple();
 314         retrycnt=0;
 315         while (ret == ErrLockTimeOut)
 316         {
 317            table->closeScan();
 318             rv = conn.rollback();
 319             if (rv != OK)  {
 320                 while (rv == ErrLockTimeOut)
 321                 {
 322                     printf("retrying abort\n");
 323                     rv = conn.rollback();
 324                 }
 325             }
 326             if (retrycnt == 10) { ret = 1; break;}
 327             rv = conn.startTransaction();
 328             if (rv != OK) {
 329                 while (rv == ErrLockTimeOut)
 330                 {
 331                     printf("retrying startTransaction\n");
 332                     rv = conn.startTransaction();
 333                 }
 334             }
 335             rv = table->execute();
 336             if (rv != OK) {
 337                 printf("Execute returned %d\n", rv);
 338                 os::usleep(500);
 339                 int retry=5;
 340                 while (rv == ErrLockTimeOut) {
 341                    if (retry == 0) { printf("DELETE:execute failed\n"); break;}
 342                    printf("retrying execute\n");
 343                    rv = table->execute();
 344                    retry--;
 345                 }
 346             }
 347             tuple = (char*)table->fetch() ;
 348             if (tuple == NULL)  { printf("DELETE:fetch failed\n");break;}
 349             ret = table->deleteTuple();
 350             retrycnt++;
 351         }
 352         if (ret != 0) { printf("RETURNING EARLY: %d with i:%d\n", ret, i); break;}
 353         table->closeScan();
 354         rv = conn.commit();
 355         if (rv != OK) {
 356             printf("COMMIT returned %d for record %d\n", rv, i);
 357             if (rv != ErrLockTimeOut)
 358             {
 359                 printf("RETURNING EARLY: with i:%d\n", i);
 360                 break;
 361             }
 362
 363             os::usleep(500);
 364             while (rv == ErrLockTimeOut) {
 365                 printf("retrying commit\n");
 366                 rv = conn.commit();
 367             }
 368         }
 369         icount++;
 370     }
 371    char msgBuf[1024];
 372    sprintf(msgBuf,"Delete: Thread %lu %d: Total rows :%d\n",os::getthrid(), val, icount);
 373    os::write(1,msgBuf,strlen(msgBuf));
 374    dbMgr->closeTable(table);
 375    rv  = conn.close();
 376    return NULL;
 377 }
 378
 379