unify.c

分布式文件系统

}

/**
 * unify_flush_cbk - 
 */
int32_t
unify_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;
}

/**
 * unify_flush -
 */
int32_t
unify_flush (call_frame_t *frame,
	     xlator_t *this,
	     fd_t *fd)
{
  xlator_t *child = NULL;

  UNIFY_CHECK_FD_CTX_AND_UNWIND_ON_ERR(fd);

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

  STACK_WIND (frame, unify_flush_cbk, child, 
	      child->fops->flush, fd);

  return 0;
}

/**
 * unify_close_cbk -
 */
int32_t
unify_ns_close_cbk (call_frame_t *frame,
		    void *cookie,
		    xlator_t *this,
		    int32_t op_ret,
		    int32_t op_errno)
{
  unify_local_t *local = frame->local;

  LOCK (&frame->lock);
  {
    if (op_ret >= 0) {
      local->op_ret = op_ret;
      local->op_errno = op_errno;
    }
  }
  UNLOCK (&frame->lock);

  STACK_UNWIND (frame, local->op_ret, local->op_errno);
  
  return 0;
}

/**
 * unify_close_cbk -
 */
int32_t
unify_close_cbk (call_frame_t *frame,
		 void *cookie,
		 xlator_t *this,
		 int32_t op_ret,
		 int32_t op_errno)
{
  unify_local_t *local = frame->local;

  LOCK (&frame->lock);
  {
    if (op_ret >= 0) { 
      local->op_ret = op_ret;
      local->op_errno = op_errno;
    }
  }
  UNLOCK (&frame->lock);

  /* to namespace */
  STACK_WIND (frame, unify_ns_close_cbk, NS(this),
	      NS(this)->fops->close, local->fd);

  return 0;
}

/**
 * unify_close - send 'close' fop to all the nodes where 'fd' is open.
 */
int32_t
unify_close (call_frame_t *frame,
	     xlator_t *this,
	     fd_t *fd)
{
  unify_local_t *local = NULL;
  xlator_t *child = NULL;

  UNIFY_CHECK_FD_CTX_AND_UNWIND_ON_ERR(fd);

  /* Init */
  INIT_LOCAL (frame, local);
  local->inode = fd->inode;
  local->fd = fd;

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

  /* to storage node */
  STACK_WIND (frame, unify_close_cbk, child,
	      child->fops->close, fd);

  return 0;
}

/**
 * unify_fsync_cbk - 
 */
int32_t
unify_fsync_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;
}

/**
 * unify_fsync - 
 */
int32_t
unify_fsync (call_frame_t *frame,
	     xlator_t *this,
	     fd_t *fd,
	     int32_t flags)
{
  xlator_t *child = NULL;

  UNIFY_CHECK_FD_CTX_AND_UNWIND_ON_ERR(fd);

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

  STACK_WIND (frame, unify_fsync_cbk, child,
	      child->fops->fsync, fd, flags);

  return 0;
}

/**
 * unify_fstat - Send fstat FOP to Namespace only if its directory, and to 
 *     both namespace and the storage node if its a file.
 */
int32_t
unify_fstat (call_frame_t *frame,
	     xlator_t *this,
	     fd_t *fd)
{
  unify_local_t *local = NULL;

  UNIFY_CHECK_FD_AND_UNWIND_ON_ERR(fd);

  INIT_LOCAL (frame, local);

  if (dict_get (fd->ctx, this->name)) {
    /* If its set, then its file */
    xlator_t *child = NULL;

    child = data_to_ptr (dict_get (fd->ctx, this->name));
    local->call_count = 2;

    STACK_WIND (frame, unify_buf_cbk, child,
		child->fops->fstat, fd);

    STACK_WIND (frame, unify_buf_cbk, NS(this),
		NS(this)->fops->fstat, fd);

  } else {
    /* this is an directory */
    local->call_count = 1;
    STACK_WIND (frame, unify_buf_cbk, NS(this),
		NS(this)->fops->fstat, fd);

  }

  return 0;
}

/**
 * unify_getdents_cbk - 
 */
int32_t
unify_getdents_cbk (call_frame_t *frame,
		   void *cookie,
		   xlator_t *this,
		   int32_t op_ret,
		   int32_t op_errno,
		   dir_entry_t *entry,
		   int32_t count)
{
  STACK_UNWIND (frame, op_ret, op_errno, entry, count);
  return 0;
}

/**
 * unify_getdents - send the FOP request to all the nodes.
 */
