gsh-digital-services/seaweedFS
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

volume: checksum remove the hashing step in Value()

Browse Source
This commit is contained in:
chrislu
2022-06-05 15:24:02 -07:00
parent ecef844dfc
commit 056c480eb0
7 changed files with 26 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
weed/storage/needle/crc.go
View File

@@ -21,6 +21,7 @@ func (c CRC) Update(b []byte) CRC {
return CRC(crc32.Update(uint32(c), table, b)) return CRC(crc32.Update(uint32(c), table, b))
} }
// Value Deprecated. Just use the raw uint32 value to compare.
func (c CRC) Value() uint32 { func (c CRC) Value() uint32 {
return uint32(c>>15|c<<17) + 0xa282ead8 return uint32(c>>15|c<<17) + 0xa282ead8
} }
@@ -51,4 +52,4 @@ func (c *CRCwriter) Write(p []byte) (n int, err error) {
return return
} }
func (c *CRCwriter) Sum() uint32 { return c.crc.Value() } // final hash func (c *CRCwriter) Sum() uint32 { return uint32(c.crc) } // final hash

3
weed/storage/needle/needle_read.go
View File

@@ -72,7 +72,8 @@ func (n *Needle) ReadBytes(bytes []byte, offset int64, size Size, version Versio
if size > 0 { if size > 0 {
checksum := util.BytesToUint32(bytes[NeedleHeaderSize+size : NeedleHeaderSize+size+NeedleChecksumSize]) checksum := util.BytesToUint32(bytes[NeedleHeaderSize+size : NeedleHeaderSize+size+NeedleChecksumSize])
newChecksum := NewCRC(n.Data) newChecksum := NewCRC(n.Data)
if checksum != newChecksum.Value() { if checksum != newChecksum.Value() && checksum != uint32(newChecksum) {
// the crc.Value() function is to be deprecated. this double checking is for backward compatible.
stats.VolumeServerRequestCounter.WithLabelValues(stats.ErrorCRC).Inc() stats.VolumeServerRequestCounter.WithLabelValues(stats.ErrorCRC).Inc()
return errors.New("CRC error! Data On Disk Corrupted") return errors.New("CRC error! Data On Disk Corrupted")
} }

19
weed/storage/needle/needle_read_page.go
View File

@@ -10,7 +10,7 @@ import (
) )
// ReadNeedleDataInto uses a needle without n.Data to read the content into an io.Writer // ReadNeedleDataInto uses a needle without n.Data to read the content into an io.Writer
func (n *Needle) ReadNeedleDataInto(r backend.BackendStorageFile, volumeOffset int64, buf []byte, writer io.Writer, needleOffset int64, size int64, expectedChecksumValue uint32) (err error) { func (n *Needle) ReadNeedleDataInto(r backend.BackendStorageFile, volumeOffset int64, buf []byte, writer io.Writer, needleOffset int64, size int64) (err error) {
crc := CRC(0) crc := CRC(0)
for x := needleOffset; x < needleOffset+size; x += int64(len(buf)) { for x := needleOffset; x < needleOffset+size; x += int64(len(buf)) {
count, err := n.ReadNeedleData(r, volumeOffset, buf, x) count, err := n.ReadNeedleData(r, volumeOffset, buf, x)
@@ -31,8 +31,9 @@ func (n *Needle) ReadNeedleDataInto(r backend.BackendStorageFile, volumeOffset i
break break
} }
} }
if needleOffset == 0 && size == int64(n.DataSize) && expectedChecksumValue != crc.Value() { if needleOffset == 0 && size == int64(n.DataSize) && (n.Checksum != crc && uint32(n.Checksum) != crc.Value()) {
return fmt.Errorf("ReadNeedleData checksum %v expected %v", crc.Value(), expectedChecksumValue) // the crc.Value() function is to be deprecated. this double checking is for backward compatible.
return fmt.Errorf("ReadNeedleData checksum %v expected %v", crc, n.Checksum)
} }
return nil return nil
} }
@@ -58,18 +59,18 @@ func (n *Needle) ReadNeedleData(r backend.BackendStorageFile, volumeOffset int64
} }
// ReadNeedleMeta fills all metadata except the n.Data // ReadNeedleMeta fills all metadata except the n.Data
func (n *Needle) ReadNeedleMeta(r backend.BackendStorageFile, offset int64, size Size, version Version) (checksumValue uint32, err error) { func (n *Needle) ReadNeedleMeta(r backend.BackendStorageFile, offset int64, size Size, version Version) (err error) {
bytes := make([]byte, NeedleHeaderSize+DataSizeSize) bytes := make([]byte, NeedleHeaderSize+DataSizeSize)
count, err := r.ReadAt(bytes, offset) count, err := r.ReadAt(bytes, offset)
if count != NeedleHeaderSize+DataSizeSize || err != nil { if count != NeedleHeaderSize+DataSizeSize || err != nil {
return 0, err return err
} }
n.ParseNeedleHeader(bytes) n.ParseNeedleHeader(bytes)
if n.Size != size { if n.Size != size {
if OffsetSize == 4 && offset < int64(MaxPossibleVolumeSize) { if OffsetSize == 4 && offset < int64(MaxPossibleVolumeSize) {
return 0, ErrorSizeMismatch return ErrorSizeMismatch
} }
} }
@@ -86,18 +87,18 @@ func (n *Needle) ReadNeedleMeta(r backend.BackendStorageFile, offset int64, size
err = nil err = nil
} }
if err != nil { if err != nil {
return 0, err return err
} }
var index int var index int
index, err = n.readNeedleDataVersion2NonData(metaSlice) index, err = n.readNeedleDataVersion2NonData(metaSlice)
checksumValue = util.BytesToUint32(metaSlice[index : index+NeedleChecksumSize]) n.Checksum = CRC(util.BytesToUint32(metaSlice[index : index+NeedleChecksumSize]))
if version == Version3 { if version == Version3 {
n.AppendAtNs = util.BytesToUint64(metaSlice[index+NeedleChecksumSize : index+NeedleChecksumSize+TimestampSize]) n.AppendAtNs = util.BytesToUint64(metaSlice[index+NeedleChecksumSize : index+NeedleChecksumSize+TimestampSize])
} }
return checksumValue, err return err
} }

