page.c

分布式文件系统

/*
   Copyright (c) 2007, 2008 Z RESEARCH, Inc. <http://www.zresearch.com>
   This file is part of GlusterFS.

   GlusterFS is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published
   by the Free Software Foundation; either version 3 of the License,
   or (at your option) any later version.

   GlusterFS is distributed in the hope that it will be useful, but
   WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
   General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program.  If not, see
   <http://www.gnu.org/licenses/>.
*/

#ifndef _CONFIG_H
#define _CONFIG_H
#include "config.h"
#endif

#include "glusterfs.h"
#include "logging.h"
#include "dict.h"
#include "xlator.h"
#include "io-cache.h"
#include <assert.h>
#include <sys/time.h>

ioc_page_t *
ioc_page_get (ioc_inode_t *ioc_inode,
	      off_t offset)
{
  ioc_table_t *table = ioc_inode->table;
  ioc_page_t *page = NULL;
  off_t rounded_offset = floor (offset, table->page_size);
  int8_t found = 0;

  if (list_empty (&ioc_inode->pages)) {
    return NULL;
  }

  list_for_each_entry (page, &ioc_inode->pages, pages) {
    if (page->offset == rounded_offset) {
      found = 1;
      break;
    }
  }

  /* was previously returning ioc_inode itself.., 1st of its type and found one more downstairs :O */
  if (!found){
    page = NULL;
  } else {
    /* push the page to the end of the lru list */
    list_move_tail (&page->page_lru, &ioc_inode->page_lru);
  }

  return page;
}


/*
 * ioc_page_destroy -
 *
 * @page:
 *
 */
int64_t
ioc_page_destroy (ioc_page_t *page)
{
  int64_t page_size = 0;

  page_size = page->size;

  if (page->waitq) {
    /* frames waiting on this page, do not destroy this page */
    page_size = -1;
  } else {

    list_del (&page->pages);
    list_del (&page->page_lru);
    
    gf_log (page->inode->table->xl->name, GF_LOG_DEBUG,
	    "destroying page = %p, offset = %lld && inode = %p",
	    page, page->offset, page->inode);
    
    if (page->vector){
      dict_unref (page->ref);
      free (page->vector);
      page->vector = NULL;
    }
    
    page->inode = NULL;
    
  }

  if (page_size != -1) {
    pthread_mutex_destroy (&page->page_lock);
    free (page);
  }

  return page_size;
}

/*
 * ioc_prune - prune the cache. we have a limit to the number of pages we
 *             can have in-memory.
 *
 * @table: ioc_table_t of this translator
 *
 */
int32_t
ioc_prune (ioc_table_t *table)
{
  ioc_inode_t *curr = NULL, *next_ioc_inode = NULL;
  ioc_page_t *page = NULL, *next = NULL;
  int32_t ret = -1;
  int32_t index = 0;
  uint64_t size_to_prune = 0;
  uint64_t size_pruned = 0;

  ioc_table_lock (table);
  
  size_to_prune = table->cache_used - table->cache_size;
  /* take out the least recently used inode */
  for (index=0; index < table->max_pri; index++) {
    
    list_for_each_entry_safe (curr, next_ioc_inode, &table->inode_lru[index], inode_lru) {
      /* prune page-by-page for this inode, till we reach the equilibrium */
      ioc_inode_lock (curr);
      list_for_each_entry_safe (page, next, &curr->page_lru, page_lru){
	/* done with all pages, and not reached equilibrium yet??
	 * continue with next inode in lru_list */
	size_pruned += page->size;
	ret = ioc_page_destroy (page);
	
	if (ret != -1)
	  table->cache_used -= ret;
	
	gf_log (table->xl->name,
		GF_LOG_DEBUG,
		"index = %d && table->cache_used = %"PRIu64" && table->cache_size = %"PRIu64, 
		index, table->cache_used, table->cache_size);
	
	if (size_pruned >= size_to_prune)
	  break;
      }
      if (list_empty (&curr->pages)) {
	list_del_init (&curr->inode_lru);
      }

      ioc_inode_unlock (curr);
      
      if (size_pruned >= size_to_prune)
	break;
    }

    if (size_pruned >= size_to_prune)
      break;
  }
    
  ioc_table_unlock (table);

  return 0;
}

/*
 * ioc_page_create - create a new page. 
 *
 * @ioc_inode: 
 * @offset:
 *
 */
ioc_page_t *
ioc_page_create (ioc_inode_t *ioc_inode,
		 off_t offset)
{
  ioc_table_t *table = ioc_inode->table;
  ioc_page_t *page = NULL;
  off_t rounded_offset = floor (offset, table->page_size);
  ioc_page_t *newpage = calloc (1, sizeof (*newpage));
  
  if (ioc_inode)
    table = ioc_inode->table;
  else {
    return NULL;
  }
   
  newpage->offset = rounded_offset;
  newpage->inode = ioc_inode;
  pthread_mutex_init (&newpage->page_lock, NULL);

  list_add_tail (&newpage->page_lru, &ioc_inode->page_lru);
  list_add_tail (&newpage->pages, &ioc_inode->pages);

  page = newpage;

  gf_log ("io-cache", GF_LOG_DEBUG,
	  "returning new page %p", page);
  return page;
}

