Repository: juicedata/juicefs
Pull request #2213 (2022-06-13 14:03:17)

hadoop:   fix sdk in chinese path


Patch:
diff --git a/go.mod b/go.mod
index 817ea624..6ad72305 100644
--- a/go.mod
+++ b/go.mod
@@ -229,7 +229,7 @@ require (
 	xorm.io/builder v0.3.7 // indirect
 )
 
-replace github.com/minio/minio v0.0.0-20210206053228-97fe57bba92c => github.com/juicedata/minio v0.0.0-20220324072921-2057662dce6d
+replace github.com/minio/minio v0.0.0-20210206053228-97fe57bba92c => github.com/juicedata/minio v0.0.0-20220613143934-cee0571a1b03
 
 replace github.com/hanwen/go-fuse/v2 v2.1.1-0.20210611132105-24a1dfe6b4f8 => github.com/juicedata/go-fuse/v2 v2.1.1-0.20210926080226-cfe1ec802a7f
 
diff --git a/go.sum b/go.sum
index f3544933..7ccbba25 100644
--- a/go.sum
+++ b/go.sum
@@ -609,8 +609,8 @@ github.com/juicedata/go-fuse/v2 v2.1.1-0.20210926080226-cfe1ec802a7f h1:EhufxJV9
 github.com/juicedata/go-fuse/v2 v2.1.1-0.20210926080226-cfe1ec802a7f/go.mod h1:oRyA5eK+pvJyv5otpO/DgccS8y/RvYMaO00GgRLGryc=
 github.com/juicedata/godaemon v0.0.0-20210629045518-3da5144a127d h1:kpQMvNZJKGY3PTt7OSoahYc4nM0HY67SvK0YyS0GLwA=
 github.com/juicedata/godaemon v0.0.0-20210629045518-3da5144a127d/go.mod h1:dlxKkLh3qAIPtgr2U/RVzsZJDuXA1ffg+Njikfmhvgw=
-github.com/juicedata/minio v0.0.0-20220324072921-2057662dce6d h1:Jkul/jGraZy7U3o8xWztM3cf3e1uiKIiPaCg+cRPaz8=
-github.com/juicedata/minio v0.0.0-20220324072921-2057662dce6d/go.mod h1:ILQUt98Xz3GF9KgJKeGQ81c0RFaRSrUZmZ8oSb/7QxU=
+github.com/juicedata/minio v0.0.0-20220613143934-cee0571a1b03 h1:+0jiEkr6qE5Zm2+LilAv5pV/+V+V1XZ5KUv0pmXZHhQ=
+github.com/juicedata/minio v0.0.0-20220613143934-cee0571a1b03/go.mod h1:ILQUt98Xz3GF9KgJKeGQ81c0RFaRSrUZmZ8oSb/7QxU=
 github.com/juicedata/mpb/v7 v7.0.4-0.20220411070607-092927ed0122 h1:2YZ6nb3kO38/WIG5L5dwrlrElXtCNKRIdrXm2c9meOk=
 github.com/juicedata/mpb/v7 v7.0.4-0.20220411070607-092927ed0122/go.mod h1:NXGsfPGx6G2JssqvEcULtDqUrxuuYs4llpv8W6ZUpzk=
 github.com/juju/ratelimit v1.0.1 h1:+7AIFJVQ0EQgq/K9+0Krm7m530Du7tIz0METWzN0RgY=
diff --git a/pkg/fs/fs.go b/pkg/fs/fs.go
index ad5dc7b0..85da5c61 100644
--- a/pkg/fs/fs.go
+++ b/pkg/fs/fs.go
@@ -438,7 +438,7 @@ func (fs *FileSystem) Rmr(ctx meta.Context, p string) (err syscall.Errno) {
 	if err != 0 {
 		return
 	}
-	err = meta.Remove(fs.m, ctx, parent.inode, path.Base(p), nil)
+	err = fs.m.Remove(ctx, parent.inode, path.Base(p), nil)
 	fs.invalidateEntry(parent.inode, path.Base(p))
 	return
 }
diff --git a/pkg/meta/base.go b/pkg/meta/base.go
index 6ad5a043..92a1b664 100644
--- a/pkg/meta/base.go
+++ b/pkg/meta/base.go
@@ -105,14 +105,14 @@ type baseMeta struct {
 	en engine
 }
 
