posix-locks.c

分布式文件系统

	      FIRST_CHILD(this), FIRST_CHILD(this)->fops->fstat, 
	      fd);
  return 0;
}

int32_t 
pl_close_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
	      int32_t op_ret, int32_t op_errno)
{
  GF_ERROR_IF_NULL (this);

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

int32_t 
pl_close (call_frame_t *frame, xlator_t *this,
	  fd_t *fd)
{
  GF_ERROR_IF_NULL (this);
  GF_ERROR_IF_NULL (fd);

  posix_locks_private_t *priv = (posix_locks_private_t *)this->private;
  pthread_mutex_lock (&priv->mutex);

  struct flock nulllock = {0, };
  data_t *fd_data = dict_get (fd->ctx, this->name);
  if (fd_data == NULL) {
    pthread_mutex_unlock (&priv->mutex);
    gf_log (this->name, GF_LOG_ERROR, "returning EBADF");
    STACK_UNWIND (frame, -1, EBADF, &nulllock);
    return 0;
  }
  pl_fd_t *pfd = (pl_fd_t *)data_to_bin (fd_data);

  data_t *inode_data = dict_get (fd->inode->ctx, this->name);
  if (inode_data == NULL) {
    pthread_mutex_unlock (&priv->mutex);
    gf_log (this->name, GF_LOG_ERROR, "returning EBADF");
    STACK_UNWIND (frame, -1, EBADF, &nulllock);
    return 0;
  }
  pl_inode_t *inode = (pl_inode_t *)data_to_bin (inode_data);

  dict_del (fd->ctx, this->name);

  do_blocked_rw (inode);
  grant_blocked_locks (inode);

  free (pfd);

  pthread_mutex_unlock (&priv->mutex);

  STACK_WIND (frame, pl_close_cbk, 
	      FIRST_CHILD(this), FIRST_CHILD(this)->fops->close, 
	      fd);
  return 0;
}

int32_t 
pl_flush_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
	      int32_t op_ret, int32_t op_errno)
{
  STACK_UNWIND (frame, op_ret, op_errno);
  return 0;
}

static void
delete_locks_of_owner (pl_inode_t *inode, transport_t *transport,
		       pid_t pid)
{
  posix_lock_t *l = inode->locks;
  while (l) {
    posix_lock_t *tmp = l;
    l = l->next;
    if ((tmp->transport == transport) && 
	(tmp->client_pid == pid)) {
      delete_lock (inode, tmp);
      destroy_lock (tmp);
    }
  }
}

int32_t 
pl_flush (call_frame_t *frame, xlator_t *this,
	  fd_t *fd)
{
  data_t *inode_data = dict_get (fd->inode->ctx, this->name);
  if (inode_data == NULL) {
    gf_log (this->name, GF_LOG_ERROR, "returning EBADF");
    STACK_UNWIND (frame, -1, EBADF);
    return 0;
  }
  pl_inode_t *inode = (pl_inode_t *)data_to_bin (inode_data);

  delete_locks_of_owner (inode, frame->root->trans, frame->root->pid);
  do_blocked_rw (inode);
  grant_blocked_locks (inode);

  STACK_WIND (frame, pl_flush_cbk, 
	      FIRST_CHILD(this), FIRST_CHILD(this)->fops->flush, 
	      fd);
  return 0;
}

struct _flags {
  int32_t flags;
};

