ITS#8321 track temporary cursors

In rebalance/split operations, temporary cursors need to be visible
to propagate fixups

diff --git a/libraries/liblmdb/mdb.c b/libraries/liblmdb/mdb.c
index 638f5a94a27510e3c2d2a12dd5599b5923bf6f16..752fdfaafcf8b20270a0453f0c1b8895ebb35501 100644 (file)
--- a/libraries/liblmdb/mdb.c
+++ b/libraries/liblmdb/mdb.c
@@ -7524,6 +7524,22 @@ mdb_update_key(MDB_cursor *mc, MDB_val *key)
 static void
 mdb_cursor_copy(const MDB_cursor *csrc, MDB_cursor *cdst);
 
+/** Track a temporary cursor */
+#define CURSOR_TMP_TRACK(mc, mn, dummy, tracked) \
+       if (mc->mc_flags & C_SUB) { \
+               dummy.mc_flags =  C_INITIALIZED; \
+               dummy.mc_xcursor = (MDB_xcursor *)&mn; \
+               tracked = &dummy; \
+       } else { \
+               tracked = &mn; \
+       } \
+       tracked->mc_next = mc->mc_txn->mt_cursors[mc->mc_dbi]; \
+       mc->mc_txn->mt_cursors[mc->mc_dbi] = tracked
+
+/** Stop tracking a temporary cursor */
+#define CURSOR_TMP_UNTRACK(mc, tracked) \
+       mc->mc_txn->mt_cursors[mc->mc_dbi] = tracked->mc_next
+
 /** Move a node from csrc to cdst.
  */
 static int
@@ -7679,6 +7695,7 @@ mdb_node_move(MDB_cursor *csrc, MDB_cursor *cdst, int fromleft)
         */
        if (csrc->mc_ki[csrc->mc_top] == 0) {
                if (csrc->mc_ki[csrc->mc_top-1] != 0) {
+                       MDB_cursor dummy, *tracked;
                        if (IS_LEAF2(csrc->mc_pg[csrc->mc_top])) {
                                key.mv_data = LEAF2KEY(csrc->mc_pg[csrc->mc_top], 0, key.mv_size);
                        } else {
@@ -7691,7 +7708,11 @@ mdb_node_move(MDB_cursor *csrc, MDB_cursor *cdst, int fromleft)
                        mdb_cursor_copy(csrc, &mn);
                        mn.mc_snum--;
                        mn.mc_top--;
-                       if ((rc = mdb_update_key(&mn, &key)) != MDB_SUCCESS)
+                       /* We want mdb_rebalance to find mn when doing fixups */
+                       CURSOR_TMP_TRACK(csrc, mn, dummy, tracked);
+                       rc = mdb_update_key(&mn, &key);
+                       CURSOR_TMP_UNTRACK(csrc, tracked);
+                       if (rc)
                                return rc;
                }
                if (IS_BRANCH(csrc->mc_pg[csrc->mc_top])) {
@@ -7707,6 +7728,7 @@ mdb_node_move(MDB_cursor *csrc, MDB_cursor *cdst, int fromleft)
 
        if (cdst->mc_ki[cdst->mc_top] == 0) {
                if (cdst->mc_ki[cdst->mc_top-1] != 0) {
+                       MDB_cursor dummy, *tracked;
                        if (IS_LEAF2(csrc->mc_pg[csrc->mc_top])) {
                                key.mv_data = LEAF2KEY(cdst->mc_pg[cdst->mc_top], 0, key.mv_size);
                        } else {
@@ -7719,7 +7741,11 @@ mdb_node_move(MDB_cursor *csrc, MDB_cursor *cdst, int fromleft)
                        mdb_cursor_copy(cdst, &mn);
                        mn.mc_snum--;
                        mn.mc_top--;
-                       if ((rc = mdb_update_key(&mn, &key)) != MDB_SUCCESS)
+                       /* We want mdb_rebalance to find mn when doing fixups */
+                       CURSOR_TMP_TRACK(cdst, mn, dummy, tracked);
+                       rc = mdb_update_key(&mn, &key);
+                       CURSOR_TMP_UNTRACK(cdst, tracked);
+                       if (rc)
                                return rc;
                }
                if (IS_BRANCH(cdst->mc_pg[cdst->mc_top])) {
@@ -8077,24 +8103,13 @@ mdb_rebalance(MDB_cursor *mc)
                if (!fromleft) {
                        rc = mdb_page_merge(&mn, mc);
                } else {
-                       MDB_cursor dummy;
+                       MDB_cursor dummy, *tracked;
                        oldki += NUMKEYS(mn.mc_pg[mn.mc_top]);
                        mn.mc_ki[mn.mc_top] += mc->mc_ki[mn.mc_top] + 1;
                        /* We want mdb_rebalance to find mn when doing fixups */
-                       if (mc->mc_flags & C_SUB) {
-                               dummy.mc_flags = C_INITIALIZED;
-                               dummy.mc_next = mc->mc_txn->mt_cursors[mc->mc_dbi];
-                               mc->mc_txn->mt_cursors[mc->mc_dbi] = &dummy;
-                               dummy.mc_xcursor = (MDB_xcursor *)&mn;
-                       } else {
-                               mn.mc_next = mc->mc_txn->mt_cursors[mc->mc_dbi];
-                               mc->mc_txn->mt_cursors[mc->mc_dbi] = &mn;
-                       }
+                       CURSOR_TMP_TRACK(mc, mn, dummy, tracked);
                        rc = mdb_page_merge(mc, &mn);
-                       if (mc->mc_flags & C_SUB)
-                               mc->mc_txn->mt_cursors[mc->mc_dbi] = dummy.mc_next;
-                       else
-                               mc->mc_txn->mt_cursors[mc->mc_dbi] = mn.mc_next;
+                       CURSOR_TMP_UNTRACK(mc, tracked);
                        mdb_cursor_copy(&mn, mc);
                }
                mc->mc_flags &= ~C_EOF;
@@ -8460,10 +8475,14 @@ mdb_page_split(MDB_cursor *mc, MDB_val *newkey, MDB_val *newdata, pgno_t newpgno
         */
        if (SIZELEFT(mn.mc_pg[ptop]) < mdb_branch_size(env, &sepkey)) {
                int snum = mc->mc_snum;
+               MDB_cursor dummy, *tracked;
                mn.mc_snum--;
                mn.mc_top--;
                did_split = 1;
+               /* We want other splits to find mn when doing fixups */
+               CURSOR_TMP_TRACK(mc, mn, dummy, tracked);
                rc = mdb_page_split(&mn, &sepkey, NULL, rp->mp_pgno, 0);
+               CURSOR_TMP_UNTRACK(mc, tracked);
                if (rc)
                        goto done;