/**
 * FSAL_getattrs:
 * Get attributes for the object specified by its filehandle.
 *
 * \param filehandle (input):
 *        The handle of the object to get parameters.
 * \param context (input):
 *        Authentication context for the operation (user, export...).
 * \param object_attributes (mandatory input/output):
 *        The retrieved attributes for the object.
 *        As input, it defines the attributes that the caller
 *        wants to retrieve (by positioning flags into this structure)
 *        and the output is built considering this input
 *        (it fills the structure according to the flags it contains).
 *
 * \return Major error codes :
 *        - ERR_FSAL_NO_ERROR     (no error)
 *        - ERR_FSAL_STALE        (object_handle does not address an existing object)
 *        - ERR_FSAL_FAULT        (a NULL pointer was passed as mandatory argument) 
 *        - Another error code if an error occured.
 */
fsal_status_t CEPHFSAL_getattrs(fsal_handle_t * exthandle,
                                fsal_op_context_t * extcontext,
                                fsal_attrib_list_t * object_attributes)
{
  int rc;
  struct stat st;
  fsal_status_t status;
  cephfsal_handle_t* filehandle = (cephfsal_handle_t*) exthandle;
  cephfsal_op_context_t* context = (cephfsal_op_context_t*) extcontext;
  int uid = FSAL_OP_CONTEXT_TO_UID(context);
  int gid = FSAL_OP_CONTEXT_TO_GID(context);

  /* sanity checks.
   * note : object_attributes is mandatory in FSAL_getattrs.
   */
  if(!filehandle || !context || !object_attributes)
    Return(ERR_FSAL_FAULT, 0, INDEX_FSAL_getattrs);

  TakeTokenFSCall();

  rc = ceph_ll_getattr(context->export_context->cmount, VINODE(filehandle),
                       &st, uid, gid);

  ReleaseTokenFSCall();

  if (rc < 0)
    Return(posix2fsal_error(rc), 0, INDEX_FSAL_getattrs);

  /* convert attributes */
  status = posix2fsal_attributes(&st, object_attributes);
  if(FSAL_IS_ERROR(status))
    {
      FSAL_CLEAR_MASK(object_attributes->asked_attributes);
      FSAL_SET_MASK(object_attributes->asked_attributes, FSAL_ATTR_RDATTR_ERR);
      Return(ERR_FSAL_NO_ERROR, 0, INDEX_FSAL_getattrs);
    }

  Return(ERR_FSAL_NO_ERROR, 0, INDEX_FSAL_getattrs);
}

fsal_status_t tank_remove_extattr_by_name(struct fsal_obj_handle *obj_hdl,
					  const char *xattr_name)
{
	struct zfs_fsal_obj_handle *obj_handle = NULL;
	int retval = 0;
	creden_t cred;

	obj_handle =
	    container_of(obj_hdl, struct zfs_fsal_obj_handle, obj_handle);

	cred.uid = op_ctx->creds->caller_uid;
	cred.gid = op_ctx->creds->caller_gid;

	retval = libzfswrap_removexattr(ZFSFSAL_GetVFS(obj_handle->handle),
					&cred,
					obj_handle->handle->zfs_handle,
					xattr_name);
	if (retval != 0)
		return fsalstat(posix2fsal_error(retval), retval);
	return fsalstat(ERR_FSAL_NO_ERROR, 0);

}

static int xattr_id_to_name(libzfswrap_vfs_t *p_vfs, creden_t *pcred,
			    inogen_t object, unsigned int xattr_id, char *name)
{
	unsigned int index;
	unsigned int curr_idx;
	char names[MAXPATHLEN], *ptr;
	size_t namesize;
	size_t len = 0;
	int retval = 0;

	if (xattr_id < XATTR_COUNT)
		return ERR_FSAL_INVAL;

	index = xattr_id - XATTR_COUNT;

	/* get xattrs */

	retval =
	    libzfswrap_listxattr(p_vfs, pcred, object, (char **)&names,
				 &namesize);
	if (retval)
		return posix2fsal_error(retval);

	if (namesize == 0)
		return ERR_FSAL_NOENT;

	errno = 0;

	for (ptr = names, curr_idx = 0;
	     ptr < names + namesize;
	     curr_idx++, ptr += len + 1) {
		len = strlen(ptr);
		if (curr_idx == index) {
			strcpy(name, ptr);
			return ERR_FSAL_NO_ERROR;
		}
	}
	return ERR_FSAL_NOENT;
}

