unify.c

分布式文件系统

    return 0;
  }
  dict_set (inode->ctx, this->name, data_from_ptr (local->list));

  local->list[0] = priv->child_count; //index of namespace in xl_array
  local->index = 1;
  local->call_count = priv->child_count;

  /* Send mkdir request to all the nodes now */
  for (index = 0; index < priv->child_count; index++) {
    loc_t tmp_loc = {
      .inode = inode,
      .path = local->name
    };
    STACK_WIND_COOKIE (frame,
		 unify_mkdir_cbk,
		 (void *)(long)index, //cookie
		 priv->xl_array[index],
		 priv->xl_array[index]->fops->mkdir,
		 &tmp_loc,
		 local->mode);
  }
  
  return 0;
}


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

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

  STACK_WIND (frame,
	      unify_ns_mkdir_cbk,
	      NS(this),
	      NS(this)->fops->mkdir,
	      loc,
	      mode);
  return 0;
}

/**
 * unify_rmdir_cbk -
 */
int32_t
unify_rmdir_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 = 0;
  }
  UNLOCK (&frame->lock);

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

  return 0;
}

/**
 * unify_ns_rmdir_cbk -
 */
int32_t
unify_ns_rmdir_cbk (call_frame_t *frame,
		    void *cookie,
		    xlator_t *this,
		    int32_t op_ret,
		    int32_t op_errno)
{
  int16_t index = 0;
  unify_private_t *priv = this->private;
  unify_local_t *local = frame->local;
  int32_t call_count = 0;
  
  if (op_ret == -1) {
     /* No need to send rmdir request to other servers, 
     * as namespace action failed 
     */
    gf_log (this->name, 
	    ((op_errno != 39) ? GF_LOG_ERROR : GF_LOG_DEBUG),
	    "rmdir on namespace failed (%d)", op_errno);
    unify_local_wipe (local);
    STACK_UNWIND (frame, op_ret, op_errno);
    return 0;
  }

  for (index = 0; local->list[index] != -1; index++) {
    if (NS(this) != priv->xl_array[local->list[index]]) {
      local->call_count++;
      call_count++;
    }
  }


  if (local->call_count) {
    for (index = 0; local->list[index] != -1; index++) {
      if (priv->xl_array[local->list[index]] != NS(this)) {
	loc_t tmp_loc = {
	  .path = local->path, 
	  .inode = local->inode
	};
	STACK_WIND (frame,
		    unify_rmdir_cbk,
		    priv->xl_array[local->list[index]],
		    priv->xl_array[local->list[index]]->fops->rmdir,
		    &tmp_loc);
	if (!--call_count)
	  break;
      }
    }
  } else {
    gf_log (this->name, GF_LOG_ERROR,
	    "rmdir sending ENOENT, as no directory found on storage nodes");
    unify_local_wipe (local);
    STACK_UNWIND (frame, -1, ENOENT);
  }

  /* */
  return 0;
}

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

  UNIFY_CHECK_INODE_CTX_AND_UNWIND_ON_ERR (loc);

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

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

  STACK_WIND (frame,
	      unify_ns_rmdir_cbk,
	      NS(this),
	      NS(this)->fops->rmdir,
	      loc);

  return 0;
}

int32_t 
unify_open_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;

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

/**
 * unify_open_cbk -
 */
