Catch most uses of finished/parent txns.

* Add MDB_TXN_FINISHED, MDB_TXN_HAS_CHILD, MDB_TXN_BLOCKED.
* Clear mt_numdbs in writers, for TXN_DBI_EXIST() to catch.
  We already do in readers.

diff --git a/libraries/liblmdb/lmdb.h b/libraries/liblmdb/lmdb.h
index c27f78bfb808d29fc2d5ebc9eb89999e821a737c..b72418e4fb78a6f2b3f17b5db423361d2ddee406 100644 (file)
--- a/libraries/liblmdb/lmdb.h
+++ b/libraries/liblmdb/lmdb.h
@@ -432,7 +432,7 @@ typedef enum MDB_cursor_op {
 #define MDB_INCOMPATIBLE       (-30784)
        /** Invalid reuse of reader locktable slot */
 #define MDB_BAD_RSLOT          (-30783)
-       /** Transaction cannot recover - it must be aborted */
+       /** Transaction must abort, has a child, or is invalid */
 #define MDB_BAD_TXN                    (-30782)
        /** Unsupported size of key/DB name/data, or wrong DUPFIXED size */
 #define MDB_BAD_VALSIZE                (-30781)

diff --git a/libraries/liblmdb/mdb.c b/libraries/liblmdb/mdb.c
index 32508dcd03915b6430fdcdbe8824f08912428553..ec34cfec50b55d682b00994c6c0b0b2365a87aba 100644 (file)
--- a/libraries/liblmdb/mdb.c
+++ b/libraries/liblmdb/mdb.c
@@ -1056,7 +1056,8 @@ typedef struct MDB_dbx {
         */
 struct MDB_txn {
        MDB_txn         *mt_parent;             /**< parent of a nested txn */
-       MDB_txn         *mt_child;              /**< nested txn under this txn */
+       /** Nested txn under this txn, set together with flag #MDB_TXN_HAS_CHILD */
+       MDB_txn         *mt_child;
        pgno_t          mt_next_pgno;   /**< next unallocated page */
        /** The ID of this transaction. IDs are integers incrementing from 1.
         *      Only committed write transactions increment the ID. If a transaction
@@ -1104,8 +1105,9 @@ struct MDB_txn {
        MDB_cursor      **mt_cursors;
        /** Array of flags for each DB */
        unsigned char   *mt_dbflags;
-       /**     Number of DB records in use. This number only ever increments;
-        *      we don't decrement it when individual DB handles are closed.
+       /**     Number of DB records in use, or 0 when the txn is finished.
+        *      This number only ever increments until the txn finishes; we
+        *      don't decrement it when individual DB handles are closed.
         */
        MDB_dbi         mt_numdbs;
 
@@ -1120,9 +1122,13 @@ struct MDB_txn {
 #define MDB_TXN_RDONLY         MDB_RDONLY      /**< read-only transaction */
        /* internal txn flags */
 #define MDB_TXN_WRITEMAP       MDB_WRITEMAP    /**< copy of #MDB_env flag in writers */
+#define MDB_TXN_FINISHED       0x01            /**< txn is finished or never began */
 #define MDB_TXN_ERROR          0x02            /**< txn is unusable after an error */
 #define MDB_TXN_DIRTY          0x04            /**< must write, even if dirty list is empty */
 #define MDB_TXN_SPILLS         0x08            /**< txn or a parent has spilled pages */
+#define MDB_TXN_HAS_CHILD      0x10            /**< txn has an #MDB_txn.%mt_child */
+       /** most operations on the txn are currently illegal */
+#define MDB_TXN_BLOCKED                (MDB_TXN_FINISHED|MDB_TXN_ERROR|MDB_TXN_HAS_CHILD)
 /** @} */
        unsigned int    mt_flags;               /**< @ref mdb_txn */
        /** #dirty_list room: Array size - \#dirty pages visible to this txn.
@@ -1418,7 +1424,7 @@ static char *const mdb_errstr[] = {
        "MDB_MAP_RESIZED: Database contents grew beyond environment mapsize",
        "MDB_INCOMPATIBLE: Operation and DB incompatible, or DB flags changed",
        "MDB_BAD_RSLOT: Invalid reuse of reader locktable slot",
-       "MDB_BAD_TXN: Transaction cannot recover - it must be aborted",
+       "MDB_BAD_TXN: Transaction must abort, has a child, or is invalid",
        "MDB_BAD_VALSIZE: Unsupported size of key/DB name/data, or wrong DUPFIXED size",
        "MDB_BAD_DBI: The specified DBI handle was closed/changed unexpectedly",
 };
@@ -2787,9 +2793,7 @@ mdb_txn_begin(MDB_env *env, MDB_txn *parent, unsigned int flags, MDB_txn **ret)
        if (parent) {
                /* Nested transactions: Max 1 child, write txns only, no writemap */
                flags |= parent->mt_flags;
-               if (parent->mt_child ||
-                       (flags & (MDB_RDONLY|MDB_WRITEMAP|MDB_TXN_ERROR)))
-               {
+               if (flags & (MDB_RDONLY|MDB_WRITEMAP|MDB_TXN_BLOCKED)) {
                        return (parent->mt_flags & MDB_TXN_RDONLY) ? EINVAL : MDB_BAD_TXN;
                }
                /* Child txns save MDB_pgstate and use own copy of cursors */
@@ -2831,6 +2835,7 @@ mdb_txn_begin(MDB_env *env, MDB_txn *parent, unsigned int flags, MDB_txn **ret)
                txn->mt_u.dirty_list[0].mid = 0;
                txn->mt_spill_pgs = NULL;
                txn->mt_next_pgno = parent->mt_next_pgno;
+               parent->mt_flags |= MDB_TXN_HAS_CHILD;
                parent->mt_child = txn;
                txn->mt_parent = parent;
                txn->mt_numdbs = parent->mt_numdbs;
@@ -2947,9 +2952,10 @@ mdb_txn_end(MDB_txn *txn, unsigned mode)
                                txn->mt_u.reader = NULL;
                        } /* else txn owns the slot until it does MDB_END_SLOT */
                }
-               txn->mt_numdbs = 0;             /* close nothing if called again */
+               txn->mt_numdbs = 0;             /* prevent further DBI activity */
+               txn->mt_flags |= MDB_TXN_FINISHED;
                txn->mt_dbxs = NULL;    /* mark txn as reset */
-       } else {
+       } else if (!F_ISSET(txn->mt_flags, MDB_TXN_FINISHED)) {
                pgno_t *pghead = env->me_pghead;
 
                if (!(mode & MDB_END_UPDATE)) /* !(already closed cursors) */
@@ -2958,6 +2964,9 @@ mdb_txn_end(MDB_txn *txn, unsigned mode)
                        mdb_dlist_free(txn);
                }
 
+               txn->mt_numdbs = 0;
+               txn->mt_flags = MDB_TXN_FINISHED;
+
                if (!txn->mt_parent) {
                        mdb_midl_shrink(&txn->mt_free_pgs);
                        env->me_free_pgs = txn->mt_free_pgs;
@@ -2973,6 +2982,7 @@ mdb_txn_end(MDB_txn *txn, unsigned mode)
                                UNLOCK_MUTEX(env->me_wmutex);
                } else {
                        txn->mt_parent->mt_child = NULL;
+                       txn->mt_parent->mt_flags &= ~MDB_TXN_HAS_CHILD;
                        env->me_pgstate = ((MDB_ntxn *)txn)->mnt_pgstate;
                        mdb_midl_free(txn->mt_free_pgs);
                        mdb_midl_free(txn->mt_spill_pgs);
@@ -3380,8 +3390,8 @@ mdb_txn_commit(MDB_txn *txn)
                goto done;
        }
 
-       if (F_ISSET(txn->mt_flags, MDB_TXN_ERROR)) {
-               DPUTS("error flag is set, can't commit");
+       if (txn->mt_flags & (MDB_TXN_FINISHED|MDB_TXN_ERROR)) {
+               DPUTS("txn has failed/finished, can't commit");
                if (txn->mt_parent)
                        txn->mt_parent->mt_flags |= MDB_TXN_ERROR;
                rc = MDB_BAD_TXN;
@@ -4842,6 +4852,7 @@ mdb_env_open(MDB_env *env, const char *path, unsigned int flags, mdb_mode_t mode
                                txn->mt_dbflags = (unsigned char *)(txn->mt_dbiseqs + env->me_maxdbs);
                                txn->mt_env = env;
                                txn->mt_dbxs = env->me_dbxs;
+                               txn->mt_flags = MDB_TXN_FINISHED;
                                env->me_txn0 = txn;
                        } else {
                                rc = ENOMEM;
@@ -5394,8 +5405,8 @@ mdb_page_search(MDB_cursor *mc, MDB_val *key, int flags)
        /* Make sure the txn is still viable, then find the root from
         * the txn's db table and set it as the root of the cursor's stack.
         */
-       if (F_ISSET(mc->mc_txn->mt_flags, MDB_TXN_ERROR)) {
-               DPUTS("transaction has failed, must abort");
+       if (mc->mc_txn->mt_flags & MDB_TXN_BLOCKED) {
+               DPUTS("transaction may not be used now");
                return MDB_BAD_TXN;
        } else {
                /* Make sure we're using an up-to-date root */
@@ -5582,7 +5593,7 @@ mdb_get(MDB_txn *txn, MDB_dbi dbi,
        if (!key || !data || !TXN_DBI_EXIST(txn, dbi, DB_USRVALID))
                return EINVAL;
 
-       if (txn->mt_flags & MDB_TXN_ERROR)
+       if (txn->mt_flags & MDB_TXN_BLOCKED)
                return MDB_BAD_TXN;
 
        mdb_cursor_init(&mc, txn, dbi, &mx);
@@ -6103,7 +6114,7 @@ mdb_cursor_get(MDB_cursor *mc, MDB_val *key, MDB_val *data,
        if (mc == NULL)
                return EINVAL;
 
-       if (mc->mc_txn->mt_flags & MDB_TXN_ERROR)
+       if (mc->mc_txn->mt_flags & MDB_TXN_BLOCKED)
                return MDB_BAD_TXN;
 
        switch (op) {
@@ -6333,7 +6344,7 @@ mdb_cursor_put(MDB_cursor *mc, MDB_val *key, MDB_val *data,
        nospill = flags & MDB_NOSPILL;
        flags &= ~MDB_NOSPILL;
 
-       if (mc->mc_txn->mt_flags & (MDB_TXN_RDONLY|MDB_TXN_ERROR))
+       if (mc->mc_txn->mt_flags & (MDB_TXN_RDONLY|MDB_TXN_BLOCKED))
                return (mc->mc_txn->mt_flags & MDB_TXN_RDONLY) ? EACCES : MDB_BAD_TXN;
 
        if (key->mv_size-1 >= ENV_MAXKEY(env))
@@ -6815,7 +6826,7 @@ mdb_cursor_del(MDB_cursor *mc, unsigned int flags)
        MDB_page        *mp;
        int rc;
 
-       if (mc->mc_txn->mt_flags & (MDB_TXN_RDONLY|MDB_TXN_ERROR))
+       if (mc->mc_txn->mt_flags & (MDB_TXN_RDONLY|MDB_TXN_BLOCKED))
                return (mc->mc_txn->mt_flags & MDB_TXN_RDONLY) ? EACCES : MDB_BAD_TXN;
 
        if (!(mc->mc_flags & C_INITIALIZED))
@@ -7345,7 +7356,7 @@ mdb_cursor_open(MDB_txn *txn, MDB_dbi dbi, MDB_cursor **ret)
        if (!ret || !TXN_DBI_EXIST(txn, dbi, DB_VALID))
                return EINVAL;
 
-       if (txn->mt_flags & MDB_TXN_ERROR)
+       if (txn->mt_flags & MDB_TXN_BLOCKED)
                return MDB_BAD_TXN;
 
        /* Allow read access to the freelist */
@@ -7380,7 +7391,7 @@ mdb_cursor_renew(MDB_txn *txn, MDB_cursor *mc)
        if ((mc->mc_flags & C_UNTRACK) || txn->mt_cursors)
                return EINVAL;
 
-       if (txn->mt_flags & MDB_TXN_ERROR)
+       if (txn->mt_flags & MDB_TXN_BLOCKED)
                return MDB_BAD_TXN;
 
        mdb_cursor_init(mc, txn, mc->mc_dbi, mc->mc_xcursor);
@@ -7399,7 +7410,7 @@ mdb_cursor_count(MDB_cursor *mc, size_t *countp)
        if (mc->mc_xcursor == NULL)
                return MDB_INCOMPATIBLE;
 
-       if (mc->mc_txn->mt_flags & MDB_TXN_ERROR)
+       if (mc->mc_txn->mt_flags & MDB_TXN_BLOCKED)
                return MDB_BAD_TXN;
 
        if (!(mc->mc_flags & C_INITIALIZED))
@@ -8126,7 +8137,7 @@ mdb_del(MDB_txn *txn, MDB_dbi dbi,
        if (!key || !TXN_DBI_EXIST(txn, dbi, DB_USRVALID))
                return EINVAL;
 
-       if (txn->mt_flags & (MDB_TXN_RDONLY|MDB_TXN_ERROR))
+       if (txn->mt_flags & (MDB_TXN_RDONLY|MDB_TXN_BLOCKED))
                return (txn->mt_flags & MDB_TXN_RDONLY) ? EACCES : MDB_BAD_TXN;
 
        if (!F_ISSET(txn->mt_dbs[dbi].md_flags, MDB_DUPSORT)) {
@@ -9314,7 +9325,7 @@ int mdb_dbi_open(MDB_txn *txn, const char *name, unsigned int flags, MDB_dbi *db
 
        if (flags & ~VALID_FLAGS)
                return EINVAL;
-       if (txn->mt_flags & MDB_TXN_ERROR)
+       if (txn->mt_flags & MDB_TXN_BLOCKED)
                return MDB_BAD_TXN;
 
        /* main DB? */
@@ -9412,7 +9423,7 @@ mdb_stat(MDB_txn *txn, MDB_dbi dbi, MDB_stat *arg)
        if (!arg || !TXN_DBI_EXIST(txn, dbi, DB_VALID))
                return EINVAL;
 
-       if (txn->mt_flags & MDB_TXN_ERROR)
+       if (txn->mt_flags & MDB_TXN_BLOCKED)
                return MDB_BAD_TXN;
 
        if (txn->mt_dbflags[dbi] & DB_STALE) {