/** @fn fsal_status_t
 *	GPFSFSAL_write(int fd, uint64_t offset, size_t buf_size, caddr_t buf,
 *		       size_t *write_amount, bool *fsal_stable,
 *		       const struct req_op_context *op_ctx)
 *  @brief Perform a write operation on an opened file.
 *
 *  @param fd The file descriptor returned by FSAL_open.
 *  @param buf_size Amount (in bytes) of data to be written.
 *  @param buf Address where the data is in memory.
 *  @param write_amount Pointer to the amount of data (in bytes) that have been written
 *        during this call.
 *
 * @return ERR_FSAL_NO_ERROR on success, error otherwise
 */
fsal_status_t
GPFSFSAL_write(int fd, uint64_t offset, size_t buf_size, caddr_t buf,
	       size_t *write_amount, bool *fsal_stable,
	       const struct req_op_context *op_ctx, int expfd)
{
	struct write_arg warg = {0};
	uint32_t stability_got = 0;
	ssize_t nb_write;
	int errsv;

	/* sanity checks. */
	if (!buf || !write_amount)
		return fsalstat(ERR_FSAL_FAULT, 0);

	warg.mountdirfd = expfd;
	warg.fd = fd;
	warg.bufP = buf;
	warg.offset = offset;
	warg.length = buf_size;
	warg.stability_wanted = *fsal_stable;
	warg.stability_got = &stability_got;
	warg.options = 0;

	fsal_set_credentials(op_ctx->creds);
	nb_write = gpfs_ganesha(OPENHANDLE_WRITE_BY_FD, &warg);
	errsv = errno;
	fsal_restore_ganesha_credentials();

	if (nb_write == -1) {
		if (errsv == EUNATCH)
			LogFatal(COMPONENT_FSAL, "GPFS Returned EUNATCH");
		return fsalstat(posix2fsal_error(errsv), errsv);
	}

	*write_amount = nb_write;
	*fsal_stable = (stability_got) ? true : false;

	return fsalstat(ERR_FSAL_NO_ERROR, 0);
}

static fsal_status_t getxattrs(struct fsal_obj_handle *obj_hdl,
				xattrname4 *xa_name,
				xattrvalue4 *xa_value)
{
	int rc;
	int errsv;
	struct getxattr_arg gxarg;
	struct gpfs_fsal_obj_handle *myself;
	struct gpfs_filesystem *gpfs_fs = obj_hdl->fs->private_data;

	myself = container_of(obj_hdl, struct gpfs_fsal_obj_handle,
				obj_handle);

	gxarg.mountdirfd = gpfs_fs->root_fd;
	gxarg.handle = myself->handle;
	gxarg.name_len = xa_name->utf8string_len;
	gxarg.name = xa_name->utf8string_val;
	gxarg.value_len = xa_value->utf8string_len;
	gxarg.value = xa_value->utf8string_val;

	rc = gpfs_ganesha(OPENHANDLE_GETXATTRS, &gxarg);
	if (rc < 0) {
		errsv = errno;
		LogDebug(COMPONENT_FSAL,
			"GETXATTRS returned rc %d errsv %d", rc, errsv);

		if (errsv == ERANGE)
			return fsalstat(ERR_FSAL_TOOSMALL, 0);
		if (errsv == ENODATA)
			return fsalstat(ERR_FSAL_NOENT, 0);
		return fsalstat(posix2fsal_error(errsv), errsv);
	}
	LogDebug(COMPONENT_FSAL,
		"GETXATTRS returned value %.*s len %d rc %d",
		gxarg.value_len, (char *)gxarg.value, gxarg.value_len, rc);

	xa_value->utf8string_len = gxarg.value_len;
	return fsalstat(ERR_FSAL_NO_ERROR, 0);
}

static fsal_status_t makedir(struct fsal_obj_handle *dir_hdl,
			     const char *name, struct attrlist *attrib,
			     struct fsal_obj_handle **handle)
{
	fsal_errors_t fsal_error = ERR_FSAL_NO_ERROR;
	int retval = 0;
	struct pt_fsal_obj_handle *hdl;
	fsal_status_t status;
	ptfsal_handle_t *fh = alloca(sizeof(ptfsal_handle_t));

