bool uid2grp( uid_t uid, struct group_data ** pgdata ) 
{
  bool success;
   
  struct gsh_buffdesc name;
  struct gsh_buffdesc * pname = &name ;

  pthread_rwlock_rdlock(&uid2grp_user_lock);

  success = uid2grp_lookup_by_uid( uid, &pname, pgdata );

  pthread_rwlock_unlock(&uid2grp_user_lock);

  if (success)
    return true;
  else
    {
      if( pwentuid2grp( uid, &name, *pgdata ) )
       {
         pthread_rwlock_wrlock(&uid2grp_user_lock);

         success = uid2grp_add_user(&name, uid, *pgdata ) ;

         pthread_rwlock_unlock(&uid2grp_user_lock);

         if (!success)
	   LogMajor(COMPONENT_IDMAPPER, "uid2grp %u", uid ) ;
         return true;
       }
      else
	return false ;
    }
}

bool name2grp(const struct gsh_buffdesc *name, struct group_data **pgdata)
{
	bool success;
	uid_t uid = -1;

	pthread_rwlock_rdlock(&uid2grp_user_lock);

	success = uid2grp_lookup_by_uname(name, &uid, pgdata);

	pthread_rwlock_unlock(&uid2grp_user_lock);

	if (success)
		return true;
	else {
		/* Something we can mutate and count on as terminated */
		char *namebuff = alloca(name->len + 1);

		memcpy(namebuff, name->addr, name->len);
		*(namebuff + name->len) = '\0';

		if (pwentname2grp(namebuff, &uid, *pgdata)) {
			pthread_rwlock_wrlock(&uid2grp_user_lock);

			success = uid2grp_add_user(name, uid, *pgdata);

			pthread_rwlock_unlock(&uid2grp_user_lock);

			if (!success)
				LogMajor(COMPONENT_IDMAPPER, "name2grp %s",
					 namebuff);
			return true;
		} else
			return false;
	}
}

/*
 *  Process NFSv4 ACLs passed in setattr call
 */
fsal_status_t glusterfs_process_acl(struct glfs *fs,
				    struct glfs_object *object,
				    struct attrlist *attrs,
				    glusterfs_fsal_xstat_t *buffxstat)
{
	if (attrs->acl) {
		LogDebug(COMPONENT_FSAL, "setattr acl = %p",
			 attrs->acl);

		/* Convert FSAL ACL to POSIX ACL */
		buffxstat->e_acl = fsal_acl_2_posix_acl(attrs->acl,
						ACL_TYPE_ACCESS);
		if (!buffxstat->e_acl) {
			LogMajor(COMPONENT_FSAL,
				 "failed to set access type posix acl");
			return fsalstat(ERR_FSAL_FAULT, 0);
		}
		/* For directories consider inherited acl too */
		if (buffxstat->is_dir) {
			buffxstat->i_acl = fsal_acl_2_posix_acl(attrs->acl,
							ACL_TYPE_DEFAULT);
			if (!buffxstat->i_acl)
				LogDebug(COMPONENT_FSAL,
				"inherited acl is not defined for directory");
		}

	} else {
		LogCrit(COMPONENT_FSAL, "setattr acl is NULL");
		return fsalstat(ERR_FSAL_FAULT, 0);
	}
	return fsalstat(ERR_FSAL_NO_ERROR, 0);
}

/**
 * @brief Read from a data-server handle.
 *
 * NFSv4.1 data server handles are disjount from normal
 * filehandles (in Ganesha, there is a ds_flag in the filehandle_v4_t
 * structure) and do not get loaded into cache_inode or processed the
 * normal way.
 *
 * @param[in]  ds_pub           FSAL DS handle
 * @param[in]  req_ctx          Credentials
 * @param[in]  stateid          The stateid supplied with the READ operation,
 *                              for validation
 * @param[in]  offset           The offset at which to read
 * @param[in]  requested_length Length of read requested (and size of buffer)
 * @param[out] buffer           The buffer to which to store read data
 * @param[out] supplied_length  Length of data read
 * @param[out] eof              True on end of file
 *
 * @return An NFSv4.1 status code.
 */
