/**
 * @brief Return gid given a group name
 *
 * @param[in]  name  group name
 * @param[out] gid   address for gid to be filled in
 *
 * @return 0 on success and errno on failure.
 *
 * NOTE: If a group name doesn't exist, getgrnam_r returns 0 with the
 * result pointer set to NULL. We turn that into ENOENT error! Also,
 * getgrnam_r fails with ERANGE if there is a group with a large number
 * of users that it can't fill all those users into the supplied buffer.
 * This need not be the group we are asking for! ERANGE is handled here,
 * so this function never ends up returning ERANGE back to the caller.
 */
static int name_to_gid(const char *name, gid_t *gid)
{
	struct group g;
	struct group *gres = NULL;
	char *buf;
	size_t buflen = sysconf(_SC_GETGR_R_SIZE_MAX);

	/* Upper bound on the buffer length. Just to bailout if there is
	 * a bug in getgrname_r returning ERANGE incorrectly. 64MB
	 * should be good enough for now.
	 */
	size_t maxlen = 64 * 1024 * 1024;
	int err;

	if (buflen == -1)
		buflen = PWENT_BEST_GUESS_LEN;

	do {
		buf = gsh_malloc(buflen);
		if (buf == NULL) {
			LogCrit(COMPONENT_IDMAPPER,
				"gsh_malloc failed, buflen: %zu", buflen);

			return ENOMEM;
		}

		err = getgrnam_r(name, &g, buf, buflen, &gres);
		if (err == ERANGE) {
			buflen *= 16;
			gsh_free(buf);
		}
	} while (buflen <= maxlen && err == ERANGE);

	if (err == 0) {
		if (gres == NULL)
			err = ENOENT;
		else
			*gid = gres->gr_gid;
	}

	if (err != ERANGE)
		gsh_free(buf);

	return err;
}

/**
 * @brief Release previously-ref'd DRC.
 *
 * Release previously-ref'd DRC.  If its refcnt drops to 0, the DRC
 * is queued for later recycling.
 *
 * @param[in] xprt  The SVCXPRT associated with DRC, if applicable
 * @param[in] drc   The DRC
 * @param[in] flags Control flags
 */
void nfs_dupreq_put_drc(SVCXPRT *xprt, drc_t *drc, uint32_t flags)
{
	if (!(flags & DRC_FLAG_LOCKED))
		PTHREAD_MUTEX_lock(&drc->mtx);
	/* drc LOCKED */

	if (drc->refcnt == 0) {
		LogCrit(COMPONENT_DUPREQ,
			"drc %p refcnt will underrun refcnt=%u", drc,
			drc->refcnt);
	}

	nfs_dupreq_unref_drc(drc);

	LogFullDebug(COMPONENT_DUPREQ, "drc %p refcnt==%u", drc, drc->refcnt);

	switch (drc->type) {
	case DRC_UDP_V234:
		/* do nothing */
		break;
	case DRC_TCP_V4:
	case DRC_TCP_V3:
		if (drc->refcnt == 0) {
			if (!(drc->flags & DRC_FLAG_RECYCLE)) {
				/* note t's lock order wrt drc->mtx is
				 * the opposite of drc->xt[*].lock */
				drc->d_u.tcp.recycle_time = time(NULL);
				drc->flags |= DRC_FLAG_RECYCLE;
				PTHREAD_MUTEX_unlock(&drc->mtx); /* !LOCKED */
				DRC_ST_LOCK();
				TAILQ_INSERT_TAIL(&drc_st->tcp_drc_recycle_q,
						  drc, d_u.tcp.recycle_q);
				++(drc_st->tcp_drc_recycle_qlen);
				LogFullDebug(COMPONENT_DUPREQ,
					     "enqueue drc %p for recycle", drc);
				DRC_ST_UNLOCK();
				return;
			}
		}
	default:
		break;
	};

	PTHREAD_MUTEX_unlock(&drc->mtx); /* !LOCKED */
}

