single-page btrees are working

src/stasis/CMakeLists.txt

@@ -37,6 +37,7 @@ ADD_LIBRARY(stasis crc32.c redblack.c lhtable.c rw.c doubleLinkedList.c
                    operations/linkedListNTA.c
                    operations/pageOrientedListNTA.c
                    operations/regions.c operations/lsmTree.c
+                   operations/bTree.c
                    io/rangeTracker.c io/memory.c io/file.c io/pfile.c
                    io/non_blocking.c io/debug.c io/handle.c
                    bufferManager/pageArray.c

src/stasis/Makefile.am

@@ -25,6 +25,7 @@ libstasis_la_SOURCES=crc32.c redblack.c lhtable.c rw.c doubleLinkedList.c common
 		   operations/lsnFreeSet.c \
 		   operations/group/logStructured.c \
 		   operations/segmentFile.c \
+		   operations/bTree.c \
 		   io/rangeTracker.c io/memory.c io/file.c io/pfile.c io/non_blocking.c \
 		   io/debug.c io/handle.c \
 		   bufferManager.c \

src/stasis/blobManager.c

@@ -96,6 +96,7 @@ page_impl stasis_page_blob_impl() {
       0, //getLength,
       0, //recordFirst,
       0, //recordNext,
+      0, //recordLast,
       stasis_page_not_supported, // is block supported
       0, //pageGenericBlockFirst,
       0, //pageGenericBlockNext,

src/stasis/page.c

@@ -231,6 +231,10 @@ recordid stasis_record_next(int xid, Page * p, recordid prev){
   return page_impls[p->pageType]
     .recordNext(xid,p,prev);
 }
+recordid stasis_record_last(int xid, Page * p) {
+  return page_impls[p->pageType]
+    .recordLast(xid,p);
+}
 recordid stasis_record_alloc_begin(int xid, Page * p, int size){
   return page_impls[p->pageType]
     .recordPreAlloc(xid,p,size);

src/stasis/page/fixed.c

@@ -140,6 +140,7 @@ page_impl fixedImpl() {
     fixedGetLength,
     fixedFirst,
     fixedNext,
+    fixedLast,
     notSupported, // notSupported,
     stasis_block_first_default_impl,
     stasis_block_next_default_impl,

src/stasis/page/segment.c

@@ -22,6 +22,7 @@ static page_impl pi =  {
     0,
     0,
     0,
+    0,
     notSupported, // is block supported
     stasis_block_first_default_impl,
     stasis_block_next_default_impl,
@@ -34,6 +35,8 @@ static page_impl pi =  {
     0, //XXX page_impl_dereference_identity,
     0,
     0,
+    0,
+    0,
     0
   };
   return pi;

src/stasis/page/slotted.c

@@ -341,6 +341,11 @@ static recordid slottedFirst(int xid, Page *p) {
   recordid rid = { p->id, -1, 0 };
   return slottedNext(xid, p, rid);
 }
+static recordid slottedLast(int xid, Page *p) {
+  recordid rid = {p->id, -1, 0 };
+  rid.slot = (*stasis_page_slotted_numslots_cptr(p)) - 1;
+  return rid;
+}
 
 static int notSupported(int xid, Page * p) { return 0; }
 
@@ -502,6 +507,7 @@ static void slottedPostRalloc(int xid, Page * page, recordid rid) {
 }
 
 static void slottedSpliceSlot(int xid, Page *p, slotid_t a, slotid_t b) {
+  if(a==b) { return; } // no-op
   assert(a < b);
   int16_t b_slot = *stasis_page_slotted_slot_cptr(p, b);
   int16_t b_slot_len = *stasis_page_slotted_slot_length_cptr(p, b);
@@ -510,7 +516,7 @@ static void slottedSpliceSlot(int xid, Page *p, slotid_t a, slotid_t b) {
     *stasis_page_slotted_slot_length_ptr(p, i+1) = *stasis_page_slotted_slot_length_cptr(p, i);
   }
   *stasis_page_slotted_slot_ptr(p, a) = b_slot;
-  *stasis_page_slotted_slot_length_ptr(p, b) = b_slot_len;
+  *stasis_page_slotted_slot_length_ptr(p, a) = b_slot_len;
 }
 
 static void slottedFree(int xid, Page * p, recordid rid) {
@@ -580,6 +586,7 @@ static page_impl pi =  {
     slottedGetLength,
     slottedFirst,
     slottedNext,
+    slottedLast,
     notSupported, // is block supported
     stasis_block_first_default_impl,
     stasis_block_next_default_impl,

src/stasis/transactional2.c

@@ -95,6 +95,7 @@ int Tinit() {
   TallocInit();
   TnaiveHashInit();
   LinearHashNTAInit();
+  BtreeInit();
   TlinkedListNTAInit();
   iterator_init();
   consumer_init();

stasis/constants.h

@@ -265,6 +265,9 @@ static const short SLOT_TYPE_LENGTHS[] = { -1, -1, sizeof(blob_record_t), -1};
 
 #define USER_DEFINED_ITERATOR 10
 
+#define BYTE_ARRAY_COMPARATOR 1
+
+#define MAX_COMPARATOR 10
 
 
 #endif

stasis/operations.h

@@ -153,6 +153,7 @@ typedef struct {
 #include "operations/linkedListNTA.h"
 #include "operations/pageOrientedListNTA.h"
 #include "operations/linearHashNTA.h"
+#include "operations/bTree.h"
 #include "operations/regions.h"
 #include "operations/lsmTree.h"
 #include "operations/lsnFreeSet.h"

stasis/page.h

@@ -544,6 +544,7 @@ void stasis_record_type_write(int xid, Page * p, recordid rid, int type);
 int stasis_record_length_read(int xid, Page *p, recordid rid);
 recordid stasis_record_first(int xid, Page * p);
 recordid stasis_record_next(int xid, Page * p, recordid prev);
+recordid stasis_record_last(int xid, Page * p);
 recordid stasis_record_alloc_begin(int xid, Page * p, int size);
 void stasis_record_alloc_done(int xid, Page * p, recordid rid);
 void stasis_record_splice(int xid, Page * p, slotid_t first, slotid_t second);
@@ -747,6 +748,15 @@ typedef struct page_impl {
   */
   recordid (*recordNext)(int xid, Page *p, recordid rid);
 
+ /**
+      This returns the last potentially occupied slot in the page (in slot order).
+      The slot is guaranteed to be valid if pageCompactSlotIDs was just called,
+      or if no records have been freed since the page was created or compacted.
+
+      @param p the page of interest.
+      @return a slot id greater than or equal to all valid slots on the page.
+  */
+  recordid (*recordLast)(int xid, Page *p);
   // -------- "Exotic" (optional) access methods.
 
   /*

stasis/page/fixed.h

@@ -25,6 +25,12 @@ static inline recordid fixedFirst(int xid, Page *p) {
   rid.size = *recordsize_ptr(p);
   return fixedNext(xid, p, rid);
 }
+static inline recordid fixedLast(int xid, Page *p) {
+  recordid rid = { p->id, -1, 0 };
+  rid.size = *recordsize_ptr(p);
+  rid.slot = -1+*recordcount_ptr(p);
+  return rid;
+}
 
 void fixedPageInit();
 void fixedPageDeinit();

test/stasis/check_bTree.c

@@ -43,7 +43,6 @@ terms specified in this license.
 #include "../check_includes.h"
 
 #include <stasis/transactional.h>
-#include <stasis/page.h>
 
 #include <assert.h>
 #include <limits.h>
@@ -51,240 +50,47 @@ terms specified in this license.
 #include <pthread.h>
 #define LOG_NAME   "check_bTree.log"
 #define NUM_ENTRIES 100000
-/** @test
-*/
-
-void testFunctions();
-int SimpleExample();
-
-START_TEST(bTreeTest)
-{
-  Tinit();
-  testFunctions();
-  SimpleExample();
-  // printf("\n end of practice run\n");
-   Tdeinit();
-} END_TEST
-
-/* This only takes in a page that is already initialized as a fixed page.
-   and has been initialized as a BTree Node, which involves placing an
-   index to tell how many entries are currently valid.
-   For now it will just return false if you try to add something to it
-   when it is already full.
-*/
-int insert(int xid, Page* p, int valueIn){
-  printf ("\nbegin insert\n");
-  int DEBUG = 0;
-  int DEBUGERROR = 1;
-  int DEBUGERROR2 = 0;
-  int DEBUGSTATEMENTS = 1;
-  int DEBUGMALLOC = 1; // 1 makes it malloc the space
-
-  //printf("\npage->id  = %d\n", p->id);
-
-  // make a copy of the rid - so we don't effect the caller's copy
-  recordid rid = { p->id, 0, sizeof(int)};
-
-
-  // if DEBUGERROR ==1 this causes a seg fault below!
-  if (DEBUGERROR) {printf("\n***page->id  = %lld\n", p->id);}
-  printf("\n***rid.page  = %lld\n\n", (long long)rid.page);
-
-
-  if(DEBUG) {printf("\nrid.page  = %lld\n", (long long)rid.page);}
-  if(DEBUG) {printf("\nrid.slot  = %lld\n", (long long)rid.slot);}
-
-
-  // Figure out how many entries are in the node
-  rid.slot = 0; // need to get the variable in slot 0
-
-  byte * countBuff;/// AHHH - i wasn't mallocing space for this!
-
-  if(DEBUGMALLOC ) {countBuff = (byte *) malloc (sizeof(int));}
-
-  if (DEBUGSTATEMENTS) {printf("\nDebug1\n");}
-
-  stasis_record_read(xid, p, rid, countBuff); // read the value of count from slot 0
-
-  if (DEBUGSTATEMENTS) {printf("\nDebug2\n");}
-
-  int * countInt = (int *) countBuff; // cast it to an int *
-
-  // set rid.slot to be the max slot entry used.
-  rid.slot = *countInt;
-
-  printf("\nrid2slot  = %d\n", rid.slot);
-
-  // *recordcount_ptr(p) = last accessible index on the page.
-  int max_index = stasis_fixed_records_per_page(rid.size); // rcs made this change.
-  //  int max_index = *recordcount_ptr(p);     // recordcount_ptr is the number of slots currently allocated on the page.
-                                               // but this code seems to do it's own allocation(?)
-
-
-  //THESE LINES WERE CAUSING A SEGFAULT! *******************************************
-  if (DEBUGERROR2) {printf("\nx  = %d\n", max_index);} // This causes a segfault after Debug2
-
-  // HACK! TO FIX THE ABOVE PROBLEM!
-  //  max_index = 1021;
-
-  // assert that max_index is greater than our record of how many
-  // entries we currently have on the page.
-  assert(max_index>rid.slot);
-
-
-  // check to see if we have room to add the entry.
-  if (rid.slot == max_index){
-    // we can't add any more entries to this node
-
-    return -1;
-  }
-
-
-
-  // the default location to put the new value is location 1.
-  int insertLocation = 1;
-
-  // Iterating DOWN through the slots. Stop when we reach 0.
-  // Will also break out of the while loop if we've found where
-  // to insert the new value.
-  while ( (rid.slot >= 1) && (insertLocation == 1)){
-
-    // TODO: JuSt haven't filled in the code here yet...
-    insertLocation =2;
-  }
-
-  // convert the input valueIn into a byte array
-  //  byte * valueInBuff = (byte *) & valueIn;
-
-  // get the rid ready to write to the insertLocation (determined above)
-  rid.slot = insertLocation;
-
-  // fixedWrite(p, rid, valueInBuff); // page/fixed.c:58: checkRid: Assertion `page->id == rid.page' failed.
-  printf("\n***page->id  = %lld\n", p->id);
-  printf("\n***rid.page  = %lld\n", (long long)rid.page);
-
-
-  return 0;
-
-}
-
-/* This takes a page that is already initialized and a
-   corresponding rid and initalizes the count value for
-   it to be a BTreeNode. Just puts the value 0 in the
-   first index of the page.
-*/
-void initializeNewBTreeNode(int xid, Page* p){
-
-  // need access to the first slot
-  recordid rid = { p->id, 0, sizeof(int)};
-
-  // prepare the value to go into the first slot
-  int countInt = 0;
-  byte * countBuff = (byte *) & countInt;
-
-  // write the count out
-  stasis_record_write(xid, p, rid, countBuff);
-  stasis_page_lsn_write(xid, p, 1);
-}
-void testFunctions(){
-  printf("testing functions");
-
-  // getting things ready
-  int xid = Tbegin();
-  pageid_t pageid1 = TfixedPageAlloc(xid, sizeof(int)); // this does the initialize
-  Page *  p1 = loadPage(xid, pageid1);
-
-  // calling functions
-  writelock(p1->rwlatch,0);
-  initializeNewBTreeNode(xid, p1);
-  insert(xid, p1, 3);
-  unlock(p1->rwlatch);
-
-  // cleaning up
-
-  releasePage(p1);
-  Tcommit(xid);
-}
-
-
-int SimpleExample(){
-
-  int DEBUGP = 0;
-  //  int DEBUGT = 0;
-  //  int DEBUGA = 0;
-  int xid = Tbegin();
-
-  /* Where to find stuff
-   * ****************************
-   *  TpageAlloc ->         stasis/src/stasis/operations/pageOperations.c
-   *  TfixedPageAlloc ->    stasis/src/stasis/operations/pageOperations.c
-   *  fixedPageInitailze -> stasis/src/stasis/page/fixed.c
-   */
-
-
-
-  pageid_t pageid1 = TfixedPageAlloc(xid, sizeof(int)); // this does the initialize
-
-  Page *  p1 = loadPage(xid, pageid1);
-  writelock(p1->rwlatch, 0);
-  if(DEBUGP) {  printf("\n**** page->id  = %lld\n", p1->id);}
-
-  /* check consistency between rid & page's values
-   * for number of slots and record size */
-  assert (p1->id == pageid1);
-
-
-  /* check to make sure page is recorded as a FIXED_PAGE */
-  assert( p1->pageType == FIXED_PAGE);
-
-  if (DEBUGP) { printf("\n%lld\n", (long long)pageid1); }
-  byte * b1 = (byte *) malloc (sizeof (int));
-  byte * b2 = (byte *) malloc (sizeof (int));
-  byte * b3 = (byte *) malloc (sizeof (int));
-  //  int x = *recordcount_ptr(p1);
-  int x = 42; //  rcs - recordcount_ptr is no longer exposed here...
-  int y = 0; //rid1.slot;
-  int z = 256;
-
-  b1 = (byte *) & x;
-  b2 = (byte *) & y;
-  b3 = (byte *) & z;
-
-  int * x1 = (int *) b1;
-
-
-  if (DEBUGP) { printf("\nx  = %d\n", x);}
-  if (DEBUGP) { printf("\nb1 = %d\n", *b1);}
-  if (DEBUGP) { printf("\nx1  = %d\n", *x1);}
-  if (DEBUGP) { printf("\ny  = %d\n", y);}
-  if (DEBUGP) { printf("\nb2  = %d\n", *b2);}
-  if (DEBUGP) { printf("\nz = %d\n", z);}
-  if (DEBUGP) { printf("\nb3 = %d\n", *b3);}
-
-  recordid rid1 = { pageid1, 0,sizeof(int)};
-
-  // @todo This is a messy way to do this...
-
-  stasis_record_write(xid, p1, rid1, b1);
-  stasis_page_lsn_write(xid, p1, 1);
-  stasis_record_read(xid, p1, rid1, b2);
-  if (DEBUGP) { printf("\nb2** = %d\n",*((int *) b2));}
-
-  //  initializeNewBTreeNode(p1, rid1);
-
-  unlock(p1->rwlatch);
-
-  releasePage(p1);
-  Tcommit(xid);
-
-
-  return 0;
-}
-
-/** @test
-*/
-
 #define NUM_ENTRIES_XACT 10000
+/** @test
+*/
+START_TEST(bTreeTest) {
+  Tinit();
+  int xid = Tbegin();
+
+  uint8_t key = 42;
+  uint64_t val = 424242424242;
+
+  recordid rid = TbtreeCreate(xid, BYTE_ARRAY_COMPARATOR);
+  TbtreeInsert(xid, rid, NULL, &key, sizeof(key), (byte*)&val, sizeof(val));
+
+  uint64_t* theval = 0;
+  size_t thevalsize = -1;
+  int found =TbtreeLookup(xid, rid, NULL, &key, sizeof(key), (byte**)&theval, &thevalsize);
+  assert(found);
+  assert(thevalsize == sizeof(val));
+  assert(*theval == val);
+  Tcommit(xid);
+
+  for(int j = 250; j > 0; j--) {
+    int i;
+    if(j % 2) {
+      i = j;
+    } else {
+      i = 200 - j;
+    }
+    void * scratch;
+    size_t scratchsize=-1;
+    assert(!TbtreeLookup(xid, rid, NULL, (byte*)&i, sizeof(i), (byte**)&scratch, &scratchsize));
+    TbtreeInsert(xid, rid, NULL, (byte*)&i, sizeof(i), (byte*)&i, sizeof(i));
+    assert(TbtreeLookup(xid, rid, NULL, (byte*)&i, sizeof(i), (byte**)&scratch, &scratchsize));
+    assert(*(typeof(&i))scratch == i);
+    assert(scratchsize == sizeof(i));
+    free(scratch);
+  }
+  Tdeinit();
+
+}END_TEST
+
 
 Suite * check_suite(void) {
   Suite *s = suite_create("bTree");