/*
 *  linux/fs/nfs/rpcsock.c
 *
 *  This is a generic RPC call interface for datagram sockets that is able
 *  to place several concurrent RPC requests at the same time. It works like
 *  this:
 *
 *  -	When a process places a call, it allocates a request slot if
 *	one is available. Otherwise, it sleeps on the backlog queue
 *	(rpc_reserve).
 *  -	Then, the message is transmitted via rpc_send (exported by name of
 *	rpc_transmit).
 *  -	Finally, the process waits for the call to complete (rpc_doio):
 *	The first process on the receive queue waits for the next RPC packet,
 *	and peeks at the XID. If it finds a matching request, it receives
 *	the datagram on behalf of that process and wakes it up. Otherwise,
 *	the datagram is discarded.
 *  -	If the process having received the datagram was the first one on
 *	the receive queue, it wakes up the next one to listen for replies.
 *  -	It then removes itself from the request queue (rpc_release).
 *	If there are more callers waiting on the backlog queue, they are
 *	woken up, too.
 *
 * Mar 1996:
 *  -	Split up large functions into smaller chunks as per Linus' coding
 *	style. Found an interesting bug this way, too.
 *  -	Added entry points for nfsiod.
 *
 *  Copyright (C) 1995, 1996, Olaf Kirch <okir@monad.swb.de>
 */

#include <linux/types.h>
#include <linux/malloc.h>
#include <linux/sched.h>
#include <linux/nfs_fs.h>
#include <linux/errno.h>
#include <linux/socket.h>
#include <linux/fcntl.h>
#include <linux/in.h>
#include <linux/net.h>
#include <linux/mm.h>
#include <linux/rpcsock.h>
#include <linux/poll.h>

#include <linux/udp.h>
#include <net/sock.h>

#include <asm/uaccess.h>

#define msleep(sec)	{ current->timeout = sec * HZ / 1000; \
			  current->state = TASK_INTERRUPTIBLE; \
			  schedule(); \
			}

#undef DEBUG_RPC
#ifdef DEBUG_RPC			
#define dprintk(args...)	printk(## args)
#else
#define	dprintk(args...)
#endif


/*
 * Insert new request into wait list. We make sure list is sorted by
 * increasing timeout value.
 */
static inline void
rpc_insque(struct rpc_sock *rsock, struct rpc_wait *slot)
{
	struct rpc_wait	*next = rsock->pending;

	slot->w_next = next;
	slot->w_prev = NULL;
	if (next)
		next->w_prev = slot;
	rsock->pending = slot;
	slot->w_queued = 1;

	dprintk("RPC: inserted %p into queue\n", slot);
}

/*
 * Remove request from request queue
 */
static inline void
rpc_remque(struct rpc_sock *rsock, struct rpc_wait *slot)
{
	struct rpc_wait	*prev = slot->w_prev,
			*next = slot->w_next;

	if (prev != NULL)
		prev->w_next = next;
	else
		rsock->pending = next;
	if (next != NULL)
		next->w_prev = prev;

	slot->w_queued = 0;
	dprintk("RPC: removed %p from queue, head now %p.\n",
			slot, rsock->pending);
}

/*
 * Write data to socket.
 */
static inline int
rpc_sendmsg(struct rpc_sock *rsock, struct iovec *iov, int nr, int len,
				struct sockaddr *sap, int salen)
{
	struct socket	*sock = rsock->sock;
	struct msghdr	msg;
	unsigned long	oldfs;
	int		result;

	msg.msg_iov	= iov;
	msg.msg_iovlen	= nr;
	msg.msg_name	= sap;
	msg.msg_namelen = salen;
	msg.msg_control = NULL;
	msg.msg_controllen = 0;
	msg.msg_flags = 0;

	oldfs = get_fs();
	set_fs(get_ds());
	result = sock_sendmsg(sock, &msg, len);
	set_fs(oldfs);

	dprintk("RPC: rpc_sendmsg(iov %p, len %d) = %d\n", iov, len, result);
	return result;
}
/*
 * Read data from socket
 */
static inline int
rpc_recvmsg(struct rpc_sock *rsock, struct iovec *iov,
			int nr, int len, int flags)
{
	struct socket	*sock = rsock->sock;
	struct sockaddr	sa;
	struct msghdr	msg;
	unsigned long	oldfs;
	int		result;

	msg.msg_iov	= iov;
	msg.msg_iovlen	= nr;
	msg.msg_name	= &sa;
	msg.msg_namelen = sizeof(sa);
	msg.msg_control = NULL;
	msg.msg_controllen = 0;
	msg.msg_flags = 0;

	oldfs = get_fs();
	set_fs(get_ds());
	result = sock_recvmsg(sock, &msg, len, flags|MSG_DONTWAIT);
	set_fs(oldfs);

	dprintk("RPC: rpc_recvmsg(iov %p, len %d) = %d\n", iov, len, result);
	return result;
}