fsal_fs_locations_t *nfs4_fs_locations_new(const char *fs_root,
					   const char *rootpath,
					   const unsigned int count)
{
	fsal_fs_locations_t *fs_locations;

	fs_locations = nfs4_fs_locations_alloc(count);
	if (fs_locations == NULL) {
		LogCrit(COMPONENT_NFS_V4, "Could not allocate fs_locations");
		return NULL;
	}

	fs_locations->fs_root = gsh_strdup(fs_root);
	fs_locations->rootpath = gsh_strdup(rootpath);
	fs_locations->ref = 1;

	return fs_locations;
}

void TestCrit(int expect, char *buff, log_components_t component, char *string)
{
	char compare[2048];
	sprintf(compare, "%s: CRITICAL ERROR: %s",
		LogComponents[component].comp_str, string);
	buff[0] = '\0';
	LogCrit(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);
}

fsal_status_t POSIXFSAL_InitClientContext(fsal_op_context_t * thr_context)
{
  posixfsal_op_context_t * p_thr_context = (posixfsal_op_context_t *) thr_context;
  fsal_posixdb_status_t st;

  /* sanity check */
  if(!p_thr_context)
    Return(ERR_FSAL_FAULT, 0, INDEX_FSAL_InitClientContext);

  /* initialy set the export entry to none */
  p_thr_context->export_context = NULL;

  st = fsal_posixdb_connect(&global_posixdb_params, &(p_thr_context->p_conn));
  if(FSAL_POSIXDB_IS_ERROR(st))
    {
      LogCrit(COMPONENT_FSAL,
              "CRITICAL ERROR: Worker could not connect to database !!!");
      Return(ERR_FSAL_SERVERFAULT, 0, INDEX_FSAL_InitClientContext);
    }
  else
    {
      LogEvent(COMPONENT_FSAL, "Worker successfuly connected to database");
    }

  /* sets the credential time */
/*  p_thr_context->credential.last_update = time( NULL );*/

  /* traces: prints p_credential structure */

  /*
     LogDebug(COMPONENT_FSAL, "credential created:");
     LogDebug(COMPONENT_FSAL, "\tuid = %d, gid = %d",
     p_thr_context->credential.hpss_usercred.SecPWent.Uid, p_thr_context->credential.hpss_usercred.SecPWent.Gid);
     LogDebug(COMPONENT_FSAL, "\tName = %s",
     p_thr_context->credential.hpss_usercred.SecPWent.Name);

     for ( i=0; i< p_thr_context->credential.hpss_usercred.NumGroups; i++ )
     LogDebug(COMPONENT_FSAL, "\tAlt grp: %d",
     p_thr_context->credential.hpss_usercred.AltGroups[i] );
   */
  Return(ERR_FSAL_NO_ERROR, 0, INDEX_FSAL_InitClientContext);

}

int mnt_Mnt(nfs_arg_t *arg,
	    nfs_worker_data_t *worker,
	    struct svc_req *req, nfs_res_t *res)
{
	struct gsh_export *export = NULL;
	struct fsal_obj_handle *pfsal_handle = NULL;
	int auth_flavor[NB_AUTH_FLAVOR];
	int index_auth = 0;
	int i = 0;
	char dumpfh[1024];
	int retval = NFS_REQ_OK;
	nfs_fh3 *fh3 = (nfs_fh3 *) &res->res_mnt3.mountres3_u.mountinfo.fhandle;
	cache_entry_t *entry = NULL;

	LogDebug(COMPONENT_NFSPROTO,
		 "REQUEST PROCESSING: Calling mnt_Mnt path=%s", arg->arg_mnt);

	/* Paranoid command to clean the result struct. */
	memset(res, 0, sizeof(nfs_res_t));

	/* Quick escape if an unsupported MOUNT version */
	if (req->rq_vers != MOUNT_V3) {
		res->res_mnt1.status = NFSERR_ACCES;
		goto out;
	}