	*handle = NULL;		/* poison it */
	if (!dir_hdl->ops->handle_is(dir_hdl, DIRECTORY)) {
		LogCrit(COMPONENT_FSAL,
			"Parent handle is not a directory. hdl = 0x%p",
			dir_hdl);
		return fsalstat(ERR_FSAL_NOTDIR, 0);
	}
	memset(fh, 0, sizeof(ptfsal_handle_t));
	fh->data.handle.handle_size = FSI_CCL_PERSISTENT_HANDLE_N_BYTES;

	attrib->mask =
	    op_ctx->fsal_export->ops->fs_supported_attrs(op_ctx->fsal_export);
	status = PTFSAL_mkdir(dir_hdl, name, op_ctx, attrib->mode, fh, attrib);
	if (FSAL_IS_ERROR(status))
		return status;

	/* allocate an obj_handle and fill it up */
	hdl = alloc_handle(fh, attrib, NULL, NULL, NULL, op_ctx->fsal_export);
	if (hdl == NULL) {
		retval = ENOMEM;
		goto fileerr;
	}
	*handle = &hdl->obj_handle;
	return fsalstat(ERR_FSAL_NO_ERROR, 0);

 fileerr:
	fsal_error = posix2fsal_error(retval);
	return fsalstat(fsal_error, retval);
}

/**
 * FSAL_dynamic_fsinfo:
 * Return dynamic filesystem info such as
 * used size, free size, number of objects...
 *
 * \param filehandle (input):
 *        Handle of an object in the filesystem
 *        whom info is to be retrieved.
 * \param cred (input):
 *        Authentication context for the operation (user,...).
 * \param dynamicinfo (output):
 *        Pointer to the static info of the filesystem.
 *
 * \return Major error codes:
 *      - ERR_FSAL_NO_ERROR: no error.
 *      - ERR_FSAL_FAULT: NULL pointer passed as input parameter.
 *      - ERR_FSAL_SERVERFAULT: Unexpected error.
 */
fsal_status_t LUSTREFSAL_dynamic_fsinfo(fsal_handle_t * p_filehandle,     /* IN */
                                        fsal_op_context_t * p_context,    /* IN */
                                        fsal_dynamicfsinfo_t * p_dynamicinfo    /* OUT */
    )
{
  fsal_path_t pathfsal;
  fsal_status_t status;
  struct statvfs buffstatvfs;
  int rc, errsv;
  /* sanity checks. */
  if(!p_filehandle || !p_dynamicinfo || !p_context)
    Return(ERR_FSAL_FAULT, 0, INDEX_FSAL_dynamic_fsinfo);

  status = fsal_internal_Handle2FidPath(p_context, p_filehandle, &pathfsal);
  if(FSAL_IS_ERROR(status))
    Return(status.major, status.minor, INDEX_FSAL_dynamic_fsinfo);

  TakeTokenFSCall();
  rc = statvfs(pathfsal.path, &buffstatvfs);
  errsv = errno;
  ReleaseTokenFSCall();
  if(rc)
    Return(posix2fsal_error(errsv), errsv, INDEX_FSAL_dynamic_fsinfo);

  p_dynamicinfo->total_bytes = buffstatvfs.f_frsize * buffstatvfs.f_blocks;
  p_dynamicinfo->free_bytes = buffstatvfs.f_frsize * buffstatvfs.f_bfree;
  p_dynamicinfo->avail_bytes = buffstatvfs.f_frsize * buffstatvfs.f_bavail;

  p_dynamicinfo->total_files = buffstatvfs.f_files;
  p_dynamicinfo->free_files = buffstatvfs.f_ffree;
  p_dynamicinfo->avail_files = buffstatvfs.f_favail;

  p_dynamicinfo->time_delta.seconds = 1;
  p_dynamicinfo->time_delta.nseconds = 0;

  Return(ERR_FSAL_NO_ERROR, 0, INDEX_FSAL_dynamic_fsinfo);

}

