Addresses #2119 refs[t:2119] Merged 2119 fix into main

git-svn-id: file:///svn/toku/tokudb@15513 c7de825b-a66e-492c-adef-691d508d4ae1

src/tests/Makefile

@@ -110,6 +110,7 @@ BDB_DONTRUN_TESTS = \
 	backwards_10_each_le_and_msg \
 	test_dupsort_set_range_reverse \
 	test_dupsort_get_both_range_reverse \
+	test_cursor_delete_2119 \
 #\ ends prev line
 
 # checkpoint tests depend on this header file, 

src/tests/test_cursor_delete_2119.c

+/* -*- mode: C; c-basic-offset: 4 -*- */
+#ident "Copyright (c) 2007 Tokutek Inc.  All rights reserved."
+#include "test.h"
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <assert.h>
+#include <unistd.h>
+#include <memory.h>
+#include <errno.h>
+#include <sys/stat.h>
+#include <db.h>
+
+
+static DB_ENV *dbenv;
+static DB *db;
+static DB_TXN * txn;
+static DBC *cursor;
+
+static void
+test_cursor_delete_2119 (u_int32_t c_del_flags, u_int32_t txn_isolation_flags) {
+    int r;
+    r = system("rm -rf " ENVDIR);
+        CKERR(r);
+    r = toku_os_mkdir(ENVDIR, S_IRWXU+S_IRWXG+S_IRWXO);
+        CKERR(r);
+    DBT key,val;
+
+    r = db_env_create(&dbenv, 0);                                                            CKERR(r);
+    r = dbenv->open(dbenv, ENVDIR, DB_PRIVATE|DB_INIT_MPOOL|DB_CREATE|DB_INIT_TXN|DB_INIT_LOCK, 0);       CKERR(r);
+
+    r = db_create(&db, dbenv, 0);                                                            CKERR(r);
+    r = dbenv->txn_begin(dbenv, 0, &txn, txn_isolation_flags);                               CKERR(r);
+    r = db->open(db, txn, "primary.db", NULL, DB_BTREE, DB_CREATE, 0600);                    CKERR(r);
+    r = txn->commit(txn, 0);                                                                 CKERR(r);
+
+    r = dbenv->txn_begin(dbenv, 0, &txn, txn_isolation_flags);                               CKERR(r);
+    r = db->put(db, txn, dbt_init(&key, "a", 2), dbt_init(&val, "b", 2), DB_YESOVERWRITE);   CKERR(r);
+    r = txn->commit(txn, 0);                                                                 CKERR(r);
+
+    r = dbenv->txn_begin(dbenv, 0, &txn, txn_isolation_flags);                               CKERR(r);
+    r = db->del(db, txn, dbt_init(&key, "a", 2), 0);                                         CKERR(r);
+    r = txn->commit(txn, 0);                                                                 CKERR(r);
+
+    r = dbenv->txn_begin(dbenv, 0, &txn, txn_isolation_flags);                               CKERR(r);
+    r = db->put(db, txn, dbt_init(&key, "a", 2), dbt_init(&val, "c", 2), DB_YESOVERWRITE);   CKERR(r);
+
+    cursor=cursor;
+
+    r = db->cursor(db, txn, &cursor, 0);                                                     CKERR(r);
+    r = cursor->c_get(cursor, dbt_init_malloc(&key), dbt_init_malloc(&val), DB_FIRST);       CKERR(r);
+    assert(strcmp(key.data, "a")==0);  toku_free(key.data);
+    assert(strcmp(val.data, "c")==0);  toku_free(val.data);
+    r = cursor->c_del(cursor, c_del_flags);                                                  CKERR(r);
+    r = cursor->c_del(cursor, c_del_flags);                                                  assert(r==DB_KEYEMPTY);
+    r = cursor->c_get(cursor, dbt_init_malloc(&key), dbt_init_malloc(&val), DB_NEXT);        assert(r==DB_NOTFOUND);
+
+    r = cursor->c_close(cursor);                                                             CKERR(r);
+    r = txn->commit(txn, 0);                                                                 CKERR(r);
+
+
+
+    r = db->close(db, 0);                                                                    CKERR(r);
+    r = dbenv->close(dbenv, 0);                                                              CKERR(r);
+}
+
+int
+test_main(int argc, char *argv[]) {
+
+    parse_args(argc, argv);
+
+    int isolation;
+    int read_prelocked;
+    int write_prelocked;
+    for (isolation = 0; isolation < 2; isolation++) {
+        u_int32_t isolation_flag = isolation ? DB_READ_UNCOMMITTED : 0;
+        for (read_prelocked = 0; read_prelocked < 2; read_prelocked++) {
+            u_int32_t read_prelocked_flag = read_prelocked ? DB_PRELOCKED : 0;
+            for (write_prelocked = 0; write_prelocked < 2; write_prelocked++) {
+                u_int32_t write_prelocked_flag = write_prelocked ? DB_PRELOCKED_WRITE : 0;
+                test_cursor_delete_2119(read_prelocked_flag | write_prelocked_flag,
+                                        isolation_flag);
+            }
+        }
+    }
+  
+    
+    return 0;
+}

