From b3fa7b3fea2dad1a6bdb9cdcee44849fe5278ab5 Mon Sep 17 00:00:00 2001 From: PeyChou Date: Thu, 21 Mar 2024 14:38:54 +0800 Subject: [PATCH 01/15] =?UTF-8?q?=E4=BC=98=E5=8C=96cache=E5=8F=96key?= =?UTF-8?q?=E6=80=A7=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pkg/fs/client/cache/cache.go | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/pkg/fs/client/cache/cache.go b/pkg/fs/client/cache/cache.go index 898c4f30..5348ae6a 100644 --- a/pkg/fs/client/cache/cache.go +++ b/pkg/fs/client/cache/cache.go @@ -17,11 +17,10 @@ limitations under the License. package cache import ( - "fmt" "io" - "path" "path/filepath" "strconv" + "strings" "sync" "syscall" "time" @@ -241,7 +240,15 @@ func (r *rCache) key(index int) string { r.store.Unlock() } hash := utils.KeyHash(keyID) - return path.Clean(fmt.Sprintf("blocks/%d/%v_%v", hash%256, keyID, index)) + + var sb strings.Builder + sb.WriteString("blocks/") + sb.WriteString(strconv.Itoa(int(hash % 256))) + sb.WriteString("/") + sb.WriteString(keyID) + sb.WriteString("_") + sb.WriteString(strconv.Itoa(index)) + return sb.String() } func (r *rCache) readCache(buf []byte, key string, off int) (int, bool) { From 9baf1df24055bbf20d400732a0dc188dfb7985db Mon Sep 17 00:00:00 2001 From: PeyChou Date: Mon, 25 Mar 2024 16:30:19 +0800 Subject: [PATCH 02/15] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E6=97=A0=E9=99=90?= =?UTF-8?q?=E5=BE=AA=E7=8E=AFbug?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pkg/fs/client/meta/meta_kv.go | 23 ++++++++++------------- pkg/fs/client/vfs/reader.go | 10 ++++++++++ 2 files changed, 20 insertions(+), 13 deletions(-) diff --git a/pkg/fs/client/meta/meta_kv.go b/pkg/fs/client/meta/meta_kv.go index 90f9c070..da64e872 100644 --- a/pkg/fs/client/meta/meta_kv.go +++ b/pkg/fs/client/meta/meta_kv.go @@ -1846,20 +1846,17 @@ func (m *kvMeta) Open(ctx *Context, inode Ino, flags uint32, attr *Attr) (ufslib return err } } else { - attrTmp := &Attr{} - err := m.GetAttr(ctx, inode, attrTmp) - if err != syscall.F_OK { - log.Errorf("get attr err %v", err) - return err + for i := 0; i < 3; i++ { + log.Errorf("open inode[%v] not exist [%v]", inode, i) + time.Sleep(time.Duration(i*100) * time.Millisecond) + a = tx.Get(m.inodeKey(inode)) + if a != nil { + break + } } - a = tx.Get(m.inodeKey(inode)) - m.parseInode(a, inodeItem_) - if !m.inodeItemExpired(*inodeItem_) { - log.Debugf("open inodeItem cache %+v and attr %+v", *inodeItem_, inodeItem_.attr) - *attr = inodeItem_.attr - inodeItem_.fileHandles += 1 - errSet := tx.Set(m.inodeKey(inode), m.marshalInode(inodeItem_)) - return errSet + if a == nil { + log.Errorf("open emtpy inode") + return syscall.ENOENT } } info, err := ufs_.GetAttr(newPath) diff --git a/pkg/fs/client/vfs/reader.go b/pkg/fs/client/vfs/reader.go index 681acdd7..ac42a9be 100644 --- a/pkg/fs/client/vfs/reader.go +++ b/pkg/fs/client/vfs/reader.go @@ -95,6 +95,7 @@ func (fh *fileReader) Read(buf []byte, off uint64) (int, syscall.Errno) { var err error var nread int bufSize := len(buf) + exceedFirstTime := time.Now() if fh.reader.store != nil { reader := fh.reader.store.NewReader(fh.path, int(fh.length), fh.flags, fh.ufs, fh.buffersCache, fh.reader.bufferPool, fh.seqReadAmount, fh.bufferMapLock) @@ -106,6 +107,15 @@ func (fh *fileReader) Read(buf []byte, off uint64) (int, syscall.Errno) { 越界的情况，返回0，如off>=length||len(buf)==0 */ nread, err = reader.ReadAt(buf[bytesRead:], int64(off)) + if nread == 0 && err == nil { + if time.Since(exceedFirstTime) > 5*time.Minute { + log.Errorf("nread time out is zero len buf %v off %v bytesRead %v size %v path[%s]", len(buf), off, bytesRead, fh.size, fh.path) + return 0, syscall.EBADF + } + } else { + exceedFirstTime = time.Now() + } + if err != nil && err != syscall.ENOMEM && err != io.EOF && err != io.ErrUnexpectedEOF { log.Errorf("reader failed: %v", err) // 重试的时候稍微等待一下 From 06617aa871982e8df6cc3dc9f53a667450f17806 Mon Sep 17 00:00:00 2001 From: PeyChou Date: Mon, 25 Mar 2024 16:30:19 +0800 Subject: [PATCH 03/15] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E6=97=A0=E9=99=90?= =?UTF-8?q?=E5=BE=AA=E7=8E=AFbug?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pkg/fs/client/meta/meta_kv.go | 1 - 1 file changed, 1 deletion(-) diff --git a/pkg/fs/client/meta/meta_kv.go b/pkg/fs/client/meta/meta_kv.go index da64e872..7f83f539 100644 --- a/pkg/fs/client/meta/meta_kv.go +++ b/pkg/fs/client/meta/meta_kv.go @@ -1918,7 +1918,6 @@ func (m *kvMeta) Close(ctx *Context, inode Ino) syscall.Errno { } } return nil - }) return utils.ToSyscallErrno(err) } From 264dd55d0b208021f10cdfb474161420db3e193d Mon Sep 17 00:00:00 2001 From: PeyChou Date: Mon, 25 Mar 2024 16:30:19 +0800 Subject: [PATCH 04/15] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E6=97=A0=E9=99=90?= =?UTF-8?q?=E5=BE=AA=E7=8E=AFbug?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pkg/fs/client/fs/file_test.go | 35 +++++++++++++++++++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/pkg/fs/client/fs/file_test.go b/pkg/fs/client/fs/file_test.go index fe639ce0..d4cff698 100644 --- a/pkg/fs/client/fs/file_test.go +++ b/pkg/fs/client/fs/file_test.go @@ -18,10 +18,12 @@ package fs import ( "fmt" + "github.com/agiledragon/gomonkey/v2" "os" "path/filepath" "strings" "sync" + "syscall" "testing" "time" @@ -69,6 +71,39 @@ func newPfsTest() (*FileSystem, error) { return pfs, nil } +func TestFSOpenFail(t *testing.T) { + os.RemoveAll("./mock") + os.RemoveAll("./mock-cache") + defer func() { + os.RemoveAll("./mock") + os.RemoveAll("./mock-cache") + }() + d := cache.Config{ + BlockSize: 2, + MaxReadAhead: 100, + Config: kv.Config{ + Driver: kv.MemType, + }, + } + SetDataCache(d) + client, err := newPfsTest() + assert.Equal(t, err, nil) + + path := "testRead" + flags := os.O_RDWR | os.O_CREATE | os.O_TRUNC + mode := 0666 + _, err = client.Create(path, uint32(flags), uint32(mode)) + assert.Equal(t, err, nil) + + patches := gomonkey.NewPatches() + patches.ApplyMethod(new(kv.KVTxn), "Get", func(_ *kv.KVTxn, key []byte) []byte { + return nil + }) + _, err = client.Open(path) + assert.Equal(t, err, syscall.ENOENT) + patches.Reset() +} + func TestFSClient_readAt_BigOff(t *testing.T) { // 测试off越界的情况 os.RemoveAll("./mock") From 39bc97ff525440035bcbe430b8d87351ba23fe3b Mon Sep 17 00:00:00 2001 From: PeyChou Date: Sun, 7 Apr 2024 10:56:12 +0800 Subject: [PATCH 05/15] =?UTF-8?q?=E5=8D=95=E7=8B=AC=E5=88=9B=E5=BB=BA?= =?UTF-8?q?=E9=A2=84=E7=83=AD=E4=BC=98=E5=8C=96=E5=8D=8F=E7=A8=8B=E6=B1=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/fs/fuse/service/warmup.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/cmd/fs/fuse/service/warmup.go b/cmd/fs/fuse/service/warmup.go index 95040d05..df9adb02 100644 --- a/cmd/fs/fuse/service/warmup.go +++ b/cmd/fs/fuse/service/warmup.go @@ -58,7 +58,7 @@ func findUniqueParentDirs(paths []string) []string { wg.Done() } - pool, _ = ants.NewPool(poolSize) + dirPool, _ := ants.NewPool(poolSize) log.Infof("Start to find unique parent dirs") // 分批提交协程池处理 @@ -72,7 +72,7 @@ func findUniqueParentDirs(paths []string) []string { end = len(paths) } wg.Add(1) - _ = pool.Submit(func() { + _ = dirPool.Submit(func() { processBatch(paths[start:end]) }) } @@ -85,7 +85,7 @@ func findUniqueParentDirs(paths []string) []string { end = len(paths) } wg.Add(1) - _ = pool.Submit(func() { + _ = dirPool.Submit(func() { processBatch(paths[start:end]) }) } From 53d020d07eb3bcc11be8672c7a168133b899bec3 Mon Sep 17 00:00:00 2001 From: PeyChou Date: Tue, 16 Apr 2024 17:22:08 +0800 Subject: [PATCH 06/15] =?UTF-8?q?=E4=BF=AE=E5=A4=8Dcmd=E5=91=BD=E4=BB=A4?= =?UTF-8?q?=E5=8F=82=E6=95=B0=E8=AF=BB=E5=8F=96=E9=94=99=E8=AF=AF=EF=BC=8C?= =?UTF-8?q?=E5=90=88=E5=B9=B6=E6=96=87=E4=BB=B6=E4=BF=9D=E7=95=99=E7=9B=AE?= =?UTF-8?q?=E5=BD=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/fs/fuse/service/warmup.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/cmd/fs/fuse/service/warmup.go b/cmd/fs/fuse/service/warmup.go index df9adb02..218e5c2c 100644 --- a/cmd/fs/fuse/service/warmup.go +++ b/cmd/fs/fuse/service/warmup.go @@ -26,12 +26,14 @@ func findUniqueParentDirs(paths []string) []string { var wg sync.WaitGroup var rwmu sync.RWMutex parentDirMap := make(map[string]map[string]struct{}) + uniqueParentDirs := make([]string, 0) // 协程任务 processBatch := func(pathBatch []string) { for _, p := range pathBatch { // 忽略目录，只处理文件 if p[len(p)-1] == '/' { + uniqueParentDirs = append(uniqueParentDirs, p) continue } // 获取文件的父目录 @@ -94,7 +96,6 @@ func findUniqueParentDirs(paths []string) []string { wg.Wait() // 从 parentDirMap 中提取结果 - uniqueParentDirs := make([]string, 0) for parentPath, dirMap := range parentDirMap { if len(dirMap) >= minxFileCount { uniqueParentDirs = append(uniqueParentDirs, parentPath) @@ -110,12 +111,11 @@ func findUniqueParentDirs(paths []string) []string { func warmup(ctx *cli.Context) error { fname := ctx.String("file") - paths := ctx.Args().Slice() threads := int(ctx.Uint("threads")) warmType := ctx.String("type") recursive := ctx.Bool("recursive") pool, _ = ants.NewPool(threads) - return warmup_(fname, paths, threads, warmType, recursive) + return warmup_(fname, []string{}, threads, warmType, recursive) } func warmup_(fname string, paths []string, threads int, warmType string, recursive bool) error { From 7dbb3ffd319d7739d04e8c2374bdb2eb32bfc25d Mon Sep 17 00:00:00 2001 From: PeyChou Date: Tue, 16 Apr 2024 17:22:08 +0800 Subject: [PATCH 07/15] =?UTF-8?q?=E4=BF=AE=E5=A4=8Dcmd=E5=91=BD=E4=BB=A4?= =?UTF-8?q?=E5=8F=82=E6=95=B0=E8=AF=BB=E5=8F=96=E9=94=99=E8=AF=AF=EF=BC=8C?= =?UTF-8?q?=E5=90=88=E5=B9=B6=E6=96=87=E4=BB=B6=E4=BF=9D=E7=95=99=E7=9B=AE?= =?UTF-8?q?=E5=BD=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/fs/fuse/service/warmup.go | 3 +++ cmd/fs/fuse/service/warmup_test.go | 5 ++++- 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/cmd/fs/fuse/service/warmup.go b/cmd/fs/fuse/service/warmup.go index 218e5c2c..0db902c3 100644 --- a/cmd/fs/fuse/service/warmup.go +++ b/cmd/fs/fuse/service/warmup.go @@ -25,6 +25,7 @@ const minxFileCount = 5 func findUniqueParentDirs(paths []string) []string { var wg sync.WaitGroup var rwmu sync.RWMutex + var mu sync.Mutex parentDirMap := make(map[string]map[string]struct{}) uniqueParentDirs := make([]string, 0) @@ -33,7 +34,9 @@ func findUniqueParentDirs(paths []string) []string { for _, p := range pathBatch { // 忽略目录，只处理文件 if p[len(p)-1] == '/' { + mu.Lock() uniqueParentDirs = append(uniqueParentDirs, p) + mu.Unlock() continue } // 获取文件的父目录 diff --git a/cmd/fs/fuse/service/warmup_test.go b/cmd/fs/fuse/service/warmup_test.go index b8dc628e..88fee1f8 100644 --- a/cmd/fs/fuse/service/warmup_test.go +++ b/cmd/fs/fuse/service/warmup_test.go @@ -116,6 +116,7 @@ func TestFindUniqueParentDirs(t *testing.T) { "/home/user/docs/assignment.docx", "/var/log/sys.log", "/usr/bin/someexecutable", + "/usr/bin/", }, }, { @@ -149,7 +150,9 @@ func TestFindUniqueParentDirs(t *testing.T) { paths: []string{ "/", }, - expected: []string{}, + expected: []string{ + "/", + }, }, } From 82fd4df25c183938d4c1240014cf5df3fcc3b6d3 Mon Sep 17 00:00:00 2001 From: PeyChou Date: Wed, 24 Apr 2024 14:59:57 +0800 Subject: [PATCH 08/15] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E5=90=8E=E5=8F=B0?= =?UTF-8?q?=E8=BF=90=E8=A1=8C=E6=A8=A1=E5=BC=8F=EF=BC=8C=E6=9B=B4=E6=96=B0?= =?UTF-8?q?go.mod?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/fs/fuse/flag/flags.go | 6 + cmd/fs/fuse/service/mount.go | 23 ++- go.mod | 4 +- go.sum | 5 + pkg/fs/client/utils/daemon.go | 267 ++++++++++++++++++++++++++++++++++ 5 files changed, 302 insertions(+), 3 deletions(-) create mode 100644 pkg/fs/client/utils/daemon.go diff --git a/cmd/fs/fuse/flag/flags.go b/cmd/fs/fuse/flag/flags.go index f4c576db..2868667f 100644 --- a/cmd/fs/fuse/flag/flags.go +++ b/cmd/fs/fuse/flag/flags.go @@ -181,6 +181,12 @@ func MountFlags(fuseConf *meta.FuseConfig) []cli.Flag { Value: false, Usage: "Assume all directory objects (\"dir/\") exist (default: false)", }, + &cli.BoolFlag{ + Name: "background", + Aliases: []string{"d"}, + Value: false, + Usage: "run in backGround", + }, } } diff --git a/cmd/fs/fuse/service/mount.go b/cmd/fs/fuse/service/mount.go index ad9669fb..c584cf11 100644 --- a/cmd/fs/fuse/service/mount.go +++ b/cmd/fs/fuse/service/mount.go @@ -19,6 +19,7 @@ package service import ( "encoding/json" "fmt" + utils2 "github.com/PaddlePaddle/PaddleFlow/pkg/fs/client/utils" "io/ioutil" "math" "net" @@ -234,6 +235,10 @@ func mount(c *cli.Context) error { return err } + if c.Bool("background") { + daemonRun() + } + if !c.Bool("local") { stopChan := make(chan struct{}) defer close(stopChan) @@ -248,7 +253,7 @@ func mount(c *cli.Context) error { } } - log.Debugf("start mount service") + log.Infof("start mount service on %v", c.String("mount-point")) server, err := fuse.Server(c.String("mount-point"), *opts) if err != nil { log.Fatalf("mount fail: %v", err) @@ -460,3 +465,19 @@ func signalHandle(mp string) { } }() } + +func daemonRun() { + err := makeDaemon() + if err != nil { + log.Fatalf("Failed to make daemon: %s", err) + } + if runtime.GOOS == "linux" { + log.SetOutput(os.Stderr) + } +} + +func makeDaemon() error { + var attrs utils2.DaemonAttr + _, _, err := utils2.MakeDaemon(&attrs) + return err +} diff --git a/go.mod b/go.mod index 7f8d33b1..238db105 100644 --- a/go.mod +++ b/go.mod @@ -20,7 +20,7 @@ require ( github.com/go-chi/chi v1.5.4 github.com/google/uuid v1.3.0 github.com/gorilla/websocket v1.4.2 - github.com/hanwen/go-fuse/v2 v2.2.0 + github.com/hanwen/go-fuse/v2 v2.5.1 github.com/jcmturner/gokrb5/v8 v8.4.4 github.com/jinzhu/copier v0.3.2 github.com/johannesboyne/gofakes3 v0.0.0-20220627085814-c3ac35da23b2 @@ -147,7 +147,7 @@ require ( go.opencensus.io v0.22.5 // indirect golang.org/x/mod v0.8.0 // indirect golang.org/x/oauth2 v0.7.0 // indirect - golang.org/x/sys v0.15.0 // indirect + golang.org/x/sys v0.19.0 // indirect golang.org/x/term v0.15.0 // indirect golang.org/x/text v0.14.0 // indirect golang.org/x/time v0.0.0-20210723032227-1f47c861a9ac // indirect diff --git a/go.sum b/go.sum index 18cce105..1749b297 100644 --- a/go.sum +++ b/go.sum @@ -468,6 +468,8 @@ github.com/grpc-ecosystem/grpc-gateway v1.14.6/go.mod h1:zdiPV4Yse/1gnckTHtghG4G github.com/grpc-ecosystem/grpc-gateway v1.16.0/go.mod h1:BDjrQk3hbvj6Nolgz8mAMFbcEtjT1g+wF4CSlocrBnw= github.com/hanwen/go-fuse/v2 v2.2.0 h1:jo5QZYmBLNcl9ovypWaQ5yXMSSV+Ch68xoC3rtZvvBM= github.com/hanwen/go-fuse/v2 v2.2.0/go.mod h1:B1nGE/6RBFyBRC1RRnf23UpwCdyJ31eukw34oAKukAc= +github.com/hanwen/go-fuse/v2 v2.5.1 h1:OQBE8zVemSocRxA4OaFJbjJ5hlpCmIWbGr7r0M4uoQQ= +github.com/hanwen/go-fuse/v2 v2.5.1/go.mod h1:xKwi1cF7nXAOBCXujD5ie0ZKsxc8GGSA1rlMJc+8IJs= github.com/hashicorp/consul/api v1.3.0/go.mod h1:MmDNSzIMUjNpY/mQ398R4bk2FnqQLoPndWW5VkKPlCE= github.com/hashicorp/consul/sdk v0.3.0/go.mod h1:VKf9jXwCTEY1QZP2MOLRhb5i/I/ssyNV1vwHyQBF0x8= github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= @@ -716,6 +718,7 @@ github.com/mitchellh/mapstructure v0.0.0-20160808181253-ca63d7c062ee/go.mod h1:F github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y= github.com/mitchellh/mapstructure v1.4.1 h1:CpVNEelQCZBooIPDn+AR3NpivK/TIKU8bDxdASFVQag= github.com/mitchellh/mapstructure v1.4.1/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo= +github.com/moby/sys/mountinfo v0.6.2/go.mod h1:IJb6JQeOklcdMU9F5xQ8ZALD+CUr5VlGpwtX+VE0rpI= github.com/moby/term v0.0.0-20200312100748-672ec06f55cd h1:aY7OQNf2XqY/JQ6qREWamhI/81os/agb2BAGpcx5yWI= github.com/moby/term v0.0.0-20200312100748-672ec06f55cd/go.mod h1:DdlQx2hp0Ss5/fLikoLlEeIYiATotOjgB//nb973jeo= github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= @@ -1242,6 +1245,8 @@ golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.15.0 h1:h48lPFYpsTvQJZF4EKyI4aLHaev3CxivZmv7yZig9pc= golang.org/x/sys v0.15.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.19.0 h1:q5f1RH2jigJ1MoAWp2KTp3gm5zAGFUTarQZ5U386+4o= +golang.org/x/sys v0.19.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= diff --git a/pkg/fs/client/utils/daemon.go b/pkg/fs/client/utils/daemon.go new file mode 100644 index 00000000..662feeb7 --- /dev/null +++ b/pkg/fs/client/utils/daemon.go @@ -0,0 +1,267 @@ +package utils + +import ( + "bytes" + "crypto/sha1" + "encoding/hex" + "fmt" + "io" + "os" + "strconv" + "strings" + "syscall" + "time" +) + +const ( + stageVar = "__DAEMON_STAGE" + fdVarPrefix = "__DAEMON_FD_" +) + +// DaemonAttr describes the options that apply to daemonization +type DaemonAttr struct { + ProgramName string // child's os.Args[0]; copied from parent if empty + CaptureOutput bool // whether to capture stdout/stderr + Files []**os.File // files to keep open in the daemon + Stdout *os.File // redirect stdout/stderr to it + OnExit func(stage int) error +} + +func MakeDaemon(attrs *DaemonAttr) (io.Reader, io.Reader, error) { + stage, advanceStage, resetEnv := getStage() + + fatal := func(err error) (io.Reader, io.Reader, error) { + if stage > 0 { + os.Exit(1) + } + resetEnv() + return nil, nil, err + } + + fileCount := 3 + len(attrs.Files) + files := make([]*os.File, fileCount, fileCount+2) + + if stage == 0 { + nullDev, err := os.OpenFile("/dev/null", 0, 0) + if err != nil { + return fatal(err) + } + files[0] = nullDev + if attrs.Stdout != nil { + files[1], files[2] = attrs.Stdout, attrs.Stdout + } else { + files[1], files[2] = nullDev, nullDev + } + + fd := 3 + for _, fPtr := range attrs.Files { + files[fd] = *fPtr + saveFileName(fd, (*fPtr).Name()) + fd++ + } + } else { + files[0], files[1], files[2] = os.Stdin, os.Stdout, os.Stderr + + fd := 3 + for _, fPtr := range attrs.Files { + *fPtr = os.NewFile(uintptr(fd), getFileName(fd)) + syscall.CloseOnExec(fd) + files[fd] = *fPtr + fd++ + } + } + + if stage < 2 { + procName, err := os.Executable() + if err != nil { + return fatal(fmt.Errorf("can't determine full path to executable: %s", err)) + } + + if len(procName) == 0 { + return fatal(fmt.Errorf("can't determine full path to executable")) + } + + if stage == 1 && attrs.CaptureOutput { + files = files[:fileCount+2] + + // stdout: write at fd:1, read at fd:fileCount + if files[fileCount], files[1], err = os.Pipe(); err != nil { + return fatal(err) + } + // stderr: write at fd:2, read at fd:fileCount+1 + if files[fileCount+1], files[2], err = os.Pipe(); err != nil { + return fatal(err) + } + } + + if err := advanceStage(); err != nil { + return fatal(err) + } + dir, _ := os.Getwd() + osAttrs := os.ProcAttr{Dir: dir, Env: os.Environ(), Files: files} + + if stage == 0 { + sysattrs := syscall.SysProcAttr{Setsid: true} + osAttrs.Sys = &sysattrs + } + + progName := attrs.ProgramName + if len(progName) == 0 { + progName = os.Args[0] + } + args := append([]string{progName}, os.Args[1:]...) + proc, err := os.StartProcess(procName, args, &osAttrs) + if err != nil { + return fatal(fmt.Errorf("can't create process %s: %s", procName, err)) + } + proc.Release() + if attrs.OnExit != nil { + attrs.OnExit(stage) + } + os.Exit(0) + } + + //os.Chdir("/") + syscall.Umask(0) + resetEnv() + + for fd := 3; fd < fileCount; fd++ { + resetFileName(fd) + } + currStage = DaemonStage(stage) + + var stdout, stderr *os.File + if attrs.CaptureOutput { + stdout = os.NewFile(uintptr(fileCount), "stdout") + stderr = os.NewFile(uintptr(fileCount+1), "stderr") + } + return stdout, stderr, nil +} + +func saveFileName(fd int, name string) { + // We encode in hex to avoid issues with filename encoding, and to be able + // to separate it from the original variable value (if set) that we want to + // keep. Otherwise, all non-zero characters are valid in the name, and we + // can't insert a zero in the var as a separator. + fdVar := fdVarPrefix + fmt.Sprint(fd) + value := fmt.Sprintf("%s:%s", + hex.EncodeToString([]byte(name)), os.Getenv(fdVar)) + + if err := os.Setenv(fdVar, value); err != nil { + fmt.Fprintf(os.Stderr, "can't set %s: %s\n", fdVar, err) + os.Exit(1) + } +} + +func getFileName(fd int) string { + fdVar := fdVarPrefix + fmt.Sprint(fd) + value := os.Getenv(fdVar) + sep := bytes.IndexByte([]byte(value), ':') + + if sep < 0 { + fmt.Fprintf(os.Stderr, "bad fd var %s\n", fdVar) + os.Exit(1) + } + name, err := hex.DecodeString(value[:sep]) + if err != nil { + fmt.Fprintf(os.Stderr, "error decoding %s\n", fdVar) + os.Exit(1) + } + return string(name) +} + +func resetFileName(fd int) { + fdVar := fdVarPrefix + fmt.Sprint(fd) + value := os.Getenv(fdVar) + sep := bytes.IndexByte([]byte(value), ':') + + if sep < 0 { + fmt.Fprintf(os.Stderr, "bad fd var %s\n", fdVar) + os.Exit(1) + } + if err := os.Setenv(fdVar, value[sep+1:]); err != nil { + fmt.Fprintf(os.Stderr, "can't reset %s\n", fdVar) + os.Exit(1) + } +} + +// DaemonStage tells in what stage in the process we are. See Stage(). +type DaemonStage int + +// Stages in the daemonizing process. +const ( + StageParent = DaemonStage(iota) // Original process + StageChild // MakeDaemon() called once: first child + StageDaemon // MakeDaemon() run twice: final daemon + + stageUnknown = DaemonStage(-1) +) + +var currStage = stageUnknown + +func Stage() DaemonStage { + if currStage == stageUnknown { + s, _, _ := getStage() + currStage = DaemonStage(s) + } + return currStage +} + +func (s DaemonStage) String() string { + switch s { + case StageParent: + return "parent" + case StageChild: + return "first child" + case StageDaemon: + return "daemon" + default: + return "unknown" + } +} + +func getStage() (stage int, advanceStage func() error, resetEnv func() error) { + var origValue string + stage = 0 + + daemonStage := os.Getenv(stageVar) + stageTag := strings.SplitN(daemonStage, ":", 2) + stageInfo := strings.SplitN(stageTag[0], "/", 3) + + if len(stageInfo) == 3 { + stageStr, tm, check := stageInfo[0], stageInfo[1], stageInfo[2] + + hash := sha1.New() + hash.Write([]byte(stageStr + "/" + tm + "/")) + + if check != hex.EncodeToString(hash.Sum([]byte{})) { + // This whole chunk is original data + origValue = daemonStage + } else { + stage, _ = strconv.Atoi(stageStr) + + if len(stageTag) == 2 { + origValue = stageTag[1] + } + } + } else { + origValue = daemonStage + } + + advanceStage = func() error { + base := fmt.Sprintf("%d/%09d/", stage+1, time.Now().Nanosecond()) + hash := sha1.New() + hash.Write([]byte(base)) + tag := base + hex.EncodeToString(hash.Sum([]byte{})) + + if err := os.Setenv(stageVar, tag+":"+origValue); err != nil { + return fmt.Errorf("can't set %s: %s", stageVar, err) + } + return nil + } + resetEnv = func() error { + return os.Setenv(stageVar, origValue) + } + + return stage, advanceStage, resetEnv +} From 54fe5e723e2b875d4f4e569874ce25d88a954e51 Mon Sep 17 00:00:00 2001 From: PeyChou Date: Wed, 24 Apr 2024 14:59:57 +0800 Subject: [PATCH 09/15] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E5=90=8E=E5=8F=B0?= =?UTF-8?q?=E8=BF=90=E8=A1=8C=E6=A8=A1=E5=BC=8F=EF=BC=8C=E6=9B=B4=E6=96=B0?= =?UTF-8?q?go.mod?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/fs/fuse/service/mount_test.go | 72 +++++++++++++++++++++++++++++++ 1 file changed, 72 insertions(+) create mode 100644 cmd/fs/fuse/service/mount_test.go diff --git a/cmd/fs/fuse/service/mount_test.go b/cmd/fs/fuse/service/mount_test.go new file mode 100644 index 00000000..321fe22e --- /dev/null +++ b/cmd/fs/fuse/service/mount_test.go @@ -0,0 +1,72 @@ +package service + +import ( + "fmt" + + "github.com/PaddlePaddle/PaddleFlow/pkg/fs/utils" + "github.com/PaddlePaddle/PaddleFlow/pkg/version" + "github.com/urfave/cli/v2" + "os" + "testing" + "time" +) + +const testMountPoint = "./mock" + +func MockMain(args []string) error { + cli.VersionFlag = &cli.BoolFlag{ + Name: "version", Aliases: []string{"V"}, + Usage: "print only the version", + } + app := &cli.App{ + Name: "pfs-fuse", + Usage: "A POSIX file system built on kv DB and object storage.", + Version: version.InfoStr(), + Copyright: "Apache License 2.0", + HideHelpCommand: true, + EnableBashCompletion: true, + Commands: []*cli.Command{ + CmdMount(), + CmdUmount(), + CmdStats(), + CmdBench(), + CmdWarmup(), + }, + } + return app.Run(args) +} + +func mountTemp(t *testing.T) { + os.MkdirAll(testMountPoint, 0755) + mountArgs := []string{"", "mount", "-d", "true"} + go func() { + if err := MockMain(mountArgs); err != nil { + t.Errorf("mount failed: %s", err) + } + }() + time.Sleep(3 * time.Second) + inode, err := utils.GetFileInode(testMountPoint) + if err != nil { + t.Fatalf("get file inode failed: %s", err) + } + if inode != 1 { + t.Fatalf("mount failed: inode of %s is not 1", testMountPoint) + } else { + t.Logf("mount %s success", testMountPoint) + } +} + +func umountTemp(t *testing.T) { + if err := MockMain([]string{"", "umount", testMountPoint}); err != nil { + t.Fatalf("umount failed: %s", err) + } +} + +func TestMountBackground(t *testing.T) { + mountTemp(t) + defer umountTemp(t) + + if err := os.WriteFile(fmt.Sprintf("%s/f1.txt", testMountPoint), []byte("test"), 0644); err != nil { + t.Fatalf("write file failed: %s", err) + } +} From 3470475033693fe6666aee8dfe28935502e3ea45 Mon Sep 17 00:00:00 2001 From: PeyChou Date: Wed, 24 Apr 2024 14:59:57 +0800 Subject: [PATCH 10/15] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E5=90=8E=E5=8F=B0?= =?UTF-8?q?=E8=BF=90=E8=A1=8C=E6=A8=A1=E5=BC=8F=EF=BC=8C=E6=9B=B4=E6=96=B0?= =?UTF-8?q?go.mod?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/fs/fuse/service/mount_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cmd/fs/fuse/service/mount_test.go b/cmd/fs/fuse/service/mount_test.go index 321fe22e..2572a17c 100644 --- a/cmd/fs/fuse/service/mount_test.go +++ b/cmd/fs/fuse/service/mount_test.go @@ -38,7 +38,7 @@ func MockMain(args []string) error { func mountTemp(t *testing.T) { os.MkdirAll(testMountPoint, 0755) - mountArgs := []string{"", "mount", "-d", "true"} + mountArgs := []string{"", "mount", "-d", "true", "--local", "true"} go func() { if err := MockMain(mountArgs); err != nil { t.Errorf("mount failed: %s", err) From 7a8a6eb3b0c5946714b13297c8b7d6fccd7fc536 Mon Sep 17 00:00:00 2001 From: PeyChou Date: Wed, 24 Apr 2024 14:59:57 +0800 Subject: [PATCH 11/15] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E5=90=8E=E5=8F=B0?= =?UTF-8?q?=E8=BF=90=E8=A1=8C=E6=A8=A1=E5=BC=8F=EF=BC=8C=E6=9B=B4=E6=96=B0?= =?UTF-8?q?go.mod?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/fs/fuse/service/mount_test.go | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/cmd/fs/fuse/service/mount_test.go b/cmd/fs/fuse/service/mount_test.go index 2572a17c..e309bc5b 100644 --- a/cmd/fs/fuse/service/mount_test.go +++ b/cmd/fs/fuse/service/mount_test.go @@ -1,8 +1,6 @@ package service import ( - "fmt" - "github.com/PaddlePaddle/PaddleFlow/pkg/fs/utils" "github.com/PaddlePaddle/PaddleFlow/pkg/version" "github.com/urfave/cli/v2" @@ -38,7 +36,7 @@ func MockMain(args []string) error { func mountTemp(t *testing.T) { os.MkdirAll(testMountPoint, 0755) - mountArgs := []string{"", "mount", "-d", "true", "--local", "true"} + mountArgs := []string{"", "mount", "--local", "true", "-d", "true"} go func() { if err := MockMain(mountArgs); err != nil { t.Errorf("mount failed: %s", err) @@ -65,8 +63,4 @@ func umountTemp(t *testing.T) { func TestMountBackground(t *testing.T) { mountTemp(t) defer umountTemp(t) - - if err := os.WriteFile(fmt.Sprintf("%s/f1.txt", testMountPoint), []byte("test"), 0644); err != nil { - t.Fatalf("write file failed: %s", err) - } } From 09706a16e0d54093dffb1c1de28f19fa1a53e302 Mon Sep 17 00:00:00 2001 From: PeyChou Date: Wed, 24 Apr 2024 14:59:57 +0800 Subject: [PATCH 12/15] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E5=90=8E=E5=8F=B0?= =?UTF-8?q?=E8=BF=90=E8=A1=8C=E6=A8=A1=E5=BC=8F=EF=BC=8C=E6=9B=B4=E6=96=B0?= =?UTF-8?q?go.mod?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/fs/fuse/service/mount.go | 4 ++++ cmd/fs/fuse/service/mount_test.go | 7 +++---- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/cmd/fs/fuse/service/mount.go b/cmd/fs/fuse/service/mount.go index c584cf11..a48bf087 100644 --- a/cmd/fs/fuse/service/mount.go +++ b/cmd/fs/fuse/service/mount.go @@ -284,6 +284,7 @@ func InitVFS(c *cli.Context, registry *prometheus.Registry) error { server := c.String("server") if c.Bool("local") == true { localRoot := c.String("local-root") + localRoot = "./mock" if localRoot == "" || localRoot == "/" { log.Errorf("invalid localRoot: [%s]", localRoot) return fmt.Errorf("invalid localRoot: [%s]", localRoot) @@ -425,6 +426,9 @@ func InitVFS(c *cli.Context, registry *prometheus.Registry) error { CachePath: c.String("data-cache-path"), }, } + if fsMeta.Properties == nil { + fsMeta.Properties = make(map[string]string) + } if c.Bool("no-implicit-dir") { fsMeta.Properties[common.ImplicitDir] = "false" } else { diff --git a/cmd/fs/fuse/service/mount_test.go b/cmd/fs/fuse/service/mount_test.go index e309bc5b..ecb9e2bf 100644 --- a/cmd/fs/fuse/service/mount_test.go +++ b/cmd/fs/fuse/service/mount_test.go @@ -26,9 +26,6 @@ func MockMain(args []string) error { Commands: []*cli.Command{ CmdMount(), CmdUmount(), - CmdStats(), - CmdBench(), - CmdWarmup(), }, } return app.Run(args) @@ -36,13 +33,15 @@ func MockMain(args []string) error { func mountTemp(t *testing.T) { os.MkdirAll(testMountPoint, 0755) - mountArgs := []string{"", "mount", "--local", "true", "-d", "true"} + mountArgs := []string{"", "mount", "-mp", testMountPoint, "--local", "true", "--local-root", testMountPoint, "-d", "true"} + go func() { if err := MockMain(mountArgs); err != nil { t.Errorf("mount failed: %s", err) } }() time.Sleep(3 * time.Second) + inode, err := utils.GetFileInode(testMountPoint) if err != nil { t.Fatalf("get file inode failed: %s", err) From 0fde44543df7a8ee3594e6dde859e57af2cd456d Mon Sep 17 00:00:00 2001 From: PeyChou Date: Wed, 24 Apr 2024 14:59:57 +0800 Subject: [PATCH 13/15] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E5=90=8E=E5=8F=B0?= =?UTF-8?q?=E8=BF=90=E8=A1=8C=E6=A8=A1=E5=BC=8F=EF=BC=8C=E6=9B=B4=E6=96=B0?= =?UTF-8?q?go.mod?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/fs/fuse/service/mount.go | 4 +++- pkg/fs/client/utils/daemon.go | 20 ++++++++++++++++---- 2 files changed, 19 insertions(+), 5 deletions(-) diff --git a/cmd/fs/fuse/service/mount.go b/cmd/fs/fuse/service/mount.go index a48bf087..5832b292 100644 --- a/cmd/fs/fuse/service/mount.go +++ b/cmd/fs/fuse/service/mount.go @@ -284,7 +284,9 @@ func InitVFS(c *cli.Context, registry *prometheus.Registry) error { server := c.String("server") if c.Bool("local") == true { localRoot := c.String("local-root") - localRoot = "./mock" + if localRoot == "" { + localRoot = "./mock" + } if localRoot == "" || localRoot == "/" { log.Errorf("invalid localRoot: [%s]", localRoot) return fmt.Errorf("invalid localRoot: [%s]", localRoot) diff --git a/pkg/fs/client/utils/daemon.go b/pkg/fs/client/utils/daemon.go index 662feeb7..190674be 100644 --- a/pkg/fs/client/utils/daemon.go +++ b/pkg/fs/client/utils/daemon.go @@ -34,7 +34,10 @@ func MakeDaemon(attrs *DaemonAttr) (io.Reader, io.Reader, error) { if stage > 0 { os.Exit(1) } - resetEnv() + err = resetEnv() + if err != nil { + return nil, nil, err + } return nil, nil, err } @@ -114,16 +117,25 @@ func MakeDaemon(attrs *DaemonAttr) (io.Reader, io.Reader, error) { if err != nil { return fatal(fmt.Errorf("can't create process %s: %s", procName, err)) } - proc.Release() + err = proc.Release() + if err != nil { + return nil, nil, err + } if attrs.OnExit != nil { - attrs.OnExit(stage) + err := attrs.OnExit(stage) + if err != nil { + return nil, nil, err + } } os.Exit(0) } //os.Chdir("/") syscall.Umask(0) - resetEnv() + err := resetEnv() + if err != nil { + return nil, nil, err + } for fd := 3; fd < fileCount; fd++ { resetFileName(fd) From cf25923bd61ae599da6d9860b3c48da117742143 Mon Sep 17 00:00:00 2001 From: PeyChou Date: Wed, 24 Apr 2024 14:59:57 +0800 Subject: [PATCH 14/15] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E5=90=8E=E5=8F=B0?= =?UTF-8?q?=E8=BF=90=E8=A1=8C=E6=A8=A1=E5=BC=8F=EF=BC=8C=E6=9B=B4=E6=96=B0?= =?UTF-8?q?go.mod?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/fs/fuse/service/mount.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cmd/fs/fuse/service/mount.go b/cmd/fs/fuse/service/mount.go index 5832b292..bbc40948 100644 --- a/cmd/fs/fuse/service/mount.go +++ b/cmd/fs/fuse/service/mount.go @@ -19,7 +19,6 @@ package service import ( "encoding/json" "fmt" - utils2 "github.com/PaddlePaddle/PaddleFlow/pkg/fs/client/utils" "io/ioutil" "math" "net" @@ -32,6 +31,7 @@ import ( "syscall" "time" + utils2 "github.com/PaddlePaddle/PaddleFlow/pkg/fs/client/utils" libfuse "github.com/hanwen/go-fuse/v2/fuse" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promhttp" From ff00f8bdf21d9c30e24be5ce61a55783462a56b4 Mon Sep 17 00:00:00 2001 From: PeyChou Date: Wed, 24 Apr 2024 14:59:57 +0800 Subject: [PATCH 15/15] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E5=90=8E=E5=8F=B0?= =?UTF-8?q?=E8=BF=90=E8=A1=8C=E6=A8=A1=E5=BC=8F=EF=BC=8C=E6=9B=B4=E6=96=B0?= =?UTF-8?q?go.mod?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pkg/fs/client/utils/daemon.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pkg/fs/client/utils/daemon.go b/pkg/fs/client/utils/daemon.go index 190674be..7ca770c1 100644 --- a/pkg/fs/client/utils/daemon.go +++ b/pkg/fs/client/utils/daemon.go @@ -34,9 +34,9 @@ func MakeDaemon(attrs *DaemonAttr) (io.Reader, io.Reader, error) { if stage > 0 { os.Exit(1) } - err = resetEnv() + resetErr := resetEnv() if err != nil { - return nil, nil, err + return nil, nil, resetErr } return nil, nil, err }