io-cache.c

分布式文件系统

/*
 * ioc_close - close fop for io cache
 * 
 * @frame:
 * @this:
 * @fd:
 *
 */
int32_t
ioc_close (call_frame_t *frame,
	   xlator_t *this,
	   fd_t *fd)
{
  STACK_WIND (frame,
	      ioc_close_cbk,
	      FIRST_CHILD(this),
	      FIRST_CHILD(this)->fops->close,
	      fd);
  return 0;
}

/* 
 * ioc_readv_disabled_cbk 
 * @frame:
 * @cookie:
 * @this:
 * @op_ret:
 * @op_errno:
 * @vector:
 * @count:
 *
 */ 
int32_t
ioc_readv_disabled_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)
{
  GF_ERROR_IF_NULL (this);
  GF_ERROR_IF_NULL (vector);

  STACK_UNWIND (frame, op_ret, op_errno, vector, count, stbuf);
  return 0;
}


int32_t
ioc_need_prune (ioc_table_t *table)
{
  int64_t cache_difference = 0;
  
  ioc_table_lock (table);
  cache_difference = table->cache_used - table->cache_size;
  ioc_table_unlock (table);

  if (cache_difference > 0)
    return ((cache_difference*100/table->cache_size) >= 20);
  else 
    return 0;
}

/*
 * dispatch_requests -
 * 
 * @frame:
 * @inode:
 *
 * 
 */
static void
dispatch_requests (call_frame_t *frame,
		   ioc_inode_t *ioc_inode,
		   fd_t *fd,
		   off_t offset,
		   size_t size)
{
  ioc_local_t *local = frame->local;
  ioc_table_t *table = ioc_inode->table;
  off_t rounded_offset = 0;
  off_t rounded_end = 0;
  off_t trav_offset = 0;
  ioc_page_t *trav = NULL;
  int32_t fault = 0;
  int8_t might_need_validate = 0;  /* if a page exists, do we need to validate it? */
  int8_t need_validate = 0;

  rounded_offset = floor (offset, table->page_size);
  rounded_end = roof (offset + size, table->page_size);
  trav_offset = rounded_offset;

  /* once a frame does read, it should be waiting on something */
  local->wait_count++;

  /* Requested region can fall in three different pages,
   * 1. Ready - region is already in cache, we just have to serve it.
   * 2. In-transit - page fault has been generated on this page, we need
   *    to wait till the page is ready
   * 3. Fault - page is not in cache, we have to generate a page fault
   */

  might_need_validate = ioc_inode_need_revalidate (ioc_inode);

  while (trav_offset < rounded_end) {
    size_t trav_size = 0;
    off_t local_offset = 0;

    ioc_inode_lock (ioc_inode);
    /* look for requested region in the cache */
    trav = ioc_page_get (ioc_inode, trav_offset);

    local_offset = max (trav_offset, offset);
    trav_size = min (((offset+size) - local_offset), table->page_size);

    if (!trav) {
      /* page not in cache, we need to generate page fault */
      trav = ioc_page_create (ioc_inode, trav_offset);
      fault = 1;
      if (!trav) {
	gf_log (frame->this->name, GF_LOG_CRITICAL,
		"ioc_page_create returned NULL");
      }
    } 

    ioc_wait_on_page (trav, frame, local_offset, trav_size);

    if (trav->ready) {
      /* page found in cache */
      if (!might_need_validate) {
	/* fresh enough */
	gf_log (frame->this->name, GF_LOG_DEBUG,
		"cache hit for trav_offset=%lld/local_offset=%lld",
		trav_offset, local_offset);
	ioc_page_wakeup (trav);
      } else {
	/* we need to validate the cache */
	need_validate = 1;
	trav->ready = 0;
      }
    }

    ioc_inode_unlock (ioc_inode);
    
    if (fault) {
      fault = 0;
      /* new page created, increase the table->cache_used */
      ioc_page_fault (ioc_inode, frame, fd, trav_offset);
    }

    if (need_validate) {
      need_validate = 0;
      gf_log (frame->this->name, GF_LOG_DEBUG,
	      "sending validate request for inode(%"PRId64") at offset=%lld",
	      fd->inode->ino, trav_offset);
      ioc_cache_validate (frame, ioc_inode, fd, trav);
    }

    
    trav_offset += table->page_size;
  }

  ioc_frame_return (frame);

  if (ioc_need_prune (ioc_inode->table)) {
    ioc_prune (ioc_inode->table);
  }

  return;
}


