diff --git a/.github/workflows/fuse-dlm-integration.yml b/.github/workflows/fuse-dlm-integration.yml new file mode 100644 index 000000000..868fbe581 --- /dev/null +++ b/.github/workflows/fuse-dlm-integration.yml @@ -0,0 +1,63 @@ +name: "FUSE DLM Integration Tests" + +on: + pull_request: + paths: + - 'weed/command/mount*.go' + - 'weed/mount/**' + - 'weed/cluster/**' + - 'test/fuse_dlm/**' + - '.github/workflows/fuse-dlm-integration.yml' + push: + branches: [master] + paths: + - 'weed/command/mount*.go' + - 'weed/mount/**' + - 'weed/cluster/**' + - 'test/fuse_dlm/**' + +concurrency: + group: ${{ github.head_ref || github.ref }}/fuse-dlm-integration + cancel-in-progress: true + +permissions: + contents: read + +jobs: + fuse-dlm-integration: + name: FUSE DLM Integration Tests + runs-on: ubuntu-22.04 + timeout-minutes: 30 + + steps: + - name: Check out code + uses: actions/checkout@v6 + + - name: Set up Go + uses: actions/setup-go@v6 + with: + go-version-file: 'go.mod' + + - name: Install FUSE dependencies + run: | + sudo apt-get update + sudo apt-get install -y libfuse3-dev + echo 'user_allow_other' | sudo tee -a /etc/fuse.conf + sudo chmod 644 /etc/fuse.conf + + - name: Build SeaweedFS + run: go build -o weed/weed -buildvcs=false ./weed + + - name: Run DLM integration tests + timeout-minutes: 25 + env: + WEED_BINARY: ${{ github.workspace }}/weed/weed + run: go test -v -count=1 -timeout=20m ./test/fuse_dlm/... + + - name: Upload logs on failure + if: failure() + uses: actions/upload-artifact@v7 + with: + name: fuse-dlm-test-logs + path: /tmp/seaweedfs-fuse-dlm-logs/ + retention-days: 3 diff --git a/test/fuse_dlm/dlm_concurrent_write_test.go b/test/fuse_dlm/dlm_concurrent_write_test.go new file mode 100644 index 000000000..5f844e622 --- /dev/null +++ b/test/fuse_dlm/dlm_concurrent_write_test.go @@ -0,0 +1,277 @@ +package fuse_dlm + +import ( + "fmt" + "os" + "path/filepath" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestDLMConcurrentWritersSameFile verifies that two mounts writing to the same +// file concurrently produce valid (non-corrupted) data. With DLM enabled, the +// writes are serialized — one blocks until the other completes. +// +// Note: cross-mount read consistency depends on FUSE kernel cache invalidation +// and filer metadata subscription, which are asynchronous. This test verifies +// write integrity, not instant read convergence. +func TestDLMConcurrentWritersSameFile(t *testing.T) { + if testing.Short() { + t.Skip("skipping DLM integration test in short mode") + } + + cluster := startDLMTestCluster(t) + t.Cleanup(cluster.Stop) + + const iterations = 5 + for iter := 0; iter < iterations; iter++ { + fileName := fmt.Sprintf("concurrent_write_%d.txt", iter) + payloadA := []byte(fmt.Sprintf("mount0-iteration-%d-payload-AAAA", iter)) + payloadB := []byte(fmt.Sprintf("mount1-iteration-%d-payload-BBBB", iter)) + + var wg sync.WaitGroup + wg.Add(2) + + go func() { + defer wg.Done() + err := os.WriteFile(filepath.Join(cluster.mountPoints[0], fileName), payloadA, 0644) + assert.NoError(t, err, "mount0 write iteration %d", iter) + }() + + go func() { + defer wg.Done() + err := os.WriteFile(filepath.Join(cluster.mountPoints[1], fileName), payloadB, 0644) + assert.NoError(t, err, "mount1 write iteration %d", iter) + }() + + wg.Wait() + + // Verify file is readable and contains one of the expected payloads + // (read from mount0 — its own view is authoritative for write success). + content, err := os.ReadFile(filepath.Join(cluster.mountPoints[0], fileName)) + require.NoError(t, err, "read from mount0 iteration %d", iter) + validPayload := string(content) == string(payloadA) || string(content) == string(payloadB) + assert.True(t, validPayload, + "iteration %d: content must be one of the expected payloads, got: %q", iter, content) + } +} + +// TestDLMRepeatedOpenWriteClose verifies that repeated open/write/close cycles +// from both mounts all succeed without errors. +func TestDLMRepeatedOpenWriteClose(t *testing.T) { + if testing.Short() { + t.Skip("skipping DLM integration test in short mode") + } + + cluster := startDLMTestCluster(t) + t.Cleanup(cluster.Stop) + + const cycles = 20 + fileName := "repeated_write.txt" + + var wg sync.WaitGroup + wg.Add(2) + + go func() { + defer wg.Done() + for i := 0; i < cycles; i++ { + data := []byte(fmt.Sprintf("mount0-cycle-%d", i)) + err := os.WriteFile(filepath.Join(cluster.mountPoints[0], fileName), data, 0644) + assert.NoError(t, err, "mount0 cycle %d", i) + } + }() + + go func() { + defer wg.Done() + for i := 0; i < cycles; i++ { + data := []byte(fmt.Sprintf("mount1-cycle-%d", i)) + err := os.WriteFile(filepath.Join(cluster.mountPoints[1], fileName), data, 0644) + assert.NoError(t, err, "mount1 cycle %d", i) + } + }() + + wg.Wait() + + // File must be readable from at least one mount + content, err := os.ReadFile(filepath.Join(cluster.mountPoints[0], fileName)) + require.NoError(t, err) + assert.NotEmpty(t, content, "file must not be empty") +} + +// TestDLMWriteBlocksSecondWriter verifies the core DLM guarantee: while one +// mount has a file open for writing, another mount's write-open blocks until +// the first mount closes the file. +func TestDLMWriteBlocksSecondWriter(t *testing.T) { + if testing.Short() { + t.Skip("skipping DLM integration test in short mode") + } + + cluster := startDLMTestCluster(t) + t.Cleanup(cluster.Stop) + + fileName := "blocking_test.txt" + path0 := filepath.Join(cluster.mountPoints[0], fileName) + path1 := filepath.Join(cluster.mountPoints[1], fileName) + + // Mount 0 opens the file for writing and holds it open + f, err := os.OpenFile(path0, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0644) + require.NoError(t, err, "mount0 open") + _, err = f.Write([]byte("mount0-holds-lock")) + require.NoError(t, err, "mount0 write") + + // Mount 1 tries to write — should block (we use a goroutine with atomic flag) + var mount1Completed atomic.Bool + mount1Done := make(chan error, 1) + go func() { + err := os.WriteFile(path1, []byte("mount1-waited"), 0644) + mount1Completed.Store(true) + mount1Done <- err + }() + + // Give mount 1 a moment — it should NOT complete while mount 0 holds the file open + time.Sleep(3 * time.Second) + require.False(t, mount1Completed.Load(), + "mount1 write must not complete while mount0 holds the file open") + t.Log("mount1 write is blocked as expected while mount0 holds the file") + + // Mount 0 closes the file — this releases the DLM lock + require.NoError(t, f.Close(), "mount0 close") + + // Mount 1 should now complete + select { + case err := <-mount1Done: + assert.NoError(t, err, "mount1 write after mount0 close") + case <-time.After(30 * time.Second): + t.Fatal("mount1 write did not complete within 30s after mount0 closed") + } +} + +// TestDLMRenameWhileWriteOpen verifies that a rename is coordinated with DLM: +// while mount0 has a file open for writing (re-opened after creation), +// mount1 cannot rename it until mount0 closes the file. +func TestDLMRenameWhileWriteOpen(t *testing.T) { + if testing.Short() { + t.Skip("skipping DLM integration test in short mode") + } + + cluster := startDLMTestCluster(t) + t.Cleanup(cluster.Stop) + + origName := "rename_source.txt" + newName := "rename_dest.txt" + + // Create and close the file first so it's flushed to the filer and + // visible on both mounts. + require.NoError(t, os.WriteFile( + filepath.Join(cluster.mountPoints[0], origName), + []byte("initial-content"), 0644)) + time.Sleep(2 * time.Second) // metadata propagation + + // Verify mount1 can see the file + _, err := os.Stat(filepath.Join(cluster.mountPoints[1], origName)) + require.NoError(t, err, "mount1 should see the file") + + // Mount 0 re-opens the file for writing and holds it open + f, err := os.OpenFile( + filepath.Join(cluster.mountPoints[0], origName), + os.O_WRONLY|os.O_TRUNC, 0644) + require.NoError(t, err, "mount0 reopen") + _, err = f.Write([]byte("data-while-holding-lock")) + require.NoError(t, err, "mount0 write") + + // Mount 1 tries to rename — should block because mount0 holds the + // DLM lock on the old path + var renameCompleted atomic.Bool + renameDone := make(chan error, 1) + go func() { + err := os.Rename( + filepath.Join(cluster.mountPoints[1], origName), + filepath.Join(cluster.mountPoints[1], newName)) + renameCompleted.Store(true) + renameDone <- err + }() + + // Rename must NOT complete while mount0 holds the file open + time.Sleep(3 * time.Second) + require.False(t, renameCompleted.Load(), + "rename must not complete while mount0 holds the file open") + t.Log("rename is blocked as expected while mount0 holds the file") + + // Mount 0 closes → releases DLM lock → rename should proceed + require.NoError(t, f.Close(), "mount0 close") + + select { + case err := <-renameDone: + assert.NoError(t, err, "rename after mount0 close") + case <-time.After(30 * time.Second): + t.Fatal("rename did not complete within 30s after mount0 closed") + } +} + +// Note: Same-mount rename while a file is open for writing is not tested here +// because macOS FUSE serializes operations on the same inode, causing a +// kernel-level deadlock between the Rename handler's internal flush and the +// pending Close. Same-mount coordination is already handled by the per-mount +// fhLockTable and FUSE kernel serialization, so DLM is not needed for it. + +// TestDLMConcurrentRenames verifies that two concurrent renames of the same +// file from different mounts don't corrupt metadata. DLM locks on both old +// and new paths ensure renames are serialized. +func TestDLMConcurrentRenames(t *testing.T) { + if testing.Short() { + t.Skip("skipping DLM integration test in short mode") + } + + cluster := startDLMTestCluster(t) + t.Cleanup(cluster.Stop) + + // Create a file first + origPath := filepath.Join(cluster.mountPoints[0], "rename_race.txt") + require.NoError(t, os.WriteFile(origPath, []byte("original-content"), 0644)) + time.Sleep(1 * time.Second) // propagation + + // Both mounts try to rename the same file concurrently + var wg sync.WaitGroup + var errA, errB error + wg.Add(2) + + go func() { + defer wg.Done() + errA = os.Rename( + filepath.Join(cluster.mountPoints[0], "rename_race.txt"), + filepath.Join(cluster.mountPoints[0], "renamed_by_mount0.txt"), + ) + }() + + go func() { + defer wg.Done() + errB = os.Rename( + filepath.Join(cluster.mountPoints[1], "rename_race.txt"), + filepath.Join(cluster.mountPoints[1], "renamed_by_mount1.txt"), + ) + }() + + wg.Wait() + + // At least one rename should succeed; the other may fail with ENOENT + // since the source was already moved. + succeeded := 0 + if errA == nil { + succeeded++ + t.Logf("mount0 rename succeeded") + } else { + t.Logf("mount0 rename failed: %v", errA) + } + if errB == nil { + succeeded++ + t.Logf("mount1 rename succeeded") + } else { + t.Logf("mount1 rename failed: %v", errB) + } + assert.GreaterOrEqual(t, succeeded, 1, "at least one rename must succeed") +} diff --git a/test/fuse_dlm/framework_test.go b/test/fuse_dlm/framework_test.go new file mode 100644 index 000000000..0a6fcc5d7 --- /dev/null +++ b/test/fuse_dlm/framework_test.go @@ -0,0 +1,516 @@ +package fuse_dlm + +import ( + "context" + "fmt" + "net" + "os" + "os/exec" + "path/filepath" + "strconv" + "sync" + "syscall" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/cluster/lock_manager" + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" +) + +const filerGroup = "fuse-dlm-test" + +// dlmTestCluster manages a full SeaweedFS cluster with 2 filers and 2 FUSE +// mounts for testing DLM-based cross-mount write coordination. +type dlmTestCluster struct { + t testing.TB + baseDir string + weedBinary string + + masterPort int + masterGrpcPort int + volumePort int + volumeGrpcPort int + filerPorts [2]int + filerGrpcPorts [2]int + mountPoints [2]string + + masterCmd *exec.Cmd + volumeCmd *exec.Cmd + filerCmds [2]*exec.Cmd + mountCmds [2]*exec.Cmd + logFiles []*os.File + + cleanupOnce sync.Once +} + +func startDLMTestCluster(t testing.TB) *dlmTestCluster { + binary := findWeedBinary() + if binary == "" { + t.Skip("weed binary not found; set WEED_BINARY or ensure it is on PATH") + } + + baseDir, err := os.MkdirTemp("", "seaweedfs_fuse_dlm_test_") + require.NoError(t, err) + + c := &dlmTestCluster{ + t: t, + baseDir: baseDir, + weedBinary: binary, + } + // Register cleanup early so processes are stopped even if a require fails below. + t.Cleanup(c.Stop) + + // Allocate ports: master(2) + volume(2) + filer0(2) + filer1(2) = 8 + ports := allocatePorts(t, 8) + c.masterPort = ports[0] + c.masterGrpcPort = ports[1] + c.volumePort = ports[2] + c.volumeGrpcPort = ports[3] + c.filerPorts[0] = ports[4] + c.filerGrpcPorts[0] = ports[5] + c.filerPorts[1] = ports[6] + c.filerGrpcPorts[1] = ports[7] + + // Write empty security.toml + configDir := filepath.Join(baseDir, "config") + require.NoError(t, os.MkdirAll(configDir, 0755)) + require.NoError(t, os.WriteFile(filepath.Join(configDir, "security.toml"), []byte(""), 0644)) + + // Start master + require.NoError(t, c.startMaster(configDir)) + require.NoError(t, c.waitForTCP(fmt.Sprintf("127.0.0.1:%d", c.masterPort), 30*time.Second), + "master not ready\n%s", c.tailLog("master")) + + // Start volume + require.NoError(t, c.startVolume(configDir)) + require.NoError(t, c.waitForTCP(fmt.Sprintf("127.0.0.1:%d", c.volumePort), 30*time.Second), + "volume not ready\n%s", c.tailLog("volume")) + + // Start 2 filers + for i := 0; i < 2; i++ { + require.NoError(t, c.startFiler(i, configDir)) + require.NoError(t, c.waitForTCP(fmt.Sprintf("127.0.0.1:%d", c.filerGrpcPorts[i]), 30*time.Second), + "filer %d not ready\n%s", i, c.tailLog(fmt.Sprintf("filer%d", i))) + } + require.NoError(t, c.waitForFilerCount(2, 30*time.Second), "filer group registration") + require.NoError(t, c.waitForLockRingConverged(30*time.Second), "lock ring convergence") + + // Start 2 mounts, both pointing at filer0 for metadata consistency. + // (filer1 exists for the DLM lock ring but both mounts share filer0's + // metadata store since leveldb is per-filer.) + for i := 0; i < 2; i++ { + mp := filepath.Join(baseDir, fmt.Sprintf("mount%d", i)) + require.NoError(t, os.MkdirAll(mp, 0755)) + c.mountPoints[i] = mp + require.NoError(t, c.startMount(i, configDir)) + require.NoError(t, c.waitForMount(mp, 30*time.Second), + "mount %d not ready\n%s", i, c.tailLog(fmt.Sprintf("mount%d", i))) + } + + return c +} + +func (c *dlmTestCluster) Stop() { + if c == nil { + return + } + c.cleanupOnce.Do(func() { + // Stop mounts first (triggers flush + DLM unlock). + // Use stopCmd for bounded wait to avoid hanging on wedged FUSE processes. + for i := 1; i >= 0; i-- { + stopCmd(c.mountCmds[i]) + // Backup unmount in case FUSE teardown didn't clean up + exec.Command("fusermount3", "-u", c.mountPoints[i]).Run() + exec.Command("fusermount", "-u", c.mountPoints[i]).Run() + } + // Stop filers, volume, master + for i := 1; i >= 0; i-- { + stopCmd(c.filerCmds[i]) + } + stopCmd(c.volumeCmd) + stopCmd(c.masterCmd) + + for _, f := range c.logFiles { + f.Close() + } + + // Copy logs for CI + c.copyLogsForCI() + + if !c.t.Failed() { + os.RemoveAll(c.baseDir) + } + + // Wait for ports to be fully released before the next test + // allocates new ports (avoids TIME_WAIT collisions). + time.Sleep(2 * time.Second) + }) +} + +// masterAddress returns the master address in the format that encodes both +// HTTP and gRPC ports: "host:httpPort.grpcPort". This is the format that +// SeaweedFS uses to communicate non-default gRPC ports between components. +func (c *dlmTestCluster) masterAddress() string { + return string(pb.NewServerAddress("127.0.0.1", c.masterPort, c.masterGrpcPort)) +} + +func (c *dlmTestCluster) startMaster(configDir string) error { + c.masterCmd = exec.Command(c.weedBinary, + "-logdir="+filepath.Join(c.baseDir, "logs"), + "master", + "-ip=127.0.0.1", + "-ip.bind=127.0.0.1", + "-port="+strconv.Itoa(c.masterPort), + "-port.grpc="+strconv.Itoa(c.masterGrpcPort), + "-mdir="+filepath.Join(c.baseDir, "master"), + ) + return c.startCmd(c.masterCmd, "master") +} + +func (c *dlmTestCluster) startVolume(configDir string) error { + volDir := filepath.Join(c.baseDir, "volume") + if err := os.MkdirAll(volDir, 0755); err != nil { + return fmt.Errorf("create volume dir: %w", err) + } + c.volumeCmd = exec.Command(c.weedBinary, + "-logdir="+filepath.Join(c.baseDir, "logs"), + "volume", + "-ip=127.0.0.1", + "-ip.bind=127.0.0.1", + "-port="+strconv.Itoa(c.volumePort), + "-port.grpc="+strconv.Itoa(c.volumeGrpcPort), + "-master="+c.masterAddress(), + "-dir="+volDir, + "-max=10", + ) + return c.startCmd(c.volumeCmd, "volume") +} + +func (c *dlmTestCluster) startFiler(idx int, configDir string) error { + filerDir := filepath.Join(c.baseDir, fmt.Sprintf("filer%d", idx)) + if err := os.MkdirAll(filerDir, 0755); err != nil { + return fmt.Errorf("create filer dir: %w", err) + } + c.filerCmds[idx] = exec.Command(c.weedBinary, + "-logdir="+filepath.Join(c.baseDir, "logs"), + "filer", + "-ip=127.0.0.1", + "-ip.bind=127.0.0.1", + "-port="+strconv.Itoa(c.filerPorts[idx]), + "-port.grpc="+strconv.Itoa(c.filerGrpcPorts[idx]), + "-master="+c.masterAddress(), + "-filerGroup="+filerGroup, + "-defaultStoreDir="+filerDir, + ) + return c.startCmd(c.filerCmds[idx], fmt.Sprintf("filer%d", idx)) +} + +func (c *dlmTestCluster) filerAddress(idx int) string { + return string(pb.NewServerAddress("127.0.0.1", c.filerPorts[idx], c.filerGrpcPorts[idx])) +} + +func (c *dlmTestCluster) startMount(idx int, configDir string) error { + cacheDir := filepath.Join(c.baseDir, fmt.Sprintf("cache%d", idx)) + if err := os.MkdirAll(cacheDir, 0755); err != nil { + return fmt.Errorf("create cache dir: %w", err) + } + c.mountCmds[idx] = exec.Command(c.weedBinary, + "-logdir="+filepath.Join(c.baseDir, "logs"), + "mount", + "-filer="+c.filerAddress(0), // both mounts use filer0 for shared metadata + "-dir="+c.mountPoints[idx], + "-filer.path=/", + "-dirAutoCreate", + "-allowOthers=false", + "-cacheDir="+cacheDir, + "-dlm", + ) + return c.startCmd(c.mountCmds[idx], fmt.Sprintf("mount%d", idx)) +} + +func (c *dlmTestCluster) startCmd(cmd *exec.Cmd, name string) error { + logPath := filepath.Join(c.baseDir, "logs") + if err := os.MkdirAll(logPath, 0755); err != nil { + return fmt.Errorf("create log dir: %w", err) + } + logFile, err := os.Create(filepath.Join(logPath, name+".log")) + if err != nil { + return err + } + c.logFiles = append(c.logFiles, logFile) + cmd.Stdout = logFile + cmd.Stderr = logFile + return cmd.Start() +} + +func (c *dlmTestCluster) tailLog(name string) string { + data, err := os.ReadFile(filepath.Join(c.baseDir, "logs", name+".log")) + if err != nil { + return fmt.Sprintf("(log %s not available: %v)", name, err) + } + const maxTail = 8192 + if len(data) > maxTail { + data = data[len(data)-maxTail:] + } + return string(data) +} + +func (c *dlmTestCluster) copyLogsForCI() { + ciLogDir := "/tmp/seaweedfs-fuse-dlm-logs" + os.MkdirAll(ciLogDir, 0755) + logsDir := filepath.Join(c.baseDir, "logs") + entries, err := os.ReadDir(logsDir) + if err != nil { + return + } + for _, e := range entries { + data, err := os.ReadFile(filepath.Join(logsDir, e.Name())) + if err != nil { + continue + } + os.WriteFile(filepath.Join(ciLogDir, e.Name()), data, 0644) + } +} + +func (c *dlmTestCluster) waitForTCP(addr string, timeout time.Duration) error { + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + conn, err := net.DialTimeout("tcp", addr, time.Second) + if err == nil { + conn.Close() + return nil + } + time.Sleep(200 * time.Millisecond) + } + return fmt.Errorf("service at %s not ready within timeout", addr) +} + +// waitForMount waits for a FUSE filesystem to actually be mounted at +// mountPoint by comparing the device ID of the mount point against its parent. +// A plain directory (pre-created before mount) has the same device as its +// parent; a mounted FUSE filesystem has a different device. +func (c *dlmTestCluster) waitForMount(mountPoint string, timeout time.Duration) error { + parentDir := filepath.Dir(mountPoint) + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + parentStat, err := os.Stat(parentDir) + if err != nil { + time.Sleep(200 * time.Millisecond) + continue + } + mountStat, err := os.Stat(mountPoint) + if err != nil { + time.Sleep(200 * time.Millisecond) + continue + } + parentSys := parentStat.Sys().(*syscall.Stat_t) + mountSys := mountStat.Sys().(*syscall.Stat_t) + if parentSys.Dev != mountSys.Dev { + // Different device = FUSE mounted + return nil + } + time.Sleep(200 * time.Millisecond) + } + return fmt.Errorf("mount point %s not ready within timeout (FUSE not detected)", mountPoint) +} + +func (c *dlmTestCluster) filerGRPCAddress(idx int) string { + return fmt.Sprintf("127.0.0.1:%d", c.filerGrpcPorts[idx]) +} + +func (c *dlmTestCluster) waitForFilerCount(expected int, timeout time.Duration) error { + addr := fmt.Sprintf("127.0.0.1:%d", c.masterGrpcPort) + conn, err := grpc.NewClient(addr, grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + return err + } + defer conn.Close() + + client := master_pb.NewSeaweedClient(conn) + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + resp, err := client.ListClusterNodes(ctx, &master_pb.ListClusterNodesRequest{ + ClientType: "filer", + FilerGroup: filerGroup, + }) + cancel() + if err == nil && len(resp.ClusterNodes) >= expected { + return nil + } + time.Sleep(200 * time.Millisecond) + } + return fmt.Errorf("timed out waiting for %d filers in group %q", expected, filerGroup) +} + +// waitForLockRingConverged verifies that both filers have a consistent view of +// the lock ring by acquiring the same lock through each filer and checking +// mutual exclusion. Adapted from test/s3/distributed_lock/. +func (c *dlmTestCluster) waitForLockRingConverged(timeout time.Duration) error { + deadline := time.Now().Add(timeout) + + owners := []pb.ServerAddress{ + pb.ServerAddress(c.filerGRPCAddress(0)), + pb.ServerAddress(c.filerGRPCAddress(1)), + } + ring := lock_manager.NewHashRing(lock_manager.DefaultVnodeCount) + ring.SetServers(owners) + + attempt := 0 + for time.Now().Before(deadline) { + testKeys := convergenceKeysPerPrimary(ring, owners, attempt) + attempt++ + + allConverged := true + for _, key := range testKeys { + converged, _ := c.checkLockMutualExclusion(key) + if !converged { + allConverged = false + break + } + } + if allConverged { + return nil + } + time.Sleep(500 * time.Millisecond) + } + return fmt.Errorf("lock ring did not converge") +} + +// convergenceKeysPerPrimary generates one test key per primary filer. +func convergenceKeysPerPrimary(ring *lock_manager.HashRing, owners []pb.ServerAddress, attempt int) []string { + found := make(map[pb.ServerAddress]bool) + var keys []string + for i := 0; len(found) < len(owners) && i < 10000; i++ { + key := fmt.Sprintf("convergence-test-%d-%d", attempt, i) + primary, _ := ring.GetPrimaryAndBackup(key) + if !found[primary] { + found[primary] = true + keys = append(keys, key) + } + } + return keys +} + +func (c *dlmTestCluster) checkLockMutualExclusion(key string) (bool, error) { + // Try to lock via filer0 + conn0, err := grpc.NewClient(c.filerGRPCAddress(0), grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + return false, err + } + defer conn0.Close() + + client0 := filer_pb.NewSeaweedFilerClient(conn0) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + resp0, err := client0.DistributedLock(ctx, &filer_pb.LockRequest{ + Name: key, + SecondsToLock: 5, + Owner: "convergence-test-0", + }) + if err != nil { + return false, err + } + if resp0.Error != "" { + return false, fmt.Errorf("lock0: %s", resp0.Error) + } + + // Try to lock via filer1 — should fail (already locked) + conn1, err := grpc.NewClient(c.filerGRPCAddress(1), grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + return false, err + } + defer conn1.Close() + + client1 := filer_pb.NewSeaweedFilerClient(conn1) + resp1, err := client1.DistributedLock(ctx, &filer_pb.LockRequest{ + Name: key, + SecondsToLock: 5, + Owner: "convergence-test-1", + }) + + // Unlock via filer0 + client0.DistributedUnlock(ctx, &filer_pb.UnlockRequest{ + Name: key, + RenewToken: resp0.RenewToken, + }) + + if err != nil { + return false, err + } + // If filer1 also got the lock, the ring hasn't converged + if resp1.Error == "" { + // Unlock the second one too + client1.DistributedUnlock(ctx, &filer_pb.UnlockRequest{ + Name: key, + RenewToken: resp1.RenewToken, + }) + return false, nil + } + + return true, nil +} + +func stopCmd(cmd *exec.Cmd) { + if cmd == nil || cmd.Process == nil { + return + } + cmd.Process.Signal(syscall.SIGTERM) + done := make(chan struct{}) + go func() { + cmd.Wait() + close(done) + }() + select { + case <-done: + case <-time.After(10 * time.Second): + cmd.Process.Kill() + <-done + } +} + +func allocatePorts(t testing.TB, n int) []int { + t.Helper() + // Hold all listeners open until all ports are collected, then close + // them together. This prevents the OS from reassigning a just-freed + // port to the next Listen call within the same allocation batch. + listeners := make([]net.Listener, 0, n) + ports := make([]int, 0, n) + defer func() { + for _, l := range listeners { + l.Close() + } + }() + for i := 0; i < n; i++ { + l, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + listeners = append(listeners, l) + ports = append(ports, l.Addr().(*net.TCPAddr).Port) + } + return ports +} + +func findWeedBinary() string { + if p := os.Getenv("WEED_BINARY"); p != "" { + return p + } + if p, err := exec.LookPath("weed"); err == nil { + return p + } + candidates := []string{ + "../../weed/weed", + "./weed", + } + for _, c := range candidates { + if info, err := os.Stat(c); err == nil && !info.IsDir() { + abs, _ := filepath.Abs(c) + return abs + } + } + return "" +} diff --git a/weed/cluster/lock_client.go b/weed/cluster/lock_client.go index dc5a6ac77..3cc61ac4e 100644 --- a/weed/cluster/lock_client.go +++ b/weed/cluster/lock_client.go @@ -62,6 +62,46 @@ func (lc *LockClient) NewShortLivedLock(key string, owner string) (lock *LiveLoc return } +// NewBlockingLongLivedLock blocks until the lock is acquired, then starts a +// background renewal goroutine that keeps the lock alive. This combines the +// synchronous acquisition of NewShortLivedLock with the auto-renewal of +// StartLongLivedLock. Release with Stop(). +func (lc *LockClient) NewBlockingLongLivedLock(key, owner string, lockTTL time.Duration) *LiveLock { + if lockTTL == 0 { + lockTTL = lock_manager.LiveLockTTL + } + lock := &LiveLock{ + key: key, + hostFiler: lc.seedFiler, + cancelCh: make(chan struct{}), + expireAtNs: time.Now().Add(lockTTL).UnixNano(), + grpcDialOption: lc.grpcDialOption, + self: owner, + lc: lc, + lockTTL: lockTTL, + } + // Block until acquired + lock.retryUntilLocked(lockTTL) + // Start renewal goroutine using a ticker for interruptible sleep + go func() { + renewInterval := lockTTL / 2 + ticker := time.NewTicker(renewInterval) + defer ticker.Stop() + for { + select { + case <-lock.cancelCh: + return + case <-ticker.C: + if err := lock.AttemptToLock(lockTTL); err != nil { + glog.V(0).Infof("lock renewal failed for %s: %v", key, err) + atomic.StoreInt32(&lock.isLocked, 0) + } + } + } + }() + return lock +} + // StartLongLivedLock starts a goroutine to lock the key and returns immediately. // lockTTL specifies how long the lock should be held. The renewal interval is // automatically derived as lockTTL / 2 to ensure timely renewals. diff --git a/weed/command/mount.go b/weed/command/mount.go index 37bf2d15a..c82fbd9ec 100644 --- a/weed/command/mount.go +++ b/weed/command/mount.go @@ -52,6 +52,9 @@ type MountOptions struct { dirIdleEvictSec *int + // Distributed lock for cross-mount write coordination + distributedLock *bool + // FUSE performance options writebackCache *bool asyncDio *bool @@ -125,6 +128,9 @@ func init() { mountMemProfile = cmdMount.Flag.String("memprofile", "", "memory profile output file") mountReadRetryTime = cmdMount.Flag.Duration("readRetryTime", 6*time.Second, "maximum read retry wait time") + // Distributed lock for cross-mount write coordination + mountOptions.distributedLock = cmdMount.Flag.Bool("dlm", false, "enable distributed lock for cross-mount write coordination (only one mount can write a file at a time)") + // FUSE performance options mountOptions.writebackCache = cmdMount.Flag.Bool("writebackCache", false, "enable FUSE writeback cache for improved write performance (at risk of data loss on crash)") mountOptions.asyncDio = cmdMount.Flag.Bool("asyncDio", false, "enable async direct I/O for better concurrency") diff --git a/weed/command/mount_std.go b/weed/command/mount_std.go index 8cb617fbf..8228bbcd2 100644 --- a/weed/command/mount_std.go +++ b/weed/command/mount_std.go @@ -352,7 +352,8 @@ func RunMount(option *MountOptions, umask os.FileMode) bool { RdmaMaxConcurrent: *option.rdmaMaxConcurrent, RdmaTimeoutMs: *option.rdmaTimeoutMs, DirIdleEvictSec: *option.dirIdleEvictSec, - WritebackCache: option.writebackCache != nil && *option.writebackCache, + EnableDistributedLock: option.distributedLock != nil && *option.distributedLock, + WritebackCache: option.writebackCache != nil && *option.writebackCache, }) // create mount root diff --git a/weed/mount/filehandle.go b/weed/mount/filehandle.go index 485a00e41..ba4a2cdb0 100644 --- a/weed/mount/filehandle.go +++ b/weed/mount/filehandle.go @@ -5,6 +5,7 @@ import ( "sync" "github.com/seaweedfs/go-fuse/v2/fuse" + "github.com/seaweedfs/seaweedfs/weed/cluster" "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" @@ -38,6 +39,11 @@ type FileHandle struct { isDeleted bool isRenamed bool // set by Rename before waiting for async flush; skips old-path metadata flush + // dlmLock holds the distributed lock for cross-mount write coordination. + // Non-nil only when -dlm is enabled and the file was opened for writing. + // Acquired in AcquireHandle, released in ReleaseHandle. + dlmLock *cluster.LiveLock + // RDMA chunk offset cache for performance optimization chunkOffsetCache []int64 chunkCacheValid bool @@ -137,6 +143,13 @@ func (fh *FileHandle) AddChunks(chunks []*filer_pb.FileChunk) { } func (fh *FileHandle) ReleaseHandle() { + // Release distributed lock before cleaning up, so other mounts can + // proceed as soon as this handle is done flushing. + if fh.dlmLock != nil { + fh.dlmLock.Stop() + fh.dlmLock = nil + glog.V(1).Infof("DLM lock released for inode %d", fh.inode) + } fhActiveLock := fh.wfs.fhLockTable.AcquireLock("ReleaseHandle", fh.fh, util.ExclusiveLock) defer fh.wfs.fhLockTable.ReleaseLock(fh.fh, fhActiveLock) diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index 877d73f48..509e6d813 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -13,6 +13,7 @@ import ( "github.com/seaweedfs/go-fuse/v2/fuse" "google.golang.org/grpc" + "github.com/seaweedfs/seaweedfs/weed/cluster" "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/mount/meta_cache" @@ -82,6 +83,12 @@ type Option struct { // Directory cache refresh/eviction controls DirIdleEvictSec int + // EnableDistributedLock enables DLM-based write coordination across mounts. + // When true, opening a file for write acquires a distributed lock that is + // held (with auto-renewal) until the file is closed. Only one mount can + // have a file open for writing at a time. + EnableDistributedLock bool + // WritebackCache enables async flush on close for improved small file write performance. // When true, Flush() returns immediately and data upload + metadata flush happen in background. WritebackCache bool @@ -139,6 +146,10 @@ type WFS struct { // mutations (create, update, delete, rename). All mutations go through one // ordered stream to prevent cross-operation reordering. streamMutate *streamMutateMux + + // lockClient is the DLM client for cross-mount write coordination. + // Non-nil only when EnableDistributedLock is true. + lockClient *cluster.LockClient } const ( @@ -196,6 +207,11 @@ func NewSeaweedFileSystem(option *Option) *WFS { dirIdleEvict: dirIdleEvict, } + if option.EnableDistributedLock && len(option.FilerAddresses) > 0 { + wfs.lockClient = cluster.NewLockClient(option.GrpcDialOption, option.FilerAddresses[0]) + glog.V(0).Infof("distributed lock manager enabled for mount") + } + wfs.option.filerIndex = int32(rand.IntN(len(option.FilerAddresses))) wfs.option.setupUniqueCacheDirectory() if option.CacheSizeMBForRead > 0 { diff --git a/weed/mount/weedfs_file_mkrm.go b/weed/mount/weedfs_file_mkrm.go index 7a26a9c9d..8e9007203 100644 --- a/weed/mount/weedfs_file_mkrm.go +++ b/weed/mount/weedfs_file_mkrm.go @@ -2,10 +2,12 @@ package mount import ( "context" + "fmt" "syscall" "time" "github.com/seaweedfs/go-fuse/v2/fuse" + "github.com/seaweedfs/seaweedfs/weed/cluster/lock_manager" "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" @@ -110,6 +112,18 @@ func (wfs *WFS) Create(cancel <-chan struct{}, in *fuse.CreateIn, name string, o // Mark dirty so the deferred filer create happens on Flush, // even if the file is closed without any writes. fileHandle.dirtyMetadata = true + + // Acquire DLM lock for new file creation (Create bypasses AcquireHandle + // so we must acquire the lock here). Always lock on Create since file + // creation is inherently a write operation. + if wfs.lockClient != nil && fileHandle.dlmLock == nil { + owner := fmt.Sprintf("mount-%d", wfs.signature) + fileHandle.dlmLock = wfs.lockClient.NewBlockingLongLivedLock( + string(entryFullPath), owner, lock_manager.LiveLockTTL, + ) + glog.V(1).Infof("DLM lock acquired for new file %s", entryFullPath) + } + out.Fh = uint64(fileHandle.fh) out.OpenFlags = 0 diff --git a/weed/mount/weedfs_file_sync.go b/weed/mount/weedfs_file_sync.go index bb46b0605..55db0b165 100644 --- a/weed/mount/weedfs_file_sync.go +++ b/weed/mount/weedfs_file_sync.go @@ -167,6 +167,10 @@ func (wfs *WFS) doFlush(fh *FileHandle, uid, gid uint32, allowAsync bool) fuse.S // flushMetadataToFiler sends the file's chunk references and attributes to the filer. // This is shared between the synchronous doFlush path and the async flush completion. +// +// When -dlm is enabled, the distributed lock is already held by the FileHandle +// from open-for-write through close, so no additional distributed lock is +// needed here. The local fhLockTable lock below serializes within this mount. func (wfs *WFS) flushMetadataToFiler(fh *FileHandle, dir, name string, uid, gid uint32) error { fileFullPath := fh.FullPath() glog.V(4).Infof("flushMetadataToFiler %s/%s inode %d fh %d", dir, name, fh.inode, fh.fh) diff --git a/weed/mount/weedfs_filehandle.go b/weed/mount/weedfs_filehandle.go index 669b3190c..0d5ea2ef8 100644 --- a/weed/mount/weedfs_filehandle.go +++ b/weed/mount/weedfs_filehandle.go @@ -1,7 +1,10 @@ package mount import ( + "fmt" + "github.com/seaweedfs/go-fuse/v2/fuse" + "github.com/seaweedfs/seaweedfs/weed/cluster/lock_manager" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/util" @@ -30,6 +33,18 @@ func (wfs *WFS) AcquireHandle(inode uint64, flags, uid, gid uint32) (fileHandle // need to AcquireFileHandle again to ensure correct handle counter fileHandle = wfs.fhMap.AcquireFileHandle(wfs, inode, entry) fileHandle.RememberPath(path) + + // Acquire distributed lock for write opens. The lock is held with + // auto-renewal until the file handle is released (close). + // Use the filer path as the lock key since inode numbers are + // assigned per-mount and differ across mount instances. + if wfs.lockClient != nil && flags&fuse.O_ANYWRITE != 0 && fileHandle.dlmLock == nil { + owner := fmt.Sprintf("mount-%d", wfs.signature) + fileHandle.dlmLock = wfs.lockClient.NewBlockingLongLivedLock( + string(path), owner, lock_manager.LiveLockTTL, + ) + glog.V(1).Infof("DLM lock acquired for %s", path) + } } return } diff --git a/weed/mount/weedfs_metadata_flush.go b/weed/mount/weedfs_metadata_flush.go index bf6e6dd2b..93a25e6c8 100644 --- a/weed/mount/weedfs_metadata_flush.go +++ b/weed/mount/weedfs_metadata_flush.go @@ -85,6 +85,10 @@ func (wfs *WFS) flushAllDirtyMetadata() { // flushFileMetadata flushes the current file metadata to the filer without // flushing dirty pages from memory. This updates chunk references in the filer // so volume.fsck can see them, while keeping data in the write buffer. +// +// When -dlm is enabled, the distributed lock is already held by the FileHandle +// from open-for-write through close, so no additional distributed lock is +// needed here. The local fhLockTable lock below serializes within this mount. func (wfs *WFS) flushFileMetadata(fh *FileHandle) error { // Acquire exclusive lock on the file handle fhActiveLock := fh.wfs.fhLockTable.AcquireLock("flushMetadata", fh.fh, util.ExclusiveLock) diff --git a/weed/mount/weedfs_rename.go b/weed/mount/weedfs_rename.go index ef863ff44..7e56a3e27 100644 --- a/weed/mount/weedfs_rename.go +++ b/weed/mount/weedfs_rename.go @@ -10,6 +10,7 @@ import ( "github.com/seaweedfs/go-fuse/v2/fs" "github.com/seaweedfs/go-fuse/v2/fuse" + "github.com/seaweedfs/seaweedfs/weed/cluster/lock_manager" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/util" @@ -233,6 +234,41 @@ func (wfs *WFS) Rename(cancel <-chan struct{}, in *fuse.RenameIn, oldName string wfs.waitForPendingAsyncFlush(inode) } + // Acquire DLM locks on both old and new paths to prevent another mount + // from opening either path for writing during the rename. Lock in + // sorted order to prevent deadlocks when two mounts rename in opposite + // directions (A→B vs B→A). + // + // Skip the old-path lock if this mount already holds it via an open + // file handle (otherwise we'd deadlock trying to re-acquire our own lock). + if wfs.lockClient != nil { + owner := fmt.Sprintf("mount-%d", wfs.signature) + + // Check if the source file handle already holds a DLM lock on oldPath + oldPathAlreadyLocked := false + if sourceInode, found := wfs.inodeToPath.GetInode(oldPath); found { + if fh, ok := wfs.fhMap.FindFileHandle(sourceInode); ok && fh.dlmLock != nil { + oldPathAlreadyLocked = true + } + } + + // Determine which paths need new DLM locks + pathsToLock := []string{string(newPath)} + if !oldPathAlreadyLocked { + pathsToLock = append(pathsToLock, string(oldPath)) + } + // Sort for consistent lock ordering + if len(pathsToLock) == 2 && pathsToLock[0] > pathsToLock[1] { + pathsToLock[0], pathsToLock[1] = pathsToLock[1], pathsToLock[0] + } + + for _, p := range pathsToLock { + dlmLock := wfs.lockClient.NewBlockingLongLivedLock(p, owner, lock_manager.LiveLockTTL) + defer dlmLock.Stop() + } + glog.V(1).Infof("DLM locks acquired for rename %s => %s (oldPathAlreadyLocked=%v)", oldPath, newPath, oldPathAlreadyLocked) + } + // update remote filer request := &filer_pb.StreamRenameEntryRequest{ OldDirectory: string(oldDir), @@ -296,6 +332,23 @@ func (wfs *WFS) handleRenameResponse(ctx context.Context, resp *filer_pb.StreamR // Keep the saved handle path current so any flush fallback // after Forget uses the post-rename location, not the old one. fh.RememberPath(newPath) + + // Migrate the DLM lock from old path to new path so the + // lock key matches the current file location. Hold the + // fhLockTable to prevent ReleaseHandle from concurrently + // stopping the lock during migration. + if wfs.lockClient != nil { + fhActiveLock := wfs.fhLockTable.AcquireLock("renameDLM", fh.fh, util.ExclusiveLock) + if fh.dlmLock != nil { + owner := fmt.Sprintf("mount-%d", wfs.signature) + fh.dlmLock.Stop() + fh.dlmLock = wfs.lockClient.NewBlockingLongLivedLock( + string(newPath), owner, lock_manager.LiveLockTTL, + ) + glog.V(1).Infof("DLM lock migrated from %s to %s", oldPath, newPath) + } + wfs.fhLockTable.ReleaseLock(fh.fh, fhActiveLock) + } } // invalidate attr and data // wfs.fuseServer.InodeNotify(sourceInode, 0, -1)