int32_t
unify_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;
  unify_local_t *local = frame->local;

  LOCK (&frame->lock);
  {
    if (op_ret >= 0) {
      local->op_ret = op_ret;
      if (NS(this) != (xlator_t *)cookie) {
	/* Store child node's ptr, used in all the f*** / FileIO calls */
	dict_set (fd->ctx, this->name,
		  data_from_static_ptr (cookie));
      }
    }
    if (op_ret == -1) {
      local->op_errno = op_errno;
      local->failed = 1;
    }
    callcnt = --local->call_count;
  }
  UNLOCK (&frame->lock);
  
  if (!callcnt) {
    if ((local->failed == 1) && (local->op_ret >= 0)) {
      local->call_count = 1;
      /* return -1 to user */
      local->op_ret = -1;
      //local->op_errno = EIO; 
      
      if (dict_get (local->fd->ctx, this->name)) {
	xlator_t *child = data_to_ptr (dict_get (local->fd->ctx, this->name));
	gf_log (this->name, GF_LOG_ERROR, 
		"Open success on child node, failed on namespace");
	STACK_WIND (frame,
		    unify_open_close_cbk,
		    child,
		    child->fops->close,
		    local->fd);
      } else {
	gf_log (this->name, GF_LOG_ERROR, 
		"Open success on namespace, failed on child node");
	STACK_WIND (frame,
		    unify_open_close_cbk,
		    NS(this),
		    NS(this)->fops->close,
		    local->fd);
      }
      return 0;
    }

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

  return 0;
}

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

  UNIFY_CHECK_INODE_CTX_AND_UNWIND_ON_ERR (loc);

  /* Init */
  INIT_LOCAL (frame, local);
  local->inode = loc->inode;
  local->fd = fd;
  list = data_to_ptr (dict_get (loc->inode->ctx, this->name));
  local->list = list;
  for (index = 0; local->list[index] != -1; index++)
    local->call_count++;

  if (local->call_count != 2) {
    /* If the lookup was done for file */
    gf_log (this->name, GF_LOG_ERROR,
	    "%s: entry_count is %d",
	    loc->path, local->call_count);
    for (index = 0; local->list[index] != -1; index++)
      gf_log (this->name, GF_LOG_ERROR, "%s: found on %s",
	      loc->path, priv->xl_array[list[index]]->name);
    
    STACK_UNWIND (frame, -1, EIO, fd);
    return 0;
  }

  for (index = 0; list[index] != -1; index++) {
    char need_break = list[index+1] == -1;
    STACK_WIND_COOKIE (frame,
		 unify_open_cbk,
		 priv->xl_array[list[index]], //cookie
		 priv->xl_array[list[index]],
		 priv->xl_array[list[index]]->fops->open,
		 loc,
		 flags,
		 fd);
    if (need_break)
      break;
  }

  return 0;
}


int32_t 
unify_create_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;
  
  STACK_UNWIND (frame, local->op_ret, local->op_errno, local->fd, 
		local->inode, &local->stbuf);
  
  return 0;
}


int32_t 
unify_create_fail_cbk (call_frame_t *frame,
		       void *cookie,
		       xlator_t *this,
		       int32_t op_ret,
		       int32_t op_errno)
{
  unify_local_t *local = frame->local;
  
  /* Create failed in storage node, but it was success in 
   * namespace node, so after closing fd, need to unlink the file
   */
  loc_t tmp_loc = {
    .inode = local->inode,
    .path = local->name
  };
  local->call_count = 1;

  STACK_WIND (frame,
	      unify_create_close_cbk,
	      NS(this),
	      NS(this)->fops->unlink,
	      &tmp_loc);

  return 0;
}

/**
 * unify_create_open_cbk -
 */