static nfsstat4 ds_read(struct fsal_ds_handle *const ds_pub,
			struct req_op_context *const req_ctx,
			const stateid4 *stateid, const offset4 offset,
			const count4 requested_length, void *const buffer,
			count4 * const supplied_length,
			bool * const end_of_file)
{
	/* The private DS handle */
	struct glfs_ds_handle *ds =
		container_of(ds_pub, struct glfs_ds_handle, ds);
	int    rc = 0;
	struct glusterfs_export *glfs_export =
	container_of(ds_pub->pds->mds_fsal_export,
		     struct glusterfs_export, export);

	if (ds->glhandle == NULL)
		LogDebug(COMPONENT_PNFS, "ds_read glhandle NULL");

	rc = glfs_h_anonymous_read(glfs_export->gl_fs->fs, ds->glhandle,
				   buffer, requested_length, offset);
	if (rc < 0) {
		LogMajor(COMPONENT_PNFS, "Read failed on DS");
		return posix2nfs4_error(-rc);
	}

	*supplied_length = rc;
	if (rc == 0 || rc < requested_length)
		*end_of_file = true;


	return NFS4_OK;
}

/**
 * FSAL_DigestHandle :
 *  Convert an posixfsal_handle_t to a buffer
 *  to be included into NFS handles,
 *  or another digest.
 *
 * \param output_type (input):
 *        Indicates the type of digest to do.
 * \param in_fsal_handle (input):
 *        The handle to be converted to digest.
 * \param fh_desc (input/output):
 *        The buffer where the digest is to be stored.
 *
 * \return The major code is ERR_FSAL_NO_ERROR is no error occured.
 *         Else, it is a non null value.
 */
fsal_status_t POSIXFSAL_DigestHandle(fsal_export_context_t * expcontext, /* IN */
                                     fsal_digesttype_t output_type,     /* IN */
                                     fsal_handle_t * in_fsal_handle,     /* IN */
                                     struct fsal_handle_desc *fh_desc     /* IN/OUT */
    )
{
  posixfsal_export_context_t * p_expcontext
    = (posixfsal_export_context_t *) expcontext;
  posixfsal_handle_t * p_in_fsal_handle = (posixfsal_handle_t *) in_fsal_handle;
  size_t fh_size;
  caddr_t fh_data;

  /* sanity checks */
  if(!p_in_fsal_handle || !fh_desc || !fh_desc->start || !p_expcontext)
    ReturnCode(ERR_FSAL_FAULT, 0);

  switch (output_type)
    {
    /* NFS handle digest */
    case FSAL_DIGEST_NFSV2:
    case FSAL_DIGEST_NFSV3:
    case FSAL_DIGEST_NFSV4:
      fh_size = sizeof(p_in_fsal_handle->data.id) + sizeof(p_in_fsal_handle->data.ts);
      fh_data = (caddr_t) p_in_fsal_handle;
      break;

    /* FileId digest */
    case FSAL_DIGEST_FILEID2:
    case FSAL_DIGEST_FILEID3:
    case FSAL_DIGEST_FILEID4:
      /* XXX: does fh_desc need to be padded to FSAL_DIGEST_SIZE_FILEID{2,3,4} ?
       *      or is setting fh_size = sizeof(ino_t) sufficient? */
      fh_size = sizeof(p_in_fsal_handle->data.info.inode);
      fh_data = (caddr_t) &p_in_fsal_handle->data.info.inode;
      break;

    /* Nodetype digest. */
    case FSAL_DIGEST_NODETYPE:
      /* XXX: should fh_desc be padded to FSAL_DIGEST_SIZE_NODETYPE ? */
      fh_size = sizeof(p_in_fsal_handle->data.info.ftype);
      fh_data = (caddr_t) &p_in_fsal_handle->data.info.ftype;
      break;

    default:
      ReturnCode(ERR_FSAL_SERVERFAULT, 0);
    }

  if(fh_desc->len < fh_size)
    {
      LogMajor( COMPONENT_FSAL, "POSIX DigestHandle: too small for handle.  need %lu, have %lu",
                fh_size, fh_desc->len);
      ReturnCode(ERR_FSAL_TOOSMALL, 0);
    }

  memcpy(fh_desc->start, fh_data, fh_size);
  fh_desc->len = fh_size;

  ReturnCode(ERR_FSAL_NO_ERROR, 0);
}