/* 
 * ioc_wait_on_page - pause a frame to wait till the arrival of a page. here we need to handle the case
 *                    when the frame who calls wait_on_page himself has caused page_fault 
 *
 * @page: page to wait on
 * @frame: call frame who is waiting on page
 *
 */
void
ioc_wait_on_page (ioc_page_t *page,
		  call_frame_t *frame,
		  off_t offset,
		  size_t size)
{
  ioc_waitq_t *waitq = calloc (1, sizeof (*waitq));
  ioc_local_t *local = frame->local;
  
  gf_log (frame->this->name, GF_LOG_DEBUG,
	  "frame(%p) waiting on page = %p, offset=%lld, size=%d",
	  frame, page, offset, size);

  waitq->data = frame;
  waitq->next = page->waitq;
  waitq->pending_offset = offset;
  waitq->pending_size = size;
  page->waitq = waitq;
  /* one frame can wait only once on a given page, 
   * local->wait_count is number of pages a frame is waiting on */
  ioc_local_lock (local);
  local->wait_count++;
  ioc_local_unlock (local);
}


/*
 * ioc_cache_still_valid - see if cached pages ioc_inode are still valid against given stbuf
 *
 * @ioc_inode:
 * @stbuf:
 *
 * assumes ioc_inode is locked
 */
int8_t
ioc_cache_still_valid (ioc_inode_t *ioc_inode,
		       struct stat *stbuf)
{
  int8_t cache_still_valid = 1;
  
#if 0
  if (!stbuf || (stbuf->st_mtime != ioc_inode->mtime) || 
      (stbuf->st_mtim.tv_nsec != ioc_inode->stbuf.st_mtim.tv_nsec))
#else
  if (!stbuf || (stbuf->st_mtime != ioc_inode->mtime))
#endif
    cache_still_valid = 0;

#if 0
  /* talk with [e-mail] to enable this section */
  if (!ioc_inode->mtime && stbuf) {
    cache_still_valid = 1;
    ioc_inode->mtime = stbuf->st_mtime;
  }
#endif

  return cache_still_valid;
}


static int32_t
ioc_fault_cbk (call_frame_t *frame,
	       void *cookie,
	       xlator_t *this,
	       int32_t op_ret,
	       int32_t op_errno,
	       struct iovec *vector,
	       int32_t count,
	       struct stat *stbuf)
{
  ioc_local_t *local = frame->local;
  off_t offset = local->pending_offset;
  ioc_inode_t *ioc_inode = local->inode;
  ioc_table_t *table = ioc_inode->table;
  ioc_page_t *page = NULL;
  off_t trav_offset = 0;
  size_t payload_size = 0;
  int32_t destroy_size = 0;

  trav_offset = offset;  
  payload_size = op_ret;

  ioc_inode_lock (ioc_inode);

  if (op_ret == -1 || (op_ret >= 0 && !ioc_cache_still_valid(ioc_inode, stbuf))) {
    gf_log (ioc_inode->table->xl->name, GF_LOG_DEBUG,
	    "cache for inode(%p) is invalid. flushing all pages", ioc_inode);
    destroy_size = __ioc_inode_flush (ioc_inode);
  }

  if (op_ret >= 0)
    ioc_inode->mtime = stbuf->st_mtime;

  gettimeofday (&ioc_inode->tv, NULL);

  if (op_ret < 0) {
    /* error, readv returned -1 */
    page = ioc_page_get (ioc_inode, offset);
    if (page)
      ioc_page_error (page, op_ret, op_errno);
  } else {
    gf_log (ioc_inode->table->xl->name, GF_LOG_DEBUG,
	    "op_ret = %d", op_ret);
    page = ioc_page_get (ioc_inode, offset);
    if (!page) {
      /* page was flushed */
      /* some serious bug ? */
      gf_log (this->name, GF_LOG_DEBUG,
	      "wasted copy: %lld[+%d] ioc_inode=%p", 
	      offset, table->page_size, ioc_inode);
    } else {
      if (page->vector) {
	dict_unref (page->ref);
	free (page->vector);
	page->vector = NULL;
      }

      /* keep a copy of the page for our cache */
      page->vector = iov_dup (vector, count);
      page->count = count;
      if (frame->root->rsp_refs) {
	page->ref = dict_ref (frame->root->rsp_refs);
      } else {
	/* TODO: we have got a response to our request and no data */
      }
      page->size = op_ret;
      if (page->waitq) {
	/* wake up all the frames waiting on this page, including 
	 * the frame which triggered fault */
	ioc_page_wakeup (page);
      }
    }
  }
  ioc_inode_unlock (ioc_inode);