read-ahead.c

分布式文件系统

          xlator_t *this,
          fd_t *fd,
          size_t size,
          off_t offset)
{
  ra_file_t *file;
  ra_local_t *local;
  ra_conf_t *conf = this->private;


  gf_log (this->name, GF_LOG_DEBUG,
	  "NEW REQ at offset=%"PRId64" for size=%d",
	  offset, size);

  file = data_to_ptr (dict_get (fd->ctx, this->name));

  if (file->offset != offset) {
    gf_log (this->name, GF_LOG_DEBUG,
	    "received unexpected offset (%"PRId64" != %"PRId64"), resetting page_count to 0",
	    file->offset, offset);
    file->expected = file->page_count = 0;
  } else {
    gf_log (this->name, GF_LOG_DEBUG,
	    "received expected offset (%"PRId64") when page_count=%d",
	    offset, file->page_count);
    if (file->expected < (conf->page_size * conf->page_count)) {
      file->expected += size;
      file->page_count = min ((file->expected / file->page_size),
			      conf->page_count);
    }
  }

  if (file->disabled) {
    STACK_WIND (frame, 
		ra_readv_disabled_cbk,
		FIRST_CHILD (frame->this), 
		FIRST_CHILD (frame->this)->fops->readv,
		file->fd, size, offset);
    return 0;
  }

  //  if (fd->inode->buf.st_mtime != file->stbuf.st_mtime)
    /* flush the whole read-ahead cache */
    //flush_region (frame, file, 0, file->pages.prev->offset + 1);

  call_frame_t *ra_frame = copy_frame (frame);


  local = (void *) calloc (1, sizeof (*local));
  local->offset = offset;
  local->size = size;
  local->file = ra_file_ref (file);
  local->wait_count = 1; /* for synchronous STACK_UNWIND from protocol in case of error */
  local->fill.next = &local->fill;
  local->fill.prev = &local->fill;
  pthread_mutex_init (&local->local_lock, NULL);

  frame->local = local;

  dispatch_requests (frame, file);

  flush_region (frame, file, 0, floor (offset, file->page_size));

  ra_frame_return (frame);

  read_ahead (ra_frame, file);

  file->offset = offset + size;

  STACK_DESTROY (ra_frame->root);

  return 0;
}

int32_t
ra_flush_cbk (call_frame_t *frame,
              void *cookie,
              xlator_t *this,
              int32_t op_ret,
              int32_t op_errno)
{
  STACK_UNWIND (frame, op_ret, op_errno);
  return 0;
}

int32_t
ra_flush (call_frame_t *frame,
          xlator_t *this,
          fd_t *fd)
{
  data_t *file_data = dict_get (fd->ctx, this->name);
  ra_file_t *file = NULL;

  if (file_data) {
    file = data_to_ptr (file_data);
    
    flush_region (frame, file, 0, file->pages.prev->offset+1);
  }

  STACK_WIND (frame,
	      ra_flush_cbk,
	      FIRST_CHILD(this),
	      FIRST_CHILD(this)->fops->flush,
	      fd);
  return 0;
}

int32_t
ra_fsync (call_frame_t *frame,
          xlator_t *this,
          fd_t *fd,
          int32_t datasync)
{
  data_t *file_data = dict_get (fd->ctx, this->name);
  ra_file_t *file = NULL;

  if (file_data) {
    file = data_to_ptr (file_data);
    flush_region (frame, file, 0, file->pages.prev->offset+1);
  }
  STACK_WIND (frame,
              ra_flush_cbk,
              FIRST_CHILD(this),
              FIRST_CHILD(this)->fops->fsync,
              fd,
              datasync);
  return 0;
}

int32_t
ra_writev_cbk (call_frame_t *frame,
               void *cookie,
               xlator_t *this,
               int32_t op_ret,
               int32_t op_errno,
	       struct stat *stbuf)
{
  fd_t *fd = frame->local;
  data_t *file_data = dict_get (fd->ctx, this->name);
  ra_file_t *file = NULL;

  if (file_data) {
    file = data_to_ptr (file_data);
    flush_region (frame, file, 0, file->pages.prev->offset+1);
  }
  frame->local = NULL;
  STACK_UNWIND (frame, op_ret, op_errno, stbuf);
  return 0;
}

int32_t
ra_writev (call_frame_t *frame,
           xlator_t *this,
           fd_t *fd,
           struct iovec *vector,
           int32_t count,
           off_t offset)
{
  data_t *file_data = dict_get (fd->ctx, this->name);
  ra_file_t *file = NULL;

  frame->local = (void *) fd;
  if (file_data) {
    file = data_to_ptr (file_data);    
    flush_region (frame, file, 0, file->pages.prev->offset+1);

    /* reset the read-ahead counters too */
    file->expected = file->page_count = 0;
  }

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

  return 0;
}

int32_t 
ra_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;
}