/*
 * This code is slightly complicated. Since the networking code does not
 * honor the current->timeout value, we have to poll on the socket.
 */
static inline int
rpc_select(struct rpc_sock *rsock)
{
	struct poll_table_entry entry;
	struct file	*file = rsock->file;
	poll_table	wait_table;

	dprintk("RPC: polling socket...\n");
	wait_table.nr = 0;
	wait_table.entry = &entry;
	current->state = TASK_INTERRUPTIBLE;
	if (!(file->f_op->poll(file, &wait_table) & POLLIN)) {
		schedule();
		remove_wait_queue(entry.wait_address, &entry.wait);
		current->state = TASK_RUNNING;
		if (current->signal & ~current->blocked)
			return -ERESTARTSYS;
		if (current->timeout == 0)
			return -ETIMEDOUT;
	} else if (wait_table.nr)
		remove_wait_queue(entry.wait_address, &entry.wait);
	current->state = TASK_RUNNING;
	dprintk("RPC: ...Okay, there appears to be some data.\n");
	return 0;
}

/*
 * Reserve an RPC call slot. nocwait determines whether we wait in case
 * of congestion or not.
 */
int
rpc_reserve(struct rpc_sock *rsock, struct rpc_ioreq *req, int nocwait)
{
	struct rpc_wait	*slot;

	req->rq_slot = NULL;

	while (!(slot = rsock->free) || rsock->cong >= rsock->cwnd) {
		if (nocwait) {
			current->timeout = 0;
			return -ENOBUFS;
		}
		dprintk("RPC: rpc_reserve waiting on backlog\n");
		interruptible_sleep_on(&rsock->backlog);
		if (current->timeout == 0)
			return -ETIMEDOUT;
		if (current->signal & ~current->blocked)
			return -ERESTARTSYS;
		if (rsock->shutdown)
			return -EIO;
	}

	rsock->free = slot->w_next;
	rsock->cong += RPC_CWNDSCALE;	/* bump congestion value */

	slot->w_queued = 0;
	slot->w_gotit = 0;
	slot->w_req = req;

	dprintk("RPC: reserved slot %p\n", slot);
	req->rq_slot = slot;
	return 0;
}

/*
 * Release an RPC call slot
 */
void
rpc_release(struct rpc_sock *rsock, struct rpc_ioreq *req)
{
	struct rpc_wait	*slot = req->rq_slot;

	if (slot != NULL) {
		dprintk("RPC: release slot %p\n", slot);

		/* Wake up the next receiver */
		if (slot == rsock->pending && slot->w_next != NULL)
			wake_up(&slot->w_next->w_wait);

		/* remove slot from queue of pending */
		if (slot->w_queued)
			rpc_remque(rsock, slot);
		slot->w_next = rsock->free;
		rsock->free = slot;

		/* decrease congestion value */
		rsock->cong -= RPC_CWNDSCALE;
		if (rsock->cong < rsock->cwnd && rsock->backlog)
			wake_up(&rsock->backlog);
		if (rsock->shutdown)
			wake_up(&rsock->shutwait);

		req->rq_slot = NULL;
	}
}

/*
 * Adjust RPC congestion window
 */
static void
rpc_cwnd_adjust(struct rpc_sock *rsock, int timeout)
{
	unsigned long	cwnd = rsock->cwnd;

	if (!timeout) {
		if (rsock->cong >= cwnd) {
			/* The (cwnd >> 1) term makes sure
			 * the result gets rounded properly. */
			cwnd += (RPC_CWNDSCALE * RPC_CWNDSCALE +
					(cwnd >> 1)) / cwnd;
			if (cwnd > RPC_MAXCWND)
				cwnd = RPC_MAXCWND;
		}
	} else {
		if ((cwnd >>= 1) < RPC_CWNDSCALE)
			cwnd = RPC_CWNDSCALE;
		dprintk("RPC: cwnd decrease %08lx\n", cwnd);
	}
	dprintk("RPC: cong %08lx, cwnd was %08lx, now %08lx\n",
			rsock->cong, rsock->cwnd, cwnd);

	rsock->cwnd = cwnd;
}

static inline void
rpc_send_check(char *where, u32 *ptr)
{
	if (ptr[1] != htonl(RPC_CALL) || ptr[2] != htonl(RPC_VERSION)) {
		printk("RPC: %s sending evil packet:\n"
		       "     %08x %08x %08x %08x %08x %08x %08x %08x\n",
		       where,
		       ptr[0], ptr[1], ptr[2], ptr[3],
		       ptr[4], ptr[5], ptr[6], ptr[7]);
	}
}

