handle.c

// SPDX-License-Identifier: LGPL-3.0-or-later
/*
 * vim:noexpandtab:shiftwidth=8:tabstop=8:
 *
 * Copyright (C) Max Matveev, 2012
 * Copyright CEA/DAM/DIF  (2008)
 *
 * contributeur : Philippe DENIEL   philippe.deniel@cea.fr
 *                Thomas LEIBOVICI  thomas.leibovici@cea.fr
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 3 of the License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
 * 02110-1301 USA
 */

/* Proxy handle methods */

#include "config.h"

#include "fsal.h"
#include <assert.h>
#include <pthread.h>
#include <arpa/inet.h>
#include <sys/poll.h>
#include <netdb.h>
#include <urcu-bp.h>

#include "gsh_list.h"
#include "abstract_atomic.h"
#include "fsal_types.h"
#include "FSAL/fsal_commonlib.h"
#include "proxyv4_fsal_methods.h"
#include "fsal_nfsv4_macros.h"
#include "nfs_core.h"
#include "nfs_proto_functions.h"
#include "nfs_proto_tools.h"
#include "export_mgr.h"
#include "common_utils.h"
#include "fsal_convert.h"

#define FSAL_PROXY_NFS_V4 4
#define FSAL_PROXY_NFS_V4_MINOR 1
#define NB_RPC_SLOT 16
#define NB_MAX_OPERATIONS 10

/* NB! nfs_prog is just an easy way to get this info into the call
 *     It should really be fetched via export pointer */
/**
 * We mutualize rpc_context and slot NFSv4.1.
 */
struct proxyv4_rpc_io_context {
	pthread_mutex_t iolock;
	pthread_cond_t iowait;
	struct glist_head calls;
	uint32_t rpc_xid;
	bool iodone;
	int ioresult;
	unsigned int nfs_prog;
	unsigned int sendbuf_sz;
	unsigned int recvbuf_sz;
	char *sendbuf;
	char *recvbuf;
	slotid4 slotid;
	sequenceid4 seqid;
};

/* Use this to estimate storage requirements for fattr4 blob */
struct proxyv4_fattr_storage {
	fattr4_type type;
	fattr4_change change_time;
	fattr4_size size;
	fattr4_fsid fsid;
	fattr4_filehandle filehandle;
	fattr4_fileid fileid;
	fattr4_mode mode;
	fattr4_numlinks numlinks;
	fattr4_owner owner;
	fattr4_owner_group owner_group;
	fattr4_space_used space_used;
	fattr4_time_access time_access;
	fattr4_time_metadata time_metadata;
	fattr4_time_modify time_modify;
	fattr4_rawdev rawdev;
	char padowner[MAXNAMLEN + 1];
	char padgroup[MAXNAMLEN + 1];
	char padfh[NFS4_FHSIZE];
};

#define FATTR_BLOB_SZ sizeof(struct proxyv4_fattr_storage)

/*
 * This is what becomes an opaque FSAL handle for the upper layers.
 *
 * The type is a placeholder for future expansion.
 */
struct proxyv4_handle_blob {
	uint8_t len;
	uint8_t type;
	uint8_t bytes[0];
};

struct proxyv4_obj_handle {
	struct fsal_obj_handle obj;
	nfs_fh4 fh4;
#ifdef PROXYV4_HANDLE_MAPPING
	nfs23_map_handle_t h23;
#endif
	fsal_openflags_t openflags;
	struct proxyv4_handle_blob blob;
};

static struct proxyv4_obj_handle *
proxyv4_alloc_handle(struct fsal_export *exp,
		     const nfs_fh4 *fh,
		     fattr4 *obj_attributes,
		     struct fsal_attrlist *attrs_out);

struct proxyv4_state {
	struct state_t state;
	stateid4 stateid;
};

struct state_t *proxyv4_alloc_state(struct fsal_export *exp_hdl,
				    enum state_type state_type,
				    struct state_t *related_state)
{
	return init_state(gsh_calloc(1, sizeof(struct proxyv4_state)), exp_hdl,
			  state_type, related_state);
}

void proxyv4_free_state(struct fsal_export *exp_hdl, struct state_t *state)
{
	struct proxyv4_state *proxyv4_state_id =
		container_of(state, struct proxyv4_state, state);

	gsh_free(proxyv4_state_id);
}

#define FSAL_VERIFIER_T_TO_VERIFIER4(verif4, fsal_verif)		\
	do {								\
		BUILD_BUG_ON(sizeof(fsal_verifier_t) != sizeof(verifier4)); \
		memcpy(verif4, fsal_verif, sizeof(fsal_verifier_t));	\
	} while (0)

static fsal_status_t nfsstat4_to_fsal(nfsstat4 nfsstatus)
{
	switch (nfsstatus) {
	case NFS4ERR_SAME:
	case NFS4ERR_NOT_SAME:
	case NFS4_OK:
		return fsalstat(ERR_FSAL_NO_ERROR, (int)nfsstatus);
	case NFS4ERR_PERM:
		return fsalstat(ERR_FSAL_PERM, (int)nfsstatus);
	case NFS4ERR_NOENT:
		return fsalstat(ERR_FSAL_NOENT, (int)nfsstatus);
	case NFS4ERR_IO:
		return fsalstat(ERR_FSAL_IO, (int)nfsstatus);
	case NFS4ERR_NXIO:
		return fsalstat(ERR_FSAL_NXIO, (int)nfsstatus);
	case NFS4ERR_EXPIRED:
	case NFS4ERR_LOCKED:
	case NFS4ERR_SHARE_DENIED:
	case NFS4ERR_LOCK_RANGE:
	case NFS4ERR_OPENMODE:
	case NFS4ERR_FILE_OPEN:
	case NFS4ERR_ACCESS:
	case NFS4ERR_DENIED:
		return fsalstat(ERR_FSAL_ACCESS, (int)nfsstatus);
	case NFS4ERR_EXIST:
		return fsalstat(ERR_FSAL_EXIST, (int)nfsstatus);
	case NFS4ERR_XDEV:
		return fsalstat(ERR_FSAL_XDEV, (int)nfsstatus);
	case NFS4ERR_NOTDIR:
		return fsalstat(ERR_FSAL_NOTDIR, (int)nfsstatus);
	case NFS4ERR_ISDIR:
		return fsalstat(ERR_FSAL_ISDIR, (int)nfsstatus);
	case NFS4ERR_FBIG:
		return fsalstat(ERR_FSAL_FBIG, 0);
	case NFS4ERR_NOSPC:
		return fsalstat(ERR_FSAL_NOSPC, (int)nfsstatus);
	case NFS4ERR_ROFS:
		return fsalstat(ERR_FSAL_ROFS, (int)nfsstatus);
	case NFS4ERR_MLINK:
		return fsalstat(ERR_FSAL_MLINK, (int)nfsstatus);
	case NFS4ERR_NAMETOOLONG:
		return fsalstat(ERR_FSAL_NAMETOOLONG, (int)nfsstatus);
	case NFS4ERR_NOTEMPTY:
		return fsalstat(ERR_FSAL_NOTEMPTY, (int)nfsstatus);
	case NFS4ERR_DQUOT:
		return fsalstat(ERR_FSAL_DQUOT, (int)nfsstatus);
	case NFS4ERR_STALE:
		return fsalstat(ERR_FSAL_STALE, (int)nfsstatus);
	case NFS4ERR_NOFILEHANDLE:
	case NFS4ERR_BADHANDLE:
		return fsalstat(ERR_FSAL_BADHANDLE, (int)nfsstatus);
	case NFS4ERR_BAD_COOKIE:
		return fsalstat(ERR_FSAL_BADCOOKIE, (int)nfsstatus);
	case NFS4ERR_NOTSUPP:
		return fsalstat(ERR_FSAL_NOTSUPP, (int)nfsstatus);
	case NFS4ERR_TOOSMALL:
		return fsalstat(ERR_FSAL_TOOSMALL, (int)nfsstatus);
	case NFS4ERR_SERVERFAULT:
		return fsalstat(ERR_FSAL_SERVERFAULT, (int)nfsstatus);
	case NFS4ERR_BADTYPE:
		return fsalstat(ERR_FSAL_BADTYPE, (int)nfsstatus);
	case NFS4ERR_GRACE:
	case NFS4ERR_DELAY:
		return fsalstat(ERR_FSAL_DELAY, (int)nfsstatus);
	case NFS4ERR_FHEXPIRED:
		return fsalstat(ERR_FSAL_FHEXPIRED, (int)nfsstatus);
	case NFS4ERR_WRONGSEC:
		return fsalstat(ERR_FSAL_SEC, (int)nfsstatus);
	case NFS4ERR_SYMLINK:
		return fsalstat(ERR_FSAL_SYMLINK, (int)nfsstatus);
	case NFS4ERR_ATTRNOTSUPP:
		return fsalstat(ERR_FSAL_ATTRNOTSUPP, (int)nfsstatus);
	case NFS4ERR_BADNAME:
		return fsalstat(ERR_FSAL_BADNAME, (int)nfsstatus);
	case NFS4ERR_INVAL:
	case NFS4ERR_CLID_INUSE:
	case NFS4ERR_MOVED:
	case NFS4ERR_RESOURCE:
	case NFS4ERR_MINOR_VERS_MISMATCH:
	case NFS4ERR_STALE_CLIENTID:
	case NFS4ERR_STALE_STATEID:
	case NFS4ERR_OLD_STATEID:
	case NFS4ERR_BAD_STATEID:
	case NFS4ERR_BAD_SEQID:
	case NFS4ERR_RESTOREFH:
	case NFS4ERR_LEASE_MOVED:
	case NFS4ERR_NO_GRACE:
	case NFS4ERR_RECLAIM_BAD:
	case NFS4ERR_RECLAIM_CONFLICT:
	case NFS4ERR_BADXDR:
	case NFS4ERR_BADCHAR:
	case NFS4ERR_BAD_RANGE:
	case NFS4ERR_BADOWNER:
	case NFS4ERR_OP_ILLEGAL:
	case NFS4ERR_LOCKS_HELD:
	case NFS4ERR_LOCK_NOTSUPP:
	case NFS4ERR_DEADLOCK:
	case NFS4ERR_ADMIN_REVOKED:
	case NFS4ERR_CB_PATH_DOWN:
	default:
		return fsalstat(ERR_FSAL_INVAL, (int)nfsstatus);
	}
}

#define PROXYV4_ATTR_BIT(b) (1U << b)
#define PROXYV4_ATTR_BIT2(b) (1U << (b - 32))

static struct bitmap4 proxyv4_bitmap_getattr = {
	.map[0] =
	(PROXYV4_ATTR_BIT(FATTR4_SUPPORTED_ATTRS) |
	 PROXYV4_ATTR_BIT(FATTR4_TYPE) | PROXYV4_ATTR_BIT(FATTR4_CHANGE) |
	 PROXYV4_ATTR_BIT(FATTR4_SIZE) | PROXYV4_ATTR_BIT(FATTR4_FSID) |
	 PROXYV4_ATTR_BIT(FATTR4_FILEID)),
	.map[1] =
	(PROXYV4_ATTR_BIT2(FATTR4_MODE) |
	 PROXYV4_ATTR_BIT2(FATTR4_NUMLINKS) |
	 PROXYV4_ATTR_BIT2(FATTR4_OWNER) |
	 PROXYV4_ATTR_BIT2(FATTR4_OWNER_GROUP) |
	 PROXYV4_ATTR_BIT2(FATTR4_SPACE_USED) |
	 PROXYV4_ATTR_BIT2(FATTR4_TIME_ACCESS) |
	 PROXYV4_ATTR_BIT2(FATTR4_TIME_METADATA) |
	 PROXYV4_ATTR_BIT2(FATTR4_TIME_MODIFY) |
	 PROXYV4_ATTR_BIT2(FATTR4_RAWDEV)),
	.bitmap4_len = 2
};

static struct bitmap4 proxyv4_bitmap_fsinfo = {
	.map[0] =
	(PROXYV4_ATTR_BIT(FATTR4_FILES_AVAIL) |
	 PROXYV4_ATTR_BIT(FATTR4_FILES_FREE) |
	 PROXYV4_ATTR_BIT(FATTR4_FILES_TOTAL)),
	.map[1] =
	(PROXYV4_ATTR_BIT2(FATTR4_SPACE_AVAIL) |
	 PROXYV4_ATTR_BIT2(FATTR4_SPACE_FREE) |
	 PROXYV4_ATTR_BIT2(FATTR4_SPACE_TOTAL)),
	.bitmap4_len = 2
};

static struct bitmap4 lease_bits = {
	.map[0] = PROXYV4_ATTR_BIT(FATTR4_LEASE_TIME),
	.bitmap4_len = 1
};

static struct bitmap4 proxyv4_bitmap_per_fs = {
	.map[0] = (PROXYV4_ATTR_BIT(FATTR4_MAXREAD) |
		   PROXYV4_ATTR_BIT(FATTR4_MAXWRITE)),
	.bitmap4_len = 1
};

#undef PROXYV4_ATTR_BIT
#undef PROXYV4_ATTR_BIT2

static struct {
	attrmask_t mask;
	int fattr_bit;
} fsal_mask2bit[] = {
	{
		ATTR_SIZE, FATTR4_SIZE}, {
		ATTR_MODE, FATTR4_MODE}, {
		ATTR_OWNER, FATTR4_OWNER}, {
		ATTR_GROUP, FATTR4_OWNER_GROUP}, {
		ATTR_ATIME, FATTR4_TIME_ACCESS_SET}, {
		ATTR_ATIME_SERVER, FATTR4_TIME_ACCESS_SET}, {
		ATTR_MTIME, FATTR4_TIME_MODIFY_SET}, {
		ATTR_MTIME_SERVER, FATTR4_TIME_MODIFY_SET}, {
		ATTR_CTIME, FATTR4_TIME_METADATA}
};

static struct bitmap4 empty_bitmap = {
	.map[0] = 0,
	.map[1] = 0,
	.map[2] = 0,
	.bitmap4_len = 2
};

static int
proxyv4_fsalattr_to_fattr4(const struct fsal_attrlist *attrs, fattr4 *data)
{
	int i;
	struct bitmap4 bmap = empty_bitmap;
	struct xdr_attrs_args args;

	for (i = 0; i < ARRAY_SIZE(fsal_mask2bit); i++) {
		if (FSAL_TEST_MASK(attrs->valid_mask, fsal_mask2bit[i].mask)) {
			if (fsal_mask2bit[i].fattr_bit > 31) {
				bmap.map[1] |=
					1U << (fsal_mask2bit[i].fattr_bit - 32);
				bmap.bitmap4_len = 2;
			} else {
				bmap.map[0] |=
					1U << fsal_mask2bit[i].fattr_bit;
			}
		}
	}

	memset(&args, 0, sizeof(args));
	args.attrs = (struct fsal_attrlist *)attrs;
	args.data = NULL;

	return nfs4_FSALattr_To_Fattr(&args, &bmap, data);
}

static GETATTR4resok *
proxyv4_fill_getattr_reply(nfs_resop4 *resop, char *blob, size_t blob_sz)
{
	GETATTR4resok *a = &resop->nfs_resop4_u.opgetattr.GETATTR4res_u.resok4;

	a->obj_attributes.attrmask = empty_bitmap;
	a->obj_attributes.attr_vals.attrlist4_val = blob;
	a->obj_attributes.attr_vals.attrlist4_len = blob_sz;

	return a;
}

