fuse: support direct IO in loopback

open(2) has supports direct IO. When opening a file in the loopback
FS, writes to the underlying would fail, because the FUSE server does
not position the bytes to write (which come directly after the WriteIn
header) on a 512-byte boundary.

Fix this by allocating the input buffer with such that the first
content byte is aligned.

Add a test.

Fixes #328.

Change-Id: Ib7ad24aff673413ab8db4112d0b12143c2654617

fs/simple_test.go

@@ -51,14 +51,18 @@ type testOptions struct {
 	entryCache    bool
 	attrCache     bool
 	suppressDebug bool
+	testDir       string
 }
 
 func newTestCase(t *testing.T, opts *testOptions) *testCase {
 	if opts == nil {
 		opts = &testOptions{}
 	}
+	if opts.testDir == "" {
+		opts.testDir = testutil.TempDir()
+	}
 	tc := &testCase{
-		dir: testutil.TempDir(),
+		dir: opts.testDir,
 		T:   t,
 	}
 	tc.origDir = tc.dir + "/orig"
@@ -356,6 +360,31 @@ func TestPosix(t *testing.T) {
 	}
 }
 
+func TestOpenDirectIO(t *testing.T) {
+	// Apparently, tmpfs does not allow O_DIRECT, so try to create
+	// a test temp directory in the home directory.
+	ext4Dir := filepath.Join(os.Getenv("HOME"), ".go-fuse-test")
+	if err := os.MkdirAll(ext4Dir, 0755); err != nil {
+		t.Fatalf("MkdirAll: %v", err)
+	}
+	defer os.RemoveAll(ext4Dir)
+
+	posixtest.DirectIO(t, ext4Dir)
+	if t.Failed() {
+		t.Skip("DirectIO failed on underlying FS")
+	}
+
+	opts := testOptions{
+		testDir:    ext4Dir,
+		attrCache:  true,
+		entryCache: true,
+	}
+
+	tc := newTestCase(t, &opts)
+	defer tc.Clean()
+	posixtest.DirectIO(t, tc.mntDir)
+}
+
 func init() {
 	syscall.Umask(0)
 }

fuse/constants.go

@@ -32,4 +32,6 @@ const (
 	CUSE_INIT = 4096
 
 	O_ANYWRITE = uint32(os.O_WRONLY | os.O_RDWR | os.O_APPEND | os.O_CREATE | os.O_TRUNC)
+
+	logicalBlockSize = 512
 )

fuse/opcode.go

@@ -95,7 +95,6 @@ func doInit(server *Server, req *request) {
 		server.kernelSettings.Flags |= CAP_FLOCK_LOCKS | CAP_POSIX_LOCKS
 	}
 