/**
 * @brief Initialize asynchronous request system
 *
 * @return State status.
 */
state_status_t state_async_init(void)
{
	int rc = 0;
	struct fridgethr_params frp;

	memset(&frp, 0, sizeof(struct fridgethr_params));
	frp.thr_max = 1;
	frp.deferment = fridgethr_defer_queue;

	rc = fridgethr_init(&state_async_fridge, "State_Async", &frp);

	if (rc != 0) {
		LogMajor(COMPONENT_STATE,
			 "Unable to initialize state async thread fridge: %d",
			 rc);
		return STATE_INIT_ENTRY_FAILED;
	}

	memset(&frp, 0, sizeof(struct fridgethr_params));
	frp.thr_max = 1;
	frp.thr_min = 1;
	frp.thread_delay = nfs_param.core_param.blocked_lock_poller_interval;
	frp.flavor = fridgethr_flavor_looper;

	rc = fridgethr_init(&state_poll_fridge, "state_poll", &frp);

	if (rc != 0) {
		LogMajor(COMPONENT_STATE,
			 "Unable to initialize state blocked lock polling thread fridge: %d",
			 rc);
		return STATE_INIT_ENTRY_FAILED;
	}

	rc = fridgethr_submit(state_poll_fridge, blocked_lock_polling, NULL);

	if (rc != 0) {
		LogMajor(COMPONENT_STATE,
			 "Unable to start blocked lock polling thread, error code %d.",
			 rc);
		return STATE_INIT_ENTRY_FAILED;
	}

	return STATE_SUCCESS;
}

void _9p_tcp_process_request(struct _9p_request_data *req9p)
{
	u32 outdatalen = 0;
	int rc = 0;
	char replydata[_9P_MSG_SIZE];

	rc = _9p_process_buffer(req9p, replydata, &outdatalen);
	if (rc != 1) {
		LogMajor(COMPONENT_9P,
			 "Could not process 9P buffer on socket #%lu",
			 req9p->pconn->trans_data.sockfd);
	} else {
		if (tcp_conn_send(req9p->pconn, replydata, outdatalen, 0) !=
		    outdatalen)
			LogMajor(COMPONENT_9P,
				 "Could not send 9P/TCP reply correclty on socket #%lu",
				 req9p->pconn->trans_data.sockfd);
	}
	_9p_DiscardFlushHook(req9p);
}				/* _9p_process_request */

int FSAL_LoadLibrary(char *path)
{
  void *handle;
  char *error;

  LogEvent(COMPONENT_FSAL, "Load shared FSAL : %s", path);

  if((handle = dlopen(path, RTLD_LAZY)) == NULL)
    {
      LogMajor(COMPONENT_FSAL,
               "FSAL_LoadLibrary: could not load fsal: %s",
               dlerror());
      return 0;
    }

  /* Clear any existing error : dlerror will be used to check if dlsym succeeded or not */
  dlerror();

  /* Map FSAL_GetFunctions */
  *(void **)(&getfunctions) = dlsym(handle, "FSAL_GetFunctions");
  if((error = dlerror()) != NULL)
    {
      LogMajor(COMPONENT_FSAL,
               "FSAL_LoadLibrary: Could not map symbol FSAL_GetFunctions:%s",
               error);
      return 0;
    }
  /* Map FSAL_GetConsts */
  *(void **)(&getconsts) = dlsym(handle, "FSAL_GetConsts");
  if((error = dlerror()) != NULL)
    {
      LogMajor(COMPONENT_FSAL,
               "FSAL_LoadLibrary: Could not map symbol FSAL_GetConsts:%s",
               error);
      return 0;
    }

  return 1;
}                               /* FSAL_LoadLibrary */

