devstream: fast sequential file access with 9p pipelining experiment

1 files changed, 14 insertions, 0 deletions

diff --git a/sys/src/9/port/devmnt.c b/sys/src/9/port/devmnt.c
index 6e220f70a..2b6610afc 100644
--- a/sys/src/9/port/devmnt.c
+++ b/sys/src/9/port/devmnt.c
@@ -30,6 +30,8 @@ struct Mntrpc
 	uint	rpclen;		/* len of buffer */
 	Block*	b;		/* reply blocks */
 	Mntrpc*	flushed;	/* message this one flushes */
+	void	*iocomarg;	/* Rpc completion callback for pipelining */
+	void	(*iocomfun)(void*, int);
 	char	done;		/* Rpc completed */
 };
 
@@ -789,6 +791,9 @@ mountio(Mnt *m, Mntrpc *r)
 	lock(m);
 	r->z = &up->sleep;
 	r->m = m;
+	r->iocomarg = up->iocomarg;
+	r->iocomfun = up->iocomfun;
+	up->iocomfun = nil;
 	r->list = m->queue;
 	m->queue = r;
 	unlock(m);
@@ -806,6 +811,10 @@ mountio(Mnt *m, Mntrpc *r)
 	if(devtab[m->c->type]->write(m->c, r->rpc, n, 0) != n)
 		error(Emountrpc);
 
+	/* Rpc commited */
+	if(r->iocomfun != nil)
+		(*r->iocomfun)(r->iocomarg, 0);
+
 	/* Gate readers onto the mount point one at a time */
 	for(;;) {
 		lock(m);
@@ -948,6 +957,11 @@ mountmux(Mnt *m, Mntrpc *r)
 		/* look for a reply to a message */
 		if(q->request.tag == r->reply.tag) {
 			*l = q->list;
+
+			/* Rpc completed */
+			if(q->iocomfun != nil)
+				(*q->iocomfun)(q->iocomarg, 1);
+
 			if(q == r) {
 				q->done = 1;
 				unlock(m);