	if (arg->arg_mnt == NULL) {
		LogCrit(COMPONENT_NFSPROTO,
			"NULL path passed as Mount argument !!!");
		retval = NFS_REQ_DROP;
		goto out;
	}

	/* If the path ends with a '/', get rid of it */
	/** @todo: should it be a while()?? */
	if (arg->arg_mnt[strlen(arg->arg_mnt) - 1] == '/')
		arg->arg_mnt[strlen(arg->arg_mnt) - 1] = '\0';

	/*  Find the export for the dirname (using as well Path or Tag) */
	if (arg->arg_mnt[0] == '/')
		export = get_gsh_export_by_path(arg->arg_mnt, false);
	else
		export = get_gsh_export_by_tag(arg->arg_mnt);

/**
 *
 * @brief Initialize the caching layer
 *
 * This function initializes the memory pools, lookup table, and weakref
 * table used for cache management.
 *
 * @return CACHE_INODE_SUCCESS or errors.
 *
 */
cache_inode_status_t cache_inode_init(void)
{
  cache_inode_status_t status = CACHE_INODE_SUCCESS;

  cache_inode_entry_pool = pool_init("Entry Pool",
                                     sizeof(cache_entry_t),
                                     pool_basic_substrate,
                                     NULL, NULL, NULL);
  if(!(cache_inode_entry_pool))
    {
      LogCrit(COMPONENT_CACHE_INODE,
              "Can't init Entry Pool");
      status = CACHE_INODE_INVALID_ARGUMENT;
    }

  cih_pkginit();

  return status;
}                               /* cache_inode_init */

/**
 * @brief Initialize a shared duplicate request cache
 */
static inline void init_shared_drc()
{
	drc_t *drc = &drc_st->udp_drc;
	int ix, code __attribute__ ((unused)) = 0;

	drc->type = DRC_UDP_V234;
	drc->refcnt = 0;
	drc->retwnd = 0;
	drc->d_u.tcp.recycle_time = 0;
	drc->maxsize = nfs_param.core_param.drc.udp.size;
	drc->cachesz = nfs_param.core_param.drc.udp.cachesz;
	drc->npart = nfs_param.core_param.drc.udp.npart;
	drc->hiwat = nfs_param.core_param.drc.udp.hiwat;

	gsh_mutex_init(&drc->mtx, NULL);

	/* init dict */
	code =
	    rbtx_init(&drc->xt, dupreq_shared_cmpf, drc->npart,
		      RBT_X_FLAG_ALLOC | RBT_X_FLAG_CACHE_WT);
	assert(!code);

	/* completed requests */
	TAILQ_INIT(&drc->dupreq_q);

	/* init closed-form "cache" partition */
	for (ix = 0; ix < drc->npart; ++ix) {
		struct rbtree_x_part *xp = &(drc->xt.tree[ix]);
		drc->xt.cachesz = drc->cachesz;
		xp->cache =
		    gsh_calloc(drc->cachesz, sizeof(struct opr_rbtree_node *));
		if (unlikely(!xp->cache)) {
			LogCrit(COMPONENT_DUPREQ,
				"UDP DRC hash partition allocation "
				"failed (ix=%d)", ix);
			drc->cachesz = 0;
			break;
		}
	}

	return;
}

/**
 * @brief Unlock Message
 *
 * @param[in]  args
 * @param[in]  req
 * @param[out] res
 *
 */
int nlm4_Unlock_Message(nfs_arg_t *args, struct svc_req *req, nfs_res_t *res)
{
	state_nlm_client_t *nlm_client = NULL;
	state_nsm_client_t *nsm_client;
	nlm4_unlockargs *arg = &args->arg_nlm4_unlock;
	int rc = NFS_REQ_OK;

	LogDebug(COMPONENT_NLM,
		 "REQUEST PROCESSING: Calling nlm_Unlock_Message");

	nsm_client = get_nsm_client(CARE_NO_MONITOR,
				    req->rq_xprt,
				    arg->alock.caller_name);

	if (nsm_client != NULL)
		nlm_client = get_nlm_client(CARE_NO_MONITOR,
					    req->rq_xprt, nsm_client,
					    arg->alock.caller_name);

	if (nlm_client == NULL)
		rc = NFS_REQ_DROP;
	else
		rc = nlm4_Unlock(args, req, res);

	if (rc == NFS_REQ_OK)
		rc = nlm_send_async_res_nlm4(nlm_client,
					     nlm4_unlock_message_resp,
					     res);

	if (rc == NFS_REQ_DROP) {
		if (nsm_client != NULL)
			dec_nsm_client_ref(nsm_client);

		if (nlm_client != NULL)
			dec_nlm_client_ref(nlm_client);

		LogCrit(COMPONENT_NLM,
			"Could not send async response for nlm_Unlock_Message");
	}

	return NFS_REQ_DROP;
}

/**
 * @brief Construct the fs opaque part of a pseudofs nfsv4 handle
 *
 * Given the components of a pseudofs nfsv4 handle, the nfsv4 handle is
 * created by concatenating the components. This is the fs opaque piece
 * of struct file_handle_v4 and what is sent over the wire.
 *
 * @param[in] pseudopath Full patch of the pseudofs node
 * @param[in] len length of the pseudopath parameter
 * @param[in] hashkey a 64 bit hash of the pseudopath parameter
 *
 * @return The nfsv4 pseudofs file handle as a char *
 */
char *package_pseudo_handle(char *pseudopath, ushort len, uint64 hashkey)
{
	char *buff = NULL;
	int opaque_bytes_used = 0, pathlen = 0;

	/* This is the size of the v4 file handle opaque area used for pseudofs
	 * or FSAL file handles.
	 */
	buff = gsh_malloc(V4_FH_OPAQUE_SIZE);
	if (buff == NULL) {
		LogCrit(COMPONENT_NFS_V4_PSEUDO,
			"Failed to malloc space for pseudofs handle.");
		return NULL;
	}

	memcpy(buff, &hashkey, sizeof(hashkey));
	opaque_bytes_used += sizeof(hashkey);

	/* include length of the path in the handle.
	 * MAXPATHLEN=4096 ... max path length can be contained in a short int.
	 */
	memcpy(buff + opaque_bytes_used, &len, sizeof(ushort));
	opaque_bytes_used += sizeof(ushort);

	/* Either the nfsv4 fh opaque size or the length of the pseudopath.
	 * Ideally we can include entire pseudofs pathname for guaranteed
	 * uniqueness of pseudofs handles.
	 */
	pathlen = MIN(V4_FH_OPAQUE_SIZE - opaque_bytes_used, len);
	memcpy(buff + opaque_bytes_used, pseudopath, pathlen);
	opaque_bytes_used += pathlen;

	/* If there is more space in the opaque handle due to a short pseudofs
	 * path ... zero it.
	 */
	if (opaque_bytes_used < V4_FH_OPAQUE_SIZE) {
		memset(buff + opaque_bytes_used, 0,
		       V4_FH_OPAQUE_SIZE - opaque_bytes_used);
	}

	return buff;
}

static fsal_status_t create(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_create(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);
}

/**
 * @brief Clean resources associated with entry
 *
 * This function frees the various resources associated wiith a cache
 * entry.
 *
 * @param[in] entry Entry to be cleaned
 *
 * @return CACHE_INODE_SUCCESS or various errors
 */
cache_inode_status_t
cache_inode_clean_internal(cache_entry_t *entry)
{
     hash_buffer_t key, val;
     hash_error_t rc = 0;

     if (entry->fh_desc.start == 0)
         return CACHE_INODE_SUCCESS;

     key.pdata = entry->fh_desc.start;
     key.len = entry->fh_desc.len;


     val.pdata = entry;
     val.len = sizeof(cache_entry_t);

     rc = HashTable_DelSafe(fh_to_cache_entry_ht,
                            &key,
                            &val);

     /* Nonexistence is as good as success. */
     if ((rc != HASHTABLE_SUCCESS) &&
         (rc != HASHTABLE_ERROR_NO_SUCH_KEY)) {
          /* XXX this seems to logically prevent relcaiming the HashTable LRU
           * reference, and it seems to indicate a very serious problem */
          LogCrit(COMPONENT_CACHE_INODE,
                  "HashTable_Del error %d in cache_inode_clean_internal", rc);
          return CACHE_INODE_INCONSISTENT_ENTRY;
     }

     /* Delete from the weakref table */
     cache_inode_weakref_delete(&entry->weakref);

     if (entry->type == SYMBOLIC_LINK) {
          pthread_rwlock_wrlock(&entry->content_lock);
          cache_inode_release_symlink(entry);
          pthread_rwlock_unlock(&entry->content_lock);
     }

     return CACHE_INODE_SUCCESS;
} /* cache_inode_clean_internal */

/**
 * nlm4_Lock_Message: Lock Message
 *
 *  @param parg        [IN]
 *  @param pexportlist [IN]
 *  @param pcontextp   [IN]
 *  @param pclient     [INOUT]
 *  @param ht          [INOUT]
 *  @param preq        [IN]
 *  @param pres        [OUT]
 *
 */
int nlm4_Lock_Message(nfs_arg_t * parg /* IN     */ ,
                      exportlist_t * pexport /* IN     */ ,
                      fsal_op_context_t * pcontext /* IN     */ ,
                      cache_inode_client_t * pclient /* INOUT  */ ,
                      hash_table_t * ht /* INOUT  */ ,
                      struct svc_req *preq /* IN     */ ,
                      nfs_res_t * pres /* OUT    */ )
{
  state_nlm_client_t * nlm_client = NULL;
  state_nsm_client_t * nsm_client;
  nlm4_lockargs      * arg = &parg->arg_nlm4_lock;
  int                  rc = NFS_REQ_OK;

  LogDebug(COMPONENT_NLM, "REQUEST PROCESSING: Calling nlm_Lock_Message");

  nsm_client = get_nsm_client(CARE_NO_MONITOR, preq->rq_xprt, arg->alock.caller_name);

  if(nsm_client != NULL)
    nlm_client = get_nlm_client(CARE_NO_MONITOR, preq->rq_xprt, nsm_client, arg->alock.caller_name);

  if(nlm_client == NULL)
    rc = NFS_REQ_DROP;
  else
    rc = nlm4_Lock(parg, pexport, pcontext, pclient, ht, preq, pres);

  if(rc == NFS_REQ_OK)
    rc = nlm_send_async_res_nlm4(nlm_client, nlm4_lock_message_resp, pres);

  if(rc == NFS_REQ_DROP)
    {
      if(nsm_client != NULL)
        dec_nsm_client_ref(nsm_client);
      if(nlm_client != NULL)
        dec_nlm_client_ref(nlm_client);
      LogCrit(COMPONENT_NLM,
            "Could not send async response for nlm_Lock_Message");
    }

  return NFS_REQ_DROP;

}

/**
 * _9p_dispatcher_svc_run: main loop for 9p dispatcher
 *
 * This function is the main loop for the 9p dispatcher.
 * It never returns because it is an infinite loop.
 *
 * @param sock accept socket for 9p dispatch
 *
 * @return nothing (void function).
 *
 */
void _9p_dispatcher_svc_run(long int sock)
{
	int rc = 0;
	struct sockaddr_in addr;
	socklen_t addrlen = sizeof(addr);
	long int newsock = -1;
	pthread_attr_t attr_thr;
	pthread_t tcp_thrid;

	/* Init for thread parameter (mostly for scheduling) */
	if (pthread_attr_init(&attr_thr) != 0)
		LogDebug(COMPONENT_9P_DISPATCH,
			 "can't init pthread's attributes");

	if (pthread_attr_setscope(&attr_thr, PTHREAD_SCOPE_SYSTEM) != 0)
		LogDebug(COMPONENT_9P_DISPATCH, "can't set pthread's scope");

	if (pthread_attr_setdetachstate(&attr_thr,
					PTHREAD_CREATE_DETACHED) != 0)
		LogDebug(COMPONENT_9P_DISPATCH,
			 "can't set pthread's join state");

	LogEvent(COMPONENT_9P_DISPATCH, "9P dispatcher started");
	while (true) {
		newsock = accept(sock, (struct sockaddr *)&addr, &addrlen);
		if (newsock < 0) {
			LogCrit(COMPONENT_9P_DISPATCH, "accept failed");
			continue;
		}

		/* Starting the thread dedicated to signal handling */
		rc = pthread_create(&tcp_thrid, &attr_thr,
				    _9p_socket_thread, (void *)newsock);
		if (rc != 0) {
			LogFatal(COMPONENT_THREAD,
				 "Could not create 9p socket manager thread, error = %d (%s)",
				 errno, strerror(errno));
		}
	}			/* while */
	return;
}				/* _9p_dispatcher_svc_run */

fsal_status_t vfs_open_my_fd(struct vfs_fsal_obj_handle *myself,
			     fsal_openflags_t openflags,
			     int posix_flags,
			     struct vfs_fd *my_fd)
{
	int fd;
	fsal_errors_t fsal_error = ERR_FSAL_NO_ERROR;
	int retval = 0;

