SQLite

#include <sqlite3.h>
#include <stdlib.h>
#include <stddef.h>
#include "tt3_core.c"

#ifdef USE_OSINST
# include "../src/test_osinst.c"
#else
# define vfslog_time() 0
#endif

typedef struct Config Config;
typedef struct ThreadCtx ThreadCtx;

#define THREAD_TIME_INSERT   0
#define THREAD_TIME_COMMIT   1
#define THREAD_TIME_ROLLBACK 2
#define THREAD_TIME_WRITER   3
#define THREAD_TIME_CKPT     4

struct ThreadCtx {
  Config *pConfig;
  Sqlite *pDb;
  Error *pErr;
  sqlite3_int64 aTime[5];
};

struct Config {
  int nIPT;                       /* --inserts-per-transaction */
  int nThread;                    /* --threads */
  int nSecond;                    /* --seconds */
  int bMutex;                     /* --mutex */
  int nAutoCkpt;                  /* --autockpt */
  int bRm;                        /* --rm */
  int bClearCache;                /* --clear-cache */
  int nMmap;                      /* mmap limit in MB */
  char *zFile;
  int bOsinst;                    /* True to use osinst */

  ThreadCtx *aCtx;                /* Array of size nThread */

  pthread_cond_t cond;
  pthread_mutex_t mutex;
  int nCondWait;                  /* Number of threads waiting on hCond */
  sqlite3_vfs *pVfs;
};








typedef struct VfsWrapperFd VfsWrapperFd;
struct VfsWrapperFd {
  sqlite3_file base;              /* Base class */
  int bWriter;                    /* True if holding shm WRITER lock */
  int iTid;
  Config *pConfig;
  sqlite3_file *pFd;              /* Underlying file descriptor */
};

/* Methods of the wrapper VFS */
static int vfsWrapOpen(sqlite3_vfs*, const char*, sqlite3_file*, int, int*);
static int vfsWrapDelete(sqlite3_vfs*, const char*, int);

    vfsWrapShmMap, vfsWrapShmLock, vfsWrapShmBarrier,
    vfsWrapShmUnmap, vfsWrapFetch, vfsWrapUnfetch
  };

  Config *pConfig = (Config*)pVfs->pAppData;
  VfsWrapperFd *pWrapper = (VfsWrapperFd*)pFd;
  int rc;

  memset(pWrapper, 0, sizeof(VfsWrapperFd));
  if( flags & SQLITE_OPEN_MAIN_DB ){
    pWrapper->iTid = (int)sqlite3_uri_int64(zName, "tid", 0);
  }

  pWrapper->pFd = (sqlite3_file*)&pWrapper[1];
  pWrapper->pConfig = pConfig;
  rc = pConfig->pVfs->xOpen(pConfig->pVfs, zName, pWrapper->pFd, flags, fout);
  if( rc==SQLITE_OK ){
    pWrapper->base.pMethods = &methods;
  }

  if(  (offset==0 && n==1)
    && (flags & SQLITE_SHM_LOCK) && (flags & SQLITE_SHM_EXCLUSIVE)
  ){
    pthread_mutex_lock(&pConfig->mutex);
    pWrapper->bWriter = 1;
    bMutex = 1;
    if( pWrapper->iTid ){
      sqlite3_int64 t = vfslog_time();
      pConfig->aCtx[pWrapper->iTid-1].aTime[THREAD_TIME_WRITER] -= t;
    }




  }

  rc = pWrapper->pFd->pMethods->xShmLock(pWrapper->pFd, offset, n, flags);

  if( (rc!=SQLITE_OK && bMutex)
   || (offset==0 && (flags & SQLITE_SHM_UNLOCK) && pWrapper->bWriter)
  ){
    assert( pWrapper->bWriter );
    pthread_mutex_unlock(&pConfig->mutex);
    pWrapper->bWriter = 0;
    if( pWrapper->iTid ){
      sqlite3_int64 t = vfslog_time();
      pConfig->aCtx[pWrapper->iTid-1].aTime[THREAD_TIME_WRITER] += t;
    }
  }

  return rc;
}
static void vfsWrapShmBarrier(sqlite3_file *pFd){
  VfsWrapperFd *pWrapper = (VfsWrapperFd*)pFd;
  return pWrapper->pFd->pMethods->xShmBarrier(pWrapper->pFd);

}


