stripe.c

分布式文件系统

  
  return 0;
}


/**
 * stripe_create - If a block-size is specified for the 'name', create the 
 *    file in all the child nodes. If not, create it in only first child.
 *
 * @name- complete path of the file to be created.
 */
int32_t
stripe_create (call_frame_t *frame,
	       xlator_t *this,
	       loc_t *loc,
	       int32_t flags,
	       mode_t mode,
	       fd_t *fd)
{
  stripe_private_t *priv = this->private;
  stripe_local_t *local = NULL;
  xlator_list_t *trav = NULL;
  off_t stripe_size = 0;

  stripe_size = stripe_get_matching_bs (loc->path, priv->pattern);
  
  /* files created in O_APPEND mode does not allow lseek() on fd */
  flags &= ~O_APPEND;

  if (priv->first_child_down || (stripe_size && priv->nodes_down)) {
    gf_log (this->name, GF_LOG_WARNING, "First node down, returning EIO");
    STACK_UNWIND (frame, -1, EIO, fd, loc->inode, NULL);
    return 0;
  }

  /* Initialization */
  local = calloc (1, sizeof (stripe_local_t));
  local->op_ret = -1;
  local->op_errno = ENOTCONN;
  local->stripe_size = stripe_size;
  local->path = strdup (loc->path);
  frame->local = local;
  local->fd = fd;
  local->inode = loc->inode;

  if (local->stripe_size) {
    /* Everytime in stripe lookup, all child nodes should be looked up */
    local->call_count = ((stripe_private_t *)this->private)->child_count;
    
    trav = this->children;
    while (trav) {
      STACK_WIND_COOKIE (frame,
		   stripe_create_cbk,
		   trav->xlator,  /* cookie */
		   trav->xlator,
		   trav->xlator->fops->create,
		   loc,
		   flags,
		   mode,
		   fd);
      trav = trav->next;
    }
  } else {
    /* This path doesn't match any pattern, create the file only in first node */
    local->call_count = 1;
    STACK_WIND_COOKIE (frame,
		 stripe_create_cbk,
		 FIRST_CHILD(this),
		 FIRST_CHILD(this),
		 FIRST_CHILD(this)->fops->create,
		 loc,
		 flags,
		 mode,
		 fd);
  }
       
  return 0;
}

/**
 * stripe_open_fail_cbk - 
 */
int32_t
stripe_open_fail_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;
  }
  UNLOCK (&frame->lock);

  if (!callcnt) {
    STACK_UNWIND (frame, local->op_ret, local->op_errno, local->fd);
  }

  return 0;
}

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

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

    if (op_ret == -1) {
      local->failed = 1;
      gf_log (this->name, GF_LOG_WARNING, "%s returned errno %d",
	      (char *)cookie, op_errno);
      local->op_ret = -1;
      local->op_errno = op_errno;
    }
    
    if (op_ret >= 0) {
      local->op_ret = op_ret;
    }
  }
  UNLOCK (&frame->lock);
  
  if (!callcnt) {
    if (local->failed && (local->striped != 1)) {
      stripe_private_t *priv = this->private;
      xlator_list_t *trav = this->children;

      local->op_ret = -1;
      local->call_count = priv->child_count;
      while (trav) {
	STACK_WIND (frame, 
		    stripe_open_fail_cbk,
		    trav->xlator,
		    trav->xlator->fops->close,
		    local->fd);
	trav = trav->next;
      }
      return 0;
    }
    if (local->op_ret >= 0) {
      dict_set (local->fd->ctx, 
		this->name, 
		data_from_uint64 (local->stripe_size));
    }
    freee (local->path);
    STACK_UNWIND (frame, local->op_ret, local->op_errno, local->fd);
  }

  return 0;
}


/**
 * stripe_getxattr_cbk - 
 */
int32_t
stripe_open_getxattr_cbk (call_frame_t *frame,
			  void *cookie,
			  xlator_t *this,
			  int32_t op_ret,
			  int32_t op_errno,
			  dict_t *dict)
{
  int32_t callcnt = 0;
  stripe_local_t *local = frame->local;
  xlator_list_t *trav = this->children;
  stripe_private_t *priv = this->private;

  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_ret = -1;
      local->op_errno = op_errno;
    }
  }
  UNLOCK (&frame->lock);
  
  if (!callcnt) {
    if (!local->failed && (local->op_ret != -1)) {
      /* If getxattr doesn't fails, call open */
      char size_key[256] = {0,};
      data_t *stripe_size_data = NULL;

      sprintf (size_key, "trusted.%s.stripe-size", this->name);
      stripe_size_data = dict_get (dict, size_key);

      if (stripe_size_data) {
	local->stripe_size = data_to_int64 (stripe_size_data);
      } else {
	/* if the file was created using earlier versions of stripe */
	local->stripe_size = stripe_get_matching_bs (local->path, 
						     ((stripe_private_t *)this->private)->pattern);
	if (local->stripe_size) {
	  gf_log (this->name, 
		  GF_LOG_WARNING,
		  "Seems like file(%s) created using earlier version",
		  local->path);
	} else {
	  gf_log (this->name,
		  GF_LOG_WARNING,
		  "no pattern found for file(%s), opening only in first node",
		  local->path);
	}
      }
    }
    
    local->call_count = priv->child_count;

    while (trav) {
      loc_t tmp_loc = {
	.path = local->path, 
	.inode = local->inode
      };
      STACK_WIND_COOKIE (frame,
		   stripe_open_cbk,
		   trav->xlator->name,
		   trav->xlator,
		   trav->xlator->fops->open,
		   &tmp_loc,
		   local->flags,
		   local->fd);
      trav = trav->next;
    }
  }
  return 0;
}

