libfuse: add buffer interface

Add a generic buffer interface for use with I/O.  Buffer vectors are
supplied and each buffer in the vector may be a memory pointer or a
file descriptor.

The fuse_reply_fd() interface is converted to using buffers.

5 files changed, 490 insertions, 34 deletions

diff --git a/lib/Makefile.am b/lib/Makefile.am
index 7b19fc2..22075b2 100644
--- a/lib/Makefile.am
+++ b/lib/Makefile.am
@@ -29,6 +29,7 @@ libfuse_la_SOURCES = 		\
 	fuse_opt.c		\
 	fuse_session.c		\
 	fuse_signals.c		\
+	buffer.c		\
 	cuse_lowlevel.c		\
 	helper.c		\
 	modules/subdir.c	\
diff --git a/lib/buffer.c b/lib/buffer.c
new file mode 100644
index 0000000..8940edd
--- /dev/null
+++ b/lib/buffer.c
@@ -0,0 +1,284 @@
+/*
+  FUSE: Filesystem in Userspace
+  Copyright (C) 2010  Miklos Szeredi <miklos@szeredi.hu>
+
+  This program can be distributed under the terms of the GNU LGPLv2.
+  See the file COPYING.LIB
+*/
+
+#define _GNU_SOURCE
+
+#include "fuse_i.h"
+#include "fuse_lowlevel.h"
+#include <string.h>
+#include <unistd.h>
+#include <errno.h>
+#include <assert.h>
+
+size_t fuse_buf_size(const struct fuse_bufvec *bufv)
+{
+	size_t i;
+	size_t size = 0;
+
+	for (i = 0; i < bufv->count; i++) {
+		if (bufv->buf[i].size == SIZE_MAX)
+			size = SIZE_MAX;
+		else
+			size += bufv->buf[i].size;
+	}
+
+	return size;
+}
+
+static size_t min_size(size_t s1, size_t s2)
+{
+	return s1 < s2 ? s1 : s2;
+}
+
+static ssize_t fuse_buf_write(const struct fuse_buf *dst, size_t dst_off,
+			      const struct fuse_buf *src, size_t src_off,
+			      size_t len)
+{
+	ssize_t res = 0;
+	size_t copied = 0;
+
+	while (len) {
+		if (dst->flags & FUSE_BUF_FD_SEEK) {
+			res = pwrite(dst->fd, src->mem + src_off, len,
+				     dst->pos + dst_off);
+		} else {
+			res = write(dst->fd, src->mem + src_off, len);
+		}
+		if (res == -1) {
+			if (!copied)
+				return -errno;
+			break;
+		}
+		if (res == 0)
+			break;
+
+		copied += res;
+		if (!(dst->flags & FUSE_BUF_FD_RETRY))
+			break;
+
+		src_off += res;
+		dst_off += res;
+		len -= res;
+	}
+
+	return copied;
+}
+
+static ssize_t fuse_buf_read(const struct fuse_buf *dst, size_t dst_off,
+			     const struct fuse_buf *src, size_t src_off,
+			     size_t len)
+{
+	ssize_t res = 0;
+	size_t copied = 0;
+
+	while (len) {
+		if (src->flags & FUSE_BUF_FD_SEEK) {
+			res = pread(src->fd, dst->mem + dst_off, len,
+				     src->pos + src_off);
+		} else {
+			res = read(src->fd, dst->mem + dst_off, len);
+		}
+		if (res == -1) {
+			if (!copied)
+				return -errno;
+			break;
+		}
+		if (res == 0)
+			break;
+
+		copied += res;
+		if (!(src->flags & FUSE_BUF_FD_RETRY))
+			break;
+
+		dst_off += res;
+		src_off += res;
+		len -= res;
+	}
+
+	return copied;
+}
+
+static ssize_t fuse_buf_fd_to_fd(const struct fuse_buf *dst, size_t dst_off,
+				 const struct fuse_buf *src, size_t src_off,
+				 size_t len)
+{
+	char buf[4096];
+	struct fuse_buf tmp = {
+		.size = sizeof(buf),
+		.flags = 0,
+	};
+	ssize_t res;
+	size_t copied = 0;
+
+	tmp.mem = buf;
+
+	while (len) {
+		size_t this_len = min_size(tmp.size, len);
+		size_t read_len;
+
+		res = fuse_buf_read(&tmp, 0, src, src_off, this_len);
+		if (res < 0) {
+			if (!copied)
+				return res;
+			break;
+		}
+		if (res == 0)
+			break;
+
+		read_len = res;
+		res = fuse_buf_write(dst, dst_off, &tmp, 0, read_len);
+		if (res < 0) {
+			if (!copied)
+				return res;
+			break;
+		}
+		if (res == 0)
+			break;
+
+		copied += res;
+
+		if (res < this_len)
+			break;
+
+		dst_off += res;
+		src_off += res;
+		len -= res;
+	}
+
+	return copied;
+}
+
+static ssize_t fuse_buf_splice(const struct fuse_buf *dst, size_t dst_off,
+			       const struct fuse_buf *src, size_t src_off,
+			       size_t len, enum fuse_buf_copy_flags flags)
+{
+	int splice_flags = 0;
+	off_t *srcpos = NULL;
+	off_t *dstpos = NULL;
+	off_t srcpos_val;
+	off_t dstpos_val;
+	ssize_t res;
+	size_t copied = 0;
+
+	if (flags & FUSE_BUF_SPLICE_MOVE)
+		splice_flags |= SPLICE_F_MOVE;
+	if (flags & FUSE_BUF_SPLICE_NONBLOCK)
+		splice_flags |= SPLICE_F_NONBLOCK;
+
+	if (src->flags & FUSE_BUF_FD_SEEK) {
+		srcpos_val = src->pos + src_off;
+		srcpos = &srcpos_val;
+	}
+	if (dst->flags & FUSE_BUF_FD_SEEK) {
+		dstpos_val = dst->pos + dst_off;
+		dstpos = &dstpos_val;
+	}
+
+	while (len) {
+		res = splice(src->fd, srcpos, dst->fd, dstpos, len,
+			     splice_flags);
+		if (res == -1) {
+			if (copied)
+				break;
+
+			if (errno != EINVAL || (flags & FUSE_BUF_FORCE_SPLICE))
+				return -errno;
+
+			/* Maybe splice is not supported for this combination */
+			return fuse_buf_fd_to_fd(dst, dst_off, src, src_off,
+						 len);
+		}
+		if (res == 0)
+			break;
+
+		copied += res;
+		if (!(src->flags & FUSE_BUF_FD_RETRY) &&
+		    !(dst->flags & FUSE_BUF_FD_RETRY)) {
+			break;
+		}
+
+		len -= res;
+	}
+
+	return copied;
+}
+
+
+static ssize_t fuse_buf_copy_one(const struct fuse_buf *dst, size_t dst_off,
+				 const struct fuse_buf *src, size_t src_off,
+				 size_t len, enum fuse_buf_copy_flags flags)
+{
+	int src_is_fd = src->flags & FUSE_BUF_IS_FD;
+	int dst_is_fd = dst->flags & FUSE_BUF_IS_FD;
+
+	if (!src_is_fd && !dst_is_fd) {
+		memcpy(dst->mem + dst_off, src->mem + src_off, len);
+		return len;
+	} else if (!src_is_fd) {
+		return fuse_buf_write(dst, dst_off, src, src_off, len);
+	} else if (!dst_is_fd) {
+		return fuse_buf_read(dst, dst_off, src, src_off, len);
+	} else if (flags & FUSE_BUF_NO_SPLICE) {
+		return fuse_buf_fd_to_fd(dst, dst_off, src, src_off, len);
+	} else {
+		return fuse_buf_splice(dst, dst_off, src, src_off, len, flags);
+	}
+}
+
+static const struct fuse_buf *fuse_bufvec_current(struct fuse_bufvec *bufv)
+{
+	return &bufv->buf[bufv->idx];
+}
+
+static int fuse_bufvec_advance(struct fuse_bufvec *bufv, size_t len)
+{
+	const struct fuse_buf *buf = fuse_bufvec_current(bufv);
+
+	bufv->off += len;
+	assert(bufv->off <= buf->size);
+	if (bufv->off == buf->size) {
+		assert(bufv->idx < bufv->count);
+		bufv->idx++;
+		if (bufv->idx == bufv->count)
+			return 0;
+		bufv->off = 0;
+	}
+	return 1;
+}
+
+ssize_t fuse_buf_copy(struct fuse_bufvec *dstv, struct fuse_bufvec *srcv,
+		      enum fuse_buf_copy_flags flags)
+{
+	size_t copied = 0;
+
+	for (;;) {
+		const struct fuse_buf *src = fuse_bufvec_current(srcv);
+		const struct fuse_buf *dst = fuse_bufvec_current(dstv);
+		size_t src_len = src->size - srcv->off;
+		size_t dst_len = dst->size - dstv->off;
+		size_t len = min_size(src_len, dst_len);
+		ssize_t res;
+
+		res = fuse_buf_copy_one(dst, dstv->off, src, srcv->off, len, flags);
+		if (res < 0) {
+			if (!copied)
+				return res;
+			break;
+		}
+		copied += res;
+
+		if (!fuse_bufvec_advance(srcv, res) ||
+		    !fuse_bufvec_advance(dstv, res))
+			break;
+
+		if (res < len)
+			break;
+	}
+
+	return copied;
+}
diff --git a/lib/fuse_i.h b/lib/fuse_i.h
index edd66f8..0206336 100644
--- a/lib/fuse_i.h
+++ b/lib/fuse_i.h
@@ -62,6 +62,7 @@ struct fuse_ll {
 	pthread_mutex_t lock;
 	int got_destroy;
 	pthread_key_t pipe_key;
+	int broken_splice_nonblock;
 };
 
 struct fuse_cmd {
diff --git a/lib/fuse_lowlevel.c b/lib/fuse_lowlevel.c
index 109f92d..388ab04 100644
--- a/lib/fuse_lowlevel.c
+++ b/lib/fuse_lowlevel.c
@@ -413,7 +413,17 @@ static struct fuse_ll_pipe *fuse_ll_get_pipe(struct fuse_ll *f)
 			free(llp);
 			return NULL;
 		}