/*
** Wal hook used by connections in thread_main().
*/
static int thread_wal_hook(
  void *pArg,                     /* Pointer to ThreadCtx object */
  sqlite3 *db,
  const char *zDb, 
  int nFrame
){
  ThreadCtx *pCtx = (ThreadCtx*)pArg;
  Config *pConfig = pCtx->pConfig;

  if( pConfig->nAutoCkpt && nFrame>=pConfig->nAutoCkpt ){
    pCtx->aTime[THREAD_TIME_CKPT] -= vfslog_time();
    pthread_mutex_lock(&pConfig->mutex);
    if( pConfig->nCondWait>=0 ){
      pConfig->nCondWait++;
      if( pConfig->nCondWait==pConfig->nThread ){
        execsql(pCtx->pErr, pCtx->pDb, "PRAGMA wal_checkpoint");
        pthread_cond_broadcast(&pConfig->cond);
      }else{
        pthread_cond_wait(&pConfig->cond, &pConfig->mutex);
      }
      pConfig->nCondWait--;
    }
    pthread_mutex_unlock(&pConfig->mutex);
    pCtx->aTime[THREAD_TIME_CKPT] += vfslog_time();
  }

  return SQLITE_OK;
}


static char *thread_main(int iTid, void *pArg){

  Config *pConfig = (Config*)pArg;
  Error err = {0};                /* Error code and message */
  Sqlite db = {0};                /* SQLite database connection */
  int nAttempt = 0;               /* Attempted transactions */
  int nCommit = 0;                /* Successful transactions */
  int j;
  ThreadCtx *pCtx = &pConfig->aCtx[iTid-1];
  char *zUri = 0;

#ifdef USE_OSINST
  char *zOsinstName = 0;
  char *zLogName = 0;
  if( pConfig->bOsinst ){
    zOsinstName = sqlite3_mprintf("osinst%d", iTid);
    zLogName = sqlite3_mprintf("bc_test1.log.%d.%d", (int)getpid(), iTid);
    zUri = sqlite3_mprintf(
        "file:%s?vfs=%s&tid=%d", pConfig->zFile, zOsinstName, iTid
    );
    sqlite3_vfslog_new(zOsinstName, 0, zLogName);
    opendb(&err, &db, zUri, 0);
  }else
#endif
  {
    zUri = sqlite3_mprintf("file:%s?tid=%d", pConfig->zFile, iTid);
    opendb(&err, &db, zUri, 0);
  }

  sqlite3_busy_handler(db.db, 0, 0);
  sql_script_printf(&err, &db, 
      "PRAGMA wal_autocheckpoint = 0;"
      "PRAGMA synchronous = 0;"
      "PRAGMA mmap_limit = %lld;",
      (i64)(pConfig->nMmap) * 1024 * 1024
  );

  pCtx->pConfig = pConfig;
  pCtx->pErr = &err;
  pCtx->pDb = &db;
  sqlite3_wal_hook(db.db, thread_wal_hook, (void*)pCtx);

  while( !timetostop(&err) ){
    execsql(&err, &db, "BEGIN CONCURRENT");

    pCtx->aTime[THREAD_TIME_INSERT] -= vfslog_time();
    for(j=0; j<pConfig->nIPT; j++){
      execsql(&err, &db, 
          "INSERT INTO t1 VALUES"
          "(randomblob(10), randomblob(20), randomblob(30), randomblob(200))"
      );
    }
    pCtx->aTime[THREAD_TIME_INSERT] += vfslog_time();

    pCtx->aTime[THREAD_TIME_COMMIT] -= vfslog_time();
    execsql(&err, &db, "COMMIT");
    pCtx->aTime[THREAD_TIME_COMMIT] += vfslog_time();

    pCtx->aTime[THREAD_TIME_ROLLBACK] -= vfslog_time();
    nAttempt++;
    if( err.rc==SQLITE_OK ){
      nCommit++;
    }else{
      clear_error(&err, SQLITE_BUSY);
      execsql(&err, &db, "ROLLBACK");
    }
    pCtx->aTime[THREAD_TIME_ROLLBACK] += vfslog_time();

    if( pConfig->bClearCache ){
      sqlite3_db_release_memory(db.db);
    }
  }

  closedb(&err, &db);

#ifdef USE_OSINST
  if( pConfig->bOsinst ){
    sqlite3_vfslog_finalize(zOsinstName);
    sqlite3_free(zOsinstName);
    sqlite3_free(zLogName);
  }
#endif
  sqlite3_free(zUri);

  pthread_mutex_lock(&pConfig->mutex);
  pConfig->nCondWait = -1;
  pthread_cond_broadcast(&pConfig->cond);
  pthread_mutex_unlock(&pConfig->mutex);

  return sqlite3_mprintf("commits: %d/%d insert: %dms"
      " commit: %dms"
      " rollback: %dms" 
      " writer: %dms"
      " checkpoint: %dms", 
      nCommit, nAttempt, 
      (int)(pCtx->aTime[THREAD_TIME_INSERT]/1000), 
      (int)(pCtx->aTime[THREAD_TIME_COMMIT]/1000), 
      (int)(pCtx->aTime[THREAD_TIME_ROLLBACK]/1000),
      (int)(pCtx->aTime[THREAD_TIME_WRITER]/1000),
      (int)(pCtx->aTime[THREAD_TIME_CKPT]/1000)
  );
}

