X-Git-Url: https://jfr.im/git/irc/quakenet/newserv.git/blobdiff_plain/529abc44a3b712601d11c7a11722bc2f0e6a7ffa..211567363bd8bfbd70cef81e4208ea02d49ebb55:/sqlite/sqlite.c

diff --git a/sqlite/sqlite.c b/sqlite/sqlite.c
index 20385791..7b972642 100644
--- a/sqlite/sqlite.c
+++ b/sqlite/sqlite.c
@@ -2,94 +2,205 @@
  * SQLite module
  */
 
+#include <stdlib.h>
+#include <stdio.h>
+#include <stdarg.h>
+#include <string.h>
+
+#ifndef __USE_POSIX199309
+#define __USE_POSIX199309
+#endif
+
+#include <time.h>
+
 #include "../core/config.h"
 #include "../core/error.h"
-#include "../irc/irc_config.h"
 #include "../core/events.h"
 #include "../core/hooks.h"
-#include "../lib/irc_string.h"
 #include "../lib/version.h"
 #include "../lib/strlfunc.h"
 #include "../core/nsmalloc.h"
-#include "sqlite.h"
+#include "../core/schedule.h"
 
 #define BUILDING_DBAPI
 #include "../dbapi/dbapi.h"
 
-#include <stdlib.h>
-#include <sys/poll.h>
-#include <stdarg.h>
-#include <string.h>
-#define __USE_POSIX199309
-#include <time.h>
+#include "sqlite.h"
 
 MODULE_VERSION("");
 
 static int dbconnected = 0;
 static struct sqlite3 *conn;
+static SQLiteModuleIdentifier modid;
+
+struct sqlitequeue {
+  sqlite3_stmt *statement;
+  SQLiteQueryHandler handler;
+  void *tag;
+  int identifier;
+  struct sqlitequeue *next;
+};
+
+static struct sqlitequeue *head, *tail;
+static int queuesize;
+static void *processsched;
+static int inited;
 
 #define SYNC_MODE "OFF"
 
+static void sqlitequeueprocessor(void *arg);
+static void dbstatus(int hooknum, void *arg);
+
 void _init(void) {
   sstring *dbfile;
   int rc;
 
   dbfile = getcopyconfigitem("sqlite", "file", "newserv.db", 100);
 
-  if(!dbfile)
+  if(!dbfile) {
+    Error("sqlite", ERR_ERROR, "Unable to get config settings.");
     return;
+  }
   
+  processsched = schedulerecurring(time(NULL), 0, 1, &sqlitequeueprocessor, NULL);
+  if(!processsched) {
+    Error("sqlite", ERR_ERROR, "Unable to schedule query processor.");
+    freesstring(dbfile);
+    return;
+  }
+
+
+  if(sqlite3_initialize() != SQLITE_OK) {
+    Error("sqlite", ERR_ERROR, "Unable to initialise sqlite");
+    return;
+  }
+  sqlite3_config(SQLITE_CONFIG_SINGLETHREAD);
+
+  inited = 1;
+
   rc = sqlite3_open(dbfile->content, &conn);
   freesstring(dbfile);
 
   if(rc) {
     Error("sqlite", ERR_ERROR, "Unable to connect to database: %s", sqlite3_errmsg(conn));
+    deleteschedule(processsched, &sqlitequeueprocessor, NULL);
     return;
   }
 
   dbconnected = 1;
 
   sqliteasyncqueryf(0, NULL, NULL, 0, "PRAGMA synchronous=" SYNC_MODE ";");
+  registerhook(HOOK_CORE_STATSREQUEST, dbstatus);
 }
 
 void _fini(void) {
-  if(!sqliteconnected())
-    return;
+  struct sqlitequeue *q, *nq;
+
+  if(sqliteconnected()) {
+    deregisterhook(HOOK_CORE_STATSREQUEST, dbstatus);
+    deleteschedule(processsched, &sqlitequeueprocessor, NULL);
+
+    /* we assume every module that's being unloaded
+     * has us as a dependency and will have cleaned up
+     * their queries by using freeid..
+     */
+    for(q=head;q;q=nq) {
+      nq = q->next;
+      sqlite3_finalize(q->statement);
+      nsfree(POOL_SQLITE, q);
+    }
+
+    sqlite3_close(conn);
 
-  sqlite3_close(conn);
+    dbconnected = 0;
+  }
+
+  if(inited) {
+    sqlite3_shutdown();
+    inited = 0;
+  }
 
-  dbconnected = 0;
   nscheckfreeall(POOL_SQLITE);
 }
 