/*
 * The data server address will be send from here
 *
 * The information about the first server present
 * in the PATH_INFO_KEY will be returned, since
 * entire file is consistent over the servers
 * (Striped volumes are not considered right now)
 *
 * On success, returns zero with ip address of
 * the server will be send
 */
int
glfs_get_ds_addr(struct glfs *fs, struct glfs_object *object, uint32_t *ds_addr)
{
	int             ret                  = 0;
	char            pathinfo[1024]       = {0, };
	char            hostname[1024]       = {0, };
	struct addrinfo hints, *res;
	struct in_addr  addr                 = {0, };
	const char      *pathinfokey         = "trusted.glusterfs.pathinfo";

	ret = glfs_h_getxattrs(fs, object, pathinfokey, pathinfo, 1024);

	LogDebug(COMPONENT_PNFS, "pathinfo %s", pathinfo);

	ret = select_ds(object, pathinfo, hostname, sizeof(hostname));
	if (ret) {
		LogMajor(COMPONENT_PNFS, "No DS found");
		goto out;
	}

	memset(&hints, 0, sizeof(hints));
	hints.ai_socktype = SOCK_STREAM;
	hints.ai_family = AF_INET;
	ret = getaddrinfo(hostname, NULL, &hints, &res);
	if (ret != 0) {
		LogMajor(COMPONENT_PNFS, "error %d\n", ret);
		goto out;
	}

	addr.s_addr = ((struct sockaddr_in *)(res->ai_addr))->sin_addr.s_addr;

	LogDebug(COMPONENT_PNFS, "ip address : %s", inet_ntoa(addr));

	freeaddrinfo(res);
out:

	*ds_addr = addr.s_addr;
	return ret;
}

/**
 *
 * fsal_internal_ClientReconnect: Redo a previously lost connection to the server.
 *
 * edo a previously lost connection to the server.
 *
 * @param p_thr_context [INOUT]  pointer to the FSAL thread context.
 *
 * @return 0 if successful, -1 if failed
 *
 */
