filehash/ffdb_pagepool.c
int
ffdb_pagepool_get_page (ffdb_pagepool_t* pgp, pgno_t* pageno,
			unsigned int flags, void** mem)
{
  int ret, found;
  ffdb_bkt_t* bp;
  struct _ffdb_hqh *head;

  /* Set memory pointer to NULL */
  *mem = 0;

  FFDB_LOCK (pgp->lock);
#ifdef _FFDB_STATISTICS
  pgp->pageget++;
#endif

  /**
   * Check flag for consistence
   */
  if (FFDB_FLAG_ISSET(flags, FFDB_PAGE_DIRTY)) {
    if (FFDB_FLAG_ISSET(pgp->fileflags, FFDB_RDONLY)) {
	fprintf (stderr, "ffdb_pagepool_get: DIRTY_PAGE flag cannot be used on readonly file.\n");
	errno = EINVAL;
	FFDB_UNLOCK (pgp->lock);
	return errno;
      }
  }
  
  /**
   * Handle the case of new page
   * No need to lock page here because each page is a new alloced page
   */
  if (FFDB_FLAG_ISSET(flags, FFDB_PAGE_NEW)) {
    if (FFDB_FLAG_ISSET(flags, FFDB_PAGE_CREATE)) {
	fprintf (stderr, "ffdb_pagepool_get: PAGE_NEW flag cannot be used with PAGE_CREATE flag\n");
	errno = EINVAL;
	FFDB_UNLOCK (pgp->lock);
	return errno;
    }
    fprintf (stderr, "calling new page i for page number %p\n", pageno);
    ret = _ffdb_pagepool_new_page_i (pgp, pageno, flags, mem);
    FFDB_UNLOCK (pgp->lock);    
    return ret;
  }

#if 0
  {
    ffdb_bkt_t* bk;

    head = &pgp->hqh[FFDB_HASHKEY(*pageno)];
    FFDB_CIRCLEQ_FOREACH(bk, head, hq) {
      if (bk->pgno == *pageno ) {
	fprintf (stderr, "Hash has this page %d alreay 0x%x\n", *pageno,
		 bk->page);
      }
    }

    FFDB_CIRCLEQ_FOREACH(bk, &pgp->lqh, lq) {
      if (bk->pgno == *pageno) {
	fprintf (stderr, "LRU has this page %d alreay 0x%x\n", *pageno,
		 bk->page);
      }
    } 
  }
#endif

  /**
   * Try to find a page from existing cache. This page has to be
   * not pinned by other threads
   */
  found = 0;
  head = &pgp->hqh[FFDB_HASHKEY(*pageno)];
  FFDB_CIRCLEQ_FOREACH(bp, head, hq) {
    if (bp->pgno == *pageno) {
      found = 1;
      break;
    }
  }

#ifdef _FFDB_STATISTICS  
  if (found)
    pgp->cachehit++;
  else
    pgp->cachemiss++;
#endif

  if (found) { /* Now I am still holding the lock */
#ifdef _FFDB_DEBUG
    fprintf (stderr, "Found page %d in pagepool at memory 0x%x\n", *pageno,
	     bp->page);
#endif
    /**
     * If I am the owner, the bp will be returned
     */
    if (FFDB_THREAD_SAME(bp->owner, FFDB_THREAD_ID)) {
      FFDB_FLAG_SET(bp->flags, FFDB_PAGE_PINNED);
      bp->ref++;
    }
    else {
      /**
       * A different thread try to access this page
       */
      if (bp->waiters > 0 || FFDB_FLAG_ISSET(bp->flags, FFDB_PAGE_PINNED)) {
	/* Create a waiter for this page */
	ffdb_bkt_waiter_t *waiter;
	ffdb_bkt_waiter_t* fw = 0;

	waiter = (ffdb_bkt_waiter_t *)malloc(sizeof(ffdb_bkt_waiter_t));
	if (!waiter) {
	  fprintf (stderr, "ffdb_pagepool_get: cannot allocate space for waiter object\n");
	  abort ();
	}
	waiter->wakeup = 0;
	waiter->bp = bp;
	FFDB_COND_INIT(waiter->cv);
	bp->waiters++;
	FFDB_CIRCLEQ_INSERT_HEAD(&bp->wqh, waiter, q);
	
	/* Now this waiter is waiting for the page */
	while (waiter->wakeup == 0) 
	  FFDB_COND_WAIT(waiter->cv, pgp->lock);
	
	/* Now waiter is done, we should have the page now */
	fw = FFDB_CIRCLEQ_LAST(&bp->wqh);
	if (fw != waiter) {
	  fprintf (stderr, "ffdb_pagepool_get_page: waiter wakeup with wrong waiter pointer\n");
	  abort ();
	}
	/* remove this from queue */
	FFDB_CIRCLEQ_REMOVE(&bp->wqh, waiter, q);

	bp->waiters--;

      	/* Destroy the conditional variable */
	FFDB_COND_FINI(waiter->cv);

	free (waiter);
      }
      /* Now I have grabed the page */
      bp->ref++;

      /* Now set owner of this page */
      bp->owner = FFDB_THREAD_ID;

      /* Now I get hold of this page */      
      FFDB_FLAG_SET(bp->flags, FFDB_PAGE_PINNED);
    }

    /* Change flags of this page since I own this page now */
    if (FFDB_FLAG_ISSET(flags, FFDB_PAGE_DIRTY) ||
	FFDB_FLAG_ISSET(flags, FFDB_PAGE_EDIT))
      FFDB_FLAG_SET(bp->flags, FFDB_PAGE_DIRTY);
  
    if (FFDB_FLAG_ISSET(flags, FFDB_PAGE_LOCKED))
      FFDB_FLAG_SET(bp->flags, FFDB_PAGE_LOCKED);
  
    *mem = bp->page; 
    
    /**
     * The following removal and reinsert must be done at the same 
     * time.
     * If the item is removed first, the lock is given up when 
     * a thread is waiting on the conditional variable.
     * Another thread come in to request the same page, it will not
     * find the page and will create a new page with the same page
     * number. One will have two entries in the hash and LRU with
     * the same page number
     */

    /* remove this page from hash and LRU */
    FFDB_CIRCLEQ_REMOVE(head, bp, hq);
    FFDB_CIRCLEQ_REMOVE(&pgp->lqh, bp, lq);
    /* We found this page in the cache so we have to 
     * move this page to the head of the hash chain and the tail of the
     * lru chain
     */
    FFDB_CIRCLEQ_INSERT_HEAD(head, bp, hq);
    FFDB_CIRCLEQ_INSERT_TAIL(&pgp->lqh, bp, lq);
#if 0
    fprintf (stderr, "Insert pageno %d\n", bp->pgno);
#endif

    FFDB_UNLOCK(pgp->lock);
    return 0;
  }
  else {
    /* We have not found a page with this page number, therefore
     * we have to create new pages using the pagenumber
     */
    /**
     * We cannot find this page with provided page number
     * if flag FFDB_PAGE_CREATE is set, we have to create this page
     */
    if (FFDB_FLAG_ISSET(flags, FFDB_PAGE_CREATE) &&
	*pageno >= pgp->npages ) {
      ret = _ffdb_pagepool_new_page_i (pgp, pageno, 
				       flags | FFDB_PAGE_REQUEST, mem);
    }
    else {
      ret = _ffdb_pagepool_load_new_page (pgp, *pageno, flags, mem);
    }


    FFDB_UNLOCK (pgp->lock);

    return ret;
  }
  /* Should never get here */
  fprintf (stderr, "ffdb_pagepool_get_page: should not get here\n");
  return 0;
}
/**
 * Get a cached page from the page poll
 * There will be no difference of treatment on the thread getting
 * this page, i.e. readonly or write
 * flags can be 0 or OR'ING the following values
 * FFDB_PAGE_CREATE if the specified page does not exist, create it.
 * FFDB_PAGE_DIRTY  this page will be modified before leaving the cache
 * FFDB_PAGE_LOCK   this page will stay in cache
 * number to the memory location of the pagno
 * FFDB_NEW create a new page in the file, and copy its page number into the
 * the memory localtion of the pageno.
 *
 * Since each page is locked by checking whether pinned flag is set,
 * so as long as pinned flag is changed, one is ok to modify other
 * attribute of the page
 */
