Re: [patch] single copy pipe rewrite

Re: [patch] single copy pipe rewrite

[David S. Miller]

   Date: Sun, 07 Jan 2001 00:25:16 +0100
   From: Manfred <manfred@colorfullife.com>

   Last march David Miller proposed using kiobuf for these data
   transfers, I've written a new patch for 2.4.

   (David's original patch contained 2 bugs: it doesn't protect
   properly against multiple writers and it causes a BUG() in
   pipe_read() when data is stored in both the kiobuf and the normal
   buffer)

A couple months ago David posted a revised version of his patch which
fixed both these and some other problems.  Most of the fixes were done
by Alexey Kuznetsov.

Later,
David S. Miller
davem@redhat.com
-
To unsubscribe from this list: send the line "unsubscribe linux-kernel" in
the body of a message to majordomo@vger.kernel.org
Please read the FAQ at http://www.tux.org/lkml/

[patch] single copy pipe rewrite

[Manfred]

[-- Attachment #1: Type: text/plain, Size: 591 bytes --]

(Linus, look away, I'll resubmit it for 2.5.0)

The Linux pipe implementation is extremely inefficient for long,
blocking data transfers with pipes (e.g. what gcc -pipe does):
* 2 memcopies: user space->kernel buffer->user space
* 2 context switches for each transfered page.

Last march David Miller proposed using kiobuf for these data transfers,
I've written a new patch for 2.4.

(David's original patch contained 2 bugs: it doesn't protect properly
against multiple writers and it causes a BUG() in pipe_read() when data
is stored in both the kiobuf and the normal buffer)


--
	Manfred

[-- Attachment #2: patch-kiopipe --]
[-- Type: text/plain, Size: 11373 bytes --]

// $Header$
// Kernel Version:
//  VERSION = 2
//  PATCHLEVEL = 4
//  SUBLEVEL = 0
//  EXTRAVERSION =
--- 2.4/fs/pipe.c	Thu Nov 16 22:18:26 2000
+++ build-2.4/fs/pipe.c	Sat Jan  6 22:43:32 2001
@@ -10,6 +10,8 @@
 #include <linux/malloc.h>
 #include <linux/module.h>
 #include <linux/init.h>
+#include <linux/iobuf.h>
+#include <linux/highmem.h>
 
 #include <asm/uaccess.h>
 
@@ -35,30 +37,100 @@
 	down(PIPE_SEM(*inode));
 }
 
+struct pipe_pio {
+	int *pdone;
+	struct kiobuf iobuf;
+};
+
+static int
+pio_copy_to_user(struct kiobuf* iobuf, char* ubuf, int chars, int offset)
+{
+	int page_nr;
+	offset += iobuf->offset;
+	page_nr = offset/PAGE_SIZE;
+	offset %= PAGE_SIZE;
+	while(chars) {
+		int pcount = PAGE_SIZE-offset;
+		void *kaddr;
+		if (pcount > chars)
+			pcount = chars;
+		kaddr = kmap(iobuf->maplist[page_nr]);
+		if (copy_to_user(ubuf, kaddr+offset, pcount))
+			return 1;
+		kunmap(iobuf->maplist[page_nr]);
+		chars -= pcount;
+		ubuf += pcount;
+		offset = 0;
+		page_nr++;
+	}
+	return 0;
+}
+
 static ssize_t
 pipe_read(struct file *filp, char *buf, size_t count, loff_t *ppos)
 {
 	struct inode *inode = filp->f_dentry->d_inode;
-	ssize_t size, read, ret;
+	ssize_t read, ret;
 
 	/* Seeks are not allowed on pipes.  */
-	ret = -ESPIPE;
-	read = 0;
 	if (ppos != &filp->f_pos)
-		goto out_nolock;
+		return -ESPIPE;
 
 	/* Always return 0 on null read.  */
-	ret = 0;
 	if (count == 0)
-		goto out_nolock;
+		return 0;
 
-	/* Get the pipe semaphore */
-	ret = -ERESTARTSYS;
-	if (down_interruptible(PIPE_SEM(*inode)))
-		goto out_nolock;
+	down(PIPE_SEM(*inode));
 
-	if (PIPE_EMPTY(*inode)) {
-do_more_read:
+	read = 0;
+	for (;;) {
+		/* read what data is available */
+		int chars = PIPE_LEN(*inode);
+		if (chars) {
+			char *pipebuf = PIPE_BASE(*inode);
+			int offset = PIPE_START(*inode);
+
+			if (chars > count)
+				chars = count;
+			ret = -EFAULT;
+			if(PIPE_IS_PIO(*inode)) {
+				struct pipe_pio* pio = ((struct pipe_pio*)pipebuf);
+				if(pio_copy_to_user(&pio->iobuf, buf, chars, offset))
+					goto out;
+
+				PIPE_LEN(*inode) -= chars;
+				if(!PIPE_LEN(*inode)) {
+					unmap_kiobuf(&pio->iobuf);
+					*pio->pdone = 1;
+					PIPE_IS_PIO(*inode) = 0;
+					PIPE_START(*inode) = 0;
+				} else {
+					PIPE_START(*inode) += chars;
+				}
+			} else {
+				if (offset+chars <= PIPE_SIZE) {
+					if (copy_to_user(buf, pipebuf+offset, chars))
+						goto out;
+				} else {
+					int p1 = PIPE_SIZE-offset;
+					if (copy_to_user(buf, pipebuf+offset, p1))
+						goto out;
+					if (copy_to_user(buf+p1, pipebuf, chars-p1))
+						goto out;
+				}
+				PIPE_LEN(*inode) -= chars;
+				if (!PIPE_LEN(*inode)) {
+					/* Cache behaviour optimization */
+					PIPE_START(*inode) = 0;
+				} else {
+					PIPE_START(*inode) += chars;
+					PIPE_START(*inode) &= (PIPE_SIZE - 1);
+				}
+			}
+			read += chars;
+			count -= chars;
+			buf += chars;
+		}
 		ret = 0;
 		if (!PIPE_WRITERS(*inode))
 			goto out;
@@ -67,65 +139,33 @@
 		if (filp->f_flags & O_NONBLOCK)
 			goto out;
 
-		for (;;) {
-			PIPE_WAITING_READERS(*inode)++;
-			pipe_wait(inode);
-			PIPE_WAITING_READERS(*inode)--;
-			ret = -ERESTARTSYS;
-			if (signal_pending(current))
-				goto out;
-			ret = 0;
+		ret = 0;
+		if (count && PIPE_MORE_DATA_WAITING(*inode) && !(filp->f_flags & O_NONBLOCK)) {
+			/*
+			 * We know that we are going to sleep: signal
+			 * writers synchronously that there is more
+			 * room.
+			 */
+			wake_up_interruptible_sync(PIPE_WAIT(*inode));
 			if (!PIPE_EMPTY(*inode))
-				break;
-			if (!PIPE_WRITERS(*inode))
-				goto out;
+				BUG();
+		} else if (read) {
+			goto out;
 		}
-	}
-
-	/* Read what data is available.  */
-	ret = -EFAULT;
-	while (count > 0 && (size = PIPE_LEN(*inode))) {
-		char *pipebuf = PIPE_BASE(*inode) + PIPE_START(*inode);
-		ssize_t chars = PIPE_MAX_RCHUNK(*inode);
-
-		if (chars > count)
-			chars = count;
-		if (chars > size)
-			chars = size;
 
-		if (copy_to_user(buf, pipebuf, chars))
+		pipe_wait(inode);
+		ret = -ERESTARTSYS;
+		if (signal_pending(current))
 			goto out;
-
-		read += chars;
-		PIPE_START(*inode) += chars;
-		PIPE_START(*inode) &= (PIPE_SIZE - 1);
-		PIPE_LEN(*inode) -= chars;
-		count -= chars;
-		buf += chars;
 	}
 
-	/* Cache behaviour optimization */
-	if (!PIPE_LEN(*inode))
-		PIPE_START(*inode) = 0;
 
-	if (count && PIPE_WAITING_WRITERS(*inode) && !(filp->f_flags & O_NONBLOCK)) {
-		/*
-		 * We know that we are going to sleep: signal
-		 * writers synchronously that there is more
-		 * room.
-		 */
-		wake_up_interruptible_sync(PIPE_WAIT(*inode));
-		if (!PIPE_EMPTY(*inode))
-			BUG();
-		goto do_more_read;
-	}
+out:
 	/* Signal writers asynchronously that there is more room.  */
-	wake_up_interruptible(PIPE_WAIT(*inode));
+	if (read && !PIPE_IS_PIO(*inode))
+		wake_up_interruptible(PIPE_WAIT(*inode));
 
-	ret = read;
-out:
 	up(PIPE_SEM(*inode));
-out_nolock:
 	if (read)
 		ret = read;
 	return ret;
@@ -136,92 +176,89 @@
 {
 	struct inode *inode = filp->f_dentry->d_inode;
 	ssize_t free, written, ret;
+	int pio_done;
 
 	/* Seeks are not allowed on pipes.  */
-	ret = -ESPIPE;
-	written = 0;
 	if (ppos != &filp->f_pos)
-		goto out_nolock;
+		return -ESPIPE;
 
 	/* Null write succeeds.  */
-	ret = 0;
 	if (count == 0)
-		goto out_nolock;
-
-	ret = -ERESTARTSYS;
-	if (down_interruptible(PIPE_SEM(*inode)))
-		goto out_nolock;
+		return 0;
 
-	/* No readers yields SIGPIPE.  */
-	if (!PIPE_READERS(*inode))
-		goto sigpipe;
+	down(PIPE_SEM(*inode));
 
 	/* If count <= PIPE_BUF, we have to make it atomic.  */
 	free = (count <= PIPE_BUF ? count : 1);
 
-	/* Wait, or check for, available space.  */
-	if (filp->f_flags & O_NONBLOCK) {
-		ret = -EAGAIN;
-		if (PIPE_FREE(*inode) < free)
+	written = 0;
+	pio_done = 1;
+	for(;;) {
+		int chars;
+
+		/* No readers yields SIGPIPE.  */
+		ret = -EPIPE;
+		if (!PIPE_READERS(*inode))
 			goto out;
-	} else {
-		while (PIPE_FREE(*inode) < free) {
-			PIPE_WAITING_WRITERS(*inode)++;
-			pipe_wait(inode);
-			PIPE_WAITING_WRITERS(*inode)--;
-			ret = -ERESTARTSYS;
-			if (signal_pending(current))
-				goto out;
-
-			if (!PIPE_READERS(*inode))
-				goto sigpipe;
-		}
-	}
-
-	/* Copy into available space.  */
-	ret = -EFAULT;
-	while (count > 0) {
-		int space;
-		char *pipebuf = PIPE_BASE(*inode) + PIPE_END(*inode);
-		ssize_t chars = PIPE_MAX_WCHUNK(*inode);
 
-		if ((space = PIPE_FREE(*inode)) != 0) {
+		/* Copy into available space.  */
+		ret = -EFAULT;
+		if (count >= PIPE_BUF && PIPE_EMPTY(*inode) &&
+		    (!(filp->f_flags & O_NONBLOCK))) {
+			struct pipe_pio* pio = (struct pipe_pio*)PIPE_BASE(*inode);
+			chars = KIO_MAX_ATOMIC_BYTES;
 			if (chars > count)
 				chars = count;
-			if (chars > space)
-				chars = space;
-
-			if (copy_from_user(pipebuf, buf, chars))
+			kiobuf_init(&pio->iobuf);
+			if(map_user_kiobuf(READ, &pio->iobuf, (unsigned long)buf, chars))
 				goto out;
+			PIPE_IS_PIO(*inode) = 1;
+			pio_done = 0;
+			pio->pdone = &pio_done;
+		} else if (!PIPE_IS_PIO(*inode) &&
+				(chars = PIPE_FREE(*inode)) >= free) {
+			char *pipebuf = PIPE_BASE(*inode);
+			int offset = PIPE_END(*inode);
 
-			written += chars;
-			PIPE_LEN(*inode) += chars;
-			count -= chars;
-			buf += chars;
-			space = PIPE_FREE(*inode);
-			continue;
+			if (chars > count)
+				chars = count;
+			if (chars+offset <= PIPE_SIZE) {
+				if (copy_from_user(pipebuf+offset, buf, chars))
+					goto out;
+			} else {
+				int p1 = PIPE_SIZE-offset;
+				if (copy_from_user(pipebuf+offset, buf, p1))
+					goto out;
+				if (copy_from_user(pipebuf, buf+p1, chars-p1))
+					goto out;
+			}
 		}
+		written += chars;
+		PIPE_LEN(*inode) += chars;
+		count -= chars;
+		buf += chars;
+		if (!count && pio_done)
+			break; /* DONE */
 
-		ret = written;
+		ret = -EAGAIN;
 		if (filp->f_flags & O_NONBLOCK)
 			break;
 
-		do {
-			/*
-			 * Synchronous wake-up: it knows that this process
-			 * is going to give up this CPU, so it doesnt have
-			 * to do idle reschedules.
-			 */
-			wake_up_interruptible_sync(PIPE_WAIT(*inode));
-			PIPE_WAITING_WRITERS(*inode)++;
-			pipe_wait(inode);
-			PIPE_WAITING_WRITERS(*inode)--;
-			if (signal_pending(current))
-				goto out;
-			if (!PIPE_READERS(*inode))
-				goto sigpipe;
-		} while (!PIPE_FREE(*inode));
-		ret = -EFAULT;
+		/*
+		 * Synchronous wake-up: it knows that this process
+		 * is going to give up this CPU, so it doesnt have
+		 * to do idle reschedules.
+		 */
+		wake_up_interruptible_sync(PIPE_WAIT(*inode));
+		if (count)
+			PIPE_MORE_DATA_WAITING(*inode)++;
+		pipe_wait(inode);
+		if (count)
+			PIPE_MORE_DATA_WAITING(*inode)--;
+		if (!count && pio_done)
+			break; /* DONE */
+		if (signal_pending(current))
+			goto out;
 	}
 
 	/* Signal readers asynchronously that there is more data.  */
@@ -231,18 +268,20 @@
 	mark_inode_dirty(inode);
 
 out:
+	if(!pio_done) {
+		struct pipe_pio* pio = (struct pipe_pio*)PIPE_BASE(*inode);
+		PIPE_IS_PIO(*inode) = 0;
+		written -= PIPE_LEN(*inode);
+		PIPE_LEN(*inode) = 0;
+		unmap_kiobuf(&pio->iobuf);
+		wake_up_interruptible(PIPE_WAIT(*inode));
+	}
 	up(PIPE_SEM(*inode));
-out_nolock:
 	if (written)
 		ret = written;
+	if (ret == -EPIPE)
+		send_sig(SIGPIPE, current, 0);
 	return ret;
-
-sigpipe:
-	if (written)
-		goto out;
-	up(PIPE_SEM(*inode));
-	send_sig(SIGPIPE, current, 0);
-	return -EPIPE;
 }
 
 static loff_t
@@ -453,9 +492,10 @@
 
 	init_waitqueue_head(PIPE_WAIT(*inode));
 	PIPE_BASE(*inode) = (char*) page;
+	PIPE_IS_PIO(*inode) = 0;
 	PIPE_START(*inode) = PIPE_LEN(*inode) = 0;
 	PIPE_READERS(*inode) = PIPE_WRITERS(*inode) = 0;
-	PIPE_WAITING_READERS(*inode) = PIPE_WAITING_WRITERS(*inode) = 0;
+	PIPE_MORE_DATA_WAITING(*inode) = 0;
 	PIPE_RCOUNTER(*inode) = PIPE_WCOUNTER(*inode) = 1;
 
 	return inode;
--- 2.4/include/linux/pipe_fs_i.h	Mon May  8 20:17:47 2000
+++ build-2.4/include/linux/pipe_fs_i.h	Sat Jan  6 17:56:33 2001
@@ -5,11 +5,11 @@
 struct pipe_inode_info {
 	wait_queue_head_t wait;
 	char *base;
+	unsigned int is_pio;
 	unsigned int start;
 	unsigned int readers;
 	unsigned int writers;
-	unsigned int waiting_readers;
-	unsigned int waiting_writers;
+	unsigned int more_data;
 	unsigned int r_counter;
 	unsigned int w_counter;
 };
@@ -21,12 +21,12 @@
 #define PIPE_SEM(inode)		(&(inode).i_sem)
 #define PIPE_WAIT(inode)	(&(inode).i_pipe->wait)
 #define PIPE_BASE(inode)	((inode).i_pipe->base)
+#define PIPE_IS_PIO(inode)	((inode).i_pipe->is_pio)
 #define PIPE_START(inode)	((inode).i_pipe->start)
 #define PIPE_LEN(inode)		((inode).i_size)
 #define PIPE_READERS(inode)	((inode).i_pipe->readers)
 #define PIPE_WRITERS(inode)	((inode).i_pipe->writers)
-#define PIPE_WAITING_READERS(inode)	((inode).i_pipe->waiting_readers)
-#define PIPE_WAITING_WRITERS(inode)	((inode).i_pipe->waiting_writers)
+#define PIPE_MORE_DATA_WAITING(inode)	((inode).i_pipe->more_data)
 #define PIPE_RCOUNTER(inode)	((inode).i_pipe->r_counter)
 #define PIPE_WCOUNTER(inode)	((inode).i_pipe->w_counter)
 
@@ -34,8 +34,6 @@
 #define PIPE_FULL(inode)	(PIPE_LEN(inode) == PIPE_SIZE)
 #define PIPE_FREE(inode)	(PIPE_SIZE - PIPE_LEN(inode))
 #define PIPE_END(inode)	((PIPE_START(inode) + PIPE_LEN(inode)) & (PIPE_SIZE-1))
-#define PIPE_MAX_RCHUNK(inode)	(PIPE_SIZE - PIPE_START(inode))
-#define PIPE_MAX_WCHUNK(inode)	(PIPE_SIZE - PIPE_END(inode))
 
 /* Drop the inode semaphore and wait for a pipe event, atomically */
 void pipe_wait(struct inode * inode);

Re: [patch] single copy pipe rewrite

[Manfred]

"David S. Miller" wrote:
> 
> A couple months ago David posted a revised version of his patch which
> fixed both these and some other problems.  Most of the fixes were done
> by Alexey Kuznetsov.
> 

Do you still have that patch?

Stephen Tweedie included the original, unrevied version in his
kiobuf.2.3.99.pre9-2.tar.gz from May, and I couldn't find a newer
version.

Was it posted to linux-kernel?
-
To unsubscribe from this list: send the line "unsubscribe linux-kernel" in
the body of a message to majordomo@vger.kernel.org
Please read the FAQ at http://www.tux.org/lkml/

Re: [patch] single copy pipe rewrite

[David S. Miller]

   Date: Sun, 07 Jan 2001 01:36:22 +0100
   From: Manfred <manfred@colorfullife.com>

   Do you still have that patch?

I think so, see below.

   Was it posted to linux-kernel?

Yes, it was.

I just found a copy, enjoy:

diff -ur ../vger3-001101/linux/fs/pipe.c linux/fs/pipe.c
--- ../vger3-001101/linux/fs/pipe.c	Sat Oct 14 18:38:24 2000
+++ linux/fs/pipe.c	Wed Nov  1 21:39:53 2000
@@ -8,6 +8,8 @@
 #include <linux/file.h>
 #include <linux/poll.h>
 #include <linux/malloc.h>
+#include <linux/iobuf.h>
+#include <linux/highmem.h>
 #include <linux/module.h>
 #include <linux/init.h>
 
@@ -22,6 +24,18 @@
  * -- Julian Bradfield 1999-06-07.
  */
 
+#define PIPE_UMAP(inode)	((inode).i_pipe->umap)
+#define PIPE_UMAPOFF(inode)	((inode).i_pipe->umap_offset)
+#define PIPE_UMAPLEN(inode)	((inode).i_pipe->umap_length)
+
+#define PIPE_UMAP_EMPTY(inode)	\
+	((PIPE_UMAP(inode) == NULL) || \
+	 (PIPE_UMAPOFF(inode) >= PIPE_UMAPLEN(inode)))
+
+#define PIPE_EMPTY(inode)	\
+	((PIPE_LEN(inode) == 0) && PIPE_UMAP_EMPTY(inode))
+
+
 /* Drop the inode semaphore and wait for a pipe event, atomically */
 void pipe_wait(struct inode * inode)
 {
@@ -36,6 +50,65 @@
 }
 
 static ssize_t
+pipe_copy_from_kiobuf(char *buf, size_t count, struct kiobuf *kio, int kio_offset)
+{
+	struct page **cur_page;
+	unsigned long cur_offset, remains_this_page;
+	char *cur_buf;
+	int kio_remains;
+
+	kio_remains = kio->length;
+	cur_page = kio->maplist;
+	cur_offset = kio->offset;
+	while (kio_offset > 0 && kio_remains > 0) {
+		remains_this_page = PAGE_SIZE - cur_offset;
+		if (kio_offset < remains_this_page) {
+			cur_offset += kio_offset;
+			kio_remains -= kio_offset;
+			break;
+		}
+		kio_offset -= remains_this_page;
+		kio_remains -= remains_this_page;
+		cur_offset = 0;
+		cur_page++;
+	}
+
+	cur_buf = buf;
+	while (kio_remains > 0) {
+		unsigned long kvaddr;
+		int err;
+
+		remains_this_page = PAGE_SIZE - cur_offset;
+		if (remains_this_page > count)
+			remains_this_page = count;
+		if (remains_this_page > kio_remains)
+			remains_this_page = kio_remains;
+
+		kvaddr = kmap(*cur_page);
+		err = copy_to_user(cur_buf, (void *)(kvaddr + cur_offset),
+				   remains_this_page);
+		kunmap(*cur_page);
+
+		if (err)
+			return -EFAULT;
+
+		cur_buf += remains_this_page;
+		count -= remains_this_page;
+		if (count <= 0)
+			break;
+
+		kio_remains -= remains_this_page;
+		if (kio_remains <= 0)
+			break;
+
+		cur_offset = 0;
+		cur_page++;
+	}
+
+	return cur_buf - buf;
+}
+
+static ssize_t
 pipe_read(struct file *filp, char *buf, size_t count, loff_t *ppos)
 {
 	struct inode *inode = filp->f_dentry->d_inode;
@@ -84,29 +157,44 @@
 
 	/* Read what data is available.  */
 	ret = -EFAULT;
-	while (count > 0 && (size = PIPE_LEN(*inode))) {
-		char *pipebuf = PIPE_BASE(*inode) + PIPE_START(*inode);
-		ssize_t chars = PIPE_MAX_RCHUNK(*inode);
-
-		if (chars > count)
-			chars = count;
-		if (chars > size)
-			chars = size;
+	if (PIPE_UMAP(*inode)) {
+		ssize_t chars;
 
-		if (copy_to_user(buf, pipebuf, chars))
+		chars = pipe_copy_from_kiobuf(buf, count,
+					      PIPE_UMAP(*inode),
+					      PIPE_UMAPOFF(*inode));
+		if (chars < 0)
 			goto out;
 
 		read += chars;
-		PIPE_START(*inode) += chars;
-		PIPE_START(*inode) &= (PIPE_SIZE - 1);
-		PIPE_LEN(*inode) -= chars;
 		count -= chars;
 		buf += chars;
-	}
+		PIPE_UMAPOFF(*inode) += chars;
+	} else {
+		while (count > 0 && (size = PIPE_LEN(*inode))) {
+			char *pipebuf = PIPE_BASE(*inode) + PIPE_START(*inode);
+			ssize_t chars = PIPE_MAX_RCHUNK(*inode);
 
-	/* Cache behaviour optimization */
-	if (!PIPE_LEN(*inode))
-		PIPE_START(*inode) = 0;
+			if (chars > count)
+				chars = count;
+			if (chars > size)
+				chars = size;
+
+			if (copy_to_user(buf, pipebuf, chars))
+				goto out;
+
+			read += chars;
+			PIPE_START(*inode) += chars;
+			PIPE_START(*inode) &= (PIPE_SIZE - 1);
+			PIPE_LEN(*inode) -= chars;
+			count -= chars;
+			buf += chars;
+		}
+
+		/* Cache behaviour optimization */
+		if (!PIPE_LEN(*inode))
+			PIPE_START(*inode) = 0;
+	}
 
 	if (count && PIPE_WAITING_WRITERS(*inode) && !(filp->f_flags & O_NONBLOCK)) {
 		/*
@@ -156,16 +244,20 @@
 	if (!PIPE_READERS(*inode))
 		goto sigpipe;
 
-	/* If count <= PIPE_BUF, we have to make it atomic.  */
-	free = (count <= PIPE_BUF ? count : 1);
+	/* If count <= PIPE_BUF, we have to make it atomic.
+	 * PIPE_BUF is not used more, pipe is atomic for blocking
+	 * IO up to KIO_MAX_ATOMIC_BYTES and for non-blocking
+	 * up to PAGE_SIZE.
+	 */
+	free = (count <= PIPE_SIZE ? count : PIPE_SIZE);
 
 	/* Wait, or check for, available space.  */
 	if (filp->f_flags & O_NONBLOCK) {
 		ret = -EAGAIN;
-		if (PIPE_FREE(*inode) < free)
+		if (PIPE_FREE(*inode) < free || PIPE_UMAP(*inode))
 			goto out;
 	} else {
-		while (PIPE_FREE(*inode) < free) {
+		while (PIPE_FREE(*inode) < free || PIPE_UMAP(*inode)) {
 			PIPE_WAITING_WRITERS(*inode)++;
 			pipe_wait(inode);
 			PIPE_WAITING_WRITERS(*inode)--;
@@ -180,11 +272,69 @@
 
 	/* Copy into available space.  */
 	ret = -EFAULT;
-	while (count > 0) {
+	if (count >= PAGE_SIZE &&
+	    !(filp->f_flags & O_NONBLOCK)) {
+		struct kiobuf *kio;
+		int err, do_sigpipe;;
+
+		/* Bulk, non-blocking, sole writer, use kio. */
+		err = alloc_kiovec(1, &kio);
+		if (err)
+			goto kio_abort;
+
+		do_sigpipe = 0;
+		PIPE_UMAP(*inode) = kio;
+		while (count > 0) {
+			unsigned long iosize = count;
+
+			if (iosize > KIO_MAX_ATOMIC_BYTES)
+				iosize = KIO_MAX_ATOMIC_BYTES;
+
+			/* Since we only use atomic sized transfers, the
+			 * only possible error here is EFAULT.
+			 */
+			err = map_user_kiobuf(WRITE, kio, (unsigned long)buf, iosize);
+			if (err)
+				break;
+
+			PIPE_UMAPOFF(*inode) = 0;
+			PIPE_UMAPLEN(*inode) = iosize;
+
+			do {
+				wake_up_interruptible_sync(PIPE_WAIT(*inode));
+				PIPE_WAITING_WRITERS(*inode)++;
+				pipe_wait(inode);
+				PIPE_WAITING_WRITERS(*inode)--;
+				if (signal_pending(current)) {
+					unmap_kiobuf(kio);
+					goto kio_break;
+				}
+				if (!PIPE_READERS(*inode)) {
+					unmap_kiobuf(kio);
+					do_sigpipe = 1;
+					goto kio_break;
+				}
+			} while (!PIPE_UMAP_EMPTY(*inode));
+
+			unmap_kiobuf(kio);
+			count -= iosize;
+			written += iosize;
+			buf += iosize;
+		}
+	kio_break:
+		free_kiovec(1, &kio);
+		PIPE_UMAP(*inode) = NULL;
+		
+		if (do_sigpipe)
+			goto sigpipe;
+	} else while (count > 0) {
 		int space;
-		char *pipebuf = PIPE_BASE(*inode) + PIPE_END(*inode);
-		ssize_t chars = PIPE_MAX_WCHUNK(*inode);
+		char *pipebuf;
+		ssize_t chars;
 
+kio_abort:
+		pipebuf = PIPE_BASE(*inode) + PIPE_END(*inode);
+		chars = PIPE_MAX_WCHUNK(*inode);
 		if ((space = PIPE_FREE(*inode)) != 0) {
 			if (chars > count)
 				chars = count;
@@ -285,8 +435,11 @@
 	poll_wait(filp, PIPE_WAIT(*inode), wait);
 
 	/* Reading only -- no need for acquiring the semaphore.  */
-	mask = POLLIN | POLLRDNORM;
-	if (PIPE_EMPTY(*inode))
+ 	/* Is this not racy occasionally? */
+ 	mask = 0;
+ 	if (!PIPE_EMPTY(*inode))
+ 		mask = POLLIN | POLLRDNORM;
+ 	else if (!PIPE_UMAP(*inode))
 		mask = POLLOUT | POLLWRNORM;
 	if (!PIPE_WRITERS(*inode) && filp->f_version != PIPE_WCOUNTER(*inode))
 		mask |= POLLHUP;
@@ -309,6 +462,8 @@
 		struct pipe_inode_info *info = inode->i_pipe;
 		inode->i_pipe = NULL;
 		free_page((unsigned long) info->base);
+		if (info->umap != NULL)
+			BUG();
 		kfree(info);
 	} else {
 		wake_up_interruptible(PIPE_WAIT(*inode));
@@ -457,6 +612,7 @@
 	PIPE_READERS(*inode) = PIPE_WRITERS(*inode) = 0;
 	PIPE_WAITING_READERS(*inode) = PIPE_WAITING_WRITERS(*inode) = 0;
 	PIPE_RCOUNTER(*inode) = PIPE_WCOUNTER(*inode) = 1;
+	PIPE_UMAP(*inode) = NULL;
 
 	return inode;
 fail_page:
diff -ur ../vger3-001101/linux/include/linux/pipe_fs_i.h linux/include/linux/pipe_fs_i.h
--- ../vger3-001101/linux/include/linux/pipe_fs_i.h	Fri May 26 21:31:37 2000
+++ linux/include/linux/pipe_fs_i.h	Wed Nov  1 21:39:54 2000
@@ -4,6 +4,9 @@
 #define PIPEFS_MAGIC 0x50495045
 struct pipe_inode_info {
 	wait_queue_head_t wait;
+	void *umap;
+	int umap_length;
+	int umap_offset;
 	char *base;
 	unsigned int start;
 	unsigned int readers;
@@ -30,7 +33,6 @@
 #define PIPE_RCOUNTER(inode)	((inode).i_pipe->r_counter)
 #define PIPE_WCOUNTER(inode)	((inode).i_pipe->w_counter)
 
-#define PIPE_EMPTY(inode)	(PIPE_LEN(inode) == 0)
 #define PIPE_FULL(inode)	(PIPE_LEN(inode) == PIPE_SIZE)
 #define PIPE_FREE(inode)	(PIPE_SIZE - PIPE_LEN(inode))
 #define PIPE_END(inode)	((PIPE_START(inode) + PIPE_LEN(inode)) & (PIPE_SIZE-1))
-
To unsubscribe from this list: send the line "unsubscribe linux-kernel" in
the body of a message to majordomo@vger.kernel.org
Please read the FAQ at http://www.tux.org/lkml/