ITS#9376 Fixes for repeated deletes with xcursor

On DUPSORT DBs, must initialize xcursor regardless of whether caller requested its data. Also in cursor_prev must check whether cursor index is still within range before using it.
4 years ago · c131c853f0
parent 01db7a099e
commit c131c853f0
1 changed files with 108 additions and 102 deletions
--- a/libraries/liblmdb/mdb.c
+++ b/libraries/liblmdb/mdb.c
@ -5942,16 +5942,12 @@ skip:

 	if (F_ISSET(leaf->mn_flags, F_DUPDATA)) {
 		mdb_xcursor_init1(mc, leaf);
-	}
-	if (data) {
-		if ((rc = mdb_node_read(mc, leaf, data)) != MDB_SUCCESS)
-			return rc;
-
-		if (F_ISSET(leaf->mn_flags, F_DUPDATA)) {
 		rc = mdb_cursor_first(&mc->mc_xcursor->mx_cursor, data, NULL);
 		if (rc != MDB_SUCCESS)
 			return rc;
-		}
+	} else if (data) {
+		if ((rc = mdb_node_read(mc, leaf, data)) != MDB_SUCCESS)
+			return rc;
 	}

 	MDB_GET_KEY(leaf, key);
@ -5975,7 +5971,8 @@ mdb_cursor_prev(MDB_cursor *mc, MDB_val *key, MDB_val *data, MDB_cursor_op op)

 	mp = mc->mc_pg[mc->mc_top];

