mirror of
https://github.com/versity/versitygw.git
synced 2026-10-22 06:12:39 +00:00
fix: keep O_DIRECT active for object uploads and reads
Uploads copied the request body with io.CopyBuffer, which issues one write per read. Network reads return arbitrary lengths, so the first short read produced an unaligned write and the tmpfile silently reopened buffered for the rest of the object. - Fill the I/O buffer completely before each O_DIRECT write so only the final tail is unaligned; write that tail buffered without treating it as a fallback. - Allocate I/O buffers 4 KiB aligned and use 4 KiB alignment for O_DIRECT lengths and offsets to cover 4Kn devices. - Replace the O_DIRECT read wrapper with a reader that issues aligned preads into an aligned buffer for full and ranged GETs. sendfile from an O_DIRECT fd fails with EINVAL after a partial socket write, so O_DIRECT reads no longer go through sendfile; buffered reads still do. - Debug log unexpected O_DIRECT fallbacks on both paths.
This commit is contained in:
Ben McClelland
+145
-110
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user