/*
 * ioc_readv -
 * 
 * @frame:
 * @this:
 * @fd:
 * @size:
 * @offset:
 *
 */
int32_t
ioc_readv (call_frame_t *frame,
	   xlator_t *this,
	   fd_t *fd,
	   size_t size,
	   off_t offset)
{
  ioc_inode_t *ioc_inode = ioc_get_inode (fd->inode->ctx, this->name);
  ioc_local_t *local = NULL;
  data_t *fd_ctx_data = dict_get (fd->ctx, this->name);
  uint32_t weight = 0;

  if (!ioc_inode) {
    /* caching disabled, go ahead with normal readv */
    STACK_WIND (frame, 
		ioc_readv_disabled_cbk,
		FIRST_CHILD (frame->this), 
		FIRST_CHILD (frame->this)->fops->readv,
		fd, 
		size, 
		offset);
    return 0;

  }

  if (fd_ctx_data) {
    /* disable caching for this fd, go ahead with normal readv */
    STACK_WIND (frame, 
		ioc_readv_disabled_cbk,
		FIRST_CHILD (frame->this), 
		FIRST_CHILD (frame->this)->fops->readv,
		fd, 
		size, 
		offset);
    return 0;
  }

  local = (ioc_local_t *) calloc (1, sizeof (ioc_local_t));
  INIT_LIST_HEAD (&local->fill_list);

  frame->local = local;  
  local->pending_offset = offset;
  local->pending_size = size;
  local->offset = offset;
  local->size = size;
  local->inode = ioc_inode;

  gf_log (this->name,
	  GF_LOG_DEBUG,
	  "NEW REQ (%p) offset = %lld && size = %d", frame, offset, size);

  weight = ioc_inode->weight;

  ioc_table_lock (ioc_inode->table);
  list_move_tail (&ioc_inode->inode_lru, &ioc_inode->table->inode_lru[weight]);
  ioc_table_unlock (ioc_inode->table);

  dispatch_requests (frame, ioc_inode, fd, offset, size);
  
  return 0;
}

/*
 * ioc_writev_cbk -
 * 
 * @frame:
 * @cookie:
 * @this:
 * @op_ret:
 * @op_errno:
 *
 */
int32_t
ioc_writev_cbk (call_frame_t *frame,
		void *cookie,
		xlator_t *this,
		int32_t op_ret,
		int32_t op_errno,
		struct stat *stbuf)
{
  ioc_local_t *local = frame->local;
  ioc_inode_t *ioc_inode = ioc_get_inode (local->fd->inode->ctx, this->name);
  
  if (ioc_inode)
    ioc_inode_flush (ioc_inode);

  STACK_UNWIND (frame, op_ret, op_errno, stbuf);
  return 0;
}

/*
 * ioc_writev
 * 
 * @frame:
 * @this:
 * @fd:
 * @vector:
 * @count:
 * @offset:
 *
 */
int32_t
ioc_writev (call_frame_t *frame,
	    xlator_t *this,
	    fd_t *fd,
	    struct iovec *vector,
	    int32_t count,
	    off_t offset)
{
  ioc_local_t *local = calloc (1, sizeof (ioc_local_t));
  ioc_inode_t *ioc_inode = ioc_get_inode (fd->inode->ctx, this->name);

  local->fd = fd;
  frame->local = local;

  if (ioc_inode)
    ioc_inode_flush (ioc_inode);

  STACK_WIND (frame,
	      ioc_writev_cbk,
	      FIRST_CHILD(this),
	      FIRST_CHILD(this)->fops->writev,
	      fd,
	      vector,
	      count,
	      offset);

  return 0;
}

