afr.c

分布式文件系统

		int32_t op_ret,
		int32_t op_errno,
		inode_t *inode,
		struct stat *buf,
		dict_t *xattr)
{
  char *child_errno = NULL;
  data_t *errno_data = NULL;
  int32_t callcnt, i, latest = -1, first = -1;
  afr_local_t *local = frame->local;
  afr_private_t *pvt = this->private;
  xlator_t **children = pvt->children;
  int32_t child_count = pvt->child_count;
  call_frame_t *prev_frame = cookie;
  struct stat *statptr = local->statptr;
  afr_selfheal_t *ashptr = local->ashptr;

  AFR_DEBUG_FMT(this, "op_ret = %d op_errno = %d, inode = %p, returned from %s", 
		op_ret, op_errno, inode, prev_frame->this->name);

  if (op_ret != 0 && op_errno != ENOTCONN)
    local->op_errno = op_errno;

  /* 'i' will be the index indicating us, which child node has returned to us */
  for (i = 0; i < child_count; i++)
    if (children[i] == prev_frame->this)
      break;

  /* child_errno is an array of one bytes, each byte corresponding to the errno returned by a child node
   * child_errno array is initialized during the first succesful return call from a child.
   */
  errno_data = dict_get (local->loc->inode->ctx, this->name);
  if (errno_data)
    child_errno = data_to_ptr (errno_data);
  if (child_errno == NULL) {
    /* first time lookup and success */
    child_errno = calloc (child_count, sizeof (char));	
    dict_set (local->loc->inode->ctx, this->name, data_from_dynptr (child_errno, child_count));
  }

  /* child_errno[i] is either 0 indicating success or op_errno indicating failure */
  if (op_ret == 0) {
    data_t *ctime_data, *version_data;
    local->op_ret = 0;
    if (inode && list_empty (&inode->fds)) {
      child_errno[i] = 0;
    }

    GF_BUG_ON (!inode);
    GF_BUG_ON (!buf);

    statptr[i] = *buf;
    if (pvt->self_heal && xattr) {
      /* self heal is 'on' and we also recieved the xattr that we requested from our children.
       * store ctime and version returned by each child */
      ctime_data = dict_get (xattr, GLUSTERFS_CREATETIME);
      if (ctime_data) {
	ashptr[i].ctime = data_to_uint32 (ctime_data);
      }

      version_data = dict_get (xattr, GLUSTERFS_VERSION);
      if (version_data) {
	ashptr[i].version = data_to_uint32 (version_data);
      }

      if (ashptr[i].ctime > local->latest_ctime || 
	  (ashptr[i].ctime == local->latest_ctime && ashptr[i].version > local->latest_version)) {
	local->latest_ctime = ashptr[i].ctime;
	local->latest_version = ashptr[i].version;
	if (local->latest_xattr)
	  dict_unref (local->latest_xattr);
	local->latest_xattr = dict_ref (xattr);
      }

      AFR_DEBUG_FMT (this, "child %s ctime %d version %d", 
		     prev_frame->this->name, ashptr[i].ctime, ashptr[i].version);
    }
  } else if (inode && list_empty (&inode->fds)) {
    /* either self-heal is turned 'off' or we didn't recieve xattr, which we requested for */
    child_errno[i] = op_errno;
  }

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

  if (callcnt == 0) {
    if (local->op_ret == 0) {
      if (pvt->self_heal) {
	for (i = 0; i < child_count; i++) {
	  if (child_errno[i] == 0)
	    break;
	}
	if (i < child_count) {
	  afr_check_ctime_version (frame);
	  return 0;
	}
      }
    }
    /* child_errno will be freed when dict is destroyed */

    if (local->op_ret == 0) {
      for (i = 0; i < child_count; i++) {
	if (child_errno[i] == 0) {
	  if (latest == -1) {
	    /* first will be the first valid stat, latest will be the stat with latest mtime */
	    first = i;
	    latest = i;
	    continue;
	  }
	  /* FIXME use ctime/version 
	  if (statptr[i].ctime > statptr[latest].ctime ||
	      (statptr[i].ctime == statptr[latest].ctime && statptr[i].version > statptr[latest]))
	  */
	  if (statptr[i].st_mtime > statptr[latest].st_mtime)
	    latest = i;
	}
      }
    }
    if (first == -1) {
      first = latest = 0;
    } else {
      /* FIXME: we preserve the ino num (whatever that was got during the initial lookup(?) */
      if (local->ino)
	statptr[latest].st_ino = local->ino;
      else
	statptr[latest].st_ino = statptr[first].st_ino;
    }
    afr_loc_free(local->loc);
    afr_free_ashptr (local->ashptr, child_count, local->latest);
    STACK_UNWIND (frame,
		  local->op_ret,
		  local->op_errno,
		  inode,
		  &statptr[latest],
		  xattr);
    freee (statptr);
  }
  return 0;
}