fsal_status_t ZFSFSAL_close(fsal_file_t * file_desc  /* IN */
    )
{
  int rc = 0;
  zfsfsal_file_t * file_descriptor = (zfsfsal_file_t *)file_desc;

  /* sanity checks. */
  if(!file_descriptor)
    Return(ERR_FSAL_FAULT, 0, INDEX_FSAL_close);

  TakeTokenFSCall();

  if(!file_descriptor->is_closed)
  {
    /* Test that the vfs still exist */
    ZFSFSAL_VFS_RDLock();
    libzfswrap_vfs_t *p_vfs = ZFSFSAL_GetVFS(&file_descriptor->handle);
    if(!p_vfs)
    {
      ZFSFSAL_VFS_Unlock();
      ReleaseTokenFSCall();
      Return(ERR_FSAL_NOENT, 0, INDEX_FSAL_close);
    }

    rc = libzfswrap_close(p_vfs, &file_descriptor->cred, file_descriptor->p_vnode,
                          file_descriptor->flags);
    ZFSFSAL_VFS_Unlock();
    file_descriptor->is_closed = 1;
  }

  ReleaseTokenFSCall();

  if(rc)
    Return(posix2fsal_error(rc), rc, INDEX_FSAL_close);

  Return(ERR_FSAL_NO_ERROR, 0, INDEX_FSAL_close);

}

fsal_status_t gpfs_seek(struct fsal_obj_handle *obj_hdl,
				struct io_info *info)
{
	struct fseek_arg arg;
	struct gpfs_fsal_obj_handle *myself;
	fsal_errors_t fsal_error = ERR_FSAL_NO_ERROR;
	struct gpfs_io_info io_info;
	int retval = 0;

	myself = container_of(obj_hdl, struct gpfs_fsal_obj_handle, obj_handle);

	assert(myself->u.file.fd >= 0
	       && myself->u.file.openflags != FSAL_O_CLOSED);

	arg.mountdirfd = myself->u.file.fd;
	arg.openfd = myself->u.file.fd;
	arg.info = &io_info;
	io_info.io_offset = info->io_content.hole.di_offset;
	if (info->io_content.what == NFS4_CONTENT_DATA)
		io_info.io_what = SEEK_DATA;
	else if (info->io_content.what == NFS4_CONTENT_HOLE)
		io_info.io_what = SEEK_HOLE;
	else
		return fsalstat(ERR_FSAL_UNION_NOTSUPP, 0);

	retval = gpfs_ganesha(OPENHANDLE_SEEK_BY_FD, &arg);
	if (retval == -1) {
		retval = errno;
		if (retval == EUNATCH)
			LogFatal(COMPONENT_FSAL, "GPFS Returned EUNATCH");
		fsal_error = posix2fsal_error(retval);
	} else {
		info->io_eof = io_info.io_eof;
		info->io_content.hole.di_offset = io_info.io_offset;
		info->io_content.hole.di_length = io_info.io_len;
	}
	return fsalstat(fsal_error, 0);
}

fsal_status_t XFSFSAL_check_quota( char              * path,  /* IN */
                                   fsal_quota_type_t   quota_type,
                                   fsal_uid_t          fsal_uid)      /* IN */
{
  struct dqblk fs_quota;
  char fs_spec[MAXPATHLEN];
  
  if(!path )
    ReturnCode(ERR_FSAL_FAULT, 0);

  if(fsal_internal_path2fsname( path, fs_spec) == -1)
    ReturnCode(ERR_FSAL_INVAL, 0);

  if( fsal_uid == 0 ) /* No quota for root */
    ReturnCode(ERR_FSAL_NO_ERROR, 0) ;
  
  memset((char *)&fs_quota, 0, sizeof(struct dqblk));

  if(quotactl(FSAL_QCMD(Q_GETQUOTA, USRQUOTA), fs_spec, fsal_uid, (caddr_t) & fs_quota) < 0 )
    ReturnCode(posix2fsal_error(errno), errno);
  
  switch( quota_type )
   {
        case FSAL_QUOTA_BLOCKS:
          if( fs_quota.dqb_curspace > fs_quota.dqb_bhardlimit )
            ReturnCode( ERR_FSAL_DQUOT, EDQUOT ) ;
          
          break ;

        case FSAL_QUOTA_INODES:
          if( fs_quota.dqb_curinodes > fs_quota.dqb_ihardlimit )
            ReturnCode( ERR_FSAL_DQUOT, EDQUOT ) ;
          
          break ;
   } /* switch( quota_type ) */
 
  ReturnCode(ERR_FSAL_NO_ERROR, 0) ;
} /* XFSFSAL_check_quota */

/**
 * VFSFSAL_getattrs:
 * Get attributes for the object specified by its filehandle.
 *
 * \param filehandle (input):
 *        The handle of the object to get parameters.
 * \param cred (input):
 *        Authentication context for the operation (user,...).
 * \param object_attributes (mandatory input/output):
 *        The retrieved attributes for the object.
 *        As input, it defines the attributes that the caller
 *        wants to retrieve (by positioning flags into this structure)
 *        and the output is built considering this input
 *        (it fills the structure according to the flags it contains).
 *
 * \return Major error codes :
 *        - ERR_FSAL_NO_ERROR     (no error)
 *        - Another error code if an error occured.
 */
