Merge pull request #20 from rouzier/feature/txn_try

mdbx: non-blocking begin/try write-txn (feature).

libmdbx.files

@@ -44,6 +44,7 @@ test/osal-windows.cc
 test/osal.h
 test/test.cc
 test/test.h
+test/try.cc
 test/utils.cc
 test/utils.h
 tutorial/README.md

mdbx.h

@@ -327,6 +327,10 @@ typedef int(MDBX_cmp_func)(const MDBX_val *a, const MDBX_val *b);
 /* Store multiple data items in one call. Only for MDBX_DUPFIXED. */
 #define MDBX_MULTIPLE 0x80000u
 
+/* Transaction Flags*/
+/* Do not block when starting a write transaction */
+#define MDBX_TRYTXN 0x10000000u
+
 /* Copy Flags */
 /* Compacting copy: Omit free space from copy, and renumber all
  * pages sequentially. */
@@ -420,8 +424,10 @@ typedef enum MDBX_cursor_op {
 #define MDBX_BAD_DBI (-30780)
 /* Unexpected problem - txn should abort */
 #define MDBX_PROBLEM (-30779)
+/* Unexpected problem - txn should abort */
+#define MDBX_BUSY (-30778)
 /* The last defined error code */
-#define MDBX_LAST_ERRCODE MDBX_PROBLEM
+#define MDBX_LAST_ERRCODE MDBX_BUSY
 
 /* The mdbx_put() or mdbx_replace() was called for key,
     that has more that one associated value. */
@@ -953,6 +959,9 @@ LIBMDBX_API int mdbx_env_set_assert(MDBX_env *env, MDBX_assert_func *func);
  *  - MDBX_RDONLY
  *      This transaction will not perform any write operations.
  *
+ *  - MDBX_TRYTXN
+ *      Do not block when starting a write transaction
+ *
  * [out] txn Address where the new MDBX_txn handle will be stored
  *
  * Returns A non-zero error value on failure and 0 on success, some
@@ -964,7 +973,8 @@ LIBMDBX_API int mdbx_env_set_assert(MDBX_env *env, MDBX_assert_func *func);
  *                         as well. See mdbx_env_set_mapsize().
  *  - MDBX_READERS_FULL  - a read-only transaction was requested and the reader
  *                         lock table is full. See mdbx_env_set_maxreaders().
- *  - MDBX_ENOMEM        - out of memory. */
+ *  - MDBX_ENOMEM        - out of memory.
+ *  - MDBX_BUSY          - a write transaction is already started. */
 LIBMDBX_API int mdbx_txn_begin(MDBX_env *env, MDBX_txn *parent, unsigned flags,
                                MDBX_txn **txn);
 

src/bits.h

@@ -553,7 +553,8 @@ struct MDBX_txn {
 
 /* Transaction Flags */
 /* mdbx_txn_begin() flags */
-#define MDBX_TXN_BEGIN_FLAGS (MDBX_NOMETASYNC | MDBX_NOSYNC | MDBX_RDONLY)
+#define MDBX_TXN_BEGIN_FLAGS_PERSISTENT (MDBX_NOMETASYNC | MDBX_NOSYNC | MDBX_RDONLY)
+#define MDBX_TXN_BEGIN_FLAGS (MDBX_TXN_BEGIN_FLAGS_PERSISTENT | MDBX_TRYTXN)
 #define MDBX_TXN_NOMETASYNC                                                    \
   MDBX_NOMETASYNC                   /* don't sync meta for this txn on commit */
 #define MDBX_TXN_NOSYNC MDBX_NOSYNC /* don't sync this txn on commit */

src/lck-posix.c

@@ -184,6 +184,13 @@ static int mdbx_robust_lock(MDBX_env *env, pthread_mutex_t *mutex) {
   return rc;
 }
 
+static int mdbx_robust_trylock(MDBX_env *env, pthread_mutex_t *mutex) {
+  int rc = pthread_mutex_trylock(mutex);
+  if (unlikely(rc != 0))
+    rc = mdbx_mutex_failed(env, mutex, rc);
+  return rc;
+}
+
 static int mdbx_robust_unlock(MDBX_env *env, pthread_mutex_t *mutex) {
   int rc = pthread_mutex_unlock(mutex);
   if (unlikely(rc != 0))
@@ -213,6 +220,13 @@ int mdbx_txn_lock(MDBX_env *env) {
   return MDBX_IS_ERROR(rc) ? rc : MDBX_SUCCESS;
 }
 
+int mdbx_txn_trylock(MDBX_env *env) {
+  mdbx_trace(">>");
+  int rc = mdbx_robust_trylock(env, &env->me_lck->mti_wmutex);
+  mdbx_trace("<< rc %d", rc);
+  return MDBX_IS_ERROR(rc) ? rc : MDBX_SUCCESS;
+}
+
 void mdbx_txn_unlock(MDBX_env *env) {
   mdbx_trace(">>");
   int rc = mdbx_robust_unlock(env, &env->me_lck->mti_wmutex);
@@ -313,7 +327,9 @@ static int __cold mdbx_mutex_failed(MDBX_env *env, pthread_mutex_t *mutex,
 #endif /* MDBX_USE_ROBUST */
 
   mdbx_error("mutex (un)lock failed, %s", mdbx_strerror(rc));
-  if (rc != EDEADLK) {
+  if (rc == EBUSY) {
+    rc = MDBX_BUSY;
+  } else if (rc != EDEADLK && rc != EBUSY) {
     env->me_flags |= MDBX_FATAL_ERROR;
     rc = MDBX_PANIC;
   }

src/lck-windows.c

@@ -132,6 +132,16 @@ int mdbx_txn_lock(MDBX_env *env) {
   return GetLastError();
 }
 
+int mdbx_txn_trylock(MDBX_env *env) {
+  if (flock(env->me_fd, LCK_EXCLUSIVE | LCK_DONTWAIT, LCK_BODY))
+    return MDBX_SUCCESS;
+  int rc = GetLastError();
+  if (rc == ERROR_LOCK_VIOLATION) {
+    rc = MDBX_BUSY;
+  }
+  return rc;
+}
+
 void mdbx_txn_unlock(MDBX_env *env) {
   if (!funlock(env->me_fd, LCK_BODY))
     mdbx_panic("%s failed: errcode %u", mdbx_func_, GetLastError());

src/mdbx.c

@@ -708,6 +708,7 @@ static const char *__mdbx_strerr(int errnum) {
       "DUPFIXED size",
       "MDBX_BAD_DBI: The specified DBI handle was closed/changed unexpectedly",
       "MDBX_PROBLEM: Unexpected problem - txn should abort",
+      "MDBX_BUSY: Another write transation is started",
   };
 
   if (errnum >= MDBX_KEYEXIST && errnum <= MDBX_LAST_ERRCODE) {
@@ -2562,7 +2563,7 @@ static int mdbx_txn_renew0(MDBX_txn *txn, unsigned flags) {
   } else {
     /* Not yet touching txn == env->me_txn0, it may be active */
     mdbx_jitter4testing(false);
-    rc = mdbx_txn_lock(env);
+    rc = F_ISSET(flags, MDBX_TRYTXN) ? mdbx_txn_trylock(env) : mdbx_txn_lock(env);
     if (unlikely(rc))
       return rc;
 
@@ -2664,6 +2665,7 @@ int mdbx_txn_begin(MDBX_env *env, MDBX_txn *parent, unsigned flags,
                    MDBX_txn **ret) {
   MDBX_txn *txn;
   MDBX_ntxn *ntxn;
+  //unsigned pflags;
   int rc, size, tsize;
 
   if (unlikely(!env || !ret))
@@ -2718,7 +2720,7 @@ int mdbx_txn_begin(MDBX_env *env, MDBX_txn *parent, unsigned flags,
   txn->mt_dbxs = env->me_dbxs; /* static */
   txn->mt_dbs = (MDBX_db *)((char *)txn + tsize);
   txn->mt_dbflags = (uint8_t *)txn + size - env->me_maxdbs;
-  txn->mt_flags = flags;
+  txn->mt_flags = flags & MDBX_TXN_BEGIN_FLAGS_PERSISTENT;
   txn->mt_env = env;
 
   if (parent) {

src/osal.h

@@ -506,6 +506,7 @@ int mdbx_rdt_lock(MDBX_env *env);
 void mdbx_rdt_unlock(MDBX_env *env);
 
 int mdbx_txn_lock(MDBX_env *env);
+int mdbx_txn_trylock(MDBX_env *env);
 void mdbx_txn_unlock(MDBX_env *env);
 
 int mdbx_rpid_set(MDBX_env *env);

test/cases.cc

@@ -67,6 +67,7 @@ void testcase_setup(const char *casename, actor_params &params,
     configure_actor(last_space_id, ac_hill, nullptr, params);
     configure_actor(last_space_id, ac_jitter, nullptr, params);
     configure_actor(last_space_id, ac_hill, nullptr, params);
+    configure_actor(last_space_id, ac_try, nullptr, params);
     log_notice("<<< testcase_setup(%s): done", casename);
   } else {
     failure("unknown testcase `%s`", casename);

test/config.h

@@ -20,7 +20,7 @@
 
 #define ACTOR_ID_MAX INT16_MAX
 
-enum actor_testcase { ac_none, ac_hill, ac_deadread, ac_deadwrite, ac_jitter };
+enum actor_testcase { ac_none, ac_hill, ac_deadread, ac_deadwrite, ac_jitter, ac_try };
 
 enum actor_status {
   as_unknown,

test/dead.cc

@@ -25,7 +25,7 @@ bool testcase_deadread::setup() {
 
 bool testcase_deadread::run() {
   db_open();
-  txn_begin(true);
+  txn_begin(MDBX_RDONLY);
   return true;
 }
 
@@ -50,7 +50,7 @@ bool testcase_deadwrite::setup() {
 
 bool testcase_deadwrite::run() {
   db_open();
-  txn_begin(false);
+  txn_begin(0);
   return true;
 }
 

test/hill.cc

@@ -28,7 +28,7 @@ bool testcase_hill::setup() {
 bool testcase_hill::run() {
   db_open();
 
-  txn_begin(false);
+  txn_begin(0);
   MDBX_dbi dbi = db_table_open(true);
   txn_end(false);
 
@@ -70,7 +70,7 @@ bool testcase_hill::run() {
   uint64_t serial_count = 0;
   unsigned txn_nops = 0;
   if (!txn_guard)
-    txn_begin(false);
+    txn_begin(0);
 
   while (should_continue()) {
     const keygen::serial_t a_serial = serial_count;
@@ -91,7 +91,7 @@ bool testcase_hill::run() {
       failure_perror("mdbx_put(insert-a.1)", rc);
 
     if (++txn_nops >= config.params.batch_write) {
-      txn_restart(false, false);
+      txn_restart(false, 0);
       txn_nops = 0;
     }
 
@@ -104,7 +104,7 @@ bool testcase_hill::run() {
       failure_perror("mdbx_put(insert-b)", rc);
 
     if (++txn_nops >= config.params.batch_write) {
-      txn_restart(false, false);
+      txn_restart(false, 0);
       txn_nops = 0;
     }
 
@@ -118,7 +118,7 @@ bool testcase_hill::run() {
       failure_perror("mdbx_put(update-a: 1->0)", rc);
 
     if (++txn_nops >= config.params.batch_write) {
-      txn_restart(false, false);
+      txn_restart(false, 0);
       txn_nops = 0;
     }
 
@@ -129,7 +129,7 @@ bool testcase_hill::run() {
       failure_perror("mdbx_del(b)", rc);
 
     if (++txn_nops >= config.params.batch_write) {
-      txn_restart(false, false);
+      txn_restart(false, 0);
       txn_nops = 0;
     }
 
@@ -164,7 +164,7 @@ bool testcase_hill::run() {
       failure_perror("mdbx_put(update-a: 0->1)", rc);
 
     if (++txn_nops >= config.params.batch_write) {
-      txn_restart(false, false);
+      txn_restart(false, 0);
       txn_nops = 0;
     }
 
@@ -177,7 +177,7 @@ bool testcase_hill::run() {
       failure_perror("mdbx_put(insert-b)", rc);
 
     if (++txn_nops >= config.params.batch_write) {
-      txn_restart(false, false);
+      txn_restart(false, 0);
       txn_nops = 0;
     }
 
@@ -189,7 +189,7 @@ bool testcase_hill::run() {
       failure_perror("mdbx_del(a)", rc);
 
     if (++txn_nops >= config.params.batch_write) {
-      txn_restart(false, false);
+      txn_restart(false, 0);
       txn_nops = 0;
     }
 
@@ -200,7 +200,7 @@ bool testcase_hill::run() {
       failure_perror("mdbx_del(b)", rc);
 
     if (++txn_nops >= config.params.batch_write) {
-      txn_restart(false, false);
+      txn_restart(false, 0);
       txn_nops = 0;
     }
 
@@ -212,7 +212,7 @@ bool testcase_hill::run() {
 
   if (dbi) {
     if (config.params.drop_table && !mode_readonly()) {
-      txn_begin(false);
+      txn_begin(0);
       db_table_drop(dbi);
       txn_end(false);
     } else

test/jitter.cc

@@ -30,7 +30,7 @@ bool testcase_jitter::run() {
 
     if (flipcoin()) {
       jitter_delay();
-      txn_begin(true);
+      txn_begin(MDBX_RDONLY);
       fetch_canary();
       jitter_delay();
       txn_end(flipcoin());
@@ -51,7 +51,7 @@ bool testcase_jitter::run() {
 
     if (flipcoin()) {
       jitter_delay();
-      txn_begin(true);
+      txn_begin(MDBX_RDONLY);
       jitter_delay();
       txn_end(flipcoin());
     }

test/test.cc

@@ -29,6 +29,8 @@ const char *testcase2str(const actor_testcase testcase) {
     return "deadwrite";
   case ac_jitter:
     return "jitter";
+  case ac_try:
+    return "try";
   }
 }
 
@@ -175,18 +177,18 @@ void testcase::db_close() {
   log_trace("<< db_close");
 }
 
-void testcase::txn_begin(bool readonly) {
+void testcase::txn_begin(unsigned flags) {
-  log_trace(">> txn_begin(%s)", readonly ? "read-only" : "read-write");
+  log_trace(">> txn_begin(%s)", flags & MDBX_RDONLY ? "read-only" : "read-write");
   assert(!txn_guard);
 
   MDBX_txn *txn = nullptr;
   int rc =
-      mdbx_txn_begin(db_guard.get(), nullptr, readonly ? MDBX_RDONLY : 0, &txn);
+      mdbx_txn_begin(db_guard.get(), nullptr, flags, &txn);
   if (unlikely(rc != MDBX_SUCCESS))
     failure_perror("mdbx_txn_begin()", rc);
   txn_guard.reset(txn);
 
-  log_trace("<< txn_begin(%s)", readonly ? "read-only" : "read-write");
+  log_trace("<< txn_begin(%s)", flags & MDBX_RDONLY ? "read-only" : "read-write");
 }
 
 void testcase::txn_end(bool abort) {
@@ -207,10 +209,10 @@ void testcase::txn_end(bool abort) {
   log_trace("<< txn_end(%s)", abort ? "abort" : "commit");
 }
 
-void testcase::txn_restart(bool abort, bool readonly) {
+void testcase::txn_restart(bool abort, unsigned flags) {
   if (txn_guard)
     txn_end(abort);
-  txn_begin(readonly);
+  txn_begin(flags);
 }
 
 bool testcase::wait4start() {
@@ -443,6 +445,9 @@ bool test_execute(const actor_config &config) {
     case ac_jitter:
       test.reset(new testcase_jitter(config, pid));
       break;
+    case ac_try:
+      test.reset(new testcase_try(config, pid));
+      break;
     default:
       test.reset(new testcase(config, pid));
       break;

test/test.h

@@ -104,9 +104,9 @@ protected:
   void db_prepare();
   void db_open();
   void db_close();
-  void txn_begin(bool readonly);
+  void txn_begin(unsigned flags);
   void txn_end(bool abort);
-  void txn_restart(bool abort, bool readonly);
+  void txn_restart(bool abort, unsigned flags);
   void fetch_canary();
   void update_canary(uint64_t increment);
   void kick_progress(bool active) const;
@@ -130,8 +130,8 @@ protected:
     generate_pair(serial, key, data, data_age);
   }
 
-  bool mode_readonly() const {
+  unsigned mode_readonly() const {
-    return (config.params.mode_flags & MDBX_RDONLY) ? true : false;
+    return (config.params.mode_flags & MDBX_RDONLY) ? MDBX_RDONLY : 0;
   }
 
 public:
@@ -190,3 +190,14 @@ public:
   bool run();
   bool teardown();
 };
+
+class testcase_try : public testcase {
+  typedef testcase inherited;
+
+public:
+  testcase_try(const actor_config &config, const mdbx_pid_t pid)
+      : testcase(config, pid) {}
+  bool setup();
+  bool run();
+  bool teardown();
+};

test/test.vcxproj

@@ -182,6 +182,7 @@
     <ClCompile Include="config.cc" />
     <ClCompile Include="dead.cc" />
     <ClCompile Include="hill.cc" />
+    <ClCompile Include="try.cc" />
     <ClCompile Include="jitter.cc" />
     <ClCompile Include="keygen.cc" />
     <ClCompile Include="log.cc" />

test/try.cc

@@ -0,0 +1,38 @@
+#include "test.h"
+
+bool testcase_try::setup() {
+  log_trace(">> setup");
+  if (!inherited::setup())
+    return false;
+
+  log_trace("<< setup");
+  return true;
+}
+
+bool testcase_try::run() {
+  db_open();
+  assert(!txn_guard);
+
+  MDBX_txn *txn = nullptr;
+  MDBX_txn *txn2 = nullptr;
+  int rc =
+      mdbx_txn_begin(db_guard.get(), nullptr, 0, &txn);
+  if (unlikely(rc != MDBX_SUCCESS))
+    failure_perror("mdbx_txn_begin(MDBX_TRYTXN)", rc);
+  else {
+    rc = mdbx_txn_begin(db_guard.get(), nullptr, MDBX_TRYTXN, &txn2);
+    if (unlikely(rc != MDBX_BUSY))
+      failure_perror("mdbx_txn_begin(MDBX_TRYTXN)", rc);
+  }
+
+  txn_guard.reset(txn);
+  return true;
+}
+
+bool testcase_try::teardown() {
+  log_trace(">> teardown");
+  cursor_guard.release();
+  txn_guard.release();
+  db_guard.release();
+  return inherited::teardown();
+}