-	if (mc->mc_db->md_flags & MDB_DUPSORT) {
+	if ((mc->mc_db->md_flags & MDB_DUPSORT) &&
+		mc->mc_ki[mc->mc_top] < NUMKEYS(mp)) {
 		leaf = NODEPTR(mp, mc->mc_ki[mc->mc_top]);
 		if (F_ISSET(leaf->mn_flags, F_DUPDATA)) {
 			if (op == MDB_PREV || op == MDB_PREV_DUP) {
@ -6014,27 +6011,25 @@ mdb_cursor_prev(MDB_cursor *mc, MDB_val *key, MDB_val *data, MDB_cursor_op op)
 	DPRINTF(("==> cursor points to page %"Z"u with %u keys, key index %u",
 	    mdb_dbg_pgno(mp), NUMKEYS(mp), mc->mc_ki[mc->mc_top]));

+	if (!IS_LEAF(mp))
+		return MDB_CORRUPTED;
+
 	if (IS_LEAF2(mp)) {
 		key->mv_size = mc->mc_db->md_pad;
 		key->mv_data = LEAF2KEY(mp, mc->mc_ki[mc->mc_top], key->mv_size);
 		return MDB_SUCCESS;
 	}

-	mdb_cassert(mc, IS_LEAF(mp));
 	leaf = NODEPTR(mp, mc->mc_ki[mc->mc_top]);

 	if (F_ISSET(leaf->mn_flags, F_DUPDATA)) {
 		mdb_xcursor_init1(mc, leaf);
-	}
-	if (data) {
-		if ((rc = mdb_node_read(mc, leaf, data)) != MDB_SUCCESS)
-			return rc;
-
-		if (F_ISSET(leaf->mn_flags, F_DUPDATA)) {
 		rc = mdb_cursor_last(&mc->mc_xcursor->mx_cursor, data, NULL);
 		if (rc != MDB_SUCCESS)
 			return rc;
-		}
+	} else if (data) {
+		if ((rc = mdb_node_read(mc, leaf, data)) != MDB_SUCCESS)
+			return rc;
 	}

 	MDB_GET_KEY(leaf, key);
@ -6190,9 +6185,6 @@ set1:

 	if (F_ISSET(leaf->mn_flags, F_DUPDATA)) {
 		mdb_xcursor_init1(mc, leaf);
-	}
-	if (data) {
-		if (F_ISSET(leaf->mn_flags, F_DUPDATA)) {
 		if (op == MDB_SET || op == MDB_SET_KEY || op == MDB_SET_RANGE) {
 			rc = mdb_cursor_first(&mc->mc_xcursor->mx_cursor, data, NULL);
 		} else {
@ -6207,7 +6199,8 @@ set1:
 			if (rc != MDB_SUCCESS)
 				return rc;
 		}
-		} else if (op == MDB_GET_BOTH || op == MDB_GET_BOTH_RANGE) {
+	} else if (data) {
+		if (op == MDB_GET_BOTH || op == MDB_GET_BOTH_RANGE) {
 			MDB_val olddata;
 			MDB_cmp_func *dcmp;
 			if ((rc = mdb_node_read(mc, leaf, &olddata)) != MDB_SUCCESS)
@ -6265,22 +6258,23 @@ mdb_cursor_first(MDB_cursor *mc, MDB_val *key, MDB_val *data)
 	mc->mc_ki[mc->mc_top] = 0;

 	if (IS_LEAF2(mc->mc_pg[mc->mc_top])) {
+		if ( key ) {
 			key->mv_size = mc->mc_db->md_pad;
 			key->mv_data = LEAF2KEY(mc->mc_pg[mc->mc_top], 0, key->mv_size);
+		}
 		return MDB_SUCCESS;
 	}

-	if (data) {
 	if (F_ISSET(leaf->mn_flags, F_DUPDATA)) {
 		mdb_xcursor_init1(mc, leaf);
 		rc = mdb_cursor_first(&mc->mc_xcursor->mx_cursor, data, NULL);
 		if (rc)
 			return rc;
-		} else {
+	} else if (data) {
 		if ((rc = mdb_node_read(mc, leaf, data)) != MDB_SUCCESS)
 			return rc;
 	}
-	}
+
 	MDB_GET_KEY(leaf, key);
 	return MDB_SUCCESS;
 }
@ -6307,22 +6301,22 @@ mdb_cursor_last(MDB_cursor *mc, MDB_val *key, MDB_val *data)
 	leaf = NODEPTR(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top]);

 	if (IS_LEAF2(mc->mc_pg[mc->mc_top])) {
+		if (key) {
 			key->mv_size = mc->mc_db->md_pad;
 			key->mv_data = LEAF2KEY(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top], key->mv_size);
+		}
 		return MDB_SUCCESS;
 	}

-	if (data) {
 	if (F_ISSET(leaf->mn_flags, F_DUPDATA)) {
 		mdb_xcursor_init1(mc, leaf);
 		rc = mdb_cursor_last(&mc->mc_xcursor->mx_cursor, data, NULL);
 		if (rc)
 			return rc;
-		} else {
+	} else if (data) {
 		if ((rc = mdb_node_read(mc, leaf, data)) != MDB_SUCCESS)
 			return rc;
 	}
-	}

 	MDB_GET_KEY(leaf, key);
 	return MDB_SUCCESS;
@ -7102,6 +7096,8 @@ mdb_cursor_del(MDB_cursor *mc, unsigned int flags)
 		return rc;

 	mp = mc->mc_pg[mc->mc_top];
+	if (!IS_LEAF(mp))
+		return MDB_CORRUPTED;
 	if (IS_LEAF2(mp))
 		goto del_key;
 	leaf = NODEPTR(mp, mc->mc_ki[mc->mc_top]);
@ -8473,28 +8469,31 @@ mdb_cursor_del0(MDB_cursor *mc)
 		}
 	}
 	rc = mdb_rebalance(mc);
+	if (rc)
+		goto fail;

-	if (rc == MDB_SUCCESS) {
 	/* DB is totally empty now, just bail out.
 	 * Other cursors adjustments were already done
 	 * by mdb_rebalance and aren't needed here.
 	 */
-		if (!mc->mc_snum)
+	if (!mc->mc_snum) {
+		mc->mc_flags |= C_EOF;
 		return rc;
+	}

+	ki = mc->mc_ki[mc->mc_top];
 	mp = mc->mc_pg[mc->mc_top];
 	nkeys = NUMKEYS(mp);

 	/* Adjust other cursors pointing to mp */
 	for (m2 = mc->mc_txn->mt_cursors[dbi]; !rc && m2; m2=m2->mc_next) {
 		m3 = (mc->mc_flags & C_SUB) ? &m2->mc_xcursor->mx_cursor : m2;
-			if (! (m2->mc_flags & m3->mc_flags & C_INITIALIZED))
+		if (!(m2->mc_flags & m3->mc_flags & C_INITIALIZED))
 			continue;
 		if (m3->mc_snum < mc->mc_snum)
 			continue;
 		if (m3->mc_pg[mc->mc_top] == mp) {
 			/* if m3 points past last node in page, find next sibling */
-				if (m3->mc_ki[mc->mc_top] >= mc->mc_ki[mc->mc_top]) {
 			if (m3->mc_ki[mc->mc_top] >= nkeys) {
 				rc = mdb_cursor_sibling(m3, 1);
 				if (rc == MDB_NOTFOUND) {
@ -8502,8 +8501,12 @@ mdb_cursor_del0(MDB_cursor *mc)
 					rc = MDB_SUCCESS;
 					continue;
 				}
+				if (rc)
+					goto fail;
 			}
-					if (mc->mc_db->md_flags & MDB_DUPSORT) {
+			if (m3->mc_ki[mc->mc_top] >= ki ||
+				/* moved to right sibling */ m3->mc_pg[mc->mc_top] != mp) {
+				if (m3->mc_xcursor && !(m3->mc_flags & C_EOF)) {
 					MDB_node *node = NODEPTR(m3->mc_pg[m3->mc_top], m3->mc_ki[m3->mc_top]);
 					/* If this node has dupdata, it may need to be reinited
 					 * because its data has moved.
@ -8517,16 +8520,19 @@ mdb_cursor_del0(MDB_cursor *mc)
 								m3->mc_xcursor->mx_cursor.mc_pg[0] = NODEDATA(node);
 						} else {
 							mdb_xcursor_init1(m3, node);
-								m3->mc_xcursor->mx_cursor.mc_flags |= C_DEL;
-							}
+							rc = mdb_cursor_first(&m3->mc_xcursor->mx_cursor, NULL, NULL);
+							if (rc)
+								goto fail;
 						}
 					}
+					m3->mc_xcursor->mx_cursor.mc_flags |= C_DEL;
 				}
+				m3->mc_flags |= C_DEL;
 			}
 		}
-		mc->mc_flags |= C_DEL;
 	}

+fail:
 	if (rc)
 		mc->mc_txn->mt_flags |= MDB_TXN_ERROR;
 	return rc;