	LogFullDebug(COMPONENT_FSAL,
		     "my_fd->fd = %d openflags = %x, posix_flags = %x",
		     my_fd->fd, openflags, posix_flags);

	assert(my_fd->fd == -1
	       && my_fd->openflags == FSAL_O_CLOSED && openflags != 0);

	LogFullDebug(COMPONENT_FSAL,
		     "openflags = %x, posix_flags = %x",
		     openflags, posix_flags);

	fd = vfs_fsal_open(myself, posix_flags, &fsal_error);

	if (fd < 0) {
		retval = -fd;
	} else {
		/* Save the file descriptor, make sure we only save the
		 * open modes that actually represent the open file.
		 */
		LogFullDebug(COMPONENT_FSAL,
			     "fd = %d, new openflags = %x",
			     fd, openflags);
		if (fd == 0)
			LogCrit(COMPONENT_FSAL,
				"fd = %d, new openflags = %x",
				fd, openflags);
		my_fd->fd = fd;
		my_fd->openflags = openflags;
	}

	return fsalstat(fsal_error, retval);
}

int mnt_UmntAll(nfs_arg_t * parg /* IN     */ ,
                exportlist_t * pexport /* IN     */ ,
                fsal_op_context_t * pcontext /* IN     */ ,
                cache_inode_client_t * pclient /* INOUT  */ ,
                hash_table_t * ht /* INOUT  */ ,
                struct svc_req *preq /* IN     */ ,
                nfs_res_t * pres /* OUT    */ )
{
    LogFullDebug(COMPONENT_NFSPROTO,
                 "REQUEST PROCESSING: Calling mnt_UmntAll");

    /* Just empty the Mount list, take void as argument and returns void */
    if(!nfs_Purge_MountList())
    {
        /* Purge mount list failed */
        LogCrit(COMPONENT_NFSPROTO,
                "UMOUNT ALL: Error when emptying the mount list");
    }

    return NFS_REQ_OK;
}                               /* mnt_UmntAll */

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

