stripe.c

分布式文件系统

}

/**
 * stripe_writedir - 
 */
int32_t
stripe_setdents (call_frame_t *frame,
		 xlator_t *this,
		 fd_t *fd,
		 int32_t flags,
		 dir_entry_t *entries,
		 int32_t count)
{
  stripe_local_t *local = NULL;
  stripe_private_t *priv = this->private;
  xlator_list_t *trav = this->children;

  STRIPE_CHECK_INODE_CTX_AND_UNWIND_ON_ERR (fd);

  /* Initialization */
  local = calloc (1, sizeof (stripe_local_t));
  local->op_ret = -1;
  frame->local = local;
  local->call_count = priv->child_count;

  while (trav) {
    STACK_WIND (frame,	      
		stripe_stack_unwind_cbk,
		trav->xlator,
		trav->xlator->fops->setdents,
		fd,
		flags,
		entries,
		count);
    trav = trav->next;
  }

  return 0;
}


/**
 * stripe_flush - 
 */
int32_t
stripe_flush (call_frame_t *frame,
	      xlator_t *this,
	      fd_t *fd)
{
  stripe_local_t *local = NULL;
  stripe_private_t *priv = this->private;
  xlator_list_t *trav = this->children;
  int8_t striped = 0;

  STRIPE_CHECK_INODE_CTX_AND_UNWIND_ON_ERR (fd);

  striped = data_to_int8 (dict_get (fd->inode->ctx, this->name));
  if (striped == 1) {
    STACK_WIND (frame,	      
		stripe_common_cbk,
		trav->xlator,
		trav->xlator->fops->flush,
		fd);
  } else {
    /* Initialization */
    local = calloc (1, sizeof (stripe_local_t));
    local->op_ret = -1;
    frame->local = local;
    local->call_count = priv->child_count;
    
    while (trav) {
      STACK_WIND (frame,	      
		  stripe_stack_unwind_cbk,
		  trav->xlator,
		  trav->xlator->fops->flush,
		  fd);
      trav = trav->next;
    }
  }
  return 0;
}

int32_t 
stripe_close_cbk (call_frame_t *frame,
		  void *cookie,
		  xlator_t *this,
		  int32_t op_ret,
		  int32_t op_errno)
{
  int32_t callcnt = 0;
  stripe_local_t *local = frame->local;

  LOCK (&frame->lock);
  {
    callcnt = --local->call_count;

    if (op_ret == -1) {
      gf_log (this->name, GF_LOG_WARNING, "%s returned errno %d",
	      ((call_frame_t *)cookie)->this->name, op_errno);
      if (op_errno == ENOTCONN) {
	local->failed = 1;
      } 
      local->op_errno = op_errno;
    }
    if (op_ret >= 0) 
      local->op_ret = op_ret;
  }
  UNLOCK (&frame->lock);

  if (!callcnt) {
    if (local->failed) {
      local->op_ret = -1;
    }
    STACK_WIND (frame,	      
		stripe_common_cbk,
		FIRST_CHILD(this),
		FIRST_CHILD(this)->fops->close,
		local->fd);
  }

  return 0;
}

/**
 * stripe_close - 
 */
int32_t
stripe_close (call_frame_t *frame,
	      xlator_t *this,
	      fd_t *fd)
{
  stripe_local_t *local = NULL;
  stripe_private_t *priv = this->private;
  xlator_list_t *trav = this->children;
  int8_t striped = 0;

  STRIPE_CHECK_INODE_CTX_AND_UNWIND_ON_ERR (fd);

  striped = data_to_int8 (dict_get (fd->inode->ctx, this->name));
  if (striped == 1) {
    STACK_WIND (frame,	      
		stripe_common_cbk,
		trav->xlator,
		trav->xlator->fops->close,
		fd);
  } else {
    /* Initialization */
    local = calloc (1, sizeof (stripe_local_t));
    local->op_ret = -1;
    local->fd = fd;
    frame->local = local;
    local->call_count = priv->child_count - 1;

    while (trav) {
      /* Send close() to the first child only after closing fd in all other 
       * nodes
       */
      if (trav->xlator != FIRST_CHILD(this)) {
	STACK_WIND (frame,	      
		    stripe_close_cbk,
		    trav->xlator,
		    trav->xlator->fops->close,
		    fd);
      }
      trav = trav->next;
    }
  }
  return 0;
}


/**
 * stripe_fsync - 
 */
int32_t
stripe_fsync (call_frame_t *frame,
	      xlator_t *this,
	      fd_t *fd,
	      int32_t flags)
{
  stripe_local_t *local = NULL;
  stripe_private_t *priv = this->private;
  xlator_list_t *trav = this->children;
  int8_t striped = 0;

  STRIPE_CHECK_INODE_CTX_AND_UNWIND_ON_ERR (fd);

  striped = data_to_int8 (dict_get (fd->inode->ctx, this->name));
  if (striped == 1) {
    STACK_WIND (frame,	      
		stripe_common_cbk,
		trav->xlator,
		trav->xlator->fops->fsync,
		fd,
		flags);
  } else {
    /* Initialization */
    local = calloc (1, sizeof (stripe_local_t));
    local->op_ret = -1;
    frame->local = local;
    local->call_count = priv->child_count;

    while (trav) {
      STACK_WIND (frame,	      
		  stripe_stack_unwind_cbk,
		  trav->xlator,
		  trav->xlator->fops->fsync,
		  fd,
		  flags);
      trav = trav->next;
    }
  }

  return 0;
}


