refs #113, add a callback function to the cursor to check for interrupts.

This allows callers to prevent a cursor that churns through many many committed
deletes to run amok.

buildheader/make_tdb.cc

@@ -207,6 +207,7 @@ enum {
         TOKUDB_BAD_CHECKSUM            = -100015,
         TOKUDB_HUGE_PAGES_ENABLED      = -100016,
         TOKUDB_OUT_OF_RANGE            = -100017,
+        TOKUDB_INTERRUPTED             = -100018,
         DONTUSE_I_JUST_PUT_THIS_HERE_SO_I_COULD_HAVE_A_COMMA_AFTER_EACH_ITEM
 };
 
@@ -359,6 +360,7 @@ static void print_defines (void) {
     dodefine(TOKUDB_BAD_CHECKSUM);
     dodefine(TOKUDB_HUGE_PAGES_ENABLED);
     dodefine(TOKUDB_OUT_OF_RANGE);
+    dodefine(TOKUDB_INTERRUPTED);
 
     /* LOADER flags */
     printf("/* LOADER flags */\n");
@@ -602,6 +604,7 @@ static void print_dbc_struct (void) {
 	"int (*c_getf_set_range)(DBC *, uint32_t, DBT *, YDB_CALLBACK_FUNCTION, void *)",
 	"int (*c_getf_set_range_reverse)(DBC *, uint32_t, DBT *, YDB_CALLBACK_FUNCTION, void *)",
 	"int (*c_set_bounds)(DBC*, const DBT*, const DBT*, bool pre_acquire, int out_of_range_error)",
+    "void (*c_set_check_interrupt_callback)(DBC*, bool (*)(void*), void *)",
 	"void (*c_remove_restriction)(DBC*)",
 	NULL};
     sort_and_dump_fields("dbc", false, extra);

ft/ft-internal.h

@@ -787,6 +787,8 @@ struct ft_cursor {
     int out_of_range_error;
     int direction;
     TOKUTXN ttxn;
+    FT_CHECK_INTERRUPT_CALLBACK interrupt_cb;
+    void *interrupt_cb_extra;
 };
 
 //

ft/ft-ops.cc

@@ -4039,6 +4039,12 @@ void toku_ft_cursor_remove_restriction(FT_CURSOR ftcursor) {
     ftcursor->direction = 0;
 }
 
+void toku_ft_cursor_set_check_interrupt_cb(FT_CURSOR ftcursor, FT_CHECK_INTERRUPT_CALLBACK cb, void *extra) {
+    ftcursor->interrupt_cb = cb;
+    ftcursor->interrupt_cb_extra = extra;
+}
+
+
 void
 toku_ft_cursor_set_temporary(FT_CURSOR ftcursor) {
     ftcursor->is_temporary = true;
@@ -4812,12 +4818,20 @@ ok: ;
             switch (search->direction) {
             case FT_SEARCH_LEFT:
                 idx++;
-                if (idx >= bn->data_buffer.omt_size())
+                if (idx >= bn->data_buffer.omt_size()) {
+                    if (ftcursor->interrupt_cb && ftcursor->interrupt_cb(ftcursor->interrupt_cb_extra)) {
+                        return TOKUDB_INTERRUPTED;
+                    }
                     return DB_NOTFOUND;
+                }
                 break;
             case FT_SEARCH_RIGHT:
-                if (idx == 0)
+                if (idx == 0) {
+                    if (ftcursor->interrupt_cb && ftcursor->interrupt_cb(ftcursor->interrupt_cb_extra)) {
+                        return TOKUDB_INTERRUPTED;
+                    }
                     return DB_NOTFOUND;
+                }
                 idx--;
                 break;
             default:

ft/ft-ops.h

@@ -114,6 +114,8 @@ PATENT RIGHTS GRANT:
 // When lock_only is true, the callback only does optional lock tree locking.
 typedef int(*FT_GET_CALLBACK_FUNCTION)(ITEMLEN keylen, bytevec key, ITEMLEN vallen, bytevec val, void *extra, bool lock_only);
 
+typedef bool(*FT_CHECK_INTERRUPT_CALLBACK)(void* extra);
+
 int toku_open_ft_handle (const char *fname, int is_create, FT_HANDLE *, int nodesize, int basementnodesize, enum toku_compression_method compression_method, CACHETABLE, TOKUTXN, int(*)(DB *,const DBT*,const DBT*)) __attribute__ ((warn_unused_result));
 
 // effect: changes the descriptor for the ft of the given handle.
@@ -258,6 +260,7 @@ void toku_ft_cursor_set_leaf_mode(FT_CURSOR);
 // the cursor duing a one query.
 void toku_ft_cursor_set_temporary(FT_CURSOR);
 void toku_ft_cursor_remove_restriction(FT_CURSOR);
+void toku_ft_cursor_set_check_interrupt_cb(FT_CURSOR ftcursor, FT_CHECK_INTERRUPT_CALLBACK cb, void *extra);
 int toku_ft_cursor_is_leaf_mode(FT_CURSOR);
 void toku_ft_cursor_set_range_lock(FT_CURSOR, const DBT *, const DBT *, bool, bool, int);
 

src/tests/CMakeLists.txt

@@ -257,6 +257,7 @@ if(BUILD_TESTING OR BUILD_SRC_TESTS)
       test_bulk_fetch
       test_compression_methods
       test_cmp_descriptor
+      test_cursor_interrupt
       test_cursor_with_read_txn
       test_db_change_pagesize
       test_db_change_xxx

src/tests/test_cursor_interrupt.cc

+/* -*- mode: C; c-basic-offset: 4; indent-tabs-mode: nil -*- */
+// vim: expandtab:ts=8:sw=4:softtabstop=4:
+/*
+COPYING CONDITIONS NOTICE:
+
+  This program is free software; you can redistribute it and/or modify
+  it under the terms of version 2 of the GNU General Public License as
+  published by the Free Software Foundation, and provided that the
+  following conditions are met:
+
+      * Redistributions of source code must retain this COPYING
+        CONDITIONS NOTICE, the COPYRIGHT NOTICE (below), the
+        DISCLAIMER (below), the UNIVERSITY PATENT NOTICE (below), the
+        PATENT MARKING NOTICE (below), and the PATENT RIGHTS
+        GRANT (below).
+
+      * Redistributions in binary form must reproduce this COPYING
+        CONDITIONS NOTICE, the COPYRIGHT NOTICE (below), the
+        DISCLAIMER (below), the UNIVERSITY PATENT NOTICE (below), the
+        PATENT MARKING NOTICE (below), and the PATENT RIGHTS
+        GRANT (below) in the documentation and/or other materials
+        provided with the distribution.
+
+  You should have received a copy of the GNU General Public License
+  along with this program; if not, write to the Free Software
+  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
+  02110-1301, USA.
+
+COPYRIGHT NOTICE:
+
+  TokuDB, Tokutek Fractal Tree Indexing Library.
+  Copyright (C) 2007-2013 Tokutek, Inc.
+
+DISCLAIMER:
+
+  This program is distributed in the hope that it will be useful, but
+  WITHOUT ANY WARRANTY; without even the implied warranty of
+  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+  General Public License for more details.
+
+UNIVERSITY PATENT NOTICE:
+
+  The technology is licensed by the Massachusetts Institute of
+  Technology, Rutgers State University of New Jersey, and the Research
+  Foundation of State University of New York at Stony Brook under
+  United States of America Serial No. 11/760379 and to the patents
+  and/or patent applications resulting from it.
+
+PATENT MARKING NOTICE:
+
+  This software is covered by US Patent No. 8,185,551.
+  This software is covered by US Patent No. 8,489,638.
+
+PATENT RIGHTS GRANT:
+
+  "THIS IMPLEMENTATION" means the copyrightable works distributed by
+  Tokutek as part of the Fractal Tree project.
+
+  "PATENT CLAIMS" means the claims of patents that are owned or
+  licensable by Tokutek, both currently or in the future; and that in
+  the absence of this license would be infringed by THIS
+  IMPLEMENTATION or by using or running THIS IMPLEMENTATION.
+
+  "PATENT CHALLENGE" shall mean a challenge to the validity,
+  patentability, enforceability and/or non-infringement of any of the
+  PATENT CLAIMS or otherwise opposing any of the PATENT CLAIMS.
+
+  Tokutek hereby grants to you, for the term and geographical scope of
+  the PATENT CLAIMS, a non-exclusive, no-charge, royalty-free,
+  irrevocable (except as stated in this section) patent license to
+  make, have made, use, offer to sell, sell, import, transfer, and
+  otherwise run, modify, and propagate the contents of THIS
+  IMPLEMENTATION, where such license applies only to the PATENT
+  CLAIMS.  This grant does not include claims that would be infringed
+  only as a consequence of further modifications of THIS
+  IMPLEMENTATION.  If you or your agent or licensee institute or order
+  or agree to the institution of patent litigation against any entity
+  (including a cross-claim or counterclaim in a lawsuit) alleging that
+  THIS IMPLEMENTATION constitutes direct or contributory patent
+  infringement, or inducement of patent infringement, then any rights
+  granted to you under this License shall terminate as of the date
+  such litigation is filed.  If you or your agent or exclusive
+  licensee institute or order or agree to the institution of a PATENT
+  CHALLENGE, then Tokutek may terminate any rights granted to you
+  under this License.
+*/
+
+#ident "Copyright (c) 2007-2013 Tokutek Inc.  All rights reserved."
+#include "test.h"
+
+
+#include <stdio.h>
+
+#include <db.h>
+
+
+int num_interrupts_called;
+static bool interrupt(void* extra UU()) {
+    num_interrupts_called++;
+    return false;
+}
+
+static bool interrupt_true(void* extra UU()) {
+    num_interrupts_called++;
+    return true;
+}
+
+
+int
+test_main(int argc, char *const argv[]) {
+    parse_args(argc, argv);
+    DB_ENV *env;
+    DB *db;
+    int r;
+
+    toku_os_recursive_delete(TOKU_TEST_FILENAME);
+    r=toku_os_mkdir(TOKU_TEST_FILENAME, S_IRWXU+S_IRWXG+S_IRWXO);       assert(r==0);
+    r=db_env_create(&env, 0); assert(r==0);
+    r=env->open(env, TOKU_TEST_FILENAME, DB_INIT_LOCK | DB_INIT_LOG | DB_INIT_MPOOL | DB_INIT_TXN | DB_CREATE | DB_PRIVATE | DB_INIT_LOG, S_IRWXU+S_IRWXG+S_IRWXO); assert(r==0);
+
+    r = db_create(&db, env, 0);
+    CKERR(r);
+    r = db->set_readpagesize(db, 1024);
+    CKERR(r);
+    r = db->set_pagesize(db, 1024*10);
+    CKERR(r);
+
+    const char * const fname = "test.change_pagesize";
+    r = db->open(db, NULL, fname, "main", DB_BTREE, DB_CREATE, 0666);
+    CKERR(r);
+    DB_TXN* txn;
+    r = env->txn_begin(env, 0, &txn, 0);
+    CKERR(r);
+    for (uint64_t i = 0; i < 10000; i++) {
+        DBT key, val;
+        uint64_t k = i;
+        uint64_t v = i;
+        dbt_init(&key, &k, sizeof k);
+        dbt_init(&val, &v, sizeof v);
+        db->put(db, txn, &key, &val, DB_PRELOCKED_WRITE); // adding DB_PRELOCKED_WRITE just to make the test go faster
+    }
+    r = txn->commit(txn, 0);
+    CKERR(r);
+
+    // create a snapshot txn so that when we delete the elements
+    // we just inserted, that they do not get garbage collected away
+    DB_TXN* snapshot_txn;
+    r = env->txn_begin(env, 0, &snapshot_txn, DB_TXN_SNAPSHOT);
+    CKERR(r);
+
+    DB_TXN* delete_txn;
+    r = env->txn_begin(env, 0, &delete_txn, DB_TXN_SNAPSHOT);
+    CKERR(r);
+
+    for (uint64_t i = 0; i < 10000; i++) {
+        DBT key;
+        uint64_t k = i;
+        dbt_init(&key, &k, sizeof k);
+        db->del(db, delete_txn, &key, DB_PRELOCKED_WRITE | DB_DELETE_ANY); // adding DB_PRELOCKED_WRITE just to make the test go faster
+    }
+    r = delete_txn->commit(delete_txn, 0);
+    CKERR(r);
+
+    // to make more than one basement node in the dictionary's leaf nodes
+    r = env->txn_checkpoint(env, 0, 0, 0);
+    CKERR(r);
+
+    // create a txn that should see an empty dictionary
+    DB_TXN* test_txn;
+    r = env->txn_begin(env, 0, &test_txn, DB_TXN_SNAPSHOT);
+    CKERR(r);
+    DBC* cursor = NULL;
+    r = db->cursor(db, test_txn, &cursor, 0);
+    cursor->c_set_check_interrupt_callback(cursor, interrupt, NULL);
+    DBT key, val;
+    r = cursor->c_get(cursor, &key, &val, DB_NEXT);
+    CKERR2(r, DB_NOTFOUND);
+    assert(num_interrupts_called > 1);
+    num_interrupts_called = 0;
+    cursor->c_set_check_interrupt_callback(cursor, interrupt_true, NULL);
+    r = cursor->c_get(cursor, &key, &val, DB_NEXT);
+    CKERR2(r, TOKUDB_INTERRUPTED);
+    assert(num_interrupts_called == 1);
+
+    r = cursor->c_close(cursor);
+    CKERR(r);    
+    r = test_txn->commit(test_txn, 0);
+    CKERR(r);
+
+    
+    r = snapshot_txn->commit(snapshot_txn, 0);
+    CKERR(r);
+
+
+    r = db->close(db, 0);
+    CKERR(r);
+
+    r = env->close(env, 0);
+    assert(r == 0);
+
+    return 0;
+}

src/ydb_cursor.cc

@@ -740,6 +740,11 @@ c_remove_restriction(DBC *dbc) {
     toku_ft_cursor_remove_restriction(dbc_struct_i(dbc)->c);
 }
 
+static void
+c_set_check_interrupt_callback(DBC* dbc, bool (*interrupt_callback)(void*), void *extra) {
+    toku_ft_cursor_set_check_interrupt_cb(dbc_struct_i(dbc)->c, interrupt_callback, extra);
+}
+
 int
 toku_c_get(DBC* c, DBT* key, DBT* val, uint32_t flag) {
     //This function exists for legacy (test compatibility) purposes/parity with bdb.
@@ -833,6 +838,7 @@ toku_db_cursor_internal(DB * db, DB_TXN * txn, DBC ** c, uint32_t flags, int is_
     SCRS(c_getf_set_range_reverse);
     SCRS(c_set_bounds);
     SCRS(c_remove_restriction);
+    SCRS(c_set_check_interrupt_callback);
 #undef SCRS
 
     result->c_get = toku_c_get;