unify.c

分布式文件系统

{
  struct sched_ops *sched_ops = NULL;
  xlator_t *sched_xl = NULL;
  unify_local_t *local = frame->local;
  unify_private_t *priv = this->private;
  int16_t *list = NULL;
  int16_t index = 0;

  if (op_ret == -1) {
    /* No need to send create request to other servers, as namespace 
     * action failed. Handle exclusive create here.
     */
    if ((op_errno != EEXIST) || 
	((op_errno == EEXIST) && ((local->flags & O_EXCL) == O_EXCL))) {
      /* If its just a create call without O_EXCL, don't do this */
      gf_log (this->name, GF_LOG_ERROR,
	      "create failed on namespace node (%d)", op_errno);
      unify_local_wipe (local);
      STACK_UNWIND (frame, op_ret, op_errno, fd, inode, buf);
      return 0;
    }
  }
  
  if (op_ret >= 0) {
    /* Get the inode number from the NS node */
    local->st_ino = buf->st_ino;
  
    local->op_ret = -1;

    /* Start the mapping list */
    list = calloc (1, sizeof (int16_t) * 3);
    dict_set (inode->ctx, this->name, data_from_ptr (list));
    list[0] = priv->child_count;
    list[2] = -1;

    /* This means, file doesn't exist anywhere in the Filesystem */
    sched_ops = priv->sched_ops;

    /* Send create request to the scheduled node now */
    sched_xl = sched_ops->schedule (this, local->name); 
    for (index = 0; index < priv->child_count; index++)
      if (sched_xl == priv->xl_array[index])
	break;
    list[1] = index;
    local->inode = inode;

    {
      loc_t tmp_loc = {
	.inode = inode,
	.path = local->name
      };
      STACK_WIND (frame, unify_create_cbk,
		  sched_xl, sched_xl->fops->create,
		  &tmp_loc, local->flags, local->mode, fd);
    }
  } else {
    /* File already exists, and there is no O_EXCL flag */

    gf_log (this->name, GF_LOG_DEBUG, 
	    "File(%s) already exists on namespace, sending open instead",
	    local->name);

    local->list = calloc (1, sizeof (int16_t) * 3);
    local->call_count = priv->child_count + 1;
    local->op_ret = -1;
    for (index = 0; index <= priv->child_count; index++) {
      /* Send the lookup to all the nodes including namespace */
      loc_t tmp_loc = {
	.path = local->name,
	.inode = inode,
      };
      STACK_WIND_COOKIE (frame,
		   unify_create_lookup_cbk,
		   (void *)(long)index,
		   priv->xl_array[index],
		   priv->xl_array[index]->fops->lookup,
		   &tmp_loc,
		   0);
    }
  }
  return 0;
}

/**
 * unify_create - create a file in global namespace first, so other 
 *    clients can see them. Create the file in storage nodes in background.
 */
int32_t
unify_create (call_frame_t *frame,
	      xlator_t *this,
	      loc_t *loc,
	      int32_t flags,
	      mode_t mode,
	      fd_t *fd)
{
  unify_local_t *local = NULL;
  
  /* Initialization */
  INIT_LOCAL (frame, local);
  local->mode = mode;
  local->flags = flags;
  local->inode = loc->inode;
  local->fd = fd;

  local->name = strdup (loc->path);
  if (!local->name) {
    gf_log (this->name, GF_LOG_CRITICAL, "Not enough memory :O");
    STACK_UNWIND (frame, -1, ENOMEM, fd, loc->inode, NULL);
    return 0;
  }

  STACK_WIND (frame,
	      unify_ns_create_cbk,
	      NS(this),
	      NS(this)->fops->create,
	      loc,
	      flags | O_EXCL,
	      mode,
	      fd);
  
  return 0;
}

int32_t 
unify_opendir_fail_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;
  }
  UNLOCK (&frame->lock);

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

  return 0;
}

/**
 * unify_opendir_cbk - 
 */
int32_t
unify_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;
  unify_local_t *local = frame->local;
  unify_private_t *priv = this->private;
  call_frame_t *prev_frame = cookie;

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

  if (!callcnt) {
    if (local->failed == 1 && dict_get (local->fd->inode->ctx, this->name)) {
      int16_t *list = NULL;
      int16_t index = 0;

      list = data_to_ptr (dict_get (local->fd->inode->ctx, this->name));
      /* return -1 to user */
      local->op_ret = -1;
      local->call_count =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_opendir_fail_cbk,
		    priv->xl_array[list[index]],
		    priv->xl_array[list[index]]->fops->closedir,
		    local->fd);
	if (need_break)
	  break;
      }
      return 0;
    }

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

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

  UNIFY_CHECK_INODE_CTX_AND_UNWIND_ON_ERR (loc);

  INIT_LOCAL (frame, local);
  local->inode = loc->inode;
  local->fd = fd;
  list = data_to_ptr (dict_get (loc->inode->ctx, this->name));

  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_opendir_cbk,
		priv->xl_array[list[index]],
		priv->xl_array[list[index]]->fops->opendir,
		loc, fd);
    if (need_break)
      break;
  }

  return 0;
}