/**
 * stripe_fstat - 
 */
int32_t
stripe_fstat (call_frame_t *frame,
	      xlator_t *this,
	      fd_t *fd)
{
  stripe_local_t *local = NULL;
  stripe_private_t *priv = this->private;
  xlator_list_t *trav = this->children;
  int8_t striped = 0;

  STRIPE_CHECK_INODE_CTX_AND_UNWIND_ON_ERR (fd);

  striped = data_to_int8 (dict_get (fd->inode->ctx, this->name));
  if (striped == 1) {
    STACK_WIND (frame,	      
		stripe_common_buf_cbk,
		trav->xlator,
		trav->xlator->fops->fstat,
		fd);
  } else {
    /* Initialization */
    local = calloc (1, sizeof (stripe_local_t));
    local->op_ret = -1;
    frame->local = local;
    local->inode = fd->inode;
    local->call_count = priv->child_count;

    while (trav) {
      STACK_WIND (frame,	      
		  stripe_stack_unwind_buf_cbk,
		  trav->xlator,
		  trav->xlator->fops->fstat,
		  fd);
      trav = trav->next;
    }
  }

  return 0;
}


/**
 * stripe_fchmod - 
 */
int32_t 
stripe_fchmod (call_frame_t *frame,
	       xlator_t *this,
	       fd_t *fd,
	       mode_t mode)
{
  stripe_local_t *local = NULL;
  stripe_private_t *priv = this->private;
  xlator_list_t *trav = this->children;
  int8_t striped = 0;

  STRIPE_CHECK_INODE_CTX_AND_UNWIND_ON_ERR (fd);

  striped = data_to_int8 (dict_get (fd->inode->ctx, this->name));
  if (striped == 1) {
    STACK_WIND (frame,	      
		stripe_common_buf_cbk,
		trav->xlator,
		trav->xlator->fops->fchmod,
		fd,
		mode);
  } else {
    /* Initialization */
    local = calloc (1, sizeof (stripe_local_t));
    local->op_ret = -1;
    frame->local = local;
    local->inode = fd->inode;
    local->call_count = priv->child_count;

    while (trav) {
      STACK_WIND (frame,	      
		  stripe_stack_unwind_buf_cbk,
		  trav->xlator,
		  trav->xlator->fops->fchmod,
		  fd,
		  mode);
      trav = trav->next;
    }
  }

  return 0;
}


/**
 * stripe_fchown - 
 */
int32_t 
stripe_fchown (call_frame_t *frame,
	       xlator_t *this,
	       fd_t *fd,
	       uid_t uid,
	       gid_t gid)
{
  stripe_local_t *local = NULL;
  stripe_private_t *priv = this->private;
  xlator_list_t *trav = this->children;
  int8_t striped = 0;

  STRIPE_CHECK_INODE_CTX_AND_UNWIND_ON_ERR (fd);

  striped = data_to_int8 (dict_get (fd->inode->ctx, this->name));
  if (striped == 1) {
    STACK_WIND (frame,	      
		stripe_common_buf_cbk,
		trav->xlator,
		trav->xlator->fops->fchown,
		fd,
		uid,
		gid);
  } else {
    /* Initialization */
    local = calloc (1, sizeof (stripe_local_t));
    local->op_ret = -1;
    frame->local = local;
    local->inode = fd->inode;
    local->call_count = priv->child_count;

    while (trav) {
      STACK_WIND (frame,	      
		  stripe_stack_unwind_buf_cbk,
		  trav->xlator,
		  trav->xlator->fops->fchown,
		  fd,
		  uid,
		  gid);
      trav = trav->next;
    }
  }

  return 0;
}


/**
 * stripe_ftruncate - 
 */
int32_t
stripe_ftruncate (call_frame_t *frame,
		  xlator_t *this,
		  fd_t *fd,
		  off_t offset)
{
  stripe_local_t *local = NULL;
  stripe_private_t *priv = this->private;
  xlator_list_t *trav = this->children;
  int8_t striped = 0;

  STRIPE_CHECK_INODE_CTX_AND_UNWIND_ON_ERR (fd);

  striped = data_to_int8 (dict_get (fd->inode->ctx, this->name));
  if (striped == 1) {
    STACK_WIND (frame,	      
		stripe_common_buf_cbk,
		trav->xlator,
		trav->xlator->fops->ftruncate,
		fd,
		offset);
  } else {
    /* Initialization */
    local = calloc (1, sizeof (stripe_local_t));
    local->op_ret = -1;
    frame->local = local;
    local->inode = fd->inode;
    local->call_count = priv->child_count;

    while (trav) {
      STACK_WIND (frame,	      
		  stripe_stack_unwind_buf_cbk,
		  trav->xlator,
		  trav->xlator->fops->ftruncate,
		  fd,
		  offset);
      trav = trav->next;
    }
  }
  return 0;
}