static int proxyv4_got_rpc_reply(struct proxyv4_rpc_io_context *ctx,
				 int sock,
				 int sz,
				 u_int xid)
{
	char *repbuf = ctx->recvbuf;
	int size;

	if (sz > ctx->recvbuf_sz)
		return -E2BIG;

	PTHREAD_MUTEX_lock(&ctx->iolock);
	memcpy(repbuf, &xid, sizeof(xid));
	/*
	 * sz includes 4 bytes of xid which have been processed
	 * together with record mark - reduce the read to avoid
	 * gobbing up next record mark.
	 */
	repbuf += 4;
	ctx->ioresult = 4;
	sz -= 4;

	while (sz > 0) {
		/* TODO: handle timeouts - use poll(2) */
		int bc = read(sock, repbuf, sz);

		if (bc <= 0) {
			ctx->ioresult = -((bc < 0) ? errno : ETIMEDOUT);
			break;
		}
		repbuf += bc;
		ctx->ioresult += bc;
		sz -= bc;
	}
	ctx->iodone = true;
	size = ctx->ioresult;
	pthread_cond_signal(&ctx->iowait);
	PTHREAD_MUTEX_unlock(&ctx->iolock);
	return size;
}

static int proxyv4_rpc_read_reply(struct proxyv4_export *proxyv4_exp)
{
	struct {
		uint recmark;
		uint xid;
	} h;
	char *buf = (char *)&h;
	struct glist_head *c;
	char sink[256];
	int cnt = 0;
	struct proxyv4_export_rpc *rpc = &proxyv4_exp->rpc;

	while (cnt < 8) {
		int bc = read(rpc->rpc_sock, buf + cnt, 8 - cnt);

		if (bc < 0)
			return -errno;
		cnt += bc;
	}

	h.recmark = ntohl(h.recmark);
	/* TODO: check for final fragment */
	h.xid = ntohl(h.xid);

	LogDebug(COMPONENT_FSAL, "Recmark %x, xid %u\n", h.recmark, h.xid);
	h.recmark &= ~(1U << 31);

	PTHREAD_MUTEX_lock(&rpc->listlock);
	glist_for_each(c, &rpc->rpc_calls) {
		struct proxyv4_rpc_io_context *ctx =
			container_of(c, struct proxyv4_rpc_io_context, calls);

		if (ctx->rpc_xid == h.xid) {
			glist_del(c);
			PTHREAD_MUTEX_unlock(&rpc->listlock);
			return proxyv4_got_rpc_reply(ctx,
						     rpc->rpc_sock,
						     h.recmark, h.xid);
		}
	}
	PTHREAD_MUTEX_unlock(&rpc->listlock);

	cnt = h.recmark - 4;
	LogDebug(COMPONENT_FSAL, "xid %u is not on the list, skip %d bytes\n",
		 h.xid, cnt);
	while (cnt > 0) {
		int rb = (cnt > sizeof(sink)) ? sizeof(sink) : cnt;

		rb = read(rpc->rpc_sock, sink, rb);
		if (rb <= 0)
			return -errno;
		cnt -= rb;
	}

	return 0;
}

/* called with listlock */
static void proxyv4_new_socket_ready(struct proxyv4_export *proxyv4_exp)
{
	struct glist_head *nxt;
	struct glist_head *c;
	struct proxyv4_export_rpc *rpc = &proxyv4_exp->rpc;

	/* If there are any outstanding calls then tell them to resend */
	glist_for_each_safe(c, nxt, &rpc->rpc_calls) {
		struct proxyv4_rpc_io_context *ctx =
			container_of(c, struct proxyv4_rpc_io_context, calls);

		glist_del(c);

		PTHREAD_MUTEX_lock(&ctx->iolock);
		ctx->iodone = true;
		ctx->ioresult = -EAGAIN;
		pthread_cond_signal(&ctx->iowait);
		PTHREAD_MUTEX_unlock(&ctx->iolock);
	}

	/* If there is anyone waiting for the socket then tell them
	 * it's ready */
	pthread_cond_broadcast(&rpc->sockless);
}

/* called with listlock */
static int proxyv4_connect(struct proxyv4_export *proxyv4_exp,
			   sockaddr_t *dest, uint16_t port)
{
	int sock;
	int socklen;

	switch (dest->ss_family) {
	case AF_INET:
		((struct sockaddr_in *)dest)->sin_port = htons(port);
		socklen = sizeof(struct sockaddr_in);
		break;
	case AF_INET6:
		((struct sockaddr_in6 *)dest)->sin6_port = htons(port);
		socklen = sizeof(struct sockaddr_in6);
		break;
	default:
		LogCrit(COMPONENT_FSAL, "Unknown address family %d",
			dest->ss_family);
		return -1;
	}

	if (proxyv4_exp->info.use_privileged_client_port) {
		int priv_port = 0;

		sock = rresvport_af(&priv_port, dest->ss_family);
		if (sock < 0)
			LogCrit(COMPONENT_FSAL,
				"Cannot create TCP socket on privileged port");
	} else {
		sock = socket(dest->ss_family, SOCK_STREAM, IPPROTO_TCP);
		if (sock < 0)
			LogCrit(COMPONENT_FSAL, "Cannot create TCP socket - %d",
				errno);
	}

	if (sock >= 0) {
		if (connect(sock, (struct sockaddr *)dest, socklen) < 0) {
			close(sock);
			sock = -1;
		} else {
			proxyv4_new_socket_ready(proxyv4_exp);
		}
	}
	return sock;
}

/*
 * NB! rpc_sock can be closed by the sending thread but it will not be
 *     changing its value. Only this function will change rpc_sock which
 *     means that it can look at the value without holding the lock.
 */
static void *proxyv4_rpc_recv(void *arg)
{
	struct proxyv4_export *proxyv4_exp = arg;
	struct proxyv4_export_rpc *rpc = &proxyv4_exp->rpc;
	pthread_mutex_t *list_lock = &rpc->listlock;
	struct pollfd pfd;
	int millisec = proxyv4_exp->info.srv_timeout * 1000;

	SetNameFunction("proxyv4_rcv_thread");

	rcu_register_thread();
	while (!rpc->close_thread) {
		int nsleeps = 0;

		PTHREAD_MUTEX_lock(list_lock);
		do {
			sockaddr_t *srv_addr =
				&proxyv4_exp->info.srv_addr;
			uint16_t srv_port =
				proxyv4_exp->info.srv_port;

			rpc->rpc_sock =
				proxyv4_connect(proxyv4_exp,
						srv_addr,
						srv_port);

			/* early stop test */
			if (rpc->close_thread) {
				PTHREAD_MUTEX_unlock(list_lock);
				goto out;
			}
			if (rpc->rpc_sock < 0) {
				if (nsleeps == 0) {
					char addr[SOCK_NAME_MAX];
					struct display_buffer dspbuf = {
						sizeof(addr), addr, addr};

					display_sockaddr(&dspbuf,
							 srv_addr);

					LogCrit(COMPONENT_FSAL,
						"Cannot connect to server %s:%u",
						addr, srv_port);
				}
				PTHREAD_MUTEX_unlock(list_lock);
				sleep(proxyv4_exp->info.retry_sleeptime);
				nsleeps++;
				PTHREAD_MUTEX_lock(list_lock);
			} else {
				LogDebug(COMPONENT_FSAL,
					 "Connected after %d sleeps, resending outstanding calls",
					 nsleeps);
			}
		} while (rpc->rpc_sock < 0 &&
			 !rpc->close_thread);
		PTHREAD_MUTEX_unlock(list_lock);
		/* early stop test */
		if (rpc->close_thread)
			goto out;

		pfd.fd = rpc->rpc_sock;
		pfd.events = POLLIN | POLLRDHUP;

		while (rpc->rpc_sock >= 0) {
			switch (poll(&pfd, 1, millisec)) {
			case 0:
				LogDebug(COMPONENT_FSAL,
					 "Timeout, wait again...");
				continue;

			case -1:
				break;

			default:
				if (pfd.revents & POLLRDHUP) {
					LogEvent(COMPONENT_FSAL,
						 "Other end has closed connection, reconnecting...");
				} else if (pfd.revents & POLLNVAL) {
					LogEvent(COMPONENT_FSAL,
						 "Socket is closed");
				}

				if (proxyv4_rpc_read_reply(proxyv4_exp) >= 0)
					continue;

				break;
			}

			PTHREAD_MUTEX_lock(list_lock);
			close(rpc->rpc_sock);
			rpc->rpc_sock = -1;
			PTHREAD_MUTEX_unlock(list_lock);
		}
	}
 out:
	rcu_unregister_thread();
	return NULL;
}

static enum clnt_stat proxyv4_process_reply(struct proxyv4_rpc_io_context *ctx,
					    COMPOUND4res *res)
{
	enum clnt_stat rc = RPC_CANTRECV;
	struct timespec ts;

	PTHREAD_MUTEX_lock(&ctx->iolock);
	ts.tv_sec = time(NULL) + 60;
	ts.tv_nsec = 0;

	while (!ctx->iodone) {
		int w = pthread_cond_timedwait(&ctx->iowait, &ctx->iolock, &ts);

		if (w == ETIMEDOUT) {
			PTHREAD_MUTEX_unlock(&ctx->iolock);
			return RPC_TIMEDOUT;
		}
	}

	ctx->iodone = false;
	PTHREAD_MUTEX_unlock(&ctx->iolock);

	if (ctx->ioresult > 0) {
		struct rpc_msg reply;
		XDR x;

		memset(&reply, 0, sizeof(reply));
		reply.RPCM_ack.ar_results.proc =
			(xdrproc_t) xdr_COMPOUND4res;
		reply.RPCM_ack.ar_results.where = res;

		memset(&x, 0, sizeof(x));
		xdrmem_create(&x, ctx->recvbuf, ctx->ioresult, XDR_DECODE);

		/* macro is defined, GCC 4.7.2 ignoring */
		if (xdr_replymsg(&x, &reply)) {
			if (reply.rm_reply.rp_stat == MSG_ACCEPTED) {
				switch (reply.rm_reply.rp_acpt.ar_stat) {
				case SUCCESS:
					rc = RPC_SUCCESS;
					break;
				case PROG_UNAVAIL:
					rc = RPC_PROGUNAVAIL;
					break;
				case PROG_MISMATCH:
					rc = RPC_PROGVERSMISMATCH;
					break;
				case PROC_UNAVAIL:
					rc = RPC_PROCUNAVAIL;
					break;
				case GARBAGE_ARGS:
					rc = RPC_CANTDECODEARGS;
					break;
				case SYSTEM_ERR:
					rc = RPC_SYSTEMERROR;
					break;
				default:
					rc = RPC_FAILED;
					break;
				}
			} else {
				switch (reply.rm_reply.rp_rjct.rj_stat) {
				case RPC_MISMATCH:
					rc = RPC_VERSMISMATCH;
					break;
				case AUTH_ERROR:
					rc = RPC_AUTHERROR;
					break;
				default:
					rc = RPC_FAILED;
					break;
				}
			}
		} else {
			rc = RPC_CANTDECODERES;
		}

		reply.RPCM_ack.ar_results.proc = (xdrproc_t) xdr_void;
		reply.RPCM_ack.ar_results.where = NULL;

		xdr_free((xdrproc_t) xdr_replymsg, &reply);
	}
	return rc;
}

static inline int proxyv4_rpc_need_sock(struct proxyv4_export *proxyv4_exp)
{
	struct proxyv4_export_rpc *rpc = &proxyv4_exp->rpc;

	PTHREAD_MUTEX_lock(&rpc->listlock);
	while (rpc->rpc_sock < 0 && !rpc->close_thread)
		pthread_cond_wait(&rpc->sockless,
				  &rpc->listlock);
	PTHREAD_MUTEX_unlock(&rpc->listlock);
	return rpc->close_thread;
}

static inline int proxyv4_rpc_renewer_wait(int timeout,
					   struct proxyv4_export *proxyv4_exp)
{
	struct timespec ts;
	int rc;
	struct proxyv4_export_rpc *rpc = &proxyv4_exp->rpc;

	PTHREAD_MUTEX_lock(&rpc->listlock);
	ts.tv_sec = time(NULL) + timeout;
	ts.tv_nsec = 0;

	rc = pthread_cond_timedwait(&rpc->sockless,
				    &rpc->listlock, &ts);
	PTHREAD_MUTEX_unlock(&rpc->listlock);
	return (rc == ETIMEDOUT);
}

static int proxyv4_compoundv4_call(struct proxyv4_rpc_io_context *pcontext,
				   const struct user_cred *cred,
				   COMPOUND4args *args, COMPOUND4res *res,
				   struct proxyv4_export *proxyv4_exp)
{
	XDR x;
	struct rpc_msg rmsg;
	AUTH *au;
	enum clnt_stat rc;
	struct proxyv4_export_rpc *rpc = &proxyv4_exp->rpc;

	PTHREAD_MUTEX_lock(&rpc->listlock);
	rmsg.rm_xid = rpc->rpc_xid++;
	PTHREAD_MUTEX_unlock(&rpc->listlock);
	rmsg.rm_direction = CALL;

	rmsg.rm_call.cb_rpcvers = RPC_MSG_VERSION;
	rmsg.cb_prog = pcontext->nfs_prog;
	rmsg.cb_vers = FSAL_PROXY_NFS_V4;
	rmsg.cb_proc = NFSPROC4_COMPOUND;

	if (cred) {
		au = authunix_ncreate(rpc->proxyv4_hostname,
				      cred->caller_uid, cred->caller_gid,
				      cred->caller_glen, cred->caller_garray);
	} else {
		au = authunix_ncreate_default();
	}
	if (AUTH_FAILURE(au)) {
		char *err = rpc_sperror(&au->ah_error, "failed");

		LogDebug(COMPONENT_FSAL, "%s", err);
		gsh_free(err);
		AUTH_DESTROY(au);
		return RPC_AUTHERROR;
	}

	rmsg.cb_cred = au->ah_cred;
	rmsg.cb_verf = au->ah_verf;

	memset(&x, 0, sizeof(x));
	xdrmem_create(&x, pcontext->sendbuf + 4, pcontext->sendbuf_sz,
		      XDR_ENCODE);
	if (xdr_callmsg(&x, &rmsg) && xdr_COMPOUND4args(&x, args)) {
		u_int pos = xdr_getpos(&x);
		u_int recmark = ntohl(pos | (1U << 31));
		int first_try = 1;

		pcontext->rpc_xid = rmsg.rm_xid;

		memcpy(pcontext->sendbuf, &recmark, sizeof(recmark));
		pos += 4;

		do {
			int bc = 0;
			char *buf = pcontext->sendbuf;

			LogDebug(COMPONENT_FSAL, "%ssend XID %u with %d bytes",
				 (first_try ? "First attempt to " : "Re"),
				 rmsg.rm_xid, pos);
			PTHREAD_MUTEX_lock(&rpc->listlock);
			while (bc < pos) {
				int wc = write(rpc->rpc_sock, buf, pos - bc);

				if (wc <= 0) {
					close(rpc->rpc_sock);
					break;
				}
				bc += wc;
				buf += wc;
			}

			if (bc == pos) {
				if (first_try) {
					glist_add_tail(&rpc->rpc_calls,
						       &pcontext->calls);
					first_try = 0;
				}
			} else {
				if (!first_try)
					glist_del(&pcontext->calls);
			}
			PTHREAD_MUTEX_unlock(&rpc->listlock);

			if (bc == pos)
				rc = proxyv4_process_reply(pcontext, res);
			else
				rc = RPC_CANTSEND;
		} while (rc == RPC_TIMEDOUT);
	} else {
		rc = RPC_CANTENCODEARGS;
	}

	auth_destroy(au);
	return rc;
}

int proxyv4_compoundv4_execute(const char *caller,
			       const struct user_cred *creds,
			       uint32_t cnt, nfs_argop4 *argoparray,
			       nfs_resop4 *resoparray,
			       struct proxyv4_export *proxyv4_exp)
{
	enum clnt_stat rc;
	struct proxyv4_rpc_io_context *ctx;
	COMPOUND4args arg = {
		.minorversion = FSAL_PROXY_NFS_V4_MINOR,
		.argarray.argarray_val = argoparray,
		.argarray.argarray_len = cnt
	};
	COMPOUND4res res = {
		.resarray.resarray_val = resoparray,
		.resarray.resarray_len = cnt
	};
	struct proxyv4_export_rpc *rpc = &proxyv4_exp->rpc;

