write-behind.c

分布式文件系统

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

int32_t
wb_open (call_frame_t *frame,
         xlator_t *this,
         loc_t *loc,
         int32_t flags,
	 fd_t *fd)
{
  frame->local = calloc (1, sizeof(int32_t));
  *((int32_t *)frame->local) = flags;
  STACK_WIND (frame,
              wb_open_cbk,
              FIRST_CHILD(this),
              FIRST_CHILD(this)->fops->open,
              loc,
              flags,
	      fd);
  return 0;
}

int32_t
wb_create_cbk (call_frame_t *frame,
               void *cookie,
               xlator_t *this,
               int32_t op_ret,
               int32_t op_errno,
               fd_t *fd,
               inode_t *inode,
               struct stat *buf)
{
  if (op_ret != -1) {
    wb_file_t *file = calloc (1, sizeof (*file));

    file->pages.next = &file->pages;
    file->pages.prev = &file->pages;
    file->fd= fd;
    file->disable_till = 1048576;

    dict_set (fd->ctx,
              this->name,
              data_from_static_ptr (file));

    /* 
     * If mandatory locking has been enabled on this file,
     * we disable caching on it
     */
    if ((fd->inode->st_mode & S_ISGID) && 
	!(fd->inode->st_mode & S_IXGRP)) {
      file->disabled = 1;
    }

    LOCK_INIT (&file->lock);
    wb_file_ref (file);
  }

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

int32_t
wb_create (call_frame_t *frame,
           xlator_t *this,
	   loc_t *loc,
           int32_t flags,
           mode_t mode,
	   fd_t *fd)
{
  STACK_WIND (frame,
              wb_create_cbk,
              FIRST_CHILD(this),
              FIRST_CHILD(this)->fops->create,
	      loc,
              flags,
              mode,
	      fd);
  return 0;
}

int32_t 
wb_writev_cbk (call_frame_t *frame,
               void *cookie,
               xlator_t *this,
               int32_t op_ret,
               int32_t op_errno,
	       struct stat *stbuf)
{
  GF_ERROR_IF_NULL (this);

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

int32_t
wb_writev (call_frame_t *frame,
           xlator_t *this,
           fd_t *fd,
           struct iovec *vector,
           int32_t count,
           off_t offset)
{
  wb_file_t *file;
  wb_conf_t *conf = this->private;
  call_frame_t *wb_frame;
  dict_t *ref = NULL;
  struct stat buf = {0, };
  size_t size = iov_length (vector, count);

  if (!dict_get (fd->ctx, this->name)) {
    gf_log (this->name, GF_LOG_ERROR, "returning EBADFD");
    STACK_UNWIND (frame, -1, EBADFD, NULL);
    return 0;
  }

  file = data_to_ptr (dict_get (fd->ctx, this->name));
  
  if (file->disabled || file->disable_till) {
    if (size > file->disable_till) {
      file->disable_till = 0;
    } else {
      file->disable_till -= size;
    }

    STACK_WIND (frame, 
                wb_writev_cbk,
                FIRST_CHILD (frame->this), 
                FIRST_CHILD (frame->this)->fops->writev,
                file->fd, 
                vector, 
                count, 
                offset);
    return 0;
  }

  if (file->op_ret == -1) {
    /* delayed error delivery */
    gf_log (this->name, GF_LOG_ERROR, "delayed error : %d", file->op_errno);
    STACK_UNWIND (frame, -1, file->op_errno, &buf);
    file->op_ret = 0;
    return 0;
  }

  if (offset != file->offset)
    wb_sync (frame, file);

  wb_frame = copy_frame (frame);
  ref = dict_ref (frame->root->req_refs);

  /* FIXME: sending back a dummy stat buffer */
  STACK_UNWIND (frame, iov_length (vector, count), 0, &buf); /* liar! liar! :O */

  file->offset = (offset + iov_length (vector, count));
  {
    wb_page_t *page = calloc (1, sizeof (*page));

    page->vector = iov_dup (vector, count);
    page->count = count;
    page->offset = offset;
    page->refs = ref;

    page->next = &file->pages;
    page->prev = file->pages.prev;
    page->next->prev = page;
    page->prev->next = page;

    file->size += iov_length (vector, count);
  }

  if (file->size >= conf->aggregate_size) {
    wb_sync (wb_frame, file);
  }

  STACK_DESTROY (wb_frame->root);
  return 0;
}

int32_t
wb_readv_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)
{
  STACK_UNWIND (frame, op_ret, op_errno, vector, count, stbuf);
  return 0;
}

int32_t
wb_readv (call_frame_t *frame,
          xlator_t *this,
          fd_t *fd,
          size_t size,
          off_t offset)
{
  wb_file_t *file;

  if (!dict_get (fd->ctx, this->name)) {
    gf_log (this->name, GF_LOG_ERROR, "returning EBADFD");
    STACK_UNWIND (frame, -1, EBADFD, NULL);
    return 0;
  }

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

  /*
  list_for_each_entry (iter_fd, &(file->fd->inode->fds), inode_list) {
    wb_file_t *iter_file;
    if (dict_get (iter_fd->ctx, this->name)) {
      iter_file = data_to_ptr (dict_get (iter_fd->ctx, this->name));
      wb_sync (frame, iter_file);
    }
  }
  */

  STACK_WIND (frame,
              wb_readv_cbk,
              FIRST_CHILD(this),
              FIRST_CHILD(this)->fops->readv,
              fd,
              size,
              offset);

  return 0;
}

int32_t
wb_ffr_cbk (call_frame_t *frame,
            void *cookie,
            xlator_t *this,
            int32_t op_ret,
            int32_t op_errno)
{
  wb_file_t *file = frame->local;
  if (file->op_ret == -1) {
    op_ret = file->op_ret;
    op_errno = file->op_errno;

    file->op_ret = 0;
  }

  frame->local = NULL;
  wb_file_unref (file);
  STACK_UNWIND (frame, op_ret, op_errno);
  return 0;
}

int32_t
wb_ffr_bg_cbk (call_frame_t *frame,
	       void *cookie,
	       xlator_t *this,
	       int32_t op_ret,
	       int32_t op_errno)
{
  wb_file_t *file = frame->local;

  if (file->op_ret == -1) {
    op_ret = file->op_ret;
    op_errno = file->op_errno;

    file->op_ret = 0;
  }
  
  frame->local = NULL;
  wb_file_unref (file);
  STACK_DESTROY (frame->root);
  return 0;
}


static int32_t
wb_flush (call_frame_t *frame,
          xlator_t *this,
          fd_t *fd)
{
  wb_conf_t *conf = this->private;
  wb_file_t *file;
  call_frame_t *flush_frame;

  if (!dict_get (fd->ctx, this->name)) {
    gf_log (this->name, GF_LOG_ERROR, "returning EBADFD");
    STACK_UNWIND (frame, -1, EBADFD);
    return 0;
  }

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

  if (conf->flush_behind && (!file->disabled)) {
    flush_frame = copy_frame (frame);
    
    STACK_UNWIND (frame, file->op_ret, file->op_errno); // liar! liar! :O

    flush_frame->local = wb_file_ref (file);

    wb_sync (flush_frame, file);

    STACK_WIND (flush_frame,
                wb_ffr_bg_cbk,
                FIRST_CHILD(this),
                FIRST_CHILD(this)->fops->flush,
                fd);
  } else {
    frame->local = wb_file_ref (file);
    
    wb_sync (frame, file);


    STACK_WIND (frame,
		wb_ffr_cbk,
		FIRST_CHILD(this),
		FIRST_CHILD(this)->fops->flush,
		fd);
  }

  return 0;
}

int32_t
wb_fsync (call_frame_t *frame,
          xlator_t *this,
          fd_t *fd,
          int32_t datasync)
{
  wb_file_t *file;

  if (!dict_get (fd->ctx, this->name)) {
    gf_log (this->name, GF_LOG_ERROR, "returning EBADFD");
    STACK_UNWIND (frame, -1, EBADFD);
    return 0;
  }

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

  frame->local = wb_file_ref (file);

  STACK_WIND (frame,
              wb_ffr_cbk,
              FIRST_CHILD(this),
              FIRST_CHILD(this)->fops->fsync,
              fd,
              datasync);
  return 0;
}

int32_t
wb_close (call_frame_t *frame,
          xlator_t *this,
          fd_t *fd)
{
  wb_file_t *file;

  if (!dict_get (fd->ctx, this->name)) {
    gf_log (this->name, GF_LOG_ERROR, "returning EBADFD");
    STACK_UNWIND (frame, -1, EBADFD);
    return 0;
  }

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

  wb_sync (frame, file);

  frame->local = wb_file_ref (file);

  wb_file_unref (file);

  STACK_WIND (frame,
              wb_ffr_cbk,
              FIRST_CHILD(this),
              FIRST_CHILD(this)->fops->close,
              fd);
  return 0;
}

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

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

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