Get a cached page from the page poll
There will be no difference of treatment on the thread getting
this page, i.e. readonly or write
flags can be 0 or OR'ING the following values
FFDB_PAGE_CREATE if the specified page does not exist, create it.
FFDB_PAGE_DIRTY  this page will be modified before leaving the cache
FFDB_PAGE_LOCK   this page will stay in cache
number to the memory location of the pagno
FFDB_NEW create a new page in the file, and copy its page number into the
the memory localtion of the pageno.

Since each page is locked by checking whether pinned flag is set,
so as long as pinned flag is changed, one is ok to modify other
attribute of the page
Get a cached page from the page poll There will be no difference of treatment on the thread getting this page i . e . readonly or write flags can be 0 or OR ' ING the following values FFDB_PAGE_CREATE if the specified page does not exist create it . FFDB_PAGE_DIRTY this page will be modified before leaving the cache FFDB_PAGE_LOCK this page will stay in cache number to the memory location of the pagno FFDB_NEW create a new page in the file and copy its page number into the the memory localtion of the pageno . Since each page is locked by checking whether pinned flag is set so as long as pinned flag is changed one is ok to modify other attribute of the page
int
ffdb_pagepool_get_page (ffdb_pagepool_t* pgp, pgno_t* pageno,
			unsigned int flags, void** mem)
{
  int ret, found;
  ffdb_bkt_t* bp;
  struct _ffdb_hqh *head;
  *mem = 0;
  FFDB_LOCK (pgp->lock);
#ifdef _FFDB_STATISTICS
  pgp->pageget++;
#endif
  if (FFDB_FLAG_ISSET(flags, FFDB_PAGE_DIRTY)) {
    if (FFDB_FLAG_ISSET(pgp->fileflags, FFDB_RDONLY)) {
	fprintf (stderr, "ffdb_pagepool_get: DIRTY_PAGE flag cannot be used on readonly file.\n");
	errno = EINVAL;
	FFDB_UNLOCK (pgp->lock);
	return errno;
      }
  }
  if (FFDB_FLAG_ISSET(flags, FFDB_PAGE_NEW)) {
    if (FFDB_FLAG_ISSET(flags, FFDB_PAGE_CREATE)) {
	fprintf (stderr, "ffdb_pagepool_get: PAGE_NEW flag cannot be used with PAGE_CREATE flag\n");
	errno = EINVAL;
	FFDB_UNLOCK (pgp->lock);
	return errno;
    }
    fprintf (stderr, "calling new page i for page number %p\n", pageno);
    ret = _ffdb_pagepool_new_page_i (pgp, pageno, flags, mem);
    FFDB_UNLOCK (pgp->lock);    
    return ret;
  }
#if 0
  {
    ffdb_bkt_t* bk;
    head = &pgp->hqh[FFDB_HASHKEY(*pageno)];
    FFDB_CIRCLEQ_FOREACH(bk, head, hq) {
      if (bk->pgno == *pageno ) {
	fprintf (stderr, "Hash has this page %d alreay 0x%x\n", *pageno,
		 bk->page);
      }
    }
    FFDB_CIRCLEQ_FOREACH(bk, &pgp->lqh, lq) {
      if (bk->pgno == *pageno) {
	fprintf (stderr, "LRU has this page %d alreay 0x%x\n", *pageno,
		 bk->page);
      }
    } 
  }
#endif
  found = 0;
  head = &pgp->hqh[FFDB_HASHKEY(*pageno)];
  FFDB_CIRCLEQ_FOREACH(bp, head, hq) {
    if (bp->pgno == *pageno) {
      found = 1;
      break;
    }
  }
#ifdef _FFDB_STATISTICS  
  if (found)
    pgp->cachehit++;
  else
    pgp->cachemiss++;
#endif
  if (found) { 
#ifdef _FFDB_DEBUG
    fprintf (stderr, "Found page %d in pagepool at memory 0x%x\n", *pageno,
	     bp->page);
#endif
    if (FFDB_THREAD_SAME(bp->owner, FFDB_THREAD_ID)) {
      FFDB_FLAG_SET(bp->flags, FFDB_PAGE_PINNED);
      bp->ref++;
    }
    else {
      if (bp->waiters > 0 || FFDB_FLAG_ISSET(bp->flags, FFDB_PAGE_PINNED)) {
	ffdb_bkt_waiter_t *waiter;
	ffdb_bkt_waiter_t* fw = 0;
	waiter = (ffdb_bkt_waiter_t *)malloc(sizeof(ffdb_bkt_waiter_t));
	if (!waiter) {
	  fprintf (stderr, "ffdb_pagepool_get: cannot allocate space for waiter object\n");
	  abort ();
	}
	waiter->wakeup = 0;
	waiter->bp = bp;
	FFDB_COND_INIT(waiter->cv);
	bp->waiters++;
	FFDB_CIRCLEQ_INSERT_HEAD(&bp->wqh, waiter, q);
	while (waiter->wakeup == 0) 
	  FFDB_COND_WAIT(waiter->cv, pgp->lock);
	fw = FFDB_CIRCLEQ_LAST(&bp->wqh);
	if (fw != waiter) {
	  fprintf (stderr, "ffdb_pagepool_get_page: waiter wakeup with wrong waiter pointer\n");
	  abort ();
	}
	FFDB_CIRCLEQ_REMOVE(&bp->wqh, waiter, q);
	bp->waiters--;
	FFDB_COND_FINI(waiter->cv);
	free (waiter);
      }
      bp->ref++;
      bp->owner = FFDB_THREAD_ID;
      FFDB_FLAG_SET(bp->flags, FFDB_PAGE_PINNED);
    }
    if (FFDB_FLAG_ISSET(flags, FFDB_PAGE_DIRTY) ||
	FFDB_FLAG_ISSET(flags, FFDB_PAGE_EDIT))
      FFDB_FLAG_SET(bp->flags, FFDB_PAGE_DIRTY);
    if (FFDB_FLAG_ISSET(flags, FFDB_PAGE_LOCKED))
      FFDB_FLAG_SET(bp->flags, FFDB_PAGE_LOCKED);
    *mem = bp->page; 
    FFDB_CIRCLEQ_REMOVE(head, bp, hq);
    FFDB_CIRCLEQ_REMOVE(&pgp->lqh, bp, lq);
    FFDB_CIRCLEQ_INSERT_HEAD(head, bp, hq);
    FFDB_CIRCLEQ_INSERT_TAIL(&pgp->lqh, bp, lq);
#if 0
    fprintf (stderr, "Insert pageno %d\n", bp->pgno);
#endif
    FFDB_UNLOCK(pgp->lock);
    return 0;
  }
  else {
    if (FFDB_FLAG_ISSET(flags, FFDB_PAGE_CREATE) &&
	*pageno >= pgp->npages ) {
      ret = _ffdb_pagepool_new_page_i (pgp, pageno, 
				       flags | FFDB_PAGE_REQUEST, mem);
    }
    else {
      ret = _ffdb_pagepool_load_new_page (pgp, *pageno, flags, mem);
    }
    FFDB_UNLOCK (pgp->lock);
    return ret;
  }
  fprintf (stderr, "ffdb_pagepool_get_page: should not get here\n");
  return 0;
}
int ffdb_pagepool_get_page ( ffdb_pagepool_t * pgp , pgno_t * pageno , unsigned int flags , void * * mem ) { int ret , found ; ffdb_bkt_t * bp ; struct _ffdb_hqh * head ; * mem = 0 ; FFDB_LOCK ( pgp -> lock ) ; #ifdef _FFDB_STATISTICS pgp -> pageget ++ ; #endif if ( FFDB_FLAG_ISSET ( flags , FFDB_PAGE_DIRTY ) ) { if ( FFDB_FLAG_ISSET ( pgp -> fileflags , FFDB_RDONLY ) ) { fprintf ( stderr , " \n " ) ; errno = EINVAL ; FFDB_UNLOCK ( pgp -> lock ) ; return errno ; } } if ( FFDB_FLAG_ISSET ( flags , FFDB_PAGE_NEW ) ) { if ( FFDB_FLAG_ISSET ( flags , FFDB_PAGE_CREATE ) ) { fprintf ( stderr , " \n " ) ; errno = EINVAL ; FFDB_UNLOCK ( pgp -> lock ) ; return errno ; } fprintf ( stderr , " \n " , pageno ) ; ret = _ffdb_pagepool_new_page_i ( pgp , pageno , flags , mem ) ; FFDB_UNLOCK ( pgp -> lock ) ; return ret ; } #if 0 { ffdb_bkt_t * bk ; head = & pgp -> hqh [ FFDB_HASHKEY ( * pageno ) ] ; FFDB_CIRCLEQ_FOREACH ( bk , head , hq ) { if ( bk -> pgno == * pageno ) { fprintf ( stderr , " \n " , * pageno , bk -> page ) ; } } FFDB_CIRCLEQ_FOREACH ( bk , & pgp -> lqh , lq ) { if ( bk -> pgno == * pageno ) { fprintf ( stderr , " \n " , * pageno , bk -> page ) ; } } } #endif found = 0 ; head = & pgp -> hqh [ FFDB_HASHKEY ( * pageno ) ] ; FFDB_CIRCLEQ_FOREACH ( bp , head , hq ) { if ( bp -> pgno == * pageno ) { found = 1 ; break ; } } #ifdef _FFDB_STATISTICS if ( found ) pgp -> cachehit ++ ; else pgp -> cachemiss ++ ; #endif if ( found ) { #ifdef _FFDB_DEBUG fprintf ( stderr , " \n " , * pageno , bp -> page ) ; #endif if ( FFDB_THREAD_SAME ( bp -> owner , FFDB_THREAD_ID ) ) { FFDB_FLAG_SET ( bp -> flags , FFDB_PAGE_PINNED ) ; bp -> ref ++ ; } else { if ( bp -> waiters > 0 || FFDB_FLAG_ISSET ( bp -> flags , FFDB_PAGE_PINNED ) ) { ffdb_bkt_waiter_t * waiter ; ffdb_bkt_waiter_t * fw = 0 ; waiter = ( ffdb_bkt_waiter_t * ) malloc ( sizeof ( ffdb_bkt_waiter_t ) ) ; if ( ! waiter ) { fprintf ( stderr , " \n " ) ; abort ( ) ; } waiter -> wakeup = 0 ; waiter -> bp = bp ; FFDB_COND_INIT ( waiter -> cv ) ; bp -> waiters ++ ; FFDB_CIRCLEQ_INSERT_HEAD ( & bp -> wqh , waiter , q ) ; while ( waiter -> wakeup == 0 ) FFDB_COND_WAIT ( waiter -> cv , pgp -> lock ) ; fw = FFDB_CIRCLEQ_LAST ( & bp -> wqh ) ; if ( fw != waiter ) { fprintf ( stderr , " \n " ) ; abort ( ) ; } FFDB_CIRCLEQ_REMOVE ( & bp -> wqh , waiter , q ) ; bp -> waiters -- ; FFDB_COND_FINI ( waiter -> cv ) ; free ( waiter ) ; } bp -> ref ++ ; bp -> owner = FFDB_THREAD_ID ; FFDB_FLAG_SET ( bp -> flags , FFDB_PAGE_PINNED ) ; } if ( FFDB_FLAG_ISSET ( flags , FFDB_PAGE_DIRTY ) || FFDB_FLAG_ISSET ( flags , FFDB_PAGE_EDIT ) ) FFDB_FLAG_SET ( bp -> flags , FFDB_PAGE_DIRTY ) ; if ( FFDB_FLAG_ISSET ( flags , FFDB_PAGE_LOCKED ) ) FFDB_FLAG_SET ( bp -> flags , FFDB_PAGE_LOCKED ) ; * mem = bp -> page ; FFDB_CIRCLEQ_REMOVE ( head , bp , hq ) ; FFDB_CIRCLEQ_REMOVE ( & pgp -> lqh , bp , lq ) ; FFDB_CIRCLEQ_INSERT_HEAD ( head , bp , hq ) ; FFDB_CIRCLEQ_INSERT_TAIL ( & pgp -> lqh , bp , lq ) ; #if 0 fprintf ( stderr , " \n " , bp -> pgno ) ; #endif FFDB_UNLOCK ( pgp -> lock ) ; return 0 ; } else { if ( FFDB_FLAG_ISSET ( flags , FFDB_PAGE_CREATE ) && * pageno >= pgp -> npages ) { ret = _ffdb_pagepool_new_page_i ( pgp , pageno , flags | FFDB_PAGE_REQUEST , mem ) ; } else { ret = _ffdb_pagepool_load_new_page ( pgp , * pageno , flags , mem ) ; } FFDB_UNLOCK ( pgp -> lock ) ; return ret ; } fprintf ( stderr , " \n " ) ; return 0 ; }
Get a cached page from the page poll
 There will be no difference of treatment on the thread getting
 this page, i.e.
Get a cached page from the page poll There will be no difference of treatment on the thread getting this page i . e .
/* Set memory pointer to NULL */ /**
   * Check flag for consistence
   */ /**
   * Handle the case of new page
   * No need to lock page here because each page is a new alloced page
   */ /**
   * Try to find a page from existing cache. This page has to be
   * not pinned by other threads
   */ /* Now I am still holding the lock */ /**
     * If I am the owner, the bp will be returned
     */ /**
       * A different thread try to access this page
       */ /* Create a waiter for this page */ /* Now this waiter is waiting for the page */ /* Now waiter is done, we should have the page now */ /* remove this from queue */ /* Destroy the conditional variable */ /* Now I have grabed the page */ /* Now set owner of this page */ /* Now I get hold of this page */ /* Change flags of this page since I own this page now */ /**
     * The following removal and reinsert must be done at the same 
     * time.
     * If the item is removed first, the lock is given up when 
     * a thread is waiting on the conditional variable.
     * Another thread come in to request the same page, it will not
     * find the page and will create a new page with the same page
     * number. One will have two entries in the hash and LRU with
     * the same page number
     */ /* remove this page from hash and LRU */ /* We found this page in the cache so we have to 
     * move this page to the head of the hash chain and the tail of the
     * lru chain
     */ /* We have not found a page with this page number, therefore
     * we have to create new pages using the pagenumber
     */ /**
     * We cannot find this page with provided page number
     * if flag FFDB_PAGE_CREATE is set, we have to create this page
     */ /* Should never get here */