From f9e8702bb4146045e34643086eaeeae63ce6127c Mon Sep 17 00:00:00 2001 From: zhangsong Date: Sun, 3 May 2020 19:22:45 +0800 Subject: [PATCH 1/4] use async write to persistent file to disk - part1 --- weed/storage/needle/async_request.go | 55 ++++++++ weed/storage/store.go | 13 +- weed/storage/volume.go | 7 +- weed/storage/volume_read_write.go | 180 ++++++++++++++++++++++++++- 4 files changed, 241 insertions(+), 14 deletions(-) create mode 100644 weed/storage/needle/async_request.go diff --git a/weed/storage/needle/async_request.go b/weed/storage/needle/async_request.go new file mode 100644 index 000000000..259419115 --- /dev/null +++ b/weed/storage/needle/async_request.go @@ -0,0 +1,55 @@ +package needle + +type AsyncRequest struct { + N *Needle + IsWriteRequest bool + ActualSize int64 + Fsync bool + offset uint64 + size uint64 + doneChan chan interface{} + isUnchanged bool + err error +} + +func NewAsyncRequest(n *Needle, isWriteRequest bool, fsync bool) *AsyncRequest { + return &AsyncRequest{ + offset: 0, + size: 0, + ActualSize: 0, + doneChan: make(chan interface{}), + N: n, + isUnchanged: false, + IsWriteRequest: isWriteRequest, + Fsync: fsync, + err: nil, + } +} + +func (r *AsyncRequest) WaitComplete() (uint64, uint64, bool, error) { + <-r.doneChan + return r.offset, r.size, r.isUnchanged, r.err +} + +func (r *AsyncRequest) Complete(offset uint64, size uint64, isUnchanged bool, err error) { + r.offset = offset + r.size = size + r.isUnchanged = isUnchanged + r.err = err + close(r.doneChan) +} + +func (r *AsyncRequest) UpdateResult(offset uint64, size uint64, isUnchanged bool, err error) { + r.offset = offset + r.size = size + r.isUnchanged = isUnchanged + r.err = err +} + +func (r *AsyncRequest) Submit() { + close(r.doneChan) +} + +func (r *AsyncRequest) IsSucceed() bool { + return r.err == nil +} diff --git a/weed/storage/store.go b/weed/storage/store.go index 0fff80aa9..14881ffde 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -264,12 +264,7 @@ func (s *Store) WriteVolumeNeedle(i needle.VolumeId, n *needle.Needle, fsync boo err = fmt.Errorf("volume %d is read only", i) return } - // using len(n.Data) here instead of n.Size before n.Size is populated in v.writeNeedle(n) - if MaxPossibleVolumeSize >= v.ContentSize()+uint64(needle.GetActualSize(uint32(len(n.Data)), v.Version())) { - _, _, isUnchanged, err = v.writeNeedle(n, fsync) - } else { - err = fmt.Errorf("volume size limit %d exceeded! current size is %d", s.GetVolumeSizeLimit(), v.ContentSize()) - } + _, _, isUnchanged, err = v.writeNeedle2(n, fsync) return } glog.V(0).Infoln("volume", i, "not found!") @@ -282,11 +277,7 @@ func (s *Store) DeleteVolumeNeedle(i needle.VolumeId, n *needle.Needle) (uint32, if v.noWriteOrDelete { return 0, fmt.Errorf("volume %d is read only", i) } - if MaxPossibleVolumeSize >= v.ContentSize()+uint64(needle.GetActualSize(0, v.Version())) { - return v.deleteNeedle(n) - } else { - return 0, fmt.Errorf("volume size limit %d exceeded! current size is %d", s.GetVolumeSizeLimit(), v.ContentSize()) - } + return v.deleteNeedle2(n) } return 0, fmt.Errorf("volume %d not found on %s:%d", i, s.Ip, s.Port) } diff --git a/weed/storage/volume.go b/weed/storage/volume.go index 755b98b79..df63360a1 100644 --- a/weed/storage/volume.go +++ b/weed/storage/volume.go @@ -33,6 +33,7 @@ type Volume struct { super_block.SuperBlock dataFileAccessLock sync.RWMutex + asyncRequestsChan chan *needle.AsyncRequest lastModifiedTsSeconds uint64 //unix time in seconds lastAppendAtNs uint64 //unix time in nanoseconds @@ -46,12 +47,15 @@ type Volume struct { func NewVolume(dirname string, collection string, id needle.VolumeId, needleMapKind NeedleMapType, replicaPlacement *super_block.ReplicaPlacement, ttl *needle.TTL, preallocate int64, memoryMapMaxSizeMb uint32) (v *Volume, e error) { // if replicaPlacement is nil, the superblock will be loaded from disk - v = &Volume{dir: dirname, Collection: collection, Id: id, MemoryMapMaxSizeMb: memoryMapMaxSizeMb} + v = &Volume{dir: dirname, Collection: collection, Id: id, MemoryMapMaxSizeMb: memoryMapMaxSizeMb, + asyncRequestsChan: make(chan *needle.AsyncRequest, 128)} v.SuperBlock = super_block.SuperBlock{ReplicaPlacement: replicaPlacement, Ttl: ttl} v.needleMapKind = needleMapKind e = v.load(true, true, needleMapKind, preallocate) + v.startWorker() return } + func (v *Volume) String() string { return fmt.Sprintf("Id:%v, dir:%s, Collection:%s, dataFile:%v, nm:%v, noWrite:%v canDelete:%v", v.Id, v.dir, v.Collection, v.DataBackend, v.nm, v.noWriteOrDelete || v.noWriteCanDelete, v.noWriteCanDelete) } @@ -65,6 +69,7 @@ func VolumeFileName(dir string, collection string, id int) (fileName string) { } return } + func (v *Volume) FileName() (fileName string) { return VolumeFileName(v.dir, v.Collection, int(v.Id)) } diff --git a/weed/storage/volume_read_write.go b/weed/storage/volume_read_write.go index bb0421724..6d2899c53 100644 --- a/weed/storage/volume_read_write.go +++ b/weed/storage/volume_read_write.go @@ -46,6 +46,7 @@ func (v *Volume) Destroy() (err error) { err = fmt.Errorf("volume %d is compacting", v.Id) return } + close(v.asyncRequestsChan) storageName, storageKey := v.RemoteStorageNameKey() if v.HasRemoteFile() && storageName != "" && storageKey != "" { if backendStorage, found := backend.BackendStorages[storageName]; found { @@ -63,7 +64,11 @@ func (v *Volume) Destroy() (err error) { return } -func (v *Volume) writeNeedle(n *needle.Needle, fsync bool) (offset uint64, size uint32, isUnchanged bool, err error) { +func (v *Volume) asyncRequestAppend(request *needle.AsyncRequest) { + v.asyncRequestsChan <- request +} + +func (v *Volume) writeNeedleDeprecated(n *needle.Needle, fsync bool) (offset uint64, size uint32, isUnchanged bool, err error) { // glog.V(4).Infof("writing needle %s", needle.NewFileIdFromNeedle(v.Id, n).String()) v.dataFileAccessLock.Lock() defer v.dataFileAccessLock.Unlock() @@ -117,7 +122,66 @@ func (v *Volume) writeNeedle(n *needle.Needle, fsync bool) (offset uint64, size return } -func (v *Volume) deleteNeedle(n *needle.Needle) (uint32, error) { +func (v *Volume) writeNeedle2(n *needle.Needle, fsync bool) (offset uint64, size uint32, isUnchanged bool, err error) { + // glog.V(4).Infof("writing needle %s", needle.NewFileIdFromNeedle(v.Id, n).String()) + if n.Ttl == needle.EMPTY_TTL && v.Ttl != needle.EMPTY_TTL { + n.SetHasTtl() + n.Ttl = v.Ttl + } + + asyncRequest := needle.NewAsyncRequest(n, true, fsync) + // using len(n.Data) here instead of n.Size before n.Size is populated in n.Append() + asyncRequest.ActualSize = needle.GetActualSize(uint32(len(n.Data)), v.Version()) + + v.asyncRequestAppend(asyncRequest) + offset, _, isUnchanged, err = asyncRequest.WaitComplete() + + return +} + +func (v *Volume) doWriteRequest(n *needle.Needle) (offset uint64, size uint32, isUnchanged bool, err error) { + // glog.V(4).Infof("writing needle %s", needle.NewFileIdFromNeedle(v.Id, n).String()) + if v.isFileUnchanged(n) { + size = n.DataSize + isUnchanged = true + return + } + + // check whether existing needle cookie matches + nv, ok := v.nm.Get(n.Id) + if ok { + existingNeedle, _, _, existingNeedleReadErr := needle.ReadNeedleHeader(v.DataBackend, v.Version(), nv.Offset.ToAcutalOffset()) + if existingNeedleReadErr != nil { + err = fmt.Errorf("reading existing needle: %v", existingNeedleReadErr) + return + } + if existingNeedle.Cookie != n.Cookie { + glog.V(0).Infof("write cookie mismatch: existing %x, new %x", existingNeedle.Cookie, n.Cookie) + err = fmt.Errorf("mismatching cookie %x", n.Cookie) + return + } + } + + // append to dat file + n.AppendAtNs = uint64(time.Now().UnixNano()) + if offset, size, _, err = n.Append(v.DataBackend, v.Version()); err != nil { + return + } + v.lastAppendAtNs = n.AppendAtNs + + // add to needle map + if !ok || uint64(nv.Offset.ToAcutalOffset()) < offset { + if err = v.nm.Put(n.Id, ToOffset(int64(offset)), n.Size); err != nil { + glog.V(4).Infof("failed to save in needle map %d: %v", n.Id, err) + } + } + if v.lastModifiedTsSeconds < n.LastModified { + v.lastModifiedTsSeconds = n.LastModified + } + return +} + +func (v *Volume) deleteNeedleDeprecated(n *needle.Needle) (uint32, error) { glog.V(4).Infof("delete needle %s", needle.NewFileIdFromNeedle(v.Id, n).String()) v.dataFileAccessLock.Lock() defer v.dataFileAccessLock.Unlock() @@ -140,6 +204,37 @@ func (v *Volume) deleteNeedle(n *needle.Needle) (uint32, error) { return 0, nil } +func (v *Volume) deleteNeedle2(n *needle.Needle) (uint32, error) { + asyncRequest := needle.NewAsyncRequest(n, false, false) + asyncRequest.ActualSize = needle.GetActualSize(0, v.Version()) + + v.asyncRequestAppend(asyncRequest) + _, size, _, err := asyncRequest.WaitComplete() + + return uint32(size), err +} + +func (v *Volume) doDeleteRequest(n *needle.Needle) (uint32, error) { + glog.V(4).Infof("delete needle %s", needle.NewFileIdFromNeedle(v.Id, n).String()) + nv, ok := v.nm.Get(n.Id) + //fmt.Println("key", n.Id, "volume offset", nv.Offset, "data_size", n.Size, "cached size", nv.Size) + if ok && nv.Size != TombstoneFileSize { + size := nv.Size + n.Data = nil + n.AppendAtNs = uint64(time.Now().UnixNano()) + offset, _, _, err := n.Append(v.DataBackend, v.Version()) + if err != nil { + return size, err + } + v.lastAppendAtNs = n.AppendAtNs + if err = v.nm.Delete(n.Id, ToOffset(int64(offset))); err != nil { + return size, err + } + return size, err + } + return 0, nil +} + // read fills in Needle content by looking up n.Id from NeedleMapper func (v *Volume) readNeedle(n *needle.Needle) (int, error) { v.dataFileAccessLock.RLock() @@ -176,6 +271,87 @@ func (v *Volume) readNeedle(n *needle.Needle) (int, error) { return -1, ErrorNotFound } +func (v *Volume) startWorker() { + go func() { + chanClosed := false + for { + // chan closed. go thread will exit + if chanClosed { + break + } + fsync := false + currentRequests := make([]*needle.AsyncRequest, 0, 128) + currentBytesToWrite := int64(0) + for { + request, ok := <-v.asyncRequestsChan + //volume may be closed + if !ok { + chanClosed = true + break + } + if MaxPossibleVolumeSize < v.ContentSize()+uint64(currentBytesToWrite+request.ActualSize) { + request.Complete(0, 0, false, + fmt.Errorf("volume size limit %d exceeded! current size is %d", MaxPossibleVolumeSize, v.ContentSize())) + break + } + currentRequests = append(currentRequests, request) + currentBytesToWrite += request.ActualSize + if request.Fsync { + fsync = true + } + // submit at most 4M bytes or 128 requests at one time to decrease request delay. + // it also need to break if there is no data in channel to avoid io hang. + if currentBytesToWrite >= 4*1024*1024 || len(currentRequests) >= 128 || len(v.asyncRequestsChan) == 0 { + break + } + } + if len(currentRequests) == 0 { + continue + } + v.dataFileAccessLock.Lock() + end, _, e := v.DataBackend.GetStat() + if e != nil { + for i := 0; i < len(currentRequests); i++ { + currentRequests[i].Complete(0, 0, false, + fmt.Errorf("cannot read current volume position: %v", e)) + } + v.dataFileAccessLock.Unlock() + continue + } + + for i := 0; i < len(currentRequests); i++ { + if currentRequests[i].IsWriteRequest { + offset, size, isUnchanged, err := v.doWriteRequest(currentRequests[i].N) + currentRequests[i].UpdateResult(offset, uint64(size), isUnchanged, err) + } else { + size, err := v.doDeleteRequest(currentRequests[i].N) + currentRequests[i].UpdateResult(0, uint64(size), false, err) + } + } + + if fsync { + // if sync error, data is not reliable, we should mark the completed request as fail and rollback + if err := v.DataBackend.Sync(); err != nil { + // todo: this may generate dirty data or cause data inconsistent, may be weed need to panic? + if te := v.DataBackend.Truncate(end); te != nil { + glog.V(0).Infof("Failed to truncate %s back to %d with error: %v", v.DataBackend.Name(), end, te) + } + for i := 0; i < len(currentRequests); i++ { + if currentRequests[i].IsSucceed() { + currentRequests[i].UpdateResult(0, 0, false, err) + } + } + } + } + + for i := 0; i < len(currentRequests); i++ { + currentRequests[i].Submit() + } + v.dataFileAccessLock.Unlock() + } + }() +} + type VolumeFileScanner interface { VisitSuperBlock(super_block.SuperBlock) error ReadNeedleBody() bool From afb6a1dbb46a2a8793c8faa0450a2609f8e01af3 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 4 May 2020 20:34:26 -0700 Subject: [PATCH 2/4] Revert "Merge pull request #1299 from song-zhang/master" This reverts commit 9016fa19ba4507bdf9ac0addc6d79f147c00302a, reversing changes made to 47234760f40e4d2cea87b4a83d2178b8181598f5. --- weed/storage/needle/async_request.go | 55 -------- weed/storage/store.go | 13 +- weed/storage/volume.go | 7 +- weed/storage/volume_read_write.go | 180 +-------------------------- 4 files changed, 14 insertions(+), 241 deletions(-) delete mode 100644 weed/storage/needle/async_request.go diff --git a/weed/storage/needle/async_request.go b/weed/storage/needle/async_request.go deleted file mode 100644 index 259419115..000000000 --- a/weed/storage/needle/async_request.go +++ /dev/null @@ -1,55 +0,0 @@ -package needle - -type AsyncRequest struct { - N *Needle - IsWriteRequest bool - ActualSize int64 - Fsync bool - offset uint64 - size uint64 - doneChan chan interface{} - isUnchanged bool - err error -} - -func NewAsyncRequest(n *Needle, isWriteRequest bool, fsync bool) *AsyncRequest { - return &AsyncRequest{ - offset: 0, - size: 0, - ActualSize: 0, - doneChan: make(chan interface{}), - N: n, - isUnchanged: false, - IsWriteRequest: isWriteRequest, - Fsync: fsync, - err: nil, - } -} - -func (r *AsyncRequest) WaitComplete() (uint64, uint64, bool, error) { - <-r.doneChan - return r.offset, r.size, r.isUnchanged, r.err -} - -func (r *AsyncRequest) Complete(offset uint64, size uint64, isUnchanged bool, err error) { - r.offset = offset - r.size = size - r.isUnchanged = isUnchanged - r.err = err - close(r.doneChan) -} - -func (r *AsyncRequest) UpdateResult(offset uint64, size uint64, isUnchanged bool, err error) { - r.offset = offset - r.size = size - r.isUnchanged = isUnchanged - r.err = err -} - -func (r *AsyncRequest) Submit() { - close(r.doneChan) -} - -func (r *AsyncRequest) IsSucceed() bool { - return r.err == nil -} diff --git a/weed/storage/store.go b/weed/storage/store.go index 14881ffde..0fff80aa9 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -264,7 +264,12 @@ func (s *Store) WriteVolumeNeedle(i needle.VolumeId, n *needle.Needle, fsync boo err = fmt.Errorf("volume %d is read only", i) return } - _, _, isUnchanged, err = v.writeNeedle2(n, fsync) + // using len(n.Data) here instead of n.Size before n.Size is populated in v.writeNeedle(n) + if MaxPossibleVolumeSize >= v.ContentSize()+uint64(needle.GetActualSize(uint32(len(n.Data)), v.Version())) { + _, _, isUnchanged, err = v.writeNeedle(n, fsync) + } else { + err = fmt.Errorf("volume size limit %d exceeded! current size is %d", s.GetVolumeSizeLimit(), v.ContentSize()) + } return } glog.V(0).Infoln("volume", i, "not found!") @@ -277,7 +282,11 @@ func (s *Store) DeleteVolumeNeedle(i needle.VolumeId, n *needle.Needle) (uint32, if v.noWriteOrDelete { return 0, fmt.Errorf("volume %d is read only", i) } - return v.deleteNeedle2(n) + if MaxPossibleVolumeSize >= v.ContentSize()+uint64(needle.GetActualSize(0, v.Version())) { + return v.deleteNeedle(n) + } else { + return 0, fmt.Errorf("volume size limit %d exceeded! current size is %d", s.GetVolumeSizeLimit(), v.ContentSize()) + } } return 0, fmt.Errorf("volume %d not found on %s:%d", i, s.Ip, s.Port) } diff --git a/weed/storage/volume.go b/weed/storage/volume.go index df63360a1..755b98b79 100644 --- a/weed/storage/volume.go +++ b/weed/storage/volume.go @@ -33,7 +33,6 @@ type Volume struct { super_block.SuperBlock dataFileAccessLock sync.RWMutex - asyncRequestsChan chan *needle.AsyncRequest lastModifiedTsSeconds uint64 //unix time in seconds lastAppendAtNs uint64 //unix time in nanoseconds @@ -47,15 +46,12 @@ type Volume struct { func NewVolume(dirname string, collection string, id needle.VolumeId, needleMapKind NeedleMapType, replicaPlacement *super_block.ReplicaPlacement, ttl *needle.TTL, preallocate int64, memoryMapMaxSizeMb uint32) (v *Volume, e error) { // if replicaPlacement is nil, the superblock will be loaded from disk - v = &Volume{dir: dirname, Collection: collection, Id: id, MemoryMapMaxSizeMb: memoryMapMaxSizeMb, - asyncRequestsChan: make(chan *needle.AsyncRequest, 128)} + v = &Volume{dir: dirname, Collection: collection, Id: id, MemoryMapMaxSizeMb: memoryMapMaxSizeMb} v.SuperBlock = super_block.SuperBlock{ReplicaPlacement: replicaPlacement, Ttl: ttl} v.needleMapKind = needleMapKind e = v.load(true, true, needleMapKind, preallocate) - v.startWorker() return } - func (v *Volume) String() string { return fmt.Sprintf("Id:%v, dir:%s, Collection:%s, dataFile:%v, nm:%v, noWrite:%v canDelete:%v", v.Id, v.dir, v.Collection, v.DataBackend, v.nm, v.noWriteOrDelete || v.noWriteCanDelete, v.noWriteCanDelete) } @@ -69,7 +65,6 @@ func VolumeFileName(dir string, collection string, id int) (fileName string) { } return } - func (v *Volume) FileName() (fileName string) { return VolumeFileName(v.dir, v.Collection, int(v.Id)) } diff --git a/weed/storage/volume_read_write.go b/weed/storage/volume_read_write.go index 6d2899c53..bb0421724 100644 --- a/weed/storage/volume_read_write.go +++ b/weed/storage/volume_read_write.go @@ -46,7 +46,6 @@ func (v *Volume) Destroy() (err error) { err = fmt.Errorf("volume %d is compacting", v.Id) return } - close(v.asyncRequestsChan) storageName, storageKey := v.RemoteStorageNameKey() if v.HasRemoteFile() && storageName != "" && storageKey != "" { if backendStorage, found := backend.BackendStorages[storageName]; found { @@ -64,11 +63,7 @@ func (v *Volume) Destroy() (err error) { return } -func (v *Volume) asyncRequestAppend(request *needle.AsyncRequest) { - v.asyncRequestsChan <- request -} - -func (v *Volume) writeNeedleDeprecated(n *needle.Needle, fsync bool) (offset uint64, size uint32, isUnchanged bool, err error) { +func (v *Volume) writeNeedle(n *needle.Needle, fsync bool) (offset uint64, size uint32, isUnchanged bool, err error) { // glog.V(4).Infof("writing needle %s", needle.NewFileIdFromNeedle(v.Id, n).String()) v.dataFileAccessLock.Lock() defer v.dataFileAccessLock.Unlock() @@ -122,66 +117,7 @@ func (v *Volume) writeNeedleDeprecated(n *needle.Needle, fsync bool) (offset uin return } -func (v *Volume) writeNeedle2(n *needle.Needle, fsync bool) (offset uint64, size uint32, isUnchanged bool, err error) { - // glog.V(4).Infof("writing needle %s", needle.NewFileIdFromNeedle(v.Id, n).String()) - if n.Ttl == needle.EMPTY_TTL && v.Ttl != needle.EMPTY_TTL { - n.SetHasTtl() - n.Ttl = v.Ttl - } - - asyncRequest := needle.NewAsyncRequest(n, true, fsync) - // using len(n.Data) here instead of n.Size before n.Size is populated in n.Append() - asyncRequest.ActualSize = needle.GetActualSize(uint32(len(n.Data)), v.Version()) - - v.asyncRequestAppend(asyncRequest) - offset, _, isUnchanged, err = asyncRequest.WaitComplete() - - return -} - -func (v *Volume) doWriteRequest(n *needle.Needle) (offset uint64, size uint32, isUnchanged bool, err error) { - // glog.V(4).Infof("writing needle %s", needle.NewFileIdFromNeedle(v.Id, n).String()) - if v.isFileUnchanged(n) { - size = n.DataSize - isUnchanged = true - return - } - - // check whether existing needle cookie matches - nv, ok := v.nm.Get(n.Id) - if ok { - existingNeedle, _, _, existingNeedleReadErr := needle.ReadNeedleHeader(v.DataBackend, v.Version(), nv.Offset.ToAcutalOffset()) - if existingNeedleReadErr != nil { - err = fmt.Errorf("reading existing needle: %v", existingNeedleReadErr) - return - } - if existingNeedle.Cookie != n.Cookie { - glog.V(0).Infof("write cookie mismatch: existing %x, new %x", existingNeedle.Cookie, n.Cookie) - err = fmt.Errorf("mismatching cookie %x", n.Cookie) - return - } - } - - // append to dat file - n.AppendAtNs = uint64(time.Now().UnixNano()) - if offset, size, _, err = n.Append(v.DataBackend, v.Version()); err != nil { - return - } - v.lastAppendAtNs = n.AppendAtNs - - // add to needle map - if !ok || uint64(nv.Offset.ToAcutalOffset()) < offset { - if err = v.nm.Put(n.Id, ToOffset(int64(offset)), n.Size); err != nil { - glog.V(4).Infof("failed to save in needle map %d: %v", n.Id, err) - } - } - if v.lastModifiedTsSeconds < n.LastModified { - v.lastModifiedTsSeconds = n.LastModified - } - return -} - -func (v *Volume) deleteNeedleDeprecated(n *needle.Needle) (uint32, error) { +func (v *Volume) deleteNeedle(n *needle.Needle) (uint32, error) { glog.V(4).Infof("delete needle %s", needle.NewFileIdFromNeedle(v.Id, n).String()) v.dataFileAccessLock.Lock() defer v.dataFileAccessLock.Unlock() @@ -204,37 +140,6 @@ func (v *Volume) deleteNeedleDeprecated(n *needle.Needle) (uint32, error) { return 0, nil } -func (v *Volume) deleteNeedle2(n *needle.Needle) (uint32, error) { - asyncRequest := needle.NewAsyncRequest(n, false, false) - asyncRequest.ActualSize = needle.GetActualSize(0, v.Version()) - - v.asyncRequestAppend(asyncRequest) - _, size, _, err := asyncRequest.WaitComplete() - - return uint32(size), err -} - -func (v *Volume) doDeleteRequest(n *needle.Needle) (uint32, error) { - glog.V(4).Infof("delete needle %s", needle.NewFileIdFromNeedle(v.Id, n).String()) - nv, ok := v.nm.Get(n.Id) - //fmt.Println("key", n.Id, "volume offset", nv.Offset, "data_size", n.Size, "cached size", nv.Size) - if ok && nv.Size != TombstoneFileSize { - size := nv.Size - n.Data = nil - n.AppendAtNs = uint64(time.Now().UnixNano()) - offset, _, _, err := n.Append(v.DataBackend, v.Version()) - if err != nil { - return size, err - } - v.lastAppendAtNs = n.AppendAtNs - if err = v.nm.Delete(n.Id, ToOffset(int64(offset))); err != nil { - return size, err - } - return size, err - } - return 0, nil -} - // read fills in Needle content by looking up n.Id from NeedleMapper func (v *Volume) readNeedle(n *needle.Needle) (int, error) { v.dataFileAccessLock.RLock() @@ -271,87 +176,6 @@ func (v *Volume) readNeedle(n *needle.Needle) (int, error) { return -1, ErrorNotFound } -func (v *Volume) startWorker() { - go func() { - chanClosed := false - for { - // chan closed. go thread will exit - if chanClosed { - break - } - fsync := false - currentRequests := make([]*needle.AsyncRequest, 0, 128) - currentBytesToWrite := int64(0) - for { - request, ok := <-v.asyncRequestsChan - //volume may be closed - if !ok { - chanClosed = true - break - } - if MaxPossibleVolumeSize < v.ContentSize()+uint64(currentBytesToWrite+request.ActualSize) { - request.Complete(0, 0, false, - fmt.Errorf("volume size limit %d exceeded! current size is %d", MaxPossibleVolumeSize, v.ContentSize())) - break - } - currentRequests = append(currentRequests, request) - currentBytesToWrite += request.ActualSize - if request.Fsync { - fsync = true - } - // submit at most 4M bytes or 128 requests at one time to decrease request delay. - // it also need to break if there is no data in channel to avoid io hang. - if currentBytesToWrite >= 4*1024*1024 || len(currentRequests) >= 128 || len(v.asyncRequestsChan) == 0 { - break - } - } - if len(currentRequests) == 0 { - continue - } - v.dataFileAccessLock.Lock() - end, _, e := v.DataBackend.GetStat() - if e != nil { - for i := 0; i < len(currentRequests); i++ { - currentRequests[i].Complete(0, 0, false, - fmt.Errorf("cannot read current volume position: %v", e)) - } - v.dataFileAccessLock.Unlock() - continue - } - - for i := 0; i < len(currentRequests); i++ { - if currentRequests[i].IsWriteRequest { - offset, size, isUnchanged, err := v.doWriteRequest(currentRequests[i].N) - currentRequests[i].UpdateResult(offset, uint64(size), isUnchanged, err) - } else { - size, err := v.doDeleteRequest(currentRequests[i].N) - currentRequests[i].UpdateResult(0, uint64(size), false, err) - } - } - - if fsync { - // if sync error, data is not reliable, we should mark the completed request as fail and rollback - if err := v.DataBackend.Sync(); err != nil { - // todo: this may generate dirty data or cause data inconsistent, may be weed need to panic? - if te := v.DataBackend.Truncate(end); te != nil { - glog.V(0).Infof("Failed to truncate %s back to %d with error: %v", v.DataBackend.Name(), end, te) - } - for i := 0; i < len(currentRequests); i++ { - if currentRequests[i].IsSucceed() { - currentRequests[i].UpdateResult(0, 0, false, err) - } - } - } - } - - for i := 0; i < len(currentRequests); i++ { - currentRequests[i].Submit() - } - v.dataFileAccessLock.Unlock() - } - }() -} - type VolumeFileScanner interface { VisitSuperBlock(super_block.SuperBlock) error ReadNeedleBody() bool From 3703c7e1765cff23a11a53e7ba99f3faaa24b878 Mon Sep 17 00:00:00 2001 From: zhangsong Date: Wed, 6 May 2020 21:10:39 +0800 Subject: [PATCH 3/4] write request keep compatible with before --- weed/storage/needle/async_request.go | 4 +- weed/storage/volume_read_write.go | 86 ++++++++++++++++------------ 2 files changed, 49 insertions(+), 41 deletions(-) diff --git a/weed/storage/needle/async_request.go b/weed/storage/needle/async_request.go index 259419115..ea02c55c5 100644 --- a/weed/storage/needle/async_request.go +++ b/weed/storage/needle/async_request.go @@ -4,7 +4,6 @@ type AsyncRequest struct { N *Needle IsWriteRequest bool ActualSize int64 - Fsync bool offset uint64 size uint64 doneChan chan interface{} @@ -12,7 +11,7 @@ type AsyncRequest struct { err error } -func NewAsyncRequest(n *Needle, isWriteRequest bool, fsync bool) *AsyncRequest { +func NewAsyncRequest(n *Needle, isWriteRequest bool) *AsyncRequest { return &AsyncRequest{ offset: 0, size: 0, @@ -21,7 +20,6 @@ func NewAsyncRequest(n *Needle, isWriteRequest bool, fsync bool) *AsyncRequest { N: n, isUnchanged: false, IsWriteRequest: isWriteRequest, - Fsync: fsync, err: nil, } } diff --git a/weed/storage/volume_read_write.go b/weed/storage/volume_read_write.go index 6d2899c53..dce800242 100644 --- a/weed/storage/volume_read_write.go +++ b/weed/storage/volume_read_write.go @@ -68,21 +68,23 @@ func (v *Volume) asyncRequestAppend(request *needle.AsyncRequest) { v.asyncRequestsChan <- request } -func (v *Volume) writeNeedleDeprecated(n *needle.Needle, fsync bool) (offset uint64, size uint32, isUnchanged bool, err error) { +func (v *Volume) syncWrite(n *needle.Needle) (offset uint64, size uint32, isUnchanged bool, err error) { // glog.V(4).Infof("writing needle %s", needle.NewFileIdFromNeedle(v.Id, n).String()) + actualSize := needle.GetActualSize(uint32(len(n.Data)), v.Version()) + v.dataFileAccessLock.Lock() defer v.dataFileAccessLock.Unlock() + + if MaxPossibleVolumeSize < v.nm.ContentSize()+uint64(actualSize) { + err = fmt.Errorf("volume size limit %d exceeded! current size is %d", MaxPossibleVolumeSize, v.ContentSize()) + return + } if v.isFileUnchanged(n) { size = n.DataSize isUnchanged = true return } - if n.Ttl == needle.EMPTY_TTL && v.Ttl != needle.EMPTY_TTL { - n.SetHasTtl() - n.Ttl = v.Ttl - } - // check whether existing needle cookie matches nv, ok := v.nm.Get(n.Id) if ok { @@ -103,11 +105,7 @@ func (v *Volume) writeNeedleDeprecated(n *needle.Needle, fsync bool) (offset uin if offset, size, _, err = n.Append(v.DataBackend, v.Version()); err != nil { return } - if fsync { - if err = v.DataBackend.Sync(); err != nil { - return - } - } + v.lastAppendAtNs = n.AppendAtNs // add to needle map @@ -129,14 +127,18 @@ func (v *Volume) writeNeedle2(n *needle.Needle, fsync bool) (offset uint64, size n.Ttl = v.Ttl } - asyncRequest := needle.NewAsyncRequest(n, true, fsync) - // using len(n.Data) here instead of n.Size before n.Size is populated in n.Append() - asyncRequest.ActualSize = needle.GetActualSize(uint32(len(n.Data)), v.Version()) + if !fsync { + return v.syncWrite(n) + } else { + asyncRequest := needle.NewAsyncRequest(n, true) + // using len(n.Data) here instead of n.Size before n.Size is populated in n.Append() + asyncRequest.ActualSize = needle.GetActualSize(uint32(len(n.Data)), v.Version()) - v.asyncRequestAppend(asyncRequest) - offset, _, isUnchanged, err = asyncRequest.WaitComplete() + v.asyncRequestAppend(asyncRequest) + offset, _, isUnchanged, err = asyncRequest.WaitComplete() - return + return + } } func (v *Volume) doWriteRequest(n *needle.Needle) (offset uint64, size uint32, isUnchanged bool, err error) { @@ -181,10 +183,17 @@ func (v *Volume) doWriteRequest(n *needle.Needle) (offset uint64, size uint32, i return } -func (v *Volume) deleteNeedleDeprecated(n *needle.Needle) (uint32, error) { +func (v *Volume) syncDelete(n *needle.Needle) (uint32, error) { glog.V(4).Infof("delete needle %s", needle.NewFileIdFromNeedle(v.Id, n).String()) + actualSize := needle.GetActualSize(0, v.Version()) v.dataFileAccessLock.Lock() defer v.dataFileAccessLock.Unlock() + + if MaxPossibleVolumeSize < v.nm.ContentSize()+uint64(actualSize) { + err := fmt.Errorf("volume size limit %d exceeded! current size is %d", MaxPossibleVolumeSize, v.ContentSize()) + return 0, err + } + nv, ok := v.nm.Get(n.Id) //fmt.Println("key", n.Id, "volume offset", nv.Offset, "data_size", n.Size, "cached size", nv.Size) if ok && nv.Size != TombstoneFileSize { @@ -205,13 +214,20 @@ func (v *Volume) deleteNeedleDeprecated(n *needle.Needle) (uint32, error) { } func (v *Volume) deleteNeedle2(n *needle.Needle) (uint32, error) { - asyncRequest := needle.NewAsyncRequest(n, false, false) - asyncRequest.ActualSize = needle.GetActualSize(0, v.Version()) + // todo: delete info is always appended no fsync, it may need fsync in future + fsync := false - v.asyncRequestAppend(asyncRequest) - _, size, _, err := asyncRequest.WaitComplete() + if !fsync { + return v.syncDelete(n) + } else { + asyncRequest := needle.NewAsyncRequest(n, false) + asyncRequest.ActualSize = needle.GetActualSize(0, v.Version()) - return uint32(size), err + v.asyncRequestAppend(asyncRequest) + _, size, _, err := asyncRequest.WaitComplete() + + return uint32(size), err + } } func (v *Volume) doDeleteRequest(n *needle.Needle) (uint32, error) { @@ -279,7 +295,6 @@ func (v *Volume) startWorker() { if chanClosed { break } - fsync := false currentRequests := make([]*needle.AsyncRequest, 0, 128) currentBytesToWrite := int64(0) for { @@ -296,9 +311,6 @@ func (v *Volume) startWorker() { } currentRequests = append(currentRequests, request) currentBytesToWrite += request.ActualSize - if request.Fsync { - fsync = true - } // submit at most 4M bytes or 128 requests at one time to decrease request delay. // it also need to break if there is no data in channel to avoid io hang. if currentBytesToWrite >= 4*1024*1024 || len(currentRequests) >= 128 || len(v.asyncRequestsChan) == 0 { @@ -329,17 +341,15 @@ func (v *Volume) startWorker() { } } - if fsync { - // if sync error, data is not reliable, we should mark the completed request as fail and rollback - if err := v.DataBackend.Sync(); err != nil { - // todo: this may generate dirty data or cause data inconsistent, may be weed need to panic? - if te := v.DataBackend.Truncate(end); te != nil { - glog.V(0).Infof("Failed to truncate %s back to %d with error: %v", v.DataBackend.Name(), end, te) - } - for i := 0; i < len(currentRequests); i++ { - if currentRequests[i].IsSucceed() { - currentRequests[i].UpdateResult(0, 0, false, err) - } + // if sync error, data is not reliable, we should mark the completed request as fail and rollback + if err := v.DataBackend.Sync(); err != nil { + // todo: this may generate dirty data or cause data inconsistent, may be weed need to panic? + if te := v.DataBackend.Truncate(end); te != nil { + glog.V(0).Infof("Failed to truncate %s back to %d with error: %v", v.DataBackend.Name(), end, te) + } + for i := 0; i < len(currentRequests); i++ { + if currentRequests[i].IsSucceed() { + currentRequests[i].UpdateResult(0, 0, false, err) } } } From 5568395eddfadef6af8dbcca6bb4839020f3445b Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 6 May 2020 15:37:17 -0700 Subject: [PATCH 4/4] Revert "Revert "Merge pull request #1299 from song-zhang/master"" This reverts commit afb6a1dbb46a2a8793c8faa0450a2609f8e01af3. --- weed/storage/needle/async_request.go | 55 ++++++++ weed/storage/store.go | 13 +- weed/storage/volume.go | 7 +- weed/storage/volume_read_write.go | 180 ++++++++++++++++++++++++++- 4 files changed, 241 insertions(+), 14 deletions(-) create mode 100644 weed/storage/needle/async_request.go diff --git a/weed/storage/needle/async_request.go b/weed/storage/needle/async_request.go new file mode 100644 index 000000000..259419115 --- /dev/null +++ b/weed/storage/needle/async_request.go @@ -0,0 +1,55 @@ +package needle + +type AsyncRequest struct { + N *Needle + IsWriteRequest bool + ActualSize int64 + Fsync bool + offset uint64 + size uint64 + doneChan chan interface{} + isUnchanged bool + err error +} + +func NewAsyncRequest(n *Needle, isWriteRequest bool, fsync bool) *AsyncRequest { + return &AsyncRequest{ + offset: 0, + size: 0, + ActualSize: 0, + doneChan: make(chan interface{}), + N: n, + isUnchanged: false, + IsWriteRequest: isWriteRequest, + Fsync: fsync, + err: nil, + } +} + +func (r *AsyncRequest) WaitComplete() (uint64, uint64, bool, error) { + <-r.doneChan + return r.offset, r.size, r.isUnchanged, r.err +} + +func (r *AsyncRequest) Complete(offset uint64, size uint64, isUnchanged bool, err error) { + r.offset = offset + r.size = size + r.isUnchanged = isUnchanged + r.err = err + close(r.doneChan) +} + +func (r *AsyncRequest) UpdateResult(offset uint64, size uint64, isUnchanged bool, err error) { + r.offset = offset + r.size = size + r.isUnchanged = isUnchanged + r.err = err +} + +func (r *AsyncRequest) Submit() { + close(r.doneChan) +} + +func (r *AsyncRequest) IsSucceed() bool { + return r.err == nil +} diff --git a/weed/storage/store.go b/weed/storage/store.go index 0fff80aa9..14881ffde 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -264,12 +264,7 @@ func (s *Store) WriteVolumeNeedle(i needle.VolumeId, n *needle.Needle, fsync boo err = fmt.Errorf("volume %d is read only", i) return } - // using len(n.Data) here instead of n.Size before n.Size is populated in v.writeNeedle(n) - if MaxPossibleVolumeSize >= v.ContentSize()+uint64(needle.GetActualSize(uint32(len(n.Data)), v.Version())) { - _, _, isUnchanged, err = v.writeNeedle(n, fsync) - } else { - err = fmt.Errorf("volume size limit %d exceeded! current size is %d", s.GetVolumeSizeLimit(), v.ContentSize()) - } + _, _, isUnchanged, err = v.writeNeedle2(n, fsync) return } glog.V(0).Infoln("volume", i, "not found!") @@ -282,11 +277,7 @@ func (s *Store) DeleteVolumeNeedle(i needle.VolumeId, n *needle.Needle) (uint32, if v.noWriteOrDelete { return 0, fmt.Errorf("volume %d is read only", i) } - if MaxPossibleVolumeSize >= v.ContentSize()+uint64(needle.GetActualSize(0, v.Version())) { - return v.deleteNeedle(n) - } else { - return 0, fmt.Errorf("volume size limit %d exceeded! current size is %d", s.GetVolumeSizeLimit(), v.ContentSize()) - } + return v.deleteNeedle2(n) } return 0, fmt.Errorf("volume %d not found on %s:%d", i, s.Ip, s.Port) } diff --git a/weed/storage/volume.go b/weed/storage/volume.go index 755b98b79..df63360a1 100644 --- a/weed/storage/volume.go +++ b/weed/storage/volume.go @@ -33,6 +33,7 @@ type Volume struct { super_block.SuperBlock dataFileAccessLock sync.RWMutex + asyncRequestsChan chan *needle.AsyncRequest lastModifiedTsSeconds uint64 //unix time in seconds lastAppendAtNs uint64 //unix time in nanoseconds @@ -46,12 +47,15 @@ type Volume struct { func NewVolume(dirname string, collection string, id needle.VolumeId, needleMapKind NeedleMapType, replicaPlacement *super_block.ReplicaPlacement, ttl *needle.TTL, preallocate int64, memoryMapMaxSizeMb uint32) (v *Volume, e error) { // if replicaPlacement is nil, the superblock will be loaded from disk - v = &Volume{dir: dirname, Collection: collection, Id: id, MemoryMapMaxSizeMb: memoryMapMaxSizeMb} + v = &Volume{dir: dirname, Collection: collection, Id: id, MemoryMapMaxSizeMb: memoryMapMaxSizeMb, + asyncRequestsChan: make(chan *needle.AsyncRequest, 128)} v.SuperBlock = super_block.SuperBlock{ReplicaPlacement: replicaPlacement, Ttl: ttl} v.needleMapKind = needleMapKind e = v.load(true, true, needleMapKind, preallocate) + v.startWorker() return } + func (v *Volume) String() string { return fmt.Sprintf("Id:%v, dir:%s, Collection:%s, dataFile:%v, nm:%v, noWrite:%v canDelete:%v", v.Id, v.dir, v.Collection, v.DataBackend, v.nm, v.noWriteOrDelete || v.noWriteCanDelete, v.noWriteCanDelete) } @@ -65,6 +69,7 @@ func VolumeFileName(dir string, collection string, id int) (fileName string) { } return } + func (v *Volume) FileName() (fileName string) { return VolumeFileName(v.dir, v.Collection, int(v.Id)) } diff --git a/weed/storage/volume_read_write.go b/weed/storage/volume_read_write.go index bb0421724..6d2899c53 100644 --- a/weed/storage/volume_read_write.go +++ b/weed/storage/volume_read_write.go @@ -46,6 +46,7 @@ func (v *Volume) Destroy() (err error) { err = fmt.Errorf("volume %d is compacting", v.Id) return } + close(v.asyncRequestsChan) storageName, storageKey := v.RemoteStorageNameKey() if v.HasRemoteFile() && storageName != "" && storageKey != "" { if backendStorage, found := backend.BackendStorages[storageName]; found { @@ -63,7 +64,11 @@ func (v *Volume) Destroy() (err error) { return } -func (v *Volume) writeNeedle(n *needle.Needle, fsync bool) (offset uint64, size uint32, isUnchanged bool, err error) { +func (v *Volume) asyncRequestAppend(request *needle.AsyncRequest) { + v.asyncRequestsChan <- request +} + +func (v *Volume) writeNeedleDeprecated(n *needle.Needle, fsync bool) (offset uint64, size uint32, isUnchanged bool, err error) { // glog.V(4).Infof("writing needle %s", needle.NewFileIdFromNeedle(v.Id, n).String()) v.dataFileAccessLock.Lock() defer v.dataFileAccessLock.Unlock() @@ -117,7 +122,66 @@ func (v *Volume) writeNeedle(n *needle.Needle, fsync bool) (offset uint64, size return } -func (v *Volume) deleteNeedle(n *needle.Needle) (uint32, error) { +func (v *Volume) writeNeedle2(n *needle.Needle, fsync bool) (offset uint64, size uint32, isUnchanged bool, err error) { + // glog.V(4).Infof("writing needle %s", needle.NewFileIdFromNeedle(v.Id, n).String()) + if n.Ttl == needle.EMPTY_TTL && v.Ttl != needle.EMPTY_TTL { + n.SetHasTtl() + n.Ttl = v.Ttl + } + + asyncRequest := needle.NewAsyncRequest(n, true, fsync) + // using len(n.Data) here instead of n.Size before n.Size is populated in n.Append() + asyncRequest.ActualSize = needle.GetActualSize(uint32(len(n.Data)), v.Version()) + + v.asyncRequestAppend(asyncRequest) + offset, _, isUnchanged, err = asyncRequest.WaitComplete() + + return +} + +func (v *Volume) doWriteRequest(n *needle.Needle) (offset uint64, size uint32, isUnchanged bool, err error) { + // glog.V(4).Infof("writing needle %s", needle.NewFileIdFromNeedle(v.Id, n).String()) + if v.isFileUnchanged(n) { + size = n.DataSize + isUnchanged = true + return + } + + // check whether existing needle cookie matches + nv, ok := v.nm.Get(n.Id) + if ok { + existingNeedle, _, _, existingNeedleReadErr := needle.ReadNeedleHeader(v.DataBackend, v.Version(), nv.Offset.ToAcutalOffset()) + if existingNeedleReadErr != nil { + err = fmt.Errorf("reading existing needle: %v", existingNeedleReadErr) + return + } + if existingNeedle.Cookie != n.Cookie { + glog.V(0).Infof("write cookie mismatch: existing %x, new %x", existingNeedle.Cookie, n.Cookie) + err = fmt.Errorf("mismatching cookie %x", n.Cookie) + return + } + } + + // append to dat file + n.AppendAtNs = uint64(time.Now().UnixNano()) + if offset, size, _, err = n.Append(v.DataBackend, v.Version()); err != nil { + return + } + v.lastAppendAtNs = n.AppendAtNs + + // add to needle map + if !ok || uint64(nv.Offset.ToAcutalOffset()) < offset { + if err = v.nm.Put(n.Id, ToOffset(int64(offset)), n.Size); err != nil { + glog.V(4).Infof("failed to save in needle map %d: %v", n.Id, err) + } + } + if v.lastModifiedTsSeconds < n.LastModified { + v.lastModifiedTsSeconds = n.LastModified + } + return +} + +func (v *Volume) deleteNeedleDeprecated(n *needle.Needle) (uint32, error) { glog.V(4).Infof("delete needle %s", needle.NewFileIdFromNeedle(v.Id, n).String()) v.dataFileAccessLock.Lock() defer v.dataFileAccessLock.Unlock() @@ -140,6 +204,37 @@ func (v *Volume) deleteNeedle(n *needle.Needle) (uint32, error) { return 0, nil } +func (v *Volume) deleteNeedle2(n *needle.Needle) (uint32, error) { + asyncRequest := needle.NewAsyncRequest(n, false, false) + asyncRequest.ActualSize = needle.GetActualSize(0, v.Version()) + + v.asyncRequestAppend(asyncRequest) + _, size, _, err := asyncRequest.WaitComplete() + + return uint32(size), err +} + +func (v *Volume) doDeleteRequest(n *needle.Needle) (uint32, error) { + glog.V(4).Infof("delete needle %s", needle.NewFileIdFromNeedle(v.Id, n).String()) + nv, ok := v.nm.Get(n.Id) + //fmt.Println("key", n.Id, "volume offset", nv.Offset, "data_size", n.Size, "cached size", nv.Size) + if ok && nv.Size != TombstoneFileSize { + size := nv.Size + n.Data = nil + n.AppendAtNs = uint64(time.Now().UnixNano()) + offset, _, _, err := n.Append(v.DataBackend, v.Version()) + if err != nil { + return size, err + } + v.lastAppendAtNs = n.AppendAtNs + if err = v.nm.Delete(n.Id, ToOffset(int64(offset))); err != nil { + return size, err + } + return size, err + } + return 0, nil +} + // read fills in Needle content by looking up n.Id from NeedleMapper func (v *Volume) readNeedle(n *needle.Needle) (int, error) { v.dataFileAccessLock.RLock() @@ -176,6 +271,87 @@ func (v *Volume) readNeedle(n *needle.Needle) (int, error) { return -1, ErrorNotFound } +func (v *Volume) startWorker() { + go func() { + chanClosed := false + for { + // chan closed. go thread will exit + if chanClosed { + break + } + fsync := false + currentRequests := make([]*needle.AsyncRequest, 0, 128) + currentBytesToWrite := int64(0) + for { + request, ok := <-v.asyncRequestsChan + //volume may be closed + if !ok { + chanClosed = true + break + } + if MaxPossibleVolumeSize < v.ContentSize()+uint64(currentBytesToWrite+request.ActualSize) { + request.Complete(0, 0, false, + fmt.Errorf("volume size limit %d exceeded! current size is %d", MaxPossibleVolumeSize, v.ContentSize())) + break + } + currentRequests = append(currentRequests, request) + currentBytesToWrite += request.ActualSize + if request.Fsync { + fsync = true + } + // submit at most 4M bytes or 128 requests at one time to decrease request delay. + // it also need to break if there is no data in channel to avoid io hang. + if currentBytesToWrite >= 4*1024*1024 || len(currentRequests) >= 128 || len(v.asyncRequestsChan) == 0 { + break + } + } + if len(currentRequests) == 0 { + continue + } + v.dataFileAccessLock.Lock() + end, _, e := v.DataBackend.GetStat() + if e != nil { + for i := 0; i < len(currentRequests); i++ { + currentRequests[i].Complete(0, 0, false, + fmt.Errorf("cannot read current volume position: %v", e)) + } + v.dataFileAccessLock.Unlock() + continue + } + + for i := 0; i < len(currentRequests); i++ { + if currentRequests[i].IsWriteRequest { + offset, size, isUnchanged, err := v.doWriteRequest(currentRequests[i].N) + currentRequests[i].UpdateResult(offset, uint64(size), isUnchanged, err) + } else { + size, err := v.doDeleteRequest(currentRequests[i].N) + currentRequests[i].UpdateResult(0, uint64(size), false, err) + } + } + + if fsync { + // if sync error, data is not reliable, we should mark the completed request as fail and rollback + if err := v.DataBackend.Sync(); err != nil { + // todo: this may generate dirty data or cause data inconsistent, may be weed need to panic? + if te := v.DataBackend.Truncate(end); te != nil { + glog.V(0).Infof("Failed to truncate %s back to %d with error: %v", v.DataBackend.Name(), end, te) + } + for i := 0; i < len(currentRequests); i++ { + if currentRequests[i].IsSucceed() { + currentRequests[i].UpdateResult(0, 0, false, err) + } + } + } + } + + for i := 0; i < len(currentRequests); i++ { + currentRequests[i].Submit() + } + v.dataFileAccessLock.Unlock() + } + }() +} + type VolumeFileScanner interface { VisitSuperBlock(super_block.SuperBlock) error ReadNeedleBody() bool