-int sqlitestep(sqlite3_stmt *s) {
-  int rc;
-  struct timespec t;
+/* busy processing is done externally as that varies depending on what you are... */
+static void processstatement(int rc, sqlite3_stmt *s, SQLiteQueryHandler handler, void *tag, char *querybuf) {
+  if(handler) { /* the handler deals with the cleanup */
+    SQLiteResult *r;
 
-  t.tv_sec = 0;
+    if((rc != SQLITE_ROW) && (rc != SQLITE_DONE)) {
+      Error("sqlite", ERR_WARNING, "SQL error %d: %s (query: %s)", rc, sqlite3_errmsg(conn), querybuf);
+      handler(NULL, tag);
+      return;
+    }
 
-  for(;;) {
-    rc = sqlite3_step(s);
-    if((rc == SQLITE_ROW) || (rc == SQLITE_OK) || (rc == SQLITE_DONE))
-      return rc;
-    if(rc == SQLITE_BUSY) {
-      t.tv_nsec = rand() % 50 + 50;
-      Error("sqlite", ERR_WARNING, "SQLite is busy, retrying in %dns...", t.tv_nsec);
-      nanosleep(&t, NULL);
-    } else {
-      Error("sqlite", ERR_WARNING, "SQL error %d: %s", rc, sqlite3_errmsg(conn));
-      return 0;
+    r = (SQLiteResult *)nsmalloc(POOL_SQLITE, sizeof(SQLiteResult));
+    r->r = s;
+    r->first = 1;
+    if(rc == SQLITE_ROW) {
+      r->final = 0;
+    } else if(rc == SQLITE_DONE) {
+      r->final = 1;
+    }
+    handler(r, tag);
+  } else {
+    if(rc == SQLITE_ROW) {
+      Error("sqlite", ERR_WARNING, "Unhandled data from query: %s", querybuf);
+    } else if(rc != SQLITE_DONE) {
+      Error("sqlite", ERR_WARNING, "SQL error %d: %s (query: %s)", rc, sqlite3_errmsg(conn), querybuf);
     }
+    sqlite3_finalize(s);
   }
 }
 
+static void pushqueue(sqlite3_stmt *s, int identifier, SQLiteQueryHandler handler, void *tag) {
+  struct sqlitequeue *q = (struct sqlitequeue *)nsmalloc(POOL_SQLITE, sizeof(struct sqlitequeue));
+
+  q->identifier = identifier;
+  q->handler = handler;
+  q->tag = tag;
+  q->next = NULL;
+  q->statement = s;
+
+  if(!tail) {
+    head = q;
+  } else {
+    tail->next = q;
+  }
+  tail = q;
+  queuesize++;
+}
+
+static struct sqlitequeue *peekqueue(void) {
+  return head;
+}
+
+/* a weird pop that doesn't return the value */
+static void popqueue(void) {
+  struct sqlitequeue *q;
+  if(!head)
+    return;
+
+  q = head;
+  if(head == tail) {
+    head = NULL;
+    tail = NULL;
+  } else {
+    head = head->next;
+  }
+
+  nsfree(POOL_SQLITE, q);
+  queuesize--;
+  return;
+}
+
 void sqliteasyncqueryf(int identifier, SQLiteQueryHandler handler, void *tag, int flags, char *format, ...) {
   char querybuf[8192];
-  va_list va;
   int len;
   int rc;
   sqlite3_stmt *s;
-  
+  va_list va;
+
   if(!sqliteconnected())
     return;
 
@@ -106,45 +217,76 @@ void sqliteasyncqueryf(int identifier, SQLiteQueryHandler handler, void *tag, in
     return;
   }
 
-  if(handler) {
-    handler(s, tag);
-  } else {
-    rc = sqlitestep(s);
-    sqlite3_finalize(s);
-    if(!rc)
-      Error("sqlite", ERR_WARNING, "SQL error %d: %s (query: %s)", rc, sqlite3_errmsg(conn), querybuf);
+  if(head) { /* stuff already queued */
+    pushqueue(s, identifier, handler, tag);
+    return;
   }
+
+  rc = sqlite3_step(s);
+  if(rc == SQLITE_BUSY) {
+    pushqueue(s, identifier, handler, tag);
+    return;
+  }
+
+  processstatement(rc, s, handler, tag, querybuf);
 }
 
 int sqliteconnected(void) {
   return dbconnected;
 }
 
-void sqliteescapestring(char *buf, char *src, unsigned int len) {
+size_t sqliteescapestring(char *buf, char *src, unsigned int len) {
   unsigned int i;
-  char *p;
+  char *p, *d;
 
-  for(p=src,i=0;i<len;i++,p++) {
+  for(p=src,d=buf,i=0;i<len;i++,p++) {
     if(*p == '\'')
-      *buf++ = *p;
-    *buf++ = *p;
+      *d++ = *p;
+    *d++ = *p;
   }
-  *buf = '\0';
+  *d = '\0';
+
+  return d - buf;
 }
 
 SQLiteResult *sqlitegetresult(SQLiteConn *r) {
-  SQLiteResult *r2;
-  if(!r)
-    return NULL;
-
-  r2 = (SQLiteResult *)nsmalloc(POOL_SQLITE, sizeof(SQLiteResult));
-  r2->r = r;
-  return r2;
+  return r;
 }
 
 int sqlitefetchrow(SQLiteResult *r) {
-  if(sqlitestep(r->r) != SQLITE_ROW)
+  int rc, v;
+  struct timespec t;
+
+  if(!r || !r->r || r->final)
+    return 0;
+
+  if(r->first) { /* we've extracted the first row already */
+    r->first = 0;
+    return 1;
+  }
+
+  t.tv_sec = 0;
+  for(;;) {
+    rc = sqlite3_step(r->r);
+    if(rc != SQLITE_BUSY)
+      break;
+
+    v = rand() % 50 + 50;
+    t.tv_nsec = v * 1000000;
+    Error("sqlite", ERR_WARNING, "SQLite is busy, retrying in %fs...", (double)v / 1000);
+    nanosleep(&t, NULL);
+  }
+
+  if(rc == SQLITE_DONE) {
+    r->final = 1;
     return 0;
+  }
+
+  if(rc != SQLITE_ROW) {
+    Error("sqlite", ERR_WARNING, "SQL error %d: %s", rc, sqlite3_errmsg(conn));
+    r->final = 1;
+    return 0;
+  }
 
   return 1;
 }
@@ -166,47 +308,150 @@ int sqlitequerysuccessful(SQLiteResult *r) {
   return 0;
 }
 
-void sqliteloadtable(char *tablename, SQLiteQueryHandler init, SQLiteQueryHandler data, SQLiteQueryHandler fini) {
-  int rc;
-  sqlite3_stmt *s;
-  char buf[1024];
-  
-  if(!sqliteconnected())
-    return;
+struct sqlitetableloader {
+  SQLiteQueryHandler init, data, fini;
+  void *tag;
+  char tablename[];
+};
 
-  snprintf(buf, sizeof(buf), "SELECT COUNT(*) FROM %s", tablename);
+static void loadtablerows(SQLiteConn *c, void *tag) {
+  struct sqlitetableloader *t = (struct sqlitetableloader *)tag;
 
-  rc = sqlite3_prepare(conn, buf, -1, &s, NULL);
-  if(rc != SQLITE_OK) {
-    Error("sqlite", ERR_ERROR, "Error getting row count for %s.", tablename);
+  if(!c) { /* pqsql doesnt call the handlers so we don't either */
+    nsfree(POOL_SQLITE, t);
     return;
   }
 
-  if(sqlitestep(s) != SQLITE_ROW) {
-    Error("sqlite", ERR_ERROR, "Error getting row count for %s.", tablename);
-    sqlite3_finalize(s);
+  /* the handlers do all the checking and cleanup */
+  if(t->init)
+    (t->init)(NULL, t->tag);
+
+  (t->data)(c, t->tag);
+
+  if(t->fini)
+    (t->fini)(NULL, t->tag);
+
+  nsfree(POOL_SQLITE, t);
+}
+
+static void loadtablecount(SQLiteConn *c, void *tag) {
+  struct sqlitetableloader *t = (struct sqlitetableloader *)tag;
+  SQLiteResult *r = NULL;
+
+  if(!c) { /* unloaded */
+    nsfree(POOL_SQLITE, t);
+    return;
+  } 
+
+  if(!(r = sqlitegetresult(c)) || !sqlitefetchrow(r)) {
+    Error("sqlite", ERR_ERROR, "Error getting row count for %s.", t->tablename);
+    nsfree(POOL_SQLITE, t);
+
+    if(r)
+      sqliteclear(r);
     return;
   }
 
-  Error("sqlite", ERR_INFO, "Found %s entries in table %s, loading...", (char *)sqlite3_column_text(s, 0), tablename);
-  sqlite3_finalize(s);
+  Error("sqlite", ERR_INFO, "Found %s entries in table %s, loading...", (char *)sqlite3_column_text(r->r, 0), t->tablename);
+  sqliteclear(r);
+  
+  sqliteasyncqueryf(0, loadtablerows, t, 0, "SELECT * FROM %s", t->tablename);
+}
 
-  snprintf(buf, sizeof(buf), "SELECT * FROM %s", tablename);
+void sqliteloadtable(char *tablename, SQLiteQueryHandler init, SQLiteQueryHandler data, SQLiteQueryHandler fini, void *tag) {
+  struct sqlitetableloader *t;
+  int len;
 
-  rc = sqlite3_prepare(conn, buf, -1, &s, NULL);
-  if(rc != SQLITE_OK) {
-    Error("sqlite", ERR_ERROR, "Error loading table %s.", tablename);
+  if(!sqliteconnected())
     return;
-  }
 
-  if(init)
-    init(s, NULL);
-  data(s, NULL);
-  if(fini)
-    fini(s, NULL);
+  len = strlen(tablename);
+
+  t = (struct sqlitetableloader *)nsmalloc(POOL_SQLITE, sizeof(struct sqlitetableloader) + len + 1);
+  memcpy(t->tablename, tablename, len + 1);
+  t->init = init;
+  t->data = data;
+  t->fini = fini;
+  t->tag = tag;
+
+  sqliteasyncqueryf(0, loadtablecount, t, 0, "SELECT COUNT(*) FROM %s", tablename);
 }
 
-void sqlitecreateschema(char *schema) {
+void sqliteattach(char *schema) {
   sqliteasyncqueryf(0, NULL, NULL, 0, "ATTACH DATABASE '%s.db' AS %s", schema, schema);
-  sqliteasyncqueryf(0, NULL, NULL, 0, "PRAGMA %s.synchronous=" SYNC_MODE ";", schema);
+  sqliteasyncqueryf(0, NULL, NULL, 0, "PRAGMA %s.synchronous=" SYNC_MODE, schema);
+}
+
+void sqlitedetach(char *schema) {
+  sqliteasyncqueryf(0, NULL, NULL, 0, "DETACH DATABASE %s", schema);
+}
+
+int sqlitegetid(void) {
+  modid++; 
+  if(modid == 0)
+    modid = 1;
+
+  return modid;
+}
+
+void sqlitefreeid(int id) {
+  struct sqlitequeue *q, *pq;
+  if(id == 0)
+    return;
+
+  for(pq=NULL,q=head;q;) {
+    if(q->identifier == id) {
+      if(pq) {
+        pq->next = q->next;
+        if(q == tail)
+          tail = pq;
+      } else { /* head */
+        head = q->next;
+        if(q == tail)
+          tail = NULL;
+      }
+      sqlite3_finalize(q->statement);
+
+      q->handler(NULL, q->tag);
+      nsfree(POOL_SQLITE, q);
+
+      queuesize--;
+      if(pq) {
+        q = pq->next;
+      } else {
+        q = head;
+      }
+    } else {
+      pq = q;
+      q = q->next;
+    }
+  }
+}
+
+static void sqlitequeueprocessor(void *arg) {
+  struct sqlitequeue *q = peekqueue();
+
+  while(q) {
+    int rc = sqlite3_step(q->statement);
+    if(rc == SQLITE_BUSY)
+      return;
+
+    processstatement(rc, q->statement, q->handler, q->tag, "??");
+    popqueue();
+
+    q = peekqueue();
+  }
+}
+
+static void dbstatus(int hooknum, void *arg) {
+  if((long)arg > 10) {
+    char message[100];
+
+    snprintf(message, sizeof(message), "SQLite  : %6d queries queued.", queuesize);
+    triggerhook(HOOK_CORE_STATSREPLY, message);
+  }
+}
+
+sqlite3 *sqlitegetconn(void) {
+  return conn;
 }