afr.c

分布式文件系统

	for (i = 0; i < child_count; i++) {
	  if (afrfdp->fdstate[i]) {
	    /* op_ret != -1 implies atleast one increment */
	    alive_children++;
	  }
	}
	rchild = local->loc->inode->ino % alive_children;
	/* read schedule among alive children */
	for (i = 0; i < child_count; i++) {
	  if (afrfdp->fdstate[i] == 1) {
	    if (rchild == 0)
	      break;
	    rchild--;
	  }
	}
	afrfdp->rchild = i;
      } else {
	afrfdp->rchild = pvt->read_node;
      }
    }

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

int32_t
afr_open (call_frame_t *frame,
	  xlator_t *this,
	  loc_t *loc,
	  int32_t flags,
	  fd_t *fd)
{
  int32_t i = 0;
  afr_local_t *local; 
  char *afr_errno = NULL;
  char *child_errno = NULL;
  afr_private_t *pvt = this->private;
  xlator_t **children = pvt->children;
  int32_t child_count = pvt->child_count;

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

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

  if (afr_errno == NULL) {
    if (loc->inode->ctx == NULL) {
      GF_ERROR (this, "inode->ctx is NULL, returning EIO");
    }
    GF_ERROR (this, "afr_errno is NULL, returning EIO");
    STACK_UNWIND (frame, -1, EIO, fd);
    return 0;
  }

  AFR_ERRNO_DUP(child_errno, afr_errno, child_count);

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

  if (i == child_count) {
    GF_ERROR (this, "%s: child_errno[] is not 0, returning ENOTCONN", loc->path);
    STACK_UNWIND (frame, -1, ENOTCONN, NULL);
    return 0;
  }

  if (frame->local == NULL) {
    frame->local = (void *) calloc (1, sizeof (afr_local_t));
  }
  local = frame->local;
  
  if (((afr_private_t *) this->private)->self_heal) {
    AFR_DEBUG_FMT (this, "self heal enabled");
    if (local->sh_return_error) {
      GF_ERROR (this, "self heal failed, returning EIO");
      STACK_UNWIND (frame, -1, EIO, fd);
      return 0;
    }
    if (local->shcalled == 0) {
      call_stub_t *stub = fop_open_stub (frame, afr_open, loc, flags, fd);
      AFR_DEBUG_FMT (this, "self heal checking...");
      afr_selfheal (frame, this, stub, loc);
      return 0;
    }
    AFR_DEBUG_FMT (this, "self heal already called");
  } else {
    AFR_DEBUG_FMT (this, "self heal disabled");
  }

  local->op_ret = -1;
  local->op_errno = ENOTCONN;
  local->flags = flags;
  local->loc = afr_loc_dup (loc);
  local->fd = fd;
  for (i = 0; i < child_count; i++) {
    if (child_errno[i] == 0)
      ++local->call_count;
  }

  for (i = 0; i < child_count; i++) {
    if (child_errno[i] == 0)
      STACK_WIND (frame,
		  afr_open_cbk,
		  children[i],
		  children[i]->fops->open,
		  loc,
		  flags,
		  fd);
  }

  return 0;
}

int32_t
afr_readv_cbk (call_frame_t *frame,
	       void *cookie,
	       xlator_t *this,
	       int32_t op_ret,
	       int32_t op_errno,
	       struct iovec *vector,
	       int32_t count,
	       struct stat *stat)
{
  afr_local_t *local = (afr_local_t *)frame->local;

  AFR_DEBUG(this);

  if (op_ret == -1) {
    call_frame_t *prev_frame = cookie;
    afrfd_t *afrfdp = local->afrfdp;
    if (op_errno == ENOTCONN || op_errno == EBADFD) {
      int i=0;
      afr_private_t *pvt = this->private;
      xlator_t **children = pvt->children;

      for (i = 0; i < pvt->child_count; i++)
	if (((call_frame_t *)cookie)->this == children[i])
	  break;

      afrfdp->fdstate[i] = 0;
      afrfdp->rchild = -1;
      for (i = 0; i < pvt->child_count; i++) {
	if (afrfdp->fdstate[i])
	  break;
      }
      GF_DEBUG (this, "reading from child %d", i);
      if (i < pvt->child_count) {
      	STACK_WIND (frame,
		    afr_readv_cbk,
		    children[i],
		    children[i]->fops->readv,
		    local->fd,
		    local->size,
		    local->offset);
	return 0;
      }
    }

    GF_ERROR (this, "(path=%s child=%s) op_ret=%d op_errno=%d", 
	      afrfdp->path, prev_frame->this->name, op_ret, op_errno);
  }

  STACK_UNWIND (frame, op_ret, op_errno, vector, count, stat);
  return 0;
}