2
weed/storage/needle/needle_read_test.go
View File

@@ -57,7 +57,7 @@ func TestPageRead(t *testing.T) {
{ {
n := new(Needle) n := new(Needle)
checksumValue, err := n.ReadNeedleMeta(datBackend, offset, size, Version3) err := n.ReadNeedleMeta(datBackend, offset, size, Version3)
if err != nil { if err != nil {
t.Fatalf("ReadNeedleHeader: %v", err) t.Fatalf("ReadNeedleHeader: %v", err)
} }

4
weed/storage/needle/needle_write.go
View File

@@ -31,7 +31,7 @@ func (n *Needle) prepareWriteBuffer(version Version, writeBytes *bytes.Buffer) (
writeBytes.Write(header) writeBytes.Write(header)
writeBytes.Write(n.Data) writeBytes.Write(n.Data)
padding := PaddingLength(n.Size, version) padding := PaddingLength(n.Size, version)
util.Uint32toBytes(header[0:NeedleChecksumSize], n.Checksum.Value()) util.Uint32toBytes(header[0:NeedleChecksumSize], uint32(n.Checksum))
writeBytes.Write(header[0 : NeedleChecksumSize+padding]) writeBytes.Write(header[0 : NeedleChecksumSize+padding])
return size, actualSize, nil return size, actualSize, nil
case Version2, Version3: case Version2, Version3:
@@ -97,7 +97,7 @@ func (n *Needle) prepareWriteBuffer(version Version, writeBytes *bytes.Buffer) (
} }
} }
padding := PaddingLength(n.Size, version) padding := PaddingLength(n.Size, version)
util.Uint32toBytes(header[0:NeedleChecksumSize], n.Checksum.Value()) util.Uint32toBytes(header[0:NeedleChecksumSize], uint32(n.Checksum))
if version == Version2 { if version == Version2 {
writeBytes.Write(header[0 : NeedleChecksumSize+padding]) writeBytes.Write(header[0 : NeedleChecksumSize+padding])
} else { } else {

9
weed/storage/store.go
View File

@@ -27,13 +27,14 @@ const (
) )
type ReadOption struct { type ReadOption struct {
// request
ReadDeleted bool ReadDeleted bool
AttemptMetaOnly bool AttemptMetaOnly bool
MustMetaOnly bool MustMetaOnly bool
IsMetaOnly bool // read status // response
ChecksumValue uint32 // read status IsMetaOnly bool // read status
VolumeRevision uint16 VolumeRevision uint16
IsOutOfRange bool // whether need to read over MaxPossibleVolumeSize IsOutOfRange bool // whether read over MaxPossibleVolumeSize
} }
/* /*

8
weed/storage/volume_read.go
View File

@@ -41,10 +41,10 @@ func (v *Volume) readNeedle(n *needle.Needle, readOption *ReadOption, onReadSize
} }
if readOption != nil && readOption.AttemptMetaOnly && readSize > PagedReadLimit { if readOption != nil && readOption.AttemptMetaOnly && readSize > PagedReadLimit {
readOption.VolumeRevision = v.SuperBlock.CompactionRevision readOption.VolumeRevision = v.SuperBlock.CompactionRevision
readOption.ChecksumValue, err = n.ReadNeedleMeta(v.DataBackend, nv.Offset.ToActualOffset(), readSize, v.Version()) err = n.ReadNeedleMeta(v.DataBackend, nv.Offset.ToActualOffset(), readSize, v.Version())
if err == needle.ErrorSizeMismatch && OffsetSize == 4 { if err == needle.ErrorSizeMismatch && OffsetSize == 4 {
readOption.IsOutOfRange = true readOption.IsOutOfRange = true
readOption.ChecksumValue, err = n.ReadNeedleMeta(v.DataBackend, nv.Offset.ToActualOffset()+int64(MaxPossibleVolumeSize), readSize, v.Version()) err = n.ReadNeedleMeta(v.DataBackend, nv.Offset.ToActualOffset()+int64(MaxPossibleVolumeSize), readSize, v.Version())
} }
if err != nil { if err != nil {
return 0, err return 0, err
@@ -105,7 +105,7 @@ func (v *Volume) readNeedleDataInto(n *needle.Needle, readOption *ReadOption, wr
if readOption.VolumeRevision != v.SuperBlock.CompactionRevision { if readOption.VolumeRevision != v.SuperBlock.CompactionRevision {
// the volume is compacted // the volume is compacted
readOption.IsOutOfRange = false readOption.IsOutOfRange = false
readOption.ChecksumValue, err = n.ReadNeedleMeta(v.DataBackend, nv.Offset.ToActualOffset(), readSize, v.Version()) err = n.ReadNeedleMeta(v.DataBackend, nv.Offset.ToActualOffset(), readSize, v.Version())
} }
buf := mem.Allocate(1024 * 1024) buf := mem.Allocate(1024 * 1024)
defer mem.Free(buf) defer mem.Free(buf)
@@ -114,7 +114,7 @@ func (v *Volume) readNeedleDataInto(n *needle.Needle, readOption *ReadOption, wr
actualOffset += int64(MaxPossibleVolumeSize) actualOffset += int64(MaxPossibleVolumeSize)
} }
return n.ReadNeedleDataInto(v.DataBackend, actualOffset, buf, writer, offset, size, readOption.ChecksumValue) return n.ReadNeedleDataInto(v.DataBackend, actualOffset, buf, writer, offset, size)
} }
// read fills in Needle content by looking up n.Id from NeedleMapper // read fills in Needle content by looking up n.Id from NeedleMapper