	PTHREAD_MUTEX_lock(&rpc->context_lock);
	while (glist_empty(&rpc->free_contexts))
		pthread_cond_wait(&rpc->need_context,
				  &rpc->context_lock);
	ctx =
		glist_first_entry(&rpc->free_contexts,
				  struct proxyv4_rpc_io_context, calls);
	glist_del(&ctx->calls);
	PTHREAD_MUTEX_unlock(&rpc->context_lock);

	/* fill slotid and sequenceid */
	if (argoparray->argop == NFS4_OP_SEQUENCE) {
		SEQUENCE4args *opsequence =
			&argoparray->nfs_argop4_u.opsequence;

		/* set slotid */
		opsequence->sa_slotid = ctx->slotid;
		/* increment and set sequence id */
		opsequence->sa_sequenceid = ++ctx->seqid;
	}

	do {
		rc = proxyv4_compoundv4_call(ctx, creds, &arg,
					     &res, proxyv4_exp);
		if (rc != RPC_SUCCESS)
			LogDebug(COMPONENT_FSAL, "%s failed with %d", caller,
				 rc);
		if (rc == RPC_CANTSEND)
			if (proxyv4_rpc_need_sock(proxyv4_exp))
				return -1;
	} while ((rc == RPC_CANTRECV && (ctx->ioresult == -EAGAIN))
		 || (rc == RPC_CANTSEND));

	PTHREAD_MUTEX_lock(&rpc->context_lock);
	pthread_cond_signal(&rpc->need_context);
	glist_add(&rpc->free_contexts, &ctx->calls);
	PTHREAD_MUTEX_unlock(&rpc->context_lock);

	if (rc == RPC_SUCCESS)
		return res.status;
	return rc;
}

static inline int
proxyv4_nfsv4_call(const struct user_cred *creds, uint32_t cnt,
				     nfs_argop4 *args, nfs_resop4 *resp)
{
	struct proxyv4_export *proxyv4_exp =
		container_of(op_ctx->fsal_export,
			     struct proxyv4_export, exp);

	return proxyv4_compoundv4_execute(__func__, creds, cnt, args, resp,
					  proxyv4_exp);
}

static inline void
proxyv4_get_clientid(struct proxyv4_export *proxyv4_exp, clientid4 *ret)
{
	struct proxyv4_export_rpc *rpc = &proxyv4_exp->rpc;

	PTHREAD_MUTEX_lock(&rpc->proxyv4_clientid_mutex);
	*ret = rpc->proxyv4_clientid;
	PTHREAD_MUTEX_unlock(&rpc->proxyv4_clientid_mutex);
}

static inline void proxyv4_get_client_sessionid(sessionid4 ret)
{
	struct proxyv4_export *proxyv4_exp =
		container_of(op_ctx->fsal_export,
			     struct proxyv4_export, exp);
	struct proxyv4_export_rpc *rpc = &proxyv4_exp->rpc;

	PTHREAD_MUTEX_lock(&rpc->proxyv4_clientid_mutex);
	while (rpc->no_sessionid)
		pthread_cond_wait(&rpc->cond_sessionid,
				  &rpc->proxyv4_clientid_mutex);
	memcpy(ret, rpc->proxyv4_client_sessionid, sizeof(sessionid4));
	PTHREAD_MUTEX_unlock(&rpc->proxyv4_clientid_mutex);
}

static inline void
proxyv4_get_client_sessionid_export(sessionid4 ret,
				    struct proxyv4_export *proxyv4_exp)
{
	struct proxyv4_export_rpc *rpc = &proxyv4_exp->rpc;

	PTHREAD_MUTEX_lock(&rpc->proxyv4_clientid_mutex);
	while (rpc->no_sessionid)
		pthread_cond_wait(&rpc->cond_sessionid,
				  &rpc->proxyv4_clientid_mutex);
	memcpy(ret, rpc->proxyv4_client_sessionid, sizeof(sessionid4));
	PTHREAD_MUTEX_unlock(&rpc->proxyv4_clientid_mutex);
}

static inline void proxyv4_get_client_seqid(struct proxyv4_export *proxyv4_exp,
					    sequenceid4 *ret)
{
	struct proxyv4_export_rpc *rpc = &proxyv4_exp->rpc;

	PTHREAD_MUTEX_lock(&rpc->proxyv4_clientid_mutex);
	*ret = rpc->proxyv4_client_seqid;
	PTHREAD_MUTEX_unlock(&rpc->proxyv4_clientid_mutex);
}

/**
 * Confirm proxyv4_clientid to set a new session.
 *
 * @param[out] new_sessionid The new session id
 * @param[out] new_lease_time Lease time from the background NFSv4.1 server
 *
 * @return 0 on success or NFS error code
 */
static int proxyv4_setsessionid(sessionid4 new_sessionid, uint32_t *lease_time,
				struct proxyv4_export *proxyv4_exp)
{
	int rc;
	int opcnt = 0;
	/* CREATE_SESSION to set session id */
	/* SEQUENCE RECLAIM_COMPLETE PUTROOTFH GETATTR to get lease time */
#define FSAL_SESSIONID_NB_OP_ALLOC 4
	nfs_argop4 arg[FSAL_SESSIONID_NB_OP_ALLOC];
	nfs_resop4 res[FSAL_SESSIONID_NB_OP_ALLOC];
	clientid4 cid;
	sequenceid4 seqid;
	CREATE_SESSION4res *s_res;
	CREATE_SESSION4resok *res_ok;
	callback_sec_parms4 sec_parms4;
	uint32_t fore_ca_rdma_ird_val_sink;
	uint32_t back_ca_rdma_ird_val_sink;

	proxyv4_get_clientid(proxyv4_exp, &cid);
	proxyv4_get_client_seqid(proxyv4_exp, &seqid);
	LogDebug(COMPONENT_FSAL, "Getting new session id for client id %"PRIx64
		 " with sequence id %"PRIx32, cid, seqid);
	s_res = &res->nfs_resop4_u.opcreate_session;
	res_ok = &s_res->CREATE_SESSION4res_u.csr_resok4;
	res_ok->csr_fore_chan_attrs.ca_rdma_ird.ca_rdma_ird_len = 0;
	res_ok->csr_fore_chan_attrs.ca_rdma_ird.ca_rdma_ird_val =
		&fore_ca_rdma_ird_val_sink;
	res_ok->csr_back_chan_attrs.ca_rdma_ird.ca_rdma_ird_len = 0;
	res_ok->csr_back_chan_attrs.ca_rdma_ird.ca_rdma_ird_val =
		&back_ca_rdma_ird_val_sink;

	COMPOUNDV4_ARG_ADD_OP_CREATE_SESSION(opcnt, arg, cid, seqid,
					     (&(proxyv4_exp->info)),
					     &sec_parms4);
	rc = proxyv4_compoundv4_execute(__func__, NULL, opcnt,
					arg, res, proxyv4_exp);
	if (rc != NFS4_OK)
		return -1;

	/*get session_id in res*/
	if (s_res->csr_status != NFS4_OK)
		return -1;

	memcpy(new_sessionid,
	       res_ok->csr_sessionid,
	       sizeof(sessionid4));

	/* Get the lease time */
	opcnt = 0;
	COMPOUNDV4_ARG_ADD_OP_SEQUENCE(opcnt, arg, new_sessionid, NB_RPC_SLOT);
	COMPOUNDV4_ARG_ADD_OP_GLOBAL_RECLAIM_COMPLETE(opcnt, arg);
	COMPOUNDV4_ARG_ADD_OP_PUTROOTFH(opcnt, arg);
	proxyv4_fill_getattr_reply(res + opcnt, (char *)lease_time,
				   sizeof(*lease_time));
	COMPOUNDV4_ARG_ADD_OP_GETATTR(opcnt, arg, lease_bits);

	rc = proxyv4_compoundv4_execute(__func__, NULL, opcnt,
					arg, res, proxyv4_exp);
	if (rc != NFS4_OK) {
		*lease_time = 60;
		LogDebug(COMPONENT_FSAL,
			 "Setting new lease_time to default %d", *lease_time);
	} else {
		*lease_time = ntohl(*lease_time);
		LogDebug(COMPONENT_FSAL,
			 "Getting new lease %d", *lease_time);
	}

	return 0;
}

static int proxyv4_setclientid(clientid4 *new_clientid, sequenceid4 *new_seqid,
			       struct proxyv4_export *proxyv4_exp)
{
	int rc;
#define FSAL_EXCHANGE_ID_NB_OP_ALLOC 1
	nfs_argop4 arg[FSAL_EXCHANGE_ID_NB_OP_ALLOC];
	nfs_resop4 res[FSAL_EXCHANGE_ID_NB_OP_ALLOC];
	client_owner4 clientid;
	char clientid_name[MAXNAMLEN + 1];
	uint64_t temp_verifier;
	EXCHANGE_ID4args opexchange_id;
	EXCHANGE_ID4res *ei_res;
	EXCHANGE_ID4resok *ei_resok;
	char so_major_id_val[NFS4_OPAQUE_LIMIT];
	char eir_server_scope_val[NFS4_OPAQUE_LIMIT];
	nfs_impl_id4 eir_server_impl_id_val;
	struct sockaddr_in sin;
	socklen_t slen = sizeof(sin);
	char addrbuf[sizeof("255.255.255.255")];
	struct proxyv4_export_rpc *rpc = &proxyv4_exp->rpc;

	LogEvent(COMPONENT_FSAL,
		 "Negotiating a new ClientId with the remote server");

	/* prepare input */
	if (getsockname(rpc->rpc_sock, &sin, &slen))
		return -errno;

	rc = snprintf(clientid_name, sizeof(clientid_name),
		      "%s(%d, %d) - GANESHA NFSv4 Proxy",
		      inet_ntop(AF_INET, &sin.sin_addr, addrbuf,
				sizeof(addrbuf)),
		      getpid(), proxyv4_exp->exp.export_id);

	if (rc < 0) {
		return -errno;
	} else if (rc >= sizeof(clientid_name)) {
		return -EINVAL;
	}

	clientid.co_ownerid.co_ownerid_len = strlen(clientid_name);
	clientid.co_ownerid.co_ownerid_val = clientid_name;

	/* copy to intermediate uint64_t to 0-fill or truncate as needed */
	temp_verifier = (uint64_t)nfs_ServerBootTime.tv_sec;
	BUILD_BUG_ON(sizeof(clientid.co_verifier) != sizeof(uint64_t));
	memcpy(&clientid.co_verifier, &temp_verifier, sizeof(uint64_t));


	arg[0].argop = NFS4_OP_EXCHANGE_ID;
	opexchange_id.eia_clientowner = clientid;
	opexchange_id.eia_flags = 0;
	opexchange_id.eia_state_protect.spa_how = SP4_NONE;
	opexchange_id.eia_client_impl_id.eia_client_impl_id_len = 0;
	opexchange_id.eia_client_impl_id.eia_client_impl_id_val = NULL;
	arg[0].nfs_argop4_u.opexchange_id = opexchange_id;

	/* prepare reply */
	ei_res = &res->nfs_resop4_u.opexchange_id;
	ei_resok = &ei_res->EXCHANGE_ID4res_u.eir_resok4;
	ei_resok->eir_server_owner.so_major_id.so_major_id_val =
		so_major_id_val;
	ei_resok->eir_server_scope.eir_server_scope_val = eir_server_scope_val;
	ei_resok->eir_server_impl_id.eir_server_impl_id_val =
		&eir_server_impl_id_val;

	rc = proxyv4_compoundv4_execute(__func__, NULL, 1,
					arg, res, proxyv4_exp);
	if (rc != NFS4_OK) {
		LogDebug(COMPONENT_FSAL,
			 "Compound setclientid res request returned %d",
			 rc);
		return -1;
	}

	/* Keep the confirmed client id and sequence id*/
	if (ei_res->eir_status != NFS4_OK) {
		LogDebug(COMPONENT_FSAL, "EXCHANGE_ID res status is %d",
			 ei_res->eir_status);
		return -1;
	}
	*new_clientid = ei_resok->eir_clientid;
	*new_seqid = ei_resok->eir_sequenceid;

	return 0;
}

static void *proxyv4_clientid_renewer(void *arg)
{
	struct proxyv4_export *proxyv4_exp = arg;
	struct proxyv4_export_rpc *rpc = &proxyv4_exp->rpc;
	int clientid_needed = 1;
	int sessionid_needed = 1;
	uint32_t lease_time = 60;

	SetNameFunction("proxyv4_clientid_renewer");

	rcu_register_thread();
	while (!rpc->close_thread) {
		clientid4 newcid = 0;
		sequenceid4 newseqid = 0;

		if (!sessionid_needed &&
		    proxyv4_rpc_renewer_wait(lease_time - 5, proxyv4_exp)) {
			/* Simply renew the session id you've got */
			nfs_argop4 seq_arg;
			nfs_resop4 res;
			int opcnt = 0;
			int rc;
			sessionid4 sid;
			clientid4 cid;
			SEQUENCE4res *s_res;
			SEQUENCE4resok *s_resok;

			proxyv4_get_clientid(proxyv4_exp, &cid);
			proxyv4_get_client_sessionid_export(sid, proxyv4_exp);
			LogDebug(COMPONENT_FSAL,
				 "Try renew session id for client id %"PRIx64,
				 cid);
			COMPOUNDV4_ARG_ADD_OP_SEQUENCE(opcnt, &seq_arg, sid,
						       NB_RPC_SLOT);
			s_res = &res.nfs_resop4_u.opsequence;
			s_resok = &s_res->SEQUENCE4res_u.sr_resok4;
			s_resok->sr_status_flags = 0;
			rc = proxyv4_compoundv4_execute(__func__, NULL, 1,
							&seq_arg,
							&res, proxyv4_exp);
			if (rc == NFS4_OK && !s_resok->sr_status_flags) {
				LogDebug(COMPONENT_FSAL,
					 "New session id for client id %"PRIu64,
					 cid);
				continue;
			} else if (rc == NFS4_OK &&
				   s_resok->sr_status_flags) {
				LogEvent(COMPONENT_FSAL,
					 "sr_status_flags received on renewing session with seqop : %"
					 PRIu32,
					 s_resok->sr_status_flags);
				continue;
			} else if (rc != NFS4_OK) {
				LogEvent(COMPONENT_FSAL,
					 "Failed to renew session");
			}
		}

		/* early stop test */
		if (rpc->close_thread)
			break;

		/* We've either failed to renew or rpc socket has been
		 * reconnected and we need new clientid or sessionid. */
		if (proxyv4_rpc_need_sock(proxyv4_exp))
			/* early stop test */
			break;

		/* We need a new session_id */
		if (!clientid_needed) {
			sessionid4 new_sessionid;

			LogDebug(COMPONENT_FSAL, "Need %d new session id",
				 sessionid_needed);
			sessionid_needed =
				proxyv4_setsessionid(new_sessionid,
						     &lease_time, proxyv4_exp);
			if (!sessionid_needed) {
				pthread_mutex_t *client_lock =
					&rpc->proxyv4_clientid_mutex;
				PTHREAD_MUTEX_lock(client_lock);
				/* Set new session id */
				memcpy(rpc->proxyv4_client_sessionid,
				       new_sessionid, sizeof(sessionid4));
				rpc->no_sessionid = false;
				pthread_cond_broadcast(&rpc->cond_sessionid);
				/*
				 * We finish our create session request next
				 * one will use the next client sequence id.
				 */
				rpc->proxyv4_client_seqid++;
				PTHREAD_MUTEX_unlock(client_lock);
				continue;
			}
		}

		LogDebug(COMPONENT_FSAL, "Need %d new client id",
			 clientid_needed);

		clientid_needed =
			proxyv4_setclientid(&newcid, &newseqid, proxyv4_exp);
		if (!clientid_needed) {
			PTHREAD_MUTEX_lock(&rpc->proxyv4_clientid_mutex);
			rpc->proxyv4_clientid = newcid;
			rpc->proxyv4_client_seqid = newseqid;
			PTHREAD_MUTEX_unlock(&rpc->proxyv4_clientid_mutex);
		}
	}
	rcu_unregister_thread();
	return NULL;
}

