c_src/bdberl_drv.c

@@ -113,7 +113,6 @@ static void do_async_txnop(void* arg);
 static void do_async_cursor_put(void* arg);
 static void do_async_cursor_get(void* arg);
 static void do_async_cursor_del(void* arg);
-static void do_async_cursor_count(void* arg);
 static void do_async_cursor_cnp(void* arg);
 static void do_async_truncate(void* arg);
 static void do_sync_data_dirs_info(PortData *p);
@@ -137,7 +136,6 @@ static void* checkpointer(void* arg);
 
 static void bdb_errcall(const DB_ENV* dbenv, const char* errpfx, const char* msg);
 static void bdb_msgcall(const DB_ENV* dbenv, const char* msg);
-static void bdb_eventcall(DB_ENV* dbenv, u_int32_t type, void* msg);
 static void send_log_message(ErlDrvTermData* msg, int elements);
 
 /**
@@ -200,7 +198,7 @@ static unsigned int G_CHECKPOINT_ACTIVE   = 1;
 static unsigned int G_CHECKPOINT_INTERVAL = 60;         /* Seconds between checkpoints */
 
 /**
- * Pipe is used to wake up the various monitors.  Instead of just sleeping
+ * Pipe to used to wake up the various monitors.  Instead of just sleeping
  * they wait for an exceptional condition on the read fd of the pipe.  When it is time to
  * shutdown, the driver closes the write fd and waits for the threads to be joined.
  */
@@ -229,20 +227,36 @@ static TPool* G_TPOOL_GENERAL = NULL;
 static TPool* G_TPOOL_TXNS    = NULL;
 
 
+/**
+ * Helpful macros
+ */
+#ifdef DEBUG
+#  define DBG(...) fprintf(stderr, __VA_ARGS__)
+#  define DBGCMD(P, ...)   bdberl_dbgcmd(P, __VA_ARGS__)
+#  define DBGCMDRC(P, ...) bdberl_dbgcmdrc(P, __VA_ARGS__)
+static void bdberl_dbgcmd(PortData *d, const char *fmt, ...);
+static void bdberl_dbgcmdrc(PortData *d, int rc);
+#else
+#  define DBG(arg1,...)
+#  define DBGCMD(d, fmt, ...)
+#  define DBGCMDRC(d, rc) { while (0) { rc++; } }  // otherwise get unused variable error
+#endif
+
+
 #define LOCK_DATABASES(P)                                               \
     do                                                                  \
     {                                                                   \
-        DBG("threadid %p port %p: locking G_DATABASES\n", erl_drv_thread_self(), P);   \
+        DBG("threadid %p port %p: locking G_DATABASES\r\n", erl_drv_thread_self(), P);   \
         erl_drv_mutex_lock(G_DATABASES_MUTEX);                          \
-        DBG("threadid %p port %p: locked G_DATABASES\n", erl_drv_thread_self(), P);    \
+        DBG("threadid %p port %p: locked G_DATABASES\r\n", erl_drv_thread_self(), P);    \
     } while(0)
 
 #define UNLOCK_DATABASES(P)                                             \
     do                                                                  \
     {                                                                   \
-        DBG("threadid %p port %p: unlocking G_DATABASES\n", erl_drv_thread_self(), P); \
+        DBG("threadid %p port %p: unlocking G_DATABASES\r\n", erl_drv_thread_self(), P); \
         erl_drv_mutex_unlock(G_DATABASES_MUTEX);                        \
-        DBG("threadid %p port %p: unlocked G_DATABASES\n", erl_drv_thread_self(), P);  \
+        DBG("threadid %p port %p: unlocked G_DATABASES\r\n", erl_drv_thread_self(), P);  \
     } while (0)
 
 
@@ -254,7 +268,7 @@ static TPool* G_TPOOL_TXNS    = NULL;
 
 DRIVER_INIT(bdberl_drv)
 {
-    DBG("DRIVER INIT");
+    DBG("DRIVER INIT\r\n");
     // Setup flags we'll use to init the environment
     int flags =
         DB_INIT_LOCK |          /* Enable support for locking */
@@ -263,9 +277,6 @@ DRIVER_INIT(bdberl_drv)
         DB_RECOVER |            /* Enable support for recovering from failures */
         DB_CREATE |             /* Create files as necessary */
         DB_REGISTER |           /* Run recovery if needed */
-        DB_FAILCHK |            /* Release any database reads locks held by the
-                                   thread of control that exited and, if needed,
-                                   abort unresolved transaction. */
         DB_USE_ENVIRON |        /* Use DB_HOME environment variable */
         DB_THREAD;              /* Make the environment free-threaded */
 
@@ -281,46 +292,25 @@ DRIVER_INIT(bdberl_drv)
     // specify where the working directory is
     DBG("db_env_create(%p, 0)", &G_DB_ENV);
     G_DB_ENV_ERROR = db_env_create(&G_DB_ENV, 0);
-    DBG(" = %d\n", G_DB_ENV_ERROR);
+    DBG(" = %d\r\n", G_DB_ENV_ERROR);
     if (G_DB_ENV_ERROR != 0)
     {
         G_DB_ENV = 0;
     }
     else
     {
-        // DB should use the safe allocation routines provided by the Erlang VM
-        DBG("G_DB_ENV->set_alloc(%p, ...)", &G_DB_ENV);
         G_DB_ENV_ERROR = G_DB_ENV->set_alloc(G_DB_ENV, driver_alloc, driver_realloc, driver_free);
-        DBG(" = %d\n", G_DB_ENV_ERROR);
+        DBG(" = %d\r\n", G_DB_ENV_ERROR);
 
-        // Inform DB of the number of threads that will be operating on the DB Environment
-        unsigned int nthreads = G_NUM_GENERAL_THREADS + G_NUM_TXN_THREADS;
-        DBG("G_DB_ENV->set_thread_count(%p, %d, ...)", &G_DB_ENV, nthreads);
-        G_DB_ENV_ERROR = G_DB_ENV->set_thread_count(G_DB_ENV, nthreads);
-        DBG(" = %d\n", G_DB_ENV_ERROR);
-
-        DBG("G_DB_ENV->set_thread_id(%p, ...)", &G_DB_ENV);
-        G_DB_ENV_ERROR = G_DB_ENV->set_thread_id(G_DB_ENV, &bdberl_tpool_thread_id);
-        DBG(" = %d\n", G_DB_ENV_ERROR);
-
-        DBG("G_DB_ENV->set_thread_id_string(%p, ...)", &G_DB_ENV);
-        G_DB_ENV_ERROR = G_DB_ENV->set_thread_id_string(G_DB_ENV, &bdberl_tpool_thread_id_string);
-        DBG(" = %d\n", G_DB_ENV_ERROR);
-
-        DBG("G_DB_ENV->set_is_alive(%p, ...)", &G_DB_ENV);
-        G_DB_ENV_ERROR = G_DB_ENV->set_isalive(G_DB_ENV, &bdberl_tpool_thread_is_alive);
-        DBG(" = %d\n", G_DB_ENV_ERROR);
-
-        // Open the DB Environment
         DBG("G_DB_ENV->open(%p, 0, %08X, 0)", &G_DB_ENV, flags);
         G_DB_ENV_ERROR = G_DB_ENV->open(G_DB_ENV, 0, flags, 0);
-        DBG(" = %d\n", G_DB_ENV_ERROR);
+        DBG(" = %d\r\n", G_DB_ENV_ERROR);
         if (G_DB_ENV_ERROR != 0)
         {
             // Something bad happened while initializing BDB; in this situation we
             // cleanup and set the environment to zero. Attempts to open ports will
             // fail and the user will have to sort out how to resolve the issue.
-            DBG("G_DB_ENV->close(%p, 0);", &G_DB_ENV);
+            DBG("G_DB_ENV->close(%p, 0);\r\n", &G_DB_ENV);
             G_DB_ENV->close(G_DB_ENV, 0);
             G_DB_ENV = 0;
         }
@@ -354,7 +344,7 @@ DRIVER_INIT(bdberl_drv)
             }
             else
             {
-                fprintf(stderr, "Ignoring \"BDBERL_PAGE_SIZE\" value %u - not power of 2",
+                fprintf(stderr, "Ignoring \"BDBERL_PAGE_SIZE\" value %u - not power of 2\r\n",
                     page_size);
             }
         }
@@ -398,7 +388,7 @@ DRIVER_INIT(bdberl_drv)
     }
     else
     {
-        DBG("DRIVER INIT FAILED - %s\n", db_strerror(G_DB_ENV_ERROR));
+        DBG("DRIVER INIT FAILED - %s\r\n", db_strerror(G_DB_ENV_ERROR));
     }
 
     return &bdberl_drv_entry;