int fsal_internal_ClientReconnect(proxyfsal_op_context_t * p_thr_context)
{
  int rc;
  struct timeval timeout = TIMEOUTRPC;
  struct sockaddr_in addr_rpc;
  fsal_status_t fsal_status;


  memset(&addr_rpc, 0, sizeof(addr_rpc));
  addr_rpc.sin_port = p_thr_context->srv_port;
  addr_rpc.sin_family = AF_INET;
  addr_rpc.sin_addr.s_addr = p_thr_context->srv_addr;
  int priv_port = 0 ;

  LogEvent( COMPONENT_FSAL, "Remote server lost, trying to reconnect to remote server" ) ;

  /* First of all, close the formerly opened socket that is now useless */
  if( close( p_thr_context->socket ) == -1 )
    LogMajor( COMPONENT_FSAL,
              "FSAL RECONNECT : got POSIX error %u while closing socket in fsal_internal_ClientReconnect", errno ) ;
  
  // clnt_destroy(p_thr_context->rpc_client);
  // clnt_destroy makes the server segfaults if no server exists on the other side

  if(!strcmp(p_thr_context->srv_proto, "udp"))
    {
      if( ( p_thr_context->socket = socket( AF_INET, SOCK_DGRAM, IPPROTO_UDP ) ) < 0)
        return -1;

      if((p_thr_context->rpc_client = clntudp_bufcreate(&addr_rpc,
                                                        p_thr_context->srv_prognum,
                                                        FSAL_PROXY_NFS_V4,
                                                        (struct timeval)
                                                        {
                                                        25, 0},
                                                        &p_thr_context->socket,
                                                        p_thr_context->srv_sendsize,
                                                        p_thr_context->srv_recvsize)) == NULL )
        {

          LogCrit(COMPONENT_FSAL,
               "FSAL RECONNECT : Cannot contact server addr=%u.%u.%u.%u port=%u prognum=%u using NFSv4 protocol",
               (ntohl(p_thr_context->srv_addr) & 0xFF000000) >> 24,
               (ntohl(p_thr_context->srv_addr) & 0x00FF0000) >> 16,
               (ntohl(p_thr_context->srv_addr) & 0x0000FF00) >> 8,
               (ntohl(p_thr_context->srv_addr) & 0x000000FF),
               ntohs(p_thr_context->srv_port), p_thr_context->srv_prognum);

          return -1;
        }
    }

/**
 * FSAL_getattrs.
 *
 * \return Major error codes :
 *        - ERR_FSAL_NO_ERROR     (no error)
 *        - ERR_FSAL_ATTRNOTSUPP  (attribute not supported)
 *        - ERR_FSAL_BADHANDLE    (illegal handle)
 *        - ERR_FSAL_FAULT        (null pointer parameter)
 *        - ERR_FSAL_IO           (corrupted FS)
 *        - ERR_FSAL_NOT_INIT     (ghostfs not initialize)
 *        - ERR_FSAL_SERVERFAULT  (unexpected error)
 */
fsal_status_t FSAL_getattrs(fsal_handle_t * filehandle, /* IN */
                            fsal_op_context_t * p_context,      /* IN */
                            fsal_attrib_list_t * object_attributes      /* IN/OUT */
    )
{

  int rc;
  GHOSTFS_Attrs_t ghost_attrs;
  fsal_attrib_mask_t supp_attr, unsupp_attr;

  /* For logging */
  SetFuncID(INDEX_FSAL_getattrs);

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

  rc = GHOSTFS_GetAttrs((GHOSTFS_handle_t) (*filehandle), &ghost_attrs);

  if(rc)
    Return(ghost2fsal_error(rc), rc, INDEX_FSAL_getattrs);

  /* Credentials are not tested because
   * we consider that if the user got an handle to the object,
   * he has the right for retrieving its attributes.
   */

  /* supported attrs for GHOSTFS */
  supp_attr = GHOSTFS_SUPPORTED_ATTRIBUTES;

  /* tests whether we can supply all asked attributes */
  unsupp_attr = (object_attributes->asked_attributes) & (~supp_attr);
  if(unsupp_attr)
    {
      LogMajor(COMPONENT_FSAL,
               "Unsupported attributes: %#llX removing it from asked attributes ",
               unsupp_attr);
      object_attributes->asked_attributes =
          object_attributes->asked_attributes & (~unsupp_attr);
    }

  /* Fills the output struct */
  ghost2fsal_attrs(object_attributes, &ghost_attrs);

  /* everything has been copied ! */

  Return(ERR_FSAL_NO_ERROR, 0, INDEX_FSAL_getattrs);

}

/**
 * @brief Schedule a lock notification
 *
 * @param[in] block Lock to schedule
 *
 * @return State status.
 */
state_status_t state_block_schedule(state_block_data_t *block)
{
	int rc;

	LogFullDebug(COMPONENT_STATE, "Schedule notification %p", block);

	rc = fridgethr_submit(state_async_fridge, state_blocked_lock_caller,
			      block);

	if (rc != 0)
		LogMajor(COMPONENT_STATE, "Unable to schedule request: %d", rc);

	return rc == 0 ? STATE_SUCCESS : STATE_SIGNAL_ERROR;
}

void TestMajor(int expect, char *buff, log_components_t component, char *string)
{
  char compare[2048];
  sprintf(compare, "%s: MAJOR ERROR: %s", LogComponents[component].comp_str, string);
  buff[0] = '\0';
  LogMajor(component, "%s", string);
  if (expect && strcmp(compare, buff) != 0 || !expect && buff[0] != '\0')
    {
      LogTest("FAILURE: %s produced \"%s\" expected \"%s\"", string, buff, compare);
      exit(1);
    }
  if (expect)
    LogTest("SUCCESS: %s produced \"%s\"", string, buff);
  else
    LogTest("SUCCESS: %s didn't produce anything", string);
}

/* Allocate and fill in group_data structure */
static struct group_data *uid2grp_allocate_by_name(
		const struct gsh_buffdesc *name)
{
	struct passwd p;
	struct passwd *pp;
	char *namebuff = alloca(name->len + 1);
	struct group_data *gdata = NULL;
	char *buff;
	long buff_size;

	memcpy(namebuff, name->addr, name->len);
	*(namebuff + name->len) = '\0';

	buff_size = sysconf(_SC_GETPW_R_SIZE_MAX);
	if (buff_size == -1) {
		LogMajor(COMPONENT_IDMAPPER, "sysconf failure: %d", errno);
		return NULL;
	}

	buff = alloca(buff_size);
	if ((getpwnam_r(namebuff, &p, buff, buff_size, &pp) != 0)
	    || (pp == NULL)) {
		LogEvent(COMPONENT_IDMAPPER, "getpwnam_r %s failed", namebuff);
		return gdata;
	}

	gdata = gsh_malloc(sizeof(struct group_data) + strlen(p.pw_name));
	if (gdata == NULL) {
		LogEvent(COMPONENT_IDMAPPER, "failed to allocate group data");
		return gdata;
	}

	gdata->uname.len = strlen(p.pw_name);
	gdata->uname.addr = (char *)gdata + sizeof(struct group_data);
	memcpy(gdata->uname.addr, p.pw_name, gdata->uname.len);
	gdata->uid = p.pw_uid;
	gdata->gid = p.pw_gid;
	if (!my_getgrouplist_alloc(p.pw_name, p.pw_gid, gdata)) {
		gsh_free(gdata);
		return NULL;
	}

	PTHREAD_MUTEX_init(&gdata->lock, NULL);
	gdata->epoch = time(NULL);
	gdata->refcount = 0;
	return gdata;
}

/**
 * @brief Release a DS object
 *
 * @param[in] obj_pub The object to release
 *
 * @return NFS Status codes.
 */
static void release(struct fsal_ds_handle *const ds_pub)
{
	int    rc                 = 0;
	struct glfs_ds_handle *ds =
		container_of(ds_pub, struct glfs_ds_handle, ds);

	fsal_ds_handle_fini(&ds->ds);
	if (ds->glhandle) {
		rc = glfs_h_close(ds->glhandle);
		if (rc) {
			LogMajor(COMPONENT_PNFS,
				 "glfs_h_close returned error %s(%d)",
				 strerror(errno), errno);
		}
	}
	gsh_free(ds);
}

int SetComponentLogFile(log_components_t component, const char *name)
{
  int newtype, changed;

  if (strcmp(name, "SYSLOG") == 0)
    newtype = SYSLOG;
  else if (strcmp(name, "STDERR") == 0)
    newtype = STDERRLOG;
  else if (strcmp(name, "STDOUT") == 0)
    newtype = STDOUTLOG;
  else if (strcmp(name, "TEST") == 0)
    newtype = TESTLOG;
  else
    newtype = FILELOG;

  if (newtype == FILELOG)
    {
      if (!isValidLogPath(name))
        {
          LogMajor(COMPONENT_LOG, "Could not set default logging to %s", name);
          errno = EINVAL;
          return -1;
        }
      }

  changed = (newtype != LogComponents[component].comp_log_type) ||
            (newtype == FILELOG && strcmp(name, LogComponents[component].comp_log_file) != 0);

  if (component != COMPONENT_LOG && changed)
    LogChanges("Changing log destination for %s from %s to %s",
               LogComponents[component].comp_name,
               LogComponents[component].comp_log_file,
               name);

  LogComponents[component].comp_log_type = newtype;
  strncpy(LogComponents[component].comp_log_file, name, MAXPATHLEN);

  if (component == COMPONENT_LOG && changed)
    LogChanges("Changing log destination for %s from %s to %s",
               LogComponents[component].comp_name,
               LogComponents[component].comp_log_file,
               name);

  return 0;
}                               /* SetComponentLogFile */

static fsal_status_t handle_digest(const struct fsal_obj_handle *obj_hdl,
				   fsal_digesttype_t output_type,
				   struct gsh_buffdesc *fh_desc)
{
	fsal_status_t status = { ERR_FSAL_NO_ERROR, 0 };
	size_t fh_size;
	struct glusterfs_handle *objhandle;
#ifdef GLTIMING
	struct timespec s_time, e_time;

	now(&s_time);
#endif

	if (!fh_desc)
		return fsalstat(ERR_FSAL_FAULT, 0);

	objhandle = container_of(obj_hdl, struct glusterfs_handle, handle);

	switch (output_type) {
	case FSAL_DIGEST_NFSV2:
	case FSAL_DIGEST_NFSV3:
	case FSAL_DIGEST_NFSV4:
		fh_size = GLAPI_HANDLE_LENGTH;
		if (fh_desc->len < fh_size) {
			LogMajor(COMPONENT_FSAL,
				 "Space too small for handle.  need %lu, have %lu",
				 fh_size, fh_desc->len);
			status.major = ERR_FSAL_TOOSMALL;
			goto out;
		}
		memcpy(fh_desc->addr, objhandle->globjhdl, fh_size);
		break;
	default:
		status.major = ERR_FSAL_SERVERFAULT;
		goto out;
	}

	fh_desc->len = fh_size;
 out:
#ifdef GLTIMING
	now(&e_time);
	latency_update(&s_time, &e_time, lat_handle_digest);
#endif
	return status;
}

/**
 * @brief Commit a byte range to a DS handle.
 *
 * NFSv4.1 data server filehandles are disjount from normal
 * filehandles (in Ganesha, there is a ds_flag in the filehandle_v4_t
 * structure) and do not get loaded into cache_inode or processed the
 * normal way.
 *
 * @param[in]  ds_pub    FSAL DS handle
 * @param[in]  req_ctx   Credentials
 * @param[in]  offset    Start of commit window
 * @param[in]  count     Length of commit window
 * @param[out] writeverf Write verifier
 *
 * @return An NFSv4.1 status code.
 */
static nfsstat4 ds_commit(struct fsal_ds_handle *const ds_pub,
			  struct req_op_context *const req_ctx,
			  const offset4 offset, const count4 count,
			  verifier4 * const writeverf)
{
	memset(writeverf, 0, NFS4_VERIFIER_SIZE);
	struct glfs_ds_handle *ds =
		container_of(ds_pub, struct glfs_ds_handle, ds);
	int rc = 0;
	fsal_status_t status = { ERR_FSAL_NO_ERROR, 0 };

	if (ds->stability_got == FILE_SYNC4) {
		struct glusterfs_export *glfs_export =
			container_of(ds_pub->pds->mds_fsal_export,
				     struct glusterfs_export, export);
		struct glfs_fd *glfd = NULL;

		SET_GLUSTER_CREDS(glfs_export, &op_ctx->creds->caller_uid,
				  &op_ctx->creds->caller_gid,
				  op_ctx->creds->caller_glen,
				  op_ctx->creds->caller_garray);

		glfd = glfs_h_open(glfs_export->gl_fs->fs, ds->glhandle,
				   O_RDWR);
		if (glfd == NULL) {
			LogDebug(COMPONENT_PNFS, "glfd in ds_handle is NULL");
			SET_GLUSTER_CREDS(glfs_export, NULL, NULL, 0, NULL);
			return NFS4ERR_SERVERFAULT;
		}

		rc = glfs_fsync(glfd);
		if (rc != 0)
			LogMajor(COMPONENT_PNFS, "ds_commit() failed  %d", -rc);
		rc = glfs_close(glfd);
		if (rc != 0)
			LogDebug(COMPONENT_PNFS, "status after close %d", -rc);

		SET_GLUSTER_CREDS(glfs_export, NULL, NULL, 0, NULL);
	}

	if ((rc != 0) || (status.major != ERR_FSAL_NO_ERROR))
		return NFS4ERR_INVAL;

	return NFS4_OK;
}

/**
 *
 * @brief Write to a data-server handle.
 *
 * @param[in]  ds_pub           FSAL DS handle
 * @param[in]  req_ctx          Credentials
 * @param[in]  stateid          The stateid supplied with the READ operation,
 *                              for validation
 * @param[in]  offset           The offset at which to read
 * @param[in]  write_length     Length of write requested (and size of buffer)
 * @param[out] buffer           The buffer to which to store read data
 * @param[in]  stability wanted Stability of write
 * @param[out] written_length   Length of data written
 * @param[out] writeverf        Write verifier
 * @param[out] stability_got    Stability used for write (must be as
 *                              or more stable than request)
 *
 * @return An NFSv4.1 status code.
 */
static nfsstat4 ds_write(struct fsal_ds_handle *const ds_pub,
			 struct req_op_context *const req_ctx,
			 const stateid4 *stateid, const offset4 offset,
			 const count4 write_length, const void *buffer,
			 const stable_how4 stability_wanted,
			 count4 * const written_length,
			 verifier4 * const writeverf,
			 stable_how4 * const stability_got)
{
	struct glfs_ds_handle *ds =
		container_of(ds_pub, struct glfs_ds_handle, ds);
	struct glusterfs_export *glfs_export =
	container_of(ds_pub->pds->mds_fsal_export,
		     struct glusterfs_export, export);
	int    rc = 0;

	memset(writeverf, 0, NFS4_VERIFIER_SIZE);

	if (ds->glhandle == NULL)
		LogDebug(COMPONENT_PNFS, "ds_write glhandle NULL");

	rc = glfs_h_anonymous_write(glfs_export->gl_fs->fs, ds->glhandle,
				    buffer, write_length, offset);
	if (rc < 0) {
		LogMajor(COMPONENT_PNFS, "status after write %d", -rc);
		return posix2nfs4_error(-rc);
	}

	/** @todo:Here DS is performing the write operation, so the MDS is not
	 *      aware of the change.We should inform MDS through upcalls about
	 *      change in file attributes such as size and time.
	 */
	*written_length = rc;

	*stability_got = stability_wanted;
	ds->stability_got = stability_wanted;

	/* Incase of MDS being DS, there shall not be upcalls sent from
	 * backend. Hence invalidate the entry here */
	(void)upcall_inode_invalidate(glfs_export->gl_fs, ds->glhandle);

	return NFS4_OK;
}

/**
 * @brief Destroy the package.
 */
void
cih_pkgdestroy(void)
{
	/* Index over partitions */
	int ix = 0;

	/* Destroy the partitions, warning if not empty */
	for (ix = 0; ix < cih_fhcache.npart; ++ix) {
		if (avltree_first(&cih_fhcache.partition[ix].t) != NULL)
			LogMajor(COMPONENT_CACHE_INODE,
				 "Cache inode AVL tree not empty");
		PTHREAD_RWLOCK_destroy(&cih_fhcache.partition[ix].lock);
		gsh_free(cih_fhcache.partition[ix].cache);
	}
	/* Destroy the partition table */
	gsh_free(cih_fhcache.partition);
	cih_fhcache.partition = NULL;
	initialized = false;
}