addresses #909

use DB_PRELOCK when we know we have lock acquired

git-svn-id: file:///svn/mysql/tokudb-engine/src@4481 c7de825b-a66e-492c-adef-691d508d4ae1

storage/tokudb/ha_tokudb.cc

@@ -699,7 +699,7 @@ ha_tokudb::ha_tokudb(handlerton * hton, TABLE_SHARE * table_arg)
     // flags defined in sql\handler.h
     int_table_flags(HA_REC_NOT_IN_SEQ | HA_FAST_KEY_READ | HA_NULL_IN_KEY | HA_CAN_INDEX_BLOBS | HA_PRIMARY_KEY_IN_READ_INDEX | 
                     HA_FILE_BASED | HA_CAN_GEOMETRY | HA_AUTO_PART_KEY | HA_TABLE_SCAN_ON_INDEX), 
-    changed_rows(0), last_dup_key((uint) - 1), version(0), using_ignore(0), last_cursor_error(0),primary_key_offsets(NULL) {
+    changed_rows(0), last_dup_key((uint) - 1), version(0), using_ignore(0), last_cursor_error(0),lock_grabbed(false), primary_key_offsets(NULL) {
 }
 
 static const char *ha_tokudb_exts[] = {
@@ -2215,6 +2215,7 @@ int ha_tokudb::index_init(uint keynr, bool sorted) {
     }
     active_index = keynr;
     last_cursor_error = 0;
+    lock_grabbed = false;
     DBUG_ASSERT(keynr <= table->s->keys);
     DBUG_ASSERT(share->key_file[keynr]);
     if ((error = share->key_file[keynr]->cursor(share->key_file[keynr], transaction, &cursor, table->reginfo.lock_type > TL_WRITE_ALLOW_READ ? 0 : 0))) {
@@ -2231,6 +2232,7 @@ int ha_tokudb::index_init(uint keynr, bool sorted) {
 int ha_tokudb::index_end() {
     TOKUDB_DBUG_ENTER("ha_tokudb::index_end %p", this);
     int error = 0;
+    lock_grabbed = false;
     if (cursor) {
         DBUG_PRINT("enter", ("table: '%s'", table_share->table_name.str));
         error = cursor->c_close(cursor);
@@ -2568,11 +2570,12 @@ int ha_tokudb::index_next(uchar * buf) {
     TOKUDB_DBUG_ENTER("ha_tokudb::index_next");
     int error; 
     DBT row;
+    u_int32_t flags = lock_grabbed ? (DB_NEXT | DB_PRELOCKED) : DB_NEXT;
     CHECK_VALID_CURSOR();
     
     statistic_increment(table->in_use->status_var.ha_read_next_count, &LOCK_status);
     bzero((void *) &row, sizeof(row));
-    error = read_row(cursor->c_get(cursor, &last_key, &row, DB_NEXT), buf, active_index, &row, &last_key, 1);
+    error = read_row(cursor->c_get(cursor, &last_key, &row, flags), buf, active_index, &row, &last_key, 1);
 cleanup:
     TOKUDB_DBUG_RETURN(error);
 }
@@ -2601,10 +2604,12 @@ int ha_tokudb::index_next_same(uchar * buf, const uchar * key, uint keylen) {
         !(table->key_info[active_index].flags & HA_NOSAME) && 
         !(table->key_info[active_index].flags & HA_END_SPACE_KEY)) {
 
-        error = cursor->c_get(cursor, &last_key, &row, DB_NEXT_DUP);
+        u_int32_t flags = lock_grabbed ? (DB_NEXT_DUP | DB_PRELOCKED) : DB_NEXT;
+        error = cursor->c_get(cursor, &last_key, &row, flags);
         error = read_row(error, buf, active_index, &row, &last_key, 1);
     } else {
-        error = read_row(cursor->c_get(cursor, &last_key, &row, DB_NEXT), buf, active_index, &row, &last_key, 1);
+        u_int32_t flags = lock_grabbed ? (DB_NEXT | DB_PRELOCKED) : DB_NEXT;
+        error = read_row(cursor->c_get(cursor, &last_key, &row, flags), buf, active_index, &row, &last_key, 1);
         if (!error &&::key_cmp_if_same(table, key, active_index, keylen))
             error = HA_ERR_END_OF_FILE;
     }
@@ -2624,11 +2629,12 @@ int ha_tokudb::index_next_same(uchar * buf, const uchar * key, uint keylen) {
 int ha_tokudb::index_prev(uchar * buf) {
     TOKUDB_DBUG_ENTER("ha_tokudb::index_prev");
     int error;
+    u_int32_t flags = lock_grabbed ? (DB_PREV | DB_PRELOCKED) : DB_NEXT;
     CHECK_VALID_CURSOR();
     DBT row;
     statistic_increment(table->in_use->status_var.ha_read_prev_count, &LOCK_status);
     bzero((void *) &row, sizeof(row));
-    error = read_row(cursor->c_get(cursor, &last_key, &row, DB_PREV), buf, active_index, &row, &last_key, 1);
+    error = read_row(cursor->c_get(cursor, &last_key, &row, flags), buf, active_index, &row, &last_key, 1);
 cleanup:
     TOKUDB_DBUG_RETURN(error);
 }
@@ -2687,12 +2693,23 @@ int ha_tokudb::rnd_init(bool scan) {
     TOKUDB_DBUG_ENTER("ha_tokudb::rnd_init");
     int error;
     current_row.flags = DB_DBT_REALLOC;
+    lock_grabbed = false;
     if (scan) {
         DB* db = share->key_file[primary_key];
         error = db->pre_acquire_read_lock(db, transaction, db->dbt_neg_infty(), NULL, db->dbt_pos_infty(), NULL);
         if (error) { last_cursor_error = error; goto cleanup; }
     }
     error = index_init(primary_key, 0);
+    if (error) { goto cleanup;}
+
+    //
+    // only want to set lock_grabbed to true after index_init
+    // successfully executed for two reasons:
+    // 1) index_init will reset it to false anyway
+    // 2) if it fails, we don't want prelocking on,
+    //
+    if (scan) { lock_grabbed = true; }
+    error = 0;
 cleanup:
     TOKUDB_DBUG_RETURN(error);
 }
@@ -2702,6 +2719,7 @@ int ha_tokudb::rnd_init(bool scan) {
 //
 int ha_tokudb::rnd_end() {
     TOKUDB_DBUG_ENTER("ha_tokudb::rnd_end");
+    lock_grabbed = false;
     TOKUDB_DBUG_RETURN(index_end());
 }
 
@@ -2718,6 +2736,7 @@ int ha_tokudb::rnd_next(uchar * buf) {
     TOKUDB_DBUG_ENTER("ha_tokudb::ha_tokudb::rnd_next");
     int error;
     DBT row;    
+    u_int32_t flags = lock_grabbed ? (DB_NEXT | DB_PRELOCKED) : DB_NEXT;
 
     CHECK_VALID_CURSOR()
     //
@@ -2727,7 +2746,7 @@ int ha_tokudb::rnd_next(uchar * buf) {
     statistic_increment(table->in_use->status_var.ha_read_rnd_next_count, &LOCK_status);
     bzero((void *) &row, sizeof(row));
     DBUG_DUMP("last_key", (uchar *) last_key.data, last_key.size);
-    error = read_row(cursor->c_get(cursor, &last_key, &row, DB_NEXT), buf, primary_key, &row, &last_key, 1);
+    error = read_row(cursor->c_get(cursor, &last_key, &row, flags), buf, primary_key, &row, &last_key, 1);
 cleanup:
     TOKUDB_DBUG_RETURN(error);
 }
@@ -2786,7 +2805,7 @@ int ha_tokudb::read_range_first(
     uchar end_key_buff [table_share->max_key_length + MAX_REF_PARTS * 3];
     bzero((void *) &start_dbt_key, sizeof(start_dbt_key));
     bzero((void *) &end_dbt_key, sizeof(end_dbt_key));
-
+    lock_grabbed = false;
 
 
     if (start_key) {
@@ -2841,7 +2860,7 @@ int ha_tokudb::read_range_first(
         }
         goto cleanup; 
     }
-
+    lock_grabbed = true;
     error = handler::read_range_first(start_key, end_key, eq_range, sorted);
 
 cleanup:
@@ -2849,7 +2868,13 @@ int ha_tokudb::read_range_first(
 }
 int ha_tokudb::read_range_next()
 {
-    return handler::read_range_next();
+    TOKUDB_DBUG_ENTER("ha_tokudb::read_range_next");
+    int error;
+    error = handler::read_range_next();
+    if (error) {
+        lock_grabbed = false;
+    }
+    TOKUDB_DBUG_RETURN(error);
 }
 
 

storage/tokudb/ha_tokudb.h

@@ -123,6 +123,13 @@ class ha_tokudb : public handler {
     //
     int last_cursor_error;
 
+    //
+    // For instances where we successfully prelock a range or a table,
+    // we set this to TRUE so that successive cursor calls can know
+    // know to limit the locking overhead in a call to the fractal tree
+    //
+    bool lock_grabbed;
+
     PRIM_KEY_PART_INFO* primary_key_offsets;
 
     bool fix_rec_buff_for_blob(ulong length);