Implement a lighter versio nof {{{DBC->c_getf_next}}}. Addresses #902.

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

src/ydb.c

@@ -1666,20 +1666,64 @@ static int toku_c_get(DBC * c, DBT * key, DBT * data, u_int32_t flag) {
 }
 
 static int locked_c_getf_next(DBC *c, u_int32_t flag, void(*f)(DBT const *key, DBT const *data, void *extra), void *extra) {
-    return toku_c_getf_next(c, flag, f, extra); // flags are grabbed inside.
+    toku_ydb_lock();  int r = toku_c_getf_next(c, flag, f, extra); toku_ydb_unlock(); return r;
 }
 
-static int toku_c_getf_next(DBC *c, u_int32_t flag, void(*f)(DBT const *key, DBT const *data, void *extra), void *extra) {
+static int toku_c_getf_next_old(DBC *c, u_int32_t flag, void(*f)(DBT const *key, DBT const *data, void *extra), void *extra) {
     DBT key,val;
     memset(&key, 0, sizeof(key));
     memset(&val, 0, sizeof(val));
     flag &= ~DB_PRELOCKED; // Get rid of the prelock flag, because c_get doesn't know about it.
     assert(flag==0);
-    int r = c->c_get(c, &key, &val, DB_NEXT);
+    int r = toku_c_get_noassociate(c, &key, &val, DB_NEXT | flag);
     if (r==0) f(&key, &val, extra);
     return r;
 }
 
+static int toku_c_getf_next(DBC *c, u_int32_t flag, void(*f)(DBT const *key, DBT const *data, void *extra), void *extra) {
+    HANDLE_PANICKED_DB(c->dbp);
+    //int prelocked = flag & DB_PRELOCKED;
+    flag &= ~DB_PRELOCKED; // Get rid of the prelock flag, because c_get doesn't know about it.
+    if (toku_c_uninitialized(c)) return toku_c_getf_next_old(c, flag, f, extra); //return toku_c_getf_first(c, flag, f, extra);
+    assert(flag==0);
+    TOKUTXN txn = c->i->txn ? c->i->txn->i->tokutxn : NULL;
+    DBT key,val;
+    memset(&key, 0, sizeof(key));
+    memset(&val, 0, sizeof(val));
+    DBT prevkey,prevval;
+    memset(&prevkey, 0, sizeof(key)); prevkey.flags = DB_DBT_MALLOC;
+    memset(&prevval, 0, sizeof(val)); prevval.flags = DB_DBT_MALLOC;
+    int r = brt_cursor_save_key_val(c->i->c, &prevkey, &prevval);
+    if (r!=0) goto cleanup;
+
+    DB *db=c->dbp;
+
+    unsigned int brtflags;
+    toku_brt_get_flags(db->i->brt, &brtflags);
+
+    int c_get_result = toku_brt_cursor_get(c->i->c, &key, &val,
+					   (brtflags & TOKU_DB_DUPSORT) ? DB_NEXT : DB_NEXT_NODUP,
+					   txn);
+    if (c_get_result!=0 && c_get_result!=DB_NOTFOUND) { r = c_get_result; goto cleanup; }
+    int found = c_get_result==0;
+    toku_lock_tree* lt = db->i->lt;
+    DB_TXN *txn_anc = toku_txn_ancestor(c->i->txn);
+    r = toku_txn_add_lt(txn_anc, lt);
+    if (r!=0) goto cleanup;
+    r = toku_lt_acquire_range_read_lock(lt, db, toku_txn_get_txnid(txn_anc->i->tokutxn),
+					&prevkey, &prevval,
+					found ? &key : toku_lt_infinity,
+					found ? &val : toku_lt_infinity);
+    if (r!=0) goto cleanup;
+    if (found) {
+	f(&key, &val, extra);
+    }
+    r = c_get_result;
+ cleanup:
+    toku_free(prevkey.data);
+    toku_free(prevval.data);
+    return r;
+}
 
 static int toku_c_close(DBC * c) {
     int r = toku_brt_cursor_close(c->i->c);
@@ -1986,7 +2030,7 @@ static int toku_db_cursor(DB * db, DB_TXN * txn, DBC ** c, u_int32_t flags, int
     result->c_close = locked_c_close;
     result->c_del = locked_c_del;
     result->c_count = locked_c_count;
-    result->c_getf_next = locked_c_getf_next; // Don't need loc
+    result->c_getf_next = locked_c_getf_next;
     MALLOC(result->i);
     assert(result->i);
     result->dbp = db;