int32_t
unify_getdents (call_frame_t *frame,
		xlator_t *this,
		fd_t *fd,
		size_t size,
		off_t offset,
		int32_t flag)
{
  UNIFY_CHECK_FD_AND_UNWIND_ON_ERR (fd);

  STACK_WIND (frame, unify_getdents_cbk, NS(this),
	      NS(this)->fops->getdents, fd, size, offset, flag);

  return 0;
}


/**
 * unify_readdir_cbk - 
 */
int32_t
unify_readdir_cbk (call_frame_t *frame,
		   void *cookie,
		   xlator_t *this,
		   int32_t op_ret,
		   int32_t op_errno,
		   gf_dirent_t *buf)
{
  STACK_UNWIND (frame, op_ret, op_errno, buf);

  return 0;
}

/**
 * unify_readdir - send the FOP request to all the nodes.
 */
int32_t
unify_readdir (call_frame_t *frame,
	       xlator_t *this,
	       fd_t *fd,
	       size_t size,
	       off_t offset)
{
  UNIFY_CHECK_FD_AND_UNWIND_ON_ERR (fd);

  STACK_WIND (frame, unify_readdir_cbk, NS(this),
	      NS(this)->fops->readdir, fd, size, offset);

  return 0;
}

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

  LOCK (&frame->lock);
  {
    callcnt = --local->call_count;
    if (op_ret >= 0)
      local->op_ret = op_ret;
  }
  UNLOCK (&frame->lock);
  
  if (!callcnt) {
    STACK_UNWIND (frame, op_ret, op_errno);
  }
  
  return 0;
}

/**
 * unify_closedir - send 'closedir' fop to all the nodes, where 'fd' is open. 
 */
int32_t
unify_closedir (call_frame_t *frame,
		xlator_t *this,
		fd_t *fd)
{
  int16_t *list = NULL;
  int16_t index = 0;
  unify_local_t *local = NULL;
  unify_private_t *priv = this->private;

  UNIFY_CHECK_FD_AND_UNWIND_ON_ERR (fd);

  INIT_LOCAL (frame, local);
  
  if (dict_get (fd->inode->ctx, this->name)) {
    list = data_to_ptr (dict_get (fd->inode->ctx, this->name));
  } else {
    gf_log (this->name, GF_LOG_ERROR, 
	    "returning EINVAL, no list found in inode ctx");
    STACK_UNWIND (frame, -1, EINVAL, NULL);
    return 0;
  }
  for (index = 0; list[index] != -1; index++)
    local->call_count++;
  
  for (index = 0; list[index] != -1; index++) {
    char need_break = list[index+1] == -1;
    STACK_WIND (frame,
		unify_closedir_cbk,
		priv->xl_array[list[index]],
		priv->xl_array[list[index]]->fops->closedir,
		fd);
    if (need_break)
      break;
  }

  return 0;
}

/**
 * unify_fsyncdir_cbk - 
 */
int32_t
unify_fsyncdir_cbk (call_frame_t *frame,
		    void *cookie,
		    xlator_t *this,
		    int32_t op_ret,
		    int32_t op_errno)
{
  int32_t callcnt = 0;
  unify_local_t *local = frame->local;
  call_frame_t *prev_frame = cookie;

  LOCK (&frame->lock);
  {
    callcnt = --local->call_count;
    
    if (op_ret == -1) {
      gf_log (this->name, GF_LOG_ERROR, 
	      "fop failed on %s (%d)", prev_frame->this->name, op_errno);
      local->op_errno = op_errno;
    } else {
      local->op_ret = op_ret;
    }
  }
  UNLOCK (&frame->lock);
  
  if (!callcnt) {
    STACK_UNWIND (frame, local->op_ret, local->op_errno);
  }
  return 0;
}

/**
 * unify_fsyncdir -
 */
int32_t
unify_fsyncdir (call_frame_t *frame,
		xlator_t *this,
		fd_t *fd,
		int32_t flags)
{
  int16_t *list = NULL;
  int16_t index = 0;
  unify_local_t *local = NULL;
  unify_private_t *priv = this->private;

  UNIFY_CHECK_FD_AND_UNWIND_ON_ERR (fd);

  INIT_LOCAL (frame, local);
  
  if (dict_get (fd->inode->ctx, this->name)) {
    list = data_to_ptr (dict_get (fd->inode->ctx, this->name));
  } else {
    gf_log (this->name, GF_LOG_ERROR, 
	    "returning EINVAL, no list found in inode ctx");
    STACK_UNWIND (frame, -1, EINVAL, NULL);
    return 0;
  }
  for (index = 0; list[index] != -1; index++)
    local->call_count++;
  
  for (index = 0; list[index] != -1; index++) {
    char need_break = list[index+1] == -1;
    STACK_WIND (frame,
		unify_fsyncdir_cbk,
		priv->xl_array[list[index]],
		priv->xl_array[list[index]]->fops->fsyncdir,
		fd,
		flags);
    if (need_break)
      break;
  }

  return 0;
}