	*handle = NULL;		/* poison it first */
	if (!path)
		return fsalstat(ERR_FSAL_FAULT, 0);
	memset(fh, 0, sizeof(ptfsal_handle_t));
	fh->data.handle.handle_size = FSI_CCL_PERSISTENT_HANDLE_N_BYTES;
	if (!parent->ops->handle_is(parent, DIRECTORY)) {
		LogCrit(COMPONENT_FSAL,
			"Parent handle is not a directory. hdl = 0x%p", parent);
		return fsalstat(ERR_FSAL_NOTDIR, 0);
	}
	attrib.mask = parent->attributes.mask;
	status = PTFSAL_lookup(op_ctx, parent, path, &attrib, fh);
	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 hdlerr;
	}
	*handle = &hdl->obj_handle;
	return fsalstat(ERR_FSAL_NO_ERROR, 0);

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

fsal_status_t
fsal_proxy_create_rpc_clnt(proxyfsal_op_context_t * ctx)
{
  int sock;
  struct sockaddr_in addr_rpc;
  struct timeval timeout = TIMEOUTRPC;
  int rc;
  int priv_port = 0 ; 
  fsal_status_t fsal_status;
  char addr[INET_ADDRSTRLEN];

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

  if(!strcmp(ctx->srv_proto, "udp"))
    {
      if((sock = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)) < 0)
        ReturnCode(ERR_FSAL_FAULT, errno);

      ctx->rpc_client = clntudp_bufcreate(&addr_rpc,
                                          ctx->srv_prognum,
                                          FSAL_PROXY_NFS_V4,
                                          (struct timeval){ 25, 0},
                                          &sock,
                                          ctx->srv_sendsize,
                                          ctx->srv_recvsize);

      if(ctx->rpc_client == NULL)
        {

          LogCrit(COMPONENT_FSAL,
                  "Cannot contact server addr=%s port=%u prognum=%u using NFSv4 protocol",
                  inet_ntop(AF_INET, &ctx->srv_addr, addr, sizeof(addr)),
                  ntohs(ctx->srv_port), ctx->srv_prognum);

          ReturnCode(ERR_FSAL_INVAL, 0);
        }
    }