/*
 * ioc_truncate_cbk -
 * 
 * @frame:
 * @cookie:
 * @this:
 * @op_ret:
 * @op_errno:
 * @buf:
 *
 */
int32_t 
ioc_truncate_cbk (call_frame_t *frame,
		  void *cookie,
		  xlator_t *this,
		  int32_t op_ret,
		  int32_t op_errno,
		  struct stat *buf)
{

  STACK_UNWIND (frame, op_ret, op_errno, buf);
  return 0;
}

/*
 * ioc_truncate -
 * 
 * @frame:
 * @this:
 * @loc:
 * @offset:
 *
 */
int32_t 
ioc_truncate (call_frame_t *frame,
	      xlator_t *this,
	      loc_t *loc,
	      off_t offset)
{
  ioc_inode_t *ioc_inode =  ioc_get_inode (loc->inode->ctx, this->name);

  if (ioc_inode)
    ioc_inode_flush (ioc_inode);

  STACK_WIND (frame,
	      ioc_truncate_cbk,
	      FIRST_CHILD(this),
	      FIRST_CHILD(this)->fops->truncate,
	      loc,
	      offset);
  return 0;
}

/*
 * ioc_ftruncate -
 * 
 * @frame:
 * @this:
 * @fd:
 * @offset:
 *
 */
int32_t
ioc_ftruncate (call_frame_t *frame,
	       xlator_t *this,
	       fd_t *fd,
	       off_t offset)
{
  ioc_inode_t *ioc_inode = ioc_get_inode (fd->inode->ctx, this->name);
  
  if (ioc_inode)
    ioc_inode_flush (ioc_inode);

  STACK_WIND (frame,
	      ioc_truncate_cbk,
	      FIRST_CHILD(this),
	      FIRST_CHILD(this)->fops->ftruncate,
	      fd,
	      offset);
  return 0;
}

int32_t
ioc_lk_cbk (call_frame_t *frame,
	    void *cookie,
	    xlator_t *this,
	    int32_t op_ret,
	    int32_t op_errno,
	    struct flock *lock)
{
  STACK_UNWIND (frame, op_ret, op_errno, lock);
  return 0;
}

int32_t 
ioc_lk (call_frame_t *frame,
	xlator_t *this,
	fd_t *fd,
	int32_t cmd,
	struct flock *lock)
{
  ioc_inode_t *ioc_inode = ioc_get_inode (fd->inode->ctx, this->name);

  if (!ioc_inode) {
    gf_log (this->name, GF_LOG_ERROR,
	    "inode context is NULL: returning EBADFD");
    STACK_UNWIND (frame, -1, EBADFD, NULL);
  }

  ioc_inode_lock (ioc_inode);
  gettimeofday (&ioc_inode->tv, NULL);
  ioc_inode_unlock (ioc_inode);

  STACK_WIND (frame, ioc_lk_cbk, 
	      FIRST_CHILD (this),
	      FIRST_CHILD (this)->fops->lk, fd, cmd, lock);
  return 0;
}