/*
 * Place the actual RPC call.
 * We have to copy the iovec because sendmsg fiddles with its contents.
 */
static inline int
rpc_send(struct rpc_sock *rsock, struct rpc_wait *slot)
{
	struct rpc_ioreq *req = slot->w_req;
	struct iovec	iov[UIO_FASTIOV];

	if (rsock->shutdown)
		return -EIO;

	memcpy(iov, req->rq_svec, req->rq_snr * sizeof(iov[0]));
	slot->w_xid = *(u32 *)(iov[0].iov_base);
	if (!slot->w_queued)
		rpc_insque(rsock, slot);

	dprintk("rpc_send(%p, %x)\n", slot, slot->w_xid);
	rpc_send_check("rpc_send", (u32 *) req->rq_svec[0].iov_base);
	return rpc_sendmsg(rsock, iov, req->rq_snr, req->rq_slen,
				req->rq_addr, req->rq_alen);
}

/*
 * This is the same as rpc_send but for the functions exported to nfsiod
 */
int
rpc_transmit(struct rpc_sock *rsock, struct rpc_ioreq *req)
{
	rpc_send_check("rpc_transmit", (u32 *) req->rq_svec[0].iov_base);
	return rpc_send(rsock, req->rq_slot);
}

/*
 * Receive and dispatch a single reply
 */
static inline int
rpc_grok(struct rpc_sock *rsock)
{
	struct rpc_wait	*rovr;
	struct rpc_ioreq *req;
	struct iovec	iov[UIO_FASTIOV];
	u32		xid;
	int		safe, result;

	iov[0].iov_base = (void *) &xid;
	iov[0].iov_len  = sizeof(xid);
	result = rpc_recvmsg(rsock, iov, 1, sizeof(xid), MSG_PEEK);

	if (result < 0) {
		switch (-result) {
		case EAGAIN: case ECONNREFUSED:
			return 0;
		case ERESTARTSYS:
			return result;
		default:
			dprintk("rpc_grok: recv error = %d\n", result);
		}
	}
	if (result < 4) {
		printk(KERN_WARNING "RPC: impossible RPC reply size %d\n",
						result);
		iov[0].iov_base=(void*)&xid;	/* xid=32bits, which is large enough */
		iov[0].iov_len=result;
		rpc_recvmsg(rsock, iov, 1, result, 0);
		return 0;
	}

	dprintk("RPC: rpc_grok: got xid %08lx\n", (unsigned long) xid);

	/* Look for the caller */
	safe = 0;
	for (rovr = rsock->pending; rovr; rovr = rovr->w_next) {
		if (rovr->w_xid == xid)
			break;
		if (safe++ > RPC_MAXREQS) {
			printk(KERN_WARNING "RPC: loop in request Q!!\n");
			rovr = NULL;
			break;
		}
	}

	if (!rovr || rovr->w_gotit) {
		/* discard dgram */
		dprintk("RPC: rpc_grok: %s.\n",
			rovr? "duplicate reply" : "bad XID");
		iov[0].iov_base = (void *) &xid;
		iov[0].iov_len  = sizeof(xid);
		rpc_recvmsg(rsock, iov, 1, sizeof(xid), 0);
		return 0;
	}
	req = rovr->w_req;

	/* Now receive the reply... Copy the iovec first because of 
	 * memcpy_fromiovec fiddling. */
	memcpy(iov, req->rq_rvec, req->rq_rnr * sizeof(iov[0]));
	result = rpc_recvmsg(rsock, iov, req->rq_rnr, req->rq_rlen, 0);
	rovr->w_result = result;
	rovr->w_gotit = 1;

	/* ... and wake up the process */
	wake_up(&rovr->w_wait);

	return result;
}

/*
 * Wait for the reply to our call.
 */
static int
rpc_recv(struct rpc_sock *rsock, struct rpc_wait *slot)
{
	int	result;

	do {
		/* If we are not the receiver, wait on the sidelines */
		dprintk("RPC: rpc_recv TP1\n");
		while (rsock->pending != slot) {
			if (!slot->w_gotit)
				interruptible_sleep_on(&slot->w_wait);
			if (slot->w_gotit)
				return slot->w_result; /* quite important */
			if (current->signal & ~current->blocked)
				return -ERESTARTSYS;
			if (rsock->shutdown)
				return -EIO;
			if (current->timeout == 0)
				return -ETIMEDOUT;
		}

		/* Wait for data to arrive */
		if ((result = rpc_select(rsock)) < 0) {
			dprintk("RPC: select error = %d\n", result);
			return result;
		}

		/* Receive and dispatch */
		if ((result = rpc_grok(rsock)) < 0)
			return result;
	} while (current->timeout && !slot->w_gotit);

	return slot->w_gotit? slot->w_result : -ETIMEDOUT;
}