fsal_status_t VFSFSAL_getattrs(fsal_handle_t * p_filehandle,       /* IN */
                            fsal_op_context_t * p_context,      /* IN */
                            fsal_attrib_list_t * p_object_attributes    /* IN/OUT */
    )
{
  fsal_status_t st;
  int rc = 0 ;
  int errsv;
  struct stat buffstat;

  /* sanity checks.
   * note : object_attributes is mandatory in VFSFSAL_getattrs.
   */
  if(!p_filehandle || !p_context || !p_object_attributes)
    Return(ERR_FSAL_FAULT, 0, INDEX_FSAL_getattrs);

  TakeTokenFSCall();
  rc = vfs_stat_by_handle( ((vfsfsal_op_context_t *)p_context)->export_context->mount_root_fd,
                           &((vfsfsal_handle_t *)p_filehandle)->data.vfs_handle,
                           &buffstat ) ;
  errsv = errno;
  ReleaseTokenFSCall();

  if( rc == -1 )
    Return(posix2fsal_error(errsv), errsv, INDEX_FSAL_getattrs);

  /* convert attributes */
  st = posix2fsal_attributes(&buffstat, p_object_attributes);
  if(FSAL_IS_ERROR(st))
    {
      FSAL_CLEAR_MASK(p_object_attributes->asked_attributes);
      FSAL_SET_MASK(p_object_attributes->asked_attributes, FSAL_ATTR_RDATTR_ERR);
      ReturnStatus(st, INDEX_FSAL_getattrs);
    }

  Return(ERR_FSAL_NO_ERROR, 0, INDEX_FSAL_getattrs);

}

fsal_status_t fsal_internal_get_handle_at(int dfd,      /* IN */
                                          const char *name,     /* IN */
                                          fsal_handle_t *p_handle      /* OUT
                                                                         */ )
{
  int errsrv = 0 ;

  if( !name || !p_handle )
    ReturnCode(ERR_FSAL_FAULT, 0);

  memset(p_handle, 0, sizeof(vfsfsal_handle_t));

  LogFullDebug(COMPONENT_FSAL, "get handle at for %s", name);

  ((vfsfsal_handle_t *)p_handle)->data.vfs_handle.handle_bytes = VFS_HANDLE_LEN ;
  if( vfs_name_by_handle_at( dfd, name, &((vfsfsal_handle_t *)p_handle)->data.vfs_handle ) != 0 )
   {
      errsrv = errno;
      ReturnCode(posix2fsal_error(errsrv), errsrv);
   }

  ReturnCode(ERR_FSAL_NO_ERROR, 0);
} /* fsal_internal_get_handle_at */

/**
 *  Removes a xattr by Name
 *
 * \param p_objecthandle Handle of the object you want to get attribute for.
 * \param p_context pointer to the current security context.
 * \param xattr_name xattr's name
 */
fsal_status_t ZFSFSAL_RemoveXAttrByName(fsal_handle_t * obj_handle,    /* IN */
                                     fsal_op_context_t * p_context,     /* IN */
                                     const fsal_name_t * xattr_name)    /* IN */
{
  int rc;
  creden_t cred;
  zfsfsal_handle_t * p_objecthandle = (zfsfsal_handle_t *)obj_handle;

  /* Hook to prevent any modification in the snapshots */
  if(p_objecthandle->data.i_snap != 0)
    Return(ERR_FSAL_ROFS, 0, INDEX_FSAL_SetXAttrValue);
  cred.uid = p_context->credential.user;
  cred.gid = p_context->credential.group;

  TakeTokenFSCall();
  rc = libzfswrap_removexattr(((zfsfsal_op_context_t *)p_context)->export_context->p_vfs, &cred,
                              p_objecthandle->data.zfs_handle, xattr_name->name);
  ReleaseTokenFSCall();

  if(rc)
    Return(posix2fsal_error(rc), 0, INDEX_FSAL_SetXAttrValue);
  Return(ERR_FSAL_NO_ERROR, 0, INDEX_FSAL_SetXAttrValue);
}