int32_t
afr_lookup (call_frame_t *frame,
	    xlator_t *this,
	    loc_t *loc,
	    int32_t need_xattr)
{
  afr_local_t *local = calloc (1, sizeof (*local));
  afr_private_t *pvt = this->private;
  xlator_t **children = pvt->children;
  int32_t child_count = pvt->child_count, i;

  AFR_DEBUG_FMT (this, "loc->path = %s loc->inode = %p", loc->path, loc->inode);

  frame->local = local;
  local->op_ret = -1;
  local->op_errno = ENOTCONN;
  local->loc = afr_loc_dup (loc);

  /* statptr[] array is used for selfheal */
  local->statptr = calloc (child_count, sizeof (struct stat));
  local->ashptr  = calloc (child_count, sizeof (afr_selfheal_t));
  local->call_count = child_count;
  local->ino = loc->ino;

  for (i = 0; i < child_count; i ++) {
    /* request for extended attributes if self heal is 'on' */
    int32_t need_xattr = pvt->self_heal;
    STACK_WIND (frame,
		afr_lookup_cbk,
		children[i],
		children[i]->fops->lookup,
		loc,
		need_xattr);
  }
  return 0;
}

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

  if (op_ret > local->op_ret)
    local->op_ret = op_ret;

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

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

int32_t
afr_incver (call_frame_t *frame,
	    xlator_t *this,
	    const char *path)
{
  afr_local_t *local = calloc (1, sizeof (afr_local_t));;
  afr_private_t *pvt = frame->this->private;
  int32_t child_count = pvt->child_count, i;
  xlator_t **children = pvt->children;
  char *state = pvt->state;

  frame->local = local;
  local->op_ret = -1;
  for (i = 0; i < child_count; i++) {
    if (state[i])
      local->call_count++;
  }

  if (local->call_count == 0) {
    GF_ERROR (this, "all children are down, returning ENOTCONN");
    STACK_UNWIND (frame, -1, ENOTCONN);
    return 0;
  }

  for (i = 0; i < child_count; i++) {
    if (state[i]) {
      STACK_WIND (frame,
		  afr_incver_cbk,
		  children[i],
		  children[i]->fops->incver,
		  path);
    }
  }
  return 0;
}

int32_t
afr_incver_internal_unlock_cbk (call_frame_t *frame,
				void *cookie,
				xlator_t *this,
				int32_t op_ret,
				int32_t op_errno)
{
  afr_local_t *local = frame->local;

  freee (local->path);
  STACK_DESTROY (frame->root);
  return 0;
}

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

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

  if (callcnt == 0) {
    char *lock_path = NULL;
    asprintf (&lock_path, "/%s%s", local->lock_node->name, local->path);
    STACK_WIND (frame,
		afr_incver_internal_unlock_cbk,
		local->lock_node,
		local->lock_node->mops->unlock,
		lock_path);
    freee (lock_path);
  }
  return 0;
}

int32_t
afr_incver_internal_lock_cbk (call_frame_t *frame,
			      void *cookie,
			      xlator_t *this,
			      int32_t op_ret,
			      int32_t op_errno)
{
  afr_local_t *local = frame->local;
  afr_private_t *pvt = frame->this->private;
  int32_t child_count = pvt->child_count, i;
  xlator_t **children = pvt->children;
  char *state = pvt->state;

  for (i = 0; i < child_count; i++) {
    if (state[i])
      local->call_count++;
  }

  /* handle call_count == 0 */
  for (i = 0; i < child_count; i++) {
    if (state[i]) {
      STACK_WIND (frame,
		  afr_incver_internal_incver_cbk,
		  children[i],
		  children[i]->fops->incver,
		  local->path);
    }
  }
  return 0;
}

int32_t
afr_incver_internal (call_frame_t *frame,
		     xlator_t *this,
		     const char *path)
{
  call_frame_t *incver_frame;
  afr_local_t *local;
  afr_private_t *pvt = frame->this->private;
  int32_t child_count = pvt->child_count, i, call_count = 0;
  xlator_t **children = pvt->children;
  char *state = pvt->state;
  char *lock_path = NULL;

  if (pvt->self_heal == 0)
    return 0;

  for (i = 0; i < child_count; i++) {
    if (state[i])
      call_count++;
  }
  /* we wont incver if all children are down or if all children are up */
  if (call_count == 0 || call_count == child_count) {
    return 0;
  }

  for (i = 0; i < child_count; i++) {
    if (state[i])
      break;
  }

  local = calloc (1, sizeof (afr_local_t));
  incver_frame = copy_frame (frame);
  incver_frame->local = local;

  local->lock_node = children[i];
  local->path = dirname (strdup(path));

  asprintf (&lock_path, "/%s%s", local->lock_node->name, local->path);
  STACK_WIND (incver_frame,
	      afr_incver_internal_lock_cbk,
	      local->lock_node,
	      local->lock_node->mops->lock,
	      lock_path);
  freee (lock_path);

  return 0;
}