src/ydb.c

@@ -1780,13 +1780,18 @@ grab_range_lock(RANGE_LOCK_REQUEST request) {
 //This is the user level callback function given to ydb layer functions like
 //toku_c_getf_first
 
+typedef struct __toku_is_write_op {
+    BOOL is_write_op;
+} WRITE_OP;
+
 typedef struct query_context_base_t {
     BRT_CURSOR  c;
     DB_TXN     *txn;
     DB         *db;
     void       *f_extra;
-    BOOL        do_locking;
     int         r_user_callback;
+    BOOL        do_locking;
+    BOOL        is_write_op;
 } *QUERY_CONTEXT_BASE, QUERY_CONTEXT_BASE_S;
 
 typedef struct query_context_t {
@@ -1803,13 +1808,15 @@ typedef struct query_context_with_input_t {
 
 
 static void
-query_context_base_init(QUERY_CONTEXT_BASE context, DBC *c, u_int32_t flag, void *extra) {
+query_context_base_init(QUERY_CONTEXT_BASE context, DBC *c, u_int32_t flag, WRITE_OP is_write_op, void *extra) {
     context->c       = dbc_struct_i(c)->c;
     context->txn     = dbc_struct_i(c)->txn;
     context->db      = c->dbp;
     context->f_extra = extra;
+    context->is_write_op = is_write_op.is_write_op;
     u_int32_t lock_flags = get_prelocked_flags(flag, dbc_struct_i(c)->txn);
     flag &= ~lock_flags;
+    if (context->is_write_op) lock_flags &= DB_PRELOCKED_WRITE; // Only care about whether already locked for write
     assert(flag==0);
     context->do_locking = (BOOL)(context->db->i->lt!=NULL && !lock_flags);
     context->r_user_callback = 0;
@@ -1817,13 +1824,22 @@ query_context_base_init(QUERY_CONTEXT_BASE context, DBC *c, u_int32_t flag, void
 
 static void
 query_context_init(QUERY_CONTEXT context, DBC *c, u_int32_t flag, YDB_CALLBACK_FUNCTION f, void *extra) {
-    query_context_base_init(&context->base, c, flag, extra);
+    WRITE_OP is_write = {FALSE};
+    query_context_base_init(&context->base, c, flag, is_write, extra);
+    context->f = f;
+}
+
+static void
+query_context_init_write_op(QUERY_CONTEXT context, DBC *c, u_int32_t flag, YDB_CALLBACK_FUNCTION f, void *extra) {
+    WRITE_OP is_write = {TRUE};
+    query_context_base_init(&context->base, c, flag, is_write, extra);
     context->f = f;
 }
 
 static void
 query_context_with_input_init(QUERY_CONTEXT_WITH_INPUT context, DBC *c, u_int32_t flag, DBT *key, DBT *val, YDB_CALLBACK_FUNCTION f, void *extra) {
-    query_context_base_init(&context->base, c, flag, extra);
+    WRITE_OP is_write = {FALSE};
+    query_context_base_init(&context->base, c, flag, is_write, extra);
     context->f         = f;
     context->input_key = key;
     context->input_val = val;
@@ -1850,7 +1866,7 @@ toku_c_del(DBC * c, u_int32_t flags) {
     else {
         if (do_locking) {
             QUERY_CONTEXT_S context;
-            query_context_init(&context, c, lock_flags, NULL, NULL);
+            query_context_init_write_op(&context, c, lock_flags, NULL, NULL);
             //We do not need a read lock, we must already have it.
             r = toku_c_getf_current_binding(c, DB_PRELOCKED, c_del_callback, &context);
         }
@@ -1872,6 +1888,7 @@ c_del_callback(DBT const *key, DBT const *val, void *extra) {
     int r;
 
     assert(context->do_locking);
+    assert(context->is_write_op);
     assert(key!=NULL);
     assert(val!=NULL);
     //Lock:
@@ -2667,7 +2684,8 @@ typedef struct {
 
 static void
 query_context_heaviside_init(QUERY_CONTEXT_HEAVISIDE context, DBC *c, u_int32_t flag, YDB_HEAVISIDE_CALLBACK_FUNCTION f, void *extra, HEAVI_WRAPPER wrapper) {
-    query_context_base_init(&context->base, c, flag, extra);
+    WRITE_OP is_write = {FALSE};
+    query_context_base_init(&context->base, c, flag, is_write, extra);
     context->f       = f;
     context->wrapper = wrapper;
 }