/*
 * Generic RPC call routine. This handles retries and timeouts etc pp.
 *
 * If sent is non-null, it assumes the called has already sent out the
 * message, so it won't need to do so unless a timeout occurs.
 */
int
rpc_doio(struct rpc_sock *rsock, struct rpc_ioreq *req,
			struct rpc_timeout *strategy, int sent)
{
	struct rpc_wait	*slot;
	int		result, retries;
	unsigned long	timeout;

	timeout = strategy->to_initval;
	retries = 0;
	slot = req->rq_slot;

	do {
		dprintk("RPC: rpc_doio: TP1 (req %p)\n", req);
		current->timeout = jiffies + timeout;
		if (slot == NULL) {
			result = rpc_reserve(rsock, req, 0);
			if (result == -ETIMEDOUT)
				goto timedout;
			if (result < 0)
				break;
			slot = req->rq_slot;
			rpc_send_check("rpc_doio",
				(u32 *) req->rq_svec[0].iov_base);
			rpc_insque(rsock, slot);
		}

		/* This check is for loopback NFS. Sometimes replies come
		 * in before biod has called rpc_doio... */
		if (slot->w_gotit) {
			result = slot->w_result;
			break;
		}

		dprintk("RPC: rpc_doio: TP2\n");
		if (sent || (result = rpc_send(rsock, slot)) >= 0) {
			result = rpc_recv(rsock, slot);
			sent = 0;
		}

		if (result != -ETIMEDOUT) {
			/* dprintk("RPC: rpc_recv returned %d\n", result); */
			rpc_cwnd_adjust(rsock, 0);
			break;
		}

		rpc_cwnd_adjust(rsock, 1);

timedout:
		dprintk("RPC: rpc_recv returned timeout.\n");
		if (strategy->to_exponential)
			timeout <<= 1;
		else
			timeout += strategy->to_increment;
		if (strategy->to_maxval && timeout >= strategy->to_maxval)
			timeout = strategy->to_maxval;
		if (strategy->to_retries && ++retries >= strategy->to_retries)
			break;
	} while (1);

	dprintk("RPC: rpc_doio: TP3\n");
	current->timeout = 0;
	return result;
}

/*
 */
int
rpc_call(struct rpc_sock *rsock, struct rpc_ioreq *req,
			struct rpc_timeout *strategy)
{
	int	result;

	result = rpc_doio(rsock, req, strategy, 0);
	if (req->rq_slot == NULL)
		printk(KERN_WARNING "RPC: bad: rq_slot == NULL\n");
	rpc_release(rsock, req);
	return result;
}

struct rpc_sock *
rpc_makesock(struct file *file)
{
	struct rpc_sock	*rsock;
	struct socket	*sock;
	struct sock	*sk;
	struct rpc_wait	*slot;
	int		i;

	dprintk("RPC: make RPC socket...\n");
	sock = &file->f_inode->u.socket_i;
	if (sock->type != SOCK_DGRAM || sock->ops->family != AF_INET) {
		printk(KERN_WARNING "RPC: only UDP sockets supported\n");
		return NULL;
	}
	sk = sock->sk;

	if ((rsock = kmalloc(sizeof(struct rpc_sock), GFP_KERNEL)) == NULL)
		return NULL;
	memset(rsock, 0, sizeof(*rsock)); /* Nnnngh! */

	rsock->sock = sock;
	rsock->inet = sk;
	rsock->file = file;
	rsock->cwnd = RPC_INITCWND;

	dprintk("RPC: slots %p, %p, ...\n", rsock->waiting, rsock->waiting + 1);
	rsock->free = rsock->waiting;
	for (i = 0, slot = rsock->waiting; i < RPC_MAXREQS-1; i++, slot++)
		slot->w_next = slot + 1;
	slot->w_next = NULL;

	dprintk("RPC: made socket %p\n", rsock);
	return rsock;
}

int
rpc_closesock(struct rpc_sock *rsock)
{
	unsigned long	t0 = jiffies;

	rsock->shutdown = 1;
	while (rsock->pending || waitqueue_active(&rsock->backlog)) {
		interruptible_sleep_on(&rsock->shutwait);
		if (current->signal & ~current->blocked)
			return -EINTR;
#if 1
		if (t0 && t0 - jiffies > 60 * HZ) {
			printk(KERN_WARNING "RPC: hanging in rpc_closesock.\n");
			t0 = 0;
		}
#endif
	}

	kfree(rsock);
	return 0;
}