int main(int argc, const char **argv){
  Error err = {0};                /* Error code and message */
  Sqlite db = {0};                /* SQLite database connection */
  Threadset threads = {0};        /* Test threads */
  Config conf = {5, 3, 5};
  int i;

  CmdlineArg apArg[] = {
    { "--seconds", CMDLINE_INT,  offsetof(Config, nSecond) },
    { "--inserts", CMDLINE_INT,  offsetof(Config, nIPT) },
    { "--threads", CMDLINE_INT,  offsetof(Config, nThread) },
    { "--mutex",   CMDLINE_BOOL, offsetof(Config, bMutex) },
    { "--rm",      CMDLINE_BOOL, offsetof(Config, bRm) },
    { "--autockpt",CMDLINE_INT,  offsetof(Config, nAutoCkpt) },
    { "--mmap",    CMDLINE_INT,  offsetof(Config, nMmap) },
    { "--clear-cache",    CMDLINE_BOOL,  offsetof(Config, bClearCache) },
    { "--file",    CMDLINE_STRING,  offsetof(Config, zFile) },
    { "--osinst",  CMDLINE_BOOL,  offsetof(Config, bOsinst) },
    { 0, 0, 0 }
  };

  cmdline_process(apArg, argc, argv, (void*)&conf);
  if( err.rc==SQLITE_OK ){
    char *z = cmdline_construct(apArg, (void*)&conf);
    printf("With: %s\n", z);
    sqlite3_free(z);
  }
  if( conf.zFile==0 ){
    conf.zFile = "xyz.db";
  }

  /* Create the special VFS - "wrapper". And the mutex and condition 
  ** variable. */
  create_vfs(&conf);
  pthread_mutex_init(&conf.mutex, 0);
  pthread_cond_init(&conf.cond, 0);

  conf.aCtx = sqlite3_malloc(sizeof(ThreadCtx) * conf.nThread);
  memset(conf.aCtx, 0, sizeof(ThreadCtx) * conf.nThread);

  /* Ensure the schema has been created */
  opendb(&err, &db, conf.zFile, conf.bRm);
  sql_script(&err, &db,
      "PRAGMA journal_mode = wal;"
      "CREATE TABLE IF NOT EXISTS t1(a PRIMARY KEY, b, c, d) WITHOUT ROWID;"
      "CREATE INDEX IF NOT EXISTS t1b ON t1(b);"
      "CREATE INDEX IF NOT EXISTS t1c ON t1(c);"
  );

  setstoptime(&err, conf.nSecond*1000);
  if( conf.nThread==1 ){
    char *z = thread_main(1, (void*)&conf);
    printf("Thread 0 says: %s\n", (z==0 ? "..." : z));
    fflush(stdout);
  }else{
    for(i=0; i<conf.nThread; i++){
      launch_thread(&err, &threads, thread_main, (void*)&conf);
    }
    join_all_threads(&err, &threads);
  }

  if( err.rc==SQLITE_OK ){
    printf("Database is %dK\n", (int)(filesize(&err, conf.zFile) / 1024));
  }
  if( err.rc==SQLITE_OK ){
    char *zWal = sqlite3_mprintf("%s-wal", conf.zFile);
    printf("Wal file is %dK\n", (int)(filesize(&err, zWal) / 1024));
  }

  closedb(&err, &db);
  print_and_free_err(&err);
  return 0;
}

