posix.c

分布式文件系统

  char *real_path;

  DECLARE_OLD_FS_UID_VAR;

  MAKE_REAL_PATH (real_path, this, loc->path);
  /*
  _fd = open (real_path, O_DIRECTORY | O_RDONLY);
  */
  SET_FS_UID (frame->root->uid, frame->root->gid);
  
  op_ret = rmdir (real_path);
  op_errno = errno;
  if (op_ret == -1 && op_errno != ENOTEMPTY) {
    gf_log (this->name, GF_LOG_WARNING, 
	    "rmdir of %s: %s", loc->path, strerror (op_errno));
  }
    
  SET_TO_OLD_FS_UID ();

  /*
  if (op_ret == -1) {
    close (_fd);
  } else {
    dict_set (loc->inode->ctx, this->name, data_from_int32 (_fd));
  }
  */

  frame->root->rsp_refs = NULL;
  STACK_UNWIND (frame, op_ret, op_errno);

  return 0;
}

int32_t 
posix_symlink (call_frame_t *frame,
	       xlator_t *this,
	       const char *linkname,
	       loc_t *loc)
{
  int32_t op_ret;
  int32_t op_errno;
  char *real_path;
  struct stat stbuf = { 0, };
  DECLARE_OLD_FS_UID_VAR;

  MAKE_REAL_PATH (real_path, this, loc->path);

  SET_FS_UID (frame->root->uid, frame->root->gid);
    
  op_ret = symlink (linkname, real_path);
  op_errno = errno;
  if (op_ret == -1) {
    gf_log (this->name, GF_LOG_WARNING, 
	    "symlink of %s --> %s: %s", 
	    loc->path, linkname, strerror (op_errno));
  }
  
  if (op_ret == 0) {
#ifndef HAVE_SET_FS_ID
    lchown (real_path, frame->root->uid, frame->root->gid);
#endif
    lstat (real_path, &stbuf);
  }
    
  SET_TO_OLD_FS_UID ();

  frame->root->rsp_refs = NULL;
  STACK_UNWIND (frame, op_ret, op_errno, loc->inode, &stbuf);

  return 0;
}

int32_t 
posix_rename (call_frame_t *frame,
	      xlator_t *this,
	      loc_t *oldloc,
	      loc_t *newloc)
{
  int32_t op_ret;
  int32_t op_errno;
  char *real_oldpath;
  char *real_newpath;
  struct stat stbuf = {0, };
  DECLARE_OLD_FS_UID_VAR;

  MAKE_REAL_PATH (real_oldpath, this, oldloc->path);
  MAKE_REAL_PATH (real_newpath, this, newloc->path);

  SET_FS_UID (frame->root->uid, frame->root->gid);
  
  op_ret = rename (real_oldpath, real_newpath);
  op_errno = errno;
  if (op_ret == -1) {
    gf_log (this->name, GF_LOG_WARNING, 
	    "rename of %s -> %s: %s", 
	    oldloc->path, newloc->path, strerror (op_errno));
  }
    
  if (op_ret == 0) {
    lstat (real_newpath, &stbuf);
  }

  SET_TO_OLD_FS_UID ();

  frame->root->rsp_refs = NULL;
  STACK_UNWIND (frame, op_ret, op_errno, &stbuf);

  return 0;
}

int32_t 
posix_link (call_frame_t *frame, 
	    xlator_t *this,
	    loc_t *oldloc,
	    const char *newpath)
{
  int32_t op_ret;
  int32_t op_errno;
  char *real_oldpath;
  char *real_newpath;
  struct stat stbuf = {0, };
  DECLARE_OLD_FS_UID_VAR;

  MAKE_REAL_PATH (real_oldpath, this, oldloc->path);
  MAKE_REAL_PATH (real_newpath, this, newpath);

  SET_FS_UID (frame->root->uid, frame->root->gid);
    
  op_ret = link (real_oldpath, real_newpath);
  op_errno = errno;
  if (op_ret == -1) {
    gf_log (this->name, GF_LOG_WARNING, 
	    "link on %s -> %s: %s", 
	    oldloc->path, newpath, strerror (op_errno));
  }
    
  if (op_ret == 0) {
#ifndef HAVE_SET_FSID
    lchown (real_newpath, frame->root->uid, frame->root->gid);
#endif
    lstat (real_newpath, &stbuf);
  }
    
  SET_TO_OLD_FS_UID ();

  frame->root->rsp_refs = NULL;
  STACK_UNWIND (frame, op_ret, op_errno, oldloc->inode, &stbuf);

  return 0;
}