/*
 * unify_normalize_stats -
 */
void
unify_normalize_stats (struct statvfs *buf,
		       unsigned long bsize,
		       unsigned long frsize)
{
  double factor;

  if (buf->f_bsize != bsize) {
    factor = ((double) buf->f_bsize) / bsize;
    buf->f_bsize  = bsize;
    buf->f_bfree  = (fsblkcnt_t) (factor * buf->f_bfree);
    buf->f_bavail = (fsblkcnt_t) (factor * buf->f_bavail);
  }
  
  if (buf->f_frsize != frsize) {
    factor = ((double) buf->f_frsize) / frsize;
    buf->f_frsize = frsize;
    buf->f_blocks = (fsblkcnt_t) (factor * buf->f_blocks);
  }
}

/**
 * unify_statfs_cbk -
 */
int32_t
unify_statfs_cbk (call_frame_t *frame,
		  void *cookie,
		  xlator_t *this,
		  int32_t op_ret,
		  int32_t op_errno,
		  struct statvfs *stbuf)
{
  int32_t callcnt = 0;
  struct statvfs *dict_buf = NULL;
  unsigned long bsize;
  unsigned long frsize;
  unify_local_t *local = (unify_local_t *)frame->local;
  call_frame_t *prev_frame = cookie;

  LOCK (&frame->lock);
  {
    if (op_ret >= 0) {
      /* when a call is successfull, add it to local->dict */
      dict_buf = &local->statvfs_buf;

      if (dict_buf->f_bsize != 0) {
	bsize  = max (dict_buf->f_bsize, stbuf->f_bsize);
	frsize = max (dict_buf->f_frsize, stbuf->f_frsize);
	unify_normalize_stats(dict_buf, bsize, frsize);
	unify_normalize_stats(stbuf, bsize, frsize);
      } else {
	dict_buf->f_bsize   = stbuf->f_bsize;
	dict_buf->f_frsize  = stbuf->f_frsize;
      }
      
      dict_buf->f_blocks += stbuf->f_blocks;
      dict_buf->f_bfree  += stbuf->f_bfree;
      dict_buf->f_bavail += stbuf->f_bavail;
      dict_buf->f_files  += stbuf->f_files;
      dict_buf->f_ffree  += stbuf->f_ffree;
      dict_buf->f_favail += stbuf->f_favail;
      dict_buf->f_fsid    = stbuf->f_fsid;
      dict_buf->f_flag    = stbuf->f_flag;
      dict_buf->f_namemax = stbuf->f_namemax;
      local->op_ret = op_ret;
    } else {
      /* fop on a storage node has failed due to some error */
      gf_log (this->name, GF_LOG_ERROR, 
	      "operation failed on %s  (%d)", prev_frame->this->name, op_errno);
      local->op_errno = op_errno;
    }
    
    callcnt = --local->call_count;
  }
  UNLOCK (&frame->lock);

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

  return 0;
}

/**
 * unify_statfs -
 */
int32_t
unify_statfs (call_frame_t *frame,
	      xlator_t *this,
	      loc_t *loc)
{
  unify_local_t *local = NULL;
  xlator_list_t *trav = this->children;

  INIT_LOCAL (frame, local);
  local->call_count = ((unify_private_t *)this->private)->child_count;

  while(trav) {
    STACK_WIND (frame,
		unify_statfs_cbk,
		trav->xlator,
		trav->xlator->fops->statfs,
		loc);
    trav = trav->next;
  }

  return 0;
}

/**
 * unify_ns_chmod_cbk - 
 */
int32_t
unify_ns_chmod_cbk (call_frame_t *frame,
		    void *cookie,
		    xlator_t *this,
		    int32_t op_ret,
		    int32_t op_errno,
		    struct stat *buf)
{
  /*  call_frame_t *bg_frame = NULL; */
  unify_local_t *local = frame->local;
  unify_private_t *priv = this->private;
  int16_t *list = local->list;
  int16_t index = 0;
  int32_t call_count = 0;

  if (op_ret == -1) {
    /* No need to send chmod request to other servers, 
     * as namespace action failed 
     */
    gf_log (this->name, GF_LOG_ERROR, 
	    "fop failed on namespace (%d)", op_errno);
    unify_local_wipe (local);
    STACK_UNWIND (frame, op_ret, op_errno, buf);
    return 0;
  }
  
  local->op_ret = op_ret;
  local->st_ino = buf->st_ino;
  local->op_errno = op_errno;
  local->stbuf = *buf;
  
