Skip to content

Commit a5e877f

Browse files
smkniazismkniazi
authored
Merge pull request #29 from smkniazi/GH-9
Improving the performance by skipping file staging for readonly workloads. Fixes GH-9
2 parents 4dbfcaf + c47cc62 commit a5e877f

16 files changed

+657
-343
lines changed

‎Dir.go

+82-58
Original file line numberDiff line numberDiff line change
@@ -17,24 +17,24 @@ import (
1717
)
1818

1919
// Encapsulates state and operations for directory node on the HDFS file system
20-
type Dir struct {
21-
FileSystem *FileSystem // Pointer to the owning filesystem
22-
Attrs Attrs // Cached attributes of the directory, TODO: add TTL
23-
Parent *Dir // Pointer to the parent directory (allows computing fully-qualified paths on demand)
24-
Entries map[string]*fs.Node // Cahed directory entries
25-
EntriesMutex sync.Mutex // Used to protect Entries
20+
type DirINode struct {
21+
FileSystem *FileSystem // Pointer to the owning filesystem
22+
Attrs Attrs // Cached attributes of the directory, TODO: add TTL
23+
Parent *DirINode // Pointer to the parent directory (allows computing fully-qualified paths on demand)
24+
Entries map[string]*fs.Node // Cahed directory entries
25+
mutex sync.Mutex // One read or write operation on a directory at a time
2626
}
2727

2828
// Verify that *Dir implements necesary FUSE interfaces
29-
var _ fs.Node = (*Dir)(nil)
30-
var _ fs.HandleReadDirAller = (*Dir)(nil)
31-
var _ fs.NodeStringLookuper = (*Dir)(nil)
32-
var _ fs.NodeMkdirer = (*Dir)(nil)
33-
var _ fs.NodeRemover = (*Dir)(nil)
34-
var _ fs.NodeRenamer = (*Dir)(nil)
29+
var _ fs.Node = (*DirINode)(nil)
30+
var _ fs.HandleReadDirAller = (*DirINode)(nil)
31+
var _ fs.NodeStringLookuper = (*DirINode)(nil)
32+
var _ fs.NodeMkdirer = (*DirINode)(nil)
33+
var _ fs.NodeRemover = (*DirINode)(nil)
34+
var _ fs.NodeRenamer = (*DirINode)(nil)
3535

3636
// Returns absolute path of the dir in HDFS namespace
37-
func (dir *Dir) AbsolutePath() string {
37+
func (dir *DirINode) AbsolutePath() string {
3838
if dir.Parent == nil {
3939
return dir.FileSystem.SrcDir
4040
} else {
@@ -43,7 +43,7 @@ func (dir *Dir) AbsolutePath() string {
4343
}
4444

4545
// Returns absolute path of the child item of this directory
46-
func (dir *Dir) AbsolutePathForChild(name string) string {
46+
func (dir *DirINode) AbsolutePathForChild(name string) string {
4747
path := dir.AbsolutePath()
4848
if path != "/" {
4949
path = path + "/"
@@ -52,7 +52,9 @@ func (dir *Dir) AbsolutePathForChild(name string) string {
5252
}
5353

5454
// Responds on FUSE request to get directory attributes
55-
func (dir *Dir) Attr(ctx context.Context, a *fuse.Attr) error {
55+
func (dir *DirINode) Attr(ctx context.Context, a *fuse.Attr) error {
56+
dir.lockMutex()
57+
defer dir.unlockMutex()
5658
if dir.Parent != nil && dir.FileSystem.Clock.Now().After(dir.Attrs.Expires) {
5759
err := dir.Parent.LookupAttrs(dir.Attrs.Name, &dir.Attrs)
5860
if err != nil {
@@ -63,54 +65,47 @@ func (dir *Dir) Attr(ctx context.Context, a *fuse.Attr) error {
6365
return dir.Attrs.Attr(a)
6466
}
6567

66-
func (dir *Dir) EntriesGet(name string) *fs.Node {
67-
dir.EntriesMutex.Lock()
68-
defer dir.EntriesMutex.Unlock()
68+
func (dir *DirINode) EntriesGet(name string) *fs.Node {
6969
if dir.Entries == nil {
7070
dir.Entries = make(map[string]*fs.Node)
7171
return nil
7272
}
7373
return dir.Entries[name]
7474
}
7575

76-
func (dir *Dir) EntriesSet(name string, node *fs.Node) {
77-
dir.EntriesMutex.Lock()
78-
defer dir.EntriesMutex.Unlock()
79-
76+
func (dir *DirINode) EntriesSet(name string, node *fs.Node) {
8077
if dir.Entries == nil {
8178
dir.Entries = make(map[string]*fs.Node)
8279
}
8380

8481
dir.Entries[name] = node
8582
}
8683

87-
func (dir *Dir) EntriesUpdate(name string, attr Attrs) {
88-
dir.EntriesMutex.Lock()
89-
defer dir.EntriesMutex.Unlock()
90-
84+
func (dir *DirINode) EntriesUpdate(name string, attr Attrs) {
9185
if dir.Entries == nil {
9286
dir.Entries = make(map[string]*fs.Node)
9387
}
9488

9589
if node, ok := dir.Entries[name]; ok {
96-
if fnode, ok := (*node).(*File); ok {
90+
if fnode, ok := (*node).(*FileINode); ok {
9791
fnode.Attrs = attr
98-
} else if dnode, ok := (*node).(*Dir); ok {
92+
} else if dnode, ok := (*node).(*DirINode); ok {
9993
dnode.Attrs = attr
10094
}
10195
}
10296
}
10397

104-
func (dir *Dir) EntriesRemove(name string) {
105-
dir.EntriesMutex.Lock()
106-
defer dir.EntriesMutex.Unlock()
98+
func (dir *DirINode) EntriesRemove(name string) {
10799
if dir.Entries != nil {
108100
delete(dir.Entries, name)
109101
}
110102
}
111103

112104
// Responds on FUSE request to lookup the directory
113-
func (dir *Dir) Lookup(ctx context.Context, name string) (fs.Node, error) {
105+
func (dir *DirINode) Lookup(ctx context.Context, name string) (fs.Node, error) {
106+
dir.lockMutex()
107+
defer dir.unlockMutex()
108+
114109
if !dir.FileSystem.IsPathAllowed(dir.AbsolutePathForChild(name)) {
115110
return nil, fuse.ENOENT
116111
}
@@ -126,7 +121,7 @@ func (dir *Dir) Lookup(ctx context.Context, name string) (fs.Node, error) {
126121
if err != nil {
127122
return nil, err
128123
}
129-
zipFile, ok := zipFileNode.(*File)
124+
zipFile, ok := zipFileNode.(*FileINode)
130125
if !ok {
131126
return nil, fuse.ENOENT
132127
}
@@ -146,11 +141,14 @@ func (dir *Dir) Lookup(ctx context.Context, name string) (fs.Node, error) {
146141
}
147142

148143
// Responds on FUSE request to read directory
149-
func (dir *Dir) ReadDirAll(ctx context.Context) ([]fuse.Dirent, error) {
144+
func (dir *DirINode) ReadDirAll(ctx context.Context) ([]fuse.Dirent, error) {
145+
dir.lockMutex()
146+
defer dir.unlockMutex()
147+
150148
absolutePath := dir.AbsolutePath()
151149
loginfo("Read directory", Fields{Operation: ReadDir, Path: absolutePath})
152150

153-
allAttrs, err := dir.FileSystem.HdfsAccessor.ReadDir(absolutePath)
151+
allAttrs, err := dir.FileSystem.getDFSConnector().ReadDir(absolutePath)
154152
if err != nil {
155153
logwarn("Failed to list DFS directory", Fields{Operation: ReadDir, Path: absolutePath, Error: err})
156154
return nil, err
@@ -184,12 +182,12 @@ func (dir *Dir) ReadDirAll(ctx context.Context) ([]fuse.Dirent, error) {
184182
}
185183

186184
// Creates typed node (Dir or File) from the attributes
187-
func (dir *Dir) NodeFromAttrs(attrs Attrs) fs.Node {
185+
func (dir *DirINode) NodeFromAttrs(attrs Attrs) fs.Node {
188186
var node fs.Node
189187
if (attrs.Mode & os.ModeDir) == 0 {
190-
node = &File{FileSystem: dir.FileSystem, Parent: dir, Attrs: attrs}
188+
node = &FileINode{FileSystem: dir.FileSystem, Parent: dir, Attrs: attrs}
191189
} else {
192-
node = &Dir{FileSystem: dir.FileSystem, Parent: dir, Attrs: attrs}
190+
node = &DirINode{FileSystem: dir.FileSystem, Parent: dir, Attrs: attrs}
193191
}
194192

195193
if n := dir.EntriesGet(attrs.Name); n != nil {
@@ -202,9 +200,10 @@ func (dir *Dir) NodeFromAttrs(attrs Attrs) fs.Node {
202200
}
203201

204202
// Performs Stat() query on the backend
205-
func (dir *Dir) LookupAttrs(name string, attrs *Attrs) error {
203+
func (dir *DirINode) LookupAttrs(name string, attrs *Attrs) error {
204+
206205
var err error
207-
*attrs, err = dir.FileSystem.HdfsAccessor.Stat(path.Join(dir.AbsolutePath(), name))
206+
*attrs, err = dir.FileSystem.getDFSConnector().Stat(path.Join(dir.AbsolutePath(), name))
208207
if err != nil {
209208
// It is a warning as each time new file write tries to stat if the file exists
210209
loginfo("Stat failed", Fields{Operation: Stat, Path: path.Join(dir.AbsolutePath(), name), Error: err})
@@ -218,72 +217,87 @@ func (dir *Dir) LookupAttrs(name string, attrs *Attrs) error {
218217
}
219218

220219
// Responds on FUSE Mkdir request
221-
func (dir *Dir) Mkdir(ctx context.Context, req *fuse.MkdirRequest) (fs.Node, error) {
222-
err := dir.FileSystem.HdfsAccessor.Mkdir(dir.AbsolutePathForChild(req.Name), req.Mode)
220+
func (dir *DirINode) Mkdir(ctx context.Context, req *fuse.MkdirRequest) (fs.Node, error) {
221+
dir.lockMutex()
222+
defer dir.unlockMutex()
223+
224+
err := dir.FileSystem.getDFSConnector().Mkdir(dir.AbsolutePathForChild(req.Name), req.Mode)
223225
if err != nil {
224226
return nil, err
225227
}
226228
return dir.NodeFromAttrs(Attrs{Name: req.Name, Mode: req.Mode | os.ModeDir}), nil
227229
}
228230

229231
// Responds on FUSE Create request
230-
func (dir *Dir) Create(ctx context.Context, req *fuse.CreateRequest, resp *fuse.CreateResponse) (fs.Node, fs.Handle, error) {
231-
loginfo("Creating a new file", Fields{Operation: Create, Path: dir.AbsolutePathForChild(req.Name), Mode: req.Mode, Flags: req.Flags})
232+
func (dir *DirINode) Create(ctx context.Context, req *fuse.CreateRequest, resp *fuse.CreateResponse) (fs.Node, fs.Handle, error) {
233+
dir.lockMutex()
234+
defer dir.unlockMutex()
232235

233-
file := dir.NodeFromAttrs(Attrs{Name: req.Name, Mode: req.Mode}).(*File)
234-
handle, err := NewFileHandle(file, false, req.Flags)
236+
loginfo("Creating a new file", Fields{Operation: Create, Path: dir.AbsolutePathForChild(req.Name), Mode: req.Mode, Flags: req.Flags})
237+
file := dir.NodeFromAttrs(Attrs{Name: req.Name, Mode: req.Mode}).(*FileINode)
238+
handle, err := file.NewFileHandle(false, req.Flags)
235239
if err != nil {
236240
logerror("File creation failed", Fields{Operation: Create, Path: dir.AbsolutePathForChild(req.Name), Mode: req.Mode, Flags: req.Flags, Error: err})
241+
//TODO remove the entry from the cache
237242
return nil, nil, err
238243
}
239244
file.AddHandle(handle)
240245
return file, handle, nil
241246
}
242247

243248
// Responds on FUSE Remove request
244-
func (dir *Dir) Remove(ctx context.Context, req *fuse.RemoveRequest) error {
249+
func (dir *DirINode) Remove(ctx context.Context, req *fuse.RemoveRequest) error {
250+
dir.lockMutex()
251+
defer dir.unlockMutex()
252+
245253
path := dir.AbsolutePathForChild(req.Name)
246254
loginfo("Removing path", Fields{Operation: Remove, Path: path})
247-
err := dir.FileSystem.HdfsAccessor.Remove(path)
255+
err := dir.FileSystem.getDFSConnector().Remove(path)
248256
if err == nil {
249257
dir.EntriesRemove(req.Name)
250258
} else {
251-
logerror("Failed to remove path", Fields{Operation: Remove, Path: path, Error: err})
259+
logwarn("Failed to remove path", Fields{Operation: Remove, Path: path, Error: err})
252260
}
253261
return err
254262
}
255263

256264
// Responds on FUSE Rename request
257-
func (dir *Dir) Rename(ctx context.Context, req *fuse.RenameRequest, newDir fs.Node) error {
265+
func (dir *DirINode) Rename(ctx context.Context, req *fuse.RenameRequest, newDir fs.Node) error {
266+
dir.lockMutex()
267+
defer dir.unlockMutex()
268+
258269
oldPath := dir.AbsolutePathForChild(req.OldName)
259-
newPath := newDir.(*Dir).AbsolutePathForChild(req.NewName)
270+
newPath := newDir.(*DirINode).AbsolutePathForChild(req.NewName)
260271
loginfo("Renaming to "+newPath, Fields{Operation: Rename, Path: oldPath})
261-
err := dir.FileSystem.HdfsAccessor.Rename(oldPath, newPath)
272+
err := dir.FileSystem.getDFSConnector().Rename(oldPath, newPath)
262273
if err == nil {
263274
// Upon successful rename, updating in-memory representation of the file entry
264275
if node := dir.EntriesGet(req.OldName); node != nil {
265-
if fnode, ok := (*node).(*File); ok {
276+
if fnode, ok := (*node).(*FileINode); ok {
266277
fnode.Attrs.Name = req.NewName
267-
} else if dnode, ok := (*node).(*Dir); ok {
278+
} else if dnode, ok := (*node).(*DirINode); ok {
268279
dnode.Attrs.Name = req.NewName
269280
}
270281
dir.EntriesRemove(req.OldName)
271-
newDir.(*Dir).EntriesSet(req.NewName, node)
282+
newDir.(*DirINode).EntriesSet(req.NewName, node)
272283
}
273284
}
274285
return err
275286
}
276287

277288
// Responds on FUSE Chmod request
278-
func (dir *Dir) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *fuse.SetattrResponse) error {
289+
func (dir *DirINode) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *fuse.SetattrResponse) error {
290+
dir.lockMutex()
291+
defer dir.unlockMutex()
292+
279293
// Get the filepath, so chmod in hdfs can work
280294
path := dir.AbsolutePath()
281295
var err error
282296

283297
if req.Valid.Mode() {
284298
loginfo("Setting attributes", Fields{Operation: Chmod, Path: path, Mode: req.Mode})
285299
(func() {
286-
err = dir.FileSystem.HdfsAccessor.Chmod(path, req.Mode)
300+
err = dir.FileSystem.getDFSConnector().Chmod(path, req.Mode)
287301
if err != nil {
288302
return
289303
}
@@ -310,7 +324,7 @@ func (dir *Dir) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *fus
310324

311325
loginfo("Setting attributes", Fields{Operation: Chown, Path: path, User: u, UID: owner, GID: group})
312326
(func() {
313-
err = dir.FileSystem.HdfsAccessor.Chown(path, owner, group)
327+
err = dir.FileSystem.getDFSConnector().Chown(path, owner, group)
314328
if err != nil {
315329
return
316330
}
@@ -326,3 +340,13 @@ func (dir *Dir) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *fus
326340

327341
return err
328342
}
343+
344+
var dirLockTime time.Time = time.Time{}
345+
346+
func (dir *DirINode) lockMutex() {
347+
dir.mutex.Lock()
348+
}
349+
350+
func (dir *DirINode) unlockMutex() {
351+
dir.mutex.Unlock()
352+
}

0 commit comments

Comments
 (0)