From e53117a1a003c7f8fb2d7d91272a48d172c4ef7a Mon Sep 17 00:00:00 2001
From: Sears Russell <sears@cs.berkeley.edu>
Date: Mon, 26 Jul 2004 22:01:09 +0000
Subject: [PATCH] Implemented (but didn't test) iterator over the records in a
 page.

---
 src/lladd/page.c     | 78 +++++++++++++++++++++++++++++++++++++-------
 src/lladd/pageFile.c | 32 +++---------------
 src/lladd/pageFile.h |  4 +++
 3 files changed, 76 insertions(+), 38 deletions(-)

diff --git a/src/lladd/page.c b/src/lladd/page.c
index 6302a3c..27a519a 100644
--- a/src/lladd/page.c
+++ b/src/lladd/page.c
@@ -97,6 +97,9 @@ terms specified in this license.
 
 #include <lladd/constants.h>
 #include <assert.h>
+#include "blobManager.h"
+#include "pageFile.h"
+
 /* TODO:  Combine with buffer size... */
 static int nextPage = 0;
 
@@ -405,6 +408,69 @@ void pageInit() {
 	pthread_mutex_init(&pageAllocMutex, NULL);
 }
 
+typedef struct {
+  int page;
+  int slot;
+  /** If pageptr is not null, then it is used by the iterator methods.
+      Otherwise, they re-load the pages and obtain short latches for
+      each call. */
+  Page * pageptr;  
+} page_iterator_t;
+
+void pageIteratorInit(recordid rid, page_iterator_t * pit, Page * p) {
+  pit->page = rid.page;
+  pit->slot = rid.slot;
+  pit->pageptr = p;
+  assert((!p) || (p->id == pit->page));
+}
+
+int nextSlot(page_iterator_t * pit, recordid * rid) {
+  Page * p;
+  int numSlots;
+  int done = 0;
+  int ret;
+  if(pit->pageptr) {
+    p = pit->pageptr;
+  } else {
+    p = loadPage(pit->page);
+  }
+
+  numSlots = readNumSlots(p->memAddr);
+  while(pit->slot < numSlots && !done) {
+    
+    if(isValidSlot(p->memAddr, pit->slot)) {
+      done = 1;
+    } else {
+      pit->slot ++;
+    }
+
+  }
+  if(!done) {
+    ret = 0;
+  } else {
+    ret = 1;
+    rid->page = pit->page;
+    rid->slot = pit->slot;
+    rid->size = getSlotLength(p->memAddr, rid->slot);
+    if(rid->size >= PAGE_SIZE) {
+
+      if(rid->size == BLOB_SLOT) {
+	blob_record_t br;
+	pageReadRecord(-1, p, *rid, (byte*)&br);
+	rid->size = br.size;
+      }
+    }
+  }
+
+  if(!pit->pageptr) {
+    releasePage(p);
+  }
+
+  return ret;
+  
+}
+
+
 void pageCommit(int xid) {
   /*	 rmTouch(xid); */
 }
@@ -574,18 +640,8 @@ void pageReadRecord(int xid, Page * page, recordid rid, byte *buff) {
 
   slot_length = getSlotLength(page->memAddr, rid.slot);
 
-  /** @todo these assertions really *should* work... is slot length storage broken? */
-  
-  /*  assert((slot_length > 0) || (rid.size >= PAGE_SIZE)); */
-  /*  assert(slot_length); */
-  /*assert */
-  
   assert((rid.size == slot_length) || (slot_length >= PAGE_SIZE));
-  /*  if((rid.size == slot_length) || (slot_length >= PAGE_SIZE)) {
-    printf ("1");
-  } else {
-    printf ("2")
-    }*/
+
   memcpy(buff, recAddress,  rid.size);
   unlock(page->rwlatch);
   
diff --git a/src/lladd/pageFile.c b/src/lladd/pageFile.c
index ea7cf5f..f231053 100644
--- a/src/lladd/pageFile.c
+++ b/src/lladd/pageFile.c
@@ -16,33 +16,6 @@ static FILE * stable = NULL;
 /** Defined in bufferManager.c */
 extern pthread_mutex_t add_pending_mutex;
 
-
-
-/** 
-    This function blocks until there are no events pending for this page.
-
-    @see addPendingEvent(), removePendingEvent()
-*/
-
-
-/*void finalize(Page * p) {
-  pthread_mutex_lock(&(add_pending_mutex));
-  p->waiting++;
-
-  while(p->pending) {
-    DEBUG("A");
-    pthread_cond_wait(&(p->noMorePending), &(add_pending_mutex));
-  }
-  
-  assert(p->pending == 0);
-  assert(p->waiting == 1);
-  pthread_cond_broadcast(&addPendingOK);
-  pthread_mutex_unlock(&(add_pending_mutex)); 
-
-
-  return;
-  }*/
-
 void pageRead(Page *ret) {
   long fileSize;
 
@@ -170,4 +143,9 @@ void myFwrite(const void * dat, long size, FILE * f) {
   
 }
 
+long pageCount() {
+  long fileSize = myFseek(stable, 0, SEEK_END);
 
+  assert(! (fileSize % PAGE_SIZE));
+  return fileSize / PAGE_SIZE;
+}
diff --git a/src/lladd/pageFile.h b/src/lladd/pageFile.h
index 8c1adc5..8f6032e 100644
--- a/src/lladd/pageFile.h
+++ b/src/lladd/pageFile.h
@@ -38,6 +38,10 @@ void pageWrite(Page * dat);
    to call them.
 */
 void pageRead(Page * ret);
+/** 
+    @return the number of pages in the store file.
+*/
+long pageCount();
 
 void openPageFile();
 void closePageFile();