int32_t
afr_readv (call_frame_t *frame,
	   xlator_t *this,
	   fd_t *fd,
	   size_t size,
	   off_t offset)
{
  int32_t i = 0;
  afrfd_t *afrfdp = NULL;
  afr_local_t *local = NULL;
  afr_private_t *pvt = this->private;
  xlator_t **children = pvt->children;
  int32_t child_count = pvt->child_count;

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

  AFR_DEBUG_FMT(this, "fd %p", fd);

  if (afrfdp == NULL) {
    GF_ERROR (this, "afrfdp is NULL, returning EBADFD");
    STACK_UNWIND (frame, -1, EBADFD, NULL, 0, NULL);
    return 0;
  }

  local = frame->local = calloc (1, sizeof (afr_local_t));
  local->afrfdp = afrfdp;
  local->offset = offset;
  local->size = size;
  local->fd = fd;

  i = afrfdp->rchild;
  if (i == -1 || afrfdp->fdstate[i] == 0) {
    for (i = 0; i < child_count; i++) {
      if (afrfdp->fdstate[i] && pvt->state[i])
	break;
    }
  }
  if (i == child_count) {
    STACK_UNWIND (frame, -1, ENOTCONN, NULL, 0, NULL);
  } else {
    STACK_WIND (frame,
		afr_readv_cbk,
		children[i],
		children[i]->fops->readv,
		fd,
		size,
		offset);
  }

  return 0;
}

/* FIXME if one write fails, we should not increment version on 
 * the other subvols
 */

int32_t
afr_writev_cbk (call_frame_t *frame,
		void *cookie,
		xlator_t *this,
		int32_t op_ret,
		int32_t op_errno,
		struct stat *stat)
{
  int32_t callcnt = 0;
  afr_local_t *local = frame->local;
  call_frame_t *prev_frame = cookie;
  call_frame_t *orig_frame = NULL;

  AFR_DEBUG_FMT(this, "op_ret %d op_errno %d", op_ret, op_errno);

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

    if (local->orig_frame && (op_ret >= 0 || !callcnt)) {
      orig_frame = local->orig_frame;
      local->orig_frame = NULL;
    }
  }
  UNLOCK (&frame->lock);


  if (op_ret == -1) {
    afr_private_t *pvt = this->private;
    int32_t child_count = pvt->child_count, i;
    xlator_t **children = pvt->children;
    afrfd_t *afrfdp = NULL;

    afrfdp = data_to_ptr (dict_get (local->fd->ctx, this->name));
    for (i = 0; i < child_count; i++)
      if (prev_frame->this == children[i])
	break;

    afrfdp->fdstate[i] = 0;

    GF_ERROR (this, "(path=%s child=%s) op_ret=%d op_errno=%d", 
	      afrfdp->path, prev_frame->this->name, op_ret, op_errno);
  }

  if (orig_frame) {
    STACK_UNWIND (orig_frame, op_ret, op_errno, stat);
  }

  if (!callcnt) {
    dict_unref (frame->root->req_refs);
    STACK_DESTROY (frame->root);
  }

  return 0;
}

int32_t
afr_writev (call_frame_t *orig_frame,
	    xlator_t *this,
	    fd_t *fd,
	    struct iovec *vector,
	    int32_t count,
	    off_t offset)
{
  afr_local_t *local = (void *) calloc (1, sizeof (afr_local_t));
  afr_private_t *pvt = this->private;
  xlator_t **children = pvt->children;
  int32_t child_count = pvt->child_count, i;
  afrfd_t *afrfdp = NULL;
  call_frame_t *frame;


  AFR_DEBUG_FMT(this, "fd %p", fd);

  afrfdp = data_to_ptr (dict_get (fd->ctx, this->name));
  if (afrfdp == NULL) {
    freee (local);
    GF_ERROR (this, "afrfdp is NULL, returning EBADFD");
    STACK_UNWIND (orig_frame, -1, EBADFD, NULL);
    return 0;
  }

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

  if (local->call_count == 0) {
    GF_ERROR (this, "afrfdp->fdstate[] is 0, returning ENOTCONN");
    STACK_UNWIND (orig_frame, -1, ENOTCONN, NULL);
    return 0;
  }

  frame = copy_frame (orig_frame);
  frame->local = local;
  frame->root->req_refs = dict_ref (orig_frame->root->req_refs);

  local->op_ret = -1;
  local->op_errno = ENOTCONN;
  local->fd = fd;
  local->orig_frame = orig_frame;
  afrfdp->write = 1;

  for (i = 0; i < child_count; i++) {
    if (afrfdp->fdstate[i]) {
      STACK_WIND (frame,
		  afr_writev_cbk,
		  children[i],
		  children[i]->fops->writev,
		  fd,
		  vector,
		  count,
		  offset);
    }
  }

  return 0;
}