static fsal_status_t lustre_linkfile(struct fsal_obj_handle *obj_hdl,
				     struct fsal_obj_handle *destdir_hdl,
				     const char *name)
{
	struct lustre_fsal_obj_handle *myself, *destdir;
	char srcpath[MAXPATHLEN];
	char destdirpath[MAXPATHLEN];
	char destnamepath[MAXPATHLEN];
	int rc = 0;
	fsal_errors_t fsal_error = ERR_FSAL_NO_ERROR;

	if (!op_ctx->fsal_export->ops->
	    fs_supports(op_ctx->fsal_export, fso_link_support)) {
		fsal_error = ERR_FSAL_NOTSUPP;
		goto out;
	}
	myself =
	    container_of(obj_hdl, struct lustre_fsal_obj_handle, obj_handle);
	lustre_handle_to_path(obj_hdl->fs->path,
			      myself->handle, srcpath);

	destdir =
	    container_of(destdir_hdl, struct lustre_fsal_obj_handle,
			 obj_handle);
	lustre_handle_to_path(obj_hdl->fs->path,
			      destdir->handle, destdirpath);

	snprintf(destnamepath, MAXPATHLEN, "%s/%s", destdirpath, name);

	rc = CRED_WRAP(op_ctx->creds, int, link, srcpath, destnamepath);
	if (rc == -1) {
		rc = errno;
		fsal_error = posix2fsal_error(rc);
	}
 out:
	return fsalstat(fsal_error, rc);
}

/** @fn fsal_status_t
 *	GPFSFSAL_alloc(int fd, uint64_t offset, uint64_t length, bool allocate)
 *  @brief Perform a de/allocc operation on an opened file.
 *  @param fd The file descriptor returned by FSAL_open.
 *  @param offset Offset
 *  @param length Length
 *  @param allocate Allocate
 *
 *  @return ERR_FSAL_NO_ERROR on success, error otherwise
 */
fsal_status_t
GPFSFSAL_alloc(int fd, uint64_t offset, uint64_t length, bool allocate)
{
	struct alloc_arg aarg = {0};
	int errsv;
	int rc;

	aarg.fd = fd;
	aarg.offset = offset;
	aarg.length = length;
	aarg.options = (allocate) ? IO_ALLOCATE : IO_DEALLOCATE;

	fsal_set_credentials(op_ctx->creds);
	rc = gpfs_ganesha(OPENHANDLE_ALLOCATE_BY_FD, &aarg);
	errsv = errno;
	fsal_restore_ganesha_credentials();

	if (rc == -1) {
		if (errsv == EUNATCH)
			LogFatal(COMPONENT_FSAL, "GPFS Returned EUNATCH");
		return fsalstat(posix2fsal_error(errsv), errsv);
	}
	return fsalstat(ERR_FSAL_NO_ERROR, 0);
}

/**
 * FSAL_dynamic_fsinfo:
 * Return dynamic filesystem info such as
 * used size, free size, number of objects...
 *
 * \param filehandle (input):
 *        Handle of an object in the filesystem
 *        whom info is to be retrieved.
 * \param p_context (input):
 *        Authentication context for the operation (user,...).
 * \param dynamicinfo (output):
 *        Pointer to the static info of the filesystem.
 *
 * \return Major error codes:
 *      - ERR_FSAL_NO_ERROR     (no error)
 *      - ERR_FSAL_FAULT        (a NULL pointer was passed as mandatory argument)
 *      - Other error codes can be returned :
 *        ERR_FSAL_IO, ...
 */
fsal_status_t ZFSFSAL_dynamic_fsinfo(fsal_handle_t * filehandle,   /* IN */
                                  fsal_op_context_t * p_context,        /* IN */
                                  fsal_dynamicfsinfo_t * dynamicinfo    /* OUT */
    )
{

  /* sanity checks. */
  if(!filehandle || !dynamicinfo || !p_context)
    Return(ERR_FSAL_FAULT, 0, INDEX_FSAL_dynamic_fsinfo);

  TakeTokenFSCall();

  struct statvfs statfs;
  int rc = libzfswrap_statfs(((zfsfsal_op_context_t *)p_context)->export_context->p_vfs,
			     &statfs);

  ReleaseTokenFSCall();

  /* >> interpret returned status << */
  if(rc)
    Return(posix2fsal_error(rc), 0, INDEX_FSAL_dynamic_fsinfo);

  dynamicinfo->total_bytes = statfs.f_frsize * statfs.f_blocks;
  dynamicinfo->free_bytes = statfs.f_frsize * statfs.f_bfree;
  dynamicinfo->avail_bytes = statfs.f_frsize * statfs.f_bavail;

  dynamicinfo->total_files = statfs.f_files;
  dynamicinfo->free_files = statfs.f_ffree;
  dynamicinfo->avail_files = statfs.f_favail;

  dynamicinfo->time_delta.seconds = 1;
  dynamicinfo->time_delta.nseconds = 0;


  Return(ERR_FSAL_NO_ERROR, 0, INDEX_FSAL_dynamic_fsinfo);
}

