mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/chrislusf/seaweedfs synced 2024-07-03 23:56:41 +02:00
Code Issues Projects Releases Wiki Activity

refactor data integrity checking code v2

Browse source
This commit is contained in:
霍晓栋 2016-07-03 14:11:25 +08:00
parent dda13def2a
commit 16f6984d02
3 changed files with 16 additions and 24 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
unmaintained/fix_dat/fix_dat.go
View file

@ -107,7 +107,7 @@ func iterateEntries(datFile, idxFile *os.File, visitNeedle func(n *storage.Needl
fmt.Println("Recovered in f", r)
}
}()
if _, err = n.ReadNeedleBody(datFile, version, offset+int64(storage.NeedleHeaderSize), rest, false); err != nil {
if err = n.ReadNeedleBody(datFile, version, offset+int64(storage.NeedleHeaderSize), rest, false); err != nil {
fmt.Printf("cannot read needle body: offset %d body %d %v\n", offset, rest, err)
}
}()

8
weed/storage/needle_read_write.go
View file

@ -226,20 +226,20 @@ func ReadNeedleHeader(r *os.File, version Version, offset int64) (n *Needle, bod
//n should be a needle already read the header
//the input stream will read until next file entry
func (n *Needle) ReadNeedleBody(r *os.File, version Version, offset int64, bodyLength uint32) (bytes []byte, err error) {
func (n *Needle) ReadNeedleBody(r *os.File, version Version, offset int64, bodyLength uint32) (err error) {
if bodyLength <= 0 {
return
return nil
}
switch version {
case Version1:
bytes = make([]byte, bodyLength)
bytes := make([]byte, bodyLength)
if _, err = r.ReadAt(bytes, offset); err != nil {
return
}
n.Data = bytes[:n.Size]
n.Checksum = NewCRC(n.Data)
case Version2:
bytes = make([]byte, bodyLength)
bytes := make([]byte, bodyLength)
if _, err = r.ReadAt(bytes, offset); err != nil {
return
}

30
weed/storage/volume.go
View file

@ -64,22 +64,14 @@ func readIndexEntryAtOffset(indexFile *os.File, offset int64, v Version) (bytes
_, err = indexFile.ReadAt(bytes, offset)
return
}
func verifyNeedleIntegrity(datFile *os.File, v Version, offset int64, key uint64) error {
if n, bodyLength, err := ReadNeedleHeader(datFile, v, offset); err != nil {
return fmt.Errorf("can not read needle header: %v", err)
} else {
if n.Id != key {
return fmt.Errorf("index key %#x does not match needle's Id %#x", key, n.Id)
} else {
if bytes, err := n.ReadNeedleBody(datFile, v, offset+int64(NeedleHeaderSize), bodyLength); err != nil {
return fmt.Errorf("dat file's body reading failed: %v", err)
} else {
checksum := util.BytesToUint32(bytes[n.Size : n.Size+NeedleChecksumSize])
if n.Checksum.Value() != checksum {
return fmt.Errorf("CRC check failed")
}
}
}
func verifyNeedleIntegrity(datFile *os.File, v Version, offset int64, key uint64, size uint32) error {
n := new(Needle)
err := n.ReadData(datFile, offset, size, v)
if err != nil {
return err
}
if n.Id != key {
return fmt.Errorf("index key %#x does not match needle's Id %#x", key, n.Id)
}
return nil
}
@ -94,8 +86,8 @@ func volumeDataIntegrityChecking(v *Volume, indexFile *os.File) error {
if lastIdxEntry, e = readIndexEntryAtOffset(indexFile, indexSize-NeedleIndexSize, v.Version()); e != nil {
return fmt.Errorf("readLastIndexEntry failed: %v", e)
}
key, offset, _ := idxFileEntry(lastIdxEntry)
if e = verifyNeedleIntegrity(v.dataFile, v.Version(), int64(offset)*NeedlePaddingSize, key); e != nil {
key, offset, size := idxFileEntry(lastIdxEntry)
if e = verifyNeedleIntegrity(v.dataFile, v.Version(), int64(offset)*NeedlePaddingSize, key, size); e != nil {
return fmt.Errorf("verifyNeedleIntegrity failed: %v", e)
}
} else {
@ -397,7 +389,7 @@ func ScanVolumeFile(dirname string, collection string, id VolumeId,
}
for n != nil {
if readNeedleBody {
if _, err = n.ReadNeedleBody(v.dataFile, version, offset+int64(NeedleHeaderSize), rest); err != nil {
if err = n.ReadNeedleBody(v.dataFile, version, offset+int64(NeedleHeaderSize), rest); err != nil {
glog.V(0).Infof("cannot read needle body: %v", err)
//err = fmt.Errorf("cannot read needle body: %v", err)
//return