change the cache table to act on the size of the objects rather than the...

change the cache table to act on the size of the objects rather than the number of objects stored in it
note that when the object size is 1, the cache table acts on the number of objects as before.

change the tree algorithms to update the object size in the cache table when the object is unpinned.



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

newbrt/brt-internal.h

@@ -86,6 +86,9 @@ void verify_counts(BRTNODE);
 int serialize_brt_header_to (int fd, struct brt_header *h);
 int deserialize_brtheader_from (int fd, diskoff off, struct brt_header **brth);
 
+/* return the size of a tree node */
+long brtnode_size (BRTNODE node);
+
 void brtnode_free (BRTNODE *node);
 
 //static inline int brtnode_n_hashtables(BRTNODE node) { if (node->height==0) return 1; else return node->u.n.n_children; }

newbrt/brt-test.c

@@ -1433,9 +1433,10 @@ void test_multiple_brt_cursor_walk(int n) {
 
     unlink(fname);
 
+    int nodesize = 1<<12;
     int h = log16(n);
-    int cachesize = 2 * h * ncursors;
-    r = brt_create_cachetable(&ct, cachesize);
+    int cachesize = 2 * h * ncursors * nodesize;
+    r = brt_create_cachetable_size(&ct, 127, cachesize);
     assert(r==0);
 
     r = open_brt(fname, 0, 1, &brt, 1<<12, ct, default_compare_fun);  

newbrt/brt.h

@@ -23,6 +23,11 @@ void brt_flush (BRT); /* fsync and clear the caches. */
 
 int brt_create_cachetable (CACHETABLE *t, int n_cachlines /* Pass 0 if you want the default. */);
 
+/* create and initialize a cache table
+   hashsize is the initialize size of the lookup table
+   cachesize is the upper limit on the size of the size of the values in the table */
+int brt_create_cachetable_size (CACHETABLE *t, int hashsize, long cachesize);
+
 extern int brt_debug_mode;
 int verify_brt (BRT brt);
 

newbrt/cachetable-test2.c

@@ -56,7 +56,7 @@ static void file_is_not_present(CACHEFILE cf) {
 }
 
 
-static void flush_forchain (CACHEFILE f __attribute__((__unused__)), CACHEKEY key, void *value, int write_me __attribute__((__unused__)), int keep_me __attribute__((__unused__))) {
+static void flush_forchain (CACHEFILE f __attribute__((__unused__)), CACHEKEY key, void *value, long size __attribute__((__unused__)), int write_me __attribute__((__unused__)), int keep_me __attribute__((__unused__))) {
     int *v = value;
     //cachetable_print_state(ct);
     //printf("Flush %lld %d\n", key, (int)value);
@@ -65,7 +65,7 @@ static void flush_forchain (CACHEFILE f __attribute__((__unused__)), CACHEKEY ke
     //print_ints();
 }
 
-static int fetch_forchain (CACHEFILE f __attribute__((__unused__)), CACHEKEY key, void**value, void*extraargs) {
+static int fetch_forchain (CACHEFILE f __attribute__((__unused__)), CACHEKEY key, void**value, long *sizep __attribute__((__unused__)), void*extraargs) {
     assert((long)extraargs==(long)key);
     *value = (void*)(long)key;
     return 0;

newbrt/cachetable.c

@@ -21,6 +21,7 @@ typedef struct ctpair *PAIR;
 struct ctpair {
     enum typ_tag tag;
     long long pinned;
+    long size;
     char     dirty;
     CACHEKEY key;
     void    *value;
@@ -39,6 +40,7 @@ struct cachetable {
     PAIR *table;
     PAIR  head,tail; // of LRU list.  head is the most recently used.  tail is least recently used.
     CACHEFILE cachefiles;
+    long size_current, size_limit;
 };
 
 struct fileid {
@@ -54,24 +56,11 @@ struct cachefile {
     struct fileid fileid;
 };
 
-void cachetable_print_state (CACHETABLE ct) {
-    int i;
-    for (i=0; i<ct->table_size; i++) {
-	PAIR p;
-	printf("t[%d]=", i);
-	for (p=ct->table[i]; p; p=p->hash_chain) {
-	    printf(" {%lld, %p, dirty=%d, pin=%lld}", p->key, p->cachefile, p->dirty, p->pinned);
-	}
-	printf("\n");
-    }
-}
-
-
-int create_cachetable (CACHETABLE *result, int n_entries) {
+int create_cachetable_size(CACHETABLE *result, int table_size, long size_limit) {
     TAGMALLOC(CACHETABLE, t);
     int i;
     t->n_in_table = 0;
-    t->table_size = n_entries;
+    t->table_size = table_size;
     MALLOC_N(t->table_size, t->table);
     assert(t->table);
     t->head = t->tail = 0;
@@ -79,6 +68,8 @@ int create_cachetable (CACHETABLE *result, int n_entries) {
 	t->table[i]=0;
     }
     t->cachefiles = 0;
+    t->size_current = 0;
+    t->size_limit = size_limit;
     *result = t;
     return 0;
 }
@@ -97,7 +88,8 @@ int cachetable_openf (CACHEFILE *cf, CACHETABLE t, const char *fname, int flags,
     fileid.st_ino = statbuf.st_ino;
     for (extant = t->cachefiles; extant; extant=extant->next) {
 	if (memcmp(&extant->fileid, &fileid, sizeof(fileid))==0) {
-	    close(fd);
+	    r = close(fd);
+            assert(r == 0);
 	    extant->refcount++;
 	    *cf = extant;
 	    return 0;
@@ -136,6 +128,8 @@ int cachefile_close (CACHEFILE *cfp) {
 	int r;
 	if ((r = cachefile_flush_and_remove(cf))) return r;
 	r = close(cf->fd);
+	assert(r == 0);
+        cf->fd = -1;
         cf->cachetable->cachefiles = remove_cf_from_list(cf, cf->cachetable->cachefiles);
 	toku_free(cf);
 	*cfp=0;
@@ -238,24 +232,25 @@ static void flush_and_remove (CACHETABLE t, PAIR remove_me, int write_me) {
     WHEN_TRACE_CT(printf("%s:%d CT flush_callback(%lld, %p, dirty=%d, 0)\n", __FILE__, __LINE__, remove_me->key, remove_me->value, remove_me->dirty && write_me)); 
     //printf("%s:%d TAG=%x p=%p\n", __FILE__, __LINE__, remove_me->tag, remove_me);
     //printf("%s:%d dirty=%d\n", __FILE__, __LINE__, remove_me->dirty);
-    remove_me->flush_callback(remove_me->cachefile, remove_me->key, remove_me->value, remove_me->dirty && write_me, 0);
+    remove_me->flush_callback(remove_me->cachefile, remove_me->key, remove_me->value, remove_me->size, remove_me->dirty && write_me, 0);
     t->n_in_table--;
     // Remove it from the hash chain.
     t->table[h] = remove_from_hash_chain (remove_me, t->table[h]);
+    t->size_current -= remove_me->size;
     toku_free(remove_me);
 }
 
 static void flush_and_keep (PAIR flush_me) {
     if (flush_me->dirty) {
 	WHEN_TRACE_CT(printf("%s:%d CT flush_callback(%lld, %p, dirty=1, 0)\n", __FILE__, __LINE__, flush_me->key, flush_me->value)); 
-        flush_me->flush_callback(flush_me->cachefile, flush_me->key, flush_me->value, 1, 1);
+        flush_me->flush_callback(flush_me->cachefile, flush_me->key, flush_me->value, flush_me->size, 1, 1);
 	flush_me->dirty=0;
     }
 }
 
-static int maybe_flush_some (CACHETABLE t) {
- again:
-    if (t->n_in_table>=t->table_size) {
+static int maybe_flush_some (CACHETABLE t, long size) {
+again:
+    if (size + t->size_current > t->size_limit) {
         /* Try to remove one. */
 	PAIR remove_me;
 	for (remove_me = t->tail; remove_me; remove_me = remove_me->prev) {
@@ -271,13 +266,14 @@ static int maybe_flush_some (CACHETABLE t) {
     return 0;
 }
 
-static int cachetable_insert_at(CACHEFILE cachefile, int h, CACHEKEY key, void *value,
+static int cachetable_insert_at(CACHEFILE cachefile, int h, CACHEKEY key, void *value, long size,
                                 cachetable_flush_func_t flush_callback,
                                 cachetable_fetch_func_t fetch_callback,
                                 void *extraargs, int dirty) {
     TAGMALLOC(PAIR, p);
     p->pinned = 1;
     p->dirty = dirty;
+    p->size = size;
     //printf("%s:%d p=%p dirty=%d\n", __FILE__, __LINE__, p, p->dirty);
     p->key = key;
     p->value = value;
@@ -286,14 +282,16 @@ static int cachetable_insert_at(CACHEFILE cachefile, int h, CACHEKEY key, void *
     p->flush_callback = flush_callback;
     p->fetch_callback = fetch_callback;
     p->extraargs = extraargs;
-    lru_add_to_list(cachefile->cachetable, p);
-    p->hash_chain = cachefile->cachetable->table[h];
-    cachefile->cachetable->table[h] = p;
-    cachefile->cachetable->n_in_table++;
+    CACHETABLE ct = cachefile->cachetable;
+    lru_add_to_list(ct, p);
+    p->hash_chain = ct->table[h];
+    ct->table[h] = p;
+    ct->n_in_table++;
+    ct->size_current += size;
     return 0;
 }
 
-int cachetable_put (CACHEFILE cachefile, CACHEKEY key, void*value,
+int cachetable_put_size(CACHEFILE cachefile, CACHEKEY key, void*value, long size,
                     cachetable_flush_func_t flush_callback, cachetable_fetch_func_t fetch_callback, void *extraargs) {
     int h = hashit(cachefile->cachetable, key);
     WHEN_TRACE_CT(printf("%s:%d CT cachetable_put(%lld)=%p\n", __FILE__, __LINE__, key, value));
@@ -309,12 +307,12 @@ int cachetable_put (CACHEFILE cachefile, CACHEKEY key, void*value,
 	    }
 	}
     }
-    if (maybe_flush_some(cachefile->cachetable)) 
+    if (maybe_flush_some(cachefile->cachetable, size)) 
         return -2;
-    return cachetable_insert_at(cachefile, h, key, value, flush_callback, fetch_callback, extraargs, 1);
+    return cachetable_insert_at(cachefile, h, key, value, size, flush_callback, fetch_callback, extraargs, 1);
 }
 
-int cachetable_get_and_pin (CACHEFILE cachefile, CACHEKEY key, void**value,
+int cachetable_get_and_pin_size (CACHEFILE cachefile, CACHEKEY key, void**value, long *sizep,
                             cachetable_flush_func_t flush_callback, cachetable_fetch_func_t fetch_callback, void *extraargs) {
     CACHETABLE t = cachefile->cachetable;
     int h = hashit(t,key);
@@ -322,20 +320,26 @@ int cachetable_get_and_pin (CACHEFILE cachefile, CACHEKEY key, void**value,
     for (p=t->table[h]; p; p=p->hash_chain) {
 	if (p->key==key && p->cachefile==cachefile) {
 	    *value = p->value;
+            *sizep = p->size;
 	    p->pinned++;
 	    lru_touch(t,p);
 	    WHEN_TRACE_CT(printf("%s:%d cachtable_get_and_pin(%lld)--> %p\n", __FILE__, __LINE__, key, *value));
 	    return 0;
 	}
     }
-    if (maybe_flush_some(t)) return -2;
+    if (maybe_flush_some(t, 1)) return -2;
     {
 	void *toku_value; 
+        long size = 1; // compat
 	int r;
 	WHEN_TRACE_CT(printf("%s:%d CT: fetch_callback(%lld...)\n", __FILE__, __LINE__, key));
-	if ((r=fetch_callback(cachefile, key, &toku_value,extraargs))) return r;
-	cachetable_insert_at(cachefile, h, key, toku_value, flush_callback, fetch_callback, extraargs, 0);
+	if ((r=fetch_callback(cachefile, key, &toku_value, &size, extraargs))) 
+            return r;
+	cachetable_insert_at(cachefile, h, key, toku_value, size, flush_callback, fetch_callback, extraargs, 0);
 	*value = toku_value;
+        if (sizep)
+            *sizep = size;
+        // maybe_flush_some(t, size);
     }
     WHEN_TRACE_CT(printf("%s:%d did fetch: cachtable_get_and_pin(%lld)--> %p\n", __FILE__, __LINE__, key, *value));
     return 0;
@@ -358,7 +362,7 @@ int cachetable_maybe_get_and_pin (CACHEFILE cachefile, CACHEKEY key, void**value
 }
 
 
-int cachetable_unpin (CACHEFILE cachefile, CACHEKEY key, int dirty) {
+int cachetable_unpin_size (CACHEFILE cachefile, CACHEKEY key, int dirty, long size) {
     CACHETABLE t = cachefile->cachetable;
     int h = hashit(t,key);
     PAIR p;
@@ -369,11 +373,15 @@ int cachetable_unpin (CACHEFILE cachefile, CACHEKEY key, int dirty) {
 	    assert(p->pinned>0);
 	    p->pinned--;
 	    p->dirty |= dirty;
+            if (size != 0) {
+                t->size_current -= p->size;
+                p->size = size;
+                t->size_current += p->size;
+            }
 	    WHEN_TRACE_CT(printf("[count=%lld]\n", p->pinned));
 	    return 0;
 	}
     }
-    printf("\n");
     return 0;
 }
 
@@ -496,16 +504,47 @@ int cachefile_fd (CACHEFILE cf) {
     return cf->fd;
 }
 
-/* debug function */
-int cachetable_get_state(CACHETABLE ct, CACHEKEY key, void **value_ptr,
-                         int *dirty_ptr, long long *pin_ptr) {
+/* debug functions */
+
+void cachetable_print_state (CACHETABLE ct) {
+    int i;
+    for (i=0; i<ct->table_size; i++) {
+	PAIR p = ct->table[i];
+        if (p != 0) {
+	printf("t[%d]=", i);
+	for (p=ct->table[i]; p; p=p->hash_chain) {
+	    printf(" {%lld, %p, dirty=%d, pin=%lld, size=%ld}", p->key, p->cachefile, p->dirty, p->pinned, p->size);
+	}
+	printf("\n");
+        }
+    }
+}
+
+void cachetable_get_state(CACHETABLE ct, int *num_entries_ptr, int *hash_size_ptr, long *size_current_ptr, long *size_limit_ptr) {
+    if (num_entries_ptr) 
+        *num_entries_ptr = ct->n_in_table;
+    if (hash_size_ptr)
+        *hash_size_ptr = ct->table_size;
+    if (size_current_ptr)
+        *size_current_ptr = ct->size_current;
+    if (size_limit_ptr)
+        *size_limit_ptr = ct->size_limit;
+}
+
+int cachetable_get_key_state(CACHETABLE ct, CACHEKEY key, void **value_ptr,
+                         int *dirty_ptr, long long *pin_ptr, long *size_ptr) {
     int h = hashit(ct, key);
     PAIR p;
     for (p = ct->table[h]; p; p = p->hash_chain) {
         if (p->key == key) {
+            if (value_ptr)
                 *value_ptr = p->value;
+            if (dirty_ptr)
                 *dirty_ptr = p->dirty;
+            if (pin_ptr)
                 *pin_ptr = p->pinned;
+            if (size_ptr)
+                *size_ptr = p->size;
             return 0;
         }
     }

newbrt/cachetable.h

 #ifndef CACHETABLE_H
 #define CACHETABLE_H
+
 #include <fcntl.h>
 
 /* Implement the cache table. */
@@ -18,28 +19,29 @@ typedef struct cachefile *CACHEFILE;
  * Note: The cachetable should use a common pool of memory, flushing things across cachetables.
  *  (The first implementation doesn't)
  * If you pin something twice, you must unpin it twice.
- * n_entries says how many items can fit into the cache table at a time.
+ * table_size is the initial size of the cache table hash table
+ * size limit is the upper bound of the sum of size of the entries in the cache table
  */
-int create_cachetable (CACHETABLE */*result*/, int /*n_entries*/);
+int create_cachetable_size(CACHETABLE */*result*/, int table_size, long size_limit);
 
 int cachetable_openf (CACHEFILE *,CACHETABLE, const char */*fname*/, int flags, mode_t mode);
 
-typedef void (*cachetable_flush_func_t)(CACHEFILE, CACHEKEY key, void*value, int write_me, int keep_me);
+typedef void (*cachetable_flush_func_t)(CACHEFILE, CACHEKEY key, void*value, long size, int write_me, int keep_me);
 
 /* If we are asked to fetch something, get it by calling this back. */
-typedef int (*cachetable_fetch_func_t)(CACHEFILE, CACHEKEY key, void**value,void*extraargs); 
+typedef int (*cachetable_fetch_func_t)(CACHEFILE, CACHEKEY key, void **value, long *sizep, void *extraargs); 
 
 /* Error if already present.  On success, pin the value. */
-int cachetable_put (CACHEFILE, CACHEKEY, void*/*value*/,
+int cachetable_put_size(CACHEFILE cf, CACHEKEY key, void* value, long size,
                     cachetable_flush_func_t flush_callback, cachetable_fetch_func_t fetch_callback, void *extraargs);
 
-int cachetable_get_and_pin (CACHEFILE, CACHEKEY, void**/*value*/,
+int cachetable_get_and_pin_size(CACHEFILE, CACHEKEY, void**/*value*/, long *sizep,
                             cachetable_flush_func_t flush_callback, cachetable_fetch_func_t fetch_callback, void *extraargs);
 
 /* If the the item is already in memory, then return 0 and store it in the void**.
  * If the item is not in memory, then return nonzero. */
 int cachetable_maybe_get_and_pin (CACHEFILE, CACHEKEY, void**);
-int cachetable_unpin (CACHEFILE, CACHEKEY, int dirty); /* Note whether it is dirty when we unpin it. */
+int cachetable_unpin_size (CACHEFILE, CACHEKEY, int dirty, long size); /* Note whether it is dirty when we unpin it. */
 int cachetable_remove (CACHEFILE, CACHEKEY, int /*write_me*/); /* Removing something already present is OK. */
 int cachetable_assert_all_unpinned (CACHETABLE);
 int cachefile_count_pinned (CACHEFILE, int /*printthem*/ );
@@ -58,6 +60,29 @@ int cachefile_fd (CACHEFILE);
 
 // Useful for debugging 
 void cachetable_print_state (CACHETABLE ct);
-int cachetable_get_state(CACHETABLE ct, CACHEKEY key, void **value_ptr,
-                         int *dirty_ptr, long long *pin_ptr);
+void cachetable_get_state(CACHETABLE ct, int *num_entries_ptr, int *hash_size_ptr, long *size_current_ptr, long *size_limit_ptr);
+int cachetable_get_key_state(CACHETABLE ct, CACHEKEY key, void **value_ptr,
+                         int *dirty_ptr, long long *pin_ptr, long *size_ptr);
+
+// Compat, will be removed
+static inline int create_cachetable (CACHETABLE *result, int table_size) {
+    return create_cachetable_size(result, table_size, table_size);
+}
+
+static inline int cachetable_put(CACHEFILE cf, CACHEKEY key, void* value,
+                                 cachetable_flush_func_t flush_callback, cachetable_fetch_func_t fetch_callback, void *extraargs) {
+    return cachetable_put_size(cf, key, value, 1, flush_callback, fetch_callback, extraargs);
+}
+
+static inline int cachetable_get_and_pin (CACHEFILE cf, CACHEKEY key, void **valuep, 
+                                          cachetable_flush_func_t flush_callback, cachetable_fetch_func_t fetch_callback, 
+                                          void *extraargs) {
+    long size;
+    return cachetable_get_and_pin_size(cf, key, valuep, &size, flush_callback, fetch_callback, extraargs);
+}
+
+static inline int cachetable_unpin(CACHEFILE cf, CACHEKEY key, int dirty) {
+    return cachetable_unpin_size(cf, key, dirty, 1);
+}
+
 #endif