summaryrefslogtreecommitdiffstats
path: root/test/threadtest4.c
diff options
context:
space:
mode:
Diffstat (limited to 'test/threadtest4.c')
-rw-r--r--test/threadtest4.c484
1 files changed, 484 insertions, 0 deletions
diff --git a/test/threadtest4.c b/test/threadtest4.c
new file mode 100644
index 0000000..da5ce75
--- /dev/null
+++ b/test/threadtest4.c
@@ -0,0 +1,484 @@
+/*
+** 2014-12-11
+**
+** The author disclaims copyright to this source code. In place of
+** a legal notice, here is a blessing:
+**
+** May you do good and not evil.
+** May you find forgiveness for yourself and forgive others.
+** May you share freely, never taking more than you give.
+**
+*************************************************************************
+** This file implements a simple standalone program used to stress the
+** SQLite library when accessing the same set of databases simultaneously
+** from multiple threads in shared-cache mode.
+**
+** This test program runs on unix-like systems only. It uses pthreads.
+** To compile:
+**
+** gcc -g -Wall -I. threadtest4.c sqlite3.c -ldl -lpthread
+**
+** To run:
+**
+** ./a.out 10
+**
+** The argument is the number of threads. There are also options, such
+** as -wal and -multithread and -serialized.
+**
+** Consider also compiling with clang instead of gcc and adding the
+** -fsanitize=thread option.
+*/
+#include "sqlite3.h"
+#include <pthread.h>
+#include <sched.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+#include <stdarg.h>
+
+/*
+** An instance of the following structure is passed into each worker
+** thread.
+*/
+typedef struct WorkerInfo WorkerInfo;
+struct WorkerInfo {
+ int tid; /* Thread ID */
+ int nWorker; /* Total number of workers */
+ unsigned wkrFlags; /* Flags */
+ sqlite3 *mainDb; /* Database connection of the main thread */
+ sqlite3 *db; /* Database connection of this thread */
+ int nErr; /* Number of errors seen by this thread */
+ int nTest; /* Number of tests run by this thread */
+ char *zMsg; /* Message returned by this thread */
+ pthread_t id; /* Thread id */
+ pthread_mutex_t *pWrMutex; /* Hold this mutex while writing */
+};
+
+/*
+** Allowed values for WorkerInfo.wkrFlags
+*/
+#define TT4_SERIALIZED 0x0000001 /* The --serialized option is used */
+#define TT4_WAL 0x0000002 /* WAL mode in use */
+#define TT4_TRACE 0x0000004 /* Trace activity */
+
+
+/*
+** Report an OOM error and die if the argument is NULL
+*/
+static void check_oom(void *x){
+ if( x==0 ){
+ fprintf(stderr, "out of memory\n");
+ exit(1);
+ }
+}
+
+/*
+** Allocate memory. If the allocation fails, print an error message and
+** kill the process.
+*/
+static void *safe_malloc(int sz){
+ void *x = sqlite3_malloc(sz>0?sz:1);
+ check_oom(x);
+ return x;
+}
+
+/*
+** Print a trace message for a worker
+*/
+static void worker_trace(WorkerInfo *p, const char *zFormat, ...){
+ va_list ap;
+ char *zMsg;
+ if( (p->wkrFlags & TT4_TRACE)==0 ) return;
+ va_start(ap, zFormat);
+ zMsg = sqlite3_vmprintf(zFormat, ap);
+ check_oom(zMsg);
+ va_end(ap);
+ fprintf(stderr, "TRACE(%02d): %s\n", p->tid, zMsg);
+ sqlite3_free(zMsg);
+}
+
+/*
+** Prepare a single SQL query
+*/
+static sqlite3_stmt *prep_sql(sqlite3 *db, const char *zFormat, ...){
+ va_list ap;
+ char *zSql;
+ int rc;
+ sqlite3_stmt *pStmt = 0;
+
+ va_start(ap, zFormat);
+ zSql = sqlite3_vmprintf(zFormat, ap);
+ va_end(ap);
+ check_oom(zSql);
+ rc = sqlite3_prepare_v2(db, zSql, -1, &pStmt, 0);
+ if( rc!=SQLITE_OK ){
+ fprintf(stderr, "SQL error (%d,%d): %s\nWhile preparing: [%s]\n",
+ rc, sqlite3_extended_errcode(db), sqlite3_errmsg(db), zSql);
+ exit(1);
+ }
+ sqlite3_free(zSql);
+ return pStmt;
+}
+
+/*
+** Run a SQL statements. Panic if unable.
+*/
+static void run_sql(WorkerInfo *p, const char *zFormat, ...){
+ va_list ap;
+ char *zSql;
+ int rc;
+ sqlite3_stmt *pStmt = 0;
+ int nRetry = 0;
+
+ va_start(ap, zFormat);
+ zSql = sqlite3_vmprintf(zFormat, ap);
+ va_end(ap);
+ check_oom(zSql);
+ rc = sqlite3_prepare_v2(p->db, zSql, -1, &pStmt, 0);
+ if( rc!=SQLITE_OK ){
+ fprintf(stderr, "SQL error (%d,%d): %s\nWhile preparing: [%s]\n",
+ rc, sqlite3_extended_errcode(p->db), sqlite3_errmsg(p->db), zSql);
+ exit(1);
+ }
+ worker_trace(p, "running [%s]", zSql);
+ while( (rc = sqlite3_step(pStmt))!=SQLITE_DONE ){
+ if( (rc&0xff)==SQLITE_BUSY || (rc&0xff)==SQLITE_LOCKED ){
+ sqlite3_reset(pStmt);
+ nRetry++;
+ if( nRetry<10 ){
+ worker_trace(p, "retry %d for [%s]", nRetry, zSql);
+ sched_yield();
+ continue;
+ }else{
+ fprintf(stderr, "Deadlock in thread %d while running [%s]\n",
+ p->tid, zSql);
+ exit(1);
+ }
+ }
+ if( rc!=SQLITE_ROW ){
+ fprintf(stderr, "SQL error (%d,%d): %s\nWhile running [%s]\n",
+ rc, sqlite3_extended_errcode(p->db), sqlite3_errmsg(p->db), zSql);
+ exit(1);
+ }
+ }
+ sqlite3_free(zSql);
+ sqlite3_finalize(pStmt);
+}
+
+
+/*
+** Open the database connection for WorkerInfo. The order in which
+** the files are opened is a function of the tid value.
+*/
+static void worker_open_connection(WorkerInfo *p, int iCnt){
+ char *zFile;
+ int x;
+ int rc;
+ static const unsigned char aOrder[6][3] = {
+ { 1, 2, 3},
+ { 1, 3, 2},
+ { 2, 1, 3},
+ { 2, 3, 1},
+ { 3, 1, 2},
+ { 3, 2, 1}
+ };
+ x = (p->tid + iCnt) % 6;
+ zFile = sqlite3_mprintf("tt4-test%d.db", aOrder[x][0]);
+ check_oom(zFile);
+ worker_trace(p, "open %s", zFile);
+ rc = sqlite3_open_v2(zFile, &p->db,
+ SQLITE_OPEN_READWRITE|SQLITE_OPEN_SHAREDCACHE, 0);
+ if( rc!=SQLITE_OK ){
+ fprintf(stderr, "sqlite_open_v2(%s) failed on thread %d\n",
+ zFile, p->tid);
+ exit(1);
+ }
+ sqlite3_free(zFile);
+ run_sql(p, "PRAGMA read_uncommitted=ON;");
+ sqlite3_busy_timeout(p->db, 10000);
+ run_sql(p, "PRAGMA synchronous=OFF;");
+ run_sql(p, "ATTACH 'tt4-test%d.db' AS aux1", aOrder[x][1]);
+ run_sql(p, "ATTACH 'tt4-test%d.db' AS aux2", aOrder[x][2]);
+}
+
+/*
+** Close the worker database connection
+*/
+static void worker_close_connection(WorkerInfo *p){
+ if( p->db ){
+ worker_trace(p, "close");
+ sqlite3_close(p->db);
+ p->db = 0;
+ }
+}
+
+/*
+** Delete all content in the three databases associated with a
+** single thread. Make this happen all in a single transaction if
+** inTrans is true, or separately for each database if inTrans is
+** false.
+*/
+static void worker_delete_all_content(WorkerInfo *p, int inTrans){
+ if( inTrans ){
+ pthread_mutex_lock(p->pWrMutex);
+ run_sql(p, "BEGIN");
+ run_sql(p, "DELETE FROM t1 WHERE tid=%d", p->tid);
+ run_sql(p, "DELETE FROM t2 WHERE tid=%d", p->tid);
+ run_sql(p, "DELETE FROM t3 WHERE tid=%d", p->tid);
+ run_sql(p, "COMMIT");
+ pthread_mutex_unlock(p->pWrMutex);
+ p->nTest++;
+ }else{
+ pthread_mutex_lock(p->pWrMutex);
+ run_sql(p, "DELETE FROM t1 WHERE tid=%d", p->tid);
+ pthread_mutex_unlock(p->pWrMutex);
+ p->nTest++;
+ pthread_mutex_lock(p->pWrMutex);
+ run_sql(p, "DELETE FROM t2 WHERE tid=%d", p->tid);
+ pthread_mutex_unlock(p->pWrMutex);
+ p->nTest++;
+ pthread_mutex_lock(p->pWrMutex);
+ run_sql(p, "DELETE FROM t3 WHERE tid=%d", p->tid);
+ pthread_mutex_unlock(p->pWrMutex);
+ p->nTest++;
+ }
+}
+
+/*
+** Create rows mn through mx in table iTab for the given worker
+*/
+static void worker_add_content(WorkerInfo *p, int mn, int mx, int iTab){
+ char *zTabDef;
+ switch( iTab ){
+ case 1: zTabDef = "t1(tid,sp,a,b,c)"; break;
+ case 2: zTabDef = "t2(tid,sp,d,e,f)"; break;
+ case 3: zTabDef = "t3(tid,sp,x,y,z)"; break;
+ }
+ pthread_mutex_lock(p->pWrMutex);
+ run_sql(p,
+ "WITH RECURSIVE\n"
+ " c(i) AS (VALUES(%d) UNION ALL SELECT i+1 FROM c WHERE i<%d)\n"
+ "INSERT INTO %s SELECT %d, zeroblob(3000), i, printf('%%d',i), i FROM c;",
+ mn, mx, zTabDef, p->tid
+ );
+ pthread_mutex_unlock(p->pWrMutex);
+ p->nTest++;
+}
+
+/*
+** Set an error message on a worker
+*/
+static void worker_error(WorkerInfo *p, const char *zFormat, ...){
+ va_list ap;
+ p->nErr++;
+ sqlite3_free(p->zMsg);
+ va_start(ap, zFormat);
+ p->zMsg = sqlite3_vmprintf(zFormat, ap);
+ va_end(ap);
+}
+
+/*
+** Each thread runs the following function.
+*/
+static void *worker_thread(void *pArg){
+ WorkerInfo *p = (WorkerInfo*)pArg;
+ int iOuter;
+ int i;
+ int rc;
+ sqlite3_stmt *pStmt;
+
+ printf("worker %d startup\n", p->tid); fflush(stdout);
+ for(iOuter=1; iOuter<=p->nWorker; iOuter++){
+ worker_open_connection(p, iOuter);
+ for(i=0; i<4; i++){
+ worker_add_content(p, i*100+1, (i+1)*100, (p->tid+iOuter)%3 + 1);
+ worker_add_content(p, i*100+1, (i+1)*100, (p->tid+iOuter+1)%3 + 1);
+ worker_add_content(p, i*100+1, (i+1)*100, (p->tid+iOuter+2)%3 + 1);
+ }
+
+ pStmt = prep_sql(p->db, "SELECT count(a) FROM t1 WHERE tid=%d", p->tid);
+ worker_trace(p, "query [%s]", sqlite3_sql(pStmt));
+ rc = sqlite3_step(pStmt);
+ if( rc!=SQLITE_ROW ){
+ worker_error(p, "Failed to step: %s", sqlite3_sql(pStmt));
+ }else if( sqlite3_column_int(pStmt, 0)!=400 ){
+ worker_error(p, "Wrong result: %d", sqlite3_column_int(pStmt,0));
+ }
+ sqlite3_finalize(pStmt);
+ if( p->nErr ) break;
+
+ if( ((iOuter+p->tid)%3)==0 ){
+ sqlite3_db_release_memory(p->db);
+ p->nTest++;
+ }
+
+ pthread_mutex_lock(p->pWrMutex);
+ run_sql(p, "BEGIN;");
+ run_sql(p, "UPDATE t1 SET c=NULL WHERE a=55");
+ run_sql(p, "UPDATE t2 SET f=NULL WHERE d=42");
+ run_sql(p, "UPDATE t3 SET z=NULL WHERE x=31");
+ run_sql(p, "ROLLBACK;");
+ p->nTest++;
+ pthread_mutex_unlock(p->pWrMutex);
+
+
+ if( iOuter==p->tid ){
+ pthread_mutex_lock(p->pWrMutex);
+ run_sql(p, "VACUUM");
+ pthread_mutex_unlock(p->pWrMutex);
+ }
+
+ pStmt = prep_sql(p->db,
+ "SELECT t1.rowid, t2.rowid, t3.rowid"
+ " FROM t1, t2, t3"
+ " WHERE t1.tid=%d AND t2.tid=%d AND t3.tid=%d"
+ " AND t1.a<>t2.d AND t2.d<>t3.x"
+ " ORDER BY 1, 2, 3"
+ ,p->tid, p->tid, p->tid);
+ worker_trace(p, "query [%s]", sqlite3_sql(pStmt));
+ for(i=0; i<p->nWorker; i++){
+ rc = sqlite3_step(pStmt);
+ if( rc!=SQLITE_ROW ){
+ worker_error(p, "Failed to step: %s", sqlite3_sql(pStmt));
+ break;
+ }
+ sched_yield();
+ }
+ sqlite3_finalize(pStmt);
+ if( p->nErr ) break;
+
+ worker_delete_all_content(p, (p->tid+iOuter)%2);
+ worker_close_connection(p);
+ p->db = 0;
+ }
+ worker_close_connection(p);
+ printf("worker %d finished\n", p->tid); fflush(stdout);
+ return 0;
+}
+
+int main(int argc, char **argv){
+ int nWorker = 0; /* Number of worker threads */
+ int i; /* Loop counter */
+ WorkerInfo *aInfo; /* Information for each worker */
+ unsigned wkrFlags = 0; /* Default worker flags */
+ int nErr = 0; /* Number of errors */
+ int nTest = 0; /* Number of tests */
+ int rc; /* Return code */
+ sqlite3 *db = 0; /* Main database connection */
+ pthread_mutex_t wrMutex; /* The write serialization mutex */
+ WorkerInfo infoTop; /* WorkerInfo for the main thread */
+ WorkerInfo *p; /* Pointer to infoTop */
+
+ sqlite3_config(SQLITE_CONFIG_MULTITHREAD);
+ for(i=1; i<argc; i++){
+ const char *z = argv[i];
+ if( z[0]=='-' ){
+ if( z[1]=='-' && z[2]!=0 ) z++;
+ if( strcmp(z,"-multithread")==0 ){
+ sqlite3_config(SQLITE_CONFIG_MULTITHREAD);
+ wkrFlags &= ~TT4_SERIALIZED;
+ }else if( strcmp(z,"-serialized")==0 ){
+ sqlite3_config(SQLITE_CONFIG_SERIALIZED);
+ wkrFlags |= TT4_SERIALIZED;
+ }else if( strcmp(z,"-wal")==0 ){
+ wkrFlags |= TT4_WAL;
+ }else if( strcmp(z,"-trace")==0 ){
+ wkrFlags |= TT4_TRACE;
+ }else{
+ fprintf(stderr, "unknown command-line option: %s\n", argv[i]);
+ exit(1);
+ }
+ }else if( z[0]>='1' && z[0]<='9' && nWorker==0 ){
+ nWorker = atoi(z);
+ if( nWorker<2 ){
+ fprintf(stderr, "minimum of 2 threads\n");
+ exit(1);
+ }
+ }else{
+ fprintf(stderr, "extra command-line argument: \"%s\"\n", argv[i]);
+ exit(1);
+ }
+ }
+ if( nWorker==0 ){
+ fprintf(stderr,
+ "usage: %s ?OPTIONS? N\n"
+ "N is the number of threads and must be at least 2.\n"
+ "Options:\n"
+ " --serialized\n"
+ " --multithread\n"
+ " --wal\n"
+ " --trace\n"
+ ,argv[0]
+ );
+ exit(1);
+ }
+ if( !sqlite3_threadsafe() ){
+ fprintf(stderr, "requires a threadsafe build of SQLite\n");
+ exit(1);
+ }
+ sqlite3_initialize();
+ sqlite3_enable_shared_cache(1);
+ pthread_mutex_init(&wrMutex, 0);
+
+ /* Initialize the test database files */
+ (void)unlink("tt4-test1.db");
+ (void)unlink("tt4-test2.db");
+ (void)unlink("tt4-test3.db");
+ rc = sqlite3_open("tt4-test1.db", &db);
+ if( rc!=SQLITE_OK ){
+ fprintf(stderr, "Unable to open test database: tt4-test2.db\n");
+ exit(1);
+ }
+ memset(&infoTop, 0, sizeof(infoTop));
+ infoTop.db = db;
+ infoTop.wkrFlags = wkrFlags;
+ p = &infoTop;
+ if( wkrFlags & TT4_WAL ){
+ run_sql(p, "PRAGMA journal_mode=WAL");
+ }
+ run_sql(p, "PRAGMA synchronous=OFF");
+ run_sql(p, "CREATE TABLE IF NOT EXISTS t1(tid INTEGER, sp, a, b, c)");
+ run_sql(p, "CREATE INDEX t1tid ON t1(tid)");
+ run_sql(p, "CREATE INDEX t1ab ON t1(a,b)");
+ run_sql(p, "ATTACH 'tt4-test2.db' AS 'test2'");
+ run_sql(p, "CREATE TABLE IF NOT EXISTS test2.t2(tid INTEGER, sp, d, e, f)");
+ run_sql(p, "CREATE INDEX test2.t2tid ON t2(tid)");
+ run_sql(p, "CREATE INDEX test2.t2de ON t2(d,e)");
+ run_sql(p, "ATTACH 'tt4-test3.db' AS 'test3'");
+ run_sql(p, "CREATE TABLE IF NOT EXISTS test3.t3(tid INTEGER, sp, x, y, z)");
+ run_sql(p, "CREATE INDEX test3.t3tid ON t3(tid)");
+ run_sql(p, "CREATE INDEX test3.t3xy ON t3(x,y)");
+ aInfo = safe_malloc( sizeof(*aInfo)*nWorker );
+ memset(aInfo, 0, sizeof(*aInfo)*nWorker);
+ for(i=0; i<nWorker; i++){
+ aInfo[i].tid = i+1;
+ aInfo[i].nWorker = nWorker;
+ aInfo[i].wkrFlags = wkrFlags;
+ aInfo[i].mainDb = db;
+ aInfo[i].pWrMutex = &wrMutex;
+ rc = pthread_create(&aInfo[i].id, 0, worker_thread, &aInfo[i]);
+ if( rc!=0 ){
+ fprintf(stderr, "thread creation failed for thread %d\n", i+1);
+ exit(1);
+ }
+ sched_yield();
+ }
+ for(i=0; i<nWorker; i++){
+ pthread_join(aInfo[i].id, 0);
+ printf("Joined thread %d: %d errors in %d tests",
+ aInfo[i].tid, aInfo[i].nErr, aInfo[i].nTest);
+ if( aInfo[i].zMsg ){
+ printf(": %s\n", aInfo[i].zMsg);
+ }else{
+ printf("\n");
+ }
+ nErr += aInfo[i].nErr;
+ nTest += aInfo[i].nTest;
+ fflush(stdout);
+ }
+ sqlite3_close(db);
+ sqlite3_free(aInfo);
+ printf("Total %d errors in %d tests\n", nErr, nTest);
+ return nErr;
+}
generated by cgit v1.2.3 (git 2.39.1) at 2024-11-16 11:54:34 +0000