/**
 * stripe_closedir - 
 */
int32_t
stripe_closedir (call_frame_t *frame,
		 xlator_t *this,
		 fd_t *fd)
{
  stripe_local_t *local = NULL;
  stripe_private_t *priv = this->private;
  xlator_list_t *trav = this->children;

  STRIPE_CHECK_INODE_CTX_AND_UNWIND_ON_ERR (fd);

  /* Initialization */
  local = calloc (1, sizeof (stripe_local_t));
  local->op_ret = -1;
  frame->local = local;
  local->call_count = priv->child_count;
  
  while (trav) {
    STACK_WIND (frame,	      
		stripe_stack_unwind_cbk,
		trav->xlator,
		trav->xlator->fops->closedir,
		fd);
    trav = trav->next;
  }

  return 0;
}


/**
 * stripe_fsyncdir - 
 */
int32_t
stripe_fsyncdir (call_frame_t *frame,
		 xlator_t *this,
		 fd_t *fd,
		 int32_t flags)
{
  stripe_local_t *local = NULL;
  stripe_private_t *priv = this->private;
  xlator_list_t *trav = this->children;

  STRIPE_CHECK_INODE_CTX_AND_UNWIND_ON_ERR (fd);

  /* Initialization */
  local = calloc (1, sizeof (stripe_local_t));
  local->op_ret = -1;
  frame->local = local;
  local->call_count = priv->child_count;

  while (trav) {
    STACK_WIND (frame,	      
		stripe_stack_unwind_cbk,
		trav->xlator,
		trav->xlator->fops->fsyncdir,
		fd,
		flags);
    trav = trav->next;
  }

  return 0;
}


/**
 * stripe_single_readv_cbk - This function is used as return fn, when the 
 *     file name doesn't match the pattern specified for striping.
 */
int32_t
stripe_single_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;
}

/**
 * stripe_readv_cbk - get all the striped reads, and order it properly, send it
 *        to above layer after putting it in a single vector.
 */
int32_t
stripe_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)
{
  int32_t index = 0;
  int32_t callcnt = 0;
  call_frame_t *main_frame = NULL;
  stripe_local_t *main_local = NULL;
  stripe_local_t *local = frame->local;

  index = local->node_index;
  main_frame = local->orig_frame;
  main_local = main_frame->local;

  LOCK (&main_frame->lock);
  {
    main_local->replies[index].op_ret = op_ret;
    main_local->replies[index].op_errno = op_errno;
    if (op_ret > 0) {
      main_local->replies[index].count  = count;
      main_local->replies[index].vector = iov_dup (vector, count);
      main_local->replies[index].stbuf = *stbuf;
      dict_copy (frame->root->rsp_refs, main_frame->root->rsp_refs);
    }
    callcnt = ++main_local->call_count;
  }
  UNLOCK(&main_frame->lock);

  if (callcnt == main_local->wind_count) {
    int32_t final_count = 0;
    struct iovec *final_vec = NULL;
    struct stat tmp_stbuf = {0,};
    dict_t *refs = main_frame->root->rsp_refs;

    op_ret = 0;
    memcpy (&tmp_stbuf, &main_local->replies[0].stbuf, sizeof (struct stat));
    for (index=0; index < main_local->wind_count; index++) {
      /* TODO: check whether each stripe returned 'expected'
       * number of bytes 
       */
      if (main_local->replies[index].op_ret == -1) {
        op_ret = -1;
        op_errno = main_local->replies[index].op_errno;
        break;
      }
      op_ret += main_local->replies[index].op_ret;
      final_count += main_local->replies[index].count;
      /* TODO: Do I need to send anything more in stbuf? */
      if (tmp_stbuf.st_size < main_local->replies[index].stbuf.st_size)
	tmp_stbuf.st_size = main_local->replies[index].stbuf.st_size;
    }

    if (op_ret != -1) {
      final_vec = calloc (final_count, sizeof (struct iovec));
      final_count = 0;

      for (index=0; index < main_local->wind_count; index++) {
        memcpy (final_vec + final_count,
                main_local->replies[index].vector,
                main_local->replies[index].count * sizeof (struct iovec));
        final_count += main_local->replies[index].count;

        free (main_local->replies[index].vector);
      }
    } else {
      final_vec = NULL;
      final_count = 0;
    }
    /* */
    freee (main_local->replies);
    refs = main_frame->root->rsp_refs;
    STACK_UNWIND (main_frame, op_ret, op_errno, final_vec, final_count, &tmp_stbuf);

    dict_unref (refs);
    if (final_vec)
      free (final_vec);
  }

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

/**
 * stripe_readv - 
 */
int32_t
stripe_readv (call_frame_t *frame,
	      xlator_t *this,
	      fd_t *fd,
	      size_t size,
	      off_t offset)
{
  int32_t index = 0