int32_t 
ra_truncate (call_frame_t *frame,
             xlator_t *this,
             loc_t *loc,
             off_t offset)
{
  ra_file_t *file = NULL;
  fd_t *iter_fd = NULL;

  if (loc->inode) {
    LOCK (&(loc->inode->lock));
    {
      list_for_each_entry (iter_fd, &(loc->inode->fds), inode_list) {
	if (dict_get (iter_fd->ctx, this->name)) {
	  file = data_to_ptr (dict_get (iter_fd->ctx, this->name));
	  flush_region (frame, file, 0, file->pages.prev->offset + 1);
	}
      }
    }
    UNLOCK (&(loc->inode->lock));
  }

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

int32_t 
ra_ftruncate_cbk (call_frame_t *frame,
                 void *cookie,
                 xlator_t *this,
                 int32_t op_ret,
                 int32_t op_errno,
                 struct stat *buf)
{
  ra_local_t *local;
  ra_file_t *file;

  local = frame->local;
  file = local->file;

  frame->local = NULL;
  STACK_UNWIND (frame, op_ret, op_errno, buf);

  ra_file_unref (file);
  free (local);
  return 0;
}

int32_t
ra_fstat_cbk (call_frame_t *frame,
	      void *cookie,
	      xlator_t *this,
	      int32_t op_ret,
	      int32_t op_errno,
	      struct stat *buf)
{
  ra_local_t *local;
  ra_file_t *file;

  local = frame->local;
  file = local->file;

  if ((op_ret == 0) && (file->stbuf.st_mtime != buf->st_mtime))
    flush_region (frame, file, 0, file->pages.prev->offset + 1);

  frame->local = NULL;
  STACK_UNWIND (frame, op_ret, op_errno, buf);

  if (file)
    ra_file_unref (file);
  free (local);
  return 0;
}

int32_t 
ra_fstat (call_frame_t *frame,
	  xlator_t *this,
	  fd_t *fd)
{
  data_t *file_data = dict_get (fd->ctx, this->name);
  ra_local_t *local;
  ra_file_t *file = NULL;

  if (file_data) {
    file = data_to_ptr (file_data);
  }

  local = calloc (1, sizeof (*local));
  if (file)
    local->file = ra_file_ref (file);
  frame->local = local;

  STACK_WIND (frame,
	      ra_fstat_cbk,
	      FIRST_CHILD (this),
	      FIRST_CHILD (this)->fops->fstat,
	      fd);
  return 0;
}

int32_t
ra_fchown_cbk (call_frame_t *frame,
	       void *cookie,
	       xlator_t *this,
	       int32_t op_ret,
	       int32_t op_errno,
	       struct stat *buf)
{
  ra_local_t *local;
  ra_file_t *file;

  local = frame->local;
  file = local->file;

  if ((op_ret == 0) && (file->stbuf.st_mtime != buf->st_mtime))
    flush_region (frame, file, 0, file->pages.prev->offset + 1);

  frame->local = NULL;
  STACK_UNWIND (frame, op_ret, op_errno, buf);

  if (file)
    ra_file_unref (file);

  free (local);
  return 0;
}

int32_t
ra_fchown (call_frame_t *frame,
	   xlator_t *this,
	   fd_t *fd,
	   uid_t uid,
	   gid_t gid)
{
  data_t *file_data = dict_get (fd->ctx, this->name);
  ra_local_t *local;
  ra_file_t *file = NULL;

  if (file_data) {
    file = data_to_ptr (file_data);
  }

  local = calloc (1, sizeof (*local));

  if (file)
    local->file = ra_file_ref (file);

  frame->local = local;

  STACK_WIND (frame,
	      ra_fchown_cbk,
	      FIRST_CHILD (this),
	      FIRST_CHILD (this)->fops->fchown,
	      fd,
	      uid,
	      gid);
  return 0;
}

int32_t
ra_ftruncate (call_frame_t *frame,
              xlator_t *this,
              fd_t *fd,
              off_t offset)
{
  data_t *file_data = dict_get (fd->ctx, this->name);
  ra_file_t *file = NULL;
  ra_local_t *local = calloc (1, sizeof (*local));
  
  if (file_data) {
    file = data_to_ptr (file_data);
    flush_region (frame, file, 0, file->pages.prev->offset + 1);
  }

  local->file = ra_file_ref (file);
  frame->local = local;
  STACK_WIND (frame,
              ra_ftruncate_cbk,
              FIRST_CHILD(this),
              FIRST_CHILD(this)->fops->ftruncate,
              fd,
              offset);
  return 0;
}

int32_t 
init (xlator_t *this)
{
  ra_conf_t *conf;
  dict_t *options = this->options;

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

  conf = (void *) calloc (1, sizeof (*conf));
  conf->page_size = 256 * 1024;
  conf->page_count = 2;

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

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

  if (dict_get (options, "force-atime-update")) {
    char *force_atime_update_str = data_to_str (dict_get (options,
							  "force-atime-update"));
    if ((!strcasecmp (force_atime_update_str, "on")) ||
	(!strcasecmp (force_atime_update_str, "yes"))) {
      conf->force_atime_update = 1;
      gf_log (this->name, GF_LOG_DEBUG, "Forcing atime updates on cache hit");
    }
  }

  conf->files.next = &conf->files;
  conf->files.prev = &conf->files;

  pthread_mutex_init (&conf->conf_lock, NULL);
  this->private = conf;
  return 0;
}

void
fini (xlator_t *this)
{
  ra_conf_t *conf = this->private;

  pthread_mutex_destroy (&conf->conf_lock);
  freee (conf);

  this->private = NULL;
  return;
}

struct xlator_fops fops = {
  .open        = ra_open,
  .create      = ra_create,
  .readv       = ra_readv,
  .writev      = ra_writev,
  .flush       = ra_flush,
  .fsync       = ra_fsync,
  .close       = ra_close,
  .truncate    = ra_truncate,
  .ftruncate   = ra_ftruncate,
  .fstat       = ra_fstat,
  .fchown      = ra_fchown,
};

struct xlator_mops mops = {
};