int32_t 
pl_open_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
	     int32_t op_ret, int32_t op_errno, fd_t *fd)
{
  GF_ERROR_IF_NULL (frame);
  GF_ERROR_IF_NULL (this);
  GF_ERROR_NO_RETURN_IF_NULL (fd);

  posix_locks_private_t *priv = (posix_locks_private_t *)this->private;
  pthread_mutex_lock (&priv->mutex);

  if (op_ret >= 0) {
    pl_fd_t *pfd = calloc (1, sizeof (pl_fd_t));
    pl_inode_t *inode;

    struct _flags *local = frame->local;
    if (frame->local)
      pfd->nonblocking = local->flags & O_NONBLOCK;

    if (!fd->inode) {
      gf_log (this->name, GF_LOG_ERROR, "fd->inode is NULL! returning EBADFD");
      STACK_UNWIND (frame, -1, EBADFD, fd);
    }

    data_t *inode_data = dict_get (fd->inode->ctx, this->name);
    if (inode_data == NULL) {
      pl_inode_t *inode = calloc (1, sizeof (pl_inode_t));

      mode_t st_mode = fd->inode->st_mode;
      if ((st_mode & S_ISGID) && !(st_mode & S_IXGRP))
	inode->mandatory = 1;

      dict_set (fd->inode->ctx, this->name, bin_to_data (inode, sizeof (inode)));
    }
    else {
      inode = data_to_bin (inode_data);
    }
    
    dict_set (fd->ctx, this->name, bin_to_data (pfd, sizeof (pfd)));
  }

  pthread_mutex_unlock (&priv->mutex);

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

int32_t 
pl_open (call_frame_t *frame, xlator_t *this,
	 loc_t *loc, int32_t flags, fd_t *fd)
{
  GF_ERROR_IF_NULL (frame);
  GF_ERROR_IF_NULL (this);
  GF_ERROR_IF_NULL (loc);

  struct _flags *f = calloc (1, sizeof (struct _flags));
  f->flags = flags;

  if (flags & O_RDONLY)
    f->flags &= ~O_TRUNC;

  frame->local = f;

  STACK_WIND (frame, pl_open_cbk, 
              FIRST_CHILD(this), FIRST_CHILD(this)->fops->open, 
              loc, flags & ~O_TRUNC, fd);

  return 0;
}

int32_t
pl_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)
{
  if (op_ret >= 0) {
    pl_inode_t *pinode = calloc (1, sizeof (pl_inode_t));
    pl_fd_t *pfd = calloc (1, sizeof (pl_fd_t));
    
    dict_set (fd->inode->ctx, this->name, bin_to_data (pinode, sizeof (pinode)));
    dict_set (fd->ctx, this->name, bin_to_data (pfd, sizeof (pfd)));
  }
  STACK_UNWIND (frame, op_ret, op_errno, fd, inode, buf);
  return 0;
}

int32_t 
pl_create (call_frame_t *frame, xlator_t *this,
	   loc_t *loc, int32_t flags, mode_t mode, fd_t *fd)
{
  GF_ERROR_IF_NULL (frame);
  GF_ERROR_IF_NULL (this);
  GF_ERROR_IF_NULL (loc->path);

  STACK_WIND (frame, pl_create_cbk,
              FIRST_CHILD (this),
              FIRST_CHILD (this)->fops->create, 
	      loc, flags, mode, fd);
  return 0;
}