static void free_io_contexts(struct proxyv4_export *proxyv4_exp)
{
	struct glist_head *cur, *n;
	struct proxyv4_export_rpc *rpc = &proxyv4_exp->rpc;

	glist_for_each_safe(cur, n, &rpc->free_contexts) {
		struct proxyv4_rpc_io_context *c =
			container_of(cur, struct proxyv4_rpc_io_context, calls);

		glist_del(cur);
		gsh_free(c);
	}
}

int proxyv4_close_thread(struct proxyv4_export *proxyv4_exp)
{
	int rc;
	struct proxyv4_export_rpc *rpc = &proxyv4_exp->rpc;

	/* setting boolean to stop thread */
	rpc->close_thread = true;

	/* waiting threads ends */

	/*
	 * proxyv4_clientid_renewer and proxyv4_rpc_recv are usually waiting on
	 * cond and rpc_sock respectively. Wake them up.
	 */

	PTHREAD_MUTEX_lock(&rpc->listlock);
	pthread_cond_broadcast(&rpc->sockless);
	close(rpc->rpc_sock);
	PTHREAD_MUTEX_unlock(&rpc->listlock);
	rc = pthread_join(rpc->proxyv4_renewer_thread, NULL);
	if (rc) {
		LogWarn(COMPONENT_FSAL,
			"Error on waiting for the proxyv4_renewer_thread: %d",
			rc);
		return rc;
	}

	rc = pthread_join(rpc->proxyv4_recv_thread, NULL);
	if (rc) {
		LogWarn(COMPONENT_FSAL,
			"Error on waiting for the proxyv4_recv_thread: %d",
			rc);
		return rc;
	}

	return 0;
}

int proxyv4_init_rpc(struct proxyv4_export *proxyv4_exp)
{
	int rc;
	struct proxyv4_export_rpc *rpc = &proxyv4_exp->rpc;
	int i = NB_RPC_SLOT-1;

	PTHREAD_MUTEX_lock(&rpc->listlock);
	glist_init(&rpc->rpc_calls);
	PTHREAD_MUTEX_unlock(&rpc->listlock);

	PTHREAD_MUTEX_lock(&rpc->context_lock);
	glist_init(&rpc->free_contexts);
	PTHREAD_MUTEX_unlock(&rpc->context_lock);

	/**
	 * @todo this lock is not really necessary so long as we can
	 *       only do one export at a time.  This is a reminder that
	 *       there is work to do to get this fnctn to truly be
	 *       per export.
	 */
	PTHREAD_MUTEX_lock(&rpc->listlock);
	if (rpc->rpc_xid == 0)
		rpc->rpc_xid = getpid() ^ time(NULL);
	PTHREAD_MUTEX_unlock(&rpc->listlock);

	if (gethostname(rpc->proxyv4_hostname,
			sizeof(rpc->proxyv4_hostname))) {
		if (strlcpy(rpc->proxyv4_hostname, "NFS-GANESHA/Proxy",
			    sizeof(rpc->proxyv4_hostname))
		    >= sizeof(rpc->proxyv4_hostname)) {
			free_io_contexts(proxyv4_exp);
			return -1;
		}
	}

	for (i = NB_RPC_SLOT-1; i >= 0; i--) {
		struct proxyv4_rpc_io_context *c =
			gsh_malloc(sizeof(*c) + proxyv4_exp->info.srv_sendsize +
				   proxyv4_exp->info.srv_recvsize);
		PTHREAD_MUTEX_init(&c->iolock, NULL);
		PTHREAD_COND_init(&c->iowait, NULL);
		c->nfs_prog = proxyv4_exp->info.srv_prognum;
		c->sendbuf_sz = proxyv4_exp->info.srv_sendsize;
		c->recvbuf_sz = proxyv4_exp->info.srv_recvsize;
		c->sendbuf = (char *)(c + 1);
		c->recvbuf = c->sendbuf + c->sendbuf_sz;
		c->slotid = i;
		c->seqid = 0;
		c->iodone = false;

		PTHREAD_MUTEX_lock(&rpc->context_lock);
		glist_add(&rpc->free_contexts, &c->calls);
		PTHREAD_MUTEX_unlock(&rpc->context_lock);
	}

	rc = pthread_create(&rpc->proxyv4_recv_thread, NULL, proxyv4_rpc_recv,
			    (void *)proxyv4_exp);
	if (rc) {
		LogCrit(COMPONENT_FSAL,
			"Cannot create proxy rpc receiver thread - %s",
			strerror(rc));
		free_io_contexts(proxyv4_exp);
		return rc;
	}

	rc = pthread_create(&rpc->proxyv4_renewer_thread, NULL,
			    proxyv4_clientid_renewer, (void *)proxyv4_exp);
	if (rc) {
		LogCrit(COMPONENT_FSAL,
			"Cannot create proxy clientid renewer thread - %s",
			strerror(rc));
		free_io_contexts(proxyv4_exp);
	}
	return rc;
}

static fsal_status_t proxyv4_make_object(struct fsal_export *export,
					 fattr4 *obj_attributes,
					 const nfs_fh4 *fh,
					 struct fsal_obj_handle **handle,
					 struct fsal_attrlist *attrs_out)
{
	struct proxyv4_obj_handle *proxyv4_hdl;

	proxyv4_hdl = proxyv4_alloc_handle(export, fh, obj_attributes,
					   attrs_out);
	if (proxyv4_hdl == NULL)
		return fsalstat(ERR_FSAL_FAULT, 0);
	*handle = &proxyv4_hdl->obj;
	return fsalstat(ERR_FSAL_NO_ERROR, 0);
}

/*
 * cap maxread and maxwrite config values to background server values
 */
static void
proxyv4_check_maxread_maxwrite(struct fsal_export *export, fattr4 *f4)
{
	fsal_dynamicfsinfo_t info;
	int rc;

	rc = nfs4_Fattr_To_fsinfo(&info, f4);
	if (rc != NFS4_OK) {
		LogWarn(COMPONENT_FSAL,
			"Unable to get maxread and maxwrite from backend NFS server : %d",
			rc);
	} else {
		struct proxyv4_fsal_module *pm =
			container_of(export->fsal,
				     struct proxyv4_fsal_module,
				     module);

		if (info.maxread != 0 &&
		    pm->module.fs_info.maxread > info.maxread) {
			LogWarn(COMPONENT_FSAL,
				"Reduced maxread from %"PRIu64
				" to align with remote server %"PRIu64,
				pm->module.fs_info.maxread, info.maxread);
			pm->module.fs_info.maxread = info.maxread;
		}

		if (info.maxwrite != 0 &&
		    pm->module.fs_info.maxwrite > info.maxwrite) {
			LogWarn(COMPONENT_FSAL,
				"Reduced maxwrite from %"PRIu64
				" to align with remote server %"PRIu64,
				pm->module.fs_info.maxwrite, info.maxwrite);
			pm->module.fs_info.maxwrite = info.maxwrite;
		}
	}
}

/*
 * NULL parent pointer is only used by lookup_path when it starts
 * from the root handle and has its own export pointer, everybody
 * else is supposed to provide a real parent pointer and matching
 * export
 */
static fsal_status_t proxyv4_lookup_impl(struct fsal_obj_handle *parent,
					 struct fsal_export *export,
					 const struct user_cred *cred,
					 const char *path,
					 struct fsal_obj_handle **handle,
					 struct fsal_attrlist *attrs_out)
{
	int rc;
	uint32_t opcnt = 0;
	GETATTR4resok *atok;
	GETATTR4resok *atok_per_file_system_attr = NULL;
	GETFH4resok *fhok;
	sessionid4 sid;
	/* SEQUENCE PUTROOTFH/PUTFH LOOKUP GETFH GETATTR (GETATTR) */
#define FSAL_LOOKUP_NB_OP_ALLOC 6
	nfs_argop4 argoparray[FSAL_LOOKUP_NB_OP_ALLOC];
	nfs_resop4 resoparray[FSAL_LOOKUP_NB_OP_ALLOC];
	char fattr_blob[FATTR_BLOB_SZ];
	char fattr_blob_per_fs[FATTR_BLOB_SZ];
	char padfilehandle[NFS4_FHSIZE];

	if (!handle)
		return fsalstat(ERR_FSAL_INVAL, 0);

	/* SEQUENCE */
	proxyv4_get_client_sessionid(sid);
	COMPOUNDV4_ARG_ADD_OP_SEQUENCE(opcnt, argoparray, sid, NB_RPC_SLOT);
	if (!parent) {
		COMPOUNDV4_ARG_ADD_OP_PUTROOTFH(opcnt, argoparray);
	} else {
		struct proxyv4_obj_handle *proxyv4_obj =
			container_of(parent, struct proxyv4_obj_handle, obj);
		switch (parent->type) {
		case DIRECTORY:
			break;

		default:
			return fsalstat(ERR_FSAL_NOTDIR, 0);
		}

		COMPOUNDV4_ARG_ADD_OP_PUTFH(opcnt, argoparray,
					    proxyv4_obj->fh4);
	}

	if (path) {
		if (!strcmp(path, ".")) {
			if (!parent)
				return fsalstat(ERR_FSAL_FAULT, 0);
		} else if (!strcmp(path, "..")) {
			if (!parent)
				return fsalstat(ERR_FSAL_FAULT, 0);
			COMPOUNDV4_ARG_ADD_OP_LOOKUPP(opcnt, argoparray);
		} else {
			COMPOUNDV4_ARG_ADD_OP_LOOKUP(opcnt, argoparray, path);
		}
	}

	fhok = &resoparray[opcnt].nfs_resop4_u.opgetfh.GETFH4res_u.resok4;
	COMPOUNDV4_ARG_ADD_OP_GETFH(opcnt, argoparray);

	atok =
		proxyv4_fill_getattr_reply(resoparray + opcnt, fattr_blob,
					   sizeof(fattr_blob));

	COMPOUNDV4_ARG_ADD_OP_GETATTR(opcnt, argoparray,
				      proxyv4_bitmap_getattr);

	/* Dynamic ask of server per file system attr */
	if (!parent) {
		atok_per_file_system_attr =
			proxyv4_fill_getattr_reply(resoparray + opcnt,
						   fattr_blob_per_fs,
						   sizeof(fattr_blob_per_fs));
		COMPOUNDV4_ARG_ADD_OP_GETATTR(opcnt, argoparray,
					      proxyv4_bitmap_per_fs);
	}
	fhok->object.nfs_fh4_val = (char *)padfilehandle;
	fhok->object.nfs_fh4_len = sizeof(padfilehandle);

	rc = proxyv4_nfsv4_call(cred, opcnt, argoparray, resoparray);
	if (rc != NFS4_OK)
		return nfsstat4_to_fsal(rc);

	/* Dynamic check of server per file system attr */
	if (!parent) {
		/* maxread and maxwrite */
		fattr4 *obj_attrs = &atok_per_file_system_attr->obj_attributes;

		proxyv4_check_maxread_maxwrite(export, obj_attrs);
	}

	return proxyv4_make_object(export, &atok->obj_attributes, &fhok->object,
				   handle, attrs_out);
}

static fsal_status_t proxyv4_lookup(struct fsal_obj_handle *parent,
				    const char *path,
				    struct fsal_obj_handle **handle,
				    struct fsal_attrlist *attrs_out)
{
	return proxyv4_lookup_impl(parent, op_ctx->fsal_export,
				   &op_ctx->creds, path, handle, attrs_out);
}

/* TODO: make this per-export */
static uint64_t fcnt;

static fsal_status_t proxyv4_mkdir(struct fsal_obj_handle *dir_hdl,
				   const char *name,
				   struct fsal_attrlist *attrib,
				   struct fsal_obj_handle **handle,
				   struct fsal_attrlist *attrs_out)
{
	int rc;
	int opcnt = 0;
	fattr4 input_attr;
	char padfilehandle[NFS4_FHSIZE];
	struct proxyv4_obj_handle *ph;
	char fattr_blob[FATTR_BLOB_SZ];
	GETATTR4resok *atok;
	GETFH4resok *fhok;
	fsal_status_t st;
	sessionid4 sid;
#define FSAL_MKDIR_NB_OP_ALLOC 5 /* SEQUENCE PUTFH CREATE GETFH GETATTR */
	nfs_argop4 argoparray[FSAL_MKDIR_NB_OP_ALLOC];
	nfs_resop4 resoparray[FSAL_MKDIR_NB_OP_ALLOC];

	/*
	 * The caller gives us partial attributes which include mode and owner
	 * and expects the full attributes back at the end of the call.
	 */
	attrib->valid_mask &= ATTR_MODE | ATTR_OWNER | ATTR_GROUP;
	if (proxyv4_fsalattr_to_fattr4(attrib, &input_attr) == -1)
		return fsalstat(ERR_FSAL_INVAL, -1);

	ph = container_of(dir_hdl, struct proxyv4_obj_handle, obj);
	/* SEQUENCE */
	proxyv4_get_client_sessionid(sid);
	COMPOUNDV4_ARG_ADD_OP_SEQUENCE(opcnt, argoparray, sid, NB_RPC_SLOT);
	COMPOUNDV4_ARG_ADD_OP_PUTFH(opcnt, argoparray, ph->fh4);

	resoparray[opcnt].nfs_resop4_u.opcreate.CREATE4res_u.resok4.attrset =
		empty_bitmap;
	COMPOUNDV4_ARG_ADD_OP_MKDIR(opcnt, argoparray, (char *)name,
				    input_attr);

	fhok = &resoparray[opcnt].nfs_resop4_u.opgetfh.GETFH4res_u.resok4;
	fhok->object.nfs_fh4_val = padfilehandle;
	fhok->object.nfs_fh4_len = sizeof(padfilehandle);
	COMPOUNDV4_ARG_ADD_OP_GETFH(opcnt, argoparray);

	atok =
		proxyv4_fill_getattr_reply(resoparray + opcnt, fattr_blob,
					   sizeof(fattr_blob));
	COMPOUNDV4_ARG_ADD_OP_GETATTR(opcnt, argoparray,
				      proxyv4_bitmap_getattr);

	rc = proxyv4_nfsv4_call(&op_ctx->creds, opcnt, argoparray, resoparray);
	nfs4_Fattr_Free(&input_attr);
	if (rc != NFS4_OK)
		return nfsstat4_to_fsal(rc);

	st = proxyv4_make_object(op_ctx->fsal_export, &atok->obj_attributes,
				 &fhok->object, handle, attrs_out);
	if (FSAL_IS_ERROR(st))
		return st;

	return (*handle)->obj_ops->getattrs(*handle, attrib);
}

static fsal_status_t proxyv4_mknod(struct fsal_obj_handle *dir_hdl,
				   const char *name,
				   object_file_type_t nodetype,
				   struct fsal_attrlist *attrib,
				   struct fsal_obj_handle **handle,
				   struct fsal_attrlist *attrs_out)
{
	int rc;
	int opcnt = 0;
	fattr4 input_attr;
	char padfilehandle[NFS4_FHSIZE];
	struct proxyv4_obj_handle *ph;
	char fattr_blob[FATTR_BLOB_SZ];
	GETATTR4resok *atok;
	GETFH4resok *fhok;
	fsal_status_t st;
	enum nfs_ftype4 nf4type;
	specdata4 specdata = { 0, 0 };
	sessionid4 sid;
#define FSAL_MKNOD_NB_OP_ALLOC 5 /* SEQUENCE PUTFH CREATE GETFH GETATTR */
	nfs_argop4 argoparray[FSAL_MKNOD_NB_OP_ALLOC];
	nfs_resop4 resoparray[FSAL_MKNOD_NB_OP_ALLOC];

	switch (nodetype) {
	case CHARACTER_FILE:
		specdata.specdata1 = attrib->rawdev.major;
		specdata.specdata2 = attrib->rawdev.minor;
		nf4type = NF4CHR;
		break;
	case BLOCK_FILE:
		specdata.specdata1 = attrib->rawdev.major;
		specdata.specdata2 = attrib->rawdev.minor;
		nf4type = NF4BLK;
		break;
	case SOCKET_FILE:
		nf4type = NF4SOCK;
		break;
	case FIFO_FILE:
		nf4type = NF4FIFO;
		break;
	default:
		return fsalstat(ERR_FSAL_FAULT, EINVAL);
	}

	/*
	 * The caller gives us partial attributes which include mode and owner
	 * and expects the full attributes back at the end of the call.
	 */
	attrib->valid_mask &= ATTR_MODE | ATTR_OWNER | ATTR_GROUP;
	if (proxyv4_fsalattr_to_fattr4(attrib, &input_attr) == -1)
		return fsalstat(ERR_FSAL_INVAL, -1);

	ph = container_of(dir_hdl, struct proxyv4_obj_handle, obj);
	/* SEQUENCE */
	proxyv4_get_client_sessionid(sid);
	COMPOUNDV4_ARG_ADD_OP_SEQUENCE(opcnt, argoparray, sid, NB_RPC_SLOT);
	COMPOUNDV4_ARG_ADD_OP_PUTFH(opcnt, argoparray, ph->fh4);

	resoparray[opcnt].nfs_resop4_u.opcreate.CREATE4res_u.resok4.attrset =
		empty_bitmap;
	COMPOUNDV4_ARG_ADD_OP_CREATE(opcnt, argoparray, (char *)name, nf4type,
				     input_attr, specdata);

	fhok = &resoparray[opcnt].nfs_resop4_u.opgetfh.GETFH4res_u.resok4;
	fhok->object.nfs_fh4_val = padfilehandle;
	fhok->object.nfs_fh4_len = sizeof(padfilehandle);
	COMPOUNDV4_ARG_ADD_OP_GETFH(opcnt, argoparray);

	atok =
		proxyv4_fill_getattr_reply(resoparray + opcnt, fattr_blob,
					   sizeof(fattr_blob));
	COMPOUNDV4_ARG_ADD_OP_GETATTR(opcnt, argoparray,
				      proxyv4_bitmap_getattr);

	rc = proxyv4_nfsv4_call(&op_ctx->creds, opcnt, argoparray, resoparray);
	nfs4_Fattr_Free(&input_attr);
	if (rc != NFS4_OK)
		return nfsstat4_to_fsal(rc);

	st = proxyv4_make_object(op_ctx->fsal_export, &atok->obj_attributes,
				 &fhok->object, handle, attrs_out);
	if (FSAL_IS_ERROR(st))
		return st;

	return (*handle)->obj_ops->getattrs(*handle, attrib);
}

static fsal_status_t proxyv4_symlink(struct fsal_obj_handle *dir_hdl,
				     const char *name, const char *link_path,
				     struct fsal_attrlist *attrib,
				     struct fsal_obj_handle **handle,
				     struct fsal_attrlist *attrs_out)
{
	int rc;
	int opcnt = 0;
	fattr4 input_attr;
	char padfilehandle[NFS4_FHSIZE];
	char fattr_blob[FATTR_BLOB_SZ];
	sessionid4 sid;
#define FSAL_SYMLINK_NB_OP_ALLOC 5 /* SEQUENCE PUTFH CREATE GETFH GETATTR */
	nfs_argop4 argoparray[FSAL_SYMLINK_NB_OP_ALLOC];
	nfs_resop4 resoparray[FSAL_SYMLINK_NB_OP_ALLOC];
	GETATTR4resok *atok;
	GETFH4resok *fhok;
	fsal_status_t st;
	struct proxyv4_obj_handle *ph;

	/* Tests if symlinking is allowed by configuration. */
	if (!op_ctx->fsal_export->exp_ops.fs_supports(op_ctx->fsal_export,
						      fso_symlink_support))
		return fsalstat(ERR_FSAL_NOTSUPP, ENOTSUP);

	attrib->valid_mask = ATTR_MODE;
	if (proxyv4_fsalattr_to_fattr4(attrib, &input_attr) == -1)
		return fsalstat(ERR_FSAL_INVAL, -1);

	ph = container_of(dir_hdl, struct proxyv4_obj_handle, obj);
	/* SEQUENCE */
	proxyv4_get_client_sessionid(sid);
	COMPOUNDV4_ARG_ADD_OP_SEQUENCE(opcnt, argoparray, sid, NB_RPC_SLOT);
	COMPOUNDV4_ARG_ADD_OP_PUTFH(opcnt, argoparray, ph->fh4);

	resoparray[opcnt].nfs_resop4_u.opcreate.CREATE4res_u.resok4.attrset =
		empty_bitmap;
	COMPOUNDV4_ARG_ADD_OP_SYMLINK(opcnt, argoparray, (char *)name,
				      (char *)link_path, input_attr);

	fhok = &resoparray[opcnt].nfs_resop4_u.opgetfh.GETFH4res_u.resok4;
	fhok->object.nfs_fh4_val = padfilehandle;
	fhok->object.nfs_fh4_len = sizeof(padfilehandle);
	COMPOUNDV4_ARG_ADD_OP_GETFH(opcnt, argoparray);

	atok =
		proxyv4_fill_getattr_reply(resoparray + opcnt, fattr_blob,
					   sizeof(fattr_blob));
	COMPOUNDV4_ARG_ADD_OP_GETATTR(opcnt, argoparray,
				      proxyv4_bitmap_getattr);

	rc = proxyv4_nfsv4_call(&op_ctx->creds, opcnt, argoparray, resoparray);
	nfs4_Fattr_Free(&input_attr);
	if (rc != NFS4_OK)
		return nfsstat4_to_fsal(rc);

	st = proxyv4_make_object(op_ctx->fsal_export, &atok->obj_attributes,
				 &fhok->object, handle, attrs_out);
	if (FSAL_IS_ERROR(st))
		return st;

	return (*handle)->obj_ops->getattrs(*handle, attrib);
}

static fsal_status_t proxyv4_readlink(struct fsal_obj_handle *obj_hdl,
				      struct gsh_buffdesc *link_content,
				      bool refresh)
{
	int rc;
	int opcnt = 0;
	struct proxyv4_obj_handle *ph;
	sessionid4 sid;
#define FSAL_READLINK_NB_OP_ALLOC 3 /* SEQUENCE PUTFH READLINK */
	nfs_argop4 argoparray[FSAL_READLINK_NB_OP_ALLOC];
	nfs_resop4 resoparray[FSAL_READLINK_NB_OP_ALLOC];
	READLINK4resok *rlok;

	ph = container_of(obj_hdl, struct proxyv4_obj_handle, obj);
	/* SEQUENCE */
	proxyv4_get_client_sessionid(sid);
	COMPOUNDV4_ARG_ADD_OP_SEQUENCE(opcnt, argoparray, sid, NB_RPC_SLOT);
	COMPOUNDV4_ARG_ADD_OP_PUTFH(opcnt, argoparray, ph->fh4);

	/* This saves us from having to do one allocation for the XDR,
	   another allocation for the return, and a copy just to get
	   the \NUL terminator. The link length should be cached in
	   the file handle. */

	link_content->len = fsal_default_linksize;
	link_content->addr = gsh_malloc(link_content->len);

	rlok = &resoparray[opcnt].nfs_resop4_u.opreadlink.READLINK4res_u.resok4;
	rlok->link.utf8string_val = link_content->addr;
	rlok->link.utf8string_len = link_content->len;
	COMPOUNDV4_ARG_ADD_OP_READLINK(opcnt, argoparray);

	rc = proxyv4_nfsv4_call(&op_ctx->creds, opcnt, argoparray, resoparray);
	if (rc != NFS4_OK) {
		gsh_free(link_content->addr);
		link_content->addr = NULL;
		link_content->len = 0;
		return nfsstat4_to_fsal(rc);
	}

	rlok->link.utf8string_val[rlok->link.utf8string_len] = '\0';
	link_content->len = rlok->link.utf8string_len + 1;
	return fsalstat(ERR_FSAL_NO_ERROR, 0);
}

static fsal_status_t proxyv4_link(struct fsal_obj_handle *obj_hdl,
				  struct fsal_obj_handle *destdir_hdl,
				  const char *name)
{
	int rc;
	struct proxyv4_obj_handle *tgt;
	struct proxyv4_obj_handle *dst;
	sessionid4 sid;
#define FSAL_LINK_NB_OP_ALLOC 5 /* SEQUENCE PUTFH SAVEFH PUTFH LINK */
	nfs_argop4 argoparray[FSAL_LINK_NB_OP_ALLOC];
	nfs_resop4 resoparray[FSAL_LINK_NB_OP_ALLOC];
	int opcnt = 0;

	/* Tests if hardlinking is allowed by configuration. */
	if (!op_ctx->fsal_export->exp_ops.fs_supports(op_ctx->fsal_export,
						      fso_link_support))
		return fsalstat(ERR_FSAL_NOTSUPP, ENOTSUP);

	tgt = container_of(obj_hdl, struct proxyv4_obj_handle, obj);
	dst = container_of(destdir_hdl, struct proxyv4_obj_handle, obj);

	/* SEQUENCE */
	proxyv4_get_client_sessionid(sid);
	COMPOUNDV4_ARG_ADD_OP_SEQUENCE(opcnt, argoparray, sid, NB_RPC_SLOT);
	COMPOUNDV4_ARG_ADD_OP_PUTFH(opcnt, argoparray, tgt->fh4);
	COMPOUNDV4_ARG_ADD_OP_SAVEFH(opcnt, argoparray);
	COMPOUNDV4_ARG_ADD_OP_PUTFH(opcnt, argoparray, dst->fh4);
	COMPOUNDV4_ARG_ADD_OP_LINK(opcnt, argoparray, (char *)name);

	rc = proxyv4_nfsv4_call(&op_ctx->creds, opcnt, argoparray, resoparray);
	return nfsstat4_to_fsal(rc);
}

#define FSAL_READDIR_NB_OP_ALLOC 3 /* SEQUENCE PUTFH READDIR */
static bool xdr_readdirres(XDR *x, nfs_resop4 *rdres)
{
	int i;
	int res = true;

	for (i = 0; i < FSAL_READDIR_NB_OP_ALLOC; i++) {
		res  = xdr_nfs_resop4(x, rdres + i);
		if (res != true)
			return res;
	}

	return res;
}

/*
 * Trying to guess how many entries can fit into a readdir buffer
 * is complicated and usually results in either gross over-allocation
 * of the memory for results or under-allocation (on large directories)
 * and buffer overruns - just pay the price of allocating the memory
 * inside XDR decoding and free it when done
 */
static fsal_status_t proxyv4_do_readdir(struct proxyv4_obj_handle *ph,
					nfs_cookie4 *cookie, fsal_readdir_cb cb,
					void *cbarg, attrmask_t attrmask,
					bool *eof, bool *again)
{
	uint32_t opcnt = 0;
	int rc;
	entry4 *e4;
	sessionid4 sid;
	nfs_argop4 argoparray[FSAL_READDIR_NB_OP_ALLOC];
	nfs_resop4 resoparray[FSAL_READDIR_NB_OP_ALLOC];
	READDIR4res *result;
	READDIR4resok *rdok;
	fsal_status_t st = { ERR_FSAL_NO_ERROR, 0 };

	/* SEQUENCE */
	proxyv4_get_client_sessionid(sid);
	COMPOUNDV4_ARG_ADD_OP_SEQUENCE(opcnt, argoparray, sid, NB_RPC_SLOT);
	/* PUTFH */
	COMPOUNDV4_ARG_ADD_OP_PUTFH(opcnt, argoparray, ph->fh4);

	/* Before pushing the READDIR op, grab the result pointer */
	result = &resoparray[opcnt].nfs_resop4_u.opreaddir;
	/* Make sure our result object is clean */
	memset(result, 0, sizeof(*result));
	rdok = &result->READDIR4res_u.resok4;

	/* READDIR */
	COMPOUNDV4_ARG_ADD_OP_READDIR(opcnt, argoparray, *cookie,
				      proxyv4_bitmap_getattr);

	rc = proxyv4_nfsv4_call(&op_ctx->creds, opcnt, argoparray, resoparray);
	if (rc != NFS4_OK)
		return nfsstat4_to_fsal(rc);

	*eof = rdok->reply.eof;

	for (e4 = rdok->reply.entries; e4; e4 = e4->nextentry) {
		struct fsal_attrlist attrs;
		struct fsal_obj_handle *handle;
		enum fsal_dir_result cb_rc;

		/* UTF8 name does not include trailing 0 */
		if (e4->name.utf8string_len > MAXNAMLEN)
			return fsalstat(ERR_FSAL_SERVERFAULT, E2BIG);

		if (nfs4_Fattr_To_FSAL_attr(&attrs, &e4->attrs, NULL))
			return fsalstat(ERR_FSAL_FAULT, 0);

		/*
		 *  If *again==false : we are in readahead,
		 *  we only call cb for next entries but don't update result
		 *  for calling readdir.
		 */
		if (*again) {
			*cookie = e4->cookie;
			*eof = rdok->reply.eof && !e4->nextentry;
		}

		/** @todo FSF: this could be handled by getting handle as part
		 *             of readdir attributes. However, if acl is
		 *             requested, we might get it separately to
		 *             avoid over large READDIR response.
		 */
		st = proxyv4_lookup_impl(&ph->obj, op_ctx->fsal_export,
					 &op_ctx->creds,
					 e4->name.utf8string_val,
					 &handle, NULL);
		if (FSAL_IS_ERROR(st))
			break;

		cb_rc = cb(e4->name.utf8string_val, handle, &attrs,
			   cbarg, e4->cookie);

		fsal_release_attrs(&attrs);

		if (cb_rc >= DIR_TERMINATE) {
			*again = false;
			break;
		}
		/*
		 * Read ahead is supported by this FSAL
		 * but limited to the current background readdir request.
		 */
		if (cb_rc >= DIR_READAHEAD && *again) {
			*again = false;
		}
	}

	xdr_free((xdrproc_t) xdr_readdirres, resoparray);
	return st;
}

/**
 *  @todo We might add a verifier to the cookie provided
 *        if server needs one ...
 */
static fsal_status_t proxyv4_readdir(struct fsal_obj_handle *dir_hdl,
				     fsal_cookie_t *whence, void *cbarg,
				     fsal_readdir_cb cb, attrmask_t attrmask,
				     bool *eof)
{
	nfs_cookie4 cookie = 0;
	struct proxyv4_obj_handle *ph;
	bool again = true;

	if (whence)
		cookie = (nfs_cookie4) *whence;

	ph = container_of(dir_hdl, struct proxyv4_obj_handle, obj);

	do {
		fsal_status_t st;

		st = proxyv4_do_readdir(ph, &cookie, cb, cbarg, attrmask, eof,
					&again);
		if (FSAL_IS_ERROR(st))
			return st;
	} while (*eof == false && again);

	return fsalstat(ERR_FSAL_NO_ERROR, 0);
}

static fsal_status_t proxyv4_rename(struct fsal_obj_handle *obj_hdl,
				    struct fsal_obj_handle *olddir_hdl,
				    const char *old_name,
				    struct fsal_obj_handle *newdir_hdl,
				    const char *new_name)
{
	int rc;
	int opcnt = 0;
	sessionid4 sid;
#define FSAL_RENAME_NB_OP_ALLOC 5 /* SEQUENCE PUTFH SAVEFH PUTFH RENAME */
	nfs_argop4 argoparray[FSAL_RENAME_NB_OP_ALLOC];
	nfs_resop4 resoparray[FSAL_RENAME_NB_OP_ALLOC];
	struct proxyv4_obj_handle *src;
	struct proxyv4_obj_handle *tgt;

	src = container_of(olddir_hdl, struct proxyv4_obj_handle, obj);
	tgt = container_of(newdir_hdl, struct proxyv4_obj_handle, obj);
	/* SEQUENCE */
	proxyv4_get_client_sessionid(sid);
	COMPOUNDV4_ARG_ADD_OP_SEQUENCE(opcnt, argoparray, sid, NB_RPC_SLOT);
	COMPOUNDV4_ARG_ADD_OP_PUTFH(opcnt, argoparray, src->fh4);
	COMPOUNDV4_ARG_ADD_OP_SAVEFH(opcnt, argoparray);
	COMPOUNDV4_ARG_ADD_OP_PUTFH(opcnt, argoparray, tgt->fh4);
	COMPOUNDV4_ARG_ADD_OP_RENAME(opcnt, argoparray, (char *)old_name,
				     (char *)new_name);

	rc = proxyv4_nfsv4_call(&op_ctx->creds, opcnt, argoparray, resoparray);
	return nfsstat4_to_fsal(rc);
}

static inline int nfs4_Fattr_To_FSAL_attr_savreqmask(
						struct fsal_attrlist *FSAL_attr,
						fattr4 *Fattr,
						compound_data_t *data)
{
	int rc = 0;
	attrmask_t saved_request_mask = FSAL_attr->request_mask;

	rc = nfs4_Fattr_To_FSAL_attr(FSAL_attr, Fattr, data);
	FSAL_attr->request_mask = saved_request_mask;
	return rc;
}

static fsal_status_t proxyv4_getattrs(struct fsal_obj_handle *obj_hdl,
				      struct fsal_attrlist *attrs)
{
	struct proxyv4_obj_handle *ph;
	int rc;
	uint32_t opcnt = 0;
	sessionid4 sid;
#define FSAL_GETATTR_NB_OP_ALLOC 3 /* SEQUENCE PUTFH GETATTR */
	nfs_argop4 argoparray[FSAL_GETATTR_NB_OP_ALLOC];
	nfs_resop4 resoparray[FSAL_GETATTR_NB_OP_ALLOC];
	GETATTR4resok *atok;
	char fattr_blob[FATTR_BLOB_SZ];

	ph = container_of(obj_hdl, struct proxyv4_obj_handle, obj);

	/* SEQUENCE */
	proxyv4_get_client_sessionid(sid);
	COMPOUNDV4_ARG_ADD_OP_SEQUENCE(opcnt, argoparray, sid, NB_RPC_SLOT);
	COMPOUNDV4_ARG_ADD_OP_PUTFH(opcnt, argoparray, ph->fh4);

	atok = proxyv4_fill_getattr_reply(resoparray + opcnt, fattr_blob,
					  sizeof(fattr_blob));
	COMPOUNDV4_ARG_ADD_OP_GETATTR(opcnt, argoparray,
				      proxyv4_bitmap_getattr);

	rc = proxyv4_nfsv4_call(&op_ctx->creds, opcnt, argoparray, resoparray);

	if (rc != NFS4_OK) {
		if (attrs->request_mask & ATTR_RDATTR_ERR) {
			/* Caller asked for error to be visible. */
			attrs->valid_mask = ATTR_RDATTR_ERR;
		}
		return nfsstat4_to_fsal(rc);
	}

	if (nfs4_Fattr_To_FSAL_attr_savreqmask(attrs, &atok->obj_attributes,
					       NULL) != NFS4_OK)
		return fsalstat(ERR_FSAL_INVAL, 0);

	return fsalstat(ERR_FSAL_NO_ERROR, 0);
}

static fsal_status_t proxyv4_unlink(struct fsal_obj_handle *dir_hdl,
				    struct fsal_obj_handle *obj_hdl,
				    const char *name)
{
	int opcnt = 0;
	int rc;
	struct proxyv4_obj_handle *ph;
	sessionid4 sid;
#define FSAL_UNLINK_NB_OP_ALLOC 4 /* SEQUENCE PUTFH REMOVE GETATTR */
	nfs_argop4 argoparray[FSAL_UNLINK_NB_OP_ALLOC];
	nfs_resop4 resoparray[FSAL_UNLINK_NB_OP_ALLOC];
#if GETATTR_AFTER
	GETATTR4resok *atok;
	char fattr_blob[FATTR_BLOB_SZ];
	struct fsal_attrlist dirattr;
#endif

	ph = container_of(dir_hdl, struct proxyv4_obj_handle, obj);
	/* SEQUENCE */
	proxyv4_get_client_sessionid(sid);
	COMPOUNDV4_ARG_ADD_OP_SEQUENCE(opcnt, argoparray, sid, NB_RPC_SLOT);
	COMPOUNDV4_ARG_ADD_OP_PUTFH(opcnt, argoparray, ph->fh4);
	COMPOUNDV4_ARG_ADD_OP_REMOVE(opcnt, argoparray, (char *)name);

#if GETATTR_AFTER
	atok =
		proxyv4_fill_getattr_reply(resoparray + opcnt, fattr_blob,
					   sizeof(fattr_blob));
	COMPOUNDV4_ARG_ADD_OP_GETATTR(opcnt, argoparray,
				      proxyv4_bitmap_getattr);
#endif

	rc = proxyv4_nfsv4_call(&op_ctx->creds, opcnt, argoparray, resoparray);
	if (rc != NFS4_OK)
		return nfsstat4_to_fsal(rc);

#if GETATTR_AFTER
	if (nfs4_Fattr_To_FSAL_attr(&dirattr, &atok->obj_attributes, NULL) ==
	    NFS4_OK)
		ph->attributes = dirattr;
#endif

	return fsalstat(ERR_FSAL_NO_ERROR, 0);
}

static fsal_status_t
proxyv4_handle_to_wire(const struct fsal_obj_handle *obj_hdl,
		       fsal_digesttype_t output_type,
		       struct gsh_buffdesc *fh_desc)
{
	struct proxyv4_obj_handle *ph =
		container_of(obj_hdl, struct proxyv4_obj_handle, obj);
	size_t fhs;
	void *data;

	/* sanity checks */
	if (!fh_desc || !fh_desc->addr)
		return fsalstat(ERR_FSAL_FAULT, 0);

	switch (output_type) {
	case FSAL_DIGEST_NFSV3:
#ifdef PROXYV4_HANDLE_MAPPING
		fhs = sizeof(ph->h23);
		data = &ph->h23;
		break;
#endif
	case FSAL_DIGEST_NFSV4:
		fhs = ph->blob.len;
		data = &ph->blob;
		break;
	default:
		return fsalstat(ERR_FSAL_SERVERFAULT, 0);
	}

	if (fh_desc->len < fhs)
		return fsalstat(ERR_FSAL_TOOSMALL, 0);
	memcpy(fh_desc->addr, data, fhs);
	fh_desc->len = fhs;
	return fsalstat(ERR_FSAL_NO_ERROR, 0);
}

static void proxyv4_handle_to_key(struct fsal_obj_handle *obj_hdl,
				  struct gsh_buffdesc *fh_desc)
{
	struct proxyv4_obj_handle *ph =
		container_of(obj_hdl, struct proxyv4_obj_handle, obj);
	fh_desc->addr = &ph->blob;
	fh_desc->len = ph->blob.len;
}

static void proxyv4_hdl_release(struct fsal_obj_handle *obj_hdl)
{
	struct proxyv4_obj_handle *ph =
		container_of(obj_hdl, struct proxyv4_obj_handle, obj);

	fsal_obj_handle_fini(obj_hdl);

	gsh_free(ph);
}

/*
 * In this first FSAL_PROXY_V4 support_ex version without state
 * nothing to do on close.
 */
static fsal_status_t proxyv4_close(struct fsal_obj_handle *obj_hdl)
{
	return fsalstat(ERR_FSAL_NOT_OPENED, 0);
}

/*
 * support_ex methods
 *
 * This first dirty version of support_ex in FSAL_PROXY_V4 doesn't take care of
 * any state.
 *
 * The goal achieves by this first dirty version is only to be compliant with
 * support_ex fsal api.
 */

/**
 * @brief Fill share_access and share_deny fields of an OPEN4args struct
 *
 * This function fills share_access and share_deny fields of an OPEN4args
 * struct to prepare an OPEN v4.1 call.
 *
 * @param[in]     openflags      fsal open flags
 * @param[in,out] share_access   share_access field to be filled
 * @param[in,out] share_deny     share_deny field to be filled
 *
 * @return FSAL status.
 */
static fsal_status_t  fill_share_OPEN4args(uint32_t *share_access,
					   uint32_t *share_deny,
					   fsal_openflags_t openflags)
{

	/* share access */
	*share_access = 0;
	if (openflags & FSAL_O_READ)
		*share_access |= OPEN4_SHARE_ACCESS_READ;
	if (openflags & FSAL_O_WRITE)
		*share_access |= OPEN4_SHARE_ACCESS_WRITE;
	/* share write */
	*share_deny = OPEN4_SHARE_DENY_NONE;
	if (openflags & FSAL_O_DENY_READ)
		*share_deny |= OPEN4_SHARE_DENY_READ;
	if (openflags & FSAL_O_DENY_WRITE ||
	    openflags & FSAL_O_DENY_WRITE_MAND)
		*share_deny |= OPEN4_SHARE_DENY_WRITE;

	return fsalstat(ERR_FSAL_NO_ERROR, 0);
}

/**
 * @brief Fill openhow field of an OPEN4args struct
 *
 * This function fills an openflags4 openhow field of an OPEN4args struct
 * to prepare an OPEN v4.1 call.
 *
 * @param[in]     attrs_in       open attributes
 * @param[in]     create_mode    open create mode
 * @param[in]     verifier       open verifier
 * @param[in,out] openhow        openhow field to be filled
 * @param[in,out] inattrs        created inattrs (need to be freed by calling
 *                               nfs4_Fattr_Free)
 * @param[out]    setattr_needed an additional setattr op is needed
 *
 * @return FSAL status.
 */
static fsal_status_t fill_openhow_OPEN4args(openflag4 *openhow,
					    fattr4 inattrs,
					    enum fsal_create_mode createmode,
					    fsal_verifier_t verifier,
					    bool *setattr_needed,
					    const char *name,
					    fsal_openflags_t openflags)
{
	if (openhow == NULL)
		return fsalstat(ERR_FSAL_INVAL, -1);

	/* openhow */
	/* not an open by handle and flag create */
	if (name && createmode != FSAL_NO_CREATE) {
		createhow4 *how = &(openhow->openflag4_u.how);
		verifier4 *verf;

		openhow->opentype = OPEN4_CREATE;
		switch (createmode) {
		case FSAL_UNCHECKED:
			how->mode = UNCHECKED4;
			how->createhow4_u.createattrs = inattrs;
			break;
		case FSAL_GUARDED:
		case FSAL_EXCLUSIVE_9P:
			how->mode = GUARDED4;
			how->createhow4_u.createattrs = inattrs;
			break;
		case FSAL_EXCLUSIVE:
			how->mode = EXCLUSIVE4;
			verf = &how->createhow4_u.createverf;
			FSAL_VERIFIER_T_TO_VERIFIER4(verf, verifier);
			/* no way to set attr in same op than old EXCLUSIVE4 */
			if (inattrs.attrmask.bitmap4_len > 0) {
				int i = 0;

				for (i = 0; i < inattrs.attrmask.bitmap4_len;
				     i++) {
					if (inattrs.attrmask.map[i]) {
						*setattr_needed = true;
						break;
					}
				}
			}
			break;
		case FSAL_EXCLUSIVE_41:
			how->mode = EXCLUSIVE4_1;
			verf = &how->createhow4_u.ch_createboth.cva_verf;
			FSAL_VERIFIER_T_TO_VERIFIER4(verf, verifier);
			how->createhow4_u.ch_createboth.cva_attrs = inattrs;
			/*
			 * We assume verifier is stored in time metadata.
			 *
			 * We had better check suppattr_exclcreat from
			 * background server.
			 */
			if (inattrs.attrmask.bitmap4_len >= 2 &&
			    (inattrs.attrmask.map[1] &
			     (1U << (FATTR4_TIME_METADATA - 32)))) {
				*setattr_needed = true;
				// cva_attrs was set to inattrs above.
				inattrs.attrmask.map[1] &=
					~(1U << (FATTR4_TIME_METADATA - 32));
			}
			break;
		default:
			return fsalstat(ERR_FSAL_FAULT,
					EINVAL);
		}
	} else
		openhow->opentype = OPEN4_NOCREATE;

	/* include open by handle and TRUNCATE case in setattr_needed */
	if (!name && openflags & FSAL_O_TRUNC)
		*setattr_needed = true;

	return fsalstat(ERR_FSAL_NO_ERROR, 0);
}

static fsal_status_t proxyv4_open2(struct fsal_obj_handle *obj_hdl,
				   struct state_t *state,
				   fsal_openflags_t openflags,
				   enum fsal_create_mode createmode,
				   const char *name,
				   struct fsal_attrlist *attrs_in,
				   fsal_verifier_t verifier,
				   struct fsal_obj_handle **new_obj,
				   struct fsal_attrlist *attrs_out,
				   bool *caller_perm_check)
{
	struct proxyv4_obj_handle *ph;
	int rc; /* return code of nfs call */
	int opcnt = 0; /* nfs arg counter */
	fsal_status_t st; /* return code of fsal call */
	char padfilehandle[NFS4_FHSIZE]; /* gotten FH */
	char owner_val[128];
	int owner_len = 0;
	uint32_t share_access = 0;
	uint32_t share_deny = 0;
	openflag4 openhow;
	fattr4 inattrs;
	open_claim4 claim;
	sessionid4 sid;
	/* SEQUENCE, PUTFH, OPEN, GETFH, GETATTR */
#define FSAL_OPEN_NB_OP 5
	nfs_argop4 argoparray[FSAL_OPEN_NB_OP];
	nfs_resop4 resoparray[FSAL_OPEN_NB_OP];
	/* SEQUENCE, PUTFH, SETATTR, GETATTR */
#define FSAL_OPEN_SETATTR_NB_OP 4
	nfs_argop4 setattr_argoparray[FSAL_OPEN_SETATTR_NB_OP];
	nfs_resop4 setattr_resoparray[FSAL_OPEN_SETATTR_NB_OP];
	nfs_resop4 *resop;
	OPEN4resok *opok;
	GETFH4resok *fhok;
	GETATTR4resok *atok;
	char fattr_blob[FATTR_BLOB_SZ];
	bool setattr_needed = false;

	/* we have not done yet any check */
	*caller_perm_check = true;

	/* get back proxy handle */
	ph = container_of(obj_hdl, struct proxyv4_obj_handle, obj);

	/* include TRUNCATE case in attrs_in */
	if (openflags & FSAL_O_TRUNC) {
		attrs_in->valid_mask |= ATTR_SIZE;
		attrs_in->filesize = 0;
	}

	/* fill inattrs */
	if (proxyv4_fsalattr_to_fattr4(attrs_in, &inattrs) == -1)
		return fsalstat(ERR_FSAL_INVAL, -1);

	/* Three cases need to do an open :
	 * -open by name to get an handle
	 * -open by handle to get attrs_out
	 * -open by handle to truncate
	 */
	if (name || attrs_out || openflags & FSAL_O_TRUNC) {
		/*
		 * We do the open to get handle, check perm, check share, trunc,
		 * create if needed ...
		 */
		/* open call will do the perm check */
		*caller_perm_check = false;

		/* prepare open call */
		/* SEQUENCE */
		proxyv4_get_client_sessionid(sid);
		COMPOUNDV4_ARG_ADD_OP_SEQUENCE(opcnt, argoparray, sid,
					       NB_RPC_SLOT);

		/* PUTFH */
		COMPOUNDV4_ARG_ADD_OP_PUTFH(opcnt, argoparray, ph->fh4);

		/* OPEN */
		/* prepare answer */
		opok = &resoparray[opcnt].nfs_resop4_u.opopen.OPEN4res_u.resok4;
		opok->rflags = 0; /* set to NULL for safety */
		opok->attrset = empty_bitmap; /* set to empty for safety */
		/* prepare open input args */
		/* share_access and share_deny */
		st = fill_share_OPEN4args(&share_access, &share_deny,
					  openflags);
		if (FSAL_IS_ERROR(st)) {
			nfs4_Fattr_Free(&inattrs);
			return st;
		}

		/* owner */
		owner_len = snprintf(owner_val, sizeof(owner_val),
				     "GANESHA/PROXY: pid=%u %" PRIu64, getpid(),
				     atomic_inc_uint64_t(&fcnt));

		if (owner_len < 0) {
			int error = errno;

			LogCrit(COMPONENT_FSAL,
				"Unexpected return from snprintf %d error %s (%d)",
				owner_len, strerror(error), error);
			return posix2fsal_status(error);
		} else if (owner_len >= sizeof(owner_val)) {
			LogMajor(COMPONENT_FSAL, "Owner length too long");
			return posix2fsal_status(EINVAL);
		}

		/* inattrs and openhow */
		st = fill_openhow_OPEN4args(&openhow, inattrs, createmode,
					    verifier, &setattr_needed, name,
					    openflags);
		if (FSAL_IS_ERROR(st)) {
			nfs4_Fattr_Free(&inattrs);
			return st;
		}

		/* claim : first support_ex version, no state -> no claim */
		if (name) {
			/* open by name */
			claim.claim = CLAIM_NULL;
			claim.open_claim4_u.file.utf8string_val = (char *)name;
			claim.open_claim4_u.file.utf8string_len = strlen(name);
		} else {
			/* open by handle */
			claim.claim = CLAIM_FH;
		}
		/* add open */
		COMPOUNDV4_ARGS_ADD_OP_OPEN_4_1(opcnt, argoparray, share_access,
						share_deny, owner_val,
						owner_len, openhow, claim);

		/* GETFH */
		/* prepare answer */

		// NOTE(boulos): This is here to make this line than 80 chars.
		resop = resoparray + opcnt;
		fhok = &resop->nfs_resop4_u.opgetfh.GETFH4res_u.resok4;

		fhok->object.nfs_fh4_val = padfilehandle;
		fhok->object.nfs_fh4_len = sizeof(padfilehandle);
		COMPOUNDV4_ARG_ADD_OP_GETFH(opcnt, argoparray);
		if (!setattr_needed && (new_obj || attrs_out)) {
			/* GETATTR */
			atok = proxyv4_fill_getattr_reply(resoparray + opcnt,
							  fattr_blob,
							  sizeof(fattr_blob));
			COMPOUNDV4_ARG_ADD_OP_GETATTR(opcnt, argoparray,
						      proxyv4_bitmap_getattr);
		}
		/* nfs call*/
		rc = proxyv4_nfsv4_call(&op_ctx->creds, opcnt, argoparray,
					resoparray);
		if (rc != NFS4_OK) {
			nfs4_Fattr_Free(&inattrs);
			return nfsstat4_to_fsal(rc);
		}


		/* update stateid in current state */
		if (state) {
			struct proxyv4_state *proxyv4_state_id =
				container_of(state,
					     struct proxyv4_state,
					     state);

			proxyv4_state_id->stateid = opok->stateid;
		}
	}

	if (setattr_needed) {
		opcnt = 0;
		/* SEQUENCE */
		proxyv4_get_client_sessionid(sid);
		COMPOUNDV4_ARG_ADD_OP_SEQUENCE(opcnt, setattr_argoparray, sid,
					       NB_RPC_SLOT);
		/* PUTFH */
		COMPOUNDV4_ARG_ADD_OP_PUTFH(opcnt, setattr_argoparray,
					    fhok->object);
		/* SETATTR for truncate */
		setattr_resoparray[opcnt].nfs_resop4_u.opsetattr.attrsset =
			empty_bitmap;
		/* We have a stateid */
		/* cause we did an open when we set setattr_needed. */
		COMPOUNDV4_ARG_ADD_OP_SETATTR(opcnt, setattr_argoparray,
					      inattrs, opok->stateid.other);

		if (new_obj || attrs_out) {
			/* GETATTR */
			nfs_resop4 *setattr_res = setattr_resoparray + opcnt;

			atok = proxyv4_fill_getattr_reply(setattr_res,
							  fattr_blob,
							  sizeof(fattr_blob));
			COMPOUNDV4_ARG_ADD_OP_GETATTR(opcnt, setattr_argoparray,
						      proxyv4_bitmap_getattr);
		}

		/* nfs call*/
		rc = proxyv4_nfsv4_call(&op_ctx->creds, opcnt,
					setattr_argoparray,
					setattr_resoparray);
		if (rc != NFS4_OK) {
			nfs4_Fattr_Free(&inattrs);
			return nfsstat4_to_fsal(rc);
		}
	}

	/* clean inattrs */
	nfs4_Fattr_Free(&inattrs);

	/* create a new object if asked and attrs_out by the way */
	if (new_obj) {
		if (name) {
			/* create new_obj and set attrs_out*/
			st = proxyv4_make_object(op_ctx->fsal_export,
						 &atok->obj_attributes,
						 &fhok->object,
						 new_obj, attrs_out);
			if (FSAL_IS_ERROR(st))
				return st;
		} else {
			*new_obj = obj_hdl;
		}
	}

	/* set attrs_out if needed and not yet done by creating new object */
	if (attrs_out && (!new_obj || (new_obj && !name))) {
		rc = nfs4_Fattr_To_FSAL_attr(attrs_out, &atok->obj_attributes,
					     NULL);
		if (rc != NFS4_OK)
			return nfsstat4_to_fsal(rc);
	}

	return fsalstat(ERR_FSAL_NO_ERROR, 0);
}

/* XXX Note that this only currently supports a vector size of 1 */
static void proxyv4_read2(struct fsal_obj_handle *obj_hdl,
			  bool bypass,
			  fsal_async_cb done_cb,
			  struct fsal_io_arg *read_arg,
			  void *caller_arg)
{
	int maxReadSize;
	int rc;
	int opcnt = 0;
	struct proxyv4_obj_handle *ph;
	sessionid4 sid;
#define FSAL_READ2_NB_OP_ALLOC 3 /* SEQUENCE + PUTFH + READ */
	nfs_argop4 argoparray[FSAL_READ2_NB_OP_ALLOC];
	nfs_resop4 resoparray[FSAL_READ2_NB_OP_ALLOC];
	READ4resok *rok;
	size_t iov_len;

	ph = container_of(obj_hdl, struct proxyv4_obj_handle, obj);

	maxReadSize =
		op_ctx->fsal_export->exp_ops.fs_maxread(op_ctx->fsal_export);
	iov_len = read_arg->iov[0].iov_len;

	if (iov_len > maxReadSize)
		iov_len = maxReadSize;


	/* SEQUENCE */
	proxyv4_get_client_sessionid(sid);
	COMPOUNDV4_ARG_ADD_OP_SEQUENCE(opcnt, argoparray, sid, NB_RPC_SLOT);
	/* prepare PUTFH */
	COMPOUNDV4_ARG_ADD_OP_PUTFH(opcnt, argoparray, ph->fh4);
	/* prepare READ */
	rok = &resoparray[opcnt].nfs_resop4_u.opread.READ4res_u.resok4;
	rok->data.data_val = read_arg->iov[0].iov_base;
	rok->data.data_len = iov_len;
	if (bypass)
		COMPOUNDV4_ARG_ADD_OP_READ_BYPASS(opcnt, argoparray,
						  read_arg->offset,
						  iov_len);
	else {
		if (read_arg->state) {
			struct proxyv4_state *proxyv4_state_id =
				container_of(read_arg->state,
					     struct proxyv4_state,
					     state);
			char *other = proxyv4_state_id->stateid.other;

			COMPOUNDV4_ARG_ADD_OP_READ(opcnt, argoparray,
						   read_arg->offset,
						   iov_len,
						   other);
		} else {
			COMPOUNDV4_ARG_ADD_OP_READ_STATELESS(opcnt, argoparray,
							     read_arg->offset,
							     iov_len);
		}
	}

	/* nfs call */
	rc = proxyv4_nfsv4_call(&op_ctx->creds, opcnt, argoparray, resoparray);
	if (rc != NFS4_OK) {
		done_cb(obj_hdl, nfsstat4_to_fsal(rc), read_arg, caller_arg);
		return;
	}

	read_arg->end_of_file = rok->eof;
	read_arg->io_amount = rok->data.data_len;
	if (read_arg->info) {
		read_arg->info->io_content.what = NFS4_CONTENT_DATA;
		read_arg->info->io_content.data.d_offset = read_arg->offset +
			read_arg->io_amount;
		read_arg->info->io_content.data.d_data.data_len =
			read_arg->io_amount;
		read_arg->info->io_content.data.d_data.data_val =
			read_arg->iov[0].iov_base;
	}
	done_cb(obj_hdl, fsalstat(0, 0), read_arg, caller_arg);
}

static void proxyv4_write2(struct fsal_obj_handle *obj_hdl,
			   bool bypass,
			   fsal_async_cb done_cb,
			   struct fsal_io_arg *write_arg,
			   void *caller_arg)
{
	int maxWriteSize;
	int rc;
	int opcnt = 0;
	sessionid4 sid;
#define FSAL_WRITE_NB_OP_ALLOC 3 /* SEQUENCE + PUTFH + WRITE */
	nfs_argop4 argoparray[FSAL_WRITE_NB_OP_ALLOC];
	nfs_resop4 resoparray[FSAL_WRITE_NB_OP_ALLOC];
	WRITE4resok *wok;
	struct proxyv4_obj_handle *ph;
	stable_how4 stable_how;
	size_t buffer_size = write_arg->iov[0].iov_len;

	ph = container_of(obj_hdl, struct proxyv4_obj_handle, obj);

	/* check max write size */
	maxWriteSize =
		op_ctx->fsal_export->exp_ops.fs_maxwrite(op_ctx->fsal_export);
	if (buffer_size > maxWriteSize)
		buffer_size = maxWriteSize;

	/* SEQUENCE */
	proxyv4_get_client_sessionid(sid);
	COMPOUNDV4_ARG_ADD_OP_SEQUENCE(opcnt, argoparray, sid, NB_RPC_SLOT);
	/* prepare PUTFH */
	COMPOUNDV4_ARG_ADD_OP_PUTFH(opcnt, argoparray, ph->fh4);
	/* prepare write */
	wok = &resoparray[opcnt].nfs_resop4_u.opwrite.WRITE4res_u.resok4;

	if (write_arg->fsal_stable)
		stable_how = DATA_SYNC4;
	else
		stable_how = UNSTABLE4;
	if (write_arg->state) {
		struct proxyv4_state *proxyv4_state_id =
			container_of(write_arg->state,
				     struct proxyv4_state,
				     state);

		COMPOUNDV4_ARG_ADD_OP_WRITE(opcnt, argoparray,
					    write_arg->offset,
					    write_arg->iov[0].iov_base,
					    buffer_size, stable_how,
					    proxyv4_state_id->stateid.other);
	} else {
		char *base = write_arg->iov[0].iov_base;

		COMPOUNDV4_ARG_ADD_OP_WRITE_STATELESS(opcnt, argoparray,
						      write_arg->offset,
						      base,
						      buffer_size,
						      stable_how);
	}

	/* nfs call */
	rc = proxyv4_nfsv4_call(&op_ctx->creds, opcnt, argoparray, resoparray);
	if (rc != NFS4_OK) {
		done_cb(obj_hdl, nfsstat4_to_fsal(rc), write_arg, caller_arg);
		return;
	}

	/* get res */
	write_arg->io_amount = wok->count;
	if (wok->committed == UNSTABLE4)
		write_arg->fsal_stable = false;
	else
		write_arg->fsal_stable = true;

	done_cb(obj_hdl, fsalstat(ERR_FSAL_NO_ERROR, 0), write_arg, caller_arg);
}

static fsal_status_t proxyv4_close2(struct fsal_obj_handle *obj_hdl,
				    struct state_t *state)
{
	struct proxyv4_obj_handle *ph;
	int rc;
	int opcnt = 0;
	sessionid4 sessionid;
	/* SEQUENCE, PUTFH, CLOSE */
#define FSAL_CLOSE_NB_OP_ALLOC 3
	nfs_argop4 argoparray[FSAL_CLOSE_NB_OP_ALLOC];
	nfs_resop4 resoparray[FSAL_CLOSE_NB_OP_ALLOC];
	char All_Zero[] = "\0\0\0\0\0\0\0\0\0\0\0\0";	/* 12 times \0 */
	struct proxyv4_state *proxyv4_state_id = NULL;

	ph = container_of(obj_hdl, struct proxyv4_obj_handle, obj);

	/* Check if this was a "stateless" open,
	 * then nothing is to be done at close */
	if (!state) {
		return fsalstat(ERR_FSAL_NO_ERROR, 0);
	} else {
		proxyv4_state_id =
			container_of(state, struct proxyv4_state, state);

		if (!memcmp(proxyv4_state_id->stateid.other, All_Zero, 12))
			return fsalstat(ERR_FSAL_NO_ERROR, 0);
	}

	/* SEQUENCE */
	proxyv4_get_client_sessionid(sessionid);
	COMPOUNDV4_ARG_ADD_OP_SEQUENCE(opcnt, argoparray, sessionid,
				       NB_RPC_SLOT);
	/* PUTFH */
	COMPOUNDV4_ARG_ADD_OP_PUTFH(opcnt, argoparray, ph->fh4);
	/* CLOSE */
	if (state)
		COMPOUNDV4_ARG_ADD_OP_CLOSE_4_1(opcnt, argoparray,
						proxyv4_state_id->stateid);
	else
		COMPOUNDV4_ARG_ADD_OP_CLOSE_4_1_STATELESS(opcnt, argoparray);

	rc = proxyv4_nfsv4_call(&op_ctx->creds, opcnt, argoparray, resoparray);
	if (rc != NFS4_OK) {
		return nfsstat4_to_fsal(rc);
	}

	/* We clean local saved stateid. */
	if (state)
		memset(&proxyv4_state_id->stateid, 0, sizeof(stateid4));

	return fsalstat(ERR_FSAL_NO_ERROR, 0);
}

static fsal_status_t proxyv4_setattr2(struct fsal_obj_handle *obj_hdl,
				      bool bypass,
				      struct state_t *state,
				      struct fsal_attrlist *attrib_set)
{
	int rc;
	fattr4 input_attr;
	uint32_t opcnt = 0;
	struct proxyv4_obj_handle *ph;
	sessionid4 sid;
#define FSAL_SETATTR2_NB_OP_ALLOC 3 /* SEQUENCE PUTFH SETATTR */
	nfs_argop4 argoparray[FSAL_SETATTR2_NB_OP_ALLOC];
	nfs_resop4 resoparray[FSAL_SETATTR2_NB_OP_ALLOC];

	/*prepare attributes */
	/*
	 * No way to update CTIME using a NFSv4 SETATTR.
	 * Server will return NFS4ERR_INVAL (22).
	 * time_metadata is a readonly attribute in NFSv4 and NFSv4.1.
	 * (section 5.7 in RFC7530 or RFC5651)
	 * Nevermind : this update is useless, we prevent it.
	 */
	FSAL_UNSET_MASK(attrib_set->valid_mask, ATTR_CTIME);

	if (FSAL_TEST_MASK(attrib_set->valid_mask, ATTR_MODE)) {
		struct fsal_export *exp = op_ctx->fsal_export;

		attrib_set->mode &= ~exp->exp_ops.fs_umask(exp);
	}

	ph = container_of(obj_hdl, struct proxyv4_obj_handle, obj);

	if (proxyv4_fsalattr_to_fattr4(attrib_set, &input_attr) == -1)
		return fsalstat(ERR_FSAL_INVAL, EINVAL);

	/* SEQUENCE */
	proxyv4_get_client_sessionid(sid);
	COMPOUNDV4_ARG_ADD_OP_SEQUENCE(opcnt, argoparray, sid, NB_RPC_SLOT);
	/* prepare PUTFH */
	COMPOUNDV4_ARG_ADD_OP_PUTFH(opcnt, argoparray, ph->fh4);

	/* prepare SETATTR */
	resoparray[opcnt].nfs_resop4_u.opsetattr.attrsset = empty_bitmap;

