recovery bind's to filenum's [t:1998]

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

newbrt/brt-internal.h

@@ -188,8 +188,10 @@ struct brt {
 
     unsigned int nodesize;
     unsigned int flags;
-    unsigned int did_set_flags;
-    unsigned int did_set_descriptor;
+    BOOL did_set_flags;
+    BOOL did_set_descriptor;
+    BOOL did_set_filenum;
+    FILENUM filenum;
     struct descriptor temp_descriptor;
     toku_dbt_upgradef dbt_userformat_upgrade;
     int (*compare_fun)(DB*,const DBT*,const DBT*);

newbrt/brt.c

@@ -2943,7 +2943,6 @@ brtheader_note_brt_open(BRT live) {
 }
 
 int toku_brt_open(BRT t, const char *fname, const char *fname_in_env, int is_create, int only_create, CACHETABLE cachetable, TOKUTXN txn, DB *db) {
-
     int r;
     BOOL txn_created = FALSE;
 
@@ -2965,7 +2964,7 @@ int toku_brt_open(BRT t, const char *fname, const char *fname_in_env, int is_cre
         BOOL did_create = FALSE;
         r = brt_open_file(t, fname, is_create, &fd, &did_create);
         if (r != 0) goto died00;
-        r=toku_cachetable_openfd(&t->cf, cachetable, fd, fname_in_env);
+        r=toku_cachetable_openfd_with_filenum(&t->cf, cachetable, fd, fname_in_env, t->did_set_filenum, t->filenum);
         if (r != 0) goto died00;
         if (did_create) {
             mode_t mode = S_IRWXU|S_IRWXG|S_IRWXO;
@@ -3098,6 +3097,12 @@ int toku_brt_set_dup_compare(BRT brt, int (*dup_compare)(DB *, const DBT*, const
     return 0;
 }
 
+int toku_brt_set_filenum(BRT brt, FILENUM filenum) {
+    brt->did_set_filenum = TRUE;
+    brt->filenum = filenum;
+    return 0;
+}
+
 int toku_brt_create_cachetable(CACHETABLE *ct, long cachesize, LSN initial_lsn, TOKULOGGER logger) {
     if (cachesize == 0)
         cachesize = 128*1024*1024;
@@ -3302,8 +3307,9 @@ int toku_brt_create(BRT *brt_ptr) {
     list_init(&brt->zombie_brt_link);
     list_init(&brt->cursors);
     brt->flags = 0;
-    brt->did_set_flags = 0;
-    brt->did_set_descriptor = 0;
+    brt->did_set_flags = FALSE;
+    brt->did_set_descriptor = FALSE;
+    brt->did_set_filenum = FALSE;
     brt->nodesize = BRT_DEFAULT_NODE_SIZE;
     brt->compare_fun = toku_default_compare_fun;
     brt->dup_compare = toku_default_compare_fun;

newbrt/brt.h

@@ -45,6 +45,8 @@ int toku_brt_get_nodesize(BRT, unsigned int *nodesize);
 int toku_brt_set_bt_compare(BRT, brt_compare_func);
 int toku_brt_set_dup_compare(BRT, brt_compare_func);
 
+int toku_brt_set_filenum(BRT brt, FILENUM filenum);
+
 int brt_set_cachetable(BRT, CACHETABLE);
 int toku_brt_open(BRT, const char *fname, const char *fname_in_env, int is_create, int only_create, CACHETABLE ct, TOKUTXN txn, DB *db);
 
@@ -54,28 +56,28 @@ int toku_brt_broadcast_commit_all (BRT brt);
 int toku_brt_lookup (BRT brt, DBT *k, DBT *v, BRT_GET_CALLBACK_FUNCTION getf, void *getf_v);
 
 // Effect: Insert a key and data pair into a brt
-// Returns 0 if successfull
+// Returns 0 if successful
 int toku_brt_insert (BRT brt, DBT *k, DBT *v, TOKUTXN txn);
 
 // Effect: Insert a key and data pair into a brt if the oplsn is newer than the brt lsn.  This function is called during recovery.
-// Returns 0 if successfull
+// Returns 0 if successful
 int toku_brt_maybe_insert (BRT brt, DBT *k, DBT *v, TOKUTXN txn, LSN oplsn);
 
 // Effect: Delete a key from a brt
-// Returns 0 if successfull
+// Returns 0 if successful
 int toku_brt_delete (BRT brt, DBT *k, TOKUTXN txn);
 
 // Effect: Delete a key from a brt if the oplsn is newer than the brt lsn.  This function is called during recovery.
-// Returns 0 if successfull
+// Returns 0 if successful
 int toku_brt_maybe_delete (BRT brt, DBT *k, TOKUTXN txn, LSN oplsn);
 
 // Effect: Delete a pair only if both k and v are equal according to the comparison function.
-// Returns 0 if successfull
+// Returns 0 if successful
 int toku_brt_delete_both (BRT brt, DBT *k, DBT *v, TOKUTXN txn); 
 
 // Effect: Delete a pair only if both k and v are equal according to the comparison function and the
 // oplsn is newer than the brt lsn.  This function is called by recovery.
-// Returns 0 if successfull
+// Returns 0 if successful
 int toku_brt_maybe_delete_both (BRT brt, DBT *k, DBT *v, TOKUTXN txn, LSN oplsn);
 
 int toku_brt_db_delay_closed (BRT brt, DB* db, int (*close_db)(DB*, u_int32_t), u_int32_t close_flags);

newbrt/cachetable.c

@@ -258,8 +258,7 @@ int toku_cachefile_of_filenum (CACHETABLE ct, FILENUM filenum, CACHEFILE *cf) {
 
 static FILENUM next_filenum_to_use={0};
 
-static void cachefile_init_filenum(CACHEFILE cf, int fd, const char *fname_relative_to_env, struct fileid fileid) \
-{
+static void cachefile_init_filenum(CACHEFILE cf, int fd, const char *fname_relative_to_env, struct fileid fileid) {
     cf->fd = fd;
     cf->fileid = fileid;
     cf->fname_relative_to_env = fname_relative_to_env ? toku_strdup(fname_relative_to_env) : 0;
@@ -275,6 +274,10 @@ cachefile_refup (CACHEFILE cf) {
 
 // If something goes wrong, close the fd.  After this, the caller shouldn't close the fd, but instead should close the cachefile.
 int toku_cachetable_openfd (CACHEFILE *cfptr, CACHETABLE ct, int fd, const char *fname_relative_to_env) {
+    return toku_cachetable_openfd_with_filenum(cfptr, ct, fd, fname_relative_to_env, FALSE, next_filenum_to_use);
+}
+
+int toku_cachetable_openfd_with_filenum (CACHEFILE *cfptr, CACHETABLE ct, int fd, const char *fname_relative_to_env, BOOL with_filenum, FILENUM filenum) {
     int r;
     CACHEFILE extant;
     struct fileid fileid;
@@ -305,19 +308,32 @@ int toku_cachetable_openfd (CACHEFILE *cfptr, CACHETABLE ct, int fd, const char
 	    goto exit;
 	}
     }
+
     //File is not open.  Make a new cachefile.
- try_again:
-    for (extant = ct->cachefiles; extant; extant=extant->next) {
-	if (next_filenum_to_use.fileid==extant->filenum.fileid) {
-	    next_filenum_to_use.fileid++;
-	    goto try_again;
-	}
+
+    if (with_filenum) {
+        // verify that filenum is not in use
+        for (extant = ct->cachefiles; extant; extant=extant->next) {
+            if (filenum.fileid == extant->filenum.fileid) {
+                r = EEXIST;
+                goto exit;
+            }
+        }
+    } else {
+        // find an unused fileid and use it
+    try_again:
+        for (extant = ct->cachefiles; extant; extant=extant->next) {
+            if (next_filenum_to_use.fileid==extant->filenum.fileid) {
+                next_filenum_to_use.fileid++;
+                goto try_again;
+            }
+        }
     }
     {
 	// create a new cachefile entry in the cachetable
         CACHEFILE XCALLOC(newcf);
         newcf->cachetable = ct;
-        newcf->filenum.fileid = next_filenum_to_use.fileid++;
+        newcf->filenum.fileid = with_filenum ? filenum.fileid : next_filenum_to_use.fileid++;
         cachefile_init_filenum(newcf, fd, fname_relative_to_env, fileid);
 	newcf->refcount = 1;
 	newcf->next = ct->cachefiles;

newbrt/cachetable.h

@@ -15,8 +15,6 @@
 extern int STRADDLE_HACK_INSIDE_CALLBACK;
 #endif
 
-
-
 // Maintain a cache mapping from cachekeys to values (void*)
 // Some of the keys can be pinned.  Don't pin too many or for too long.
 // If the cachetable is too full, it will call the flush_callback() function with the key, the value, and the otherargs
@@ -32,6 +30,8 @@ extern int STRADDLE_HACK_INSIDE_CALLBACK;
 
 typedef BLOCKNUM CACHEKEY;
 
+// cachetable operations
+
 int toku_create_cachetable(CACHETABLE */*result*/, long size_limit, LSN initial_lsn, TOKULOGGER);
 // Create a new cachetable.
 // Effects: a new cachetable is created and initialized.
@@ -49,10 +49,6 @@ int toku_cachefile_of_filenum (CACHETABLE t, FILENUM filenum, CACHEFILE *cf);
 int toku_cachetable_begin_checkpoint (CACHETABLE ct, TOKULOGGER);
 int toku_cachetable_end_checkpoint(CACHETABLE ct, TOKULOGGER logger, char **error_string, void (*testcallback_f)(void*),  void * testextra);
 
-// Does an fsync of a cachefile.
-// Handles the case where cf points to /dev/null
-int toku_cachefile_fsync(CACHEFILE cf);
-
 // Shuts down checkpoint thread
 // Requires no locks be held that are taken by the checkpoint function
 void toku_cachetable_minicron_shutdown(CACHETABLE ct);
@@ -67,13 +63,20 @@ int toku_cachetable_openf (CACHEFILE *,CACHETABLE, const char */*fname*/, const
 
 // Bind a file to a new cachefile object.
 int toku_cachetable_openfd (CACHEFILE *,CACHETABLE, int /*fd*/, const char *fname_relative_to_env /*(used for logging)*/);
+int toku_cachetable_openfd_with_filenum (CACHEFILE *,CACHETABLE, int /*fd*/, const char *fname_relative_to_env, BOOL with_filenum, FILENUM filenum);
 
 // Get access to the asynchronous work queue
 // Returns: a pointer to the work queue
 WORKQUEUE toku_cachetable_get_workqueue (CACHETABLE);
 
+// cachefile operations
+
 void toku_cachefile_get_workqueue_load (CACHEFILE, int *n_in_queue, int *n_threads);
 
+// Does an fsync of a cachefile.
+// Handles the case where cf points to /dev/null
+int toku_cachefile_fsync(CACHEFILE cf);
+
 // The flush callback is called when a key value pair is being written to storage and possibly removed from the cachetable.
 // When write_me is true, the value should be written to storage.
 // When keep_me is false, the value should be freed.