/**
 * stripe_open - 
 */
int32_t
stripe_open (call_frame_t *frame,
	     xlator_t *this,
	     loc_t *loc,
	     int32_t flags,
	     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 (loc);
  
  /* files opened in O_APPEND mode does not allow lseek() on fd */
  flags &= ~O_APPEND;

  if (priv->first_child_down) {
    gf_log (this->name, GF_LOG_WARNING, "First node down, returning ENOTCONN");
    STACK_UNWIND (frame, -1, ENOTCONN, NULL);
    return 0;
  }

  /* Initialization */
  local = calloc (1, sizeof (stripe_local_t));
  local->inode = loc->inode;
  local->fd = fd;
  frame->local = local;
  local->path = strdup (loc->path);

  striped = data_to_int8 (dict_get (loc->inode->ctx, this->name));
  local->striped = striped;

  if (striped == 1) {
    local->call_count = 1;

    /* File is present only in one node, no xattr's present */
    STACK_WIND_COOKIE (frame,
		 stripe_open_cbk,
		 trav->xlator->name,
		 trav->xlator,
		 trav->xlator->fops->open,
		 loc,
		 flags,
		 fd);
  } else {
    /* Striped files */
    local->flags = flags;
    local->call_count = priv->child_count;
    while (trav) {
      STACK_WIND (frame,
		  stripe_open_getxattr_cbk,
		  trav->xlator,
		  trav->xlator->fops->getxattr,
		  loc);
      trav = trav->next;
    }
  }

  return 0;
}


/**
 * stripe_opendir_fail_cbk- 
 */
int32_t
stripe_opendir_fail_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;
  }
  UNLOCK (&frame->lock);

  if (!callcnt) {
    STACK_UNWIND (frame, local->op_ret, local->op_errno, local->fd);
  }
  return 0;
}


/**
 * stripe_opendir_cbk - 
 */
int32_t
stripe_opendir_cbk (call_frame_t *frame,
		    void *cookie,
		    xlator_t *this,
		    int32_t op_ret,
		    int32_t op_errno,
		    fd_t *fd)
{
  int32_t callcnt = 0;
  stripe_local_t *local = frame->local;
  stripe_private_t *priv = this->private;
  xlator_list_t *trav = this->children;

  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);
      local->op_ret = -1;
      local->failed = 1;
      local->op_errno = op_errno;
    }
    
    if (op_ret >= 0) {
      local->op_ret = op_ret;
    }
  }
  UNLOCK (&frame->lock);

  if (!callcnt) {
    if ((local->op_ret >= 0) && local->failed) {
      /* Send closedir to nodes as opendir failed */
      local->op_ret = -1;
      local->call_count = priv->child_count;
      while (trav) {
	STACK_WIND (frame, 
		    stripe_opendir_fail_cbk,
		    trav->xlator,
		    trav->xlator->fops->closedir,
		    local->fd);
	trav = trav->next;
      }
      return 0;
    }
    STACK_UNWIND (frame, local->op_ret, local->op_errno, local->fd);
  }

  return 0;
}


/**
 * stripe_opendir - 
 */
int32_t
stripe_opendir (call_frame_t *frame,
		xlator_t *this,
		loc_t *loc,
		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 (loc);

  if (priv->first_child_down) {
    gf_log (this->name, GF_LOG_WARNING, "First node down, returning EIO");
    STACK_UNWIND (frame, -1, EIO, NULL);
    return 0;
  }

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

  while (trav) {
    STACK_WIND_COOKIE (frame,
		 stripe_opendir_cbk,
		 trav->xlator->name, //cookie
		 trav->xlator,
		 trav->xlator->fops->opendir,
		 loc,
		 fd);
    trav = trav->next;
  }
  
  return 0;
}


/**
 * stripe_getxattr_cbk - 
 */
int32_t
stripe_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;
}


/**
 * stripe_getxattr - 
 */
int32_t
stripe_getxattr (call_frame_t *frame,
		 xlator_t *this,
		 loc_t *loc)
{
  STRIPE_CHECK_INODE_CTX_AND_UNWIND_ON_ERR (loc);

  STACK_WIND (frame,
	      stripe_getxattr_cbk,
	      FIRST_CHILD(this),
	      FIRST_CHILD(this)->fops->getxattr,
	      loc);

  return 0;
}

/**
 * stripe_removexattr - 
 */
int32_t
stripe_removexattr (call_frame_t *frame,
		     xlator_t *this,
		     loc_t *loc,
		     const char *name)
{
  stripe_private_t *priv = this->private;

  STRIPE_CHECK_INODE_CTX_AND_UNWIND_ON_ERR (loc);

  if (priv->first_child_down) {
    gf_log (this->name, GF_LOG_WARNING, "First node down, returning ENOTCONN");
    STACK_UNWIND (frame, -1, ENOTCONN, NULL);
    return 0;
  }

  STACK_WIND (frame,
	      stripe_common_cbk,
	      FIRST_CHILD(this),
	      FIRST_CHILD(this)->fops->removexattr,
	      loc,
	      name);

  return 0;
}


/**
 * stripe_lk_cbk - 
 */
int32_t
stripe_lk_cbk (call_frame_t *frame,
	       void *cookie,
	       xlator_t *this,
	       int32_t op_ret,
	       int32_t op_errno,
	       struct flock *lock)
{
  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 == -1) {
      /* First successful call, copy the *lock */
      local->op_ret = 0;
      local->lock = *lock;
    }
  }
  UNLOCK (&frame->lock);

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


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

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

  return 0;