  conf->aggregate_size = 0;
  if (dict_get (options, "aggregate-size")) {
    conf->aggregate_size = gf_str_to_long_long (data_to_str (dict_get (options,
								       "aggregate-size")));
  }
  gf_log (this->name,
    GF_LOG_DEBUG,
    "using aggregate-size = %d", conf->aggregate_size);

  conf->flush_behind = 0;  
  if (dict_get (options, "flush-behind")) {
    if ((!strcasecmp (data_to_str (dict_get (options, "flush-behind")),
		      "on")) ||
	(!strcasecmp (data_to_str (dict_get (options, "flush-behind")),
		      "yes"))) {
      gf_log (this->name,
	      GF_LOG_DEBUG,
	      "%s: enabling flush-behind",
	      this->name);
      conf->flush_behind = 1;
    }
  }

  this->private = conf;
  return 0;
}

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

  freee (conf);
  return;
}

struct xlator_fops fops = {
  .writev      = wb_writev,
  .open        = wb_open,
  .create      = wb_create,
  .readv       = wb_readv,
  .flush       = wb_flush,
  .fsync       = wb_fsync,
  .close       = wb_close,
  .stat        = wb_stat,
  .fstat       = wb_fstat,
  .truncate    = wb_truncate,
  .ftruncate   = wb_ftruncate,
  .utimens     = wb_utimens,
};

struct xlator_mops mops = {
};