int32_t 
posix_chmod (call_frame_t *frame,
	     xlator_t *this,
	     loc_t *loc,
	     mode_t mode)
{
  int32_t op_ret;
  int32_t op_errno;
  char *real_path;
  struct stat stbuf;
  DECLARE_OLD_FS_UID_VAR;
  
  MAKE_REAL_PATH (real_path, this, loc->path);

  SET_FS_UID (frame->root->uid, frame->root->gid);
  
  op_ret = chmod (real_path, mode);
  op_errno = errno;
  if (op_ret == -1) {
    gf_log (this->name, GF_LOG_WARNING, 
	    "chmod on %s: %s", loc->path, strerror (op_errno));
  }
    
  if (op_ret == 0)
    lstat (real_path, &stbuf);
    
  SET_TO_OLD_FS_UID ();

  frame->root->rsp_refs = NULL;
  STACK_UNWIND (frame, op_ret, op_errno, &stbuf);

  return 0;
}


int32_t 
posix_chown (call_frame_t *frame,
	     xlator_t *this,
	     loc_t *loc,
	     uid_t uid,
	     gid_t gid)
{
  int32_t op_ret;
  int32_t op_errno;
  char *real_path;
  struct stat stbuf;
  DECLARE_OLD_FS_UID_VAR;

  MAKE_REAL_PATH (real_path, this, loc->path);

  SET_FS_UID (frame->root->uid, frame->root->gid);
    
  op_ret = lchown (real_path, uid, gid);
  op_errno = errno;
  if (op_ret == -1) {
    gf_log (this->name, GF_LOG_WARNING, 
	    "chown on %s: %s", loc->path, strerror (op_errno));
  }
    
  if (op_ret == 0)
    lstat (real_path, &stbuf);
    
  SET_TO_OLD_FS_UID ();

  frame->root->rsp_refs = NULL;  
  STACK_UNWIND (frame, op_ret, op_errno, &stbuf);

  return 0;
}


int32_t 
posix_truncate (call_frame_t *frame,
		xlator_t *this,
		loc_t *loc,
		off_t offset)
{
  int32_t op_ret;
  int32_t op_errno;
  char *real_path;
  struct stat stbuf;
  DECLARE_OLD_FS_UID_VAR;

  MAKE_REAL_PATH (real_path, this, loc->path);

  SET_FS_UID (frame->root->uid, frame->root->gid);

  op_ret = truncate (real_path, offset);
  op_errno = errno;
  if (op_ret == -1) {
    gf_log (this->name, GF_LOG_WARNING, 
	    "truncate of %s: %s", loc->path, strerror (op_errno));
  }
    
  if (op_ret == 0) {
    lstat (real_path, &stbuf);
  }
    
  SET_TO_OLD_FS_UID ();

  frame->root->rsp_refs = NULL;
  STACK_UNWIND (frame, op_ret, op_errno, &stbuf);

  return 0;
}


int32_t 
posix_utimens (call_frame_t *frame,
	       xlator_t *this,
	       loc_t *loc,
	       struct timespec ts[2])
{
  int32_t op_ret;
  int32_t op_errno;
  char *real_path;
  struct stat stbuf = {0, };
  struct timeval tv[2];
  DECLARE_OLD_FS_UID_VAR;
  
  MAKE_REAL_PATH (real_path, this, loc->path);

  /* TODO: fix timespec to timeval converstion 
   * Done: Check if this is correct */

  tv[0].tv_sec = ts[0].tv_sec;
  tv[0].tv_usec = ts[0].tv_nsec / 1000;
  tv[1].tv_sec = ts[1].tv_sec;
  tv[1].tv_usec = ts[1].tv_nsec / 1000;

  SET_FS_UID (frame->root->uid, frame->root->gid);
    
  op_ret = lutimes (real_path, tv);
  if (op_ret == -1 && errno == ENOSYS) {
    op_ret = utimes (real_path, tv);
  }
  op_errno = errno;
  if (op_ret == -1) {
    gf_log (this->name, GF_LOG_WARNING, 
	    "utimes on %s: %s", loc->path, strerror (op_errno));
  }

  if (op_ret == 0)
    lstat (real_path, &stbuf);
 
  SET_TO_OLD_FS_UID ();

  frame->root->rsp_refs = NULL;
  STACK_UNWIND (frame, op_ret, op_errno, &stbuf);

  return 0;
}