/**
 *
 *  utf82gid: converts a utf8 string descriptorto a gid .
 *
 * Converts a utf8 string descriptor to a gid .
 *
 * @param utf8str [IN]  group's name as UTF8 string.
 * @param Gid     [OUT] pointer to the computed gid.
 *
 * @return 0 in all cases
 */
int utf82gid(utf8string * utf8str, gid_t * Gid)
{
  char buff[2 * NFS4_MAX_DOMAIN_LEN];
  char gidname[NFS4_MAX_DOMAIN_LEN];
#ifndef _USE_NFSIDMAP
  char domainname[NFS4_MAX_DOMAIN_LEN];
#endif
  int  rc;

  if(utf8str->utf8string_len == 0)
    {
      *Gid = -1;                /* Nobody */
      LogCrit(COMPONENT_IDMAPPER,
              "utf82gid: empty group name");
      return 0;
    }

  utf82str(buff, sizeof(buff), utf8str);

#ifndef _USE_NFSIDMAP
  /* Group is shown as a string '[email protected]' , remove it if libnfsidmap is not used */
  nfs4_stringid_split(buff, gidname, domainname);
#else
  strncpy(gidname, buff, NFS4_MAX_DOMAIN_LEN);
#endif

  rc = name2gid(gidname, Gid);

  if(rc == 0)
    {
      *Gid = -1;                /* Nobody */
      return 0;
    }

  LogDebug(COMPONENT_IDMAPPER,
           "utf82gid: Mapped %s to gid = %d",
           buff, *Gid);

  return 0;
}                               /* utf82gid */

fsal_status_t GPFSFSAL_CleanUpExportContext(fsal_export_context_t * export_context) 
{
  gpfsfsal_export_context_t *p_export_context = (gpfsfsal_export_context_t *)export_context;

  if(export_context == NULL) 
  {
    LogCrit(COMPONENT_FSAL,
            "NULL mandatory argument passed to %s()", __FUNCTION__);
    Return(ERR_FSAL_FAULT, 0, INDEX_FSAL_CleanUpExportContext);
  }

  if(p_export_context->mount_root_fd != 0)
    close(p_export_context->mount_root_fd);

  if(p_export_context->fe_fsal_up_ctx != NULL)
    {
      /* Start to clean up FSAL_UP stuff. There is actually more to do here...*/
      glist_del(&p_export_context->fe_list);
    }

  Return(ERR_FSAL_NO_ERROR, 0, INDEX_FSAL_CleanUpExportContext);
}