int32_t
afr_ftruncate_cbk (call_frame_t *frame,
		   void *cookie,
		   xlator_t *this,
		   int32_t op_ret,
		   int32_t op_errno,
		   struct stat *stbuf)
{
  int32_t callcnt = 0;
  afr_local_t *local = frame->local;
  call_frame_t *prev_frame = cookie;

  AFR_DEBUG(this);

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

  if (op_ret == -1) {
    afrfd_t *afrfdp = data_to_ptr (dict_get (local->fd->ctx, this->name));
    GF_ERROR (this, "(path=%s child=%s) op_ret=%d op_errno=%d", 
	      afrfdp->path, prev_frame->this->name, op_ret, op_errno);
  }
  if (op_ret == 0)
    GF_BUG_ON (!stbuf);

  LOCK (&frame->lock);
  {
    if (op_ret == 0 && local->op_ret == -1) {
      local->op_ret = op_ret;
      local->stbuf = *stbuf;
    }
    callcnt = --local->call_count;
  }
  UNLOCK (&frame->lock);

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


int32_t
afr_ftruncate (call_frame_t *frame,
	       xlator_t *this,
	       fd_t *fd,
	       off_t offset)
{
  afr_local_t *local = (void *) calloc (1, sizeof (afr_local_t));
  afr_private_t *pvt = this->private;
  xlator_t **children = pvt->children;
  int32_t child_count = pvt->child_count, i;
  afrfd_t *afrfdp = NULL;

  AFR_DEBUG_FMT(this, "fd %p", fd);

  afrfdp = data_to_ptr (dict_get (fd->ctx, this->name));
  if (afrfdp == NULL) {
    free (local);
    GF_ERROR (this, "afrfdp is NULL, returning EBADFD");
    STACK_UNWIND (frame, -1, EBADFD, NULL);
    return 0;
  }

  frame->local = local;
  local->op_ret = -1;
  local->op_errno = ENOTCONN;
  local->fd = fd;
  afrfdp->write = 1;
  for (i = 0; i < child_count; i++) {
    if (afrfdp->fdstate[i])
      ++local->call_count;
  }
  if (local->call_count == 0) {
    GF_ERROR (this, "afrfdp->fdstate[] is 0, returning ENOTCONN");
    STACK_UNWIND (frame, -1, ENOTCONN, NULL);
    return 0;
  }

  for ( i = 0; i < child_count; i++) {
    if (afrfdp->fdstate[i]) {
      STACK_WIND(frame,
		 afr_ftruncate_cbk,
		 children[i],
		 children[i]->fops->ftruncate,
		 fd,
		 offset);
    }
  }

  return 0;
}

int32_t
afr_fstat_cbk (call_frame_t *frame,
	       void *cookie,
	       xlator_t *this,
	       int32_t op_ret,
	       int32_t op_errno,
	       struct stat *stbuf)
{

  AFR_DEBUG(this);

  if (op_ret == -1) {
    afrfd_t *afrfdp = frame->local;
    call_frame_t *prev_frame = cookie;

    GF_ERROR (this, "(path=%s child=%s) op_ret=%d op_errno=%d", 
	      afrfdp->path, prev_frame->this->name, op_ret, op_errno);
  }

  frame->local = NULL; /* so that STACK_UNWIND does not try to free */
  STACK_UNWIND (frame, op_ret, op_errno, stbuf);
  return 0;
}

int32_t
afr_fstat (call_frame_t *frame,
	   xlator_t *this,
	   fd_t *fd)
{
  afr_private_t *pvt = this->private;
  xlator_t **children = pvt->children;
  int32_t child_count = pvt->child_count, i;
  afrfd_t *afrfdp = NULL;

  AFR_DEBUG(this);

  afrfdp = data_to_ptr (dict_get (fd->ctx, this->name));
  if (afrfdp == NULL) {
    GF_ERROR (this, "afrfdp is NULL, returning EBADFD");
    STACK_UNWIND (frame, -1, EBADFD, NULL);
    return 0;
  }

  frame->local = afrfdp;
  for (i = 0; i < child_count; i++) {
    if (afrfdp->fdstate[i])
      break;
  }

  if (i == child_count) {
    GF_ERROR (this, "afrfdp->fdstate[i] is 0, returning ENOTCONN");
    STACK_UNWIND (frame, -1, ENOTCONN, NULL);
    return 0;
  }

  STACK_WIND (frame,
	      afr_fstat_cbk,
	      children[i],
	      children[i]->fops->fstat,
	      fd);

  return 0;
}

int32_t
afr_flush_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;
  call_frame_t *prev_frame = cookie;

  AFR_DEBUG(this);

  if (op_ret != 0 && op_errno != ENOTCONN) {
    local->op_errno = op_errno;
  }
  if (op_ret == -1) {
    afrfd_t *afrfdp = data_to_ptr (dict_get (local->fd->