Use a freelist for outgoing messages.

This eliminates about half of the allocations by bytes for gcsfuse's sequential
read benchmark.

connection.go

@@ -79,7 +79,8 @@ type Connection struct {
 	cancelFuncs map[uint64]func()
 
 	// Freelists, serviced by freelists.go.
-	inMessages freelist.Freelist // GUARDED_BY(mu)
+	inMessages  freelist.Freelist // GUARDED_BY(mu)
+	outMessages freelist.Freelist // GUARDED_BY(mu)
 }
 
 // State that is maintained for each in-flight op. This is stuffed into the
@@ -448,17 +449,15 @@ func (c *Connection) Reply(ctx context.Context, opErr error) {
 		c.errorLogger.Printf("%T error: %v", op, opErr)
 	}
 
-	// Send the reply to the kernel.
-	replyMsg := kernelResponse(m.Header().Unique, op, opErr, c.protocol)
-	if replyMsg != nil {
-		if err := c.writeMessage(replyMsg); err != nil {
-			if c.errorLogger != nil {
-				c.errorLogger.Printf("writeMessage: %v", err)
-			}
+	// Send the reply to the kernel, if one is required.
+	outMsg := c.kernelResponse(m.Header().Unique, op, opErr)
+	if outMsg != nil {
+		err := c.writeMessage(outMsg.Bytes())
+		c.putOutMessage(outMsg)
 
-			return
+		if err != nil && c.errorLogger != nil {
+			c.errorLogger.Printf("writeMessage: %v", err)
 		}
-
 	}
 }
 

conversions.go

@@ -387,35 +387,31 @@ func convertInMessage(
 // Outgoing messages
 ////////////////////////////////////////////////////////////////////////
 
-// Return the response that should be sent to the kernel. If the op requires no
-// response, return a nil response.
-func kernelResponse(
+// Return the response that should be sent to the kernel, or nil if the op
+// requires no response.
+func (c *Connection) kernelResponse(
 	fuseID uint64,
 	op interface{},
-	opErr error,
-	protocol fusekernel.Protocol) (msg []byte) {
-	// If the user replied with an error, create room enough just for the result
-	// header and fill it in with an error. Otherwise create an appropriate
+	opErr error) (m *buffer.OutMessage) {
+	// If the user replied with an error, create a response containing just the
+	// result header with the error filled in. Otherwise create an appropriate
 	// response.
-	var b buffer.OutMessage
 	if opErr != nil {
-		b = buffer.NewOutMessage(0)
+		m = c.getOutMessage()
 		if errno, ok := opErr.(syscall.Errno); ok {
-			b.OutHeader().Error = -int32(errno)
+			m.OutHeader().Error = -int32(errno)
 		} else {
-			b.OutHeader().Error = -int32(syscall.EIO)
+			m.OutHeader().Error = -int32(syscall.EIO)
 		}
 	} else {
-		b = kernelResponseForOp(op, protocol)
+		m = c.kernelResponseForOp(op)
 	}
 
-	msg = b.Bytes()
-
 	// Fill in the rest of the header, if a response is required.
-	if msg != nil {
-		h := b.OutHeader()
+	if m != nil {
+		h := m.OutHeader()
 		h.Unique = fuseID
-		h.Len = uint32(len(msg))
+		h.Len = uint32(m.Len())
 	}
 
 	return
@@ -423,29 +419,28 @@ func kernelResponse(
 
 // Like kernelResponse, but assumes the user replied with a nil error to the
 // op. Returns a nil response if no response is required.
-func kernelResponseForOp(
-	op interface{},
-	protocol fusekernel.Protocol) (b buffer.OutMessage) {
+func (c *Connection) kernelResponseForOp(
+	op interface{}) (m *buffer.OutMessage) {
 	// Create the appropriate output message
 	switch o := op.(type) {
 	case *fuseops.LookUpInodeOp:
-		size := fusekernel.EntryOutSize(protocol)
-		b = buffer.NewOutMessage(size)
-		out := (*fusekernel.EntryOut)(b.Grow(size))
+		size := fusekernel.EntryOutSize(c.protocol)
+		m = c.getOutMessage()
+		out := (*fusekernel.EntryOut)(m.Grow(size))
 		convertChildInodeEntry(&o.Entry, out)
 
 	case *fuseops.GetInodeAttributesOp:
-		size := fusekernel.AttrOutSize(protocol)
-		b = buffer.NewOutMessage(size)
-		out := (*fusekernel.AttrOut)(b.Grow(size))
+		size := fusekernel.AttrOutSize(c.protocol)
+		m = c.getOutMessage()
+		out := (*fusekernel.AttrOut)(m.Grow(size))
 		out.AttrValid, out.AttrValidNsec = convertExpirationTime(
 			o.AttributesExpiration)
 		convertAttributes(o.Inode, &o.Attributes, &out.Attr)
 
 	case *fuseops.SetInodeAttributesOp:
-		size := fusekernel.AttrOutSize(protocol)
-		b = buffer.NewOutMessage(size)
-		out := (*fusekernel.AttrOut)(b.Grow(size))
+		size := fusekernel.AttrOutSize(c.protocol)
+		m = c.getOutMessage()
+		out := (*fusekernel.AttrOut)(m.Grow(size))
 		out.AttrValid, out.AttrValidNsec = convertExpirationTime(
 			o.AttributesExpiration)
 		convertAttributes(o.Inode, &o.Attributes, &out.Attr)
@@ -454,85 +449,85 @@ func kernelResponseForOp(
 		// No response.
 
 	case *fuseops.MkDirOp:
-		size := fusekernel.EntryOutSize(protocol)
-		b = buffer.NewOutMessage(size)
-		out := (*fusekernel.EntryOut)(b.Grow(size))
+		size := fusekernel.EntryOutSize(c.protocol)
+		m = c.getOutMessage()
+		out := (*fusekernel.EntryOut)(m.Grow(size))
 		convertChildInodeEntry(&o.Entry, out)
 
 	case *fuseops.CreateFileOp:
-		eSize := fusekernel.EntryOutSize(protocol)
-		b = buffer.NewOutMessage(eSize + unsafe.Sizeof(fusekernel.OpenOut{}))
+		eSize := fusekernel.EntryOutSize(c.protocol)
+		m = c.getOutMessage()
 
-		e := (*fusekernel.EntryOut)(b.Grow(eSize))
+		e := (*fusekernel.EntryOut)(m.Grow(eSize))
 		convertChildInodeEntry(&o.Entry, e)
 
-		oo := (*fusekernel.OpenOut)(b.Grow(unsafe.Sizeof(fusekernel.OpenOut{})))
+		oo := (*fusekernel.OpenOut)(m.Grow(unsafe.Sizeof(fusekernel.OpenOut{})))
 		oo.Fh = uint64(o.Handle)
 
 	case *fuseops.CreateSymlinkOp:
-		size := fusekernel.EntryOutSize(protocol)
-		b = buffer.NewOutMessage(size)
-		out := (*fusekernel.EntryOut)(b.Grow(size))
+		size := fusekernel.EntryOutSize(c.protocol)
+		m = c.getOutMessage()
+		out := (*fusekernel.EntryOut)(m.Grow(size))
 		convertChildInodeEntry(&o.Entry, out)
 
 	case *fuseops.RenameOp:
-		b = buffer.NewOutMessage(0)
+		m = c.getOutMessage()
 
 	case *fuseops.RmDirOp:
-		b = buffer.NewOutMessage(0)
+		m = c.getOutMessage()
 
 	case *fuseops.UnlinkOp:
-		b = buffer.NewOutMessage(0)
+		m = c.getOutMessage()
 
 	case *fuseops.OpenDirOp:
-		b = buffer.NewOutMessage(unsafe.Sizeof(fusekernel.OpenOut{}))
-		out := (*fusekernel.OpenOut)(b.Grow(unsafe.Sizeof(fusekernel.OpenOut{})))
+		m = c.getOutMessage()
+		out := (*fusekernel.OpenOut)(m.Grow(unsafe.Sizeof(fusekernel.OpenOut{})))
 		out.Fh = uint64(o.Handle)
 
 	case *fuseops.ReadDirOp:
-		b = buffer.NewOutMessage(uintptr(len(o.Data)))
-		b.Append(o.Data)
+		m = c.getOutMessage()
+		m.Append(o.Data)
 
 	case *fuseops.ReleaseDirHandleOp:
-		b = buffer.NewOutMessage(0)
+		m = c.getOutMessage()
 
 	case *fuseops.OpenFileOp:
-		b = buffer.NewOutMessage(unsafe.Sizeof(fusekernel.OpenOut{}))
-		out := (*fusekernel.OpenOut)(b.Grow(unsafe.Sizeof(fusekernel.OpenOut{})))
+		m = c.getOutMessage()
+		out := (*fusekernel.OpenOut)(m.Grow(unsafe.Sizeof(fusekernel.OpenOut{})))
 		out.Fh = uint64(o.Handle)
 
 	case *fuseops.ReadFileOp:
-		b = buffer.NewOutMessage(uintptr(len(o.Data)))
-		b.Append(o.Data)
+		m = c.getOutMessage()
+		m.Append(o.Data)
 
 	case *fuseops.WriteFileOp:
-		b = buffer.NewOutMessage(unsafe.Sizeof(fusekernel.WriteOut{}))
-		out := (*fusekernel.WriteOut)(b.Grow(unsafe.Sizeof(fusekernel.WriteOut{})))
+		m = c.getOutMessage()
+		out := (*fusekernel.WriteOut)(m.Grow(unsafe.Sizeof(fusekernel.WriteOut{})))
 		out.Size = uint32(len(o.Data))
 
 	case *fuseops.SyncFileOp:
-		b = buffer.NewOutMessage(0)
+		m = c.getOutMessage()
 
 	case *fuseops.FlushFileOp:
-		b = buffer.NewOutMessage(0)
+		m = c.getOutMessage()
 
 	case *fuseops.ReleaseFileHandleOp:
-		b = buffer.NewOutMessage(0)
+		m = c.getOutMessage()
 
 	case *fuseops.ReadSymlinkOp:
-		b = buffer.NewOutMessage(uintptr(len(o.Target)))
-		b.AppendString(o.Target)
+		m = c.getOutMessage()
+		m.AppendString(o.Target)
 
 	case *statFSOp:
-		b = buffer.NewOutMessage(unsafe.Sizeof(fusekernel.StatfsOut{}))
-		b.Grow(unsafe.Sizeof(fusekernel.StatfsOut{}))
+		m = c.getOutMessage()
+		m.Grow(unsafe.Sizeof(fusekernel.StatfsOut{}))
 
 	case *interruptOp:
 		// No response.
 
 	case *initOp:
-		b = buffer.NewOutMessage(unsafe.Sizeof(fusekernel.InitOut{}))
-		out := (*fusekernel.InitOut)(b.Grow(unsafe.Sizeof(fusekernel.InitOut{})))
+		m = c.getOutMessage()
+		out := (*fusekernel.InitOut)(m.Grow(unsafe.Sizeof(fusekernel.InitOut{})))
 
 		out.Major = o.Library.Major
 		out.Minor = o.Library.Minor

freelists.go

@@ -20,14 +20,20 @@ import (
 	"github.com/jacobsa/fuse/internal/buffer"
 )
 
+////////////////////////////////////////////////////////////////////////
+// buffer.InMessage
+////////////////////////////////////////////////////////////////////////
+
 // LOCKS_EXCLUDED(c.mu)
-func (c *Connection) getInMessage() (m *buffer.InMessage) {
+func (c *Connection) getInMessage() (x *buffer.InMessage) {
 	c.mu.Lock()
-	m = (*buffer.InMessage)(c.inMessages.Get())
-	if m == nil {
-		m = new(buffer.InMessage)
-	}
+	x = (*buffer.InMessage)(c.inMessages.Get())
 	c.mu.Unlock()
+
+	if x == nil {
+		x = new(buffer.InMessage)
+	}
+
 	return
 }
 
@@ -37,3 +43,28 @@ func (c *Connection) putInMessage(x *buffer.InMessage) {
 	c.inMessages.Put(unsafe.Pointer(x))
 	c.mu.Unlock()
 }
+
+////////////////////////////////////////////////////////////////////////
+// buffer.OutMessage
+////////////////////////////////////////////////////////////////////////
+
+// LOCKS_EXCLUDED(c.mu)
+func (c *Connection) getOutMessage() (x *buffer.OutMessage) {
+	c.mu.Lock()
+	x = (*buffer.OutMessage)(c.outMessages.Get())
+	c.mu.Unlock()
+
+	if x == nil {
+		x = new(buffer.OutMessage)
+	}
+	x.Reset()
+
+	return
+}
+
+// LOCKS_EXCLUDED(c.mu)
+func (c *Connection) putOutMessage(x *buffer.OutMessage) {
+	c.mu.Lock()
+	c.outMessages.Put(unsafe.Pointer(x))
+	c.mu.Unlock()
+}

internal/buffer/out_message.go

@@ -15,70 +15,112 @@
 package buffer
 
 import (
+	"fmt"
+	"log"
 	"reflect"
 	"unsafe"
 
 	"github.com/jacobsa/fuse/internal/fusekernel"
 )
 
+const outHeaderSize = unsafe.Sizeof(fusekernel.OutHeader{})
+
+// We size out messages to be large enough to hold a header for the response
+// plus the largest read that may come in.
+const outMessageSize = outHeaderSize + MaxReadSize
+
 // OutMessage provides a mechanism for constructing a single contiguous fuse
 // message from multiple segments, where the first segment is always a
 // fusekernel.OutHeader message.
 //
-// Must be created with NewOutMessage. Exception: the zero value has
-// Bytes() == nil.
+// Must be initialized with Reset.
 type OutMessage struct {
-	slice []byte
+	offset  uintptr
+	storage [outMessageSize]byte
 }
 
-// Create a new buffer whose initial contents are a zeroed fusekernel.OutHeader
-// message, and with room enough to grow by extra bytes.
-func NewOutMessage(extra uintptr) (b OutMessage) {
-	const headerSize = unsafe.Sizeof(fusekernel.OutHeader{})
-	b.slice = make([]byte, headerSize, headerSize+extra)
-	return
+// Make sure alignment works out correctly, at least for the header.
+func init() {
+	a := unsafe.Alignof(OutMessage{})
+	o := unsafe.Offsetof(OutMessage{}.storage)
+	e := unsafe.Alignof(fusekernel.OutHeader{})
+
+	if a%e != 0 || o%e != 0 {
+		log.Panicf("Bad alignment or offset: %d, %d, need %d", a, o, e)
+	}
 }
 
-// Return a pointer to the header at the start of the buffer.
+// Reset the message so that it is ready to be used again. Afterward, the
+// contents are solely a zeroed header.
+func (m *OutMessage) Reset() {
+	m.offset = outHeaderSize
+	memclr(unsafe.Pointer(&m.storage), outHeaderSize)
+}
+
+// Return a pointer to the header at the start of the message.
 func (b *OutMessage) OutHeader() (h *fusekernel.OutHeader) {
-	sh := (*reflect.SliceHeader)(unsafe.Pointer(&b.slice))
-	h = (*fusekernel.OutHeader)(unsafe.Pointer(sh.Data))
+	h = (*fusekernel.OutHeader)(unsafe.Pointer(&b.storage))
 	return
 }
 
 // Grow the buffer by the supplied number of bytes, returning a pointer to the
-// start of the new segment. The sum of the arguments given to Grow must not
-// exceed the argument given to New when creating the buffer.
+// start of the new segment, which is zeroed. If there is no space left, return
+// the nil pointer.
 func (b *OutMessage) Grow(size uintptr) (p unsafe.Pointer) {
-	sh := (*reflect.SliceHeader)(unsafe.Pointer(&b.slice))
-	p = unsafe.Pointer(sh.Data + uintptr(sh.Len))
-	b.slice = b.slice[:len(b.slice)+int(size)]
+	p = b.GrowNoZero(size)
+	if p != nil {
+		memclr(p, size)
+	}
+
 	return
 }
 
-// Equivalent to growing by the length of p, then copying p into the new segment.
-func (b *OutMessage) Append(p []byte) {
-	sh := reflect.SliceHeader{
-		Data: uintptr(b.Grow(uintptr(len(p)))),
-		Len:  len(p),
-		Cap:  len(p),
+// Equivalent to Grow, except the new segment is not zeroed. Use with caution!
+func (b *OutMessage) GrowNoZero(size uintptr) (p unsafe.Pointer) {
+	if outMessageSize-b.offset < size {
+		return
 	}
 
-	copy(*(*[]byte)(unsafe.Pointer(&sh)), p)
+	p = unsafe.Pointer(uintptr(unsafe.Pointer(&b.storage)) + b.offset)
+	b.offset += size
+
+	return
 }
 
-// Equivalent to growing by the length of s, then copying s into the new segment.
-func (b *OutMessage) AppendString(s string) {
-	sh := reflect.SliceHeader{
-		Data: uintptr(b.Grow(uintptr(len(s)))),
-		Len:  len(s),
-		Cap:  len(s),
+// Equivalent to growing by the length of p, then copying p over the new
+// segment. Panics if there is not enough room available.
+func (b *OutMessage) Append(src []byte) {
+	p := b.GrowNoZero(uintptr(len(src)))
+	if p == nil {
+		panic(fmt.Sprintf("Can't grow %d bytes", len(src)))
 	}
 
-	copy(*(*[]byte)(unsafe.Pointer(&sh)), s)
+	sh := (*reflect.SliceHeader)(unsafe.Pointer(&src))
+	memmove(p, unsafe.Pointer(sh.Data), uintptr(sh.Len))
+
+	return
+}
+
+// Equivalent to growing by the length of s, then copying s over the new
+// segment. Panics if there is not enough room available.
+func (b *OutMessage) AppendString(src string) {
+	p := b.GrowNoZero(uintptr(len(src)))
+	if p == nil {
+		panic(fmt.Sprintf("Can't grow %d bytes", len(src)))
+	}
+
+	sh := (*reflect.StringHeader)(unsafe.Pointer(&src))
+	memmove(p, unsafe.Pointer(sh.Data), uintptr(sh.Len))
+
+	return
+}
+
+// Return the current size of the buffer.
+func (b *OutMessage) Len() int {
+	return int(b.offset)
 }
 
 // Return a reference to the current contents of the buffer.
 func (b *OutMessage) Bytes() []byte {
-	return b.slice
+	return b.storage[:int(b.offset)]
 }

internal/buffer/out_message_darwin.go

+// Copyright 2015 Google Inc. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package buffer
+
+// The maximum read size that we expect to ever see from the kernel, used for
+// calculating the size of out messages.
+//
+// Experimentally determined on OS X.
+const MaxReadSize = 1 << 20

internal/buffer/out_message_linux.go

+// Copyright 2015 Google Inc. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package buffer
+
+// The maximum read size that we expect to ever see from the kernel, used for
+// calculating the size of out messages.
+//
+// For 4 KiB pages, this is 128 KiB (cf. https://goo.gl/HOiEYo)
+const MaxReadSize = 1 << 17

internal/buffer/runtime.go

+// Copyright 2015 Google Inc. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package buffer
+
+import "unsafe"
+
+//go:noescape
+
+// Zero the n bytes starting at p.
+func memclr(p unsafe.Pointer, n uintptr)
+
+//go:noescape
+
+// Copy from src to dst, allowing overlap.
+func memmove(dst unsafe.Pointer, src unsafe.Pointer, n uintptr)

internal/buffer/runtime.s

+// Copyright 2015 Google Inc. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// +build amd64 arm64 ppc64 ppc64le
+
+// Assembly code isn't subject to visibility restrictions, so we can jump
+// directly into package runtime.
+//
+// Technique copied from here:
+//     https://github.com/golang/go/blob/d8c6dac/src/os/signal/sig.s
+
+#include "textflag.h"
+
+#ifdef GOARCH_ppc64
+#define JMP BR
+#endif
+#ifdef GOARCH_ppc64le
+#define JMP BR
+#endif
+
+TEXT ·memclr(SB),NOSPLIT,$0-16
+	JMP runtime·memclr(SB)
+
+TEXT ·memmove(SB),NOSPLIT,$0-24
+	JMP runtime·memmove(SB)