|
|
@ -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 |
|
|
|