-func newBaseMeta(addr string, conf *Config) baseMeta {
+func newBaseMeta(addr string, conf *Config) *baseMeta {
 	if conf.Retries == 0 {
 		conf.Retries = 10
 	}
 	if conf.Heartbeat == 0 {
 		conf.Heartbeat = 12 * time.Second
 	}
-	return baseMeta{
+	return &baseMeta{
 		addr:         utils.RemovePassword(addr),
 		conf:         conf,
 		root:         1,
@@ -127,6 +127,10 @@ func newBaseMeta(addr string, conf *Config) baseMeta {
 	}
 }
 
+func (m *baseMeta) getBase() *baseMeta {
+	return m
+}
+
 func (m *baseMeta) checkRoot(inode Ino) Ino {
 	switch inode {
 	case 0:
diff --git a/pkg/meta/interface.go b/pkg/meta/interface.go
index 72154eb6..ca42d873 100644
--- a/pkg/meta/interface.go
+++ b/pkg/meta/interface.go
@@ -337,6 +337,8 @@ type Meta interface {
 	CompactAll(ctx Context, bar *utils.Bar) syscall.Errno
 	// ListSlices returns all slices used by all files.
 	ListSlices(ctx Context, slices map[Ino][]Slice, delete bool, showProgress func()) syscall.Errno
+	// Remove all files and directories recursively.
+	Remove(ctx Context, parent Ino, name string, count *uint64) syscall.Errno
 
 	// OnMsg add a callback for the given message type.
 	OnMsg(mtype uint32, cb MsgCallback)
@@ -344,6 +346,9 @@ type Meta interface {
 	// Dump the tree under root, which may be modified by checkRoot
 	DumpMeta(w io.Writer, root Ino) error
 	LoadMeta(r io.Reader) error
+
+	// getBase return the base engine.
+	getBase() *baseMeta
 }
 
 type Creator func(driver, addr string, conf *Config) (Meta, error)
diff --git a/pkg/meta/redis.go b/pkg/meta/redis.go
index f94ecabc..08d3f08d 100644
--- a/pkg/meta/redis.go
+++ b/pkg/meta/redis.go
@@ -72,7 +72,7 @@ import (
 */
 
 type redisMeta struct {
-	baseMeta
+	*baseMeta
 	rdb        redis.UniversalClient
 	prefix     string
 	shaLookup  string // The SHA returned by Redis for the loaded `scriptLookup`
diff --git a/pkg/meta/sql.go b/pkg/meta/sql.go
index 2975ecf5..e083e5bf 100644
--- a/pkg/meta/sql.go
+++ b/pkg/meta/sql.go
@@ -155,7 +155,7 @@ type delfile struct {
 }
 
 type dbMeta struct {
-	baseMeta
+	*baseMeta
 	db   *xorm.Engine
 	snap *dbSnap
 }
diff --git a/pkg/meta/tkv.go b/pkg/meta/tkv.go
index 91f2fe64..9ff8a970 100644
--- a/pkg/meta/tkv.go
+++ b/pkg/meta/tkv.go
@@ -61,7 +61,7 @@ type tkvClient interface {
 }
 
 type kvMeta struct {
-	baseMeta
+	*baseMeta
 	client tkvClient
 	snap   map[Ino]*DumpedEntry
 }
diff --git a/pkg/meta/utils.go b/pkg/meta/utils.go
index 33f7d149..8e7c80f2 100644
--- a/pkg/meta/utils.go
+++ b/pkg/meta/utils.go
@@ -222,67 +222,72 @@ func updateLocks(ls []plockRecord, nl plockRecord) []plockRecord {
 	return ls
 }
 
-func emptyDir(r Meta, ctx Context, inode Ino, count *uint64, concurrent chan int) syscall.Errno {
-	if st := r.Access(ctx, inode, 3, nil); st != 0 {
+func (m *baseMeta) emptyDir(ctx Context, inode Ino, count *uint64, concurrent chan int) syscall.Errno {
+	if st := m.Access(ctx, inode, 3, nil); st != 0 {
 		return st
 	}
-	var entries []*Entry
-	if st := r.Readdir(ctx, inode, 0, &entries); st != 0 {
-		return st
-	}
-	var wg sync.WaitGroup
-	var status syscall.Errno
-	// try directories first to increase parallel
-	var dirs int
-	for i, e := range entries {
-		if e.Attr.Typ == TypeDirectory {
-			entries[dirs], entries[i] = entries[i], entries[dirs]
-			dirs++
+	for {
+		var entries []*Entry
+		if st := m.en.doReaddir(ctx, inode, 0, &entries, 10000); st != 0 {
+			return st
 		}
-	}
-	for _, e := range entries {
-		if e.Inode == inode || len(e.Name) == 2 && string(e.Name) == ".." {
-			continue
+		if len(entries) == 0 {
+			return 0
+		}
+		var wg sync.WaitGroup
+		var status syscall.Errno
+		// try directories first to increase parallel
+		var dirs int
+		for i, e := range entries {
+			if e.Attr.Typ == TypeDirectory {
+				entries[dirs], entries[i] = entries[i], entries[dirs]
+				dirs++
+			}
 		}
-		if e.Attr.Typ == TypeDirectory {
-			select {
-			case concurrent <- 1:
-				wg.Add(1)
-				go func(child Ino, name string) {
-					defer wg.Done()
-					e := emptyEntry(r, ctx, inode, name, child, count, concurrent)
-					if e != 0 {
-						status = e
+		for i, e := range entries {
+			if e.Attr.Typ == TypeDirectory {
+				select {
+				case concurrent <- 1:
+					wg.Add(1)
+					go func(child Ino, name string) {
+						defer wg.Done()
+						e := m.emptyEntry(ctx, inode, name, child, count, concurrent)
+						if e != 0 {
+							status = e
+						}
+						<-concurrent
+					}(e.Inode, string(e.Name))
+				default:
+					if st := m.emptyEntry(ctx, inode, string(e.Name), e.Inode, count, concurrent); st != 0 {
+						return st
 					}
-					<-concurrent
-				}(e.Inode, string(e.Name))
-			default:
-				if st := emptyEntry(r, ctx, inode, string(e.Name), e.Inode, count, concurrent); st != 0 {
+				}
+			} else {
+				if count != nil {
+					atomic.AddUint64(count, 1)
+				}
+				if st := m.Unlink(ctx, inode, string(e.Name)); st != 0 {
 					return st
 				}
 			}
-		} else {
-			if count != nil {
-				atomic.AddUint64(count, 1)
-			}
-			if st := r.Unlink(ctx, inode, string(e.Name)); st != 0 {
-				return st
+			if ctx.Canceled() {
+				return syscall.EINTR
 			}
+			entries[i] = nil // release memory
 		}
-		if ctx.Canceled() {
-			return syscall.EINTR
+		wg.Wait()
+		if status != 0 {
+			return status
 		}
 	}
-	wg.Wait()
-	return status
 }
 
-func emptyEntry(r Meta, ctx Context, parent Ino, name string, inode Ino, count *uint64, concurrent chan int) syscall.Errno {
-	st := emptyDir(r, ctx, inode, count, concurrent)
+func (m *baseMeta) emptyEntry(ctx Context, parent Ino, name string, inode Ino, count *uint64, concurrent chan int) syscall.Errno {
+	st := m.emptyDir(ctx, inode, count, concurrent)
 	if st == 0 {
-		st = r.Rmdir(ctx, parent, name)
+		st = m.Rmdir(ctx, parent, name)
 		if st == syscall.ENOTEMPTY {
-			st = emptyEntry(r, ctx, parent, name, inode, count, concurrent)
+			st = m.emptyEntry(ctx, parent, name, inode, count, concurrent)
 		} else if count != nil {
 			atomic.AddUint64(count, 1)
 		}
@@ -290,23 +295,24 @@ func emptyEntry(r Meta, ctx Context, parent Ino, name string, inode Ino, count *
 	return st
 }
 
-func Remove(r Meta, ctx Context, parent Ino, name string, count *uint64) syscall.Errno {
-	if st := r.Access(ctx, parent, 3, nil); st != 0 {
+func (m *baseMeta) Remove(ctx Context, parent Ino, name string, count *uint64) syscall.Errno {
+	parent = m.checkRoot(parent)
+	if st := m.Access(ctx, parent, 3, nil); st != 0 {
 		return st
 	}
 	var inode Ino
 	var attr Attr
-	if st := r.Lookup(ctx, parent, name, &inode, &attr); st != 0 {
+	if st := m.Lookup(ctx, parent, name, &inode, &attr); st != 0 {
 		return st
 	}
 	if attr.Typ != TypeDirectory {
 		if count != nil {
 			atomic.AddUint64(count, 1)
 		}
-		return r.Unlink(ctx, parent, name)
+		return m.Unlink(ctx, parent, name)
 	}
 	concurrent := make(chan int, 50)
-	return emptyEntry(r, ctx, parent, name, inode, count, concurrent)
+	return m.emptyEntry(ctx, parent, name, inode, count, concurrent)
 }
 
 func GetSummary(r Meta, ctx Context, inode Ino, summary *Summary, recursive bool) syscall.Errno {
diff --git a/pkg/vfs/internal.go b/pkg/vfs/internal.go
index 9267e51d..eede654d 100644
--- a/pkg/vfs/internal.go
+++ b/pkg/vfs/internal.go
@@ -192,7 +192,7 @@ func (v *VFS) handleInternalMsg(ctx meta.Context, cmd uint32, r *utils.Buffer, d
 		go func() {
 			inode := Ino(r.Get64())
 			name := string(r.Get(int(r.Get8())))
-			st = meta.Remove(v.Meta, ctx, inode, name, &count)
+			st = v.Meta.Remove(ctx, inode, name, &count)
 			if st != 0 {
 				logger.Errorf("remove %d/%s: %s", inode, name, st)
 			}
diff --git a/sdk/java/src/main/java/io/juicefs/JuiceFileSystemImpl.java b/sdk/java/src/main/java/io/juicefs/JuiceFileSystemImpl.java
index 3e3e6f44..697f697f 100644
--- a/sdk/java/src/main/java/io/juicefs/JuiceFileSystemImpl.java
+++ b/sdk/java/src/main/java/io/juicefs/JuiceFileSystemImpl.java
@@ -530,7 +530,12 @@ public class JuiceFileSystemImpl extends FileSystem {
     File libFile = new File(dir, name);
 
     URL res = null;
-    String jarPath = JuiceFileSystemImpl.class.getProtectionDomain().getCodeSource().getLocation().getPath();
+    String jarPath;
+    try {
+      jarPath = JuiceFileSystemImpl.class.getProtectionDomain().getCodeSource().getLocation().toURI().getPath();
+    } catch (URISyntaxException e) {
+      throw new IOException("get jar path failed", e);
+    }
     Enumeration<URL> resources = JuiceFileSystemImpl.class.getClassLoader().getResources(resource);
     while (resources.hasMoreElements()) {
       res = resources.nextElement();


Test patch:
diff --git a/pkg/meta/base_test.go b/pkg/meta/base_test.go
index 53f90e09..1eab3cb7 100644
--- a/pkg/meta/base_test.go
+++ b/pkg/meta/base_test.go
@@ -55,15 +55,6 @@ func testMeta(t *testing.T, m Meta) {
 	if err := m.Reset(); err != nil {
 		t.Fatalf("reset meta: %s", err)
 	}
-	var base *baseMeta
-	switch m := m.(type) {
-	case *redisMeta:
-		base = &m.baseMeta
-	case *dbMeta:
-		base = &m.baseMeta
-	case *kvMeta:
-		base = &m.baseMeta
-	}
 	testMetaClient(t, m)
 	testTruncateAndDelete(t, m)
 	testTrash(t, m)
@@ -78,6 +69,7 @@ func testMeta(t *testing.T, m Meta) {
 	testCopyFileRange(t, m)
 	testCloseSession(t, m)
 	testConcurrentDir(t, m)
+	base := m.getBase()
 	base.conf.OpenCache = time.Second
 	base.of.expire = time.Second
 	testOpenCache(t, m)
@@ -116,19 +108,9 @@ func testMetaClient(t *testing.T, m Meta) {
 	if err != nil || len(ses) != 1 {
 		t.Fatalf("list sessions %+v: %s", ses, err)
 	}
-	switch r := m.(type) {
-	case *redisMeta:
-		if r.sid != ses[0].Sid {
-			t.Fatalf("my sid %d != registered sid %d", r.sid, ses[0].Sid)
-		}
-	case *dbMeta:
-		if r.sid != ses[0].Sid {
-			t.Fatalf("my sid %d != registered sid %d", r.sid, ses[0].Sid)
-		}
-	case *kvMeta:
-		if r.sid != ses[0].Sid {
-			t.Fatalf("my sid %d != registered sid %d", r.sid, ses[0].Sid)
-		}
+	base := m.getBase()
+	if base.sid != ses[0].Sid {
+		t.Fatalf("my sid %d != registered sid %d", base.sid, ses[0].Sid)
 	}
 	go m.CleanStaleSessions()
 
@@ -737,7 +719,7 @@ func testRemove(t *testing.T, m Meta) {
 	if st := m.Create(ctx, 1, "f", 0644, 0, 0, &inode, attr); st != 0 {
 		t.Fatalf("create f: %s", st)
 	}
-	if st := Remove(m, ctx, 1, "f", nil); st != 0 {
+	if st := m.Remove(ctx, 1, "f", nil); st != 0 {
 		t.Fatalf("rmr f: %s", st)
 	}
 	if st := m.Mkdir(ctx, 1, "d", 0755, 0, 0, &parent, attr); st != 0 {
@@ -766,7 +748,7 @@ func testRemove(t *testing.T, m Meta) {
 	} else if len(entries) != 4099 {
 		t.Fatalf("entries: %d", len(entries))
 	}
-	if st := Remove(m, ctx, 1, "d", nil); st != 0 {
+	if st := m.Remove(ctx, 1, "d", nil); st != 0 {
 		t.Fatalf("rmr d: %s", st)
 	}
 }
@@ -1112,15 +1094,7 @@ func testCloseSession(t *testing.T, m Meta) {
 	if st := m.Unlink(ctx, 1, "f"); st != 0 {
 		t.Fatalf("unlink f: %s", st)
 	}
-	var sid uint64
-	switch m := m.(type) {
-	case *redisMeta:
-		sid = m.sid
-	case *dbMeta:
-		sid = m.sid
-	case *kvMeta:
-		sid = m.sid
-	}
+	sid := m.getBase().sid
 	s, err := m.GetSession(sid, true)
 	if err != nil {
 		t.Fatalf("get session: %s", err)
@@ -1245,14 +1219,7 @@ func testTrash(t *testing.T, m Meta) {
 	if st := m.Rename(ctx2, TrashInode+1, "d", 1, "f", 0, &inode, attr); st != syscall.EPERM {
 		t.Fatalf("rename d -> f: %s", st)
 	}
-	switch bm := m.(type) {
-	case *redisMeta:
-		bm.doCleanupTrash(true)
-	case *dbMeta:
-		bm.doCleanupTrash(true)
-	case *kvMeta:
-		bm.doCleanupTrash(true)
-	}
+	m.getBase().doCleanupTrash(true)
 	if st := m.GetAttr(ctx2, TrashInode+1, attr); st != syscall.ENOENT {
 		t.Fatalf("getattr: %s", st)
 	}
diff --git a/pkg/meta/benchmarks_test.go b/pkg/meta/benchmarks_test.go
index 98d60db5..3b2e03a2 100644
--- a/pkg/meta/benchmarks_test.go
+++ b/pkg/meta/benchmarks_test.go
@@ -116,7 +116,7 @@ func BenchmarkReadSliceBuf(b *testing.B) {
 
 func prepareParent(m Meta, name string, inode *Ino) error {
 	ctx := Background
-	if err := Remove(m, ctx, 1, name, nil); err != 0 && err != syscall.ENOENT {
+	if err := m.Remove(ctx, 1, name, nil); err != 0 && err != syscall.ENOENT {
 		return fmt.Errorf("remove: %s", err)
 	}
 	if err := m.Mkdir(ctx, 1, name, 0755, 0, 0, inode, nil); err != 0 {