int32_t
pl_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 *stbuf)
{
  GF_ERROR_IF_NULL (this);
  GF_ERROR_IF_NULL (vector);

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

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

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

static void
do_blocked_rw (pl_inode_t *inode)
{
  pl_rw_req_t *rw = inode->rw_reqs;

  while (rw) {
    if (rw_allowable (inode, rw->region, rw->op)) {
      switch (rw->op) {
      case OP_READ:
	STACK_WIND (rw->frame, pl_readv_cbk,
		    FIRST_CHILD (rw->this), FIRST_CHILD (rw->this)->fops->readv,
		    rw->fd, rw->size, rw->region->fl_start);
	break;
      case OP_WRITE: 
	{
	  dict_t *req_refs = rw->frame->root->req_refs;
	  STACK_WIND (rw->frame, pl_writev_cbk,
		      FIRST_CHILD (rw->this), 
		      FIRST_CHILD (rw->this)->fops->writev,
		      rw->fd, rw->vector, rw->size, rw->region->fl_start);
	  dict_unref (req_refs);
	  break;
	}
      }
      
      delete_rw_req (inode, rw);
      free (rw);
    }
    
    rw = rw->next;
  }
}

static int
rw_allowable (pl_inode_t *inode, posix_lock_t *region,
	      rw_op_t op)
{
  posix_lock_t *l = inode->locks;
  while (l) {
    if (locks_overlap (l, region) && !same_owner (l, region)) {
      if ((op == OP_READ) && (l->fl_type != F_WRLCK))
	continue;
      return 0;
    }
    l = l->next;
  }

  return 1;
}

int32_t
pl_readv (call_frame_t *frame, xlator_t *this,
	  fd_t *fd, size_t size, off_t offset)
{
  GF_ERROR_IF_NULL (this);
  GF_ERROR_IF_NULL (fd);

  posix_locks_private_t *priv = (posix_locks_private_t *)this->private;
  struct stat nullbuf = {0, };

  pthread_mutex_lock (&priv->mutex);

  data_t *fd_data = dict_get (fd->ctx, this->name);
  if (fd_data == NULL) {
    pthread_mutex_unlock (&priv->mutex);
    gf_log (this->name, GF_LOG_ERROR, "returning EBADF");
    STACK_UNWIND (frame, -1, EBADF, &nullbuf);
    return 0;
  }
  pl_fd_t *pfd = (pl_fd_t *) data_to_bin (fd_data);

  data_t *inode_data = dict_get (fd->inode->ctx, this->name);
  if (inode_data == NULL) {
    pthread_mutex_unlock (&priv->mutex);
    gf_log (this->name, GF_LOG_ERROR, "returning EBADF");
    STACK_UNWIND (frame, -1, EBADF, &nullbuf);
    return 0;
  }
  pl_inode_t *inode = (pl_inode_t *)data_to_bin (inode_data);

  if (priv->mandatory && inode->mandatory) {
    posix_lock_t *region = calloc (1, sizeof (posix_lock_t));
    region->fl_start = offset;
    region->fl_end   = offset + size - 1;
    region->transport = frame->root->trans;
    region->client_pid = frame->root->pid;
    
    if (!rw_allowable (inode, region, OP_READ)) {
      if (pfd->nonblocking) {
	pthread_mutex_unlock (&priv->mutex);
	gf_log (this->name, GF_LOG_ERROR, "returning EWOULDBLOCK");
	STACK_UNWIND (frame, -1, EWOULDBLOCK, &nullbuf);
	return -1;
      }

      pl_rw_req_t *rw = calloc (1, sizeof (pl_rw_req_t));
      rw->frame  = frame;
      rw->this   = this;
      rw->fd     = fd;
      rw->op     = OP_READ;
      rw->size   = size;
      rw->region = region;

      insert_rw_req (inode, rw);
      pthread_mutex_unlock (&priv->mutex);
      return 0;
    }
  }

  pthread_mutex_unlock (&priv->mutex);

  STACK_WIND (frame, pl_readv_cbk, 
	      FIRST_CHILD (this), FIRST_CHILD (this)->fops->readv,
	      fd, size, offset);
  return 0;
}

static int32_t
iovec_total_length (struct iovec *vector, int count)
{
  int32_t i;
  int32_t total_length = 0;
  for (i = 0; i < count; i++) {
    total_length += vector[i].iov_len;
  }

  return total_length;
}

int32_t 
pl_writev (call_frame_t *frame, xlator_t *this,
	   fd_t *fd, struct iovec *vector, int32_t count, off_t offset)
{
  GF_ERROR_IF_NULL (this);
  GF_ERROR_IF_NULL (fd);
  GF_ERROR_IF_NULL (vector);

  posix_locks_private_t *priv = (posix_locks_private_t *)this->private;
  struct stat nullbuf = {0, };

  pthread_mutex_lock (&priv->mutex);

  data_t *fd_data = dict_get (fd->ctx, this->name);
  if (fd_data == NULL) {
    pthread_mutex_unlock (&priv->mutex);
    gf_log (this->name, GF_LOG_ERROR, "returning EBADF");
    STACK_UNWIND (frame, -1, EBADF, &nullbuf);
    return 0;
  }
  pl_fd_t *pfd = (pl_fd_t *)data_to_bin (fd_data);

  data_t *inode_data = dict_get (fd->inode->ctx, this->name);
  if (inode_data == NULL) {
    pthread_mutex_unlock (&priv->mutex);
    gf_log (this->name, GF_LOG_ERROR, "returning EBADF");
    STACK_UNWIND (frame, -1, EBADF, &nullbuf);
    return 0;
  }
  pl_inode_t *inode = (pl_inode_t *)data_to_bin (inode_data);

  if (priv->mandatory && inode->mandatory) {
    int size = iovec_total_length (vector, count);

    posix_lock_t *region = calloc (1, sizeof (posix_lock_t));
    region->fl_start = offset;
    region->fl_end   = offset + size - 1;
    region->transport = frame->root->trans;
    region->client_pid = frame->root->pid;

    if (!rw_allowable (inode, region, OP_WRITE)) {
      if (pfd->nonblocking) {
	pthread_mutex_unlock (&priv->mutex);
	gf_log (this->name, GF_LOG_ERROR, "returning EWOULDBLOCK");
	STACK_UNWIND (frame, -1, EWOULDBLOCK, &nullbuf);
	return -1;
      }

      pl_rw_req_t *rw = calloc (1, sizeof (pl_rw_req_t));

      dict_ref (frame->root->req_refs);
      rw->frame  = frame;
      rw->this   = this;
      rw->fd     = fd;
      rw->op     = OP_WRITE;
      rw->size   = count;
      rw->vector = iov_dup (vector, count);
      rw->region = region;

      insert_rw_req (inode, rw);
      pthread_mutex_unlock (&priv->mutex);
      return 0;
    }
  }

  pthread_mutex_unlock (&priv->mutex);

  STACK_WIND (frame, pl_writev_cbk,
	      FIRST_CHILD(this), FIRST_CHILD(this)->fops->writev, 
	      fd, vector, count, offset);
  return 0;
}

int32_t
pl_lk (call_frame_t *frame, xlator_t *this,
       fd_t *fd, int32_t cmd,
       struct flock *flock)
{
  GF_ERROR_IF_NULL (frame);
  GF_ERROR_IF_NULL (this);
  GF_ERROR_IF_NULL (fd);
  GF_ERROR_IF_NULL (flock);

  transport_t *transport = frame->root->trans;
  pid_t client_pid = frame->root->pid;
  posix_locks_private_t *priv = (posix_locks_private_t *)this->private;
  pthread_mutex_lock (&priv->mutex);

  struct flock nulllock = {0, };
  data_t *fd_data = dict_get (fd->ctx, this->name);
  if (fd_data == NULL) {
    pthread_mutex_unlock (&priv->mutex);
    gf_log (this->name, GF_LOG_ERROR, "returning EBADF");
    STACK_UNWIND (frame, -1, EBADF, &nulllock);
    return 0;
  }
  pl_fd_t *pfd = (pl_fd_t *)data_to_bin (fd_data);

  if (!pfd) {
    pthread_mutex_unlock (&priv->mutex);
    gf_log (this->name, GF_LOG_ERROR, "returning EBADF");
    STACK_UNWIND (frame, -1, EBADF, nulllock);
    return -1;
  }

  data_t *inode_data = dict_get (fd->inode->ctx, this->name);
  if (inode_data == NULL) {
    pthread_mutex_unlock (&priv->mutex);
    gf_log (this->name, GF_LOG_ERROR, "returning EBADF");
    STACK_UNWIND (frame, -1, EBADF, &nulllock);
    return 0;
  }
  pl_inode_t *inode = (pl_inode_t *)data_to_bin (inode_data);
  
  posix_lock_t *reqlock = new_posix_lock (flock, transport, client_pid);

  int can_block = 0;

  switch (cmd) {
#if F_GETLK != F_GETLK64
  case F_GETLK64:
#endif

  case F_GETLK: {
    posix_lock_t *conf = posix_getlk (inode, reqlock);
    posix_lock_to_flock (conf, flock);
    pthread_mutex_unlock (&priv->mutex);
    destroy_lock (reqlock);

#ifdef _POSIX_LOCKS_DEBUG
    printf ("[GET] "); print_lock (reqlock);
#endif

    STACK_UNWIND (frame, 0, 0, flock);
    return 0;
  }

#if F_SETLKW != F_SETLKW64
  case F_SETLKW64:
#endif
  case F_SETLKW:
    can_block = 1;
    reqlock->frame = frame;
    reqlock->this  = this;
    reqlock->fd    = fd;
    reqlock->user_flock = calloc (1, sizeof (struct flock));
    memcpy (reqlock->user_flock, flock, sizeof (struct flock));

#if F_SETLK != F_SETLK64
  case F_SETLK64:
#endif
  case F_SETLK: {
    int ret = posix_setlk (inode, reqlock, can_block);

#ifdef _POSIX_LOCKS_DEBUG
    printf ("[SET] (ret=%d)", ret); print_lock (reqlock);
#endif

    pthread_mutex_unlock (&priv->mutex);

    if (ret == -1) {
      if (can_block)
	return -1;

      gf_log (this->name, GF_LOG_ERROR, "returning EAGAIN");
      STACK_UNWIND (frame, ret, EAGAIN, flock);
      return -1;
    }

    if (ret == 0) {
      STACK_UNWIND (frame, ret, 0, flock);
      return 0;
    }
  }
  }

  pthread_mutex_unlock (&priv->mutex);
  gf_log (this->name, GF_LOG_ERROR, "returning EINVAL");
  STACK_UNWIND (frame, -1, EINVAL, flock); /* Normally this shouldn't be reached */
  return -1;
}

int32_t
pl_forget (call_frame_t *frame,
	   xlator_t *this,
	   inode_t *inode)
{
  data_t *inode_data = dict_get (inode->ctx, this->name);
  if (inode_data) {
    pl_inode_t *pl_inode = (pl_inode_t *)data_to_bin (inode_data);
    if (pl_inode->rw_reqs) {
      gf_log (this->name, GF_LOG_ERROR,
	      "Pending R/W requests found!");
    }
    if (pl_inode->locks) {
      gf_log (this->name, GF_LOG_ERROR,
	      "Active locks found!");
    }

    free (pl_inode);
  }
  return 0;
}

int32_t
init (xlator_t *this)
{
  posix_locks_private_t *priv = NULL;
  data_t *mandatory = NULL;
  if (!this->children) {
    gf_log (this->name, 
	    GF_LOG_ERROR, 
	    "FATAL: posix-locks should have exactly one child");
    return -1;
  }

  if (this->children->next) {
    gf_log (this->name, 
	    GF_LOG_ERROR, 
	    "FATAL: posix-locks should have exactly one child");
    return -1;
  }

  priv = calloc (1, sizeof (posix_locks_private_t));
  pthread_mutex_init (&priv->mutex, NULL);

  mandatory = dict_get (this->options, "mandatory");
  if (mandatory) {
    if (strcasecmp (mandatory->data, "on") == 0)
      priv->mandatory = 1;
  }

  this->private = priv;
  return 0;
}

int32_t
fini (xlator_t *this)
{
  posix_locks_private_t *priv = this->private;
  free (priv);
  return 0;
}

struct xlator_fops fops = {
  .create      = pl_create,
  .truncate    = pl_truncate,
  .ftruncate   = pl_ftruncate,
  .open        = pl_open,
  .readv       = pl_readv,
  .writev      = pl_writev,
  .close       = pl_close,
  .lk          = pl_lk,
  .flush       = pl_flush,
  .forget      = pl_forget
};

struct xlator_mops mops = {
};