int32_t
unify_create_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;
  unify_local_t *local = frame->local;
  call_frame_t *prev_frame = cookie;

  LOCK (&frame->lock);
  {
    if (op_ret >= 0) {
      local->op_ret = op_ret;
      if (NS(this) != (xlator_t *)cookie) {
	/* Store child node's ptr, used in all the f*** / FileIO calls */
	dict_set (fd->ctx, this->name, data_from_static_ptr (cookie));
      }
    } 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;
    }
    callcnt = --local->call_count;
  }
  UNLOCK (&frame->lock);
  
  if (!callcnt) {
    if (local->failed == 1 && (local->op_ret >= 0)) {
      local->call_count = 1;
      /* return -1 to user */
      local->op_ret = -1;
      local->op_errno = EIO;
      local->fd = fd;
      if (dict_get (local->fd->ctx, this->name)) {
	xlator_t *child = data_to_ptr (dict_get (local->fd->ctx, this->name));
	gf_log (this->name, GF_LOG_ERROR, 
		"Open success on child node, failed on namespace");
	STACK_WIND (frame,
		    unify_create_close_cbk,
		    child,
		    child->fops->close,
		    local->fd);
      } else {
	gf_log (this->name, GF_LOG_ERROR, 
		"Open success on namespace, failed on child node");
	STACK_WIND (frame,
		    unify_create_close_cbk,
		    NS(this),
		    NS(this)->fops->close,
		    local->fd);
      }
      return 0;
    }

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

/**
 * unify_create_lookup_cbk - 
 */
int32_t 
unify_create_lookup_cbk (call_frame_t *frame,
			 void *cookie,
			 xlator_t *this,
			 int32_t op_ret,
			 int32_t op_errno,
			 inode_t *inode,
			 struct stat *buf,
			 dict_t *dict)
{
  int32_t callcnt = 0;
  int16_t index = 0;
  unify_private_t *priv = this->private;
  unify_local_t *local = frame->local;

  LOCK (&frame->lock);
  {
    callcnt = --local->call_count;
    if (op_ret == -1) {
      gf_log (this->name, GF_LOG_ERROR,
	      "operation failed on %s (%d)", priv->xl_array[(long)cookie]->name, op_errno);
      local->op_errno = op_errno;
      local->failed = 1;
    }

    if (op_ret >= 0) {
      local->op_ret = op_ret; 
      local->list[local->index++] = (int16_t)(long)cookie;
      if (NS(this) == (xlator_t *)cookie) {
	local->st_ino = buf->st_ino;
      } else {
	local->stbuf = *buf;
      }
    }
  }
  UNLOCK (&frame->lock);

  if (!callcnt) {
    local->stbuf.st_ino = local->st_ino;
    local->list [local->index] = -1;
    dict_set (local->inode->ctx, this->name, 
	      data_from_ptr (local->list));

    if (local->index == 2) {
      /* Everything is perfect :) */
      int16_t *list = local->list;

      local->op_ret = -1;

      local->call_count = 2;
      
      for (index = 0; list[index] != -1; index++) {
	char need_break = list[index+1] == -1;
	loc_t tmp_loc = {
	  .inode = inode,
	  .path = local->name,
	};
	STACK_WIND_COOKIE (frame,
		     unify_create_open_cbk,
		     priv->xl_array[list[index]], //cookie
		     priv->xl_array[list[index]],
		     priv->xl_array[list[index]]->fops->open,
		     &tmp_loc,
		     local->flags,
		     local->fd);
	if (need_break)
	  break;
      }
    } else {
      /* Lookup failed, can't do open */
      gf_log (this->name, GF_LOG_ERROR,
	      "%s: entry_count is %d",
	      local->path, local->index);
      local->op_ret = -1;
      unify_local_wipe (local);
      STACK_UNWIND (frame, local->op_ret, local->op_errno, local->fd, 
		    local->inode, NULL);
    }
  }

  return 0;
}


/**
 * unify_create_cbk -
 */
int32_t
unify_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)
{
  unify_local_t *local = frame->local;
  call_frame_t *prev_frame = cookie;

  if (op_ret == -1) {
    /* send close () on Namespace */
    local->op_errno = op_errno;
    local->op_ret = -1;
    local->call_count = 1;
    gf_log (this->name, GF_LOG_ERROR,
	    "create failed on %s (%d), sending close to namespace", 
	    prev_frame->this->name, op_errno);

    STACK_WIND (frame,
		unify_create_fail_cbk,
		NS(this),
		NS(this)->fops->close,
		fd);

    return 0;
  }

  if (op_ret >= 0) {
    local->op_ret = op_ret;
    local->stbuf = *buf;
    /* Just inode number should be from NS node */
    local->stbuf.st_ino = local->st_ino;

    dict_set (fd->ctx, this->name,
	      data_from_static_ptr (prev_frame->this));
  }
  
  unify_local_wipe (local);
  STACK_UNWIND (frame, local->op_ret, local->op_errno, local->fd, 
		local->inode, &local->stbuf);

  return 0;
}

/**
 * unify_ns_create_cbk -
 * 
 */
int32_t
unify_ns_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)