Updated for notify locking

Author: djthorpe

cmd/indexer/main.go

@@ -185,8 +185,8 @@ func CreateSchema(ctx context.Context, pool SQPool) error {
 	// Create table
 	return conn.Do(ctx, 0, func(txn SQTransaction) error {
 		if _, err := txn.Query(Q(`CREATE TABLE IF NOT EXISTS files (
-			name TEXT NOT NULL,
-			path TEXT NOT NULL,
+			name       TEXT NOT NULL,
+			path       TEXT NOT NULL,
 			PRIMARY KEY (name, path)
 		)`)); err != nil {
 			return err

sys/sqlite3/statement.go

@@ -2,73 +2,96 @@ package sqlite3
 
 /*
 #cgo pkg-config: sqlite3
-#include <sqlite3.h>
 #include <stdlib.h>
+#include <sqlite3.h>
 #include <pthread.h>
 #include <assert.h>
 
+// sqlite library needs to be compiled with -DSQLITE_ENABLE_UNLOCK_NOTIFY
 // https://www.sqlite.org/unlock_notify.html
 
 typedef struct UnlockNotification UnlockNotification;
 
 struct UnlockNotification {
-  int fired;                         // True after unlock event has occurred
-  pthread_cond_t cond;               // Condition variable to wait on
-  pthread_mutex_t mutex;             // Mutex to protect structure
+	int fired;                         // True after unlock event has occurred
+	pthread_cond_t cond;               // Condition variable to wait on
+	pthread_mutex_t mutex;             // Mutex to protect structure
 };
 
 static void _unlock_notify_cb(void** apArg, int nArg) {
-  int i;
-  for(i = 0; i < nArg; i++) {
-    UnlockNotification *p = (UnlockNotification* )apArg[i];
-    pthread_mutex_lock(&p->mutex);
-    p->fired = 1;
-    pthread_cond_signal(&p->cond);
-    pthread_mutex_unlock(&p->mutex);
-  }
+	int i;
+	for(i = 0; i < nArg; i++) {
+	    UnlockNotification *p = (UnlockNotification* )apArg[i];
+	    pthread_mutex_lock(&p->mutex);
+	    p->fired = 1;
+	    pthread_cond_signal(&p->cond);
+	    pthread_mutex_unlock(&p->mutex);
+	}
 }
 
 int _wait_for_unlock_notify(sqlite3* db) {
-  int rc;
-  UnlockNotification un;
-
-  // Initialize the UnlockNotification structure
-  un.fired = 0;
-  pthread_mutex_init(&un.mutex, 0);
-  pthread_cond_init(&un.cond, 0);
-
-  // Register for an unlock-notify callback
-  rc = sqlite3_unlock_notify(db, _unlock_notify_cb, (void* )&un);
-  assert(rc==SQLITE_LOCKED || rc==SQLITE_OK);
-
-  // The call to sqlite3_unlock_notify() always returns either SQLITE_LOCKED
-  // or SQLITE_OK
-  if(rc == SQLITE_OK) {
-    pthread_mutex_lock(&un.mutex);
-    if (!un.fired) {
-      pthread_cond_wait(&un.cond, &un.mutex);
-    }
-    pthread_mutex_unlock(&un.mutex);
-  }
-
-  // Destroy the mutex and condition variables
-  pthread_cond_destroy(&un.cond);
-  pthread_mutex_destroy(&un.mutex);
-
-  return rc;
-}
+	int rc;
+	UnlockNotification un;
+
+	// Initialize the UnlockNotification structure
+	un.fired = 0;
+	pthread_mutex_init(&un.mutex, 0);
+	pthread_cond_init(&un.cond, 0);
+
+	// Register for an unlock-notify callback
+	// The call to sqlite3_unlock_notify() always returns either SQLITE_LOCKED
+	// or SQLITE_OK
+	rc = sqlite3_unlock_notify(db, _unlock_notify_cb, (void* )&un);
+	assert(rc == SQLITE_LOCKED || rc == SQLITE_OK);
+	if(rc == SQLITE_OK) {
+    	pthread_mutex_lock(&un.mutex);
+		if (!un.fired) {
+			pthread_cond_wait(&un.cond, &un.mutex);
+		}
+		pthread_mutex_unlock(&un.mutex);
+	}
 
+	// Destroy the mutex and condition variables
+	pthread_cond_destroy(&un.cond);
+	pthread_mutex_destroy(&un.mutex);
+	return rc;
+}
 
+// This code is a wrapper around sqlite3_step
 static int _sqlite3_blocking_step(sqlite3_stmt* stmt) {
-  int rc;
-  while(SQLITE_LOCKED == (rc = sqlite3_step(stmt))) {
-    rc = _wait_for_unlock_notify(sqlite3_db_handle(stmt));
-    if(rc != SQLITE_OK) break;
-    sqlite3_reset(stmt);
-  }
-  return rc;
+	while (1) {
+		int rc = sqlite3_step(stmt);
+		if ((rc & 0xFF) != SQLITE_LOCKED) {
+			return rc;
+		}
+		rc = _wait_for_unlock_notify(sqlite3_db_handle(stmt));
+		if (rc != SQLITE_OK) {
+			return rc;
+		}
+		sqlite3_reset(stmt);
+	}
 }
 
+// This code is a wrapper around sqlite3_prepare_v2
+static int _sqlite3_blocking_prepare_v2(
+  sqlite3* db,            // Database handle
+  const char* sql,        // UTF-8 encoded SQL statement
+  int nSql,               // Length of zSql in bytes
+  sqlite3_stmt** stmt,    // OUT: A pointer to the prepared statement
+  const char** pz         // OUT: End of parsed string
+){
+	int rc;
+	while (1) {
+		int rc = sqlite3_prepare_v2(db, sql, nSql, stmt, pz);
+		if ((rc & 0xFF) != SQLITE_LOCKED) {
+			return rc;
+		}
+		rc = _wait_for_unlock_notify(sqlite3_db_handle(stmt));
+		if (rc != SQLITE_OK) {
+			return rc;
+		}
+	}
+}
 */
 import "C"
 
@@ -149,7 +172,7 @@ func (c *Conn) Prepare(query string) (*Statement, string, error) {
 	}
 
 	// Prepare statement
-	if err := SQError(C.sqlite3_prepare_v2((*C.sqlite3)(c), cQuery, -1, &s, &cExtra)); err != SQLITE_OK {
+	if err := SQError(C._sqlite3_blocking_prepare_v2((*C.sqlite3)(c), cQuery, -1, &s, &cExtra)); err != SQLITE_OK {
 		return nil, "", err.With(C.GoString(C.sqlite3_errmsg((*C.sqlite3)(c))))
 	}