Remember oldest reader txnid

[openldap] / libraries / liblmdb / mdb.c

diff --git a/libraries/liblmdb/mdb.c b/libraries/liblmdb/mdb.c
index b7a6ac4894447ab8abfbc99b4776e5d9e763e3a7..1e784ae11de1b8ad549037fd2b11aceed37a7cef 100644 (file)
--- a/libraries/liblmdb/mdb.c
+++ b/libraries/liblmdb/mdb.c
@@ -1083,6 +1083,7 @@ typedef struct MDB_xcursor {
 typedef struct MDB_pgstate {
        pgno_t          *mf_pghead;     /**< Reclaimed freeDB pages, or NULL before use */
        txnid_t         mf_pglast;      /**< ID of last used record, or 0 if !mf_pghead */
+       txnid_t         mf_pgoldest;    /**< ID of oldest reader last time we looked */
 } MDB_pgstate;
 
        /** The database environment. */
@@ -1110,6 +1111,7 @@ struct MDB_env {
        MDB_meta        *me_metas[2];   /**< pointers to the two meta pages */
        void            *me_pbuf;               /**< scratch area for DUPSORT put() */
        MDB_txn         *me_txn;                /**< current write transaction */
+       MDB_txn         *me_txn0;               /**< prealloc'd write transaction */
        size_t          me_mapsize;             /**< size of the data memory map */
        off_t           me_size;                /**< current file size */
        pgno_t          me_maxpg;               /**< me_mapsize / me_psize */
@@ -1120,6 +1122,7 @@ struct MDB_env {
        MDB_pgstate     me_pgstate;             /**< state of old pages from freeDB */
 #      define          me_pglast       me_pgstate.mf_pglast
 #      define          me_pghead       me_pgstate.mf_pghead
+#      define          me_pgoldest     me_pgstate.mf_pgoldest
        MDB_page        *me_dpages;             /**< list of malloc'd blocks for re-use */
        /** IDL of pages that became unused in a write txn */
        MDB_IDL         me_free_pgs;
@@ -1955,6 +1958,7 @@ mdb_page_alloc(MDB_cursor *mc, int num, MDB_page **mp)
        txnid_t oldest = 0, last;
        MDB_cursor_op op;
        MDB_cursor m2;
+       int found_old = 0;
 
        /* If there are any loose pages, just use them */
        if (num == 1 && txn->mt_loose_pgs) {
@@ -1996,8 +2000,8 @@ mdb_page_alloc(MDB_cursor *mc, int num, MDB_page **mp)
 
                if (op == MDB_FIRST) {  /* 1st iteration */
                        /* Prepare to fetch more and coalesce */
-                       oldest = mdb_find_oldest(txn);
                        last = env->me_pglast;
+                       oldest = env->me_pgoldest;
                        mdb_cursor_init(&m2, txn, FREE_DBI, NULL);
                        if (last) {
                                op = MDB_SET_RANGE;
@@ -2012,8 +2016,15 @@ mdb_page_alloc(MDB_cursor *mc, int num, MDB_page **mp)
 
                last++;
                /* Do not fetch more if the record will be too recent */
-               if (oldest <= last)
-                       break;
+               if (oldest <= last) {
+                       if (!found_old) {
+                               oldest = mdb_find_oldest(txn);
+                               env->me_pgoldest = oldest;
+                               found_old = 1;
+                       }
+                       if (oldest <= last)
+                               break;
+               }
                rc = mdb_cursor_get(&m2, &key, NULL, op);
                if (rc) {
                        if (rc == MDB_NOTFOUND)
@@ -2021,8 +2032,15 @@ mdb_page_alloc(MDB_cursor *mc, int num, MDB_page **mp)
                        goto fail;
                }
                last = *(txnid_t*)key.mv_data;
-               if (oldest <= last)
-                       break;
+               if (oldest <= last) {
+                       if (!found_old) {
+                               oldest = mdb_find_oldest(txn);
+                               env->me_pgoldest = oldest;
+                               found_old = 1;
+                       }
+                       if (oldest <= last)
+                               break;
+               }
                np = m2.mc_pg[m2.mc_top];
                leaf = NODEPTR(np, m2.mc_ki[m2.mc_top]);
                if ((rc = mdb_node_read(txn, leaf, &data)) != MDB_SUCCESS)
@@ -2607,6 +2625,10 @@ mdb_txn_begin(MDB_env *env, MDB_txn *parent, unsigned int flags, MDB_txn **ret)
        }
        size = tsize + env->me_maxdbs * (sizeof(MDB_db)+1);
        if (!(flags & MDB_RDONLY)) {
+               if (!parent) {
+                       txn = env->me_txn0;
+                       goto ok;
+               }
                size += env->me_maxdbs * sizeof(MDB_cursor *);
                /* child txns use parent's dbiseqs */
                if (!parent)
@@ -2634,6 +2656,7 @@ mdb_txn_begin(MDB_env *env, MDB_txn *parent, unsigned int flags, MDB_txn **ret)
        }
        txn->mt_env = env;
 
+ok:
        if (parent) {
                unsigned int i;
                txn->mt_u.dirty_list = malloc(sizeof(MDB_ID2)*MDB_IDL_UM_SIZE);
@@ -2676,9 +2699,10 @@ mdb_txn_begin(MDB_env *env, MDB_txn *parent, unsigned int flags, MDB_txn **ret)
        } else {
                rc = mdb_txn_renew0(txn);
        }
-       if (rc)
-               free(txn);
-       else {
+       if (rc) {
+               if (txn != env->me_txn0)
+                       free(txn);
+       } else {
                *ret = txn;
                DPRINTF(("begin txn %"Z"u%c %p on mdbenv %p, root page %"Z"u",
                        txn->mt_txnid, (txn->mt_flags & MDB_TXN_RDONLY) ? 'r' : 'w',
@@ -2805,7 +2829,8 @@ mdb_txn_abort(MDB_txn *txn)
        if ((txn->mt_flags & MDB_TXN_RDONLY) && txn->mt_u.reader)
                txn->mt_u.reader->mr_pid = 0;
 
-       free(txn);
+       if (txn != txn->mt_env->me_txn0)
+               free(txn);
 }
 
 /** Save the freelist as of this transaction to the freeDB.
@@ -3358,7 +3383,8 @@ done:
 
        if (env->me_txns)
                UNLOCK_MUTEX_W(env);
-       free(txn);
+       if (txn != env->me_txn0)
+               free(txn);
 
        return MDB_SUCCESS;
 
@@ -4489,6 +4515,22 @@ mdb_env_open(MDB_env *env, const char *path, unsigned int flags, mdb_mode_t mode
                if (!((flags & MDB_RDONLY) ||
                          (env->me_pbuf = calloc(1, env->me_psize))))
                        rc = ENOMEM;
+               if (!(flags & MDB_RDONLY)) {
+                       MDB_txn *txn;
+                       int tsize = sizeof(MDB_txn), size = tsize + env->me_maxdbs *
+                               (sizeof(MDB_db)+sizeof(MDB_cursor)+sizeof(unsigned int)+1);
+                       txn = calloc(1, size);
+                       if (txn) {
+                               txn->mt_dbs = (MDB_db *)((char *)txn + tsize);
+                               txn->mt_cursors = (MDB_cursor **)(txn->mt_dbs + env->me_maxdbs);
+                               txn->mt_dbiseqs = (unsigned int *)(txn->mt_cursors + env->me_maxdbs);
+                               txn->mt_dbflags = (unsigned char *)(txn->mt_dbiseqs + env->me_maxdbs);
+                               txn->mt_env = env;
+                               env->me_txn0 = txn;
+                       } else {
+                               rc = ENOMEM;
+                       }
+               }
        }
 
 leave: