[AF_RXRPC]: Make the in-kernel AFS filesystem use AF_RXRPC.

Make the in-kernel AFS filesystem use AF_RXRPC instead of the old RxRPC code.

Signed-off-by: David Howells <dhowells@redhat.com>
Signed-off-by: David S. Miller <davem@davemloft.net>

1 files changed, 310 insertions, 732 deletions

diff --git a/fs/afs/fsclient.c b/fs/afs/fsclient.c
index f1c3a18..167ca61 100644
--- a/fs/afs/fsclient.c
+++ b/fs/afs/fsclient.c
@@ -1,6 +1,6 @@
-/* fsclient.c: AFS File Server client stubs
+/* AFS File Server client stubs
  *
- * Copyright (C) 2002 Red Hat, Inc. All Rights Reserved.
+ * Copyright (C) 2002, 2007 Red Hat, Inc. All Rights Reserved.
  * Written by David Howells (dhowells@redhat.com)
  *
  * This program is free software; you can redistribute it and/or
@@ -11,818 +11,396 @@
 
 #include <linux/init.h>
 #include <linux/sched.h>
-#include <rxrpc/rxrpc.h>
-#include <rxrpc/transport.h>
-#include <rxrpc/connection.h>
-#include <rxrpc/call.h>
-#include "fsclient.h"
-#include "cmservice.h"
-#include "vnode.h"
-#include "server.h"
-#include "errors.h"
+#include <linux/circ_buf.h>
 #include "internal.h"
-
-#define FSFETCHSTATUS		132	/* AFS Fetch file status */
-#define FSFETCHDATA		130	/* AFS Fetch file data */
-#define FSGIVEUPCALLBACKS	147	/* AFS Discard callback promises */
-#define FSGETVOLUMEINFO		148	/* AFS Get root volume information */
-#define FSGETROOTVOLUME		151	/* AFS Get root volume name */
-#define FSLOOKUP		161	/* AFS lookup file in directory */
+#include "afs_fs.h"
 
 /*
- * map afs abort codes to/from Linux error codes
- * - called with call->lock held
+ * decode an AFSFetchStatus block
  */