int32_t
ioc_get_priority_list (const char *opt_str, struct list_head *first)
{
  int32_t max_pri = 0;
  char *tmp_str = NULL;
  char *tmp_str1 = NULL;
  char *tmp_str2 = NULL;
  char *dup_str = NULL;
  char *stripe_str = NULL;
  char *pattern = NULL;
  char *priority = NULL;
  char *string = strdup (opt_str);
  struct ioc_priority *curr = NULL;

  /* Get the pattern for cache priority. "option priority *.jpg:1,abc*:2" etc */
  /* TODO: inode_lru in table is statically hard-coded to 5, should be changed to 
   *      run-time configuration */
  stripe_str = strtok_r (string, ",", &tmp_str);
  while (stripe_str) {
    curr = calloc (1, sizeof (struct ioc_priority));
    list_add_tail (&curr->list, first);

    dup_str = strdup (stripe_str);
    pattern = strtok_r (dup_str, ":", &tmp_str1);
    if (!pattern)
      return -1;
    priority = strtok_r (NULL, ":", &tmp_str1);
    if (!priority)
      return -1;
    gf_log ("io-cache", 
	    GF_LOG_DEBUG, 
	    "ioc priority : pattern %s : priority %s", 
	    pattern,
	    priority);
    curr->pattern = strdup (pattern);
    curr->priority = strtol (priority, &tmp_str2, 0);
    if (tmp_str2 && (*tmp_str2))
      return -1;
    else
      max_pri = max (max_pri, curr->priority);
    stripe_str = strtok_r (NULL, ",", &tmp_str);
  }

  return max_pri;
}

/*
 * init - 
 * @this:
 *
 */
int32_t 
init (xlator_t *this)
{
  ioc_table_t *table;
  dict_t *options = this->options;
  uint32_t index = 0;

  if (!this->children || this->children->next) {
    gf_log (this->name, GF_LOG_ERROR,
	    "FATAL: io-cache not configured with exactly one child");
    return -1;
  }

  table = (void *) calloc (1, sizeof (*table));

  table->xl = this;
  table->page_size = IOC_PAGE_SIZE;
  table->cache_size = IOC_CACHE_SIZE;

  if (dict_get (options, "page-size")) {
    table->page_size = gf_str_to_long_long (data_to_str (dict_get (options,
								   "page-size")));
    gf_log (this->name, GF_LOG_DEBUG,
	    "Using table->page_size = 0x%x",
	    table->page_size);
  }

  if (dict_get (options, "cache-size")) {
    table->cache_size = gf_str_to_long_long (data_to_str (dict_get (options,
								      "cache-size")));
    gf_log (this->name, GF_LOG_DEBUG,
	    "Using table->cache_size = 0x%x",
	    table->cache_size);
  }

  table->force_revalidate_timeout = 1;

  if (dict_get (options, "force-revalidate-timeout")) {
    table->force_revalidate_timeout = data_to_uint32 (dict_get (options,
								"force-revalidate-timeout"));
    gf_log (this->name, GF_LOG_DEBUG,
	    "Using %d seconds to force revalidate cache",
	    table->force_revalidate_timeout);
  }

  INIT_LIST_HEAD (&table->priority_list);
  if (dict_get (options, "priority")) {
    char *option_list = data_to_str (dict_get (options, "priority"));
    gf_log (this->name,
	    GF_LOG_DEBUG,
	    "option path %s", option_list);
    /* parse the list of pattern:priority */
    table->max_pri = ioc_get_priority_list (option_list, &table->priority_list);
    
    if (table->max_pri == -1)
      return -1;
  }
  table->max_pri ++;
  INIT_LIST_HEAD (&table->inodes);
  
  table->inode_lru = calloc (table->max_pri, sizeof (struct list_head));
  for (index = 0; index < (table->max_pri); index++)
    INIT_LIST_HEAD (&table->inode_lru[index]);

  pthread_mutex_init (&table->table_lock, NULL);
  this->private = table;
  return 0;
}

/*
 * fini -
 * 
 * @this:
 *
 */
void
fini (xlator_t *this)
{
  ioc_table_t *table = this->private;

  pthread_mutex_destroy (&table->table_lock);
  freee (table);

  this->private = NULL;
  return;
}

struct xlator_fops fops = {
  .open        = ioc_open,
  .create      = ioc_create,
  .readv       = ioc_readv,
  .writev      = ioc_writev,
  .close       = ioc_close,
  .truncate    = ioc_truncate,
  .ftruncate   = ioc_ftruncate,
  .forget      = ioc_forget,
  .utimens     = ioc_utimens,
  .lookup      = ioc_lookup,
  .lk          = ioc_lk
};

struct xlator_mops mops = {
};