ReadAt may return io.EOF t end of file
related to https://github.com/seaweedfs/seaweedfs/issues/6219
This commit is contained in:
@@ -4,6 +4,7 @@ import (
|
|||||||
"github.com/seaweedfs/seaweedfs/weed/glog"
|
"github.com/seaweedfs/seaweedfs/weed/glog"
|
||||||
"github.com/seaweedfs/seaweedfs/weed/util"
|
"github.com/seaweedfs/seaweedfs/weed/util"
|
||||||
"github.com/seaweedfs/seaweedfs/weed/util/mem"
|
"github.com/seaweedfs/seaweedfs/weed/util/mem"
|
||||||
|
"io"
|
||||||
"os"
|
"os"
|
||||||
"sync"
|
"sync"
|
||||||
)
|
)
|
||||||
@@ -130,7 +131,10 @@ func (sc *SwapFileChunk) ReadDataAt(p []byte, off int64, tsNs int64) (maxStop in
|
|||||||
logicStop := min(off+int64(len(p)), chunkStartOffset+t.stopOffset)
|
logicStop := min(off+int64(len(p)), chunkStartOffset+t.stopOffset)
|
||||||
if logicStart < logicStop {
|
if logicStart < logicStop {
|
||||||
actualStart := logicStart - chunkStartOffset + int64(sc.actualChunkIndex)*sc.swapfile.chunkSize
|
actualStart := logicStart - chunkStartOffset + int64(sc.actualChunkIndex)*sc.swapfile.chunkSize
|
||||||
if _, err := sc.swapfile.file.ReadAt(p[logicStart-off:logicStop-off], actualStart); err != nil {
|
if n, err := sc.swapfile.file.ReadAt(p[logicStart-off:logicStop-off], actualStart); err != nil {
|
||||||
|
if err == io.EOF && n == int(logicStop-logicStart) {
|
||||||
|
err = nil
|
||||||
|
}
|
||||||
glog.Errorf("failed to reading swap file %s: %v", sc.swapfile.file.Name(), err)
|
glog.Errorf("failed to reading swap file %s: %v", sc.swapfile.file.Name(), err)
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ package backend
|
|||||||
import (
|
import (
|
||||||
"github.com/seaweedfs/seaweedfs/weed/glog"
|
"github.com/seaweedfs/seaweedfs/weed/glog"
|
||||||
. "github.com/seaweedfs/seaweedfs/weed/storage/types"
|
. "github.com/seaweedfs/seaweedfs/weed/storage/types"
|
||||||
|
"io"
|
||||||
"os"
|
"os"
|
||||||
"runtime"
|
"runtime"
|
||||||
"time"
|
"time"
|
||||||
@@ -43,7 +44,11 @@ func (df *DiskFile) ReadAt(p []byte, off int64) (n int, err error) {
|
|||||||
if df.File == nil {
|
if df.File == nil {
|
||||||
return 0, os.ErrClosed
|
return 0, os.ErrClosed
|
||||||
}
|
}
|
||||||
return df.File.ReadAt(p, off)
|
n, err = df.File.ReadAt(p, off)
|
||||||
|
if err == io.EOF && n == len(p) {
|
||||||
|
err = nil
|
||||||
|
}
|
||||||
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
func (df *DiskFile) WriteAt(p []byte, off int64) (n int, err error) {
|
func (df *DiskFile) WriteAt(p []byte, off int64) (n int, err error) {
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ package erasure_coding
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"io"
|
||||||
"os"
|
"os"
|
||||||
"path"
|
"path"
|
||||||
"strconv"
|
"strconv"
|
||||||
@@ -93,6 +94,10 @@ func (shard *EcVolumeShard) Destroy() {
|
|||||||
|
|
||||||
func (shard *EcVolumeShard) ReadAt(buf []byte, offset int64) (int, error) {
|
func (shard *EcVolumeShard) ReadAt(buf []byte, offset int64) (int, error) {
|
||||||
|
|
||||||
return shard.ecdFile.ReadAt(buf, offset)
|
n, err := shard.ecdFile.ReadAt(buf, offset)
|
||||||
|
if err == io.EOF && n == len(buf) {
|
||||||
|
err = nil
|
||||||
|
}
|
||||||
|
return n, err
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -255,8 +255,10 @@ func SearchNeedleFromSortedIndex(ecxFile *os.File, ecxFileSize int64, needleId t
|
|||||||
l, h := int64(0), ecxFileSize/types.NeedleMapEntrySize
|
l, h := int64(0), ecxFileSize/types.NeedleMapEntrySize
|
||||||
for l < h {
|
for l < h {
|
||||||
m := (l + h) / 2
|
m := (l + h) / 2
|
||||||
if _, err := ecxFile.ReadAt(buf, m*types.NeedleMapEntrySize); err != nil {
|
if n, err := ecxFile.ReadAt(buf, m*types.NeedleMapEntrySize); err != nil {
|
||||||
return types.Offset{}, types.TombstoneFileSize, fmt.Errorf("ecx file %d read at %d: %v", ecxFileSize, m*types.NeedleMapEntrySize, err)
|
if n != types.NeedleMapEntrySize {
|
||||||
|
return types.Offset{}, types.TombstoneFileSize, fmt.Errorf("ecx file %d read at %d: %v", ecxFileSize, m*types.NeedleMapEntrySize, err)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
key, offset, size = idx.IdxFileEntry(buf)
|
key, offset, size = idx.IdxFileEntry(buf)
|
||||||
if key == needleId {
|
if key == needleId {
|
||||||
|
|||||||
@@ -196,6 +196,9 @@ func ReadNeedleHeader(r backend.BackendStorageFile, version Version, offset int6
|
|||||||
|
|
||||||
var count int
|
var count int
|
||||||
count, err = r.ReadAt(bytes, offset)
|
count, err = r.ReadAt(bytes, offset)
|
||||||
|
if err == io.EOF && count == NeedleHeaderSize {
|
||||||
|
err = nil
|
||||||
|
}
|
||||||
if count <= 0 || err != nil {
|
if count <= 0 || err != nil {
|
||||||
return nil, bytes, 0, err
|
return nil, bytes, 0, err
|
||||||
}
|
}
|
||||||
@@ -230,7 +233,12 @@ func (n *Needle) ReadNeedleBody(r backend.BackendStorageFile, version Version, o
|
|||||||
return nil, nil
|
return nil, nil
|
||||||
}
|
}
|
||||||
bytes = make([]byte, bodyLength)
|
bytes = make([]byte, bodyLength)
|
||||||
if _, err = r.ReadAt(bytes, offset); err != nil {
|
readCount, err := r.ReadAt(bytes, offset)
|
||||||
|
if err == io.EOF && int64(readCount) == bodyLength {
|
||||||
|
err = nil
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
glog.Errorf("%s read %d bodyLength %d offset %d: %v", r.Name(), readCount, bodyLength, offset, err)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -21,6 +21,9 @@ func (n *Needle) ReadNeedleData(r backend.BackendStorageFile, volumeOffset int64
|
|||||||
startOffset := volumeOffset + NeedleHeaderSize + DataSizeSize + needleOffset
|
startOffset := volumeOffset + NeedleHeaderSize + DataSizeSize + needleOffset
|
||||||
|
|
||||||
count, err = r.ReadAt(data[:sizeToRead], startOffset)
|
count, err = r.ReadAt(data[:sizeToRead], startOffset)
|
||||||
|
if err == io.EOF && int64(count) == sizeToRead {
|
||||||
|
err = nil
|
||||||
|
}
|
||||||
if err != nil {
|
if err != nil {
|
||||||
fileSize, _, _ := r.GetStat()
|
fileSize, _, _ := r.GetStat()
|
||||||
glog.Errorf("%s read %d %d size %d at offset %d fileSize %d: %v", r.Name(), n.Id, needleOffset, sizeToRead, volumeOffset, fileSize, err)
|
glog.Errorf("%s read %d %d size %d at offset %d fileSize %d: %v", r.Name(), n.Id, needleOffset, sizeToRead, volumeOffset, fileSize, err)
|
||||||
@@ -40,6 +43,9 @@ func (n *Needle) ReadNeedleMeta(r backend.BackendStorageFile, offset int64, size
|
|||||||
bytes := make([]byte, NeedleHeaderSize+DataSizeSize)
|
bytes := make([]byte, NeedleHeaderSize+DataSizeSize)
|
||||||
|
|
||||||
count, err := r.ReadAt(bytes, offset)
|
count, err := r.ReadAt(bytes, offset)
|
||||||
|
if err == io.EOF && count == NeedleHeaderSize+DataSizeSize {
|
||||||
|
err = nil
|
||||||
|
}
|
||||||
if count != NeedleHeaderSize+DataSizeSize || err != nil {
|
if count != NeedleHeaderSize+DataSizeSize || err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ package storage
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"io"
|
||||||
"os"
|
"os"
|
||||||
"sync/atomic"
|
"sync/atomic"
|
||||||
|
|
||||||
@@ -152,8 +153,11 @@ func reverseWalkIndexFile(r *os.File, initFn func(entryCount int64), fn func(key
|
|||||||
remainingCount := entryCount - nextBatchSize
|
remainingCount := entryCount - nextBatchSize
|
||||||
|
|
||||||
for remainingCount >= 0 {
|
for remainingCount >= 0 {
|
||||||
_, e := r.ReadAt(bytes[:NeedleMapEntrySize*nextBatchSize], NeedleMapEntrySize*remainingCount)
|
n, e := r.ReadAt(bytes[:NeedleMapEntrySize*nextBatchSize], NeedleMapEntrySize*remainingCount)
|
||||||
// glog.V(0).Infoln("file", r.Name(), "readerOffset", NeedleMapEntrySize*remainingCount, "count", count, "e", e)
|
// glog.V(0).Infoln("file", r.Name(), "readerOffset", NeedleMapEntrySize*remainingCount, "count", count, "e", e)
|
||||||
|
if e == io.EOF && n == int(NeedleMapEntrySize*nextBatchSize) {
|
||||||
|
e = nil
|
||||||
|
}
|
||||||
if e != nil {
|
if e != nil {
|
||||||
return e
|
return e
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -201,9 +201,11 @@ func (s *Store) readOneEcShardInterval(needleId types.NeedleId, ecVolume *erasur
|
|||||||
shardId, actualOffset := interval.ToShardIdAndOffset(erasure_coding.ErasureCodingLargeBlockSize, erasure_coding.ErasureCodingSmallBlockSize)
|
shardId, actualOffset := interval.ToShardIdAndOffset(erasure_coding.ErasureCodingLargeBlockSize, erasure_coding.ErasureCodingSmallBlockSize)
|
||||||
data = make([]byte, interval.Size)
|
data = make([]byte, interval.Size)
|
||||||
if shard, found := ecVolume.FindEcVolumeShard(shardId); found {
|
if shard, found := ecVolume.FindEcVolumeShard(shardId); found {
|
||||||
if _, err = shard.ReadAt(data, actualOffset); err != nil {
|
if n, err = shard.ReadAt(data, actualOffset); err != nil {
|
||||||
glog.V(0).Infof("read local ec shard %d.%d offset %d: %v", ecVolume.VolumeId, shardId, actualOffset, err)
|
if n != interval.Size {
|
||||||
return
|
glog.V(0).Infof("read local ec shard %d.%d offset %d: %v", ecVolume.VolumeId, shardId, actualOffset, err)
|
||||||
|
return
|
||||||
|
}
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
ecVolume.ShardLocationsLock.RLock()
|
ecVolume.ShardLocationsLock.RLock()
|
||||||
|
|||||||
@@ -15,9 +15,11 @@ import (
|
|||||||
func ReadSuperBlock(datBackend backend.BackendStorageFile) (superBlock SuperBlock, err error) {
|
func ReadSuperBlock(datBackend backend.BackendStorageFile) (superBlock SuperBlock, err error) {
|
||||||
|
|
||||||
header := make([]byte, SuperBlockSize)
|
header := make([]byte, SuperBlockSize)
|
||||||
if _, e := datBackend.ReadAt(header, 0); e != nil {
|
if n, e := datBackend.ReadAt(header, 0); e != nil {
|
||||||
err = fmt.Errorf("cannot read volume %s super block: %v", datBackend.Name(), e)
|
if n != SuperBlockSize {
|
||||||
return
|
err = fmt.Errorf("cannot read volume %s super block: %v", datBackend.Name(), e)
|
||||||
|
return
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
superBlock.Version = needle.Version(header[0])
|
superBlock.Version = needle.Version(header[0])
|
||||||
|
|||||||
@@ -80,7 +80,11 @@ func readIndexEntryAtOffset(indexFile *os.File, offset int64) (bytes []byte, err
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
bytes = make([]byte, NeedleMapEntrySize)
|
bytes = make([]byte, NeedleMapEntrySize)
|
||||||
_, err = indexFile.ReadAt(bytes, offset)
|
var readCount int
|
||||||
|
readCount, err = indexFile.ReadAt(bytes, offset)
|
||||||
|
if err == io.EOF && readCount == NeedleMapEntrySize {
|
||||||
|
err = nil
|
||||||
|
}
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -97,7 +101,11 @@ func verifyNeedleIntegrity(datFile backend.BackendStorageFile, v needle.Version,
|
|||||||
}
|
}
|
||||||
if v == needle.Version3 {
|
if v == needle.Version3 {
|
||||||
bytes := make([]byte, TimestampSize)
|
bytes := make([]byte, TimestampSize)
|
||||||
_, err = datFile.ReadAt(bytes, offset+NeedleHeaderSize+int64(size)+needle.NeedleChecksumSize)
|
var readCount int
|
||||||
|
readCount, err = datFile.ReadAt(bytes, offset+NeedleHeaderSize+int64(size)+needle.NeedleChecksumSize)
|
||||||
|
if err == io.EOF && readCount == TimestampSize {
|
||||||
|
err = nil
|
||||||
|
}
|
||||||
if err == io.EOF {
|
if err == io.EOF {
|
||||||
return 0, err
|
return 0, err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -110,8 +110,10 @@ func (v *ChunkCacheVolume) GetNeedle(key types.NeedleId) ([]byte, error) {
|
|||||||
}
|
}
|
||||||
data := make([]byte, nv.Size)
|
data := make([]byte, nv.Size)
|
||||||
if readSize, readErr := v.DataBackend.ReadAt(data, nv.Offset.ToActualOffset()); readErr != nil {
|
if readSize, readErr := v.DataBackend.ReadAt(data, nv.Offset.ToActualOffset()); readErr != nil {
|
||||||
return nil, fmt.Errorf("read %s.dat [%d,%d): %v",
|
if readSize != int(nv.Size) {
|
||||||
v.fileName, nv.Offset.ToActualOffset(), nv.Offset.ToActualOffset()+int64(nv.Size), readErr)
|
return nil, fmt.Errorf("read %s.dat [%d,%d): %v",
|
||||||
|
v.fileName, nv.Offset.ToActualOffset(), nv.Offset.ToActualOffset()+int64(nv.Size), readErr)
|
||||||
|
}
|
||||||
} else {
|
} else {
|
||||||
if readSize != int(nv.Size) {
|
if readSize != int(nv.Size) {
|
||||||
return nil, fmt.Errorf("read %d, expected %d", readSize, nv.Size)
|
return nil, fmt.Errorf("read %d, expected %d", readSize, nv.Size)
|
||||||
@@ -133,8 +135,10 @@ func (v *ChunkCacheVolume) getNeedleSlice(key types.NeedleId, offset, length uin
|
|||||||
}
|
}
|
||||||
data := make([]byte, wanted)
|
data := make([]byte, wanted)
|
||||||
if readSize, readErr := v.DataBackend.ReadAt(data, nv.Offset.ToActualOffset()+int64(offset)); readErr != nil {
|
if readSize, readErr := v.DataBackend.ReadAt(data, nv.Offset.ToActualOffset()+int64(offset)); readErr != nil {
|
||||||
return nil, fmt.Errorf("read %s.dat [%d,%d): %v",
|
if readSize != wanted {
|
||||||
v.fileName, nv.Offset.ToActualOffset()+int64(offset), int(nv.Offset.ToActualOffset())+int(offset)+wanted, readErr)
|
return nil, fmt.Errorf("read %s.dat [%d,%d): %v",
|
||||||
|
v.fileName, nv.Offset.ToActualOffset()+int64(offset), int(nv.Offset.ToActualOffset())+int(offset)+wanted, readErr)
|
||||||
|
}
|
||||||
} else {
|
} else {
|
||||||
if readSize != wanted {
|
if readSize != wanted {
|
||||||
return nil, fmt.Errorf("read %d, expected %d", readSize, wanted)
|
return nil, fmt.Errorf("read %d, expected %d", readSize, wanted)
|
||||||
@@ -155,8 +159,10 @@ func (v *ChunkCacheVolume) readNeedleSliceAt(data []byte, key types.NeedleId, of
|
|||||||
return 0, ErrorOutOfBounds
|
return 0, ErrorOutOfBounds
|
||||||
}
|
}
|
||||||
if n, err = v.DataBackend.ReadAt(data, nv.Offset.ToActualOffset()+int64(offset)); err != nil {
|
if n, err = v.DataBackend.ReadAt(data, nv.Offset.ToActualOffset()+int64(offset)); err != nil {
|
||||||
return n, fmt.Errorf("read %s.dat [%d,%d): %v",
|
if n != wanted {
|
||||||
v.fileName, nv.Offset.ToActualOffset()+int64(offset), int(nv.Offset.ToActualOffset())+int(offset)+wanted, err)
|
return n, fmt.Errorf("read %s.dat [%d,%d): %v",
|
||||||
|
v.fileName, nv.Offset.ToActualOffset()+int64(offset), int(nv.Offset.ToActualOffset())+int(offset)+wanted, err)
|
||||||
|
}
|
||||||
} else {
|
} else {
|
||||||
if n != wanted {
|
if n != wanted {
|
||||||
return n, fmt.Errorf("read %d, expected %d", n, wanted)
|
return n, fmt.Errorf("read %d, expected %d", n, wanted)
|
||||||
|
|||||||
Reference in New Issue
Block a user