-		/* the default size is 16 pages on linux
+
+		if (fcntl(llp->pipe[0], F_SETFL, O_NONBLOCK) == -1 ||
+		    fcntl(llp->pipe[1], F_SETFL, O_NONBLOCK) == -1) {
+			close(llp->pipe[0]);
+			close(llp->pipe[1]);
+			free(llp);
+			return NULL;
+		}
+
+		/*
+		 *the default size is 16 pages on linux
 		 */
 		llp->size = getpagesize() * 16;
 		llp->can_grow = 1;
@@ -424,21 +434,81 @@ static struct fuse_ll_pipe *fuse_ll_get_pipe(struct fuse_ll *f)
 	return llp;
 }
 
-int fuse_reply_fd(fuse_req_t req, int fd, loff_t *off, size_t len,
-		  unsigned int flags)
+static int send_reply_iov_buf(fuse_req_t req, const struct iovec *iov,
+			      int count, const char *buf, size_t len)
+{
+	int res;
+	struct iovec *new_iov;
+
+	new_iov = malloc((count + 1) * sizeof(struct iovec));
+	if (new_iov == NULL)
+		return fuse_reply_err(req, ENOMEM);
+
+	memcpy(new_iov, iov, count * sizeof(struct iovec));
+	new_iov[count].iov_base = (void *) buf;
+	new_iov[count].iov_len = len;
+	count++;
+
+	res = send_reply_iov(req, 0, new_iov, count);
+	free(new_iov);
+
+	return res;
+}
+
+static int read_back(int fd, char *buf, size_t len)
+{
+	int res;
+
+	res = read(fd, buf, len);
+	if (res == -1) {
+		fprintf(stderr, "fuse: internal error: failed to read back from pipe: %s\n", strerror(errno));
+		return -EIO;
+	}
+	if (res != len) {
+		fprintf(stderr, "fuse: internal error: short read back from pipe: %i from %zi\n", res, len);
+		return -EIO;
+	}
+	return 0;
+}
+
+static int fuse_reply_data_iov(fuse_req_t req, struct iovec *iov, int iov_count,
+			       struct fuse_bufvec *buf, unsigned int flags)
 {
 	int res;
-	void *buf;
+	size_t len = fuse_buf_size(buf);
 	struct fuse_out_header out;
-	struct iovec iov;
 	struct fuse_ll_pipe *llp;
 	int splice_flags;
 	size_t pipesize;
+	size_t total_fd_size;
+	size_t idx;
+	size_t headerlen;
+	struct fuse_buf pbuf = {
+		.size = len,
+	};
+	struct fuse_bufvec pipe_buf = {
+		.buf = &pbuf,
+		.count = 1,
+	};
 
 	static size_t pagesize = 0;
 	if (!pagesize)
 		pagesize = getpagesize();
 
+	if (req->f->broken_splice_nonblock)
+		goto fallback;
+
+	total_fd_size = 0;
+	for (idx = buf->idx; idx < buf->count; idx++) {
+		if (buf->buf[idx].flags & FUSE_BUF_IS_FD) {
+			total_fd_size = buf->buf[idx].size;
+			if (idx == buf->idx)
+				total_fd_size -= buf->off;
+		}
+	}
+	if (total_fd_size < 2 * pagesize)
+		goto fallback;
+
 	if (req->f->conn.proto_minor < 14 ||
 	    !(req->f->conn.want & FUSE_CAP_SPLICE_WRITE))
 		goto fallback;
@@ -447,12 +517,20 @@ int fuse_reply_fd(fuse_req_t req, int fd, loff_t *off, size_t len,
 	if (llp == NULL)
 		goto fallback;
 
+	iov[0].iov_base = &out;
+	iov[0].iov_len = sizeof(struct fuse_out_header);
+
+	headerlen = iov_length(iov, iov_count);
+
+	out.unique = req->unique;
+	out.error = 0;
+	out.len = headerlen + len;
 
 	/*
 	 * Heuristic for the required pipe size, does not work if the
 	 * source contains less than page size fragments
 	 */
-	pipesize = pagesize * 2 + len;
+	pipesize = pagesize * (iov_count + buf->count + 1) + out.len;
 
 	if (llp->size < pipesize) {
 		if (llp->can_grow) {
@@ -467,14 +545,8 @@ int fuse_reply_fd(fuse_req_t req, int fd, loff_t *off, size_t len,
 			goto fallback;
 	}
 
-	out.unique = req->unique;
-	out.error = 0;
-	out.len = len + sizeof(struct fuse_out_header);
-
-	iov.iov_base = &out;
-	iov.iov_len = sizeof(struct fuse_out_header);
 
-	res = vmsplice(llp->pipe[1], &iov, 1, 0);
+	res = vmsplice(llp->pipe[1], iov, iov_count, SPLICE_F_NONBLOCK);
 	if (res == -1) {
 		res = -errno;
 		perror("fuse: vmsplice to pipe");
@@ -487,11 +559,94 @@ int fuse_reply_fd(fuse_req_t req, int fd, loff_t *off, size_t len,
 		goto clear_pipe;
 	}
 
-	res = splice(fd, off, llp->pipe[1], NULL, len, 0);
-	if (res == -1) {
+	pbuf.flags = FUSE_BUF_IS_FD;
+	pbuf.fd = llp->pipe[1];
+
+	res = fuse_buf_copy(&pipe_buf, buf,
+			    FUSE_BUF_FORCE_SPLICE | FUSE_BUF_SPLICE_NONBLOCK);
+	if (res < 0) {
+		if (res == -EAGAIN || res == -EINVAL) {
+			/*
+			 * Should only get EAGAIN on kernels with
+			 * broken SPLICE_F_NONBLOCK support (<=
+			 * 2.6.35) where this error or a short read is
+			 * returned even if the pipe itself is not
+			 * full
+			 *
+			 * EINVAL might mean that splice can't handle
+			 * this combination of input and output.
+			 */
+			if (res == -EAGAIN)
+				req->f->broken_splice_nonblock = 1;
+
+			pthread_setspecific(req->f->pipe_key, NULL);
+			fuse_ll_pipe_free(llp);
+			goto fallback;
+		}
 		res = fuse_reply_err(req, errno);
 		goto clear_pipe;
 	}
+
+	if (res != 0 && res < len) {
+		struct fuse_buf mbuf = {
+			.size = len,
+		};
+		struct fuse_bufvec mem_buf = {
+			.buf = &mbuf,
+			.count = 1,
+		};
+		size_t now_len = res;
+		/*
+		 * For regular files a short count is either
+		 *  1) due to EOF, or
+		 *  2) because of broken SPLICE_F_NONBLOCK (see above)
+		 *
+		 * For other inputs it's possible that we overflowed
+		 * the pipe because of small buffer fragments.
+		 */
+
+		res = posix_memalign(&mbuf.mem, pagesize, len);
+		if (res != 0) {
+			res = fuse_reply_err(req, res);
+			goto clear_pipe;
+		}
+
+		mem_buf.off = now_len;
+		res = fuse_buf_copy(&mem_buf, buf, 0);
+		if (res > 0) {
+			char *tmpbuf;
+			size_t extra_len = res;
+			/*
+			 * Trickiest case: got more data.  Need to get
+			 * back the data from the pipe and then fall
+			 * back to regular write.
+			 */
+			tmpbuf = malloc(headerlen);
+			if (tmpbuf == NULL) {
+				free(mbuf.mem);
+				res = fuse_reply_err(req, ENOMEM);
+				goto clear_pipe;
+			}
+			res = read_back(llp->pipe[0], tmpbuf, headerlen);
+			if (res != 0) {
+				free(mbuf.mem);
+				goto clear_pipe;
+			}
+			free(tmpbuf);
+			res = read_back(llp->pipe[0], mbuf.mem, now_len);
+			if (res != 0) {
+				free(mbuf.mem);
+				goto clear_pipe;
+			}
+			len = now_len + extra_len;
+			res = send_reply_iov_buf(req, iov, iov_count,
+						 mbuf.mem, len);
+			free(mbuf.mem);
+			return res;
+		}
+		free(mbuf.mem);
+		res = now_len;
+	}
 	len = res;
 	out.len = len + sizeof(struct fuse_out_header);
 
@@ -502,12 +657,12 @@ int fuse_reply_fd(fuse_req_t req, int fd, loff_t *off, size_t len,
 	}
 
 	splice_flags = 0;
-	if ((flags & FUSE_REPLY_FD_MOVE) &&
+	if ((flags & FUSE_BUF_SPLICE_MOVE) &&
 	    (req->f->conn.want & FUSE_CAP_SPLICE_MOVE))
 		splice_flags |= SPLICE_F_MOVE;
 
 	res = splice(llp->pipe[0], NULL,
-		     fuse_chan_fd(req->ch), NULL, out.len, flags);
+		     fuse_chan_fd(req->ch), NULL, out.len, splice_flags);
 	if (res == -1) {
 		res = -errno;
 		perror("fuse: splice from pipe");
@@ -527,24 +682,37 @@ clear_pipe:
 	return res;
 
 fallback:
-	res = posix_memalign(&buf, pagesize, len);
-	if (res != 0)
-		return fuse_reply_err(req, res);
-
-	if (off != NULL) {
-		res = pread(fd, buf, len, *off);
-		if (res > 0)
-			*off += res;
-	} else {
-		res = read(fd, buf, len);
+	{
+		struct fuse_buf mbuf = {
+			.size = len,
+		};
+		struct fuse_bufvec mem_buf = {
+			.buf = &mbuf,
+			.count = 1,
+		};
+
+		res = posix_memalign(&mbuf.mem, pagesize, len);
+		if (res != 0)
+			return fuse_reply_err(req, res);
+
+		res = fuse_buf_copy(&mem_buf, buf, 0);
+		if (res < 0) {
+			free(mbuf.mem);
+			return fuse_reply_err(req, -res);
+		}
+		len = res;
+		res = send_reply_iov_buf(req, iov, iov_count, mbuf.mem, len);
+		free(mbuf.mem);
+
+		return res;
 	}
-	if (res == -1)
-		res = fuse_reply_err(req, errno);
-	else
-		res = fuse_reply_buf(req, buf, res);
-	free(buf);
+}
 
-	return res;
+int fuse_reply_data(fuse_req_t req, struct fuse_bufvec *bufv,
+		    enum fuse_buf_copy_flags flags)
+{
+	struct iovec iov[1];
+	return fuse_reply_data_iov(req, iov, 1, bufv, flags);
 }
 
 int fuse_reply_statfs(fuse_req_t req, const struct statvfs *stbuf)
diff --git a/lib/fuse_versionscript b/lib/fuse_versionscript
index a919870..bd0186d 100644
--- a/lib/fuse_versionscript
+++ b/lib/fuse_versionscript
@@ -183,7 +183,9 @@ FUSE_2.8 {
 
 FUSE_2.9 {
 	global:
-		fuse_reply_fd;
+		fuse_buf_copy;
+		fuse_buf_size;
+		fuse_reply_data;
 
 	local:
 		*;