fsal_status_t fsal_internal_fd2handle( vfsfsal_op_context_t * p_context,
                                       int fd, 
				       vfsfsal_handle_t * phandle)
{
  int rc = 0 ;
  int errsv = 0 ; 
  int mnt_id = 0 ;


  phandle->data.vfs_handle.handle_bytes = VFS_HANDLE_LEN ;
  if( ( rc = vfs_fd_to_handle( fd, &phandle->data.vfs_handle, &mnt_id ) ) )
   ReturnCode(posix2fsal_error(errsv), errsv);

#if 0 
  {
    char str[1024] ;
    sprint_mem( str, phandle->data.vfs_handle.handle, phandle->data.vfs_handle.handle_bytes ) ;
    printf( "=====> fsal_internal_fd2handle: type=%u bytes=%u|%s\n",  
            phandle->data.vfs_handle.handle_type, phandle->data.vfs_handle.handle_bytes, str ) ;
  }
#endif

  ReturnCode(ERR_FSAL_NO_ERROR, 0);
}                               /* fsal_internal_fd2handle */

/**
 * VFSFSAL_getattrs_descriptor:
 * Get attributes for the object specified by its descriptor or by it's filehandle.
 *
 * \param p_file_descriptor (input):
 *        The file descriptor of the object to get parameters.
 * \param p_filehandle (input):
 *        The handle of the object to get parameters.
 * \param p_context (input):
 *        Authentication context for the operation (user,...).
 * \param p_object_attributes (mandatory input/output):
 *        The retrieved attributes for the object.
 *        As input, it defines the attributes that the caller
 *        wants to retrieve (by positioning flags into this structure)
 *        and the output is built considering this input
 *        (it fills the structure according to the flags it contains).
 *
 * \return Major error codes :
 *        - ERR_FSAL_NO_ERROR     (no error)
 *        - Another error code if an error occured.
 */
fsal_status_t VFSFSAL_getattrs_descriptor(fsal_file_t * p_file_descriptor,     /* IN */
                                           fsal_handle_t * p_filehandle,        /* IN */
                                           fsal_op_context_t * p_context,       /* IN */
                                           fsal_attrib_list_t * p_object_attributes /* IN/OUT */
    )
{
  fsal_status_t st;
  struct stat64 buffstat;
  int rc, errsv;

  /* sanity checks.
   * note : object_attributes is mandatory in VFSFSAL_getattrs.
   */
  if(!p_file_descriptor || !p_filehandle || !p_context || !p_object_attributes)
    Return(ERR_FSAL_FAULT, 0, INDEX_FSAL_getattrs_descriptor);

  TakeTokenFSCall();
  rc = fstat64(((vfsfsal_file_t *)p_file_descriptor)->fd, &buffstat);
  errsv = errno;
  ReleaseTokenFSCall();

  if(rc == -1)
    Return(posix2fsal_error(errsv), errsv, INDEX_FSAL_getattrs_descriptor);

  /* convert attributes */
  st = posixstat64_2_fsal_attributes(&buffstat, p_object_attributes);
  if(FSAL_IS_ERROR(st))
    {
      FSAL_CLEAR_MASK(p_object_attributes->asked_attributes);
      FSAL_SET_MASK(p_object_attributes->asked_attributes, FSAL_ATTR_RDATTR_ERR);
      ReturnStatus(st, INDEX_FSAL_getattrs_descriptor);
    }

  Return(ERR_FSAL_NO_ERROR, 0, INDEX_FSAL_getattrs_descriptor);

}

/**
 * FSAL_closedir :
 * Free the resources allocated for reading directory entries.
 *     
 * \param dir_descriptor (input):
 *        Pointer to a directory descriptor filled by FSAL_opendir.
 * 
 * \return Major error codes :
 *        - ERR_FSAL_NO_ERROR     (no error)
 *        - Another error code if an error occured.
 */