  for (index = 0; local->list[index] != -1; index++) {
    if (NS(this) != priv->xl_array[local->list[index]]) {
      local->call_count++;
      call_count++;
    }
  }

  /* Send chmod request to all the nodes now */
  if (local->call_count) {
    for (index = 0; list[index] != -1; index++) {
      if (priv->xl_array[list[index]] != NS(this)) {
	loc_t tmp_loc = {
	  .inode = local->inode,
	  .path = local->path,
	};
	STACK_WIND (frame,
		    unify_buf_cbk,
		    priv->xl_array[list[index]],
		    priv->xl_array[list[index]]->fops->chmod,
		    &tmp_loc,
		    local->mode);
	if (!--call_count)
	  break;
      }
    }
    return 0;
  } 

  unify_local_wipe (local);
  STACK_UNWIND (frame, 0, 0, &local->stbuf);

  return 0;
}

/**
 * unify_chmod - 
 */
int32_t
unify_chmod (call_frame_t *frame,
	     xlator_t *this,
	     loc_t *loc,
	     mode_t mode)
{
  unify_local_t *local = NULL;

  UNIFY_CHECK_INODE_CTX_AND_UNWIND_ON_ERR (loc);

  /* Initialization */
  INIT_LOCAL (frame, local);
  local->inode = loc->inode;
  local->mode = mode;
  local->path = strdup (loc->path);
  if (!local->path) {
    gf_log (this->name, GF_LOG_CRITICAL, "Not enough memory :O");
    STACK_UNWIND (frame, -1, ENOMEM, NULL);
    return 0;
  }

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

  STACK_WIND (frame,
	      unify_ns_chmod_cbk,
	      NS(this),
	      NS(this)->fops->chmod,
	      loc,
	      mode);

  return 0;
}

/**
 * unify_ns_chown_cbk - 
 */
int32_t
unify_ns_chown_cbk (call_frame_t *frame,
		    void *cookie,
		    xlator_t *this,
		    int32_t op_ret,
		    int32_t op_errno,
		    struct stat *buf)
{
  /*  call_frame_t *bg_frame = NULL; */
  unify_local_t *local = frame->local;
  unify_private_t *priv = this->private;
  int16_t *list = local->list;
  int16_t index = 0;
  int32_t call_count = 0;

  if (op_ret == -1) {
    /* No need to send chown request to other servers, as namespace action 
     * failed 
     */
    gf_log (this->name, GF_LOG_ERROR, 
	    "fop failed on namespace (%d)", op_errno);
    unify_local_wipe (local);
    STACK_UNWIND (frame, op_ret, op_errno, buf);
    return 0;
  }
  
  local->op_ret = op_ret;
  local->st_ino = buf->st_ino;
  local->op_errno = op_errno;
  local->stbuf = *buf;
  
  local->call_count = 0;
  for (index = 0; list[index] != -1; index++) {
    if (NS(this) != priv->xl_array[list[index]]) {
      local->call_count++;
      call_count++;
    }
  }
  
  if (local->call_count) {
    /* Send chown request to all the nodes now */
    for (index = 0; list[index] != -1; index++) {
      if (priv->xl_array[list[index]] != NS(this)) {
	loc_t tmp_loc = {
	  .inode = local->inode,
	  .path = local->path,
	};
	STACK_WIND (frame,
		    unify_buf_cbk,
		    priv->xl_array[list[index]],
		    priv->xl_array[list[index]]->fops->chown,
		    &tmp_loc,
		    local->uid,
		    local->gid);
	if (!--call_count)
	  break;
      }
    }
    return 0;
  }

  unify_local_wipe (local);
  STACK_UNWIND (frame, 0, 0, &local->stbuf);

  return 0;
}

/**
 * unify_chown - 
 */
int32_t
unify_chown (call_frame_t *frame,
	     xlator_t *this,
	     loc_t *loc,
	     uid_t uid,
	     gid_t gid)
{
  unify_local_t *local = NULL;
  
  UNIFY_CHECK_INODE_CTX_AND_UNWIND_ON_ERR (loc);

  /* Initialization */
  INIT_LOCAL (frame, local);
  local->inode = loc->inode;
  local->uid = uid;
  local->gid = gid;
  local->path = strdup (loc->path);
  if (!local->path) {
    gf_log (this->name, GF_LOG_CRITICAL, "Not enough memory :O");
    STACK_UNWIND (frame, -1, ENOMEM, NULL);
    return 0;
  }

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

  STACK_WIND (frame,
	      unify_ns_chown_cbk,
	      NS(this),
	      NS(this)->fops->chown,
	      loc,
	      uid,
	      gid);

  return 0;
}

/**
 * unify_ns_truncate_cbk - 
 */
int32_t
unify_ns_truncate_cbk (call_frame_t *frame,
		       void *cookie,
		       xlator_t *this,
		       int32_t op_ret,
		       int32_t op_errno,