int32_t 
posix_create (call_frame_t *frame,
	      xlator_t *this,
	      loc_t *loc,
	      int32_t flags,
	      mode_t mode,
	      fd_t *fd)
{
  int32_t op_ret = -1;
  int32_t op_errno = 0;
  int32_t _fd;
  char *real_path;
  struct stat stbuf = {0, };
  DECLARE_OLD_FS_UID_VAR;

  MAKE_REAL_PATH (real_path, this, loc->path);

  frame->root->rsp_refs = NULL;
  SET_FS_UID (frame->root->uid, frame->root->gid);
    
  if (!flags) {
    _fd = open (real_path, 
		O_CREAT|O_RDWR|O_EXCL,
		mode);
  } else {
    _fd = open (real_path, 
		flags|O_CREAT,
		mode);
  }

  op_errno = errno;
  if (_fd == -1) {
    gf_log (this->name, GF_LOG_WARNING, 
	    "open on %s: %s", loc->path, strerror (op_errno));
  }
    
  if (_fd >= 0) {
      /* trigger readahead in the kernel */
#if 0
    char buf[1024 * 64];
    read (_fd, buf, 1024 * 64);
    lseek (_fd, 0, SEEK_SET);
#endif
#ifndef HAVE_SET_FSID
    chown (real_path, frame->root->uid, frame->root->gid);
#endif
    fstat (_fd, &stbuf);
  }
  SET_TO_OLD_FS_UID ();

  if (_fd >= 0) {
    struct posix_fd *pfd;

    pfd = calloc (1, sizeof (*pfd));

    if (!pfd) {
      close (_fd);
      STACK_UNWIND (frame, -1, ENOMEM, fd, loc->inode, &stbuf);
      return 0;
    }

    pfd->flags = flags;
    pfd->fd = _fd;
    dict_set (fd->ctx, this->name, data_from_static_ptr (pfd));
    ((struct posix_private *)this->private)->stats.nr_files++;
    op_ret = 0;
  }

  frame->root->rsp_refs = NULL;
  STACK_UNWIND (frame, op_ret, op_errno, fd, loc->inode, &stbuf);

  return 0;
}

int32_t 
posix_open (call_frame_t *frame,
	    xlator_t *this,
	    loc_t *loc,
	    int32_t flags,
	    fd_t *fd)
{
  int32_t op_ret = -1;
  int32_t op_errno = 0;
  char *real_path;
  int32_t _fd;
  DECLARE_OLD_FS_UID_VAR;

  MAKE_REAL_PATH (real_path, this, loc->path);

  SET_FS_UID (frame->root->uid, frame->root->gid);
    
  _fd = open (real_path, flags, 0);
  op_errno = errno;
  if (_fd == -1) {
    gf_log (this->name, GF_LOG_WARNING, 
	    "open on %s: %s", loc->path, strerror (op_errno));
  }
    
  SET_TO_OLD_FS_UID ();

  if (_fd >= 0) {
    struct posix_fd *pfd = calloc (1, sizeof (*pfd));

    if (!pfd) {
      close (_fd);
      STACK_UNWIND (frame, -1, ENOMEM, fd);
      return 0;
    }

    pfd->flags = flags;
    pfd->fd = _fd;
    dict_set (fd->ctx, this->name, data_from_static_ptr (pfd));

    ((struct posix_private *)this->private)->stats.nr_files++;
    op_ret = 0;
#ifndef HAVE_SET_FSID
    if (flags & O_CREAT)
      chown (real_path, frame->root->uid, frame->root->gid);
#endif
  }

  frame->root->rsp_refs = NULL;
  STACK_UNWIND (frame, op_ret, op_errno, fd);

  return 0;
}

int32_t 
posix_readv (call_frame_t *frame,
	     xlator_t *this,
	     fd_t *fd,
	     size_t size,
	     off_t offset)
{
  int32_t op_ret = -1;
  int32_t op_errno = 0;
  char *buf = NULL, *alloc_buf = NULL;
  int32_t _fd;
  struct posix_private *priv = this->private;
  dict_t *reply_dict = NULL;
  struct iovec vec;
  data_t *pfd_data;
  struct posix_fd *pfd;
  struct stat stbuf = {0,};
  int32_t align = 1;

  frame->root->rsp_refs = NULL;
  pfd_data = dict_get (fd->ctx, this->name);

  if (pfd_data == NULL) {
    gf_log (this->name, GF_LOG_ERROR, "pfd_data NULL from fd=%p", fd);
    STACK_UNWIND (frame, -1, EBADF, &vec, 0, &stbuf);
    return 0;
  }

  pfd = data_to_ptr (pfd_data);

  if (!pfd) {
    gf_log (this->name, GF_LOG_ERROR, "pfd is NULL from fd=%p", fd);
    STACK_UNWIND (frame, -1, EBADF, &vec, 0, &stbuf);
    return 0;
  }

  if (!size) {
    gf_log (this->name, GF_LOG_WARNING, "size == 0");
    STACK_UNWIND (frame, 0, 0, &vec, 0, &stbuf);
    return 0;
  }

  if (pfd->flags & O_DIRECT) {
    align = 4096;
  }

  alloc_buf = malloc (1 * (size + align));
  if (!alloc_buf) {
    gf_log (this->name, GF_LOG_ERROR,
	    "unable to allocate read buffer of %d + %d bytes",
	    size, align);
    STACK_UNWIND (frame, -1, ENOMEM, &vec, 0, &stbuf);
    return 0;
  }

 /* page aligned buffer */
  buf = (void *)((unsigned long)(alloc_buf + align - 1) &
		 (unsigned long)(~(align - 1)));

  _fd = pfd->fd;

  priv->read_value += size;
  priv->interval_read += size;

  if (lseek (_fd, offset, SEEK_SET) == -1) {
    frame->root->rsp_refs = NULL;
    gf_log (this->name, GF_LOG_ERROR, "lseek(%"PRId64") failed", offset); 
    STACK_UNWIND (frame, -1, errno, &vec, 0, &stbuf);
    return 0;
  }
  
  op_ret = read (_fd, buf, size);
  op_errno = errno;
  if (op_ret == -1) {
    gf_log (this->name, GF_LOG_WARNING, 
	    "read failed: %s", strerror (op_errno));
  }

  vec.iov_base = buf;
  vec.iov_len = op_ret;
    
  if (op_ret >= 0) {
    data_t *buf_data = get_new_data ();
    reply_dict = get_new_dict ();

    reply_dict->is_locked = 1;
    buf_data->is_locked = 1;
    buf_data->data = alloc_buf;
    buf_data->len = op_ret;

    dict_set (reply_dict, NULL, buf_data);
    frame->root->rsp_refs = dict_ref (reply_dict);
    /* readv successful, we also need to get the stat of the file we read from */
    fstat (_fd, &stbuf);
  }

  STACK_UNWIND (frame, op_ret, op_errno, &vec, 1, &stbuf);

  if (reply_dict)
    dict_unref (reply_dict);
  return 0;
}