@@ -406,7 +396,8 @@ DRIVER_INIT(bdberl_drv)
 
 static ErlDrvData bdberl_drv_start(ErlDrvPort port, char* buffer)
 {
-    DBG("threadid %p port %p: BDB DRIVER STARTING\n", erl_drv_thread_self(), port);
+    DBG("threadid %p port %p: BDB DRIVER STARTING\r\n",
+        erl_drv_thread_self(), port);
 
     // Make sure we have a functional environment -- if we don't,
     // bail...
@@ -434,7 +425,7 @@ static ErlDrvData bdberl_drv_start(ErlDrvPort port, char* buffer)
     // Make sure port is running in binary mode
     set_port_control_flags(port, PORT_CONTROL_FLAG_BINARY);
 
-    DBGCMD(d, "BDB DRIVER STARTED\n");
+    DBGCMD(d, "BDB DRIVER STARTED");
 
     return (ErlDrvData)d;
 }
@@ -443,7 +434,7 @@ static void bdberl_drv_stop(ErlDrvData handle)
 {
     PortData* d = (PortData*)handle;
 
-    DBG("Stopping port %p\n", d->port);
+    DBG("Stopping port %p\r\n", d->port);
 
     // Grab the port lock, in case we have an async job running
     erl_drv_mutex_lock(d->port_lock);
@@ -452,13 +443,13 @@ static void bdberl_drv_stop(ErlDrvData handle)
     // block until the job has either been removed or has run
     if (d->async_job)
     {
-        DBGCMD(d, "Stopping port %p - cancelling async job %p\n", d->port, d->async_job);
+        DBGCMD(d, "Stopping port %p - cancelling async job %p\r\n", d->port, d->async_job);
 
         // Drop the lock prior to starting the wait for the async process
         erl_drv_mutex_unlock(d->port_lock);
 
         bdberl_tpool_cancel(d->async_pool, d->async_job);
-        DBGCMD(d, "Canceled async job for port: %p\n", d->port);
+        DBGCMD(d, "Canceled async job for port: %p\r\n", d->port);
     }
     else
     {
@@ -470,7 +461,7 @@ static void bdberl_drv_stop(ErlDrvData handle)
     erl_drv_mutex_destroy(d->port_lock);
 
     // If a cursor is open, close it
-    DBG("Stopping port %p - cleaning up cursors (%p) and transactions (%p)\n", d->port,
+    DBG("Stopping port %p - cleaning up cursors (%p) and transactions (%p)\r\n", d->port,
         d->cursor, d->txn);
 
     if (d->cursor)
@@ -482,13 +473,13 @@ static void bdberl_drv_stop(ErlDrvData handle)
     abort_txn(d);
 
     // Close all the databases we previously opened
-    DBG("Stopping port %p - closing all dbrefs\n", d->port);
+    DBG("Stopping port %p - closing all dbrefs\r\n", d->port);
     while (d->dbrefs)
     {
         int dbref = d->dbrefs->dbref;
         if (close_database(dbref, 0, d) != ERROR_NONE)
         {
-            DBG("Stopping port %p could not close dbref %d\n", d->port, dbref);
+            DBG("Stopping port %p could not close dbref %d\r\n", d->port, dbref);
         }
     }
 
@@ -497,7 +488,7 @@ static void bdberl_drv_stop(ErlDrvData handle)
     // unregister if it's already initialized to this port.
     if (G_LOG_PORT == d->port)
     {
-        DBG("Stopping port %p - removing logging port\n", d->port);
+        DBG("Stopping port %p - removing logging port\r\n", d->port);
 
         WRITE_LOCK(G_LOG_RWLOCK);
 
@@ -508,12 +499,11 @@ static void bdberl_drv_stop(ErlDrvData handle)
         // Unregister with BDB -- MUST DO THIS WITH WRITE LOCK HELD!
         G_DB_ENV->set_msgcall(G_DB_ENV, 0);
         G_DB_ENV->set_errcall(G_DB_ENV, 0);
-        G_DB_ENV->set_event_notify(G_DB_ENV, 0);
 
         WRITE_UNLOCK(G_LOG_RWLOCK);
     }
 
-    DBG("Stopped port: %p\n", d->port);
+    DBG("Stopped port: %p\r\n", d->port);
 
     // Release the port instance data
     driver_free(d->work_buffer);
@@ -522,7 +512,7 @@ static void bdberl_drv_stop(ErlDrvData handle)
 
 static void bdberl_drv_finish()
 {
-    DBG("BDB DRIVER FINISHING");
+    DBG("BDB DRIVER FINISHING\r\n");
     // Stop the thread pools
     if (G_TPOOL_GENERAL != NULL)
     {
@@ -601,7 +591,7 @@ static void bdberl_drv_finish()
         G_LOG_RWLOCK = NULL;
     }
 
-    DBG("BDB DRIVER FINISHED");
+    DBG("BDB DRIVER FINISHED\r\n");
 }
 
 static int bdberl_drv_control(ErlDrvData handle, unsigned int cmd,
@@ -800,8 +790,8 @@ static int bdberl_drv_control(ErlDrvData handle, unsigned int cmd,
 
         // Inbuf is <<Flags:32/native, KeyLen:32/native, KeyBin/bytes>>,
 
-        // If the working buffer is large enough, copy the data to put/get into it.
+        // If the working buffer is large enough, copy the data to put/get into it. Otherwise, realloc
-        // Otherwise, realloc until it is large enough
+        // until it is large enough
         if (d->work_buffer_sz < inbuf_sz)
         {
             d->work_buffer = driver_realloc(d->work_buffer, inbuf_sz);
@@ -855,26 +845,13 @@ static int bdberl_drv_control(ErlDrvData handle, unsigned int cmd,
         // Let caller know operation is in progress
         RETURN_INT(0, outbuf);
     }
-    case CMD_CURSOR_COUNT:
-    {
-        FAIL_IF_ASYNC_PENDING(d, outbuf);
-        FAIL_IF_NO_CURSOR(d, outbuf);
-
-        // Schedule the operation
-        d->async_op = cmd;
-        bdberl_general_tpool_run(&do_async_cursor_count, d, 0, &d->async_job);
-
-        // Let caller know operation is in progress
-        // Outbuf is: <<0:32>>
-        RETURN_INT(0, outbuf);
-    }
     case CMD_CURSOR_CLOSE:
     {
         FAIL_IF_ASYNC_PENDING(d, outbuf);
         FAIL_IF_NO_CURSOR(d, outbuf);
 
-        // It's possible to get a deadlock when closing a cursor,
+        // It's possible to get a deadlock when closing a cursor -- in that situation we also
-        // in that situation we also need to go ahead and abort the txn.
+        // need to go ahead and abort the txn
         int rc = d->cursor->close(d->cursor);
         if (d->txn && (rc == DB_LOCK_NOTGRANTED || rc == DB_LOCK_DEADLOCK))
         {
@@ -942,7 +919,6 @@ static int bdberl_drv_control(ErlDrvData handle, unsigned int cmd,
 
             G_DB_ENV->set_msgcall(G_DB_ENV, &bdb_msgcall);
             G_DB_ENV->set_errcall(G_DB_ENV, &bdb_errcall);
-            G_DB_ENV->set_event_notify(G_DB_ENV, &bdb_eventcall);
 
             WRITE_UNLOCK(G_LOG_RWLOCK);
         }
@@ -1032,16 +1008,16 @@ static int check_non_neg_env(char *env, unsigned int *val_ptr)
         long long val = strtoll(val_str, NULL, 0);
         if (val == 0 && errno == EINVAL)
         {
-            fprintf(stderr, "Ignoring \"%s\" value \"%s\" - invalid value\n", env, val_str);
+            fprintf(stderr, "Ignoring \"%s\" value \"%s\" - invalid value\r\n", env, val_str);
             return 0;
         }
         if (val <= 0 || val > UINT_MAX)
         {
-            fprintf(stderr, "Ignoring \"%s\" value \"%lld\" - out of range\n", env, val);
+            fprintf(stderr, "Ignoring \"%s\" value \"%lld\" - out of range\r\n", env, val);
             return 0;
         }
         unsigned int uival = (unsigned int) val;
-        DBG("Using \"%s\" value %u\n", env, uival);
+        DBG("Using \"%s\" value %u\r\n", env, uival);
         *val_ptr = uival;
         return 1;
     }
@@ -1067,7 +1043,7 @@ static int check_pos_env(char *env, unsigned int *val_ptr)
         }
         else
         {
-            fprintf(stderr, "Ignoring \"%s\" value \"%u\" - out of range\n", env, *val_ptr);
+            fprintf(stderr, "Ignoring \"%s\" value \"%u\" - out of range\r\n", env, *val_ptr);
             *val_ptr = original_val;
             return 0;
         }
@@ -1158,9 +1134,9 @@ static int open_database(const char* name, DBTYPE type, unsigned int flags, Port
 
         // Create the DB handle
         DB* db = NULL;
-        DBGCMD(data, "db_create(&db, %p, 0);\n", G_DB_ENV);
+        DBGCMD(data, "db_create(&db, %p, 0);", G_DB_ENV);
         int rc = db_create(&db, G_DB_ENV, 0);
-        DBGCMD(data, " = %s (%d) db = %p\n", rc == 0 ? "ok" : bdberl_rc_to_atom_str(rc), rc, db);
+        DBGCMD(data, "rc = %s (%d) db = %p", rc == 0 ? "ok" : bdberl_rc_to_atom_str(rc), rc, db);
         if (rc != 0)
         {
             // Failure while creating the database handle -- drop our lock and return
@@ -1174,14 +1150,12 @@ static int open_database(const char* name, DBTYPE type, unsigned int flags, Port
         {
             if (db->set_pagesize(db, G_PAGE_SIZE) != 0)
             {
-                bdb_errcall(G_DB_ENV, "\n", "Failed to set page size.");
+                bdb_errcall(G_DB_ENV, "", "Failed to set page size.");
             }
         }
 
-        flags |= DB_AUTO_COMMIT;
-
         // Attempt to open our database
-        DBGCMD(data, "db->open(%p, 0, '%s', 0, %x, %08x, 0);\n", db, name, type, flags);
+        DBGCMD(data, "db->open(%p, 0, '%s', 0, %x, %08x, 0);", db, name, type, flags);
         rc = db->open(db, 0, name, 0, type, flags, 0);
         DBGCMDRC(data, rc);
         if (rc != 0)
@@ -1235,7 +1209,7 @@ static int close_database(int dbref, unsigned flags, PortData* data)
         if (database->ports == 0)
         {
             // Close out the BDB handle
-            DBGCMD(data, "database->db->close(%p, %08x) (for dbref %d)\n", database->db, flags, dbref);
+            DBGCMD(data, "database->db->close(%p, %08x) (for dbref %d)", database->db, flags, dbref);
             rc = database->db->close(database->db, flags);
             DBGCMDRC(data, rc);
 
@@ -1271,7 +1245,7 @@ static void check_all_databases_closed()
         Database* database = &G_DATABASES[dbref];
         if (database->ports != NULL)
         {
-            fprintf(stderr, "BDBERL: Ports still open on '%s' dbref %d\n",
+            fprintf(stderr, "BDBERL: Ports still open on '%s' dbref %d\r\n",
                     database->name ? database->name : "no name", dbref);
         }
 
@@ -1280,7 +1254,7 @@ static void check_all_databases_closed()
             int flags = 0;
             DBG("final db->close(%p, %08x) (for dbref %d)", database->db, flags, dbref);
             rc = database->db->close(database->db, flags);
-            DBG(" = %s (%d)\n", rc == 0 ? "ok" : bdberl_rc_to_atom_str(rc), rc);
+            DBG(" = %s (%d)\r\n", rc == 0 ? "ok" : bdberl_rc_to_atom_str(rc), rc);
         }
     }
 
@@ -1293,7 +1267,7 @@ static void abort_txn(PortData* d)
 {
     if (d->txn)
     {
-        DBGCMD(d, "d->txn->abort(%p)\n", d->txn);
+        DBGCMD(d, "d->txn->abort(%p)", d->txn);
         int rc = d->txn->abort(d->txn);
         DBGCMDRC(d, rc);
         d->txn = NULL;
@@ -1313,7 +1287,7 @@ static int delete_database(const char* name, PortData *data)
     }
 
     // Good, database doesn't seem to be open -- attempt the delete
-    DBG("Attempting to delete database: %s\n", name);
+    DBG("Attempting to delete database: %s\r\n", name);
     int rc = G_DB_ENV->dbremove(G_DB_ENV, 0, name, 0, DB_AUTO_COMMIT);
     UNLOCK_DATABASES(data->port);
 
@@ -1489,29 +1463,6 @@ static int send_dir_info(ErlDrvPort port, ErlDrvTermData pid, const char *path)
 }
 
 
-static void send_error_response(ErlDrvPort port, ErlDrvTermData pid, int rc)
-{
-    // See if this is a standard errno that we have an erlang code for
-    char *error = bdberl_rc_to_atom_str(rc);
-    if (error != NULL)
-    {
-        ErlDrvTermData response[] = { ERL_DRV_ATOM,  driver_mk_atom("error"),
-                                      ERL_DRV_ATOM,  driver_mk_atom(error),
-                                      ERL_DRV_TUPLE, 2};
-        driver_send_term(port, pid, response, sizeof(response) / sizeof(response[0]));
-    }
-    else
-    {
-        ErlDrvTermData response[] = { ERL_DRV_ATOM, driver_mk_atom("error"),
-                                      ERL_DRV_ATOM, driver_mk_atom("unknown"),
-                                      ERL_DRV_INT,  rc,
-                                      ERL_DRV_TUPLE, 2,
-                                      ERL_DRV_TUPLE, 2};
-        driver_send_term(port, pid, response, sizeof(response) / sizeof(response[0]));
-    }
- }
-
-
 void bdberl_send_rc(ErlDrvPort port, ErlDrvTermData pid, int rc)
 {
     // TODO: May need to tag the messages a bit more explicitly so that if another async
@@ -1524,7 +1475,24 @@ void bdberl_send_rc(ErlDrvPort port, ErlDrvTermData pid, int rc)
     }
     else
     {
-      send_error_response(port, pid, rc);
+        // See if this is a standard errno that we have an erlang code for
+        char *error = bdberl_rc_to_atom_str(rc);
+        if (error != NULL)
+        {
+            ErlDrvTermData response[] = { ERL_DRV_ATOM,  driver_mk_atom("error"),
+                                          ERL_DRV_ATOM,  driver_mk_atom(error),
+                                          ERL_DRV_TUPLE, 2};
+            driver_send_term(port, pid, response, sizeof(response) / sizeof(response[0]));
+        }
+        else
+        {
+            ErlDrvTermData response[] = { ERL_DRV_ATOM, driver_mk_atom("error"),
+                                          ERL_DRV_ATOM, driver_mk_atom("unknown"),
+                                          ERL_DRV_INT,  rc,
+                                          ERL_DRV_TUPLE, 2,
+                                          ERL_DRV_TUPLE, 2};
+            driver_send_term(port, pid, response, sizeof(response) / sizeof(response[0]));
+        }
     }
 }
 
@@ -1542,32 +1510,6 @@ void bdberl_async_cleanup_and_send_rc(PortData* d, int rc)
     bdberl_send_rc(port, pid, rc);
 }
 
-static void async_cleanup_and_send_uint32(PortData* d, int rc, unsigned int value)
-{
-    // Save the port and pid references -- we need copies independent from the PortData
-    // structure. Once we release the port_lock after clearing the cmd, it's possible that
-    // the port could go away without waiting on us to finish. This is acceptable, but we need
-    // to be certain that there is no overlap of data between the two threads. driver_send_term
-    // is safe to use from a thread, even if the port you're sending from has already expired.
-    ErlDrvPort port = d->port;
-    ErlDrvTermData pid = d->port_owner;
-
-    bdberl_async_cleanup(d);
-
-    // Notify port of result
-    if (rc == 0)
-    {
-        ErlDrvTermData response[] = { ERL_DRV_ATOM, driver_mk_atom("ok"),
-                                      ERL_DRV_UINT, value,
-                                      ERL_DRV_TUPLE, 2};
-        driver_send_term(port, pid, response, sizeof(response) / sizeof(response[0]));
-    }
-    else
-    {
-        send_error_response(port, pid, rc);
-    }
-}
-
 static void async_cleanup_and_send_kv(PortData* d, int rc, DBT* key, DBT* value)
 {
     // Save the port and pid references -- we need copies independent from the PortData
@@ -1596,7 +1538,24 @@ static void async_cleanup_and_send_kv(PortData* d, int rc, DBT* key, DBT* value)
     }
     else
     {
-        send_error_response(port, pid, rc);
+        // See if this is a standard errno that we have an erlang code for
+        char *error = bdberl_rc_to_atom_str(rc);
+        if (error != NULL)
+        {
+            ErlDrvTermData response[] = { ERL_DRV_ATOM,  driver_mk_atom("error"),
+                                          ERL_DRV_ATOM,  driver_mk_atom(error),
+                                          ERL_DRV_TUPLE, 2};
+            driver_send_term(port, pid, response, sizeof(response) / sizeof(response[0]));
+        }
+        else
+        {
+            ErlDrvTermData response[] = { ERL_DRV_ATOM, driver_mk_atom("error"),
+                                          ERL_DRV_ATOM, driver_mk_atom("unknown"),
+                                          ERL_DRV_INT,  rc,
+                                          ERL_DRV_TUPLE, 2,
+                                          ERL_DRV_TUPLE, 2};
+            driver_send_term(port, pid, response, sizeof(response) / sizeof(response[0]));
+        }
     }
 }
 
@@ -1631,14 +1590,14 @@ static void do_async_put(void* arg)
     int rc;
     if (calc_crc32 != buf_crc32)
     {
-        DBGCMD(d, "CRC-32 error on put data - buffer %08X calculated %08X.\n", buf_crc32, calc_crc32);
+        DBGCMD(d, "CRC-32 error on put data - buffer %08X calculated %08X.", buf_crc32, calc_crc32);
         rc = ERROR_INVALID_VALUE;
     }
     else
     {
         // Execute the actual put. All databases are opened with AUTO_COMMIT, so if msg->port->txn
         // is NULL, the put will still be atomic
-        DBGCMD(d, "db->put(%p, %p, %p, %p, %08X) dbref %d key=%p(%d) value=%p(%d)\n",
+        DBGCMD(d, "db->put(%p, %p, %p, %p, %08X) dbref %d key=%p(%d) value=%p(%d)",
                db, d->txn, &key, &value, flags, dbref, key.data, key.size, value.data, value.size);
         rc = db->put(db, d->txn, &key, &value, flags);
         DBGCMDRC(d, rc);
@@ -1698,7 +1657,7 @@ static void do_async_get(void* arg)
 
         if (calc_crc32 != buf_crc32)
         {
-            DBGCMD(d, "CRC-32 error on get data - buffer %08X calculated %08X.\n",
+            DBGCMD(d, "CRC-32 error on get data - buffer %08X calculated %08X.",
                    buf_crc32, calc_crc32);
             rc = ERROR_INVALID_VALUE;
         }
@@ -1757,15 +1716,15 @@ static void do_async_txnop(void* arg)
     int rc = 0;
     if (d->async_op == CMD_TXN_BEGIN)
     {
-        DBGCMD(d, "G_DB_ENV->txn_begin(%p, 0, %p, %08X)\n", G_DB_ENV, d->txn, d->async_flags);
+        DBGCMD(d, "G_DB_ENV->txn_begin(%p, 0, %p, %08X)", G_DB_ENV, d->txn, d->async_flags);
         rc = G_DB_ENV->txn_begin(G_DB_ENV, 0, &(d->txn), d->async_flags);
-        DBGCMD(d, "rc = %s (%d) d->txn = %p\n", rc == 0 ? "ok" : bdberl_rc_to_atom_str(rc), rc, d->txn);
+        DBGCMD(d, "rc = %s (%d) d->txn = %p", rc == 0 ? "ok" : bdberl_rc_to_atom_str(rc), rc, d->txn);
 
     }
     else if (d->async_op == CMD_TXN_COMMIT)
     {
         assert(d->txn != NULL);
-        DBGCMD(d, "d->txn->txn_commit(%p, %08X)\n", d->txn, d->async_flags);
+        DBGCMD(d, "d->txn->txn_commit(%p, %08X)", d->txn, d->async_flags);
         rc = d->txn->commit(d->txn, d->async_flags);
         DBGCMDRC(d, rc);
         d->txn = 0;
@@ -1782,10 +1741,10 @@ static void do_async_txnop(void* arg)
 
 static void do_async_cursor_put(void* arg)
 {
-    PortData* d = (PortData*)arg;
+  PortData* d = (PortData*)arg;
-    assert(d->cursor != NULL);
+  assert(d->cursor != NULL);
-    DBGCMD(d, "cursor_put/2 not yet implemented...\n"); /* TODO: implement this. */
+  DBGCMD(d, "cursor_put/2 not yet implemented..."); /* TODO: implement this. */
-    bdberl_async_cleanup_and_send_rc(d, ERROR_DB_ACTIVE);
+  bdberl_async_cleanup_and_send_rc(d, ERROR_DB_ACTIVE);
 }
 
 
@@ -1812,7 +1771,7 @@ static void do_async_cursor_get(void* arg)
     value.flags = DB_DBT_MALLOC;
 
     // Execute the operation
-    DBGCMD(d, "d->cursor->get(%p, %p, %p, %08X\n);", d->cursor, &key, &value, flags);
+    DBGCMD(d, "d->cursor->get(%p, %p, %p, %08X);", d->cursor, &key, &value, flags);
     int rc = d->cursor->get(d->cursor, &key, &value, flags);
     DBGCMDRC(d, rc);
 
@@ -1825,7 +1784,7 @@ static void do_async_cursor_get(void* arg)
 
         if (calc_crc32 != buf_crc32)
         {
-            DBGCMD(d, "CRC-32 error on get data - buffer %08X calculated %08X.\n",
+            DBGCMD(d, "CRC-32 error on get data - buffer %08X calculated %08X.",
                    buf_crc32, calc_crc32);
             rc = ERROR_INVALID_VALUE;
         }
@@ -1850,27 +1809,10 @@ static void do_async_cursor_get(void* arg)
 
 static void do_async_cursor_del(void* arg)
 {
-    PortData* d = (PortData*)arg;
+  PortData* d = (PortData*)arg;
-    assert(d->cursor != NULL);
+  assert(d->cursor != NULL);
-    DBGCMD(d, "cursor_del/2 not yet implemented...\n"); /* TODO: implement this. */
+  DBGCMD(d, "cursor_del/2 not yet implemented..."); /* TODO: implement this. */
-    bdberl_async_cleanup_and_send_rc(d, ERROR_DB_ACTIVE);
+  bdberl_async_cleanup_and_send_rc(d, ERROR_DB_ACTIVE);
-}
-
-
-static void do_async_cursor_count(void* arg)
-{
-    PortData* d = (PortData*)arg;
-    assert(d->cursor != NULL);
-
-    // Place to store the record count.
-    db_recno_t count = 0;
-
-    // Execute the operation
-    DBGCMD(d, "d->cursor->count(%p, %p, %08X);\n", d->cursor, &count, 0);
-    int rc = d->cursor->count(d->cursor, &count, 0);
-    DBGCMDRC(d, rc);
-
-    async_cleanup_and_send_uint32(d, rc, count);
 }
 
 
@@ -1899,7 +1841,7 @@ static void do_async_cursor_cnp(void* arg)
     }
 
     // Execute the operation
-    DBGCMD(d, "d->cursor->get(%p, %p, %p, %08X);\n", d->cursor, &key, &value, flags);
+    DBGCMD(d, "d->cursor->get(%p, %p, %p, %08X);", d->cursor, &key, &value, flags);
     int rc = d->cursor->get(d->cursor, &key, &value, flags);
     DBGCMDRC(d, rc);
 
@@ -1940,7 +1882,7 @@ static void do_async_truncate(void* arg)
 
     if (d->async_dbref == -1)
     {
-        DBG("Truncating all open databases...");
+        DBG("Truncating all open databases...\r\n");
 
         // Iterate over the whole database list skipping null entries
         int i = 0; // I hate C
@@ -1952,9 +1894,10 @@ static void do_async_truncate(void* arg)
                 DB* db = database->db;
                 u_int32_t count = 0;
 
-                DBGCMD(d, "db->truncate(%p, %p, %p, 0) dbref=%d\n", db, d->txn, &count, i);
+                DBGCMD(d, "db->truncate(%p, %p, %p, 0) dbref=%d", db, d->txn, &count, i);
                 rc = db->truncate(db, d->txn, &count, 0);
-                DBGCMD(d, "rc = %s (%d) count=%d\n", rc == 0 ? "ok" : bdberl_rc_to_atom_str(rc), rc, count);
+                DBGCMD(d, "rc = %s (%d) count=%d",
+                       rc == 0 ? "ok" : bdberl_rc_to_atom_str(rc), rc, count);
 
                 if (rc != 0)
                 {
@@ -1967,9 +1910,9 @@ static void do_async_truncate(void* arg)
     {
         DB* db = G_DATABASES[d->async_dbref].db;
         u_int32_t count = 0;
-        DBGCMD(d, "db->truncate(%p, %p, %p, 0) dbref=%d\n", db, d->txn, &count, d->async_dbref);
+        DBGCMD(d, "db->truncate(%p, %p, %p, 0) dbref=%d", db, d->txn, &count, d->async_dbref);
         rc = db->truncate(db, d->txn, &count, 0);
-        DBGCMD(d, "rc = %s (%d) count=%d\n", rc == 0 ? "ok" : bdberl_rc_to_atom_str(rc),
+        DBGCMD(d, "rc = %s (%d) count=%d", rc == 0 ? "ok" : bdberl_rc_to_atom_str(rc),
                rc, count);
     }
 
@@ -2112,7 +2055,7 @@ static void* driver_calloc(unsigned int size)
 
 static int add_portref(int dbref, ErlDrvPort port)
 {
-    DBG("Adding port %p to dbref %d\n", port, dbref);
+    DBG("Adding port %p to dbref %d\r\n", port, dbref);
     PortList* current = G_DATABASES[dbref].ports;
     if (current)
     {
@@ -2147,7 +2090,7 @@ static int add_portref(int dbref, ErlDrvPort port)
 
 static int del_portref(int dbref, ErlDrvPort port)
 {
-    DBG("Deleting port %p from dbref %d\n", port, dbref);
+    DBG("Deleting port %p from dbref %d\r\n", port, dbref);
     PortList* current = G_DATABASES[dbref].ports;
     PortList* last = 0;
     assert(current != NULL);
@@ -2183,7 +2126,7 @@ static int del_portref(int dbref, ErlDrvPort port)
  */
 static int add_dbref(PortData* data, int dbref)
 {
-    DBG("Adding dbref %d to port %p\n", dbref, data->port);
+    DBG("Adding dbref %d to port %p\r\n", dbref, data->port);
     DbRefList* current = data->dbrefs;
     if (current)
     {
@@ -2220,7 +2163,7 @@ static int add_dbref(PortData* data, int dbref)
  */
 static int del_dbref(PortData* data, int dbref)
 {
-    DBG("Deleting dbref %d from port %p\n", dbref, data->port);
+    DBG("Deleting dbref %d from port %p\r\n", dbref, data->port);
 
     DbRefList* current = data->dbrefs;
     DbRefList* last = 0;
@@ -2372,7 +2315,7 @@ static void* deadlock_check(void* arg)
         }
         if (count > 0)
         {
-            DBG("Rejected deadlocks: %d\n", count);
+            DBG("Rejected deadlocks: %d\r\n", count);
         }
 
         if (G_DEADLOCK_CHECK_INTERVAL > 0)
@@ -2381,7 +2324,7 @@ static void* deadlock_check(void* arg)
         }
     }
 
-    DBG("Deadlock checker exiting.");
+    DBG("Deadlock checker exiting.\r\n");
     return 0;
 }
 
@@ -2450,7 +2393,6 @@ static void bdb_errcall(const DB_ENV* dbenv, const char* errpfx, const char* msg
                                   ERL_DRV_STRING, (ErlDrvTermData)msg, (ErlDrvUInt)strlen(msg),
                                   ERL_DRV_TUPLE, 2};
     send_log_message(response, sizeof(response));
-    DBG("BDB error call: %s\n", msg);
 }
 
 static void bdb_msgcall(const DB_ENV* dbenv, const char* msg)
@@ -2459,35 +2401,6 @@ static void bdb_msgcall(const DB_ENV* dbenv, const char* msg)
                                   ERL_DRV_STRING, (ErlDrvTermData)msg, (ErlDrvUInt)strlen(msg),
                                   ERL_DRV_TUPLE, 2};
     send_log_message(response, sizeof(response));
-    DBG("BDB message call: %s\n", msg);
-}
-
-static void bdb_eventcall(DB_ENV* dbenv, u_int32_t type, void* info)
-{
-    switch(type)
-    {
-    case DB_EVENT_PANIC:
-    {
-        const char *msg = "panic";
-        ErlDrvTermData response[] = { ERL_DRV_ATOM, driver_mk_atom("bdb_event_notify"),
-                                      ERL_DRV_STRING, (ErlDrvTermData)msg, (ErlDrvUInt)strlen(msg),
-                                      ERL_DRV_TUPLE, 2};
-        // TODO clearly something should be done to shut things down cleanly and restart (how?)
-        send_log_message(response, sizeof(response));
-        DBG("BDB panic event\n");
-        break;
-    }
-    case DB_EVENT_WRITE_FAILED:
-    {
-        const char *msg = "write failed";
-        ErlDrvTermData response[] = { ERL_DRV_ATOM, driver_mk_atom("bdb_event_notify"),
-                                      ERL_DRV_STRING, (ErlDrvTermData)msg, (ErlDrvUInt)strlen(msg),
-                                      ERL_DRV_TUPLE, 2};
-        send_log_message(response, sizeof(response));
-        DBG("BDB write failed event\n");
-        break;
-    }
-    }
 }
 
 static void send_log_message(ErlDrvTermData* msg, int elements)
@@ -2501,7 +2414,7 @@ static void send_log_message(ErlDrvTermData* msg, int elements)
 }
 
 #ifdef DEBUG
-void bdberl_dbg(const char *fmt, ...)
+static void bdberl_dbgcmd(PortData *d, const char *fmt, ...)
 {
     char buf[1024];
 
@@ -2510,27 +2423,12 @@ void bdberl_dbg(const char *fmt, ...)
     vsnprintf(buf, sizeof(buf), fmt, ap);
     va_end(ap);
 
-    (void)fprintf(stderr, "%s", buf);
+    (void)fprintf(stderr, "threadid %p port %p: %s\r\n", erl_drv_thread_self(), d->port, buf);
-    (void)fflush(stderr);
 }
 
-void bdberl_dbgcmd(PortData *d, const char *fmt, ...)
+static void bdberl_dbgcmdrc(PortData *d, int rc)
 {
-    char buf[1024];
+    (void)fprintf(stderr, "threadid %p port %p: rc = %s (%d)\r\n",
-
-    va_list ap;
-    va_start(ap, fmt);
-    vsnprintf(buf, sizeof(buf), fmt, ap);
-    va_end(ap);
-
-    (void)fprintf(stderr, "threadid %p port %p: %s", erl_drv_thread_self(), d->port, buf);
-    (void)fflush(stderr);
-}
-
-void bdberl_dbgcmdrc(PortData *d, int rc)
-{
-    (void)fprintf(stderr, "threadid %p port %p: rc = %s (%d)\n",
                   erl_drv_thread_self(), d->port, rc == 0 ? "ok" : bdberl_rc_to_atom_str(rc), rc);
-    (void)fflush(stderr);
 }
 #endif // DEBUG

c_src/bdberl_drv.h

@@ -53,6 +53,9 @@
 #define CMD_CURSOR_NEXT      12
 #define CMD_CURSOR_PREV      13
 #define CMD_CURSOR_CLOSE     14
+#define CMD_CURSOR_GET       35 /* TODO: renumber these next 3 and match them with bdberl.hrl */
+#define CMD_CURSOR_PUT       36
+#define CMD_CURSOR_DEL       37
 #define CMD_PUT_COMMIT       15
 #define CMD_REMOVE_DB        16
 #define CMD_TRUNCATE         17
@@ -73,10 +76,6 @@
 #define CMD_DATA_DIRS_INFO   32
 #define CMD_LOG_DIR_INFO     33
 #define CMD_DRIVER_INFO      34
-#define CMD_CURSOR_GET       35
-#define CMD_CURSOR_PUT       36
-#define CMD_CURSOR_DEL       37
-#define CMD_CURSOR_COUNT     38
 
 /**
  * Command status values
@@ -250,21 +249,4 @@ void bdberl_txn_tpool_run(TPoolJobFunc main_fn,  PortData* d, TPoolJobFunc cance
         RETURN_INT(0, outbuf);                                  \
     }}
 
-
-/**
- * Helpful macros
- */
-#ifdef DEBUG
-#  define DBG(...) bdberl_dbg(__VA_ARGS__)
-#  define DBGCMD(P, ...)   bdberl_dbgcmd(P, __VA_ARGS__)
-#  define DBGCMDRC(P, ...) bdberl_dbgcmdrc(P, __VA_ARGS__)
-extern void bdberl_dbg(const char * fmt, ...);
-extern void bdberl_dbgcmd(PortData *d, const char *fmt, ...);
-extern void bdberl_dbgcmdrc(PortData *d, int rc);
-#else
-#  define DBG(arg1,...)
-#  define DBGCMD(d, fmt, ...)
-#  define DBGCMDRC(d, rc) { while (0) { rc++; } }  // otherwise get unused variable error
-#endif
-
 #endif //_BDBERL_DRV

c_src/bdberl_tpool.c

@@ -1,6 +1,6 @@
 /* -------------------------------------------------------------------
  *
- * bdberl: Thread Pool
+ * bdberl: Thread Pool 
  * Copyright (c) 2008-9 The Hive http://www.thehive.com/
  * Authors: Dave "dizzyd" Smith <dizzyd@dizzyd.com>
  *          Phil Toland <phil.toland@gmail.com>
@@ -25,17 +25,11 @@
  * THE SOFTWARE.
  *
  * ------------------------------------------------------------------- */
-
-#include <db.h>
-#include "bdberl_drv.h"
 #include "bdberl_tpool.h"
 
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
-#include <signal.h>
-#include <errno.h>
-#include <pthread.h>
 
 static void* bdberl_tpool_main(void* tpool);
 static TPoolJob* next_job(TPool* tpool);
@@ -43,9 +37,8 @@ static int remove_pending_job(TPool* tpool, TPoolJob* job);
 static void cleanup_job(TPool* tpool, TPoolJob* job);
 static int is_active_job(TPool* tpool, TPoolJob* job);
 
-#define LOCK(tpool) erl_drv_mutex_lock(tpool->lock)
+#define LOCK(t) erl_drv_mutex_lock(tpool->lock)
-#define UNLOCK(tpool) erl_drv_mutex_unlock(tpool->lock)
+#define UNLOCK(t) erl_drv_mutex_unlock(tpool->lock)
-
 
 TPool* bdberl_tpool_start(unsigned int thread_count)
 {
@@ -63,13 +56,10 @@ TPool* bdberl_tpool_start(unsigned int thread_count)
     int i;
     for (i = 0; i < thread_count; i++)
     {
-        int rc = erl_drv_thread_create("bdberl_tpool_thread", &(tpool->threads[i]), &bdberl_tpool_main, (void*)tpool, 0);
+        // TODO: Figure out good way to deal with errors in this situation (should be rare, but still...)
-        if (0 != rc) {
+        erl_drv_thread_create("bdberl_tpool_thread", &(tpool->threads[i]), &bdberl_tpool_main, (void*)tpool, 0);
-            // TODO: Figure out good way to deal with errors in this situation (should be rare, but still...)
-            fprintf(stderr, "Failed to spawn an erlang thread for the BDB thread pools! %s\n", erl_errno_id(rc));
-            fflush(stderr);
-        }
     }
+
     return tpool;
 }
 
@@ -88,7 +78,7 @@ void bdberl_tpool_stop(TPool* tpool)
     {
         erl_drv_cond_wait(tpool->work_cv, tpool->lock);
     }
-
+    
     // Join up with all the workers
     int i = 0;
     for (i = 0; i < tpool->thread_count; i++)
@@ -96,7 +86,7 @@ void bdberl_tpool_stop(TPool* tpool)
         erl_drv_thread_join(tpool->threads[i], 0);
     }
 
-    // Cleanup
+    // Cleanup 
     erl_drv_cond_destroy(tpool->work_cv);
     erl_drv_cond_destroy(tpool->cancel_cv);
     driver_free(tpool->threads);
@@ -105,7 +95,7 @@ void bdberl_tpool_stop(TPool* tpool)
     driver_free(tpool);
 }
 
-void bdberl_tpool_run(TPool* tpool, TPoolJobFunc main_fn, void* arg, TPoolJobFunc cancel_fn,
+void bdberl_tpool_run(TPool* tpool, TPoolJobFunc main_fn, void* arg, TPoolJobFunc cancel_fn, 
                       TPoolJob** job_ptr)
 {
     // Allocate and fill a new job structure
@@ -117,7 +107,7 @@ void bdberl_tpool_run(TPool* tpool, TPoolJobFunc main_fn, void* arg, TPoolJobFun
 
     // Sync up with the tpool and add the job to the pending queue
     LOCK(tpool);
-
+    
     if (tpool->pending_jobs)
     {
         // Make sure the current last job points to this one next
@@ -132,7 +122,7 @@ void bdberl_tpool_run(TPool* tpool, TPoolJobFunc main_fn, void* arg, TPoolJobFun
     tpool->last_pending_job = job;
     tpool->pending_job_count++;
 
-    // Generate a notification that there is work todo.
+    // Generate a notification that there is work todo. 
     // TODO: I think this may not be necessary, in the case where there are already other
     // pending jobs. Not sure ATM, however, so will be on safe side
     erl_drv_cond_broadcast(tpool->work_cv);
@@ -143,7 +133,7 @@ void bdberl_tpool_cancel(TPool* tpool, TPoolJob* job)
 {
     LOCK(tpool);
 
-    // Remove the job from the pending queue
+    // Remove the job from the pending queue 
     if (remove_pending_job(tpool, job))
     {
         // Job was removed from pending -- unlock and notify the job that it got canceled
@@ -159,7 +149,7 @@ void bdberl_tpool_cancel(TPool* tpool, TPoolJob* job)
         return;
     }
 
-    // Job not in the pending queue -- check the active queue.
+    // Job not in the pending queue -- check the active queue. 
     if (is_active_job(tpool, job))
     {
         // Job is currently active -- mark it as cancelled (so we get notified) and wait for it
@@ -169,7 +159,7 @@ void bdberl_tpool_cancel(TPool* tpool, TPoolJob* job)
             erl_drv_cond_wait(tpool->cancel_cv, tpool->lock);
         }
 
-        // Job is no longer running and should now be considered dead. Cleanup is handled by
+        // Job is no longer running and should now be considered dead. Cleanup is handled by 
         // the worker.
         UNLOCK(tpool);
         return;
@@ -222,7 +212,7 @@ static void* bdberl_tpool_main(void* arg)
             {
                 erl_drv_cond_broadcast(tpool->cancel_cv);
             }
-
+        
             // Cleanup the job (remove from active list, free, etc.)
             cleanup_job(tpool, job);
         }
@@ -328,7 +318,7 @@ static void cleanup_job(TPool* tpool, TPoolJob* job)
             {
                 tpool->active_jobs = current->next;
             }
-
+            
             break;
         }
 
@@ -358,7 +348,7 @@ static int is_active_job(TPool* tpool, TPoolJob* job)
 }
 
 // Return the number of pending and active jobs
-void bdberl_tpool_job_count(TPool* tpool, unsigned int *pending_count_ptr,
+void bdberl_tpool_job_count(TPool* tpool, unsigned int *pending_count_ptr, 
                              unsigned int *active_count_ptr)
 {
     LOCK(tpool);
@@ -366,38 +356,3 @@ void bdberl_tpool_job_count(TPool* tpool, unsigned int *pending_count_ptr,
     *active_count_ptr = tpool->active_job_count;
     UNLOCK(tpool);
 }
-
-// Returns a unique identifier pair for the current thread of control
-void bdberl_tpool_thread_id(DB_ENV *env, pid_t *pid, db_threadid_t *tid)
-{
-    if (pid)
-        *pid = getpid();
-    if (tid)
-      *tid = (db_threadid_t)pthread_self();
-}
-
-char *bdberl_tpool_thread_id_string(DB_ENV *dbenv, pid_t pid, db_threadid_t tid, char *buf)
-{
-    snprintf(buf, DB_THREADID_STRLEN, "%d/%p", (unsigned int)pid, (void *)tid);
-    return buf;
-}
-
-// Returns non-zero if the thread of control, identified by the pid and tid arguments,
-// is still running.
-// If DB_MUTEX_PROCESS_ONLY is set in flags then return only if the process (pid) is
-// alive, ignore the thread ID.
-int bdberl_tpool_thread_is_alive(DB_ENV *dbenv, pid_t pid, db_threadid_t tid, u_int32_t flags)
-{
-    int alive = 0;
-
-    if (kill(pid, 0) != ESRCH)
-    {
-        if (flags & DB_MUTEX_PROCESS_ONLY)
-            alive = 1;
-        else
-            if (pthread_kill(tid, 0) != ESRCH)
-              alive = 1;
-    }
-    DBG("bdberl_tpool_thread_is_alive(%08X, %08X, %d) = %d\n", pid, tid, flags, alive);
-    return alive;
-}

c_src/bdberl_tpool.h

@@ -1,6 +1,6 @@
 /* -------------------------------------------------------------------
  *
- * bdberl: Thread Pool
+ * bdberl: Thread Pool 
  * Copyright (c) 2008-9 The Hive http://www.thehive.com/
  * Authors: Dave "dizzyd" Smith <dizzyd@dizzyd.com>
  *          Phil Toland <phil.toland@gmail.com>
@@ -74,7 +74,7 @@ typedef struct
     unsigned int active_threads;
 
     unsigned int shutdown;
-
+    
 } TPool;
 
 TPool* bdberl_tpool_start(unsigned int thread_count);
@@ -86,13 +86,7 @@ void bdberl_tpool_run(TPool* tpool, TPoolJobFunc main_fn, void* arg, TPoolJobFun
 
 void bdberl_tpool_cancel(TPool* tpool, TPoolJob* job);
 
-void bdberl_tpool_job_count(TPool* tpool, unsigned int *pending_count_ptr,
+void bdberl_tpool_job_count(TPool* tpool, unsigned int *pending_count_ptr, 
                             unsigned int *active_count_ptr);
 
-void bdberl_tpool_thread_id(DB_ENV *env, pid_t *pid, db_threadid_t *tid);
-
-char *bdberl_tpool_thread_id_string(DB_ENV *dbenv, pid_t pid, db_threadid_t tid, char *buf);
-
-int bdberl_tpool_thread_is_alive(DB_ENV *dbenv, pid_t pid, db_threadid_t tid, u_int32_t flags);
-
 #endif

c_src/configure_bdb.sh

@@ -2,8 +2,7 @@
 
 cd $1/build_unix && \
     ../dist/configure --disable-shared --enable-static --with-pic \
-        --disable-heap --disable-queue --disable-replication \
+        --disable-heap --disable-queue \
-        --enable-o_direct --enable-o_direct \
+        --disable-partition --disable-replication \
-        --enable-debug --enable-diagnostics \
+        --enable-o_direct \
-        --enable-dtrace \
         --prefix=$2

include/bdberl.hrl

@@ -46,6 +46,9 @@
 -define(CMD_CURSOR_NEXT,     12).
 -define(CMD_CURSOR_PREV,     13).
 -define(CMD_CURSOR_CLOSE,    14).
+-define(CMD_CURSOR_GET,      35). %% TODO: renumber these 3 and match them to bdberl_drv.h
+-define(CMD_CURSOR_PUT,      36).
+-define(CMD_CURSOR_DEL,      37).
 -define(CMD_PUT_COMMIT,      15).
 -define(CMD_REMOVE_DB,       16).
 -define(CMD_TRUNCATE,        17).
@@ -66,10 +69,6 @@
 -define(CMD_DATA_DIRS_INFO,  32).
 -define(CMD_LOG_DIR_INFO,    33).
 -define(CMD_DRIVER_INFO,     34).
--define(CMD_CURSOR_GET,      35).
--define(CMD_CURSOR_PUT,      36).
--define(CMD_CURSOR_DEL,      37).
--define(CMD_CURSOR_COUNT,    38).
 
 -define(DB_TYPE_BTREE, 1).
 -define(DB_TYPE_HASH,  2).
@@ -385,28 +384,3 @@
 -define(DB_UPDATE_SECONDARY,          29).
 -define(DB_SET_LTE,          30).
 -define(DB_GET_BOTH_LTE,     31).
-
-%% DB Event notification types.
--define(DB_EVENT_PANIC,                  0).
--define(DB_EVENT_REG_ALIVE,              1).
--define(DB_EVENT_REG_PANIC,              2).
--define(DB_EVENT_REP_CLIENT,             3).
--define(DB_EVENT_REP_CONNECT_BROKEN,     4).
--define(DB_EVENT_REP_CONNECT_ESTD,       5).
--define(DB_EVENT_REP_CONNECT_TRY_FAILED,         6).
--define(DB_EVENT_REP_DUPMASTER,          7).
--define(DB_EVENT_REP_ELECTED,            8).
--define(DB_EVENT_REP_ELECTION_FAILED,    9).
--define(DB_EVENT_REP_INIT_DONE,         10).
--define(DB_EVENT_REP_JOIN_FAILURE,      11).
--define(DB_EVENT_REP_LOCAL_SITE_REMOVED,        12).
--define(DB_EVENT_REP_MASTER,            13).
--define(DB_EVENT_REP_MASTER_FAILURE,    14).
--define(DB_EVENT_REP_NEWMASTER,         15).
--define(DB_EVENT_REP_PERM_FAILED,       16).
--define(DB_EVENT_REP_SITE_ADDED,                17).
--define(DB_EVENT_REP_SITE_REMOVED,      18).
--define(DB_EVENT_REP_STARTUPDONE,       19).
--define(DB_EVENT_REP_WOULD_ROLLBACK,    20).
--define(DB_EVENT_WRITE_FAILED,          21).
--define(DB_EVENT_NO_SUCH_EVENT,          0xffffffff).

src/bdberl.erl

@@ -68,7 +68,6 @@
          delete_database/1,
          cursor_open/1, cursor_next/0, cursor_prev/0, cursor_current/0, cursor_close/0,
          cursor_get/0, cursor_get/1, cursor_get/2, %TODO: cursor_del/2, cursor_del/3, cursor_put/2, cursor_put/3,
-         cursor_count/0,
          driver_info/0,
          register_logger/0,
          stop/0]).
@@ -105,7 +104,7 @@
 %% @spec open(Name, Type) -> {ok, Db} | {error, Error}
 %% where
 %%    Name = string()
-%%    Type = btree | hash | queue
+%%    Type = btree | hash
 %%    Db = integer()
 %%
 %% @equiv open(Name, Type, [create])
@@ -195,8 +194,8 @@ open(Name, Type, Opts) ->
     end,
     Flags = process_flags(lists:umerge(Opts, [auto_commit, threaded])),
     Cmd = <<Flags:32/native, TypeCode:8/signed-native, (list_to_binary(Name))/bytes, 0:8/native>>,
-    <<Result:32/signed-native>> = erlang:port_control(get_port(), ?CMD_OPEN_DB, Cmd),
+    <<Rc:32/signed-native>> = erlang:port_control(get_port(), ?CMD_OPEN_DB, Cmd),
-    recv_val(Result).
+    recv_val(Rc).
 
 
 %%--------------------------------------------------------------------
@@ -257,8 +256,8 @@ close(Db) ->
 close(Db, Opts) ->
     Flags = process_flags(Opts),
     Cmd = <<Db:32/signed-native, Flags:32/native>>,
-    <<Result:32/signed-native>> = erlang:port_control(get_port(), ?CMD_CLOSE_DB, Cmd),
+    <<Rc:32/signed-native>> = erlang:port_control(get_port(), ?CMD_CLOSE_DB, Cmd),
-    recv_ok(Result).
+    recv_ok(Rc).
 
 %%--------------------------------------------------------------------
 %% @doc
@@ -273,7 +272,7 @@ close(Db, Opts) ->
 -spec txn_begin() -> ok | db_error().
 
 txn_begin() ->
-    txn_begin([]).
+    txn_begin([txn_snapshot]).
 
 
 %%--------------------------------------------------------------------
@@ -365,8 +364,8 @@ txn_begin() ->
 txn_begin(Opts) ->
     Flags = process_flags(Opts),
     Cmd = <<Flags:32/native>>,
-    <<Result:32/signed-native>> = erlang:port_control(get_port(), ?CMD_TXN_BEGIN, Cmd),
+    <<Rc:32/signed-native>> = erlang:port_control(get_port(), ?CMD_TXN_BEGIN, Cmd),
-    recv_ok(Result).
+    recv_ok(Rc).
 
 %%--------------------------------------------------------------------
 %% @doc
@@ -428,8 +427,8 @@ txn_commit() ->
 txn_commit(Opts) ->
     Flags = process_flags(Opts),
     Cmd = <<Flags:32/native>>,
-    <<Result:32/signed-native>> = erlang:port_control(get_port(), ?CMD_TXN_COMMIT, Cmd),
+    <<Rc:32/signed-native>> = erlang:port_control(get_port(), ?CMD_TXN_COMMIT, Cmd),
-    recv_ok(Result).
+    recv_ok(Rc).
 
 %%--------------------------------------------------------------------
 %% @doc
@@ -596,7 +595,7 @@ transaction(Fun, Retries, TimeLeft, Opts) ->
                     transaction(Fun, R, T, Opts);
 
                 _ : Reason ->
-                    lager:info("function threw non-lock error - ~p", [{Reason, erlang:get_stacktrace()}]),
+                    lager:info("function threw non-lock error - ~p", [Reason]),
                     ok = txn_abort(),
                     {error, {transaction_failed, Reason}}
             end;
@@ -834,7 +833,7 @@ put_commit_r(Db, Key, Value, Opts) ->
 %% @end
 %%--------------------------------------------------------------------
 -spec get(Db :: db(), Key :: db_key()) ->
-    not_found | {ok, db_ret_value()} | {error, db_error()}.
+    not_found | {ok, db_ret_value()} | db_error().
 
 get(Db, Key) ->
     get(Db, Key, []).
@@ -885,7 +884,7 @@ get(Db, Key) ->
 %% @end
 %%--------------------------------------------------------------------
 -spec get(Db :: db(), Key :: db_key(), Opts :: db_flags()) ->
-    not_found | {ok, db_ret_value()} | {error, db_error()}.
+    not_found | {ok, db_ret_value()} | db_error().
 
 get(Db, Key, Opts) ->
     {KeyLen, KeyBin} = to_binary(Key),
@@ -964,6 +963,8 @@ get_r(Db, Key, Opts) ->
     end.
 
 
+
+
 %%--------------------------------------------------------------------
 %% @doc
 %% Delete a value based on key.
@@ -1172,8 +1173,8 @@ truncate() ->
 
 truncate(Db) ->
     Cmd = <<Db:32/signed-native>>,
-    <<Result:32/signed-native>> = erlang:port_control(get_port(), ?CMD_TRUNCATE, Cmd),
+    <<Rc:32/signed-native>> = erlang:port_control(get_port(), ?CMD_TRUNCATE, Cmd),
-    recv_ok(Result).
+    recv_ok(Rc).
 
 %%--------------------------------------------------------------------
 %% @doc
@@ -1193,8 +1194,8 @@ truncate(Db) ->
 
 cursor_open(Db) ->
     Cmd = <<Db:32/signed-native, 0:32/native>>,
-    <<Result:32/signed-native>> = erlang:port_control(get_port(), ?CMD_CURSOR_OPEN, Cmd),
+    <<Rc:32/signed-native>> = erlang:port_control(get_port(), ?CMD_CURSOR_OPEN, Cmd),
-    recv_ok(Result).
+    recv_ok(Rc).
 
 %%--------------------------------------------------------------------
 %% @doc
@@ -1372,12 +1373,7 @@ cursor_get(Key) ->
     not_found | {ok, db_key(), db_value()} | db_error().
 
 cursor_get(Key, Opts) ->
-    case Key of
+    {KeyLen, KeyBin} = to_binary(Key),
-        undefined ->
-            {KeyLen, KeyBin} = {0, <<>>};
-        _ ->
-            {KeyLen, KeyBin} = to_binary(Key)
-    end,
     Flags = process_flags(Opts),
     Cmd = <<Flags:32/native, KeyLen:32/native, KeyBin/bytes>>,
     <<Result:32/signed-native>> = erlang:port_control(get_port(), ?CMD_CURSOR_GET, Cmd),
@@ -1401,25 +1397,6 @@ cursor_get(Key, Opts) ->
     end.
 
 
-%%--------------------------------------------------------------------
-%% @doc
-%% Returns the count of duplicate records for the key to which the
-%% cursor currently refers.
-%%
-%% If this function fails for any reason, the state of the cursor will
-%% be unchanged.
-%%
-%% @spec cursor_count() -> {ok, Count} | {error, Error}
-%%
-%% @end
-%%--------------------------------------------------------------------
--spec cursor_count() -> {ok, Count :: number()} | {error, db_error()}.
-
-cursor_count() ->
-    <<Result:32/signed-native>> = erlang:port_control(get_port(), ?CMD_CURSOR_COUNT, <<>>),
-    recv_val(Result).
-
-
 %%--------------------------------------------------------------------
 %% @doc
 %% Closes the cursor.
@@ -1441,8 +1418,8 @@ cursor_count() ->
 -spec cursor_close() -> ok | db_error().
 
 cursor_close() ->
-    <<Result:32/signed-native>> = erlang:port_control(get_port(), ?CMD_CURSOR_CLOSE, <<>>),
+    <<Rc:32/signed-native>> = erlang:port_control(get_port(), ?CMD_CURSOR_CLOSE, <<>>),
-    recv_ok(Result).
+    recv_ok(Rc).
 
 %%--------------------------------------------------------------------
 %% @doc
@@ -1468,8 +1445,8 @@ cursor_close() ->
 
 delete_database(Filename) ->
     Cmd = <<(list_to_binary(Filename))/binary, 0:8>>,
-    <<Result:32/signed-native>> = erlang:port_control(get_port(), ?CMD_REMOVE_DB, Cmd),
+    <<Rc:32/signed-native>> = erlang:port_control(get_port(), ?CMD_REMOVE_DB, Cmd),
-    recv_ok(Result).
+    recv_ok(Rc).
 
 %%--------------------------------------------------------------------
 %% @doc
@@ -1657,8 +1634,8 @@ get_txn_timeout() ->
 stat(Db, Opts) ->
     Flags = process_flags(Opts),
     Cmd = <<Db:32/signed-native, Flags:32/native>>,
-    <<Result:32/signed-native>> = erlang:port_control(get_port(), ?CMD_DB_STAT, Cmd),
+    <<Rc:32/signed-native>> = erlang:port_control(get_port(), ?CMD_DB_STAT, Cmd),
-    recv_val(Result).
+    recv_val(Rc).
 
 %%--------------------------------------------------------------------
 %% @doc
@@ -1762,8 +1739,8 @@ stat_print(Db) ->
 lock_stat(Opts) ->
     Flags = process_flags(Opts),
     Cmd = <<Flags:32/native>>,
-    <<Result:32/signed-native>> = erlang:port_control(get_port(), ?CMD_LOCK_STAT, Cmd),
+    <<Rc:32/signed-native>> = erlang:port_control(get_port(), ?CMD_LOCK_STAT, Cmd),
-    recv_val(Result).
+    recv_val(Rc).
 
 %%--------------------------------------------------------------------
 %% @doc
@@ -1858,8 +1835,8 @@ lock_stat_print() ->
 log_stat(Opts) ->
     Flags = process_flags(Opts),
     Cmd = <<Flags:32/native>>,
-    <<Result:32/signed-native>> = erlang:port_control(get_port(), ?CMD_LOG_STAT, Cmd),
+    <<Rc:32/signed-native>> = erlang:port_control(get_port(), ?CMD_LOG_STAT, Cmd),
-    recv_val(Result).
+    recv_val(Rc).
 
 %%--------------------------------------------------------------------
 %% @doc
@@ -2039,8 +2016,8 @@ memp_stat_print() ->
 mutex_stat(Opts) ->
     Flags = process_flags(Opts),
     Cmd = <<Flags:32/native>>,
-    <<Result:32/signed-native>> = erlang:port_control(get_port(), ?CMD_MUTEX_STAT, Cmd),
+    <<Rc:32/signed-native>> = erlang:port_control(get_port(), ?CMD_MUTEX_STAT, Cmd),
-    recv_val(Result).
+    recv_val(Rc).
 
 %%--------------------------------------------------------------------
 %% @doc
@@ -2255,9 +2232,8 @@ env_stat_print() ->
     {ok, [{atom(), number()}]} | db_error().
 
 driver_info() ->
-    <<Result:32/signed-native>> = erlang:port_control(get_port(), ?CMD_DRIVER_INFO, <<>>),
+    <<Rc:32/signed-native>> = erlang:port_control(get_port(), ?CMD_DRIVER_INFO, <<>>),
-    recv_val(Result).
+    recv_val(Rc).
-
 
 %%--------------------------------------------------------------------
 %% @doc
@@ -2275,7 +2251,6 @@ register_logger() ->
     [] = erlang:port_control(get_port(), ?CMD_REGISTER_LOGGER, <<>>),
     ok.
 
-
 %%--------------------------------------------------------------------
 %% @doc
 %% Stop bdberl - stops the bdberl_logger process so that when
@@ -2447,15 +2422,15 @@ do_put(Action, Db, Key, Value, Opts) ->
     Flags = process_flags(Opts),
     Cmd = <<Db:32/signed-native, Flags:32/native, KeyLen:32/native, KeyBin/bytes,
            FinalValBinLen:32/native, FinalValBin/bytes>>,
-    <<Result:32/signed-native>> = erlang:port_control(get_port(), Action, Cmd),
+    <<Rc:32/signed-native>> = erlang:port_control(get_port(), Action, Cmd),
-    recv_ok(Result).
+    recv_ok(Rc).
 
 %%
 %% Move the cursor in a given direction. Invoked by cursor_next/prev/current.
 %%
 do_cursor_move(Direction) ->
-    <<Result:32/signed-native>> = erlang:port_control(get_port(), Direction, <<>>),
+    <<Rc:32/signed-native>> = erlang:port_control(get_port(), Direction, <<>>),
-    case decode_rc(Result) of
+    case decode_rc(Rc) of
         ok ->
             receive
                 {ok, KeyBin, ValueBin} ->

src/bdberl_logger.erl

@@ -79,11 +79,11 @@ handle_cast(_Msg, State) ->
     {stop, unsupportedOperation, State}.
 
 handle_info({bdb_error_log, Msg}, State) ->
-    lager:error("BDB: ~s\n", [Msg]),
+    lager:error("BDB Error: ~s\n", [Msg]),
     {noreply, State};
 
 handle_info({bdb_info_log, Msg}, State) ->
-    lager:info("BDB: ~s\n", [Msg]),
+    lager:info("BDB Info: ~s\n", [Msg]),
     {noreply, State};
 
 handle_info({bdb_checkpoint_stats, CheckpointSecs, ArchiveSecs, 0, 0}, State) ->
@@ -98,7 +98,7 @@ handle_info({bdb_checkpoint_stats, CheckpointSecs, ArchiveSecs, 0, 0}, State) ->
     {noreply, State};
 
 handle_info({bdb_checkpoint_stats, _CheckpointSecs, _ArchiveSecs, CheckpointRc, ArchiveRc}, State) ->
-    lager:error("BDB Checkpoint: ~w ~w\n", [CheckpointRc, ArchiveRc]),
+    lager:error("BDB Checkpoint error: ~w ~w\n", [CheckpointRc, ArchiveRc]),
     {noreply, State};
 
 handle_info({bdb_trickle_stats, ElapsedSecs, Pages, 0}, State) ->
@@ -112,11 +112,11 @@ handle_info({bdb_trickle_stats, ElapsedSecs, Pages, 0}, State) ->
     end,
     {noreply, State};
 handle_info({bdb_trickle_stats, _ElapsedSecs, _Pages, Rc}, State) ->
-    lager:error("BDB Trickle Write: ~w\n", [Rc]),
+    lager:error("BDB Trickle Write error: ~w\n", [Rc]),
     {noreply, State};
 
 handle_info(Msg, State) ->
-    lager:info("Unexpected message: ~p\n", [Msg]),
+    io:format("Unexpected message: ~p\n", [Msg]),
     {noreply, State}.
 
 terminate(_Reason, _State) ->

test/bdberl_SUITE.erl

@@ -55,7 +55,6 @@ all() ->
      update_should_accept_args_for_fun,
      port_should_return_transaction_timeouts,
      cursor_should_iterate, cursor_get_should_pos, cursor_should_fail_if_not_open,
-     cursor_should_return_count,
      put_commit_should_end_txn,
      data_dir_should_be_priv_dir,
      delete_should_remove_file,
@@ -271,20 +270,6 @@ cursor_get_should_pos(Config) ->
 
     ok = bdberl:cursor_close().
 
-cursor_should_return_count(Config) ->
-    Db = ?config(db, Config),
-
-    %% Store some sample values in the db
-    ok = bdberl:put(Db, key, value1),
-%    ok = bdberl:put(Db, key, value2),
-%    ok = bdberl:put(Db, key, value3),
-%    ok = bdberl:put(Db, key, value4),
-    ok = bdberl:cursor_open(Db),
-    {ok, _} = bdberl:cursor_get(key),
-    %% Validate the count of duplicate values
-    {ok, 1} = bdberl:cursor_count(),
-    ok = bdberl:cursor_close().
-
 cursor_should_fail_if_not_open(_Config) ->
     {error, no_cursor} = bdberl:cursor_next(),
     {error, no_cursor} = bdberl:cursor_prev(),
@@ -439,3 +424,5 @@ start_after_stop_should_be_safe(_Config) ->
     end,
     true = lists:keymember(bdberl_logger, 1, supervisor:which_children(kernel_safe_sup)),
     ok.
+
+