/* no need to do anything in forget, as the mem will be just free'd in dict_destroy(inode->ctx) */

int32_t
afr_forget (call_frame_t *frame,
	    xlator_t *this,
	    inode_t *inode)
{
  return 0;
}

int32_t
afr_setxattr_cbk (call_frame_t *frame,
		  void *cookie,
		  xlator_t *this,
		  int32_t op_ret,
		  int32_t op_errno)
{
  afr_local_t *local = (afr_local_t *) frame->local;
  call_frame_t *prev_frame = cookie;
  int32_t callcnt;

  AFR_DEBUG(this);

  if (op_ret != 0  && op_errno != ENOTCONN) {
    local->op_errno = op_errno;
  }
  if (op_ret == 0) {
    local->op_ret = op_ret;
  } else {
    GF_ERROR (this, "(path=%s child=%s) op_ret=%d op_errno=%d", 
	      local->loc->path, prev_frame->this->name, op_ret, op_errno);
  }

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

  if (callcnt == 0) {
    afr_loc_free (local->loc);
    STACK_UNWIND (frame, local->op_ret, local->op_errno);
  }
  return 0;
}

int32_t
afr_setxattr (call_frame_t *frame,
	      xlator_t *this,
	      loc_t *loc,
	      dict_t *dict,
	      int32_t flags)
{
  afr_local_t *local = (void *) calloc (1, sizeof (afr_local_t));
  afr_private_t *pvt = this->private;
  char *afr_errno = NULL;
  xlator_t **children = pvt->children;
  int32_t child_count = pvt->child_count, i;
  char *child_errno = NULL;

  AFR_DEBUG_FMT (this, "loc->path = %s", loc->path);

  afr_errno = data_to_ptr (dict_get (loc->inode->ctx, this->name));
 
  AFR_ERRNO_DUP(child_errno, afr_errno, child_count);

  frame->local = local;
  local->op_ret = -1;
  local->op_errno = ENOTCONN;
  local->loc = afr_loc_dup (loc);

  for (i = 0; i < child_count; i++) {
    if (child_errno[i] == 0)
      ++local->call_count;
  }

  if (local->call_count == 0) {
    GF_ERROR (this, "child_errno[] is not 0, returning ENOTCONN");
    STACK_UNWIND (frame, -1, ENOTCONN);
    return 0;
  }

  for (i = 0; i < child_count; i++) {
    if (child_errno[i] == 0)
      STACK_WIND(frame,
		 afr_setxattr_cbk,
		 children[i],
		 children[i]->fops->setxattr,
		 loc,
		 dict,
		 flags);
  }
  return 0;
}

int32_t
afr_getxattr_cbk (call_frame_t *frame,
		  void *cookie,
		  xlator_t *this,
		  int32_t op_ret,
		  int32_t op_errno,
		  dict_t *dict)
{
  call_frame_t *prev_frame = cookie;

  AFR_DEBUG(this);

  if (op_ret >= 0) {
    GF_BUG_ON (!dict);
  } else if (op_errno != ENODATA) {
    GF_ERROR (this, "(path=%s child=%s) op_ret=%d op_errno=%d", 
	      frame->local, prev_frame->this->name, op_ret, op_errno);
  }

  STACK_UNWIND (frame, op_ret, op_errno, dict);
  return 0;
}

int32_t
afr_getxattr (call_frame_t *frame,
	      xlator_t *this,
	      loc_t *loc)
{
  afr_private_t *pvt = this->private;
  char *afr_errno = NULL;
  xlator_t **children = pvt->children;
  int32_t child_count = pvt->child_count, i;
  char *child_errno = NULL;

  AFR_DEBUG_FMT (this, "loc->path = %s", loc->path);

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

  AFR_ERRNO_DUP(child_errno, afr_errno, child_count);

  frame->local = strdup (loc->path);
  for (i = 0; i < child_count; i++) {
    if (child_errno[i] == 0)
      break;
  }

  if (i == child_count) {
    STACK_UNWIND (frame, -1, ENOTCONN, NULL);
    return 0;
  }
  /* send getxattr command to the first child where the file is available */
  STACK_WIND (frame,
	      afr_getxattr_cbk,
	      children[i],
	      children[i]->fops->getxattr,
	      loc);
  return 0;
}

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