fsal_status_t POSIXFSAL_closedir(fsal_dir_t * dir_descriptor     /* IN */  )
{
  posixfsal_dir_t * p_dir_descriptor = (posixfsal_dir_t *) dir_descriptor;
  int rc;

  /* sanity checks */
  if(!p_dir_descriptor)
    Return(ERR_FSAL_FAULT, 0, INDEX_FSAL_closedir);

#ifdef _USE_POSIXDB_READDIR_BLOCK
  if(p_dir_descriptor->p_dbentries)
    Mem_Free(p_dir_descriptor->p_dbentries);
#endif

  rc = closedir(p_dir_descriptor->p_dir);
  if(rc != 0)
    Return(posix2fsal_error(errno), errno, INDEX_FSAL_closedir);

  /* fill dir_descriptor with zeros */
  memset(p_dir_descriptor, 0, sizeof(posixfsal_dir_t));

  Return(ERR_FSAL_NO_ERROR, 0, INDEX_FSAL_closedir);

}

fsal_status_t tank_list_ext_attrs(struct fsal_obj_handle *obj_hdl,
				  unsigned int argcookie,
				  fsal_xattrent_t *xattrs_tab,
				  unsigned int xattrs_tabsize,
				  unsigned int *p_nb_returned, int *end_of_list)
{
	unsigned int index;
	unsigned int out_index;
	unsigned int cookie = argcookie;
	struct zfs_fsal_obj_handle *obj_handle = NULL;

	char names[MAXPATHLEN], *ptr;
	size_t namesize;
	int xattr_idx;
	int retval;
	creden_t cred;

	/* sanity checks */
	if (!obj_hdl || !xattrs_tab || !p_nb_returned || !end_of_list)
		return fsalstat(ERR_FSAL_FAULT, 0);

	obj_handle =
	    container_of(obj_hdl, struct zfs_fsal_obj_handle, obj_handle);

	cred.uid = op_ctx->creds->caller_uid;
	cred.gid = op_ctx->creds->caller_gid;

	/* Deal with special cookie */
	if (cookie == XATTR_RW_COOKIE)
		cookie = XATTR_COUNT;

	for (index = cookie, out_index = 0;
	     index < XATTR_COUNT && out_index < xattrs_tabsize; index++) {
		if (do_match_type(xattr_list[index].flags,
				  obj_hdl->attributes.type)) {
			/* fills an xattr entry */
			xattrs_tab[out_index].xattr_id = index;
			strncpy(xattr_list[index].xattr_name,
				xattrs_tab[out_index].xattr_name, MAXNAMLEN);
			xattr_list[index].xattr_name[MAXNAMLEN] = '\0';
			xattrs_tab[out_index].xattr_cookie = index + 1;

			/* set asked attributes (all supported) */
			xattrs_tab[out_index].attributes.mask =
			    obj_hdl->attributes.mask;

			if (file_attributes_to_xattr_attrs(&obj_hdl->attributes,
							   &xattrs_tab
							   [out_index]
							   .attributes,
							   index)) {
				/* set error flag */
				xattrs_tab[out_index].attributes.mask =
				    ATTR_RDATTR_ERR;
			}

			/* next output slot */
			out_index++;
		}
	}

	/* save a call if output array is full */
	if (out_index == xattrs_tabsize) {
		*end_of_list = false;
		*p_nb_returned = out_index;
		return fsalstat(ERR_FSAL_NO_ERROR, 0);
	}

	/* get the path of the file in Lustre */

	/* get xattrs */

	retval = libzfswrap_listxattr(ZFSFSAL_GetVFS(obj_handle->handle),
				      &cred,
				      obj_handle->handle->zfs_handle,
				      (char **)&names, &namesize);
	if (retval)
		return fsalstat(posix2fsal_error(retval), retval);

	if (namesize > 0) {
		size_t len = 0;

		errno = 0;

		for (ptr = names, xattr_idx = 0;
		     (ptr < names + namesize) && (out_index < xattrs_tabsize);
		     xattr_idx++, ptr += len + 1) {
			len = strlen(ptr);
			index = XATTR_COUNT + xattr_idx;

			/* skip if index is before cookie */
			if (index < cookie)
				continue;

			/* fills an xattr entry */
			xattrs_tab[out_index].xattr_id = index;
			strncpy(xattrs_tab[out_index].xattr_name, ptr, len + 1);
			xattrs_tab[out_index].xattr_cookie = index + 1;

			/* set asked attributes (all supported) */
//.........这里部分代码省略.........