Implement splice to fuse device.

fuse/loopback_test.go

@@ -135,7 +135,25 @@ func TestTouch(t *testing.T) {
 	}
 }
 
-func (tc *testCase) TestReadThrough(t *testing.T) {
+func TestReadLarge(t *testing.T) {
+	ts := NewTestCase(t)
+	defer ts.Cleanup()
+
+	content := make([]byte, 1024*1024)
+	for i := range content {
+		content[i] = byte(i)
+	}
+	err := ioutil.WriteFile(ts.origFile, []byte(content), 0644)
+	CheckSuccess(err)
+	
+	back, err := ioutil.ReadFile(ts.mountFile)
+	CheckSuccess(err)
+	if bytes.Compare(content, back) != 0 {
+		t.Errorf("content comparison failed")
+	}
+}
+
+func TestReadThrough(t *testing.T) {
 	ts := NewTestCase(t)
 	defer ts.Cleanup()
 

fuse/mountstate.go

@@ -10,6 +10,7 @@ import (
 	"unsafe"
 
 	"github.com/hanwen/go-fuse/raw"
+	"github.com/hanwen/go-fuse/splice"
 )
 
 const (
@@ -44,6 +45,7 @@ type MountState struct {
 	reqReaders int
 	outstandingReadBufs int
 
+	canSplice    bool
 	loops sync.WaitGroup
 }
 
@@ -159,7 +161,7 @@ func (ms *MountState) BufferPoolStats() string {
 	r = len(ms.readPool) + ms.reqReaders
 	ms.reqMu.Unlock()
 
-	s += fmt.Sprintf("read buffers: %d (sz %d )",
+	s += fmt.Sprintf(" read buffers: %d (sz %d )",
 		r, ms.opts.MaxWrite/PAGESIZE + 1)
 	return s
 }
@@ -359,11 +361,12 @@ func (ms *MountState) write(req *request) Status {
 	}
 
 	if req.flatData.FdSize > 0 {
-		if ms.TrySplice(req) {
+		if err := ms.TrySplice(header, req); err == nil {
 			return OK
 		} else {
 			buf := ms.AllocOut(req, uint32(req.flatData.FdSize))
 			req.flatData.Read(buf)
+			header = req.serializeHeader()
 		}
 	}
 
@@ -371,9 +374,38 @@ func (ms *MountState) write(req *request) Status {
 	return ToStatus(err)
 }
 
-func (ms *MountState) TrySplice(req *request) bool {
-	// TODO - implement.
-	return false
+func (ms *MountState) TrySplice(header []byte, req *request) error {
+	finalSplice, err := splice.Get()
+	if err != nil {
+		return err
+	}
+	defer splice.Done(finalSplice)
+
+	total := len(header) + req.flatData.FdSize
+	if !finalSplice.Grow(total) {
+		return fmt.Errorf("splice.Grow failed.")
+	}
+	
+	_, err = finalSplice.Write(header)
+	if err != nil {
+		return err
+	}
+
+	n, err := finalSplice.LoadFromAt(req.flatData.Fd, req.flatData.FdSize, req.flatData.FdOff)
+	if err != nil {
+		// TODO - handle EOF.
+		// TODO - extract the data from splice.
+		return err
+	}
+	if n != req.flatData.FdSize {
+		return fmt.Errorf("Size mismatch %d %d", n, req.flatData.FdSize)		
+	}
+
+	_, err = finalSplice.WriteTo(ms.mountFile.Fd(), total)
+	if err != nil {
+		return fmt.Errorf("Splice write %v", err)
+	}
+	return nil
 }
 
 func (ms *MountState) writeInodeNotify(entry *raw.NotifyInvalInodeOut) Status {

fuse/opcode.go

@@ -8,6 +8,7 @@ import (
 	"unsafe"
 
 	"github.com/hanwen/go-fuse/raw"
+	"github.com/hanwen/go-fuse/splice"
 )
 
 var _ = log.Printf
@@ -85,6 +86,14 @@ func doInit(state *MountState, req *request) {
 
 	state.kernelSettings = *input
 	state.kernelSettings.Flags = input.Flags & (raw.CAP_ASYNC_READ | raw.CAP_BIG_WRITES | raw.CAP_FILE_OPS)
+	if input.Minor >= 13 {
+		state.canSplice = true
+		maxW := splice.MaxPipeSize() - 4096
+		if !splice.Resizable() && state.opts.MaxWrite > maxW {
+			state.opts.MaxWrite = maxW
+		}
+	}
+	
 	out := &raw.InitOut{
 		Major:               FUSE_KERNEL_VERSION,
 		Minor:               OUR_MINOR_VERSION,
@@ -261,10 +270,6 @@ func doRead(state *MountState, req *request) {
 	in := (*ReadIn)(req.inData)
 	buf := state.AllocOut(req, in.Size)
 	res := state.fileSystem.Read(req.inHeader, in, buf)
-	if res.Ok() {
-		res.Read(buf)
-	}
-	
 	req.flatData = res
 	req.status = res.Status
 }

fuse/request.go

@@ -105,7 +105,11 @@ func (r *request) OutputDebug() string {
 			s := strings.TrimRight(string(r.flatData.Data), "\x00")
 			flatStr = fmt.Sprintf(" %q", s)
 		} else {
-			flatStr = fmt.Sprintf(" %d bytes data\n", r.flatData.Size())
+			spl := ""
+			if r.flatData.FdSize > 0 {
+				spl = " (splice)"
+			}
+			flatStr = fmt.Sprintf(" %d bytes data%s\n", r.flatData.Size(), spl)
 		}
 	}