mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-06 14:45:51 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
bb1be61602 | ||
|
|
fca6152c3a | ||
|
|
a990ef2106 | ||
|
|
630f72f8c5 | ||
|
|
1b0bfbaf59 | ||
|
|
2f15e93466 | ||
|
|
04019aa161 | ||
|
|
6a40c688e0 | ||
|
|
39edcfde4c | ||
|
|
e9bb0c179f | ||
|
|
061a61e018 | ||
|
|
230a0e4a33 | ||
|
|
e4b660699f | ||
|
|
d1ab16b6e3 |
@@ -145,9 +145,9 @@ addresses = [
|
||||
"localhost:30006",
|
||||
]
|
||||
password = ""
|
||||
// allows reads from slave servers or the master, but all writes still go to the master
|
||||
# allows reads from slave servers or the master, but all writes still go to the master
|
||||
readOnly = true
|
||||
// automatically use the closest Redis server for reads
|
||||
# automatically use the closest Redis server for reads
|
||||
routeByLatency = true
|
||||
|
||||
[etcd]
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
|
||||
"github.com/chrislusf/seaweedfs/weed/filer2"
|
||||
"github.com/chrislusf/seaweedfs/weed/glog"
|
||||
"github.com/chrislusf/seaweedfs/weed/util"
|
||||
)
|
||||
|
||||
type AbstractSqlStore struct {
|
||||
@@ -65,7 +66,7 @@ func (store *AbstractSqlStore) InsertEntry(ctx context.Context, entry *filer2.En
|
||||
return fmt.Errorf("encode %s: %s", entry.FullPath, err)
|
||||
}
|
||||
|
||||
res, err := store.getTxOrDB(ctx).ExecContext(ctx, store.SqlInsert, hashToLong(dir), name, dir, meta)
|
||||
res, err := store.getTxOrDB(ctx).ExecContext(ctx, store.SqlInsert, util.HashStringToLong(dir), name, dir, meta)
|
||||
if err != nil {
|
||||
return fmt.Errorf("insert %s: %s", entry.FullPath, err)
|
||||
}
|
||||
@@ -85,7 +86,7 @@ func (store *AbstractSqlStore) UpdateEntry(ctx context.Context, entry *filer2.En
|
||||
return fmt.Errorf("encode %s: %s", entry.FullPath, err)
|
||||
}
|
||||
|
||||
res, err := store.getTxOrDB(ctx).ExecContext(ctx, store.SqlUpdate, meta, hashToLong(dir), name, dir)
|
||||
res, err := store.getTxOrDB(ctx).ExecContext(ctx, store.SqlUpdate, meta, util.HashStringToLong(dir), name, dir)
|
||||
if err != nil {
|
||||
return fmt.Errorf("update %s: %s", entry.FullPath, err)
|
||||
}
|
||||
@@ -100,7 +101,7 @@ func (store *AbstractSqlStore) UpdateEntry(ctx context.Context, entry *filer2.En
|
||||
func (store *AbstractSqlStore) FindEntry(ctx context.Context, fullpath filer2.FullPath) (*filer2.Entry, error) {
|
||||
|
||||
dir, name := fullpath.DirAndName()
|
||||
row := store.getTxOrDB(ctx).QueryRowContext(ctx, store.SqlFind, hashToLong(dir), name, dir)
|
||||
row := store.getTxOrDB(ctx).QueryRowContext(ctx, store.SqlFind, util.HashStringToLong(dir), name, dir)
|
||||
var data []byte
|
||||
if err := row.Scan(&data); err != nil {
|
||||
return nil, filer2.ErrNotFound
|
||||
@@ -120,7 +121,7 @@ func (store *AbstractSqlStore) DeleteEntry(ctx context.Context, fullpath filer2.
|
||||
|
||||
dir, name := fullpath.DirAndName()
|
||||
|
||||
res, err := store.getTxOrDB(ctx).ExecContext(ctx, store.SqlDelete, hashToLong(dir), name, dir)
|
||||
res, err := store.getTxOrDB(ctx).ExecContext(ctx, store.SqlDelete, util.HashStringToLong(dir), name, dir)
|
||||
if err != nil {
|
||||
return fmt.Errorf("delete %s: %s", fullpath, err)
|
||||
}
|
||||
@@ -135,7 +136,7 @@ func (store *AbstractSqlStore) DeleteEntry(ctx context.Context, fullpath filer2.
|
||||
|
||||
func (store *AbstractSqlStore) DeleteFolderChildren(ctx context.Context, fullpath filer2.FullPath) error {
|
||||
|
||||
res, err := store.getTxOrDB(ctx).ExecContext(ctx, store.SqlDeleteFolderChildren, hashToLong(string(fullpath)), fullpath)
|
||||
res, err := store.getTxOrDB(ctx).ExecContext(ctx, store.SqlDeleteFolderChildren, util.HashStringToLong(string(fullpath)), fullpath)
|
||||
if err != nil {
|
||||
return fmt.Errorf("deleteFolderChildren %s: %s", fullpath, err)
|
||||
}
|
||||
@@ -155,7 +156,7 @@ func (store *AbstractSqlStore) ListDirectoryEntries(ctx context.Context, fullpat
|
||||
sqlText = store.SqlListInclusive
|
||||
}
|
||||
|
||||
rows, err := store.getTxOrDB(ctx).QueryContext(ctx, sqlText, hashToLong(string(fullpath)), startFileName, string(fullpath), limit)
|
||||
rows, err := store.getTxOrDB(ctx).QueryContext(ctx, sqlText, util.HashStringToLong(string(fullpath)), startFileName, string(fullpath), limit)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("list %s : %v", fullpath, err)
|
||||
}
|
||||
|
||||
@@ -1,32 +0,0 @@
|
||||
package abstract_sql
|
||||
|
||||
import (
|
||||
"crypto/md5"
|
||||
"io"
|
||||
)
|
||||
|
||||
// returns a 64 bit big int
|
||||
func hashToLong(dir string) (v int64) {
|
||||
h := md5.New()
|
||||
io.WriteString(h, dir)
|
||||
|
||||
b := h.Sum(nil)
|
||||
|
||||
v += int64(b[0])
|
||||
v <<= 8
|
||||
v += int64(b[1])
|
||||
v <<= 8
|
||||
v += int64(b[2])
|
||||
v <<= 8
|
||||
v += int64(b[3])
|
||||
v <<= 8
|
||||
v += int64(b[4])
|
||||
v <<= 8
|
||||
v += int64(b[5])
|
||||
v <<= 8
|
||||
v += int64(b[6])
|
||||
v <<= 8
|
||||
v += int64(b[7])
|
||||
|
||||
return
|
||||
}
|
||||
@@ -25,7 +25,7 @@ type FilerClient interface {
|
||||
WithFilerClient(ctx context.Context, fn func(filer_pb.SeaweedFilerClient) error) error
|
||||
}
|
||||
|
||||
func ReadIntoBuffer(ctx context.Context, filerClient FilerClient, fullFilePath string, buff []byte, chunkViews []*ChunkView, baseOffset int64) (totalRead int64, err error) {
|
||||
func ReadIntoBuffer(ctx context.Context, filerClient FilerClient, fullFilePath FullPath, buff []byte, chunkViews []*ChunkView, baseOffset int64) (totalRead int64, err error) {
|
||||
var vids []string
|
||||
for _, chunkView := range chunkViews {
|
||||
vids = append(vids, VolumeId(chunkView.FileId))
|
||||
@@ -93,9 +93,9 @@ func ReadIntoBuffer(ctx context.Context, filerClient FilerClient, fullFilePath s
|
||||
return
|
||||
}
|
||||
|
||||
func GetEntry(ctx context.Context, filerClient FilerClient, fullFilePath string) (entry *filer_pb.Entry, err error) {
|
||||
func GetEntry(ctx context.Context, filerClient FilerClient, fullFilePath FullPath) (entry *filer_pb.Entry, err error) {
|
||||
|
||||
dir, name := FullPath(fullFilePath).DirAndName()
|
||||
dir, name := fullFilePath.DirAndName()
|
||||
|
||||
err = filerClient.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
|
||||
|
||||
@@ -126,14 +126,14 @@ func GetEntry(ctx context.Context, filerClient FilerClient, fullFilePath string)
|
||||
return
|
||||
}
|
||||
|
||||
func ReadDirAllEntries(ctx context.Context, filerClient FilerClient, fullDirPath, prefix string, fn func(entry *filer_pb.Entry, isLast bool)) (err error) {
|
||||
func ReadDirAllEntries(ctx context.Context, filerClient FilerClient, fullDirPath FullPath, prefix string, fn func(entry *filer_pb.Entry, isLast bool)) (err error) {
|
||||
|
||||
err = filerClient.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
|
||||
|
||||
lastEntryName := ""
|
||||
|
||||
request := &filer_pb.ListEntriesRequest{
|
||||
Directory: fullDirPath,
|
||||
Directory: string(fullDirPath),
|
||||
Prefix: prefix,
|
||||
StartFromFileName: lastEntryName,
|
||||
Limit: math.MaxUint32,
|
||||
|
||||
@@ -3,6 +3,8 @@ package filer2
|
||||
import (
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
"github.com/chrislusf/seaweedfs/weed/util"
|
||||
)
|
||||
|
||||
type FullPath string
|
||||
@@ -34,3 +36,7 @@ func (fp FullPath) Child(name string) FullPath {
|
||||
}
|
||||
return FullPath(dir + "/" + name)
|
||||
}
|
||||
|
||||
func (fp FullPath) AsInode() uint64 {
|
||||
return uint64(util.HashStringToLong(string(fp)))
|
||||
}
|
||||
|
||||
+63
-41
@@ -3,7 +3,6 @@ package filesys
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"path"
|
||||
"time"
|
||||
|
||||
"github.com/chrislusf/seaweedfs/weed/filer2"
|
||||
@@ -31,29 +30,35 @@ var _ = fs.NodeGetxattrer(&Dir{})
|
||||
var _ = fs.NodeSetxattrer(&Dir{})
|
||||
var _ = fs.NodeRemovexattrer(&Dir{})
|
||||
var _ = fs.NodeListxattrer(&Dir{})
|
||||
var _ = fs.NodeForgetter(&Dir{})
|
||||
|
||||
func (dir *Dir) Attr(ctx context.Context, attr *fuse.Attr) error {
|
||||
|
||||
glog.V(3).Infof("dir Attr %s", dir.Path)
|
||||
glog.V(3).Infof("dir Attr %s, existing attr: %+v", dir.Path, attr)
|
||||
|
||||
// https://github.com/bazil/fuse/issues/196
|
||||
attr.Valid = time.Second
|
||||
|
||||
if dir.Path == dir.wfs.option.FilerMountRootPath {
|
||||
dir.setRootDirAttributes(attr)
|
||||
glog.V(3).Infof("root dir Attr %s, attr: %+v", dir.Path, attr)
|
||||
return nil
|
||||
}
|
||||
|
||||
if err := dir.maybeLoadEntry(ctx); err != nil {
|
||||
glog.V(3).Infof("dir Attr %s,err: %+v", dir.Path, err)
|
||||
return err
|
||||
}
|
||||
|
||||
attr.Inode = filer2.FullPath(dir.Path).AsInode()
|
||||
attr.Mode = os.FileMode(dir.entry.Attributes.FileMode) | os.ModeDir
|
||||
attr.Mtime = time.Unix(dir.entry.Attributes.Mtime, 0)
|
||||
attr.Ctime = time.Unix(dir.entry.Attributes.Crtime, 0)
|
||||
attr.Gid = dir.entry.Attributes.Gid
|
||||
attr.Uid = dir.entry.Attributes.Uid
|
||||
|
||||
glog.V(3).Infof("dir Attr %s, attr: %+v", dir.Path, attr)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -69,6 +74,8 @@ func (dir *Dir) Getxattr(ctx context.Context, req *fuse.GetxattrRequest, resp *f
|
||||
}
|
||||
|
||||
func (dir *Dir) setRootDirAttributes(attr *fuse.Attr) {
|
||||
attr.Inode = 1 // filer2.FullPath(dir.Path).AsInode()
|
||||
attr.Valid = time.Hour
|
||||
attr.Uid = dir.wfs.option.MountUid
|
||||
attr.Gid = dir.wfs.option.MountGid
|
||||
attr.Mode = dir.wfs.option.MountMode
|
||||
@@ -76,16 +83,25 @@ func (dir *Dir) setRootDirAttributes(attr *fuse.Attr) {
|
||||
attr.Ctime = dir.wfs.option.MountCtime
|
||||
attr.Mtime = dir.wfs.option.MountMtime
|
||||
attr.Atime = dir.wfs.option.MountMtime
|
||||
attr.BlockSize = 1024 * 1024
|
||||
}
|
||||
|
||||
func (dir *Dir) newFile(name string, entry *filer_pb.Entry) *File {
|
||||
return &File{
|
||||
Name: name,
|
||||
dir: dir,
|
||||
wfs: dir.wfs,
|
||||
entry: entry,
|
||||
entryViewCache: nil,
|
||||
}
|
||||
func (dir *Dir) newFile(name string, entry *filer_pb.Entry) fs.Node {
|
||||
return dir.wfs.getNode(filer2.NewFullPath(dir.Path, name), func() fs.Node {
|
||||
return &File{
|
||||
Name: name,
|
||||
dir: dir,
|
||||
wfs: dir.wfs,
|
||||
entry: entry,
|
||||
entryViewCache: nil,
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func (dir *Dir) newDirectory(fullpath filer2.FullPath, entry *filer_pb.Entry) fs.Node {
|
||||
return dir.wfs.getNode(fullpath, func() fs.Node {
|
||||
return &Dir{Path: string(fullpath), wfs: dir.wfs, entry: entry}
|
||||
})
|
||||
}
|
||||
|
||||
func (dir *Dir) Create(ctx context.Context, req *fuse.CreateRequest,
|
||||
@@ -122,7 +138,8 @@ func (dir *Dir) Create(ctx context.Context, req *fuse.CreateRequest,
|
||||
}
|
||||
}
|
||||
|
||||
file := dir.newFile(req.Name, request.Entry)
|
||||
node := dir.newFile(req.Name, request.Entry)
|
||||
file := node.(*File)
|
||||
if !request.Entry.IsDirectory {
|
||||
file.isOpen = true
|
||||
}
|
||||
@@ -161,7 +178,7 @@ func (dir *Dir) Mkdir(ctx context.Context, req *fuse.MkdirRequest) (fs.Node, err
|
||||
})
|
||||
|
||||
if err == nil {
|
||||
node := &Dir{Path: path.Join(dir.Path, req.Name), wfs: dir.wfs}
|
||||
node := dir.newDirectory(filer2.NewFullPath(dir.Path, req.Name), nil)
|
||||
return node, nil
|
||||
}
|
||||
|
||||
@@ -172,35 +189,31 @@ func (dir *Dir) Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse.
|
||||
|
||||
glog.V(4).Infof("dir Lookup %s: %s", dir.Path, req.Name)
|
||||
|
||||
var entry *filer_pb.Entry
|
||||
fullFilePath := path.Join(dir.Path, req.Name)
|
||||
|
||||
item := dir.wfs.listDirectoryEntriesCache.Get(fullFilePath)
|
||||
if item != nil && !item.Expired() {
|
||||
entry = item.Value().(*filer_pb.Entry)
|
||||
}
|
||||
fullFilePath := filer2.NewFullPath(dir.Path, req.Name)
|
||||
entry := dir.wfs.cacheGet(fullFilePath)
|
||||
|
||||
if entry == nil {
|
||||
glog.V(3).Infof("dir Lookup cache miss %s", fullFilePath)
|
||||
entry, err = filer2.GetEntry(ctx, dir.wfs, fullFilePath)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if entry != nil {
|
||||
dir.wfs.listDirectoryEntriesCache.Set(fullFilePath, entry, 5*time.Minute)
|
||||
glog.V(1).Infof("dir GetEntry %s: %v", fullFilePath, err)
|
||||
return nil, fuse.ENOENT
|
||||
}
|
||||
dir.wfs.cacheSet(fullFilePath, entry, 5*time.Minute)
|
||||
} else {
|
||||
glog.V(4).Infof("dir Lookup cache hit %s", fullFilePath)
|
||||
}
|
||||
|
||||
if entry != nil {
|
||||
if entry.IsDirectory {
|
||||
node = &Dir{Path: path.Join(dir.Path, req.Name), wfs: dir.wfs, entry: entry}
|
||||
node = dir.newDirectory(fullFilePath, entry)
|
||||
} else {
|
||||
node = dir.newFile(req.Name, entry)
|
||||
}
|
||||
|
||||
resp.EntryValid = time.Duration(0)
|
||||
// resp.EntryValid = time.Second
|
||||
resp.Attr.Inode = fullFilePath.AsInode()
|
||||
resp.Attr.Valid = time.Second
|
||||
resp.Attr.Mtime = time.Unix(entry.Attributes.Mtime, 0)
|
||||
resp.Attr.Ctime = time.Unix(entry.Attributes.Crtime, 0)
|
||||
resp.Attr.Mode = os.FileMode(entry.Attributes.FileMode)
|
||||
@@ -210,6 +223,7 @@ func (dir *Dir) Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse.
|
||||
return node, nil
|
||||
}
|
||||
|
||||
glog.V(1).Infof("not found dir GetEntry %s: %v", fullFilePath, err)
|
||||
return nil, fuse.ENOENT
|
||||
}
|
||||
|
||||
@@ -219,15 +233,18 @@ func (dir *Dir) ReadDirAll(ctx context.Context) (ret []fuse.Dirent, err error) {
|
||||
|
||||
cacheTtl := 5 * time.Minute
|
||||
|
||||
readErr := filer2.ReadDirAllEntries(ctx, dir.wfs, dir.Path, "", func(entry *filer_pb.Entry, isLast bool) {
|
||||
readErr := filer2.ReadDirAllEntries(ctx, dir.wfs, filer2.FullPath(dir.Path), "", func(entry *filer_pb.Entry, isLast bool) {
|
||||
fullpath := filer2.NewFullPath(dir.Path, entry.Name)
|
||||
inode := fullpath.AsInode()
|
||||
if entry.IsDirectory {
|
||||
dirent := fuse.Dirent{Name: entry.Name, Type: fuse.DT_Dir}
|
||||
dirent := fuse.Dirent{Inode: inode, Name: entry.Name, Type: fuse.DT_Dir}
|
||||
ret = append(ret, dirent)
|
||||
} else {
|
||||
dirent := fuse.Dirent{Name: entry.Name, Type: fuse.DT_File}
|
||||
dirent := fuse.Dirent{Inode: inode, Name: entry.Name, Type: fuse.DT_File}
|
||||
ret = append(ret, dirent)
|
||||
}
|
||||
dir.wfs.listDirectoryEntriesCache.Set(path.Join(dir.Path, entry.Name), entry, cacheTtl)
|
||||
glog.V(4).Infof("dir ReadDirAll : %s %+v", fullpath, entry)
|
||||
dir.wfs.cacheSet(fullpath, entry, cacheTtl)
|
||||
})
|
||||
if readErr != nil {
|
||||
glog.V(0).Infof("list %s: %v", dir.Path, err)
|
||||
@@ -249,14 +266,15 @@ func (dir *Dir) Remove(ctx context.Context, req *fuse.RemoveRequest) error {
|
||||
|
||||
func (dir *Dir) removeOneFile(ctx context.Context, req *fuse.RemoveRequest) error {
|
||||
|
||||
entry, err := filer2.GetEntry(ctx, dir.wfs, path.Join(dir.Path, req.Name))
|
||||
filePath := filer2.NewFullPath(dir.Path, req.Name)
|
||||
entry, err := filer2.GetEntry(ctx, dir.wfs, filePath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
dir.wfs.deleteFileChunks(ctx, entry.Chunks)
|
||||
|
||||
dir.wfs.listDirectoryEntriesCache.Delete(path.Join(dir.Path, req.Name))
|
||||
dir.wfs.cacheDelete(filePath)
|
||||
|
||||
return dir.wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
|
||||
|
||||
@@ -269,7 +287,7 @@ func (dir *Dir) removeOneFile(ctx context.Context, req *fuse.RemoveRequest) erro
|
||||
glog.V(3).Infof("remove file: %v", request)
|
||||
_, err := client.DeleteEntry(ctx, request)
|
||||
if err != nil {
|
||||
glog.V(3).Infof("remove file %s/%s: %v", dir.Path, req.Name, err)
|
||||
glog.V(3).Infof("not found remove file %s/%s: %v", dir.Path, req.Name, err)
|
||||
return fuse.ENOENT
|
||||
}
|
||||
|
||||
@@ -280,7 +298,7 @@ func (dir *Dir) removeOneFile(ctx context.Context, req *fuse.RemoveRequest) erro
|
||||
|
||||
func (dir *Dir) removeFolder(ctx context.Context, req *fuse.RemoveRequest) error {
|
||||
|
||||
dir.wfs.listDirectoryEntriesCache.Delete(path.Join(dir.Path, req.Name))
|
||||
dir.wfs.cacheDelete(filer2.NewFullPath(dir.Path, req.Name))
|
||||
|
||||
return dir.wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
|
||||
|
||||
@@ -293,7 +311,7 @@ func (dir *Dir) removeFolder(ctx context.Context, req *fuse.RemoveRequest) error
|
||||
glog.V(3).Infof("remove directory entry: %v", request)
|
||||
_, err := client.DeleteEntry(ctx, request)
|
||||
if err != nil {
|
||||
glog.V(3).Infof("remove %s/%s: %v", dir.Path, req.Name, err)
|
||||
glog.V(3).Infof("not found remove %s/%s: %v", dir.Path, req.Name, err)
|
||||
return fuse.ENOENT
|
||||
}
|
||||
|
||||
@@ -304,11 +322,12 @@ func (dir *Dir) removeFolder(ctx context.Context, req *fuse.RemoveRequest) error
|
||||
|
||||
func (dir *Dir) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *fuse.SetattrResponse) error {
|
||||
|
||||
glog.V(3).Infof("%v dir setattr %+v", dir.Path, req)
|
||||
|
||||
if err := dir.maybeLoadEntry(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
glog.V(3).Infof("%v dir setattr %+v, fh=%d", dir.Path, req, req.Handle)
|
||||
if req.Valid.Mode() {
|
||||
dir.entry.Attributes.FileMode = uint32(req.Mode)
|
||||
}
|
||||
@@ -325,7 +344,7 @@ func (dir *Dir) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *fus
|
||||
dir.entry.Attributes.Mtime = req.Mtime.Unix()
|
||||
}
|
||||
|
||||
dir.wfs.listDirectoryEntriesCache.Delete(dir.Path)
|
||||
dir.wfs.cacheDelete(filer2.FullPath(dir.Path))
|
||||
|
||||
return dir.saveEntry(ctx)
|
||||
|
||||
@@ -343,7 +362,7 @@ func (dir *Dir) Setxattr(ctx context.Context, req *fuse.SetxattrRequest) error {
|
||||
return err
|
||||
}
|
||||
|
||||
dir.wfs.listDirectoryEntriesCache.Delete(dir.Path)
|
||||
dir.wfs.cacheDelete(filer2.FullPath(dir.Path))
|
||||
|
||||
return dir.saveEntry(ctx)
|
||||
|
||||
@@ -361,7 +380,7 @@ func (dir *Dir) Removexattr(ctx context.Context, req *fuse.RemovexattrRequest) e
|
||||
return err
|
||||
}
|
||||
|
||||
dir.wfs.listDirectoryEntriesCache.Delete(dir.Path)
|
||||
dir.wfs.cacheDelete(filer2.FullPath(dir.Path))
|
||||
|
||||
return dir.saveEntry(ctx)
|
||||
|
||||
@@ -383,6 +402,12 @@ func (dir *Dir) Listxattr(ctx context.Context, req *fuse.ListxattrRequest, resp
|
||||
|
||||
}
|
||||
|
||||
func (dir *Dir) Forget() {
|
||||
glog.V(3).Infof("Forget dir %s", dir.Path)
|
||||
|
||||
dir.wfs.forgetNode(filer2.FullPath(dir.Path))
|
||||
}
|
||||
|
||||
func (dir *Dir) maybeLoadEntry(ctx context.Context) error {
|
||||
if dir.entry == nil {
|
||||
parentDirPath, name := filer2.FullPath(dir.Path).DirAndName()
|
||||
@@ -390,9 +415,6 @@ func (dir *Dir) maybeLoadEntry(ctx context.Context) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if entry == nil {
|
||||
return fuse.ENOENT
|
||||
}
|
||||
dir.entry = entry
|
||||
}
|
||||
return nil
|
||||
|
||||
@@ -2,7 +2,9 @@ package filesys
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"github.com/chrislusf/seaweedfs/weed/filer2"
|
||||
"github.com/chrislusf/seaweedfs/weed/glog"
|
||||
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/seaweedfs/fuse"
|
||||
"github.com/seaweedfs/fuse/fs"
|
||||
@@ -11,8 +13,9 @@ import (
|
||||
func (dir *Dir) Rename(ctx context.Context, req *fuse.RenameRequest, newDirectory fs.Node) error {
|
||||
|
||||
newDir := newDirectory.(*Dir)
|
||||
glog.V(4).Infof("dir Rename %s/%s => %s/%s", dir.Path, req.OldName, newDir.Path, req.NewName)
|
||||
|
||||
return dir.wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
|
||||
err := dir.wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
|
||||
|
||||
request := &filer_pb.AtomicRenameEntryRequest{
|
||||
OldDirectory: dir.Path,
|
||||
@@ -23,11 +26,34 @@ func (dir *Dir) Rename(ctx context.Context, req *fuse.RenameRequest, newDirector
|
||||
|
||||
_, err := client.AtomicRenameEntry(ctx, request)
|
||||
if err != nil {
|
||||
return fmt.Errorf("renaming %s/%s => %s/%s: %v", dir.Path, req.OldName, newDir.Path, req.NewName, err)
|
||||
glog.V(0).Infof("dir Rename %s/%s => %s/%s : %v", dir.Path, req.OldName, newDir.Path, req.NewName, err)
|
||||
return fuse.EIO
|
||||
}
|
||||
|
||||
return nil
|
||||
|
||||
})
|
||||
|
||||
if err == nil {
|
||||
oldPath := filer2.NewFullPath(dir.Path, req.OldName)
|
||||
dir.wfs.cacheDelete(filer2.NewFullPath(newDir.Path, req.NewName))
|
||||
dir.wfs.cacheDelete(oldPath)
|
||||
|
||||
oldFileNode := dir.wfs.getNode(oldPath, func() fs.Node {
|
||||
return nil
|
||||
})
|
||||
newDirNode := dir.wfs.getNode(filer2.FullPath(dir.Path), func() fs.Node {
|
||||
return nil
|
||||
})
|
||||
if oldFileNode != nil {
|
||||
oldFile := oldFileNode.(*File)
|
||||
oldFile.Name = req.NewName
|
||||
if newDirNode != nil {
|
||||
oldFile.dir = newDirNode.(*Dir)
|
||||
}
|
||||
}
|
||||
dir.wfs.forgetNode(oldPath)
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -65,7 +65,7 @@ func (pages *ContinuousDirtyPages) AddPage(ctx context.Context, offset int64, da
|
||||
// or buffer is full if adding new data,
|
||||
// flush current buffer and add new data
|
||||
|
||||
// println("offset", offset, "size", len(data), "existing offset", pages.Offset, "size", pages.Size)
|
||||
glog.V(4).Infof("offset=%d, size=%d, existing pages offset=%d, pages size=%d, data=%d", offset, len(data), pages.Offset, pages.Size, len(pages.Data))
|
||||
|
||||
if chunk, err = pages.saveExistingPagesToStorage(ctx); err == nil {
|
||||
if chunk != nil {
|
||||
@@ -77,6 +77,7 @@ func (pages *ContinuousDirtyPages) AddPage(ctx context.Context, offset int64, da
|
||||
return
|
||||
}
|
||||
pages.Offset = offset
|
||||
glog.V(4).Infof("copy data0: offset=%d, size=%d, existing pages offset=%d, pages size=%d, data=%d", offset, len(data), pages.Offset, pages.Size, len(pages.Data))
|
||||
copy(pages.Data, data)
|
||||
pages.Size = int64(len(data))
|
||||
return
|
||||
@@ -86,7 +87,7 @@ func (pages *ContinuousDirtyPages) AddPage(ctx context.Context, offset int64, da
|
||||
// when this happens, debug shows the data overlapping with existing data is empty
|
||||
// the data is not just append
|
||||
if offset == pages.Offset && int(pages.Size) < len(data) {
|
||||
// glog.V(2).Infof("pages[%d,%d) pages.Data len=%v, data len=%d, pages.Size=%d", pages.Offset, pages.Offset+pages.Size, len(pages.Data), len(data), pages.Size)
|
||||
glog.V(4).Infof("copy data1: offset=%d, size=%d, existing pages offset=%d, pages size=%d, data=%d", offset, len(data), pages.Offset, pages.Size, len(pages.Data))
|
||||
copy(pages.Data[pages.Size:], data[pages.Size:])
|
||||
} else {
|
||||
if pages.Size != 0 {
|
||||
@@ -95,6 +96,7 @@ func (pages *ContinuousDirtyPages) AddPage(ctx context.Context, offset int64, da
|
||||
return pages.flushAndSave(ctx, offset, data)
|
||||
}
|
||||
} else {
|
||||
glog.V(4).Infof("copy data2: offset=%d, size=%d, existing pages offset=%d, pages size=%d, data=%d", offset, len(data), pages.Offset, pages.Size, len(pages.Data))
|
||||
copy(pages.Data[offset-pages.Offset:], data)
|
||||
}
|
||||
|
||||
@@ -159,6 +161,8 @@ func (pages *ContinuousDirtyPages) saveExistingPagesToStorage(ctx context.Contex
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
glog.V(0).Infof("%s/%s saveExistingPagesToStorage [%d,%d): Data len=%d", pages.f.dir.Path, pages.f.Name, pages.Offset, pages.Size, len(pages.Data))
|
||||
|
||||
return pages.saveToStorage(ctx, pages.Data[:pages.Size], pages.Offset)
|
||||
}
|
||||
|
||||
|
||||
+37
-12
@@ -3,7 +3,6 @@ package filesys
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"time"
|
||||
|
||||
@@ -24,6 +23,7 @@ var _ = fs.NodeGetxattrer(&File{})
|
||||
var _ = fs.NodeSetxattrer(&File{})
|
||||
var _ = fs.NodeRemovexattrer(&File{})
|
||||
var _ = fs.NodeListxattrer(&File{})
|
||||
var _ = fs.NodeForgetter(&File{})
|
||||
|
||||
type File struct {
|
||||
Name string
|
||||
@@ -34,20 +34,28 @@ type File struct {
|
||||
isOpen bool
|
||||
}
|
||||
|
||||
func (file *File) fullpath() string {
|
||||
return filepath.Join(file.dir.Path, file.Name)
|
||||
func (file *File) fullpath() filer2.FullPath {
|
||||
return filer2.NewFullPath(file.dir.Path, file.Name)
|
||||
}
|
||||
|
||||
func (file *File) Attr(ctx context.Context, attr *fuse.Attr) error {
|
||||
|
||||
glog.V(4).Infof("file Attr %s", file.fullpath())
|
||||
glog.V(4).Infof("file Attr %s, open:%v, existing attr: %+v", file.fullpath(), file.isOpen, attr)
|
||||
|
||||
if err := file.maybeLoadEntry(ctx); err != nil {
|
||||
return err
|
||||
if !file.isOpen {
|
||||
if err := file.maybeLoadEntry(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
attr.Inode = file.fullpath().AsInode()
|
||||
attr.Valid = time.Second
|
||||
attr.Mode = os.FileMode(file.entry.Attributes.FileMode)
|
||||
attr.Size = filer2.TotalSize(file.entry.Chunks)
|
||||
if file.isOpen {
|
||||
attr.Size = file.entry.Attributes.FileSize
|
||||
}
|
||||
attr.Crtime = time.Unix(file.entry.Attributes.Crtime, 0)
|
||||
attr.Mtime = time.Unix(file.entry.Attributes.Mtime, 0)
|
||||
attr.Gid = file.entry.Attributes.Gid
|
||||
attr.Uid = file.entry.Attributes.Uid
|
||||
@@ -87,17 +95,27 @@ func (file *File) Open(ctx context.Context, req *fuse.OpenRequest, resp *fuse.Op
|
||||
|
||||
func (file *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *fuse.SetattrResponse) error {
|
||||
|
||||
glog.V(3).Infof("%v file setattr %+v, old:%+v", file.fullpath(), req, file.entry.Attributes)
|
||||
|
||||
if err := file.maybeLoadEntry(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
glog.V(3).Infof("%v file setattr %+v, old:%+v", file.fullpath(), req, file.entry.Attributes)
|
||||
if req.Valid.Size() {
|
||||
|
||||
glog.V(3).Infof("%v file setattr set size=%v", file.fullpath(), req.Size)
|
||||
if req.Size == 0 {
|
||||
if req.Size < filer2.TotalSize(file.entry.Chunks) {
|
||||
// fmt.Printf("truncate %v \n", fullPath)
|
||||
file.entry.Chunks = nil
|
||||
var chunks []*filer_pb.FileChunk
|
||||
for _, chunk := range file.entry.Chunks {
|
||||
if uint64(chunk.Offset)+chunk.Size > req.Size {
|
||||
chunk.Size = req.Size - uint64(chunk.Offset)
|
||||
}
|
||||
if chunk.Size > 0 {
|
||||
chunks = append(chunks, chunk)
|
||||
}
|
||||
}
|
||||
file.entry.Chunks = chunks
|
||||
file.entryViewCache = nil
|
||||
}
|
||||
file.entry.Attributes.FileSize = req.Size
|
||||
@@ -126,7 +144,7 @@ func (file *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *f
|
||||
return nil
|
||||
}
|
||||
|
||||
file.wfs.listDirectoryEntriesCache.Delete(file.fullpath())
|
||||
file.wfs.cacheDelete(file.fullpath())
|
||||
|
||||
return file.saveEntry(ctx)
|
||||
|
||||
@@ -144,7 +162,7 @@ func (file *File) Setxattr(ctx context.Context, req *fuse.SetxattrRequest) error
|
||||
return err
|
||||
}
|
||||
|
||||
file.wfs.listDirectoryEntriesCache.Delete(file.fullpath())
|
||||
file.wfs.cacheDelete(file.fullpath())
|
||||
|
||||
return file.saveEntry(ctx)
|
||||
|
||||
@@ -162,7 +180,7 @@ func (file *File) Removexattr(ctx context.Context, req *fuse.RemovexattrRequest)
|
||||
return err
|
||||
}
|
||||
|
||||
file.wfs.listDirectoryEntriesCache.Delete(file.fullpath())
|
||||
file.wfs.cacheDelete(file.fullpath())
|
||||
|
||||
return file.saveEntry(ctx)
|
||||
|
||||
@@ -192,6 +210,13 @@ func (file *File) Fsync(ctx context.Context, req *fuse.FsyncRequest) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (file *File) Forget() {
|
||||
glog.V(3).Infof("Forget file %s/%s", file.dir.Path, file.Name)
|
||||
|
||||
file.wfs.forgetNode(filer2.NewFullPath(file.dir.Path, file.Name))
|
||||
|
||||
}
|
||||
|
||||
func (file *File) maybeLoadEntry(ctx context.Context) error {
|
||||
if file.entry == nil || !file.isOpen {
|
||||
entry, err := file.wfs.maybeLoadEntry(ctx, file.dir.Path, file.Name)
|
||||
|
||||
+65
-23
@@ -11,6 +11,7 @@ import (
|
||||
"github.com/karlseguin/ccache"
|
||||
"google.golang.org/grpc"
|
||||
|
||||
"github.com/chrislusf/seaweedfs/weed/filer2"
|
||||
"github.com/chrislusf/seaweedfs/weed/glog"
|
||||
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/chrislusf/seaweedfs/weed/util"
|
||||
@@ -45,13 +46,18 @@ type WFS struct {
|
||||
option *Option
|
||||
listDirectoryEntriesCache *ccache.Cache
|
||||
|
||||
// contains all open handles
|
||||
handles []*FileHandle
|
||||
pathToHandleIndex map[string]int
|
||||
pathToHandleLock sync.Mutex
|
||||
bufPool sync.Pool
|
||||
// contains all open handles, protected by handlesLock
|
||||
handlesLock sync.Mutex
|
||||
handles []*FileHandle
|
||||
|
||||
bufPool sync.Pool
|
||||
|
||||
stats statsCache
|
||||
|
||||
// nodes, protected by nodesLock
|
||||
nodesLock sync.Mutex
|
||||
nodes map[uint64]fs.Node
|
||||
root fs.Node
|
||||
}
|
||||
type statsCache struct {
|
||||
filer_pb.StatisticsResponse
|
||||
@@ -62,19 +68,21 @@ func NewSeaweedFileSystem(option *Option) *WFS {
|
||||
wfs := &WFS{
|
||||
option: option,
|
||||
listDirectoryEntriesCache: ccache.New(ccache.Configure().MaxSize(option.DirListCacheLimit * 3).ItemsToPrune(100)),
|
||||
pathToHandleIndex: make(map[string]int),
|
||||
bufPool: sync.Pool{
|
||||
New: func() interface{} {
|
||||
return make([]byte, option.ChunkSizeLimit)
|
||||
},
|
||||
},
|
||||
nodes: make(map[uint64]fs.Node),
|
||||
}
|
||||
|
||||
wfs.root = &Dir{Path: wfs.option.FilerMountRootPath, wfs: wfs}
|
||||
|
||||
return wfs
|
||||
}
|
||||
|
||||
func (wfs *WFS) Root() (fs.Node, error) {
|
||||
return &Dir{Path: wfs.option.FilerMountRootPath, wfs: wfs}, nil
|
||||
return wfs.root, nil
|
||||
}
|
||||
|
||||
func (wfs *WFS) WithFilerClient(ctx context.Context, fn func(filer_pb.SeaweedFilerClient) error) error {
|
||||
@@ -87,42 +95,35 @@ func (wfs *WFS) WithFilerClient(ctx context.Context, fn func(filer_pb.SeaweedFil
|
||||
}
|
||||
|
||||
func (wfs *WFS) AcquireHandle(file *File, uid, gid uint32) (fileHandle *FileHandle) {
|
||||
wfs.pathToHandleLock.Lock()
|
||||
defer wfs.pathToHandleLock.Unlock()
|
||||
|
||||
fullpath := file.fullpath()
|
||||
glog.V(4).Infof("%s AcquireHandle uid=%d gid=%d", fullpath, uid, gid)
|
||||
|
||||
index, found := wfs.pathToHandleIndex[fullpath]
|
||||
if found && wfs.handles[index] != nil {
|
||||
glog.V(2).Infoln(fullpath, "found fileHandle id", index)
|
||||
return wfs.handles[index]
|
||||
}
|
||||
wfs.handlesLock.Lock()
|
||||
defer wfs.handlesLock.Unlock()
|
||||
|
||||
fileHandle = newFileHandle(file, uid, gid)
|
||||
for i, h := range wfs.handles {
|
||||
if h == nil {
|
||||
wfs.handles[i] = fileHandle
|
||||
fileHandle.handle = uint64(i)
|
||||
wfs.pathToHandleIndex[fullpath] = i
|
||||
glog.V(4).Infoln(fullpath, "reuse fileHandle id", fileHandle.handle)
|
||||
glog.V(4).Infof( "%s reuse fh %d", fullpath,fileHandle.handle)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
wfs.handles = append(wfs.handles, fileHandle)
|
||||
fileHandle.handle = uint64(len(wfs.handles) - 1)
|
||||
glog.V(2).Infoln(fullpath, "new fileHandle id", fileHandle.handle)
|
||||
wfs.pathToHandleIndex[fullpath] = int(fileHandle.handle)
|
||||
glog.V(4).Infof( "%s new fh %d", fullpath,fileHandle.handle)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (wfs *WFS) ReleaseHandle(fullpath string, handleId fuse.HandleID) {
|
||||
wfs.pathToHandleLock.Lock()
|
||||
defer wfs.pathToHandleLock.Unlock()
|
||||
func (wfs *WFS) ReleaseHandle(fullpath filer2.FullPath, handleId fuse.HandleID) {
|
||||
wfs.handlesLock.Lock()
|
||||
defer wfs.handlesLock.Unlock()
|
||||
|
||||
glog.V(4).Infof("%s releasing handle id %d current handles length %d", fullpath, handleId, len(wfs.handles))
|
||||
delete(wfs.pathToHandleIndex, fullpath)
|
||||
glog.V(4).Infof("%s ReleaseHandle id %d current handles length %d", fullpath, handleId, len(wfs.handles))
|
||||
if int(handleId) < len(wfs.handles) {
|
||||
wfs.handles[int(handleId)] = nil
|
||||
}
|
||||
@@ -191,3 +192,44 @@ func (wfs *WFS) Statfs(ctx context.Context, req *fuse.StatfsRequest, resp *fuse.
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (wfs *WFS) cacheGet(path filer2.FullPath) *filer_pb.Entry {
|
||||
item := wfs.listDirectoryEntriesCache.Get(string(path))
|
||||
if item != nil && !item.Expired() {
|
||||
return item.Value().(*filer_pb.Entry)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func (wfs *WFS) cacheSet(path filer2.FullPath, entry *filer_pb.Entry, ttl time.Duration) {
|
||||
if entry == nil {
|
||||
wfs.listDirectoryEntriesCache.Delete(string(path))
|
||||
} else {
|
||||
wfs.listDirectoryEntriesCache.Set(string(path), entry, ttl)
|
||||
}
|
||||
}
|
||||
func (wfs *WFS) cacheDelete(path filer2.FullPath) {
|
||||
wfs.listDirectoryEntriesCache.Delete(string(path))
|
||||
}
|
||||
|
||||
func (wfs *WFS) getNode(fullpath filer2.FullPath, fn func() fs.Node) fs.Node {
|
||||
wfs.nodesLock.Lock()
|
||||
defer wfs.nodesLock.Unlock()
|
||||
|
||||
node, found := wfs.nodes[fullpath.AsInode()]
|
||||
if found {
|
||||
return node
|
||||
}
|
||||
node = fn()
|
||||
if node != nil {
|
||||
wfs.nodes[fullpath.AsInode()] = node
|
||||
}
|
||||
return node
|
||||
}
|
||||
|
||||
func (wfs *WFS) forgetNode(fullpath filer2.FullPath) {
|
||||
wfs.nodesLock.Lock()
|
||||
defer wfs.nodesLock.Unlock()
|
||||
|
||||
delete(wfs.nodes, fullpath.AsInode())
|
||||
|
||||
}
|
||||
|
||||
@@ -50,7 +50,10 @@ func deleteFileIds(ctx context.Context, grpcDialOption grpc.DialOption, client f
|
||||
VolumeId: vid,
|
||||
Locations: nil,
|
||||
}
|
||||
locations := resp.LocationsMap[vid]
|
||||
locations, found := resp.LocationsMap[vid]
|
||||
if !found {
|
||||
continue
|
||||
}
|
||||
for _, loc := range locations.Locations {
|
||||
lr.Locations = append(lr.Locations, operation.Location{
|
||||
Url: loc.Url,
|
||||
|
||||
+12
-10
@@ -2,8 +2,9 @@ package filesys
|
||||
|
||||
import (
|
||||
"context"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
"github.com/chrislusf/seaweedfs/weed/filer2"
|
||||
"github.com/chrislusf/seaweedfs/weed/glog"
|
||||
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/seaweedfs/fuse"
|
||||
@@ -109,10 +110,9 @@ func listxattr(entry *filer_pb.Entry, req *fuse.ListxattrRequest, resp *fuse.Lis
|
||||
|
||||
func (wfs *WFS) maybeLoadEntry(ctx context.Context, dir, name string) (entry *filer_pb.Entry, err error) {
|
||||
|
||||
fullpath := filepath.Join(dir, name)
|
||||
item := wfs.listDirectoryEntriesCache.Get(fullpath)
|
||||
if item != nil && !item.Expired() {
|
||||
entry = item.Value().(*filer_pb.Entry)
|
||||
fullpath := filer2.NewFullPath(dir, name)
|
||||
entry = wfs.cacheGet(fullpath)
|
||||
if entry != nil {
|
||||
return
|
||||
}
|
||||
glog.V(3).Infof("read entry cache miss %s", fullpath)
|
||||
@@ -125,15 +125,17 @@ func (wfs *WFS) maybeLoadEntry(ctx context.Context, dir, name string) (entry *fi
|
||||
}
|
||||
|
||||
resp, err := client.LookupDirectoryEntry(ctx, request)
|
||||
if err != nil {
|
||||
if err != nil || resp == nil || resp.Entry == nil {
|
||||
if err == filer2.ErrNotFound || strings.Contains(err.Error(), filer2.ErrNotFound.Error()) {
|
||||
glog.V(3).Infof("file attr read not found file %v: %v", request, err)
|
||||
return fuse.ENOENT
|
||||
}
|
||||
glog.V(3).Infof("file attr read file %v: %v", request, err)
|
||||
return fuse.ENOENT
|
||||
return fuse.EIO
|
||||
}
|
||||
|
||||
entry = resp.Entry
|
||||
if entry != nil {
|
||||
wfs.listDirectoryEntriesCache.Set(fullpath, entry, wfs.option.EntryCacheTtl)
|
||||
}
|
||||
wfs.cacheSet(fullpath, entry, wfs.option.EntryCacheTtl)
|
||||
|
||||
return nil
|
||||
})
|
||||
|
||||
@@ -335,8 +335,10 @@ func (fs *WebDavFileSystem) stat(ctx context.Context, fullFilePath string) (os.F
|
||||
return nil, err
|
||||
}
|
||||
|
||||
fullpath := filer2.FullPath(fullFilePath)
|
||||
|
||||
var fi FileInfo
|
||||
entry, err := filer2.GetEntry(ctx, fs, fullFilePath)
|
||||
entry, err := filer2.GetEntry(ctx, fs, fullpath)
|
||||
if entry == nil {
|
||||
return nil, os.ErrNotExist
|
||||
}
|
||||
@@ -344,14 +346,12 @@ func (fs *WebDavFileSystem) stat(ctx context.Context, fullFilePath string) (os.F
|
||||
return nil, err
|
||||
}
|
||||
fi.size = int64(filer2.TotalSize(entry.GetChunks()))
|
||||
fi.name = fullFilePath
|
||||
fi.name = string(fullpath)
|
||||
fi.mode = os.FileMode(entry.Attributes.FileMode)
|
||||
fi.modifiledTime = time.Unix(entry.Attributes.Mtime, 0)
|
||||
fi.isDirectory = entry.IsDirectory
|
||||
|
||||
_, fi.name = path.Split(path.Clean(fi.name))
|
||||
if fi.name == "" {
|
||||
fi.name = "/"
|
||||
if fi.name == "/" {
|
||||
fi.modifiledTime = time.Now()
|
||||
fi.isDirectory = true
|
||||
}
|
||||
@@ -372,7 +372,7 @@ func (f *WebDavFile) Write(buf []byte) (int, error) {
|
||||
var err error
|
||||
ctx := context.Background()
|
||||
if f.entry == nil {
|
||||
f.entry, err = filer2.GetEntry(ctx, f.fs, f.name)
|
||||
f.entry, err = filer2.GetEntry(ctx, f.fs, filer2.FullPath(f.name))
|
||||
}
|
||||
|
||||
if f.entry == nil {
|
||||
@@ -470,7 +470,7 @@ func (f *WebDavFile) Read(p []byte) (readSize int, err error) {
|
||||
ctx := context.Background()
|
||||
|
||||
if f.entry == nil {
|
||||
f.entry, err = filer2.GetEntry(ctx, f.fs, f.name)
|
||||
f.entry, err = filer2.GetEntry(ctx, f.fs, filer2.FullPath(f.name))
|
||||
}
|
||||
if f.entry == nil {
|
||||
return 0, err
|
||||
@@ -486,7 +486,7 @@ func (f *WebDavFile) Read(p []byte) (readSize int, err error) {
|
||||
}
|
||||
chunkViews := filer2.ViewFromVisibleIntervals(f.entryViewCache, f.off, len(p))
|
||||
|
||||
totalRead, err := filer2.ReadIntoBuffer(ctx, f.fs, f.name, p, chunkViews, f.off)
|
||||
totalRead, err := filer2.ReadIntoBuffer(ctx, f.fs, filer2.FullPath(f.name), p, chunkViews, f.off)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
@@ -507,12 +507,9 @@ func (f *WebDavFile) Readdir(count int) (ret []os.FileInfo, err error) {
|
||||
glog.V(2).Infof("WebDavFileSystem.Readdir %v count %d", f.name, count)
|
||||
ctx := context.Background()
|
||||
|
||||
dir := f.name
|
||||
if dir != "/" && strings.HasSuffix(dir, "/") {
|
||||
dir = dir[:len(dir)-1]
|
||||
}
|
||||
dir, _ := filer2.FullPath(f.name).DirAndName()
|
||||
|
||||
err = filer2.ReadDirAllEntries(ctx, f.fs, dir, "", func(entry *filer_pb.Entry, isLast bool) {
|
||||
err = filer2.ReadDirAllEntries(ctx, f.fs, filer2.FullPath(dir), "", func(entry *filer_pb.Entry, isLast bool) {
|
||||
fi := FileInfo{
|
||||
size: int64(filer2.TotalSize(entry.GetChunks())),
|
||||
name: entry.Name,
|
||||
|
||||
@@ -59,7 +59,7 @@ func (c *commandFsDu) Do(args []string, commandEnv *CommandEnv, writer io.Writer
|
||||
|
||||
func duTraverseDirectory(ctx context.Context, writer io.Writer, filerClient filer2.FilerClient, dir, name string) (blockCount uint64, byteCount uint64, err error) {
|
||||
|
||||
err = filer2.ReadDirAllEntries(ctx, filerClient, dir, name, func(entry *filer_pb.Entry, isLast bool) {
|
||||
err = filer2.ReadDirAllEntries(ctx, filerClient, filer2.FullPath(dir), name, func(entry *filer_pb.Entry, isLast bool) {
|
||||
if entry.IsDirectory {
|
||||
subDir := fmt.Sprintf("%s/%s", dir, entry.Name)
|
||||
if dir == "/" {
|
||||
|
||||
@@ -69,7 +69,7 @@ func (c *commandFsLs) Do(args []string, commandEnv *CommandEnv, writer io.Writer
|
||||
dir, name := filer2.FullPath(path).DirAndName()
|
||||
entryCount := 0
|
||||
|
||||
err = filer2.ReadDirAllEntries(ctx, commandEnv.getFilerClient(filerServer, filerPort), dir, name, func(entry *filer_pb.Entry, isLast bool) {
|
||||
err = filer2.ReadDirAllEntries(ctx, commandEnv.getFilerClient(filerServer, filerPort), filer2.FullPath(dir), name, func(entry *filer_pb.Entry, isLast bool) {
|
||||
|
||||
if !showHidden && strings.HasPrefix(entry.Name, ".") {
|
||||
return
|
||||
|
||||
@@ -168,7 +168,7 @@ func processOneDirectory(ctx context.Context, writer io.Writer, filerClient file
|
||||
parentPath filer2.FullPath, queue *util.Queue, jobQueueWg *sync.WaitGroup,
|
||||
fn func(parentPath filer2.FullPath, entry *filer_pb.Entry)) (err error) {
|
||||
|
||||
return filer2.ReadDirAllEntries(ctx, filerClient, string(parentPath), "", func(entry *filer_pb.Entry, isLast bool) {
|
||||
return filer2.ReadDirAllEntries(ctx, filerClient, parentPath, "", func(entry *filer_pb.Entry, isLast bool) {
|
||||
|
||||
fn(parentPath, entry)
|
||||
|
||||
|
||||
@@ -39,7 +39,7 @@ func (c *commandFsTree) Do(args []string, commandEnv *CommandEnv, writer io.Writ
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
dirCount, fCount, terr := treeTraverseDirectory(ctx, writer, commandEnv.getFilerClient(filerServer, filerPort), dir, name, newPrefix(), -1)
|
||||
dirCount, fCount, terr := treeTraverseDirectory(ctx, writer, commandEnv.getFilerClient(filerServer, filerPort), filer2.FullPath(dir), name, newPrefix(), -1)
|
||||
|
||||
if terr == nil {
|
||||
fmt.Fprintf(writer, "%d directories, %d files\n", dirCount, fCount)
|
||||
@@ -49,7 +49,7 @@ func (c *commandFsTree) Do(args []string, commandEnv *CommandEnv, writer io.Writ
|
||||
|
||||
}
|
||||
|
||||
func treeTraverseDirectory(ctx context.Context, writer io.Writer, filerClient filer2.FilerClient, dir, name string, prefix *Prefix, level int) (directoryCount, fileCount int64, err error) {
|
||||
func treeTraverseDirectory(ctx context.Context, writer io.Writer, filerClient filer2.FilerClient, dir filer2.FullPath, name string, prefix *Prefix, level int) (directoryCount, fileCount int64, err error) {
|
||||
|
||||
prefix.addMarker(level)
|
||||
|
||||
@@ -64,10 +64,7 @@ func treeTraverseDirectory(ctx context.Context, writer io.Writer, filerClient fi
|
||||
|
||||
if entry.IsDirectory {
|
||||
directoryCount++
|
||||
subDir := fmt.Sprintf("%s/%s", dir, entry.Name)
|
||||
if dir == "/" {
|
||||
subDir = "/" + entry.Name
|
||||
}
|
||||
subDir := dir.Child(entry.Name)
|
||||
dirCount, fCount, terr := treeTraverseDirectory(ctx, writer, filerClient, subDir, "", prefix, level+1)
|
||||
directoryCount += dirCount
|
||||
fileCount += fCount
|
||||
|
||||
@@ -60,7 +60,12 @@ func NewTopology(id string, seq sequence.Sequencer, volumeSizeLimit uint64, puls
|
||||
|
||||
func (t *Topology) IsLeader() bool {
|
||||
if t.RaftServer != nil {
|
||||
return t.RaftServer.State() == raft.Leader
|
||||
if t.RaftServer.State() == raft.Leader {
|
||||
return true
|
||||
}
|
||||
if t.RaftServer.Leader() == "" {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -75,7 +80,7 @@ func (t *Topology) Leader() (string, error) {
|
||||
|
||||
if l == "" {
|
||||
// We are a single node cluster, we are the leader
|
||||
return t.RaftServer.Name(), errors.New("Raft Server not initialized!")
|
||||
return t.RaftServer.Name(), nil
|
||||
}
|
||||
|
||||
return l, nil
|
||||
@@ -152,7 +157,7 @@ func (t *Topology) ListCollections(includeNormalVolumes, includeEcVolumes bool)
|
||||
t.ecShardMapLock.RUnlock()
|
||||
}
|
||||
|
||||
for k, _ := range mapOfCollections {
|
||||
for k := range mapOfCollections {
|
||||
ret = append(ret, k)
|
||||
}
|
||||
return ret
|
||||
|
||||
@@ -1,5 +1,10 @@
|
||||
package util
|
||||
|
||||
import (
|
||||
"crypto/md5"
|
||||
"io"
|
||||
)
|
||||
|
||||
// big endian
|
||||
|
||||
func BytesToUint64(b []byte) (v uint64) {
|
||||
@@ -43,3 +48,29 @@ func Uint16toBytes(b []byte, v uint16) {
|
||||
func Uint8toBytes(b []byte, v uint8) {
|
||||
b[0] = byte(v)
|
||||
}
|
||||
|
||||
// returns a 64 bit big int
|
||||
func HashStringToLong(dir string) (v int64) {
|
||||
h := md5.New()
|
||||
io.WriteString(h, dir)
|
||||
|
||||
b := h.Sum(nil)
|
||||
|
||||
v += int64(b[0])
|
||||
v <<= 8
|
||||
v += int64(b[1])
|
||||
v <<= 8
|
||||
v += int64(b[2])
|
||||
v <<= 8
|
||||
v += int64(b[3])
|
||||
v <<= 8
|
||||
v += int64(b[4])
|
||||
v <<= 8
|
||||
v += int64(b[5])
|
||||
v <<= 8
|
||||
v += int64(b[6])
|
||||
v <<= 8
|
||||
v += int64(b[7])
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
@@ -5,5 +5,5 @@ import (
|
||||
)
|
||||
|
||||
var (
|
||||
VERSION = fmt.Sprintf("%s %d.%d", sizeLimit, 1, 50)
|
||||
VERSION = fmt.Sprintf("%s %d.%d", sizeLimit, 1, 51)
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user