Merge pull request #2462 from versity/ben/fix-odirect

fix: keep O_DIRECT active for object uploads and reads
This commit is contained in:
Ben McClelland
2026-10-02 14:03:14 -07:00
committed by GitHub
5 changed files with 495 additions and 119 deletions
+145 -110
View File
@@ -26,6 +26,8 @@ import (
"syscall"
"github.com/versity/versitygw/backend"
"github.com/versity/versitygw/debuglogger"
"golang.org/x/sys/unix"
)
// openDataRead opens object data for reading and applies O_DIRECT when
@@ -49,144 +51,177 @@ func openDataRead(name string, useODirect bool) (*os.File, error) {
return f, nil
}
func buildGetObjectBody(f *os.File, path string, startOffset, length, objSize int64, useODirect bool, readBufferSize int) (io.ReadCloser, error) {
func buildGetObjectBody(f *os.File, _ string, startOffset, length, objSize int64, useODirect bool, readBufferSize int) (io.ReadCloser, error) {
// openDataRead may have silently opened without O_DIRECT.
if useODirect && isODirectFile(f) {
return newODirectReader(f, startOffset, length, readBufferSize), nil
}
if startOffset == 0 && length == objSize {
if useODirect {
return newODirectReadFallbackFile(path, f, true), nil
}
return f, nil
}
if !useODirect {
rdr := io.NewSectionReader(f, startOffset, length)
return withReadBufferSize(&backend.FileSectionReadCloser{R: rdr, F: f}, readBufferSize), nil
rdr := io.NewSectionReader(f, startOffset, length)
return withReadBufferSize(&backend.FileSectionReadCloser{R: rdr, F: f}, readBufferSize), nil
}
func isODirectFile(f *os.File) bool {
flags, err := unix.FcntlInt(f.Fd(), unix.F_GETFL, 0)
return err == nil && flags&unix.O_DIRECT != 0
}
var odirectReadBufPool sync.Pool
// odirectReader serves [off, end) of f using block-aligned preads into an
// aligned buffer, as O_DIRECT requires for offset, length, and memory.
type odirectReader struct {
mu sync.Mutex
f *os.File
direct bool
off int64
end int64
buf []byte
bufOff int64
bufLen int
}
func newODirectReader(f *os.File, off, length int64, bufSize int) *odirectReader {
size := max(odirectAlign, (bufSize+odirectAlign-1)/odirectAlign*odirectAlign)
buf, ok := odirectReadBufPool.Get().(*[]byte)
if !ok || len(*buf) != size {
b := alignedBuffer(size)
buf = &b
}
rf := newODirectReadFallbackFile(path, f, true)
if _, err := rf.Seek(startOffset, io.SeekStart); err != nil {
_ = rf.Close()
return nil, fmt.Errorf("seek range start: %w", err)
}
return withReadBufferSize(&readerWithCloser{r: io.LimitReader(rf, length), c: rf}, readBufferSize), nil
}
type readerWithCloser struct {
r io.Reader
c io.Closer
}
func (r *readerWithCloser) Read(p []byte) (int, error) {
return r.r.Read(p)
}
func (r *readerWithCloser) Close() error {
return r.c.Close()
}
type odirectReadFallbackFile struct {
mu sync.Mutex
path string
f *os.File
useODirect bool
}
func newODirectReadFallbackFile(path string, f *os.File, useODirect bool) *odirectReadFallbackFile {
return &odirectReadFallbackFile{
path: path,
f: f,
useODirect: useODirect,
return &odirectReader{
f: f,
direct: true,
off: off,
end: off + length,
buf: *buf,
}
}
func (r *odirectReadFallbackFile) Read(p []byte) (int, error) {
func (r *odirectReader) Read(p []byte) (int, error) {
r.mu.Lock()
defer r.mu.Unlock()
n, err := r.f.Read(p)
if err != nil && n == 0 && r.useODirect && isODirectRuntimeFallbackErr(err) {
if fallbackErr := r.switchToBufferedAtCurrentOffsetLocked(); fallbackErr != nil {
return 0, fallbackErr
chunk, err := r.chunkLocked()
if err != nil {
return 0, err
}
n := copy(p, chunk)
r.off += int64(n)
return n, nil
}
func (r *odirectReader) WriteTo(w io.Writer) (int64, error) {
r.mu.Lock()
defer r.mu.Unlock()
var written int64
for {
chunk, err := r.chunkLocked()
if err == io.EOF {
return written, nil
}
if err != nil {
return written, err
}
n, err := w.Write(chunk)
r.off += int64(n)
written += int64(n)
if err != nil {
return written, err
}
return r.f.Read(p)
}
return n, err
}
func (r *odirectReadFallbackFile) WriteTo(w io.Writer) (int64, error) {
r.mu.Lock()
useODirect := r.useODirect
f := r.f
r.mu.Unlock()
if !useODirect {
return io.Copy(w, &onlyRead{r})
}
writerTo, ok := interface{}(f).(io.WriterTo)
if !ok {
return io.Copy(w, &onlyRead{r})
}
n, err := writerTo.WriteTo(w)
if err == nil || !isODirectRuntimeFallbackErr(err) {
return n, err
}
r.mu.Lock()
fallbackErr := r.switchToBufferedAtCurrentOffsetLocked()
r.mu.Unlock()
if fallbackErr != nil {
return n, fallbackErr
}
m, err := io.Copy(w, &onlyRead{r})
return n + m, err
}
func (r *odirectReadFallbackFile) Seek(offset int64, whence int) (int64, error) {
func (r *odirectReader) Close() error {
r.mu.Lock()
defer r.mu.Unlock()
return r.f.Seek(offset, whence)
}
func (r *odirectReadFallbackFile) Close() error {
r.mu.Lock()
defer r.mu.Unlock()
if r.buf != nil {
buf := r.buf
odirectReadBufPool.Put(&buf)
r.buf = nil
}
return r.f.Close()
}
func (r *odirectReadFallbackFile) switchToBufferedAtOffset(offset int64) error {
fd := strconv.Itoa(int(r.f.Fd()))
bf, openErr := os.Open(filepath.Join(procfddir, fd))
if openErr != nil {
return openErr
// chunkLocked returns the buffered bytes starting at r.off, reading more if needed.
func (r *odirectReader) chunkLocked() ([]byte, error) {
if r.buf == nil {
return nil, os.ErrClosed
}
if r.off >= r.end {
return nil, io.EOF
}
if r.off < r.bufOff || r.off >= r.bufOff+int64(r.bufLen) {
if err := r.fillLocked(); err != nil {
return nil, err
}
}
if _, seekErr := bf.Seek(offset, io.SeekStart); seekErr != nil {
stop := int64(r.bufLen)
if rem := r.end - r.bufOff; rem < stop {
stop = rem
}
return r.buf[r.off-r.bufOff : stop], nil
}
func (r *odirectReader) fillLocked() error {
alignedOff := r.off &^ (odirectAlign - 1)
buf := r.buf
if need := (r.end - alignedOff + odirectAlign - 1) &^ (odirectAlign - 1); need < int64(len(buf)) {
buf = buf[:int(need)]
}
n, err := pread(r.f, buf, alignedOff)
if n == 0 && r.direct && isODirectRuntimeFallbackErr(err) {
debuglogger.Logf("O_DIRECT read of %s failed at offset %d, falling back to buffered I/O: %v", r.f.Name(), alignedOff, err)
if ferr := r.switchToBufferedLocked(); ferr != nil {
return fmt.Errorf("reopen object in buffered mode after O_DIRECT read failure: %w", ferr)
}
n, err = pread(r.f, buf, alignedOff)
}
r.bufOff, r.bufLen = alignedOff, n
if r.off < alignedOff+int64(n) {
return nil
}
if err == nil || err == io.EOF {
return io.ErrUnexpectedEOF
}
return err
}
func (r *odirectReader) switchToBufferedLocked() error {
bf, err := os.Open(filepath.Join(procfddir, strconv.Itoa(int(r.f.Fd()))))
if err != nil {
return err
}
if err := r.f.Close(); err != nil {
_ = bf.Close()
return seekErr
return err
}
if closeErr := r.f.Close(); closeErr != nil {
_ = bf.Close()
return closeErr
}
r.f = bf
r.useODirect = false
r.direct = false
return nil
}
func (r *odirectReadFallbackFile) switchToBufferedAtCurrentOffsetLocked() error {
offset, seekErr := r.f.Seek(0, io.SeekCurrent)
if seekErr != nil {
return seekErr
// pread issues a single pread; os.File.ReadAt would loop into an unaligned
// offset after a short read at EOF, which O_DIRECT can reject.
func pread(f *os.File, b []byte, off int64) (int, error) {
for {
n, err := unix.Pread(int(f.Fd()), b, off)
if err == unix.EINTR {
continue
}
if err != nil {
return 0, err
}
if n == 0 {
return 0, io.EOF
}
return n, nil
}
return r.switchToBufferedAtOffset(offset)
}
+102
View File
@@ -0,0 +1,102 @@
// Copyright 2026 Versity Software
// This file is 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.
//go:build linux
package posix
import (
"bytes"
"io"
"math/rand"
"os"
"path/filepath"
"testing"
)
func TestODirectReaderRanges(t *testing.T) {
const objSize = 3*1024*1024 + 777
data := make([]byte, objSize)
rand.New(rand.NewSource(3)).Read(data)
path := filepath.Join(t.TempDir(), "obj")
if err := os.WriteFile(path, data, 0o644); err != nil {
t.Fatal(err)
}
ranges := []struct{ off, length int64 }{
{0, objSize},
{0, 0},
{1, 1},
{4095, 2},
{123457, 1024*1024 + 3},
{objSize - 777, 777},
{objSize - 1, 1},
}
for _, direct := range []bool{false, true} {
for _, rg := range ranges {
for _, useWriteTo := range []bool{false, true} {
f, err := openDataRead(path, direct)
if err != nil {
t.Fatal(err)
}
if direct && !isODirectFile(f) {
f.Close()
t.Skip("O_DIRECT not supported on temp filesystem")
}
r := newODirectReader(f, rg.off, rg.length, 64*1024+1)
var got bytes.Buffer
if useWriteTo {
_, err = r.WriteTo(&got)
} else {
_, err = io.Copy(&got, struct{ io.Reader }{r})
}
if err != nil {
t.Fatalf("direct=%v off=%d len=%d writeTo=%v: %v", direct, rg.off, rg.length, useWriteTo, err)
}
if !bytes.Equal(got.Bytes(), data[rg.off:rg.off+rg.length]) {
t.Fatalf("direct=%v off=%d len=%d writeTo=%v: content mismatch", direct, rg.off, rg.length, useWriteTo)
}
if direct && rg.off == 1 && rg.length == 1 && r.bufLen != odirectAlign {
t.Fatalf("small range read %d bytes, want %d", r.bufLen, odirectAlign)
}
if direct && !r.direct {
t.Fatalf("off=%d len=%d: fell back to buffered I/O", rg.off, rg.length)
}
if err := r.Close(); err != nil {
t.Fatal(err)
}
}
}
}
}
func TestODirectReaderTruncatedFile(t *testing.T) {
path := filepath.Join(t.TempDir(), "obj")
if err := os.WriteFile(path, make([]byte, 1000), 0o644); err != nil {
t.Fatal(err)
}
f, err := os.Open(path)
if err != nil {
t.Fatal(err)
}
r := newODirectReader(f, 0, 2000, 4096)
defer r.Close()
if _, err := io.Copy(io.Discard, struct{ io.Reader }{r}); err != io.ErrUnexpectedEOF {
t.Fatalf("got %v, want %v", err, io.ErrUnexpectedEOF)
}
}
+91 -3
View File
@@ -23,9 +23,28 @@ import (
"strconv"
"syscall"
"time"
"unsafe"
"github.com/versity/versitygw/debuglogger"
)
const odirectMinWriteAlign = 512
// odirectAlign covers both 512e and 4Kn logical block sizes for O_DIRECT
// offsets, lengths, and memory.
const odirectAlign = 4096
// alignedBuffer returns a size-byte slice whose first byte is odirectAlign aligned.
func alignedBuffer(size int) []byte {
b := make([]byte, size+odirectAlign)
off := 0
if rem := int(uintptr(unsafe.Pointer(&b[0])) % odirectAlign); rem != 0 {
off = odirectAlign - rem
}
return b[off : off+size : off+size]
}
func isODirectMemAligned(b []byte) bool {
return len(b) == 0 || uintptr(unsafe.Pointer(&b[0]))%odirectAlign == 0
}
func (tmp *tmpfile) Write(b []byte) (int, error) {
if int64(len(b)) > tmp.size {
@@ -33,7 +52,9 @@ func (tmp *tmpfile) Write(b []byte) (int, error) {
}
if tmp.useODirect && !isODirectLenAligned(len(b)) {
if err := tmp.switchToBufferedAtCurrentOffset(fmt.Sprintf("unaligned write length: len=%d len%%512=%d", len(b), len(b)%odirectMinWriteAlign)); err != nil {
reason := fmt.Sprintf("unaligned write length: len=%d len%%%d=%d", len(b), odirectAlign, len(b)%odirectAlign)
debuglogger.Logf("O_DIRECT tmpfile falling back to buffered I/O (%s/%s): %s", tmp.bucket, tmp.objname, reason)
if err := tmp.switchToBufferedAtCurrentOffset(reason); err != nil {
return 0, err
}
}
@@ -51,6 +72,73 @@ func (tmp *tmpfile) Write(b []byte) (int, error) {
return n, err
}
// copyFrom copies r into tmp. With O_DIRECT it fills buf completely before
// each write so that only the final tail of the object is unaligned.
func (tmp *tmpfile) copyFrom(r io.Reader, buf []byte) (int64, error) {
if !tmp.useODirect || len(buf) < odirectAlign {
return io.CopyBuffer(tmp, r, buf)
}
if !isODirectMemAligned(buf) {
buf = alignedBuffer(len(buf))
}
buf = buf[:len(buf)-len(buf)%odirectAlign]
var written int64
for {
// Not io.ReadFull: it drops errors returned alongside a full buffer,
// which is how HashReader reports checksum mismatches.
n := 0
var rerr error
for n < len(buf) && rerr == nil {
var nn int
nn, rerr = r.Read(buf[n:])
n += nn
}
if n > 0 {
w, werr := tmp.writeAlignedThenTail(buf[:n])
written += int64(w)
if werr != nil {
return written, werr
}
}
if rerr == io.EOF {
return written, nil
}
if rerr != nil {
return written, rerr
}
}
}
// writeAlignedThenTail writes the block-aligned prefix of b with O_DIRECT and
// any unaligned remainder buffered, without treating the tail as a fallback.
func (tmp *tmpfile) writeAlignedThenTail(b []byte) (int, error) {
if !tmp.useODirect || isODirectLenAligned(len(b)) {
return tmp.Write(b)
}
aligned := len(b) - len(b)%odirectAlign
n := 0
if aligned > 0 {
var err error
n, err = tmp.Write(b[:aligned])
if err != nil {
return n, err
}
}
if tmp.useODirect {
if err := tmp.switchToBufferedAtCurrentOffset("final unaligned tail"); err != nil {
return n, err
}
}
m, err := tmp.Write(b[aligned:])
return n + m, err
}
func (tmp *tmpfile) switchToBufferedAtCurrentOffset(reason string) error {
offset, seekErr := tmp.f.Seek(0, io.SeekCurrent)
if seekErr != nil {
@@ -83,7 +171,7 @@ func (tmp *tmpfile) switchToBufferedAtCurrentOffset(reason string) error {
}
func isODirectLenAligned(n int) bool {
return n%odirectMinWriteAlign == 0
return n%odirectAlign == 0
}
func isODirectRuntimeFallbackErr(err error) bool {
+152
View File
@@ -0,0 +1,152 @@
// Copyright 2026 Versity Software
// This file is 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 posix
import (
"bytes"
"errors"
"io"
"math/rand"
"os"
"path/filepath"
"testing"
)
// shortReader returns at most max bytes per Read, and errAtEOF with the final data.
type shortReader struct {
r io.Reader
max int
errAtEOF error
}
func (s *shortReader) Read(p []byte) (int, error) {
if len(p) > s.max {
p = p[:s.max]
}
n, err := s.r.Read(p)
if err == nil && s.errAtEOF != nil {
if br, ok := s.r.(*bytes.Reader); ok && br.Len() == 0 {
return n, s.errAtEOF
}
}
return n, err
}
// newTestODirectTmpfile builds a tmpfile over a regular file with useODirect
// set, so alignment decisions can be observed without O_DIRECT support.
func newTestODirectTmpfile(t *testing.T, size int64) *tmpfile {
t.Helper()
f, err := os.Create(filepath.Join(t.TempDir(), "obj"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { f.Close() })
return &tmpfile{f: f, useODirect: true, size: size}
}
func TestTmpfileCopyFromShortReads(t *testing.T) {
tests := []struct {
name string
size int
wantODirect bool
bufSize int
maxReadBytes int
}{
{"aligned size stays O_DIRECT", 4 * 1024 * 1024, true, 1024 * 1024, 7777},
{"unaligned tail switches at end", 4*1024*1024 + 123, false, 1024 * 1024, 7777},
{"512 aligned tail is not 4k aligned", 3*4096 + 512, false, 1024 * 1024, 7777},
{"small unaligned object", 100, false, 1024 * 1024, 13},
{"unaligned buffer size", 64 * 1024, true, 10000, 999},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
data := make([]byte, tc.size)
rand.New(rand.NewSource(1)).Read(data)
tmp := newTestODirectTmpfile(t, int64(tc.size))
rdr := &shortReader{r: bytes.NewReader(data), max: tc.maxReadBytes}
n, err := tmp.copyFrom(rdr, make([]byte, tc.bufSize))
if err != nil {
t.Fatalf("copyFrom: %v", err)
}
if n != int64(tc.size) {
t.Fatalf("copied %d, want %d", n, tc.size)
}
if tmp.useODirect != tc.wantODirect {
t.Fatalf("useODirect=%v, want %v", tmp.useODirect, tc.wantODirect)
}
got, err := os.ReadFile(tmp.f.Name())
if err != nil {
t.Fatal(err)
}
if !bytes.Equal(got, data) {
t.Fatal("file content mismatch")
}
})
}
}
func TestAlignedBuffer(t *testing.T) {
for _, size := range []int{1, 512, 4096, 10000, 1024 * 1024} {
b := alignedBuffer(size)
if len(b) != size || cap(b) != size {
t.Fatalf("size %d: len=%d cap=%d", size, len(b), cap(b))
}
if !isODirectMemAligned(b) {
t.Fatalf("size %d: buffer not %d-byte aligned", size, odirectAlign)
}
}
}
func TestTmpfileCopyFromUnalignedMemory(t *testing.T) {
data := make([]byte, 64*1024)
rand.New(rand.NewSource(2)).Read(data)
tmp := newTestODirectTmpfile(t, int64(len(data)))
buf := alignedBuffer(32*1024 + 1)[1:]
if isODirectMemAligned(buf) {
t.Fatal("test buffer unexpectedly aligned")
}
if _, err := tmp.copyFrom(&shortReader{r: bytes.NewReader(data), max: 1000}, buf); err != nil {
t.Fatalf("copyFrom: %v", err)
}
if !tmp.useODirect {
t.Fatal("fell back to buffered I/O")
}
got, err := os.ReadFile(tmp.f.Name())
if err != nil {
t.Fatal(err)
}
if !bytes.Equal(got, data) {
t.Fatal("file content mismatch")
}
}
func TestTmpfileCopyFromPropagatesErrorWithFinalData(t *testing.T) {
errBad := errors.New("bad digest")
data := make([]byte, 1024*1024)
tmp := newTestODirectTmpfile(t, int64(len(data)))
rdr := &shortReader{r: bytes.NewReader(data), max: 4096, errAtEOF: errBad}
_, err := tmp.copyFrom(rdr, make([]byte, len(data)))
if !errors.Is(err, errBad) {
t.Fatalf("got err %v, want %v", err, errBad)
}
}
+5 -6
View File
@@ -422,7 +422,7 @@ func New(rootdir string, ms meta.MetadataStorer, opts PosixOpts) (*Posix, error)
defaultEtag: opts.DefaultEtag,
ioBufferSize: ioBufferSize,
ioBufferPool: sync.Pool{New: func() any {
b := make([]byte, ioBufferSize)
b := alignedBuffer(ioBufferSize)
return &b
}},
dataIntegrityEtag: opts.DataIntegrityEtag,
@@ -484,7 +484,7 @@ func ioBufferSizeOrDefault(n int) int {
func (p *Posix) getIOBuffer() []byte {
bp, ok := p.ioBufferPool.Get().(*[]byte)
if !ok || bp == nil || cap(*bp) < p.ioBufferSize {
return make([]byte, p.ioBufferSize)
return alignedBuffer(p.ioBufferSize)
}
return (*bp)[:p.ioBufferSize]
}
@@ -3921,7 +3921,7 @@ func (p *Posix) UploadPartWithPostFunc(ctx context.Context, input *s3.UploadPart
buf := p.getIOBuffer()
defer p.putIOBuffer(buf)
_, err = io.CopyBuffer(f, tr, buf)
_, err = f.copyFrom(tr, buf)
if err != nil {
if errors.Is(err, syscall.EDQUOT) {
drainBody(tr)
@@ -4805,7 +4805,7 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
buf := p.getIOBuffer()
defer p.putIOBuffer(buf)
_, err = io.CopyBuffer(f, rdr, buf)
_, err = f.copyFrom(rdr, buf)
if err != nil {
if errors.Is(err, syscall.EDQUOT) {
drainBody(rdr)
@@ -6038,8 +6038,7 @@ func (p *Posix) GetObject(ctx context.Context, input *s3.GetObjectInput) (*s3.Ge
}
}
// Full-object responses can keep the underlying *os.File for sendfile.
// Linux range reads on O_DIRECT may need runtime fallback to buffered I/O.
// Buffered full-object responses keep the underlying *os.File for sendfile.
body, err := buildGetObjectBody(f, objPath, startOffset, length, objSize, p.enableODirect, p.ioBufferSize)
if err != nil {
return nil, fmt.Errorf("build get object body: %w", err)