Subversion Repositories or1k

/*

 *  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/udp.h>

#include <net/sock.h>

#include <asm/segment.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;

        oldfs = get_fs();

        set_fs(get_ds());

        result = sock->ops->sendmsg(sock, &msg, len, 0, 0);

        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, alen;

        msg.msg_iov     = iov;

        msg.msg_iovlen  = nr;

        msg.msg_name    = &sa;

        msg.msg_namelen = sizeof(sa);

        msg.msg_control = NULL;

        oldfs = get_fs();

        set_fs(get_ds());

        result = sock->ops->recvmsg(sock, &msg, len, 1, flags, &alen);

        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 select on the socket.

 */

static inline int

rpc_select(struct rpc_sock *rsock)

{

        struct select_table_entry entry;

        struct file     *file = rsock->file;

        select_table    wait_table;

        dprintk("RPC: selecting on socket...\n");

        wait_table.nr = 0;

        wait_table.entry = &entry;

        current->state = TASK_INTERRUPTIBLE;

        if (!file->f_op->select(file->f_inode, file, SEL_IN, &wait_table)

         && !file->f_op->select(file->f_inode, file, SEL_IN, NULL)) {

                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_MAXIOV];

        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_MAXIOV];

        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 = (struct sock *) sock->data;

        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;

}