/**
 * unify_lk_cbk - UNWIND frame with the proper return arguments.
 */
int32_t
unify_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;
}

/**
 * unify_lk - Send it to all the storage nodes, (should be 1) which has file.
 */
int32_t
unify_lk (call_frame_t *frame,
	  xlator_t *this,
	  fd_t *fd,
	  int32_t cmd,
	  struct flock *lock)
{
  xlator_t *child = NULL;

  UNIFY_CHECK_FD_CTX_AND_UNWIND_ON_ERR (fd);

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

  STACK_WIND (frame, unify_lk_cbk, child,
	      child->fops->lk, fd, cmd, lock);

  return 0;
}

/**
 * unify_setxattr_cbk - When all the child nodes return, UNWIND frame.
 */
int32_t
unify_setxattr_cbk (call_frame_t *frame,
		    void *cookie,
		    xlator_t *this,
		    int32_t op_ret,
		    int32_t op_errno)
{
  int32_t callcnt = 0;
  unify_local_t *local = frame->local;
  call_frame_t *prev_frame = cookie;

  LOCK (&frame->lock);
  {
    callcnt = --local->call_count;
    
    if (op_ret == -1) {
      gf_log (this->name, GF_LOG_ERROR, 
	      "fop failed on %s (%d)", prev_frame->this->name, op_errno);
      local->op_errno = op_errno;
    } else {
      local->op_ret = op_ret;
    }
  }
  UNLOCK (&frame->lock);
  
  if (!callcnt) {
    STACK_UNWIND (frame, local->op_ret, local->op_errno);
  }

  return 0;
}

/**
 * unify_sexattr - This function should be sent to all the storage nodes, which 
 *    contains the file, (excluding namespace).
 */
int32_t
unify_setxattr (call_frame_t *frame,
		xlator_t *this,
		loc_t *loc,
		dict_t *dict,
		int32_t flags)
{
  unify_private_t *priv = this->private;
  unify_local_t *local = NULL;
  int16_t *list = NULL;
  int16_t index = 0;
  int32_t call_count = 0;

  UNIFY_CHECK_INODE_CTX_AND_UNWIND_ON_ERR (loc);

  /* Initialization */
  INIT_LOCAL (frame, local);

  list = data_to_ptr (dict_get (loc->inode->ctx, this->name));

  for (index = 0; list[index] != -1; index++) {
    if (NS(this) != priv->xl_array[list[index]]) {
      local->call_count++;
      call_count++;
    }
  }
   
  if (local->call_count) {
    for (index = 0; list[index] != -1; index++) {
      if (priv->xl_array[list[index]] != NS(this)) {
	STACK_WIND (frame,
		    unify_setxattr_cbk,
		    priv->xl_array[list[index]],
		    priv->xl_array[list[index]]->fops->setxattr,
		    loc,
		    dict,
		    flags);
	if (!--call_count)
	  break;
      }
    }
  } else {
    /* No entry in storage nodes */
    gf_log (this->name, GF_LOG_ERROR, 
	    "returning ENOENT, file not found on storage node.");
    STACK_UNWIND (frame, -1, ENOENT);
  }

  return 0;
}


/**
 * unify_getxattr_cbk - This function is called from only one child, so, no
 *     need of any lock or anything else, just send it to above layer 
 */
int32_t
unify_getxattr_cbk (call_frame_t *frame,
		    void *cookie,
		    xlator_t *this,
		    int32_t op_ret,
		    int32_t op_errno,
		    dict_t *value)
{
  STACK_UNWIND (frame, op_ret, op_errno, value);
  return 0;
}


/** 
 * unify_getxattr - This FOP is sent to only the storage node.
 */
int32_t
unify_getxattr (call_frame_t *frame,
		xlator_t *this,
		loc_t *loc)
{
  unify_private_t *priv = this->private;
  int16_t *list = NULL;
  int16_t index = 0;
  int16_t count = 0;

  UNIFY_CHECK_INODE_CTX_AND_UNWIND_ON_ERR (loc);

  list = data_to_ptr (dict_get (loc->inode->ctx, this->name));

  for (index = 0; list[index] != -1; index++)
    count++;
  count--; //done for namespace entry

  if (count) {
    for (index = 0; list[index] != -1; index++) {
      if (priv->xl_array[list[index]] != NS(this)) {
	STACK_WIND (frame,
		    unify_getxattr_cbk,