-
 	dataCacheMode := input.Flags & CAP_AUTO_INVAL_DATA
 	if server.opts.ExplicitDataCacheControl {
 		// we don't want CAP_AUTO_INVAL_DATA even if we cannot go into fully explicit mode
@@ -520,6 +519,8 @@ func getHandler(o uint32) *operationHandler {
 	return operationHandlers[o]
 }
 
+var maxInputSize uintptr
+
 func init() {
 	operationHandlers = make([]*operationHandler, _OPCODE_COUNT)
 	for i := range operationHandlers {
@@ -531,6 +532,7 @@ func init() {
 		operationHandlers[op].FileNameOut = true
 	}
 
+	maxInputSize = 0
 	for op, sz := range map[uint32]uintptr{
 		_OP_FORGET:          unsafe.Sizeof(ForgetIn{}),
 		_OP_BATCH_FORGET:    unsafe.Sizeof(_BatchForgetIn{}),
@@ -571,6 +573,9 @@ func init() {
 		_OP_COPY_FILE_RANGE: unsafe.Sizeof(CopyFileRangeIn{}),
 	} {
 		operationHandlers[op].InputSize = sz
+		if sz > maxInputSize {
+			maxInputSize = sz
+		}
 	}
 
 	for op, sz := range map[uint32]uintptr{

fuse/server.go

@@ -15,6 +15,7 @@ import (
 	"sync"
 	"syscall"
 	"time"
+	"unsafe"
 )
 
 const (
@@ -175,8 +176,11 @@ func NewServer(fs RawFileSystem, mountPoint string, opts *MountOptions) (*Server
 			cancel: make(chan struct{}),
 		}
 	}
-	ms.readPool.New = func() interface{} { return make([]byte, o.MaxWrite+pageSize) }
-
+	ms.readPool.New = func() interface{} {
+		buf := make([]byte, o.MaxWrite+int(maxInputSize)+logicalBlockSize)
+		buf = alignSlice(buf, unsafe.Sizeof(WriteIn{}), logicalBlockSize, uintptr(o.MaxWrite)+maxInputSize)
+		return buf
+	}
 	mountPoint = filepath.Clean(mountPoint)
 	if !filepath.IsAbs(mountPoint) {
 		cwd, err := os.Getwd()
@@ -479,6 +483,15 @@ func (ms *Server) handleRequest(req *request) Status {
 	return Status(errNo)
 }
 
+// alignSlice ensures that the byte at alignedByte is aligned with the
+// given logical block size.  The input slice should be at least (size
+// + blockSize)
+func alignSlice(buf []byte, alignedByte, blockSize, size uintptr) []byte {
+	misaligned := uintptr(unsafe.Pointer(&buf[alignedByte])) & (blockSize - 1)
+	buf = buf[blockSize-misaligned:]
+	return buf[:size]
+}
+
 func (ms *Server) allocOut(req *request, size uint32) []byte {
 	if cap(req.bufferPoolOutputBuf) >= int(size) {
 		req.bufferPoolOutputBuf = req.bufferPoolOutputBuf[:size]
@@ -486,7 +499,10 @@ func (ms *Server) allocOut(req *request, size uint32) []byte {
 	}
 	if req.bufferPoolOutputBuf != nil {
 		ms.buffers.FreeBuffer(req.bufferPoolOutputBuf)
+		req.bufferPoolOutputBuf = nil
 	}
+	// As this allocated a multiple of the page size, very likely
+	// this is aligned to logicalBlockSize too, which is smaller.
 	req.bufferPoolOutputBuf = ms.buffers.AllocBuffer(size)
 	return req.bufferPoolOutputBuf
 }

posixtest/test.go

@@ -35,6 +35,46 @@ var All = map[string]func(*testing.T, string){
 	"RenameOverwriteDestExist":   RenameOverwriteDestExist,
 	"ReadDir":                    ReadDir,
 	"ReadDirPicksUpCreate":       ReadDirPicksUpCreate,
+	"DirectIO":                   DirectIO,
+}
+
+func DirectIO(t *testing.T, mnt string) {
+	fn := mnt + "/file.txt"
+	fd, err := syscall.Open(fn, syscall.O_TRUNC|syscall.O_CREAT|syscall.O_DIRECT|syscall.O_WRONLY, 0644)
+	if err == syscall.EINVAL {
+		t.Skip("FS does not support O_DIRECT")
+	}
+	if err != nil {
+		t.Fatalf("Open: %v", err)
+	}
+	defer func() {
+		if fd != 0 {
+			syscall.Close(fd)
+		}
+	}()
+	data := bytes.Repeat([]byte("bye"), 4096)
+	if n, err := syscall.Write(fd, data); err != nil || n != len(data) {
+		t.Fatalf("Write: %v (%d)", err, n)
+	}
+
+	err = syscall.Close(fd)
+	fd = 0
+	if err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+	fd, err = syscall.Open(fn, syscall.O_DIRECT|syscall.O_RDONLY, 0644)
+	if err != nil {
+		t.Fatalf("Open 2: %v", err)
+	}
+
+	roundtrip := bytes.Repeat([]byte("xxx"), 4096)
+	if n, err := syscall.Read(fd, roundtrip); err != nil || n != len(data) {
+		t.Fatalf("ReadAt: %v (%d)", err, n)
+	}
+
+	if bytes.Compare(roundtrip, data) != 0 {
+		t.Errorf("roundtrip made changes: got %q.., want %q..", roundtrip[:10], data[:10])
+	}
 }
 
 // SymlinkReadlink tests basic symlink functionality