Make fopen parse, and log only the fname relative to the env, not the whole name with the env

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

newbrt/brt.c

@@ -1499,7 +1499,7 @@ int brt_set_dup_compare(BRT brt, int (*dup_compare)(DB *, const DBT*, const DBT*
     return 0;
 }
 
-int brt_open(BRT t, const char *fname, const char *dbname, int is_create, int only_create, CACHETABLE cachetable, TOKUTXN txn) {
+int brt_open(BRT t, const char *fname, const char *fname_in_env, const char *dbname, int is_create, int only_create, CACHETABLE cachetable, TOKUTXN txn) {
 
     /* If dbname is NULL then we setup to hold a single tree.  Otherwise we setup an array. */
     int r;
@@ -1533,9 +1533,10 @@ int brt_open(BRT t, const char *fname, const char *dbname, int is_create, int on
 		t->database_name=0;
 		goto died0a;
 	    }
-	    tokulogger_log_fcreate(txn, fname, 0777);
+	    tokulogger_log_fcreate(txn, fname_in_env, 0777);
 	}
 	r=toku_cachetable_openfd(&t->cf, cachetable, fd);
+	tokulogger_log_fopen(txn, fname_in_env, toku_cachefile_filenum(t->cf));
     }
     if (r!=0) {
 	if (0) { died1: toku_cachefile_close(&t->cf); }
@@ -1673,6 +1674,7 @@ int brt_remove_subdb(BRT brt, const char *dbname, u_int32_t flags) {
     return r ? r : r2;
 }
 
+// This one has no env
 int open_brt (const char *fname, const char *dbname, int is_create, BRT *newbrt, int nodesize, CACHETABLE cachetable, TOKUTXN txn,
 	      int (*compare_fun)(DB*,const DBT*,const DBT*)) {
     BRT brt;
@@ -1685,7 +1687,7 @@ int open_brt (const char *fname, const char *dbname, int is_create, BRT *newbrt,
     brt_set_nodesize(brt, nodesize);
     brt_set_bt_compare(brt, compare_fun);
 
-    r = brt_open(brt, fname, dbname, is_create, only_create, cachetable, txn);
+    r = brt_open(brt, fname, fname, dbname, is_create, only_create, cachetable, txn);
     if (r != 0) {
         return r;
     }

newbrt/brt.h

@@ -24,7 +24,7 @@ int brt_set_nodesize(BRT, int nodesize);
 int brt_set_bt_compare(BRT, int (*bt_compare)(DB *, const DBT*, const DBT*));
 int brt_set_dup_compare(BRT, int (*dup_compare)(DB *, const DBT*, const DBT*));
 int brt_set_cachetable(BRT, CACHETABLE);
-int brt_open(BRT, const char *fname, const char *dbname, int is_create, int only_create, CACHETABLE ct, TOKUTXN txn);
+int brt_open(BRT, const char *fname, const char *fname_in_env, const char *dbname, int is_create, int only_create, CACHETABLE ct, TOKUTXN txn);
 int brt_remove_subdb(BRT brt, const char *dbname, u_int32_t flags);
 
 int brt_insert (BRT, DBT *, DBT *, DB*, TOKUTXN);

newbrt/cachetable.c

@@ -89,6 +89,7 @@ int toku_create_cachetable(CACHETABLE *result, long size_limit, LSN initial_lsn,
 int toku_cachetable_openfd (CACHEFILE *cf, CACHETABLE t, int fd) {
     int r;
     CACHEFILE extant;
+    FILENUM max_filenum_in_use={0};
     struct stat statbuf;
     struct fileid fileid;
     memset(&fileid, 0, sizeof(fileid));
@@ -97,6 +98,7 @@ int toku_cachetable_openfd (CACHEFILE *cf, CACHETABLE t, int fd) {
     fileid.st_dev = statbuf.st_dev;
     fileid.st_ino = statbuf.st_ino;
     for (extant = t->cachefiles; extant; extant=extant->next) {
+	if (max_filenum_in_use.fileid<extant->filenum.fileid) max_filenum_in_use=extant->filenum;
 	if (memcmp(&extant->fileid, &fileid, sizeof(fileid))==0) {
 	    r = close(fd);
             assert(r == 0);
@@ -107,6 +109,7 @@ int toku_cachetable_openfd (CACHEFILE *cf, CACHETABLE t, int fd) {
     }
     {
 	CACHEFILE MALLOC(newcf);
+	newcf->filenum.fileid = 1+max_filenum_in_use.fileid;
 	newcf->next = t->cachefiles;
 	newcf->refcount = 1;
 	newcf->fd = fd;

newbrt/log-internal.h

@@ -23,6 +23,7 @@ enum {
     LT_DELETE                   = 'D',
     LT_FCREATE                  = 'F',
     LT_INSERT_WITH_NO_OVERWRITE = 'I',
+    LT_FOPEN                    = 'O',
     LT_CHECKPOINT               = 'P',
     LT_BLOCK_RENAME             = 'R',
     LT_UNLINK                   = 'U'

newbrt/log.c

@@ -286,6 +286,32 @@ int tokulogger_log_fcreate (TOKUTXN txn, const char *fname, int mode) {
     return tokulogger_finish(txn->logger, &wbuf);
 }
 
+/* fopen isn't really an action.  It's just for bookkeeping.  We need to know the filename that goes with a filenum. */
+int tokulogger_log_fopen (TOKUTXN txn, const char * fname, FILENUM filenum) {
+    if (txn==0) return 0;
+    const int fnamelen = strlen(fname);
+    const int buflen = (+1 // log command
+			+8 // lsn
+			+8 // txnid
+			+4 // length of fname
+			+fnamelen
+			+4 // filenum len
+			+8 // crc & len
+			);
+    unsigned char buf[buflen];
+    struct wbuf wbuf;
+    wbuf_init (&wbuf, buf, buflen);
+    wbuf_char (&wbuf, LT_FOPEN);
+    wbuf_lsn    (&wbuf, txn->logger->lsn);
+    txn->logger->lsn.lsn++;
+    wbuf_txnid(&wbuf, txn->txnid64);
+    wbuf_bytes(&wbuf, fname, fnamelen);
+    wbuf_filenum(&wbuf, filenum);
+    return tokulogger_finish(txn->logger, &wbuf);
+    
+}
+
+
 int tokulogger_log_unlink (TOKUTXN txn, const char *fname) {
     if (txn==0) return 0;
     const int fnamelen = strlen(fname);

newbrt/log.h

@@ -18,6 +18,8 @@ int tokutxn_begin (TOKUTXN /*parent*/,TOKUTXN *, TXNID txnid64, TOKULOGGER logge
 
 int tokulogger_log_fcreate (TOKUTXN, const char */*fname*/, int /*mode*/);
 
+int tokulogger_log_fopen (TOKUTXN, const char * /*fname*/, FILENUM filenum);
+
 int tokulogger_log_unlink (TOKUTXN, const char */*fname*/);
 
 #endif

newbrt/logdump.c

@@ -70,6 +70,11 @@ void transcribe_mode (void) {
     printf(" mode=0%o", value);
 }
 
+void transcribe_filenum(void) {
+    u_int32_t value = get_uint32();
+    printf(" filenum=%d", value);
+}
+
 void transcribe_len (void) {
     u_int32_t l = get_uint32();
     printf(" len=%d", l);
@@ -77,6 +82,7 @@ void transcribe_len (void) {
     assert(l==actual_len);
 }
 
+
 void transcribe_key_or_data (char *what) {
     u_int32_t l = get_uint32();
     unsigned int i;
@@ -142,6 +148,17 @@ int main (int argc, char *argv[]) {
 	    printf("\n");
 	    break;
 
+	case LT_FOPEN:
+	    printf("FOPEN:");
+	    transcribe_lsn();
+	    transcribe_txnid();
+	    transcribe_key_or_data("fname");
+	    transcribe_filenum();
+	    transcribe_crc32();
+	    transcribe_len();
+	    printf("\n");
+	    break;
+	    
 	case LT_COMMIT:
 	    printf("COMMIT:");
 	    transcribe_lsn();
@@ -152,7 +169,7 @@ int main (int argc, char *argv[]) {
 	    break;
 
 	default:
-	    fprintf(stderr, "Huh?, found command %c\n", cmd);
+	    fprintf(stderr, "Huh?, found command '%c'\n", cmd);
 	    assert(0);
 	}
     }

src/ydb.c

@@ -650,7 +650,7 @@ int __toku_db_open(DB * db, DB_TXN * txn, const char *fname, const char *dbname,
     db->i->open_flags = flags;
     db->i->open_mode = mode;
 
-    r = brt_open(db->i->brt, db->i->full_fname, dbname,
+    r = brt_open(db->i->brt, db->i->full_fname, fname, dbname,
 		 flags & DB_CREATE, flags & DB_EXCL,
 		 db->dbenv->i->cachetable,
 		 txn ? txn->i->tokutxn : NULL_TXN);