int32_t 
posix_writev (call_frame_t *frame,
	      xlator_t *this,
	      fd_t *fd,
	      struct iovec *vector,
	      int32_t count,
	      off_t offset)
{
  int32_t op_ret;
  int32_t op_errno = 0;
  int32_t _fd;
  struct posix_private *priv = this->private;
  data_t *pfd_data = dict_get (fd->ctx, this->name);
  struct posix_fd *pfd;
  struct stat stbuf = {0,};

  frame->root->rsp_refs = NULL;

  if (pfd_data == NULL) {
    gf_log (this->name, GF_LOG_ERROR, "pfd_data is NULL from fd=%p", fd);
    STACK_UNWIND (frame, -1, EBADF, &stbuf);
    return 0;
  }
  
  pfd = data_to_ptr (pfd_data); 

  if (!pfd) {
    gf_log (this->name, GF_LOG_ERROR, "pfd is NULL from fd=%p", fd);
    STACK_UNWIND (frame, -1, EBADF, &stbuf);
    return 0;
  }

  _fd = pfd->fd;

  if (lseek (_fd, offset, SEEK_SET) == -1) {
    gf_log (this->name, GF_LOG_ERROR, "lseek(%"PRId64") failed", offset);
    frame->root->rsp_refs = NULL;
    STACK_UNWIND (frame, -1, errno, &stbuf);
    return 0;
  }

  /* Check for the O_DIRECT flag during open() */
  if (pfd->flags & O_DIRECT) {
    /* This is O_DIRECT'd file */
    int32_t idx = 0;
    int32_t align = 4096;
    int32_t max_buf_size = 0;
    int32_t retval = 0;
    char *buf = NULL; 
    char *alloc_buf = NULL;
    if (offset % align) {
      /* Return EINVAL */
      gf_log (this->name, GF_LOG_ERROR, "O_DIRECT: offset is Invalid");
      frame->root->rsp_refs = NULL;
      STACK_UNWIND (frame, -1, EINVAL, &stbuf);
      return 0; 
    }
    
    op_ret = 0;
    for (idx = 0; idx < count; idx++) {
      if (max_buf_size < vector[idx].iov_len)
	max_buf_size = vector[idx].iov_len;
    }
    alloc_buf = malloc (1 * (max_buf_size + align));
    if (!alloc_buf) {
      gf_log (this->name, GF_LOG_ERROR,
	      "unable to allocate read buffer of %d + %d bytes",
	      vector[idx].iov_len, align);
      STACK_UNWIND (frame, -1, ENOMEM, &stbuf);
      return 0;
    }

    for (idx = 0; idx < count; idx++) {
      /* page aligned buffer */
      buf = (void *)((unsigned long)(alloc_buf + align - 1) &
		     (unsigned long)(~(align - 1)));

      memcpy (buf, vector[idx].iov_base, vector[idx].iov_len);
      
      /* not sure whether writev works on O_DIRECT'd fd */
      retval = write (_fd, buf, vector[idx].iov_len);
      
      if (retval == -1) {
	op_ret = -1;
	op_errno = errno;
	if (op_ret == -1) {
	  gf_log (this->name, GF_LOG_WARNING,