-static void afs_rxfs_aemap(struct rxrpc_call *call)
+static void xdr_decode_AFSFetchStatus(const __be32 **_bp,
+				      struct afs_vnode *vnode)
 {
-	switch (call->app_err_state) {
-	case RXRPC_ESTATE_LOCAL_ABORT:
-		call->app_abort_code = -call->app_errno;
-		break;
-	case RXRPC_ESTATE_PEER_ABORT:
-		call->app_errno = afs_abort_to_error(call->app_abort_code);
-		break;
-	default:
-		break;
+	const __be32 *bp = *_bp;
+	umode_t mode;
+	u64 data_version;
+	u32 changed = 0; /* becomes non-zero if ctime-type changes seen */
+
+#define EXTRACT(DST)				\
+	do {					\
+		u32 x = ntohl(*bp++);		\
+		changed |= DST - x;		\
+		DST = x;			\
+	} while (0)
+
+	vnode->status.if_version = ntohl(*bp++);
+	EXTRACT(vnode->status.type);
+	vnode->status.nlink = ntohl(*bp++);
+	EXTRACT(vnode->status.size);
+	data_version = ntohl(*bp++);
+	EXTRACT(vnode->status.author);
+	EXTRACT(vnode->status.owner);
+	EXTRACT(vnode->status.caller_access); /* call ticket dependent */
+	EXTRACT(vnode->status.anon_access);
+	EXTRACT(vnode->status.mode);
+	vnode->status.parent.vid = vnode->fid.vid;
+	EXTRACT(vnode->status.parent.vnode);
+	EXTRACT(vnode->status.parent.unique);
+	bp++; /* seg size */
+	vnode->status.mtime_client = ntohl(*bp++);
+	vnode->status.mtime_server = ntohl(*bp++);
+	bp++; /* group */
+	bp++; /* sync counter */
+	data_version |= (u64) ntohl(*bp++) << 32;
+	bp++; /* spare2 */
+	bp++; /* spare3 */
+	bp++; /* spare4 */
+	*_bp = bp;
+
+	if (changed) {
+		_debug("vnode changed");
+		set_bit(AFS_VNODE_CHANGED, &vnode->flags);
+		vnode->vfs_inode.i_uid		= vnode->status.owner;
+		vnode->vfs_inode.i_size		= vnode->status.size;
+		vnode->vfs_inode.i_version	= vnode->fid.unique;
+
+		vnode->status.mode &= S_IALLUGO;
+		mode = vnode->vfs_inode.i_mode;
+		mode &= ~S_IALLUGO;
+		mode |= vnode->status.mode;
+		vnode->vfs_inode.i_mode = mode;
+	}
+
+	_debug("vnode time %lx, %lx",
+	       vnode->status.mtime_client, vnode->status.mtime_server);
+	vnode->vfs_inode.i_ctime.tv_sec	= vnode->status.mtime_server;
+	vnode->vfs_inode.i_mtime	= vnode->vfs_inode.i_ctime;
+	vnode->vfs_inode.i_atime	= vnode->vfs_inode.i_ctime;
+
+	if (vnode->status.data_version != data_version) {
+		_debug("vnode modified %llx", data_version);
+		vnode->status.data_version = data_version;
+		set_bit(AFS_VNODE_MODIFIED, &vnode->flags);
+		set_bit(AFS_VNODE_ZAP_DATA, &vnode->flags);
 	}
 }
 
 /*
- * get the root volume name from a fileserver
- * - this operation doesn't seem to work correctly in OpenAFS server 1.2.2
+ * decode an AFSCallBack block
  */
-#if 0
-int afs_rxfs_get_root_volume(struct afs_server *server,
-			     char *buf, size_t *buflen)
+static void xdr_decode_AFSCallBack(const __be32 **_bp, struct afs_vnode *vnode)
 {
-	struct rxrpc_connection *conn;
-	struct rxrpc_call *call;
-	struct kvec piov[2];
-	size_t sent;
-	int ret;
-	u32 param[1];
-
-	DECLARE_WAITQUEUE(myself, current);
-
-	kenter("%p,%p,%u",server, buf, *buflen);
-
-	/* get hold of the fileserver connection */
-	ret = afs_server_get_fsconn(server, &conn);
-	if (ret < 0)
-		goto out;
-
-	/* create a call through that connection */
-	ret = rxrpc_create_call(conn, NULL, NULL, afs_rxfs_aemap, &call);
-	if (ret < 0) {
-		printk("kAFS: Unable to create call: %d\n", ret);
-		goto out_put_conn;
-	}
-	call->app_opcode = FSGETROOTVOLUME;
-
-	/* we want to get event notifications from the call */
-	add_wait_queue(&call->waitq, &myself);
-
-	/* marshall the parameters */
-	param[0] = htonl(FSGETROOTVOLUME);
-
-	piov[0].iov_len = sizeof(param);
-	piov[0].iov_base = param;
-
-	/* send the parameters to the server */
-	ret = rxrpc_call_write_data(call, 1, piov, RXRPC_LAST_PACKET, GFP_NOFS,
-				    0, &sent);
-	if (ret < 0)
-		goto abort;
-
-	/* wait for the reply to completely arrive */
-	for (;;) {
-		set_current_state(TASK_INTERRUPTIBLE);
-		if (call->app_call_state != RXRPC_CSTATE_CLNT_RCV_REPLY ||
-		    signal_pending(current))
-			break;
-		schedule();
-	}
-	set_current_state(TASK_RUNNING);
-
-	ret = -EINTR;
-	if (signal_pending(current))
-		goto abort;
-
-	switch (call->app_call_state) {
-	case RXRPC_CSTATE_ERROR:
-		ret = call->app_errno;
-		kdebug("Got Error: %d", ret);
-		goto out_unwait;
-
-	case RXRPC_CSTATE_CLNT_GOT_REPLY:
-		/* read the reply */
-		kdebug("Got Reply: qty=%d", call->app_ready_qty);
-
-		ret = -EBADMSG;
-		if (call->app_ready_qty <= 4)
-			goto abort;
-
-		ret = rxrpc_call_read_data(call, NULL, call->app_ready_qty, 0);
-		if (ret < 0)
-			goto abort;
-
-#if 0
-		/* unmarshall the reply */
-		bp = buffer;
-		for (loop = 0; loop < 65; loop++)
-			entry->name[loop] = ntohl(*bp++);
-		entry->name[64] = 0;
-
-		entry->type = ntohl(*bp++);
-		entry->num_servers = ntohl(*bp++);
-
-		for (loop = 0; loop < 8; loop++)
-			entry->servers[loop].addr.s_addr = *bp++;
-
-		for (loop = 0; loop < 8; loop++)
-			entry->servers[loop].partition = ntohl(*bp++);
-
-		for (loop = 0; loop < 8; loop++)
-			entry->servers[loop].flags = ntohl(*bp++);
-
-		for (loop = 0; loop < 3; loop++)
-			entry->volume_ids[loop] = ntohl(*bp++);
-
-		entry->clone_id = ntohl(*bp++);
-		entry->flags = ntohl(*bp);
-#endif
+	const __be32 *bp = *_bp;
 
-		/* success */
-		ret = 0;
-		goto out_unwait;
-
-	default:
-		BUG();
-	}
-
-abort:
-	set_current_state(TASK_UNINTERRUPTIBLE);
-	rxrpc_call_abort(call, ret);
-	schedule();
-out_unwait:
-	set_current_state(TASK_RUNNING);
-	remove_wait_queue(&call->waitq, &myself);
-	rxrpc_put_call(call);
-out_put_conn:
-	afs_server_release_fsconn(server, conn);
-out:
-	kleave("");
-	return ret;
+	vnode->cb_version	= ntohl(*bp++);
+	vnode->cb_expiry	= ntohl(*bp++);
+	vnode->cb_type		= ntohl(*bp++);
+	vnode->cb_expires	= vnode->cb_expiry + get_seconds();
+	*_bp = bp;
 }
-#endif
 
 /*
- * get information about a volume
+ * decode an AFSVolSync block
  */
-#if 0
-int afs_rxfs_get_volume_info(struct afs_server *server,
-			     const char *name,
-			     struct afs_volume_info *vinfo)
+static void xdr_decode_AFSVolSync(const __be32 **_bp,
+				  struct afs_volsync *volsync)
 {
-	struct rxrpc_connection *conn;
-	struct rxrpc_call *call;
-	struct kvec piov[3];
-	size_t sent;
-	int ret;
-	u32 param[2], *bp, zero;
+	const __be32 *bp = *_bp;
 
-	DECLARE_WAITQUEUE(myself, current);
+	volsync->creation = ntohl(*bp++);
+	bp++; /* spare2 */
+	bp++; /* spare3 */
+	bp++; /* spare4 */
+	bp++; /* spare5 */
+	bp++; /* spare6 */
+	*_bp = bp;
+}
 
-	_enter("%p,%s,%p", server, name, vinfo);
+/*
+ * deliver reply data to an FS.FetchStatus
+ */
+static int afs_deliver_fs_fetch_status(struct afs_call *call,
+				       struct sk_buff *skb, bool last)
+{
+	const __be32 *bp;
 
-	/* get hold of the fileserver connection */
-	ret = afs_server_get_fsconn(server, &conn);
-	if (ret < 0)
-		goto out;
+	_enter(",,%u", last);
 
-	/* create a call through that connection */
-	ret = rxrpc_create_call(conn, NULL, NULL, afs_rxfs_aemap, &call);
-	if (ret < 0) {
-		printk("kAFS: Unable to create call: %d\n", ret);
-		goto out_put_conn;
-	}
-	call->app_opcode = FSGETVOLUMEINFO;
+	afs_transfer_reply(call, skb);
+	if (!last)
+		return 0;
 
-	/* we want to get event notifications from the call */
-	add_wait_queue(&call->waitq, &myself);
+	if (call->reply_size != call->reply_max)
+		return -EBADMSG;
 
-	/* marshall the parameters */
-	piov[1].iov_len = strlen(name);
-	piov[1].iov_base = (char *) name;
-
-	zero = 0;
-	piov[2].iov_len = (4 - (piov[1].iov_len & 3)) & 3;
-	piov[2].iov_base = &zero;
-
-	param[0] = htonl(FSGETVOLUMEINFO);
-	param[1] = htonl(piov[1].iov_len);
-
-	piov[0].iov_len = sizeof(param);
-	piov[0].iov_base = param;
-
-	/* send the parameters to the server */
-	ret = rxrpc_call_write_data(call, 3, piov, RXRPC_LAST_PACKET, GFP_NOFS,
-				    0, &sent);
-	if (ret < 0)
-		goto abort;
-
-	/* wait for the reply to completely arrive */
-	bp = rxrpc_call_alloc_scratch(call, 64);
-
-	ret = rxrpc_call_read_data(call, bp, 64,
-				   RXRPC_CALL_READ_BLOCK |
-				   RXRPC_CALL_READ_ALL);
-	if (ret < 0) {
-		if (ret == -ECONNABORTED) {
-			ret = call->app_errno;
-			goto out_unwait;
-		}
-		goto abort;
-	}
+	/* unmarshall the reply once we've received all of it */
+	bp = call->buffer;
+	xdr_decode_AFSFetchStatus(&bp, call->reply);
+	xdr_decode_AFSCallBack(&bp, call->reply);
+	if (call->reply2)
+		xdr_decode_AFSVolSync(&bp, call->reply2);
 
-	/* unmarshall the reply */
-	vinfo->vid = ntohl(*bp++);
-	vinfo->type = ntohl(*bp++);
-
-	vinfo->type_vids[0] = ntohl(*bp++);
-	vinfo->type_vids[1] = ntohl(*bp++);
-	vinfo->type_vids[2] = ntohl(*bp++);
-	vinfo->type_vids[3] = ntohl(*bp++);
-	vinfo->type_vids[4] = ntohl(*bp++);
-
-	vinfo->nservers = ntohl(*bp++);
-	vinfo->servers[0].addr.s_addr = *bp++;
-	vinfo->servers[1].addr.s_addr = *bp++;
-	vinfo->servers[2].addr.s_addr = *bp++;
-	vinfo->servers[3].addr.s_addr = *bp++;
-	vinfo->servers[4].addr.s_addr = *bp++;
-	vinfo->servers[5].addr.s_addr = *bp++;
-	vinfo->servers[6].addr.s_addr = *bp++;
-	vinfo->servers[7].addr.s_addr = *bp++;
-
-	ret = -EBADMSG;
-	if (vinfo->nservers > 8)
-		goto abort;
-
-	/* success */
-	ret = 0;
-
-out_unwait:
-	set_current_state(TASK_RUNNING);
-	remove_wait_queue(&call->waitq, &myself);
-	rxrpc_put_call(call);
-out_put_conn:
-	afs_server_release_fsconn(server, conn);
-out:
-	_leave("");
-	return ret;
-
-abort:
-	set_current_state(TASK_UNINTERRUPTIBLE);
-	rxrpc_call_abort(call, ret);
-	schedule();
-	goto out_unwait;
+	_leave(" = 0 [done]");
+	return 0;
 }
-#endif
+
+/*
+ * FS.FetchStatus operation type
+ */
+static const struct afs_call_type afs_RXFSFetchStatus = {
+	.deliver	= afs_deliver_fs_fetch_status,
+	.abort_to_error	= afs_abort_to_error,
+	.destructor	= afs_flat_call_destructor,
+};
 
 /*
  * fetch the status information for a file
  */
-int afs_rxfs_fetch_file_status(struct afs_server *server,
-			       struct afs_vnode *vnode,
-			       struct afs_volsync *volsync)
+int afs_fs_fetch_file_status(struct afs_server *server,
+			     struct afs_vnode *vnode,
+			     struct afs_volsync *volsync,
+			     const struct afs_wait_mode *wait_mode)
 {
-	struct afs_server_callslot callslot;
-	struct rxrpc_call *call;
-	struct kvec piov[1];
-	size_t sent;
-	int ret;
+	struct afs_call *call;
 	__be32 *bp;
 
-	DECLARE_WAITQUEUE(myself, current);
+	_enter("");
 
-	_enter("%p,{%u,%u,%u}",
-	       server, vnode->fid.vid, vnode->fid.vnode, vnode->fid.unique);
+	call = afs_alloc_flat_call(&afs_RXFSFetchStatus, 16, 120);
+	if (!call)
+		return -ENOMEM;
 
-	/* get hold of the fileserver connection */
-	ret = afs_server_request_callslot(server, &callslot);
-	if (ret < 0)
-		goto out;
-
-	/* create a call through that connection */
-	ret = rxrpc_create_call(callslot.conn, NULL, NULL, afs_rxfs_aemap,
-				&call);
-	if (ret < 0) {
-		printk("kAFS: Unable to create call: %d\n", ret);
-		goto out_put_conn;
-	}
-	call->app_opcode = FSFETCHSTATUS;
-
-	/* we want to get event notifications from the call */
-	add_wait_queue(&call->waitq, &myself);
+	call->reply = vnode;
+	call->reply2 = volsync;
+	call->service_id = FS_SERVICE;
+	call->port = htons(AFS_FS_PORT);
 
 	/* marshall the parameters */
-	bp = rxrpc_call_alloc_scratch(call, 16);
+	bp = call->request;
 	bp[0] = htonl(FSFETCHSTATUS);
 	bp[1] = htonl(vnode->fid.vid);
 	bp[2] = htonl(vnode->fid.vnode);
 	bp[3] = htonl(vnode->fid.unique);
 
-	piov[0].iov_len = 16;
-	piov[0].iov_base = bp;
-
-	/* send the parameters to the server */
-	ret = rxrpc_call_write_data(call, 1, piov, RXRPC_LAST_PACKET, GFP_NOFS,
-				    0, &sent);
-	if (ret < 0)
-		goto abort;
-
-	/* wait for the reply to completely arrive */
-	bp = rxrpc_call_alloc_scratch(call, 120);
-
-	ret = rxrpc_call_read_data(call, bp, 120,
-				   RXRPC_CALL_READ_BLOCK |
-				   RXRPC_CALL_READ_ALL);
-	if (ret < 0) {
-		if (ret == -ECONNABORTED) {
-			ret = call->app_errno;
-			goto out_unwait;
-		}
-		goto abort;
-	}
-
-	/* unmarshall the reply */
-	vnode->status.if_version	= ntohl(*bp++);
-	vnode->status.type		= ntohl(*bp++);
-	vnode->status.nlink		= ntohl(*bp++);
-	vnode->status.size		= ntohl(*bp++);
-	vnode->status.version		= ntohl(*bp++);
-	vnode->status.author		= ntohl(*bp++);
-	vnode->status.owner		= ntohl(*bp++);
-	vnode->status.caller_access	= ntohl(*bp++);
-	vnode->status.anon_access	= ntohl(*bp++);
-	vnode->status.mode		= ntohl(*bp++);
-	vnode->status.parent.vid	= vnode->fid.vid;
-	vnode->status.parent.vnode	= ntohl(*bp++);
-	vnode->status.parent.unique	= ntohl(*bp++);
-	bp++; /* seg size */
-	vnode->status.mtime_client	= ntohl(*bp++);
-	vnode->status.mtime_server	= ntohl(*bp++);
-	bp++; /* group */
-	bp++; /* sync counter */
-	vnode->status.version |= ((unsigned long long) ntohl(*bp++)) << 32;
-	bp++; /* spare2 */
-	bp++; /* spare3 */
-	bp++; /* spare4 */
-
-	vnode->cb_version		= ntohl(*bp++);
-	vnode->cb_expiry		= ntohl(*bp++);
-	vnode->cb_type			= ntohl(*bp++);
-
-	if (volsync) {
-		volsync->creation	= ntohl(*bp++);
-		bp++; /* spare2 */
-		bp++; /* spare3 */
-		bp++; /* spare4 */
-		bp++; /* spare5 */
-		bp++; /* spare6 */
-	}
-
-	/* success */
-	ret = 0;
-
-out_unwait:
-	set_current_state(TASK_RUNNING);
-	remove_wait_queue(&call->waitq, &myself);
-	rxrpc_put_call(call);
-out_put_conn:
-	afs_server_release_callslot(server, &callslot);
-out:
-	_leave("");
-	return ret;
-
-abort:
-	set_current_state(TASK_UNINTERRUPTIBLE);
-	rxrpc_call_abort(call, ret);
-	schedule();
-	goto out_unwait;
+	return afs_make_call(&server->addr, call, GFP_NOFS, wait_mode);
 }
 
 /*
- * fetch the contents of a file or directory
+ * deliver reply data to an FS.FetchData
  */
-int afs_rxfs_fetch_file_data(struct afs_server *server,
-			     struct afs_vnode *vnode,
-			     struct afs_rxfs_fetch_descriptor *desc,
-			     struct afs_volsync *volsync)
+static int afs_deliver_fs_fetch_data(struct afs_call *call,
+				     struct sk_buff *skb, bool last)
 {
-	struct afs_server_callslot callslot;
-	struct rxrpc_call *call;
-	struct kvec piov[1];
-	size_t sent;
+	const __be32 *bp;
+	struct page *page;
+	void *buffer;
 	int ret;
-	__be32 *bp;
 
-	DECLARE_WAITQUEUE(myself, current);
-
-	_enter("%p,{fid={%u,%u,%u},sz=%Zu,of=%lu}",
-	       server,
-	       desc->fid.vid,
-	       desc->fid.vnode,
-	       desc->fid.unique,
-	       desc->size,
-	       desc->offset);
-
-	/* get hold of the fileserver connection */
-	ret = afs_server_request_callslot(server, &callslot);
-	if (ret < 0)
-		goto out;
-
-	/* create a call through that connection */
-	ret = rxrpc_create_call(callslot.conn, NULL, NULL, afs_rxfs_aemap, &call);
-	if (ret < 0) {
-		printk("kAFS: Unable to create call: %d\n", ret);
-		goto out_put_conn;
-	}
-	call->app_opcode = FSFETCHDATA;
+	_enter("{%u},{%u},%d", call->unmarshall, skb->len, last);
+
+	switch (call->unmarshall) {
+	case 0:
+		call->offset = 0;
+		call->unmarshall++;
+
+		/* extract the returned data length */
+	case 1:
+		_debug("extract data length");
+		ret = afs_extract_data(call, skb, last, &call->tmp, 4);
+		switch (ret) {
+		case 0:		break;
+		case -EAGAIN:	return 0;
+		default:	return ret;
+		}
 
-	/* we want to get event notifications from the call */
-	add_wait_queue(&call->waitq, &myself);
+		call->count = ntohl(call->tmp);
+		_debug("DATA length: %u", call->count);
+		if (call->count > PAGE_SIZE)
+			return -EBADMSG;
+		call->offset = 0;
+		call->unmarshall++;
+
+		if (call->count < PAGE_SIZE) {
+			buffer = kmap_atomic(call->reply3, KM_USER0);
+			memset(buffer + PAGE_SIZE - call->count, 0,
+			       call->count);
+			kunmap_atomic(buffer, KM_USER0);
+		}
 
-	/* marshall the parameters */
-	bp = rxrpc_call_alloc_scratch(call, 24);
-	bp[0] = htonl(FSFETCHDATA);
-	bp[1] = htonl(desc->fid.vid);
-	bp[2] = htonl(desc->fid.vnode);
-	bp[3] = htonl(desc->fid.unique);
-	bp[4] = htonl(desc->offset);
-	bp[5] = htonl(desc->size);
-
-	piov[0].iov_len = 24;
-	piov[0].iov_base = bp;
-
-	/* send the parameters to the server */
-	ret = rxrpc_call_write_data(call, 1, piov, RXRPC_LAST_PACKET, GFP_NOFS,
-				    0, &sent);
-	if (ret < 0)
-		goto abort;
-
-	/* wait for the data count to arrive */
-	ret = rxrpc_call_read_data(call, bp, 4, RXRPC_CALL_READ_BLOCK);
-	if (ret < 0)
-		goto read_failed;
-
-	desc->actual = ntohl(bp[0]);
-	if (desc->actual != desc->size) {
-		ret = -EBADMSG;
-		goto abort;
-	}
+		/* extract the returned data */
+	case 2:
+		_debug("extract data");
+		page = call->reply3;
+		buffer = kmap_atomic(page, KM_USER0);
+		ret = afs_extract_data(call, skb, last, buffer, call->count);
+		kunmap_atomic(buffer, KM_USER0);
+		switch (ret) {
+		case 0:		break;
+		case -EAGAIN:	return 0;
+		default:	return ret;
+		}
 
-	/* call the app to read the actual data */
-	rxrpc_call_reset_scratch(call);
-
-	ret = rxrpc_call_read_data(call, desc->buffer, desc->actual,
-				   RXRPC_CALL_READ_BLOCK);
-	if (ret < 0)
-		goto read_failed;
-
-	/* wait for the rest of the reply to completely arrive */
-	rxrpc_call_reset_scratch(call);
-	bp = rxrpc_call_alloc_scratch(call, 120);
-
-	ret = rxrpc_call_read_data(call, bp, 120,
-				   RXRPC_CALL_READ_BLOCK |
-				   RXRPC_CALL_READ_ALL);
-	if (ret < 0)
-		goto read_failed;
-
-	/* unmarshall the reply */
-	vnode->status.if_version	= ntohl(*bp++);
-	vnode->status.type		= ntohl(*bp++);
-	vnode->status.nlink		= ntohl(*bp++);
-	vnode->status.size		= ntohl(*bp++);
-	vnode->status.version		= ntohl(*bp++);
-	vnode->status.author		= ntohl(*bp++);
-	vnode->status.owner		= ntohl(*bp++);
-	vnode->status.caller_access	= ntohl(*bp++);
-	vnode->status.anon_access	= ntohl(*bp++);
-	vnode->status.mode		= ntohl(*bp++);
-	vnode->status.parent.vid	= desc->fid.vid;
-	vnode->status.parent.vnode	= ntohl(*bp++);
-	vnode->status.parent.unique	= ntohl(*bp++);
-	bp++; /* seg size */
-	vnode->status.mtime_client	= ntohl(*bp++);
-	vnode->status.mtime_server	= ntohl(*bp++);
-	bp++; /* group */
-	bp++; /* sync counter */
-	vnode->status.version |= ((unsigned long long) ntohl(*bp++)) << 32;
-	bp++; /* spare2 */
-	bp++; /* spare3 */
-	bp++; /* spare4 */
+		call->offset = 0;
+		call->unmarshall++;
 
-	vnode->cb_version		= ntohl(*bp++);
-	vnode->cb_expiry		= ntohl(*bp++);
-	vnode->cb_type			= ntohl(*bp++);
-
-	if (volsync) {
-		volsync->creation	= ntohl(*bp++);
-		bp++; /* spare2 */
-		bp++; /* spare3 */
-		bp++; /* spare4 */
-		bp++; /* spare5 */
-		bp++; /* spare6 */
-	}
+		/* extract the metadata */
+	case 3:
+		ret = afs_extract_data(call, skb, last, call->buffer, 120);
+		switch (ret) {
+		case 0:		break;
+		case -EAGAIN:	return 0;
+		default:	return ret;
+		}
+
+		bp = call->buffer;
+		xdr_decode_AFSFetchStatus(&bp, call->reply);
+		xdr_decode_AFSCallBack(&bp, call->reply);
+		if (call->reply2)
+			xdr_decode_AFSVolSync(&bp, call->reply2);
+
+		call->offset = 0;
+		call->unmarshall++;
 
-	/* success */
-	ret = 0;
-
-out_unwait:
-	set_current_state(TASK_RUNNING);
-	remove_wait_queue(&call->waitq,&myself);
-	rxrpc_put_call(call);
-out_put_conn:
-	afs_server_release_callslot(server, &callslot);
-out:
-	_leave(" = %d", ret);
-	return ret;
-
-read_failed:
-	if (ret == -ECONNABORTED) {
-		ret = call->app_errno;
-		goto out_unwait;
+	case 4:
+		_debug("trailer");
+		if (skb->len != 0)
+			return -EBADMSG;
+		break;
 	}
 
-abort:
-	set_current_state(TASK_UNINTERRUPTIBLE);
-	rxrpc_call_abort(call, ret);
-	schedule();
-	goto out_unwait;
+	if (!last)
+		return 0;
+
+	_leave(" = 0 [done]");
+	return 0;
 }
 
 /*
- * ask the AFS fileserver to discard a callback request on a file
+ * FS.FetchData operation type
  */
-int afs_rxfs_give_up_callback(struct afs_server *server,
-			      struct afs_vnode *vnode)
+static const struct afs_call_type afs_RXFSFetchData = {
+	.deliver	= afs_deliver_fs_fetch_data,
+	.abort_to_error	= afs_abort_to_error,
+	.destructor	= afs_flat_call_destructor,
+};
+
+/*
+ * fetch data from a file
+ */
+int afs_fs_fetch_data(struct afs_server *server,
+		      struct afs_vnode *vnode,
+		      off_t offset, size_t length,
+		      struct page *buffer,
+		      struct afs_volsync *volsync,
+		      const struct afs_wait_mode *wait_mode)
 {
-	struct afs_server_callslot callslot;
-	struct rxrpc_call *call;
-	struct kvec piov[1];
-	size_t sent;
-	int ret;
+	struct afs_call *call;
 	__be32 *bp;
 
-	DECLARE_WAITQUEUE(myself, current);
+	_enter("");
 
-	_enter("%p,{%u,%u,%u}",
-	       server, vnode->fid.vid, vnode->fid.vnode, vnode->fid.unique);
+	call = afs_alloc_flat_call(&afs_RXFSFetchData, 24, 120);
+	if (!call)
+		return -ENOMEM;
 
-	/* get hold of the fileserver connection */
-	ret = afs_server_request_callslot(server, &callslot);
-	if (ret < 0)
-		goto out;
-
-	/* create a call through that connection */
-	ret = rxrpc_create_call(callslot.conn, NULL, NULL, afs_rxfs_aemap, &call);
-	if (ret < 0) {
-		printk("kAFS: Unable to create call: %d\n", ret);
-		goto out_put_conn;
-	}
-	call->app_opcode = FSGIVEUPCALLBACKS;
-
-	/* we want to get event notifications from the call */
-	add_wait_queue(&call->waitq, &myself);
+	call->reply = vnode;
+	call->reply2 = volsync;
+	call->reply3 = buffer;
+	call->service_id = FS_SERVICE;
+	call->port = htons(AFS_FS_PORT);
 
 	/* marshall the parameters */
-	bp = rxrpc_call_alloc_scratch(call, (1 + 4 + 4) * 4);
-
-	piov[0].iov_len = (1 + 4 + 4) * 4;
-	piov[0].iov_base = bp;
-
-	*bp++ = htonl(FSGIVEUPCALLBACKS);
-	*bp++ = htonl(1);
-	*bp++ = htonl(vnode->fid.vid);
-	*bp++ = htonl(vnode->fid.vnode);
-	*bp++ = htonl(vnode->fid.unique);
-	*bp++ = htonl(1);
-	*bp++ = htonl(vnode->cb_version);
-	*bp++ = htonl(vnode->cb_expiry);
-	*bp++ = htonl(vnode->cb_type);
-
-	/* send the parameters to the server */
-	ret = rxrpc_call_write_data(call, 1, piov, RXRPC_LAST_PACKET, GFP_NOFS,
-				    0, &sent);
-	if (ret < 0)
-		goto abort;
-
-	/* wait for the reply to completely arrive */
-	for (;;) {
-		set_current_state(TASK_INTERRUPTIBLE);
-		if (call->app_call_state != RXRPC_CSTATE_CLNT_RCV_REPLY ||
-		    signal_pending(current))
-			break;
-		schedule();
-	}
-	set_current_state(TASK_RUNNING);
-
-	ret = -EINTR;
-	if (signal_pending(current))
-		goto abort;
-
-	switch (call->app_call_state) {
-	case RXRPC_CSTATE_ERROR:
-		ret = call->app_errno;
-		goto out_unwait;
+	bp = call->request;
+	bp[0] = htonl(FSFETCHDATA);
+	bp[1] = htonl(vnode->fid.vid);
+	bp[2] = htonl(vnode->fid.vnode);
+	bp[3] = htonl(vnode->fid.unique);
+	bp[4] = htonl(offset);
+	bp[5] = htonl(length);
 
-	case RXRPC_CSTATE_CLNT_GOT_REPLY:
-		ret = 0;
-		goto out_unwait;
+	return afs_make_call(&server->addr, call, GFP_NOFS, wait_mode);
+}
 
-	default:
-		BUG();
-	}
+/*
+ * deliver reply data to an FS.GiveUpCallBacks
+ */
+static int afs_deliver_fs_give_up_callbacks(struct afs_call *call,
+					    struct sk_buff *skb, bool last)
+{
+	_enter(",{%u},%d", skb->len, last);
 
-out_unwait:
-	set_current_state(TASK_RUNNING);
-	remove_wait_queue(&call->waitq, &myself);
-	rxrpc_put_call(call);
-out_put_conn:
-	afs_server_release_callslot(server, &callslot);
-out:
-	_leave("");
-	return ret;
-
-abort:
-	set_current_state(TASK_UNINTERRUPTIBLE);
-	rxrpc_call_abort(call, ret);
-	schedule();
-	goto out_unwait;
+	if (skb->len > 0)
+		return -EBADMSG; /* shouldn't be any reply data */
+	return 0;
 }
 
 /*
- * look a filename up in a directory
- * - this operation doesn't seem to work correctly in OpenAFS server 1.2.2
+ * FS.GiveUpCallBacks operation type
+ */
+static const struct afs_call_type afs_RXFSGiveUpCallBacks = {
+	.deliver	= afs_deliver_fs_give_up_callbacks,
+	.abort_to_error	= afs_abort_to_error,
+	.destructor	= afs_flat_call_destructor,
+};
+
+/*
+ * give up a set of callbacks
+ * - the callbacks are held in the server->cb_break ring
  */
-#if 0
-int afs_rxfs_lookup(struct afs_server *server,
-		    struct afs_vnode *dir,
-		    const char *filename,
-		    struct afs_vnode *vnode,
-		    struct afs_volsync *volsync)
+int afs_fs_give_up_callbacks(struct afs_server *server,
+			     const struct afs_wait_mode *wait_mode)
 {
-	struct rxrpc_connection *conn;
-	struct rxrpc_call *call;
-	struct kvec piov[3];
-	size_t sent;
-	int ret;
-	u32 *bp, zero;
+	struct afs_call *call;
+	size_t ncallbacks;
+	__be32 *bp, *tp;
+	int loop;
 
-	DECLARE_WAITQUEUE(myself, current);
+	ncallbacks = CIRC_CNT(server->cb_break_head, server->cb_break_tail,
+			      ARRAY_SIZE(server->cb_break));
 
-	kenter("%p,{%u,%u,%u},%s",
-	       server, fid->vid, fid->vnode, fid->unique, filename);
+	_enter("{%zu},", ncallbacks);
 
-	/* get hold of the fileserver connection */
-	ret = afs_server_get_fsconn(server, &conn);
-	if (ret < 0)
-		goto out;
+	if (ncallbacks == 0)
+		return 0;
+	if (ncallbacks > AFSCBMAX)
+		ncallbacks = AFSCBMAX;
 
-	/* create a call through that connection */
-	ret = rxrpc_create_call(conn, NULL, NULL, afs_rxfs_aemap, &call);
-	if (ret < 0) {
-		printk("kAFS: Unable to create call: %d\n", ret);
-		goto out_put_conn;
-	}
-	call->app_opcode = FSLOOKUP;
+	_debug("break %zu callbacks", ncallbacks);
 
-	/* we want to get event notifications from the call */
-	add_wait_queue(&call->waitq,&myself);
+	call = afs_alloc_flat_call(&afs_RXFSGiveUpCallBacks,
+				   12 + ncallbacks * 6 * 4, 0);
+	if (!call)
+		return -ENOMEM;
+
+	call->service_id = FS_SERVICE;
+	call->port = htons(AFS_FS_PORT);
 
 	/* marshall the parameters */
-	bp = rxrpc_call_alloc_scratch(call, 20);
-
-	zero = 0;
-
-	piov[0].iov_len = 20;
-	piov[0].iov_base = bp;
-	piov[1].iov_len = strlen(filename);
-	piov[1].iov_base = (char *) filename;
-	piov[2].iov_len = (4 - (piov[1].iov_len & 3)) & 3;
-	piov[2].iov_base = &zero;
-
-	*bp++ = htonl(FSLOOKUP);
-	*bp++ = htonl(dirfid->vid);
-	*bp++ = htonl(dirfid->vnode);
-	*bp++ = htonl(dirfid->unique);
-	*bp++ = htonl(piov[1].iov_len);
-
-	/* send the parameters to the server */
-	ret = rxrpc_call_write_data(call, 3, piov, RXRPC_LAST_PACKET, GFP_NOFS,
-				    0, &sent);
-	if (ret < 0)
-		goto abort;
-
-	/* wait for the reply to completely arrive */
-	bp = rxrpc_call_alloc_scratch(call, 220);
-
-	ret = rxrpc_call_read_data(call, bp, 220,
-				   RXRPC_CALL_READ_BLOCK |
-				   RXRPC_CALL_READ_ALL);
-	if (ret < 0) {
-		if (ret == -ECONNABORTED) {
-			ret = call->app_errno;
-			goto out_unwait;
-		}
-		goto abort;
+	bp = call->request;
+	tp = bp + 2 + ncallbacks * 3;
+	*bp++ = htonl(FSGIVEUPCALLBACKS);
+	*bp++ = htonl(ncallbacks);
+	*tp++ = htonl(ncallbacks);
+
+	atomic_sub(ncallbacks, &server->cb_break_n);
+	for (loop = ncallbacks; loop > 0; loop--) {
+		struct afs_callback *cb =
+			&server->cb_break[server->cb_break_tail];
+
+		*bp++ = htonl(cb->fid.vid);
+		*bp++ = htonl(cb->fid.vnode);
+		*bp++ = htonl(cb->fid.unique);
+		*tp++ = htonl(cb->version);
+		*tp++ = htonl(cb->expiry);
+		*tp++ = htonl(cb->type);
+		smp_mb();
+		server->cb_break_tail =
+			(server->cb_break_tail + 1) &
+			(ARRAY_SIZE(server->cb_break) - 1);
 	}
 
-	/* unmarshall the reply */
-	fid->vid		= ntohl(*bp++);
-	fid->vnode		= ntohl(*bp++);
-	fid->unique		= ntohl(*bp++);
-
-	vnode->status.if_version	= ntohl(*bp++);
-	vnode->status.type		= ntohl(*bp++);
-	vnode->status.nlink		= ntohl(*bp++);
-	vnode->status.size		= ntohl(*bp++);
-	vnode->status.version		= ntohl(*bp++);
-	vnode->status.author		= ntohl(*bp++);
-	vnode->status.owner		= ntohl(*bp++);
-	vnode->status.caller_access	= ntohl(*bp++);
-	vnode->status.anon_access	= ntohl(*bp++);
-	vnode->status.mode		= ntohl(*bp++);
-	vnode->status.parent.vid	= dirfid->vid;
-	vnode->status.parent.vnode	= ntohl(*bp++);
-	vnode->status.parent.unique	= ntohl(*bp++);
-	bp++; /* seg size */
-	vnode->status.mtime_client	= ntohl(*bp++);
-	vnode->status.mtime_server	= ntohl(*bp++);
-	bp++; /* group */
-	bp++; /* sync counter */
-	vnode->status.version |= ((unsigned long long) ntohl(*bp++)) << 32;
-	bp++; /* spare2 */
-	bp++; /* spare3 */
-	bp++; /* spare4 */
-
-	dir->status.if_version		= ntohl(*bp++);
-	dir->status.type		= ntohl(*bp++);
-	dir->status.nlink		= ntohl(*bp++);
-	dir->status.size		= ntohl(*bp++);
-	dir->status.version		= ntohl(*bp++);
-	dir->status.author		= ntohl(*bp++);
-	dir->status.owner		= ntohl(*bp++);
-	dir->status.caller_access	= ntohl(*bp++);
-	dir->status.anon_access		= ntohl(*bp++);
-	dir->status.mode		= ntohl(*bp++);
-	dir->status.parent.vid		= dirfid->vid;
-	dir->status.parent.vnode	= ntohl(*bp++);
-	dir->status.parent.unique	= ntohl(*bp++);
-	bp++; /* seg size */
-	dir->status.mtime_client	= ntohl(*bp++);
-	dir->status.mtime_server	= ntohl(*bp++);
-	bp++; /* group */
-	bp++; /* sync counter */
-	dir->status.version |= ((unsigned long long) ntohl(*bp++)) << 32;
-	bp++; /* spare2 */
-	bp++; /* spare3 */
-	bp++; /* spare4 */
-
-	callback->fid		= *fid;
-	callback->version	= ntohl(*bp++);
-	callback->expiry	= ntohl(*bp++);
-	callback->type		= ntohl(*bp++);
-
-	if (volsync) {
-		volsync->creation	= ntohl(*bp++);
-		bp++; /* spare2 */
-		bp++; /* spare3 */
-		bp++; /* spare4 */
-		bp++; /* spare5 */
-		bp++; /* spare6 */
-	}
+	ASSERT(ncallbacks > 0);
+	wake_up_nr(&server->cb_break_waitq, ncallbacks);
 
-	/* success */
-	ret = 0;
-
-out_unwait:
-	set_current_state(TASK_RUNNING);
-	remove_wait_queue(&call->waitq, &myself);
-	rxrpc_put_call(call);
-out_put_conn:
-	afs_server_release_fsconn(server, conn);
-out:
-	kleave("");
-	return ret;
-
-abort:
-	set_current_state(TASK_UNINTERRUPTIBLE);
-	rxrpc_call_abort(call, ret);
-	schedule();
-	goto out_unwait;
+	return afs_make_call(&server->addr, call, GFP_NOFS, wait_mode);
 }
-#endif