diff --git a/backend/posix/data_io_linux.go b/backend/posix/data_io_linux.go index 4c1b552c..358aefa7 100644 --- a/backend/posix/data_io_linux.go +++ b/backend/posix/data_io_linux.go @@ -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) } diff --git a/backend/posix/data_io_linux_test.go b/backend/posix/data_io_linux_test.go new file mode 100644 index 00000000..b950a59f --- /dev/null +++ b/backend/posix/data_io_linux_test.go @@ -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) + } +} diff --git a/backend/posix/otmpfile_common.go b/backend/posix/otmpfile_common.go index 0530d9b0..37177a76 100644 --- a/backend/posix/otmpfile_common.go +++ b/backend/posix/otmpfile_common.go @@ -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 { diff --git a/backend/posix/otmpfile_common_test.go b/backend/posix/otmpfile_common_test.go new file mode 100644 index 00000000..650e1f04 --- /dev/null +++ b/backend/posix/otmpfile_common_test.go @@ -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) + } +} diff --git a/backend/posix/posix.go b/backend/posix/posix.go index 4c36ca48..b30a63a6 100644 --- a/backend/posix/posix.go +++ b/backend/posix/posix.go @@ -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)