mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2024-01-19 02:48:24 +00:00
adjust hardlink update
simplify logic, pass entity content directly to hard link. The "weed mount" handles the logic to calculate hard link counter.
This commit is contained in:
Chris Lu
parent
5e239afdfc
commit
1295347958
|
|
@ -3,8 +3,6 @@ package filer
|
|||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"github.com/chrislusf/seaweedfs/weed/glog"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
|
|
@ -82,30 +80,9 @@ func (fsw *FilerStoreWrapper) InsertEntry(ctx context.Context, entry *Entry) err
|
|||
entry.Mime = ""
|
||||
}
|
||||
|
||||
if entry.HardLinkId != 0 {
|
||||
// check what is existing entry
|
||||
existingEntry, err := fsw.ActualStore.FindEntry(ctx, entry.FullPath)
|
||||
|
||||
if err == nil && entry.HardLinkId == existingEntry.HardLinkId {
|
||||
// updating the same entry
|
||||
if err := fsw.updateHardLink(ctx, entry); err != nil {
|
||||
if err := fsw.handleUpdateToHardLinks(ctx, entry); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
} else {
|
||||
if err == nil && existingEntry.HardLinkId != 0 {
|
||||
// break away from the old hard link
|
||||
if err := fsw.DeleteHardLink(ctx, entry.HardLinkId); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
// CreateLink 1.2 : update new file to hardlink mode
|
||||
// update one existing hard link, counter ++
|
||||
if err := fsw.increaseHardLink(ctx, entry.HardLinkId); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return fsw.ActualStore.InsertEntry(ctx, entry)
|
||||
}
|
||||
|
|
@ -122,50 +99,8 @@ func (fsw *FilerStoreWrapper) UpdateEntry(ctx context.Context, entry *Entry) err
|
|||
entry.Mime = ""
|
||||
}
|
||||
|
||||
if entry.HardLinkId != 0 {
|
||||
// handle hard link
|
||||
|
||||
// check what is existing entry
|
||||
existingEntry, err := fsw.ActualStore.FindEntry(ctx, entry.FullPath)
|
||||
if err != nil {
|
||||
return fmt.Errorf("update existing entry %s: %v", entry.FullPath, err)
|
||||
}
|
||||
|
||||
err = fsw.maybeReadHardLink(ctx, &Entry{HardLinkId: entry.HardLinkId})
|
||||
if err == ErrKvNotFound {
|
||||
|
||||
// CreateLink 1.1 : split source entry into hardlink+empty_entry
|
||||
|
||||
// create hard link from existing entry, counter ++
|
||||
existingEntry.HardLinkId = entry.HardLinkId
|
||||
if err = fsw.createHardLink(ctx, existingEntry); err != nil {
|
||||
return fmt.Errorf("createHardLink %d: %v", existingEntry.HardLinkId, err)
|
||||
}
|
||||
|
||||
// create the empty entry
|
||||
if err = fsw.ActualStore.UpdateEntry(ctx, &Entry{
|
||||
FullPath: entry.FullPath,
|
||||
HardLinkId: entry.HardLinkId,
|
||||
}); err != nil {
|
||||
return fmt.Errorf("UpdateEntry to link %d: %v", entry.FullPath, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return fmt.Errorf("update entry %s: %v", entry.FullPath, err)
|
||||
}
|
||||
|
||||
if entry.HardLinkId != existingEntry.HardLinkId {
|
||||
// if different hard link id, moving to a new hard link
|
||||
glog.Fatalf("unexpected. update entry to a new link. not implemented yet.")
|
||||
} else {
|
||||
// updating hardlink with new metadata
|
||||
if err = fsw.updateHardLink(ctx, entry); err != nil {
|
||||
return fmt.Errorf("updateHardLink %d from %s: %v", entry.HardLinkId, entry.FullPath, err)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
if err := fsw.handleUpdateToHardLinks(ctx, entry); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return fsw.ActualStore.UpdateEntry(ctx, entry)
|
||||
|
|
@ -318,131 +253,3 @@ func (fsw *FilerStoreWrapper) KvGet(ctx context.Context, key []byte) (value []by
|
|||
func (fsw *FilerStoreWrapper) KvDelete(ctx context.Context, key []byte) (err error) {
|
||||
return fsw.ActualStore.KvDelete(ctx, key)
|
||||
}
|
||||
|
||||
func (fsw *FilerStoreWrapper) createHardLink(ctx context.Context, entry *Entry) error {
|
||||
if entry.HardLinkId == 0 {
|
||||
return nil
|
||||
}
|
||||
key := entry.HardLinkId.Key()
|
||||
|
||||
_, err := fsw.KvGet(ctx, key)
|
||||
if err != ErrKvNotFound {
|
||||
return fmt.Errorf("create hardlink %d: already exists: %v", entry.HardLinkId, err)
|
||||
}
|
||||
|
||||
entry.HardLinkCounter = 1
|
||||
|
||||
newBlob, encodeErr := entry.EncodeAttributesAndChunks()
|
||||
if encodeErr != nil {
|
||||
return encodeErr
|
||||
}
|
||||
|
||||
return fsw.KvPut(ctx, key, newBlob)
|
||||
}
|
||||
|
||||
func (fsw *FilerStoreWrapper) updateHardLink(ctx context.Context, entry *Entry) error {
|
||||
if entry.HardLinkId == 0 {
|
||||
return nil
|
||||
}
|
||||
key := entry.HardLinkId.Key()
|
||||
|
||||
value, err := fsw.KvGet(ctx, key)
|
||||
if err == ErrKvNotFound {
|
||||
return fmt.Errorf("update hardlink %d: missing", entry.HardLinkId)
|
||||
}
|
||||
if err != nil {
|
||||
return fmt.Errorf("update hardlink %d err: %v", entry.HardLinkId, err)
|
||||
}
|
||||
|
||||
existingEntry := &Entry{}
|
||||
if err = existingEntry.DecodeAttributesAndChunks(value); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
entry.HardLinkCounter = existingEntry.HardLinkCounter
|
||||
|
||||
newBlob, encodeErr := entry.EncodeAttributesAndChunks()
|
||||
if encodeErr != nil {
|
||||
return encodeErr
|
||||
}
|
||||
|
||||
return fsw.KvPut(ctx, key, newBlob)
|
||||
}
|
||||
|
||||
func (fsw *FilerStoreWrapper) increaseHardLink(ctx context.Context, hardLinkId HardLinkId) error {
|
||||
if hardLinkId == 0 {
|
||||
return nil
|
||||
}
|
||||
key := hardLinkId.Key()
|
||||
|
||||
value, err := fsw.KvGet(ctx, key)
|
||||
if err == ErrKvNotFound {
|
||||
return fmt.Errorf("increaseHardLink %d: missing", hardLinkId)
|
||||
}
|
||||
if err != nil {
|
||||
return fmt.Errorf("increaseHardLink %d err: %v", hardLinkId, err)
|
||||
}
|
||||
|
||||
existingEntry := &Entry{}
|
||||
if err = existingEntry.DecodeAttributesAndChunks(value); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
existingEntry.HardLinkCounter++
|
||||
|
||||
newBlob, encodeErr := existingEntry.EncodeAttributesAndChunks()
|
||||
if encodeErr != nil {
|
||||
return encodeErr
|
||||
}
|
||||
|
||||
return fsw.KvPut(ctx, key, newBlob)
|
||||
}
|
||||
|
||||
func (fsw *FilerStoreWrapper) maybeReadHardLink(ctx context.Context, entry *Entry) error {
|
||||
if entry.HardLinkId == 0 {
|
||||
return nil
|
||||
}
|
||||
key := entry.HardLinkId.Key()
|
||||
|
||||
value, err := fsw.KvGet(ctx, key)
|
||||
if err != nil {
|
||||
glog.Errorf("read %s hardlink %d: %v", entry.FullPath, entry.HardLinkId, err)
|
||||
return err
|
||||
}
|
||||
|
||||
if err = entry.DecodeAttributesAndChunks(value); err != nil {
|
||||
glog.Errorf("decode %s hardlink %d: %v", entry.FullPath, entry.HardLinkId, err)
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (fsw *FilerStoreWrapper) DeleteHardLink(ctx context.Context, hardLinkId HardLinkId) error {
|
||||
key := hardLinkId.Key()
|
||||
value, err := fsw.KvGet(ctx, key)
|
||||
if err == ErrKvNotFound {
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
entry := &Entry{}
|
||||
if err = entry.DecodeAttributesAndChunks(value); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
entry.HardLinkCounter--
|
||||
if entry.HardLinkCounter <= 0 {
|
||||
return fsw.KvDelete(ctx, key)
|
||||
}
|
||||
|
||||
newBlob, encodeErr := entry.EncodeAttributesAndChunks()
|
||||
if encodeErr != nil {
|
||||
return encodeErr
|
||||
}
|
||||
|
||||
return fsw.KvPut(ctx, key, newBlob)
|
||||
|
||||
}
|
||||
|
|
|
|||
95
weed/filer/filerstore_hardlink.go
Normal file
95
weed/filer/filerstore_hardlink.go
Normal file
|
|
@ -0,0 +1,95 @@
|
|||
package filer
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"github.com/chrislusf/seaweedfs/weed/glog"
|
||||
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
|
||||
)
|
||||
|
||||
func (fsw *FilerStoreWrapper) handleUpdateToHardLinks(ctx context.Context, entry *Entry) error {
|
||||
if entry.HardLinkId == 0 {
|
||||
return nil
|
||||
}
|
||||
// handle hard links
|
||||
if err := fsw.setHardLink(ctx, entry); err != nil {
|
||||
return fmt.Errorf("setHardLink %d: %v", entry.HardLinkId, err)
|
||||
}
|
||||
|
||||
// check what is existing entry
|
||||
existingEntry, err := fsw.ActualStore.FindEntry(ctx, entry.FullPath)
|
||||
if err != nil && err != filer_pb.ErrNotFound {
|
||||
return fmt.Errorf("update existing entry %s: %v", entry.FullPath, err)
|
||||
}
|
||||
|
||||
// remove old hard link
|
||||
if err == nil && existingEntry.HardLinkId != entry.HardLinkId {
|
||||
if err = fsw.DeleteHardLink(ctx, existingEntry.HardLinkId); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (fsw *FilerStoreWrapper) setHardLink(ctx context.Context, entry *Entry) error {
|
||||
if entry.HardLinkId == 0 {
|
||||
return nil
|
||||
}
|
||||
key := entry.HardLinkId.Key()
|
||||
|
||||
newBlob, encodeErr := entry.EncodeAttributesAndChunks()
|
||||
if encodeErr != nil {
|
||||
return encodeErr
|
||||
}
|
||||
|
||||
return fsw.KvPut(ctx, key, newBlob)
|
||||
}
|
||||
|
||||
func (fsw *FilerStoreWrapper) maybeReadHardLink(ctx context.Context, entry *Entry) error {
|
||||
if entry.HardLinkId == 0 {
|
||||
return nil
|
||||
}
|
||||
key := entry.HardLinkId.Key()
|
||||
|
||||
value, err := fsw.KvGet(ctx, key)
|
||||
if err != nil {
|
||||
glog.Errorf("read %s hardlink %d: %v", entry.FullPath, entry.HardLinkId, err)
|
||||
return err
|
||||
}
|
||||
|
||||
if err = entry.DecodeAttributesAndChunks(value); err != nil {
|
||||
glog.Errorf("decode %s hardlink %d: %v", entry.FullPath, entry.HardLinkId, err)
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (fsw *FilerStoreWrapper) DeleteHardLink(ctx context.Context, hardLinkId HardLinkId) error {
|
||||
key := hardLinkId.Key()
|
||||
value, err := fsw.KvGet(ctx, key)
|
||||
if err == ErrKvNotFound {
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
entry := &Entry{}
|
||||
if err = entry.DecodeAttributesAndChunks(value); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
entry.HardLinkCounter--
|
||||
if entry.HardLinkCounter <= 0 {
|
||||
return fsw.KvDelete(ctx, key)
|
||||
}
|
||||
|
||||
newBlob, encodeErr := entry.EncodeAttributesAndChunks()
|
||||
if encodeErr != nil {
|
||||
return encodeErr
|
||||
}
|
||||
|
||||
return fsw.KvPut(ctx, key, newBlob)
|
||||
|
||||
}
|
||||
|
|
@ -32,22 +32,15 @@ func (dir *Dir) Link(ctx context.Context, req *fuse.LinkRequest, old fs.Node) (f
|
|||
}
|
||||
|
||||
// update old file to hardlink mode
|
||||
var updateOldEntryRequest *filer_pb.UpdateEntryRequest
|
||||
var hardLinkId filer.HardLinkId
|
||||
if oldFile.entry.HardLinkId != 0 {
|
||||
hardLinkId = filer.HardLinkId(oldFile.entry.HardLinkId)
|
||||
} else {
|
||||
// CreateLink 1.1 : split source entry into hardlink+empty_entry
|
||||
hardLinkId = filer.HardLinkId(util.RandomInt64())
|
||||
updateOldEntryRequest = &filer_pb.UpdateEntryRequest{
|
||||
Directory: oldFile.dir.FullPath(),
|
||||
Entry: &filer_pb.Entry{
|
||||
Name: oldFile.entry.Name,
|
||||
IsDirectory: oldFile.entry.IsDirectory,
|
||||
HardLinkId: int64(hardLinkId),
|
||||
},
|
||||
Signatures: []int32{dir.wfs.signature},
|
||||
if oldFile.entry.HardLinkId == 0 {
|
||||
oldFile.entry.HardLinkId = util.RandomInt64()
|
||||
oldFile.entry.HardLinkCounter = 1
|
||||
}
|
||||
oldFile.entry.HardLinkCounter++
|
||||
updateOldEntryRequest := &filer_pb.UpdateEntryRequest{
|
||||
Directory: oldFile.dir.FullPath(),
|
||||
Entry: oldFile.entry,
|
||||
Signatures: []int32{dir.wfs.signature},
|
||||
}
|
||||
|
||||
// CreateLink 1.2 : update new file to hardlink mode
|
||||
|
|
@ -56,7 +49,11 @@ func (dir *Dir) Link(ctx context.Context, req *fuse.LinkRequest, old fs.Node) (f
|
|||
Entry: &filer_pb.Entry{
|
||||
Name: req.NewName,
|
||||
IsDirectory: false,
|
||||
HardLinkId: int64(hardLinkId),
|
||||
Attributes: oldFile.entry.Attributes,
|
||||
Chunks: oldFile.entry.Chunks,
|
||||
Extended: oldFile.entry.Extended,
|
||||
HardLinkId: oldFile.entry.HardLinkId,
|
||||
HardLinkCounter: oldFile.entry.HardLinkCounter,
|
||||
},
|
||||
Signatures: []int32{dir.wfs.signature},
|
||||
}
|
||||
|
|
@ -67,14 +64,11 @@ func (dir *Dir) Link(ctx context.Context, req *fuse.LinkRequest, old fs.Node) (f
|
|||
dir.wfs.mapPbIdFromLocalToFiler(request.Entry)
|
||||
defer dir.wfs.mapPbIdFromFilerToLocal(request.Entry)
|
||||
|
||||
if updateOldEntryRequest != nil {
|
||||
if err := filer_pb.UpdateEntry(client, updateOldEntryRequest); err != nil {
|
||||
glog.V(0).Infof("Link %v/%v -> %s/%s: %v", oldFile.dir.FullPath(), oldFile.Name, dir.FullPath(), req.NewName, err)
|
||||
return fuse.EIO
|
||||
}
|
||||
dir.wfs.metaCache.UpdateEntry(context.Background(), filer.FromPbEntry(updateOldEntryRequest.Directory, updateOldEntryRequest.Entry))
|
||||
oldFile.entry.HardLinkId = int64(hardLinkId)
|
||||
}
|
||||
|
||||
if err := filer_pb.CreateEntry(client, request); err != nil {
|
||||
glog.V(0).Infof("Link %v/%v -> %s/%s: %v", oldFile.dir.FullPath(), oldFile.Name, dir.FullPath(), req.NewName, err)
|
||||
|
|
|
|||
Loading…
Reference in a new issue