** End of test code/infrastructure interface macros.
*************************************************************************/


/************************************************************************
** Start of command line processing utilities.
*/
#define CMDLINE_INT     1
#define CMDLINE_BOOL    2
#define CMDLINE_STRING  3

typedef struct CmdlineArg CmdlineArg;
struct CmdlineArg {
  const char *zSwitch;
  int eType;
  int iOffset;
};

  int i;
  fprintf(stderr, "Usage: %s SWITCHES\n", zPrg);
  fprintf(stderr, "\n");
  fprintf(stderr, "where switches are\n");
  for(i=0; apArg[i].zSwitch; i++){
    const char *zExtra = "";
    switch( apArg[i].eType ){
      case CMDLINE_STRING: zExtra = "STRING"; break;
      case CMDLINE_INT: zExtra = "N"; break;
      case CMDLINE_BOOL: zExtra = ""; break;
      default:
        zExtra = "???";
        break;
    }
    fprintf(stderr, "  %s %s\n", apArg[i].zSwitch, zExtra);

  int iArg;

  for(iArg=0; apArg[iArg].zSwitch; iArg++){
    const char *zSpace = (zRet ? " " : "");
    CmdlineArg *pArg = &apArg[iArg];

    switch( pArg->eType ){
      case CMDLINE_STRING: {
        char *zVal = *(char**)(p + pArg->iOffset);
        if( zVal ){
          zRet = sqlite3_mprintf("%z%s%s %s", zRet, zSpace, pArg->zSwitch,zVal);
        }
        break;
      };

      case CMDLINE_INT: {
        zRet = sqlite3_mprintf("%z%s%s %d", zRet, zSpace, pArg->zSwitch, 
            *(int*)(p + pArg->iOffset)
        );
        break;
      };

          case CMDLINE_INT:
            i++;
            if( i==argc ){
              cmdline_error("option requires an argument: %s", z);
            }
            *(int*)(p + apArg[iArg].iOffset) = atoi(argv[i]);
            break;

          case CMDLINE_STRING:
            i++;
            if( i==argc ){
              cmdline_error("option requires an argument: %s", z);
            }
            *(char**)(p + apArg[iArg].iOffset) = sqlite3_mprintf("%s", argv[i]);
            break;

          case CMDLINE_BOOL:
            *(int*)(p + apArg[iArg].iOffset) = 1;
            break;

          default:
            assert( 0 );