gitlab-com-seaweedfs/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-01-19 02:48:24 +00:00
Code Issues Projects Releases Wiki Activity

fix error for deleted files during compaction

Browse source 
deletion during commit may cause trouble when make up the difference during commitCompact()
This commit is contained in:
Chris Lu 2018-06-23 18:24:59 -07:00
parent 3699996985
commit 7f7e4e9885
3 changed files with 112 additions and 94 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
weed/storage/needle_read_write.go
View file

@ -162,7 +162,7 @@ func (n *Needle) ReadData(r *os.File, offset int64, size uint32, version Version
}
n.ParseNeedleHeader(bytes)
if n.Size != size {
return fmt.Errorf("File Entry Not Found. Needle %d Memory %d", n.Size, size)
return fmt.Errorf("File Entry Not Found. Needle id %d expected size %d Memory %d", n.Id, n.Size, size)
}
switch version {
case Version1:

140
weed/storage/volume_vacuum.go
View file

@ -119,7 +119,7 @@ func (v *Volume) makeupDiff(newDatFileName, newIdxFileName, oldDatFileName, oldI
oldDatCompactRevision, err := fetchCompactRevisionFromDatFile(oldDatFile)
if err != nil {
return
return fmt.Errorf("fetchCompactRevisionFromDatFile src %s failed: %v", oldDatFile.Name(), err)
}
if oldDatCompactRevision != v.lastCompactRevision {
return fmt.Errorf("current old dat file's compact revision %d is not the expected one %d", oldDatCompactRevision, v.lastCompactRevision)
@ -137,6 +137,7 @@ func (v *Volume) makeupDiff(newDatFileName, newIdxFileName, oldDatFileName, oldI
return fmt.Errorf("readIndexEntry %s at offset %d failed: %v", oldIdxFileName, idx_offset, err)
}
key, offset, size := idxFileEntry(IdxEntry)
glog.V(0).Infof("key %d offset %d size %d", key, offset, size)
if _, found := incrementedHasUpdatedIndexEntry[key]; !found {
incrementedHasUpdatedIndexEntry[key] = keyField{
offset: offset,
@ -145,77 +146,82 @@ func (v *Volume) makeupDiff(newDatFileName, newIdxFileName, oldDatFileName, oldI
}
}
if len(incrementedHasUpdatedIndexEntry) > 0 {
var (
dst, idx *os.File
)
if dst, err = os.OpenFile(newDatFileName, os.O_RDWR, 0644); err != nil {
// no updates during commit step
if len(incrementedHasUpdatedIndexEntry) == 0 {
return nil
}
// deal with updates during commit step
var (
dst, idx *os.File
)
if dst, err = os.OpenFile(newDatFileName, os.O_RDWR, 0644); err != nil {
return fmt.Errorf("open dat file %s failed: %v", newDatFileName, err)
}
defer dst.Close()
if idx, err = os.OpenFile(newIdxFileName, os.O_RDWR, 0644); err != nil {
return fmt.Errorf("open idx file %s failed: %v", newIdxFileName, err)
}
defer idx.Close()
var newDatCompactRevision uint16
newDatCompactRevision, err = fetchCompactRevisionFromDatFile(dst)
if err != nil {
return fmt.Errorf("fetchCompactRevisionFromDatFile dst %s failed: %v", dst.Name(), err)
}
if oldDatCompactRevision+1 != newDatCompactRevision {
return fmt.Errorf("oldDatFile %s 's compact revision is %d while newDatFile %s 's compact revision is %d", oldDatFileName, oldDatCompactRevision, newDatFileName, newDatCompactRevision)
}
idx_entry_bytes := make([]byte, 16)
for key, incre_idx_entry := range incrementedHasUpdatedIndexEntry {
util.Uint64toBytes(idx_entry_bytes[0:8], key)
util.Uint32toBytes(idx_entry_bytes[8:12], incre_idx_entry.offset)
util.Uint32toBytes(idx_entry_bytes[12:16], incre_idx_entry.size)
var offset int64
if offset, err = dst.Seek(0, 2); err != nil {
glog.V(0).Infof("failed to seek the end of file: %v", err)
return
}
defer dst.Close()
if idx, err = os.OpenFile(newIdxFileName, os.O_RDWR, 0644); err != nil {
return
}
defer idx.Close()
var newDatCompactRevision uint16
newDatCompactRevision, err = fetchCompactRevisionFromDatFile(dst)
if err != nil {
return
}
if oldDatCompactRevision+1 != newDatCompactRevision {
return fmt.Errorf("oldDatFile %s 's compact revision is %d while newDatFile %s 's compact revision is %d", oldDatFileName, oldDatCompactRevision, newDatFileName, newDatCompactRevision)
}
idx_entry_bytes := make([]byte, 16)
for key, incre_idx_entry := range incrementedHasUpdatedIndexEntry {
util.Uint64toBytes(idx_entry_bytes[0:8], key)
util.Uint32toBytes(idx_entry_bytes[8:12], incre_idx_entry.offset)
util.Uint32toBytes(idx_entry_bytes[12:16], incre_idx_entry.size)
var offset int64
if offset, err = dst.Seek(0, 2); err != nil {
glog.V(0).Infof("failed to seek the end of file: %v", err)
//ensure file writing starting from aligned positions
if offset%NeedlePaddingSize != 0 {
offset = offset + (NeedlePaddingSize - offset%NeedlePaddingSize)
if offset, err = v.dataFile.Seek(offset, 0); err != nil {
glog.V(0).Infof("failed to align in datafile %s: %v", v.dataFile.Name(), err)
return
}
//ensure file writing starting from aligned positions
if offset%NeedlePaddingSize != 0 {
offset = offset + (NeedlePaddingSize - offset%NeedlePaddingSize)
if offset, err = v.dataFile.Seek(offset, 0); err != nil {
glog.V(0).Infof("failed to align in datafile %s: %v", v.dataFile.Name(), err)
return
}
}
//updated needle
if incre_idx_entry.offset != 0 && incre_idx_entry.size != 0 {
//even the needle cache in memory is hit, the need_bytes is correct
var needle_bytes []byte
needle_bytes, err = ReadNeedleBlob(oldDatFile, int64(incre_idx_entry.offset)*NeedlePaddingSize, incre_idx_entry.size)
if err != nil {
return
}
dst.Write(needle_bytes)
util.Uint32toBytes(idx_entry_bytes[8:12], uint32(offset/NeedlePaddingSize))
} else { //deleted needle
//fakeDelNeedle 's default Data field is nil
fakeDelNeedle := new(Needle)
fakeDelNeedle.Id = key
fakeDelNeedle.Cookie = 0x12345678
_, _, err = fakeDelNeedle.Append(dst, v.Version())
if err != nil {
return
}
util.Uint32toBytes(idx_entry_bytes[8:12], uint32(0))
}
if _, err := idx.Seek(0, 2); err != nil {
return fmt.Errorf("cannot seek end of indexfile %s: %v",
newIdxFileName, err)
}
_, err = idx.Write(idx_entry_bytes)
}
//updated needle
if incre_idx_entry.offset != 0 && incre_idx_entry.size != 0 && incre_idx_entry.size != TombstoneFileSize {
//even the needle cache in memory is hit, the need_bytes is correct
glog.V(0).Infof("file %d offset %d size %d", key, int64(incre_idx_entry.offset)*NeedlePaddingSize, incre_idx_entry.size)
var needle_bytes []byte
needle_bytes, err = ReadNeedleBlob(oldDatFile, int64(incre_idx_entry.offset)*NeedlePaddingSize, incre_idx_entry.size)
if err != nil {
return fmt.Errorf("ReadNeedleBlob %s key %d offset %d size %d failed: %v", oldDatFile.Name(), key, int64(incre_idx_entry.offset)*NeedlePaddingSize, incre_idx_entry.size, err)
}
dst.Write(needle_bytes)
util.Uint32toBytes(idx_entry_bytes[8:12], uint32(offset/NeedlePaddingSize))
} else { //deleted needle
//fakeDelNeedle 's default Data field is nil
fakeDelNeedle := new(Needle)
fakeDelNeedle.Id = key
fakeDelNeedle.Cookie = 0x12345678
_, _, err = fakeDelNeedle.Append(dst, v.Version())
if err != nil {
return fmt.Errorf("append deleted %d failed: %v", key, err)
}
util.Uint32toBytes(idx_entry_bytes[8:12], uint32(0))
}
if _, err := idx.Seek(0, 2); err != nil {
return fmt.Errorf("cannot seek end of indexfile %s: %v",
newIdxFileName, err)
}
_, err = idx.Write(idx_entry_bytes)
}
return nil

64
weed/storage/volume_vacuum_test.go
View file

@ -69,36 +69,21 @@ func TestCompaction(t *testing.T) {
t.Fatalf("volume creation: %v", err)
}
FILE_COUNT := 234
beforeCommitFileCount := 10
afterCommitFileCount := 10
infos := make([]*needleInfo, FILE_COUNT)
infos := make([]*needleInfo, beforeCommitFileCount+afterCommitFileCount)
for i := 1; i <= FILE_COUNT; i++ {
n := newRandomNeedle(uint64(i))
size, err := v.writeNeedle(n)
if err != nil {
t.Fatalf("write file %d: %v", i, err)
}
infos[i-1] = &needleInfo{
size: size,
crc: n.Checksum,
}
println("written file", i, "checksum", n.Checksum.Value(), "size", size)
if rand.Float64() < 0.5 {
toBeDeleted := rand.Intn(i) + 1
oldNeedle := newEmptyNeedle(uint64(toBeDeleted))
v.deleteNeedle(oldNeedle)
println("deleted file", toBeDeleted)
infos[toBeDeleted-1] = &needleInfo{
size: 0,
crc: n.Checksum,
}
}
for i := 1; i <= beforeCommitFileCount; i++ {
doSomeWritesDeletes(i, v, t, infos)
}
v.Compact(0)
for i := 1; i <= afterCommitFileCount; i++ {
doSomeWritesDeletes(i+beforeCommitFileCount, v, t, infos)
}
v.commitCompact()
v.Close()
@ -108,7 +93,12 @@ func TestCompaction(t *testing.T) {
t.Fatalf("volume reloading: %v", err)
}
for i := 1; i <= FILE_COUNT; i++ {
for i := 1; i <= beforeCommitFileCount+afterCommitFileCount; i++ {
if infos[i-1] == nil {
t.Fatal("not found file", i)
continue
}
if infos[i-1].size == 0 {
continue
@ -129,6 +119,28 @@ func TestCompaction(t *testing.T) {
}
}
func doSomeWritesDeletes(i int, v *Volume, t *testing.T, infos []*needleInfo) {
n := newRandomNeedle(uint64(i))
size, err := v.writeNeedle(n)
if err != nil {
t.Fatalf("write file %d: %v", i, err)
}
infos[i-1] = &needleInfo{
size: size,
crc: n.Checksum,
}
println("written file", i, "checksum", n.Checksum.Value(), "size", size)
if rand.Float64() < 0.5 {
toBeDeleted := rand.Intn(i) + 1
oldNeedle := newEmptyNeedle(uint64(toBeDeleted))
v.deleteNeedle(oldNeedle)
println("deleted file", toBeDeleted)
infos[toBeDeleted-1] = &needleInfo{
size: 0,
crc: n.Checksum,
}
}
}
type needleInfo struct {
size uint32