	/* We don't use the special "bypass" stateid. */
	/* Indeed, bypass state will be treated like anonymous value. */
	/* RFC 5661, section 8.2.3 */
	if (state) {
		struct proxyv4_state *proxyv4_state_id =
			container_of(state,
				     struct proxyv4_state,
				     state);

		COMPOUNDV4_ARG_ADD_OP_SETATTR(opcnt, argoparray, input_attr,
					      proxyv4_state_id->stateid.other);
	} else {
		COMPOUNDV4_ARG_ADD_OP_SETATTR_STATELESS(opcnt, argoparray,
							input_attr);
	}

	/* nfs call */
	rc = proxyv4_nfsv4_call(&op_ctx->creds, opcnt, argoparray, resoparray);
	nfs4_Fattr_Free(&input_attr);
	if (rc != NFS4_OK)
		return nfsstat4_to_fsal(rc);

	return fsalstat(ERR_FSAL_NO_ERROR, 0);
}

static fsal_openflags_t proxyv4_status2(struct fsal_obj_handle *obj_hdl,
					struct state_t *state)
{
	/* first version of support_ex, no state, no saved openflags */
	fsal_openflags_t null_flags = 0; /* closed and deny_none*/

	return null_flags;
}

static fsal_status_t proxyv4_reopen2(struct fsal_obj_handle *obj_hdl,
				     struct state_t *state,
				     fsal_openflags_t openflags)
{
	/* no way to open by handle in v4 */
	/* waiting for v4.1 or solid state to really do the job */

	return fsalstat(ERR_FSAL_NO_ERROR, 0);
}

static fsal_status_t proxyv4_commit2(struct fsal_obj_handle *obj_hdl,
				     off_t offset,
				     size_t len)
{
	struct proxyv4_obj_handle *ph;
	int rc; /* return code of nfs call */
	int opcnt = 0; /* nfs arg counter */
	sessionid4 sid;
#define FSAL_COMMIT2_NB_OP 3 /* SEQUENCE, PUTFH, COMMIT */
	nfs_argop4 argoparray[FSAL_COMMIT2_NB_OP];
	nfs_resop4 resoparray[FSAL_COMMIT2_NB_OP];

	ph = container_of(obj_hdl, struct proxyv4_obj_handle, obj);

	/* SEQUENCE */
	proxyv4_get_client_sessionid(sid);
	COMPOUNDV4_ARG_ADD_OP_SEQUENCE(opcnt, argoparray, sid, NB_RPC_SLOT);
	/* prepare PUTFH */
	COMPOUNDV4_ARG_ADD_OP_PUTFH(opcnt, argoparray, ph->fh4);

	/* prepare COMMIT */
	COMPOUNDV4_ARG_ADD_OP_COMMIT(opcnt, argoparray, offset, len);

	/* nfs call */
	rc = proxyv4_nfsv4_call(&op_ctx->creds, opcnt, argoparray, resoparray);
	if (rc != NFS4_OK)
		return nfsstat4_to_fsal(rc);

	return fsalstat(ERR_FSAL_NO_ERROR, 0);
}

void proxyv4_handle_ops_init(struct fsal_obj_ops *ops)
{
	fsal_default_obj_ops_init(ops);

	ops->release = proxyv4_hdl_release;
	ops->lookup = proxyv4_lookup;
	ops->readdir = proxyv4_readdir;
	ops->mkdir = proxyv4_mkdir;
	ops->mknode = proxyv4_mknod;
	ops->symlink = proxyv4_symlink;
	ops->readlink = proxyv4_readlink;
	ops->getattrs = proxyv4_getattrs;
	ops->link = proxyv4_link;
	ops->rename = proxyv4_rename;
	ops->unlink = proxyv4_unlink;
	ops->close = proxyv4_close;
	ops->handle_to_wire = proxyv4_handle_to_wire;
	ops->handle_to_key = proxyv4_handle_to_key;
	ops->open2 = proxyv4_open2;
	ops->read2 = proxyv4_read2;
	ops->write2 = proxyv4_write2;
	ops->close2 = proxyv4_close2;
	ops->setattr2 = proxyv4_setattr2;
	ops->status2 = proxyv4_status2;
	ops->reopen2 = proxyv4_reopen2;
	ops->commit2 = proxyv4_commit2;
}

#ifdef PROXYV4_HANDLE_MAPPING
static unsigned int hash_nfs_fh4(const nfs_fh4 *fh, unsigned int cookie)
{
	const char *cpt;
	unsigned int sum = cookie;
	unsigned int extract;
	unsigned int mod = fh->nfs_fh4_len % sizeof(unsigned int);

	for (cpt = fh->nfs_fh4_val;
	     cpt - fh->nfs_fh4_val < fh->nfs_fh4_len - mod;
	     cpt += sizeof(unsigned int)) {
		memcpy(&extract, cpt, sizeof(unsigned int));
		sum = (3 * sum + 5 * extract + 1999);
	}

	/*
	 * If the handle is not 32 bits-aligned, the last loop will
	 * get uninitialized chars after the end of the handle. We
	 * must avoid this by skipping the last loop and doing a
	 * special processing for the last bytes
	 */
	if (mod) {
		extract = 0;
		while (cpt - fh->nfs_fh4_val < fh->nfs_fh4_len) {
			extract <<= 8;
			extract |= (uint8_t) (*cpt++);
		}
		sum = (3 * sum + 5 * extract + 1999);
	}

	return sum;
}
#endif

static struct proxyv4_obj_handle *
proxyv4_alloc_handle(struct fsal_export *exp,
		     const nfs_fh4 *fh,
		     fattr4 *obj_attributes,
		     struct fsal_attrlist *attrs_out)
{
	struct proxyv4_obj_handle *n =
		gsh_calloc(1, sizeof(*n) + fh->nfs_fh4_len);

	compound_data_t data;
	struct fsal_attrlist attributes;

	memset(&attributes, 0, sizeof(attributes));
	memset(&data, 0, sizeof(data));

	data.current_obj = &n->obj;

	if (nfs4_Fattr_To_FSAL_attr(&attributes,
				    obj_attributes,
				    &data) != NFS4_OK) {
		gsh_free(n);
		return NULL;
	}

	n->fh4 = *fh;
	n->fh4.nfs_fh4_val = n->blob.bytes;
	memcpy(n->blob.bytes, fh->nfs_fh4_val, fh->nfs_fh4_len);
	n->blob.len = fh->nfs_fh4_len + sizeof(n->blob);
	n->blob.type = attributes.type;
#ifdef PROXYV4_HANDLE_MAPPING
	int rc;

	memset(&n->h23, 0, sizeof(n->h23));
	n->h23.len = sizeof(n->h23);
	n->h23.type = PROXYV4_HANDLE_MAPPED;
	n->h23.object_id = n->obj.fileid;
	n->h23.handle_hash = hash_nfs_fh4(fh, n->obj.fileid);

	rc = HandleMap_SetFH(&n->h23, &n->blob, n->blob.len);
	if ((rc != HANDLEMAP_SUCCESS) && (rc != HANDLEMAP_EXISTS)) {
		gsh_free(n);
		return NULL;
	}
#endif

	fsal_obj_handle_init(&n->obj, exp, attributes.type);
	n->obj.fs = NULL;
	n->obj.state_hdl = NULL;
	n->obj.fsid = attributes.fsid;
	n->obj.fileid = attributes.fileid;
	n->obj.obj_ops = &PROXY_V4.handle_ops;
	if (attrs_out != NULL) {
		/* We aren't keeping ACL ref ourself, so pass it
		 * to the caller.
		 */
		fsal_copy_attrs(attrs_out, &attributes, true);
	} else {
		/* Make sure we release the attributes. */
		fsal_release_attrs(&attributes);
	}

	return n;
}

/* export methods that create object handles
 */

fsal_status_t proxyv4_lookup_path(struct fsal_export *exp_hdl,
				  const char *path,
				  struct fsal_obj_handle **handle,
				  struct fsal_attrlist *attrs_out)
{
	struct fsal_obj_handle *next;
	struct fsal_obj_handle *parent = NULL;
	char *saved;
	char *pcopy;
	char *p, *pnext;
	struct user_cred *creds = &op_ctx->creds;

	pcopy = gsh_strdup(path);

	p = strtok_r(pcopy, "/", &saved);
	if (!p) {
		fsal_status_t st = proxyv4_lookup_impl(parent, exp_hdl, creds,
						       NULL, &next, attrs_out);
		if (FSAL_IS_ERROR(st)) {
			gsh_free(pcopy);
			return st;
		}
	}
	while (p) {
		if (strcmp(p, "..") == 0) {
			/* Don't allow lookup of ".." */
			LogInfo(COMPONENT_FSAL,
				"Attempt to use \"..\" element in path %s",
				path);
			gsh_free(pcopy);
			return fsalstat(ERR_FSAL_ACCESS, EACCES);
		}
		/* Get the next token now, so we know if we are at the
		 * terminal token or not.
		 */
		pnext = strtok_r(NULL, "/", &saved);

		/* Note that if any element is a symlink, the following will
		 * fail, thus no security exposure. Only pass back the
		 * attributes of the terminal lookup.
		 */
		fsal_status_t st =
			proxyv4_lookup_impl(parent, exp_hdl, creds, p,
					    &next, pnext == NULL ?
					    attrs_out : NULL);
		if (FSAL_IS_ERROR(st)) {
			gsh_free(pcopy);
			return st;
		}

		p = pnext;
		parent = next;
	}
	/* The final element could be a symlink, but either way we are called
	 * will not work with a symlink, so no security exposure there.
	 */

	gsh_free(pcopy);
	*handle = next;
	return fsalstat(ERR_FSAL_NO_ERROR, 0);
}

/*
 * Create an FSAL 'object' from the handle - used
 * to construct objects from a handle which has been
 * 'extracted' by .wire_to_host.
 */
fsal_status_t proxyv4_create_handle(struct fsal_export *exp_hdl,
				    struct gsh_buffdesc *hdl_desc,
				    struct fsal_obj_handle **handle,
				    struct fsal_attrlist *attrs_out)
{
	nfs_fh4 fh4;
	struct proxyv4_obj_handle *ph;
	struct proxyv4_handle_blob *blob;
	int rc;
	uint32_t opcnt = 0;
	sessionid4 sid;
#define FSAL_CREATE_HANDLE_NB_OP_ALLOC 3 /* SEQUENCE PUTFH GETATTR */
	nfs_argop4 argoparray[FSAL_CREATE_HANDLE_NB_OP_ALLOC];
	nfs_resop4 resoparray[FSAL_CREATE_HANDLE_NB_OP_ALLOC];
	GETATTR4resok *atok;
	char fattr_blob[FATTR_BLOB_SZ];

	blob = (struct proxyv4_handle_blob *)hdl_desc->addr;
	if (blob->len != hdl_desc->len)
		return fsalstat(ERR_FSAL_INVAL, 0);

	fh4.nfs_fh4_val = blob->bytes;
	fh4.nfs_fh4_len = blob->len - sizeof(*blob);

	/* SEQUENCE */
	proxyv4_get_client_sessionid(sid);
	COMPOUNDV4_ARG_ADD_OP_SEQUENCE(opcnt, argoparray, sid, NB_RPC_SLOT);
	COMPOUNDV4_ARG_ADD_OP_PUTFH(opcnt, argoparray, fh4);

	atok = proxyv4_fill_getattr_reply(resoparray + opcnt, fattr_blob,
					  sizeof(fattr_blob));
	COMPOUNDV4_ARG_ADD_OP_GETATTR(opcnt, argoparray,
				      proxyv4_bitmap_getattr);

	rc = proxyv4_nfsv4_call(&op_ctx->creds, opcnt, argoparray, resoparray);

	if (rc != NFS4_OK)
		return nfsstat4_to_fsal(rc);

	ph = proxyv4_alloc_handle(exp_hdl, &fh4, &atok->obj_attributes,
				  attrs_out);
	if (!ph)
		return fsalstat(ERR_FSAL_FAULT, 0);

	*handle = &ph->obj;
	return fsalstat(ERR_FSAL_NO_ERROR, 0);
}

fsal_status_t proxyv4_get_dynamic_info(struct fsal_export *exp_hdl,
				       struct fsal_obj_handle *obj_hdl,
				       fsal_dynamicfsinfo_t *infop)
{
	int rc;
	int opcnt = 0;
	sessionid4 sid;
#define FSAL_FSINFO_NB_OP_ALLOC 3 /* SEQUENCE PUTFH GETATTR */
	nfs_argop4 argoparray[FSAL_FSINFO_NB_OP_ALLOC];
	nfs_resop4 resoparray[FSAL_FSINFO_NB_OP_ALLOC];
	GETATTR4resok *atok;
	char fattr_blob[48];	/* 6 values, 8 bytes each */
	struct proxyv4_obj_handle *ph;

	ph = container_of(obj_hdl, struct proxyv4_obj_handle, obj);

	/* SEQUENCE */
	proxyv4_get_client_sessionid(sid);
	COMPOUNDV4_ARG_ADD_OP_SEQUENCE(opcnt, argoparray, sid, NB_RPC_SLOT);
	COMPOUNDV4_ARG_ADD_OP_PUTFH(opcnt, argoparray, ph->fh4);
	atok =
		proxyv4_fill_getattr_reply(resoparray + opcnt, fattr_blob,
					   sizeof(fattr_blob));
	COMPOUNDV4_ARG_ADD_OP_GETATTR(opcnt, argoparray, proxyv4_bitmap_fsinfo);

	rc = proxyv4_nfsv4_call(&op_ctx->creds, opcnt, argoparray, resoparray);
	if (rc != NFS4_OK)
		return nfsstat4_to_fsal(rc);

	if (nfs4_Fattr_To_fsinfo(infop, &atok->obj_attributes) != NFS4_OK)
		return fsalstat(ERR_FSAL_INVAL, 0);

	/* At the moment, we don't actually decode time_delta from the
	 * proxied server, so let's just hardcode.
	 */
	infop->time_delta.tv_sec = 0;
	infop->time_delta.tv_nsec = FSAL_DEFAULT_TIME_DELTA_NSEC;

	return fsalstat(ERR_FSAL_NO_ERROR, 0);
}

/* Convert of-the-wire digest into unique 'handle' which
 * can be used to identify the object */
fsal_status_t proxyv4_wire_to_host(struct fsal_export *exp_hdl,
				   fsal_digesttype_t in_type,
				   struct gsh_buffdesc *fh_desc,
				   int flags)
{
	struct proxyv4_handle_blob *proxyv4_blob;
	size_t fh_size;

	if (!fh_desc || !fh_desc->addr)
		return fsalstat(ERR_FSAL_FAULT, EINVAL);

	proxyv4_blob = (struct proxyv4_handle_blob *)fh_desc->addr;
	fh_size = proxyv4_blob->len;
#ifdef PROXYV4_HANDLE_MAPPING
	if (in_type == FSAL_DIGEST_NFSV3)
		fh_size = sizeof(nfs23_map_handle_t);
#endif
	if (fh_desc->len != fh_size) {
		LogMajor(COMPONENT_FSAL,
			 "Size mismatch for handle.  should be %zu, got %zu",
			 fh_size, fh_desc->len);
		return fsalstat(ERR_FSAL_SERVERFAULT, 0);
	}
#ifdef PROXYV4_HANDLE_MAPPING
	if (in_type == FSAL_DIGEST_NFSV3) {
		nfs23_map_handle_t *h23 = (nfs23_map_handle_t *) fh_desc->addr;

		if (h23->type != PROXYV4_HANDLE_MAPPED)
			return fsalstat(ERR_FSAL_STALE, ESTALE);

		/* As long as HandleMap_GetFH copies nfs23 handle into
		 * the key before lookup I can get away with using
		 * the same buffer for input and output */
		if (HandleMap_GetFH(h23, fh_desc) != HANDLEMAP_SUCCESS)
			return fsalstat(ERR_FSAL_STALE, 0);
		fh_size = fh_desc->len;
	}
#endif

	fh_desc->len = fh_size;
	return fsalstat(ERR_FSAL_NO_ERROR, 0);
}