mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-08-19 05:36:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

windows mount: cache file data, resolved paths and attributes (#10703)

Browse Source 
* benchmark tool for mounted filesystems

* ci: on-demand mount benchmark, native WinFsp vs rclone plus a Linux reference

* windows mount: let the Windows cache manager cache file data

WinFsp only turns the cache manager on for a file when FileInfoTimeout
is infinite; at any finite value every application read and write is a
synchronous trip into the mount process at whatever size the application
issued. Metadata events already reach FspFileSystemNotify, which purges
a changed file's cached pages and attributes, so an infinite timeout
stays coherent. The dir listing, volume info and EA timeouts are pinned
to one second so they do not silently inherit the infinity.

* windows mount: cache resolved paths and attributes in the adapter

WinFsp addresses every operation by path and has no FORGET, so the
adapter walked the whole path through Lookup on each one, and in a
directory the filer has not listed yet every walk was a filer round
trip; nothing played the part of the kernel's dentry and attribute
caches. The path cache owns one lookup reference per entry the way the
kernel holds one until FORGET, serves attribute reads for files without
an open handle, and is purged by the mount's own mutations and by
metadata events, with the timeout as backstop.

* windows mount: keep a closed file's attributes cached

Open steals the path's cache entry for its handle and Release returned
the reference with a purge, so the stat that follows every copied file
walked to the filer again. Reading the handle's final attributes before
it goes away and moving the reference back into the cache serves that
stat locally, the way the kernel's attribute cache does after a close.

Only if the path still names that inode, though: WinFsp reports the
path the handle opened with, and after a delete-on-close or a rename
caching it would resurrect an entry that is gone.

* windows mount: persist entries at create, and let the flush stay at close

WinFsp posts the cleanup and close that carry the flush after
CloseHandle has returned, so deferring the filer entry to the flush let
everything that reads through the filer race an unflushed close: a
listing missed just-written files, and a directory rename moved a
directory on the filer before its newest child existed there, leaving
the straggler flush to recreate the child under the dead path.

Flush-at-cleanup is not the answer either: it makes every handle's
cleanup flush, and those flushes race the unlinks of delete-on-close,
re-inserting the entry the unlink just removed. Persisting the entry at
create takes the ordering question away.

* mount: flush written pages before a truncate shrinks past them

The shrink trims chunks, but written pages that have not become chunks
yet are invisible to it, so the next flush wrote them back and the file
grew again, resurrecting the truncated bytes. Windows hits this on
every write-then-shrink because its flush runs after CloseHandle, but
the gap is platform-neutral.

* mount: order a file's unlink against its in-flight flush

Unlink set the handle's deleted flag bare, so a flush already past its
own check of that flag wrote the entry back right after the delete
removed it, and a delete-on-close file outlived its last handle. The
flag is now set under the handle's flush lock and re-checked under it,
so a flush either completes before the delete or sees the flag and
skips. An eagerly created handle also starts clean: the dirty mark
existed to make the deferred filer create happen at flush, and eager
creates have nothing to flush.
This commit is contained in:
Chris Lu
2026-08-10 18:46:18 -07:00
committed by GitHub
parent c6e1387f59
commit 214d3599d3
18 changed files with 1261 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/mount-benchmark.yml
+203
View File
@@ -0,0 +1,203 @@
name: "mount: benchmark"
# Manual benchmark: native WinFsp mount vs rclone+WebDAV on the same Windows
# runner, with a Linux FUSE mount of the same build as a reference. Numbers
# from shared runners are noisy; this is for finding factor-of-N gaps, not
# regressions of a few percent.
on:
workflow_dispatch:
push:
branches: [ 'winfsp-bench**' ]
concurrency:
group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
cancel-in-progress: true
permissions:
contents: read
jobs:
bench-windows:
name: Windows native vs rclone
runs-on: windows-latest
timeout-minutes: 60
env:
CGO_ENABLED: 0
steps:
- uses: actions/checkout@v7
with:
persist-credentials: false
- uses: actions/setup-go@v7
with:
go-version-file: 'go.mod'
- name: Install WinFsp and rclone
run: choco install winfsp rclone -y --no-progress
- name: Build weed.exe
run: go build -o weed.exe ./weed
- name: Benchmark both mounts
shell: pwsh
run: |
$ErrorActionPreference = 'Stop'
function Test-Port($port) {
$client = New-Object System.Net.Sockets.TcpClient
try { $client.Connect('127.0.0.1', $port); return $client.Connected }
catch { return $false }
finally { $client.Dispose() }
}
function Wait-Drive($drive, $what) {
$deadline = (Get-Date).AddMinutes(2)
while ((Get-Date) -lt $deadline) {
if (Test-Path "${drive}\") { Write-Host "$what is mounted on $drive"; return }
Start-Sleep -Seconds 2
}
throw "$what never appeared on $drive"
}
function Invoke-Bench($dir, $label, $out) {
Write-Host "::group::bench $label"
& go run ./test/mount_bench -dir $dir -label $label -filer 127.0.0.1:8888 -out $out
$code = $LASTEXITCODE
Write-Host "::endgroup::"
if ($code -ne 0) { throw "bench $label failed with exit $code" }
}
New-Item -ItemType Directory -Force -Path C:\seaweed-data | Out-Null
Start-Process -FilePath .\weed.exe `
-ArgumentList '-logtostderr','mini','-dir=C:\seaweed-data','-ip=127.0.0.1' `
-RedirectStandardOutput C:\seaweed-mini.log -RedirectStandardError C:\seaweed-mini.err.log
$deadline = (Get-Date).AddMinutes(3)
while ((Get-Date) -lt $deadline) {
if ((Test-Port 8888) -and (Test-Port 18888) -and (Test-Port 7333)) { break }
Start-Sleep -Seconds 3
}
if (-not ((Test-Port 8888) -and (Test-Port 18888) -and (Test-Port 7333))) {
Get-Content C:\seaweed-mini.log, C:\seaweed-mini.err.log -ErrorAction SilentlyContinue
throw "mini cluster never came up"
}
Write-Host "filer on 8888/18888, webdav on 7333"
# --- native WinFsp mount ---
Start-Process -FilePath .\weed.exe `
-ArgumentList '-logtostderr','mount','-filer=127.0.0.1:8888','-dir=S:' `
-RedirectStandardOutput C:\seaweed-mount.log -RedirectStandardError C:\seaweed-mount.err.log
Wait-Drive 'S:' 'weed mount'
Invoke-Bench 'S:\bench-native' 'winfsp-native' 'C:\results-native.json'
Get-CimInstance Win32_Process -Filter "Name = 'weed.exe'" |
Where-Object { $_.CommandLine -like '*mount*' } |
ForEach-Object { Stop-Process -Id $_.ProcessId -Force }
$deadline = (Get-Date).AddMinutes(1)
while ((Get-Date) -lt $deadline -and (Test-Path S:\)) { Start-Sleep -Seconds 2 }
# --- rclone + WebDAV on the same WinFsp ---
# rclone serves listings from a directory cache it fills lazily, so the
# big-listing files have to exist before it mounts or it never sees them.
& go run ./test/mount_bench -filer 127.0.0.1:8888 -seed bench-rclone/biglist
if ($LASTEXITCODE -ne 0) { throw "seeding failed" }
$env:RCLONE_CONFIG_SEAWEED_TYPE = 'webdav'
$env:RCLONE_CONFIG_SEAWEED_URL = 'http://127.0.0.1:7333'
$env:RCLONE_CONFIG_SEAWEED_VENDOR = 'other'
Start-Process -FilePath rclone `
-ArgumentList 'mount','seaweed:','T:','--vfs-cache-mode=writes','-v','--log-file=C:\rclone.log'
Wait-Drive 'T:' 'rclone mount'
Invoke-Bench 'T:\bench-rclone' 'rclone-webdav' 'C:\results-rclone.json'
Stop-Process -Name rclone -Force -ErrorAction SilentlyContinue
# --- comparison ---
$table = & go run ./test/mount_bench -compare C:\results-native.json,C:\results-rclone.json
$table | Write-Host
"## Windows: native WinFsp vs rclone+WebDAV" | Out-File -Append $env:GITHUB_STEP_SUMMARY
$table | Out-File -Append $env:GITHUB_STEP_SUMMARY
- name: Logs
if: always()
shell: pwsh
run: |
foreach ($f in 'C:\seaweed-mount.log','C:\seaweed-mount.err.log','C:\rclone.log','C:\seaweed-mini.log','C:\seaweed-mini.err.log') {
if (Test-Path $f) { Write-Host "===== $f"; Get-Content $f -Tail 100 }
}
- name: Results
if: always()
uses: actions/upload-artifact@v4
with:
name: results-windows
path: C:\results-*.json
if-no-files-found: ignore
bench-linux:
name: Linux FUSE reference
runs-on: ubuntu-latest
timeout-minutes: 60
steps:
- uses: actions/checkout@v7
with:
persist-credentials: false
- uses: actions/setup-go@v7
with:
go-version-file: 'go.mod'
- name: Repair the fusermount3 setuid bit
uses: ./.github/actions/fix-fusermount-setuid
- name: Allow non-root FUSE mounts with allow_other
run: |
echo 'user_allow_other' | sudo tee -a /etc/fuse.conf
sudo chmod 644 /etc/fuse.conf
- name: Build weed
run: go build -o /tmp/weed ./weed
- name: Benchmark FUSE mount
run: |
set -e
mkdir -p /tmp/seaweed-data
/tmp/weed -logtostderr mini -dir=/tmp/seaweed-data -ip=127.0.0.1 > /tmp/mini.log 2>&1 &
for i in $(seq 1 60); do
if nc -z 127.0.0.1 8888 && nc -z 127.0.0.1 18888; then break; fi
sleep 3
done
nc -z 127.0.0.1 8888 || { cat /tmp/mini.log; echo "filer never came up"; exit 1; }
mkdir -p "$HOME/mnt"
/tmp/weed -logtostderr mount -filer=127.0.0.1:8888 -dir="$HOME/mnt" > /tmp/mount.log 2>&1 &
for i in $(seq 1 60); do
if mountpoint -q "$HOME/mnt"; then break; fi
sleep 2
done
mountpoint -q "$HOME/mnt" || { cat /tmp/mount.log; echo "mount never appeared"; exit 1; }
go run ./test/mount_bench -dir "$HOME/mnt/bench-linux" -label linux-fuse -filer 127.0.0.1:8888 -out /tmp/results-linux.json
{
echo "## Linux FUSE reference"
go run ./test/mount_bench -compare /tmp/results-linux.json
} >> "$GITHUB_STEP_SUMMARY"
- name: Logs
if: always()
run: |
tail -n 100 /tmp/mount.log /tmp/mini.log 2>/dev/null || true
- name: Results
if: always()
uses: actions/upload-artifact@v4
with:
name: results-linux
path: /tmp/results-linux.json
if-no-files-found: ignore
test/mount_bench/main.go
+408
View File
@@ -0,0 +1,408 @@
// Command mount_bench times common filesystem operations against a mounted
// directory and writes the results as JSON, so runs against different mounts
// (WinFsp, rclone, libfuse) can be compared with -compare.
package main
import (
"bytes"
"encoding/json"
"flag"
"fmt"
"io"
"log"
"math/rand"
"mime/multipart"
"net/http"
"os"
"path/filepath"
"strings"
"sync"
"time"
)
const (
seqFileSize = 256 << 20
seqBufSize = 1 << 20
randReads = 500
randReadSize = 4 << 10
smallCount = 400
smallSize = 64 << 10
overwriteN = 100
statRounds = 3
listRounds = 10
bigListCount = 5000
bigListRounds = 3
)
type phaseResult struct {
Name string `json:"name"`
Seconds float64 `json:"seconds"`
Bytes int64 `json:"bytes,omitempty"`
Ops int64 `json:"ops,omitempty"`
}
type runResult struct {
Label string `json:"label"`
Phases []phaseResult `json:"phases"`
}
func main() {
dir := flag.String("dir", "", "directory on the mount to benchmark in; created if missing, must be a direct child of the mount root")
label := flag.String("label", "bench", "name for this result set")
out := flag.String("out", "", "write results as JSON to this file")
filer := flag.String("filer", "", "filer http address for out-of-band setup of the big-listing phase")
seed := flag.String("seed", "", "seed the big-listing directory at this filer-relative path and exit; for mounts that cache listings and would miss files created behind them")
compare := flag.String("compare", "", "comma-separated result JSON files; print a comparison table and exit")
flag.Parse()
if *compare != "" {
if err := printComparison(strings.Split(*compare, ",")); err != nil {
log.Fatal(err)
}
return
}
if *seed != "" {
if *filer == "" {
log.Fatal("-seed requires -filer")
}
if err := seedFilerDir(*filer, *seed, bigListCount); err != nil {
log.Fatal(err)
}
return
}
if *dir == "" {
log.Fatal("either -dir or -compare is required")
}
if err := os.MkdirAll(*dir, 0755); err != nil {
log.Fatal(err)
}
r := &runner{dir: *dir, filer: *filer, result: runResult{Label: *label}}
r.run()
data, err := json.MarshalIndent(r.result, "", " ")
if err != nil {
log.Fatal(err)
}
if *out != "" {
if err := os.WriteFile(*out, data, 0644); err != nil {
log.Fatal(err)
}
}
fmt.Println(string(data))
}
type runner struct {
dir string
filer string
result runResult
}
func (r *runner) phase(name string, bytes, ops int64, fn func() error) {
start := time.Now()
if err := fn(); err != nil {
log.Fatalf("%s: %v", name, err)
}
elapsed := time.Since(start)
r.result.Phases = append(r.result.Phases, phaseResult{
Name: name, Seconds: elapsed.Seconds(), Bytes: bytes, Ops: ops,
})
fmt.Printf("%-14s %10.3fs%s\n", name, elapsed.Seconds(), rateSuffix(bytes, ops, elapsed.Seconds()))
}
func rateSuffix(bytes, ops int64, seconds float64) string {
if seconds <= 0 {
return ""
}
var parts []string
if bytes > 0 {
parts = append(parts, fmt.Sprintf("%8.1f MB/s", float64(bytes)/seconds/1e6))
}
if ops > 0 {
parts = append(parts, fmt.Sprintf("%8.1f ops/s", float64(ops)/seconds))
}
if len(parts) == 0 {
return ""
}
return " " + strings.Join(parts, " ")
}
func (r *runner) run() {
buf := make([]byte, seqBufSize)
rnd := rand.New(rand.NewSource(42))
rnd.Read(buf)
bigFile := filepath.Join(r.dir, "big.dat")
r.phase("seq_write", seqFileSize, 0, func() error {
f, err := os.Create(bigFile)
if err != nil {
return err
}
for written := 0; written < seqFileSize; written += len(buf) {
if _, err := f.Write(buf); err != nil {
f.Close()
return err
}
}
return f.Close()
})
r.phase("seq_read", seqFileSize, 0, func() error {
f, err := os.Open(bigFile)
if err != nil {
return err
}
defer f.Close()
n, err := io.CopyBuffer(io.Discard, f, buf)
if err != nil {
return err
}
if n != seqFileSize {
return fmt.Errorf("read %d bytes, want %d", n, seqFileSize)
}
return nil
})
r.phase("rand_read", randReads*randReadSize, randReads, func() error {
f, err := os.Open(bigFile)
if err != nil {
return err
}
defer f.Close()
block := make([]byte, randReadSize)
for i := 0; i < randReads; i++ {
off := rnd.Int63n(seqFileSize - randReadSize)
if _, err := f.ReadAt(block, off); err != nil {
return err
}
}
return nil
})
smallDir := filepath.Join(r.dir, "small")
small := buf[:smallSize]
r.phase("small_write", smallCount*smallSize, smallCount, func() error {
if err := os.MkdirAll(smallDir, 0755); err != nil {
return err
}
for i := 0; i < smallCount; i++ {
if err := os.WriteFile(filepath.Join(smallDir, fmt.Sprintf("f%04d.dat", i)), small, 0644); err != nil {
return err
}
}
return nil
})
r.phase("small_read", smallCount*smallSize, smallCount, func() error {
for i := 0; i < smallCount; i++ {
data, err := os.ReadFile(filepath.Join(smallDir, fmt.Sprintf("f%04d.dat", i)))
if err != nil {
return err
}
if len(data) != smallSize {
return fmt.Errorf("file %d: read %d bytes, want %d", i, len(data), smallSize)
}
}
return nil
})
r.phase("stat_files", 0, smallCount*statRounds, func() error {
for round := 0; round < statRounds; round++ {
for i := 0; i < smallCount; i++ {
if _, err := os.Stat(filepath.Join(smallDir, fmt.Sprintf("f%04d.dat", i))); err != nil {
return err
}
}
}
return nil
})
r.phase("list_dir", 0, listRounds, func() error {
for round := 0; round < listRounds; round++ {
entries, err := os.ReadDir(smallDir)
if err != nil {
return err
}
if len(entries) != smallCount {
return fmt.Errorf("listed %d entries, want %d", len(entries), smallCount)
}
}
return nil
})
r.phase("overwrite", overwriteN*smallSize, overwriteN, func() error {
for i := 0; i < overwriteN; i++ {
if err := os.WriteFile(filepath.Join(smallDir, fmt.Sprintf("f%04d.dat", i)), small, 0644); err != nil {
return err
}
}
return nil
})
r.phase("delete", 0, smallCount, func() error {
for i := 0; i < smallCount; i++ {
if err := os.Remove(filepath.Join(smallDir, fmt.Sprintf("f%04d.dat", i))); err != nil {
return err
}
}
return nil
})
if r.filer != "" {
r.bigListPhase()
}
}
// bigListPhase enumerates a directory whose files were created directly
// against the filer, so the listing is measured on its own rather than after
// this process has created (and possibly cached) every entry itself.
func (r *runner) bigListPhase() {
// The bench dir is a direct child of the mount root, so its base name is
// also its path on the filer.
mountName := filepath.Base(r.dir)
bigDir := filepath.Join(r.dir, "biglist")
if err := seedFilerDir(r.filer, mountName+"/biglist", bigListCount); err != nil {
log.Fatalf("seeding %d files via filer: %v", bigListCount, err)
}
r.phase("list_5k", 0, bigListCount*bigListRounds, func() error {
for round := 0; round < bigListRounds; round++ {
entries, err := os.ReadDir(bigDir)
if err != nil {
return err
}
if len(entries) != bigListCount {
return fmt.Errorf("listed %d entries, want %d", len(entries), bigListCount)
}
}
return nil
})
r.phase("walk_5k", 0, bigListCount, func() error {
entries, err := os.ReadDir(bigDir)
if err != nil {
return err
}
for _, entry := range entries {
if _, err := entry.Info(); err != nil {
return err
}
}
return nil
})
}
// seedFilerDir creates count tiny files under dir on the filer over plain
// HTTP, in parallel, bypassing the mount entirely.
func seedFilerDir(filer, dir string, count int) error {
content := []byte("x")
var wg sync.WaitGroup
errs := make(chan error, count)
sem := make(chan struct{}, 64)
client := &http.Client{Timeout: 30 * time.Second}
for i := 0; i < count; i++ {
wg.Add(1)
sem <- struct{}{}
go func(i int) {
defer wg.Done()
defer func() { <-sem }()
url := fmt.Sprintf("http://%s/%s/img%05d.jpg", filer, dir, i)
if err := uploadOne(client, url, content); err != nil {
errs <- fmt.Errorf("%s: %w", url, err)
}
}(i)
}
wg.Wait()
close(errs)
return <-errs
}
func uploadOne(client *http.Client, url string, content []byte) error {
var body bytes.Buffer
writer := multipart.NewWriter(&body)
part, err := writer.CreateFormFile("file", filepath.Base(url))
if err != nil {
return err
}
if _, err := part.Write(content); err != nil {
return err
}
if err := writer.Close(); err != nil {
return err
}
req, err := http.NewRequest(http.MethodPost, url, &body)
if err != nil {
return err
}
req.Header.Set("Content-Type", writer.FormDataContentType())
resp, err := client.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
io.Copy(io.Discard, resp.Body)
if resp.StatusCode >= 300 {
return fmt.Errorf("status %s", resp.Status)
}
return nil
}
// printComparison renders the named result files side by side as a markdown
// table, one row per phase, with throughput where the phase moved bytes and
// operation rate otherwise.
func printComparison(files []string) error {
var runs []runResult
for _, file := range files {
data, err := os.ReadFile(strings.TrimSpace(file))
if err != nil {
return err
}
var run runResult
if err := json.Unmarshal(data, &run); err != nil {
return fmt.Errorf("%s: %w", file, err)
}
runs = append(runs, run)
}
if len(runs) == 0 {
return fmt.Errorf("no result files")
}
// Preserve the phase order of the first run, appending any extras.
var order []string
seen := map[string]bool{}
for _, run := range runs {
for _, p := range run.Phases {
if !seen[p.Name] {
seen[p.Name] = true
order = append(order, p.Name)
}
}
}
byName := make([]map[string]phaseResult, len(runs))
for i, run := range runs {
byName[i] = map[string]phaseResult{}
for _, p := range run.Phases {
byName[i][p.Name] = p
}
}
header := []string{"phase"}
for _, run := range runs {
header = append(header, run.Label+" (s)", run.Label+" rate")
}
fmt.Println("| " + strings.Join(header, " | ") + " |")
fmt.Println("|" + strings.Repeat("---|", len(header)))
for _, name := range order {
row := []string{name}
for i := range runs {
p, ok := byName[i][name]
if !ok {
row = append(row, "-", "-")
continue
}
row = append(row, fmt.Sprintf("%.2f", p.Seconds), strings.TrimSpace(rateSuffix(p.Bytes, p.Ops, p.Seconds)))
}
fmt.Println("| " + strings.Join(row, " | ") + " |")
}
return nil
}
weed/command/mount_common.go
+7
View File
@@ -175,6 +175,12 @@ type fileSystemParams struct {
chunkSizeLimitMB int
cacheDirForRead string
cacheDirForWrite string
// eagerFilerCreate persists a created file's entry at create time rather
// than at flush. A platform sets it when it cannot run the flush before
// the application's close returns, so nothing that reads through the
// filer can race an unflushed close.
eagerFilerCreate bool
}
func buildSeaweedFileSystem(option *MountOptions, p fileSystemParams) *mount.WFS {
@@ -225,6 +231,7 @@ func buildSeaweedFileSystem(option *MountOptions, p fileSystemParams) *mount.WFS
DirIdleEvictSec: *option.dirIdleEvictSec,
EnableDistributedLock: option.distributedLock != nil && *option.distributedLock,
WritebackCache: option.writebackCache != nil && *option.writebackCache,
EagerFilerCreate: p.eagerFilerCreate,
PosixDirNlink: option.posixDirNlink != nil && *option.posixDirNlink,
// Peer chunk sharing
PeerEnabled: option.peerEnabled != nil && *option.peerEnabled,
weed/command/mount_windows.go
+10 -5
View File
@@ -23,10 +23,11 @@ import (
// every other client would read the entry as owned by uid 4294967295.
const ownedByMounter = ^uint32(0)
// windowsAttrTimeoutSec bounds how long WinFsp may serve cached attributes.
// Nothing invalidates that cache from this side, so it stays short rather than
// borrowing an unrelated flag's value.
const windowsAttrTimeoutSec = 1.0
// windowsCacheTimeout bounds how long the adapter serves cached paths and
// attributes and WinFsp serves cached directory listings, matching what the
// kernel caches would be allowed on a unix mount. Metadata events purge
// entries earlier; this is the backstop for anything an event misses.
const windowsCacheTimeout = time.Second
func RunMount(option *MountOptions, umask os.FileMode) bool {
chunkSizeLimitMB := *mountOptions.chunkSizeLimitMB
@@ -92,6 +93,10 @@ func RunMount(option *MountOptions, umask os.FileMode) bool {
chunkSizeLimitMB: chunkSizeLimitMB,
cacheDirForRead: cacheDirForRead,
cacheDirForWrite: cacheDirForWrite,
// WinFsp posts the cleanup and close that carry the flush after
// CloseHandle has already returned, so entry creation cannot wait
// for the flush the way it does when close(2) runs it synchronously.
eagerFilerCreate: true,
})
if !createMountRoot(seaweedFileSystem, mountRoot, bucketRootPath, filerAddresses) {
@@ -102,7 +107,7 @@ func RunMount(option *MountOptions, umask os.FileMode) bool {
VolumeName: strings.ReplaceAll(*option.filer, ",", "+"),
Uid: ownedByMounter,
Gid: ownedByMounter,
AttrTimeout: windowsAttrTimeoutSec,
CacheTimeout: windowsCacheTimeout,
ReadOnly: *option.readOnly,
Debug: *option.debugFuse,
ExtraOptions: option.extraOptions,
weed/mount/dirty_pages_chunked.go
+4
View File
@@ -55,6 +55,10 @@ func (pages *ChunkedDirtyPages) AddPage(offset int64, data []byte, isSequential
return err
}
func (pages *ChunkedDirtyPages) HasWrites() bool {
return pages.hasWrites
}
func (pages *ChunkedDirtyPages) FlushData() error {
if !pages.hasWrites {
return nil
weed/mount/page_writer.go
+4
View File
@@ -53,6 +53,10 @@ func (pw *PageWriter) FlushData() error {
return pw.randomWriter.FlushData()
}
func (pw *PageWriter) HasWrites() bool {
return pw.randomWriter.HasWrites()
}
func (pw *PageWriter) ReadDirtyDataAt(data []byte, offset int64, tsNs int64) (maxStop int64) {
glog.V(4).Infof("ReadDirtyDataAt %v [%d, %d)", pw.fh.inode, offset, offset+int64(len(data)))
weed/mount/page_writer/dirty_pages.go
+1
View File
@@ -3,6 +3,7 @@ package page_writer
type DirtyPages interface {
AddPage(offset int64, data []byte, isSequential bool, tsNs int64) error
FlushData() error
HasWrites() bool
ReadDirtyDataAt(data []byte, startOffset int64, tsNs int64) (maxStop int64)
Destroy()
LockForRead(startOffset, stopOffset int64)
weed/mount/weedfs.go
+23
View File
@@ -120,6 +120,13 @@ type Option struct {
// When true, Flush() returns immediately and data upload + metadata flush happen in background.
WritebackCache bool
// EagerFilerCreate persists a created file's entry at create time instead
// of deferring it to flush. Deferring is only safe when the flush runs
// before the application's close returns; a platform that cannot
// guarantee that sets this so listings and reopens through the filer
// cannot race an unflushed close.
EagerFilerCreate bool
// PosixDirNlink enables POSIX-compliant directory nlink counting
// (nlink = 2 + number_of_subdirectories). This requires listing
// cached directory entries on every stat, which has a performance cost.
@@ -784,6 +791,22 @@ func sameEntryContent(a, b *filer_pb.Entry) bool {
// subscription event. No filer lookup here: it can fail transiently, and with
// the subscription cursor already past the event, nothing would retry.
// IsFileOpen reports whether any open file handle refers to the inode, which
// tells a caching front end that the handle's view of the file, not its own
// cached copy, is current.
func (wfs *WFS) IsFileOpen(inode uint64) bool {
_, found := wfs.fhMap.FindFileHandle(inode)
return found
}
// PathForInode reports the path the mount currently tracks for the inode, so
// a caching front end can tell whether a path it resolved earlier still names
// the same file. An unlinked file has no path while its handles drain.
func (wfs *WFS) PathForInode(inode uint64) (util.FullPath, bool) {
path, status := wfs.inodeToPath.GetPath(inode)
return path, status == fuse.OK
}
// SetEntryChangeListener registers a callback for every metadata event this
// mount applies. A front end whose client caches entries on its own side, and
// which the mount cannot invalidate directly, uses it to push the change out.
weed/mount/weedfs_attr.go
+13
View File
@@ -1,6 +1,7 @@
package mount
import (
"context"
"os"
"syscall"
"time"
@@ -75,6 +76,18 @@ func (wfs *WFS) SetAttr(cancel <-chan struct{}, input *fuse.SetAttrIn, out *fuse
if status != fuse.OK || entry == nil {
return status
}
if size, ok := input.GetSize(); ok && fh != nil && fh.dirtyPages.HasWrites() && size < filer.FileSize(entry) {
// The truncation below trims chunks; dirty pages it cannot see. Left
// alone, pages beyond the new size come back with the next flush and
// grow the file again, so turn them into chunks first. Runs before
// the entry locks below: the flush takes its own.
ctx, cancelFunc := context.WithTimeout(context.Background(), metadataFlushTimeout)
flushStatus := wfs.doFlush(ctx, fh, input.Uid, input.Gid, false)
cancelFunc()
if flushStatus != fuse.OK {
return flushStatus
}
}
if fh != nil {
fh.entryLock.Lock()
defer fh.entryLock.Unlock()
weed/mount/weedfs_attr_race_test.go
+2
View File
@@ -45,6 +45,7 @@ func TestAttrChunkRace(t *testing.T) {
entry: &LockedEntry{Entry: entry},
entryChunkGroup: chunkGroup,
}
fh.dirtyPages = newPageWriter(fh, 1<<20)
wfs.fhMap.inode2fh[inode] = fh
wfs.fhMap.fh2inode[fh.fh] = inode
@@ -124,6 +125,7 @@ func TestReadFromChunksRace(t *testing.T) {
entry: &LockedEntry{Entry: entry},
entryChunkGroup: chunkGroup,
}
fh.dirtyPages = newPageWriter(fh, 1<<20)
wfs.fhMap.inode2fh[inode] = fh
wfs.fhMap.fh2inode[fh.fh] = inode
weed/mount/weedfs_file_mkrm.go
+22 -10
View File
@@ -72,7 +72,7 @@ func (wfs *WFS) Create(cancel <-chan struct{}, in *fuse.CreateIn, name string, o
return code
}
inode, newEntry, code = wfs.createRegularFile(dirFullPath, name, in.Mode, in.Uid, in.Gid, 0, true, true)
inode, newEntry, code = wfs.createRegularFile(dirFullPath, name, in.Mode, in.Uid, in.Gid, 0, !wfs.option.EagerFilerCreate, true)
if code == fuse.Status(syscall.EEXIST) && in.Flags&syscall.O_EXCL == 0 {
// Race: another process created the file between our check and create.
// Reopen the winner's entry.
@@ -117,9 +117,10 @@ func (wfs *WFS) Create(cancel <-chan struct{}, in *fuse.CreateIn, name string, o
fileHandle.SetEntry(newEntry)
}
fileHandle.RememberPath(entryFullPath)
// Mark dirty so the deferred filer create happens on Flush,
// even if the file is closed without any writes.
fileHandle.dirtyMetadata = true
// Mark dirty so the deferred filer create happens on Flush, even if the
// file is closed without any writes. An eager create has already
// persisted the entry, so its handle starts clean.
fileHandle.dirtyMetadata = !wfs.option.EagerFilerCreate
// Acquire DLM lock for new file creation (Create bypasses AcquireHandle
// so we must acquire the lock here). Always lock on Create since file
@@ -244,15 +245,11 @@ func (wfs *WFS) Unlink(cancel <-chan struct{}, header *fuse.InHeader, name strin
// finishes first; even if it recreated the entry, the filer delete below
// will remove it again.
if inode, found := wfs.inodeToPath.GetInode(entryFullPath); found {
if fh, fhFound := wfs.fhMap.FindFileHandle(inode); fhFound {
fh.isDeleted = true
}
wfs.markHandleDeleted(inode)
wfs.waitForPendingAsyncFlush(inode)
} else if entry != nil && entry.Attributes != nil && entry.Attributes.Inode != 0 {
inodeFromEntry := entry.Attributes.Inode
if fh, fhFound := wfs.fhMap.FindFileHandle(inodeFromEntry); fhFound {
fh.isDeleted = true
}
wfs.markHandleDeleted(inodeFromEntry)
wfs.waitForPendingAsyncFlush(inodeFromEntry)
}
@@ -438,6 +435,21 @@ func (wfs *WFS) createRegularFile(dirFullPath util.FullPath, name string, mode u
return inode, newEntry, fuse.OK
}
// markHandleDeleted flags the inode's open handle so its flushes stop writing
// the entry back. Taken and released under the handle's flush lock: a flush
// already holding it finishes before the caller's delete runs, and any later
// flush sees the flag; setting the flag bare raced the flush's own check and
// resurrected the entry right after the delete.
func (wfs *WFS) markHandleDeleted(inode uint64) {
fh, found := wfs.fhMap.FindFileHandle(inode)
if !found {
return
}
fhActiveLock := wfs.fhLockTable.AcquireLock("Unlink", fh.fh, util.ExclusiveLock)
fh.isDeleted = true
wfs.fhLockTable.ReleaseLock(fh.fh, fhActiveLock)
}
// asyncCreateEntry sends a CreateEntry RPC to the filer in the background.
// The entry is already in the local meta cache; this persists it to the filer.
// Used by Mknod with writeback caching — the node is visible locally right away.
weed/mount/weedfs_file_sync.go
+8
View File
@@ -210,6 +210,14 @@ func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, n
fhActiveLock := fh.wfs.fhLockTable.AcquireLock("doFlush", fh.fh, util.ExclusiveLock)
defer fh.wfs.fhLockTable.ReleaseLock(fh.fh, fhActiveLock)
// Re-check under the lock: Unlink sets the flag under it, so a flush that
// was already past the earlier check cannot write the entry back after
// the delete removed it.
if fh.isDeleted {
glog.V(3).Infof("flushMetadataToFiler %s fh %d: file was unlinked, skipping", fileFullPath, fh.fh)
return nil
}
entry := fh.GetEntry()
entry.Name = name // this flush may be just after a rename operation
weed/mount/winfsp/fs_windows.go
+198 -98
View File
@@ -1,14 +1,17 @@
package winfsp
import (
"strings"
"sync"
"syscall"
"time"
cgofuse "github.com/winfsp/cgofuse/fuse"
"github.com/seaweedfs/go-fuse/v2/fuse"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/mount"
"github.com/seaweedfs/seaweedfs/weed/util"
)
const (
@@ -29,6 +32,11 @@ const (
// WinFsp. Bounded so a directory with millions of children does not
// materialise in one slice.
readdirBatch = 4096
// stealAttempts bounds the resolve-then-steal loop in the open paths. A
// miss needs the entry to be evicted in the instant between the two calls,
// so a second round is already unlikely.
stealAttempts = 4
)
// never is a nil channel: receiving blocks forever, which is what the raw
@@ -39,12 +47,17 @@ var never chan struct{}
type WinFS struct {
cgofuse.FileSystemBase
wfs *mount.WFS
uid uint32
gid uint32
readOnly bool
wfs *mount.WFS
mountRoot util.FullPath
uid uint32
gid uint32
readOnly bool
// An open handle holds the lookup reference for its inode until Release,
// paths holds the lookup references for resolved paths, standing in for
// the kernel caches of the unix mounts.
paths *pathCache
// An open handle holds a lookup reference for its inode until Release,
// the way the kernel keeps one for an open file. WinFsp hands back only
// the handle, and the raw filesystem reuses one handle for repeated opens
// of the same inode, so the references are counted.
@@ -53,15 +66,37 @@ type WinFS struct {
dirInodes map[uint64]*handleRef
}
func NewWinFS(wfs *mount.WFS, uid, gid uint32, readOnly bool) *WinFS {
return &WinFS{
func NewWinFS(wfs *mount.WFS, uid, gid uint32, readOnly bool, cacheTimeout time.Duration) *WinFS {
w := &WinFS{
wfs: wfs,
mountRoot: wfs.MountRoot(),
uid: uid,
gid: gid,
readOnly: readOnly,
fileInodes: make(map[uint64]*handleRef),
dirInodes: make(map[uint64]*handleRef),
}
w.paths = newPathCache(cacheTimeout, w.forget)
return w
}
// stillNames reports whether the mount still tracks inode at the mount-relative
// key, which a delete-on-close or a rename while the handle was open has
// changed.
func (w *WinFS) stillNames(key string, inode uint64) bool {
current, ok := w.wfs.PathForInode(inode)
if !ok {
return false
}
rel, ok := relativeToMount(w.mountRoot, current)
return ok && cacheKey(rel) == key
}
// invalidatePath is what the notifier calls when a metadata event arrives:
// whatever is cached for that path is out of date. Directories take their
// subtree with them, since the children's cached paths point through them.
func (w *WinFS) invalidatePath(path string, isDirectory bool) {
w.paths.purge(cacheKey(path), isDirectory)
}
// handleRef is the lookup references an open handle is holding, one per open
@@ -142,36 +177,6 @@ func (w *WinFS) caller(inode uint64) fuse.InHeader {
}
}
// lookupRef collects the references a resolution took. Every operation that
// hands back an EntryOut grants the caller one, which the Linux kernel returns
// with FORGET. WinFsp has no FORGET, so anything the adapter looks up it has
// to release itself or the mount's inode table grows for the life of the
// process.
type lookupRef struct {
w *WinFS
inodes []uint64
}
// release returns every reference still held.
func (r *lookupRef) release() {
if r == nil {
return
}
for _, inode := range r.inodes {
r.w.forget(inode)
}
r.inodes = nil
}
// keepLast hands the reference on the final component to the caller, which
// becomes responsible for releasing it. An open handle takes it this way and
// gives it back on Release.
func (r *lookupRef) keepLast() {
if r != nil && len(r.inodes) > 0 {
r.inodes = r.inodes[:len(r.inodes)-1]
}
}
// forget returns one reference. The root is never looked up, so it never holds
// one to give back.
func (w *WinFS) forget(inode uint64) {
@@ -181,42 +186,66 @@ func (w *WinFS) forget(inode uint64) {
w.wfs.Forget(inode, 1)
}
// walk resolves a path one component at a time. Lookup does more than find an
// inode: it refreshes what the mount knows about the entry, so the result of a
// preceding truncate or write is visible to the caller.
func (w *WinFS) walk(parts []string) (uint64, *lookupRef, fuse.Status) {
ref := &lookupRef{w: w}
// walk resolves a path one component at a time, filling the path cache as it
// goes. Every reference a lookup grants is owned by the cache, which keeps the
// returned inode alive for at least one cache sweep; an operation that needs
// to hold the inode longer steals the reference from the cache.
func (w *WinFS) walk(parts []string) (uint64, fuse.Status) {
inode := uint64(rootInode)
for _, name := range parts {
for i, name := range parts {
key := strings.Join(parts[:i+1], "/")
if cached, _, ok := w.paths.lookup(key); ok {
inode = cached
continue
}
var out fuse.EntryOut
if status := w.wfs.Lookup(never, ptr(w.caller(inode)), name, &out); status != fuse.OK {
ref.release()
return 0, nil, status
return 0, status
}
w.paths.insert(key, out.NodeId, out.Attr)
inode = out.NodeId
ref.inodes = append(ref.inodes, inode)
}
return inode, ref, fuse.OK
return inode, fuse.OK
}
// resolve walks a WinFsp path down to an inode. The caller must release the
// returned reference.
func (w *WinFS) resolve(path string) (uint64, *lookupRef, fuse.Status) {
// resolve walks a WinFsp path down to an inode, valid at least until the next
// cache sweep.
func (w *WinFS) resolve(path string) (uint64, fuse.Status) {
return w.walk(splitPath(path))
}
// resolveParent resolves everything but the last component, which the create
// and delete operations need separately. The caller must release the reference.
func (w *WinFS) resolveParent(path string) (uint64, string, *lookupRef, fuse.Status) {
// and delete operations need separately.
func (w *WinFS) resolveParent(path string) (uint64, string, fuse.Status) {
parentParts, name, ok := splitParent(path)
if !ok {
return 0, "", nil, fuse.EINVAL
return 0, "", fuse.EINVAL
}
parent, ref, status := w.walk(parentParts)
parent, status := w.walk(parentParts)
if status != fuse.OK {
return 0, "", nil, status
return 0, "", status
}
return parent, name, ref, fuse.OK
return parent, name, fuse.OK
}
// resolveAndSteal resolves path and takes over the cache's reference on the
// final inode, for the open paths that hold it for the life of a handle. The
// root is handed out without a reference; it does not need one.
func (w *WinFS) resolveAndSteal(path string) (uint64, fuse.Status) {
key := cacheKey(path)
for attempt := 0; attempt < stealAttempts; attempt++ {
inode, status := w.resolve(path)
if status != fuse.OK {
return 0, status
}
if key == "" {
return inode, fuse.OK
}
if stolen, ok := w.paths.steal(key); ok {
return stolen, fuse.OK
}
}
return 0, fuse.EIO
}
func (w *WinFS) attrToStat(attr *fuse.Attr, stat *cgofuse.Stat_t) {
@@ -292,17 +321,39 @@ func (w *WinFS) Statfs(path string, stat *cgofuse.Statfs_t) int {
return 0
}
// getattrFromCache serves attributes the way the kernel would from its
// attribute cache. A file with an open handle is excluded: its live size is
// on the handle, not in the cache.
func (w *WinFS) getattrFromCache(key string, stat *cgofuse.Stat_t) bool {
if key == "" {
return false
}
inode, attr, ok := w.paths.lookup(key)
if !ok || w.wfs.IsFileOpen(inode) {
return false
}
w.attrToStat(&attr, stat)
return true
}
func (w *WinFS) Getattr(path string, stat *cgofuse.Stat_t, fh uint64) int {
inode := w.inodeForHandle(w.fileInodes, fh)
if inode == 0 {
var ref *lookupRef
key := cacheKey(path)
if fh == noHandle && w.getattrFromCache(key, stat) {
return 0
}
var status fuse.Status
inode, ref, status = w.resolve(path)
inode, status = w.resolve(path)
if status != fuse.OK {
logResolveFailure("getattr", path, status)
return toErrno(status)
}
defer ref.release()
// The walk just refreshed the cache, so this hits unless the file is
// open or is the root, and saves the second load of the same entry.
if fh == noHandle && w.getattrFromCache(key, stat) {
return 0
}
}
in := &fuse.GetAttrIn{InHeader: w.caller(inode)}
if fh != noHandle {
@@ -317,21 +368,31 @@ func (w *WinFS) Getattr(path string, stat *cgofuse.Stat_t, fh uint64) int {
return 0
}
// purgeWithParent drops a mutated path from the cache along with its parent,
// whose cached mtime the mutation just outdated.
func (w *WinFS) purgeWithParent(path string, prefix bool) {
key := cacheKey(path)
w.paths.purge(key, prefix)
if i := strings.LastIndexByte(key, '/'); i > 0 {
w.paths.purge(key[:i], false)
}
}
func (w *WinFS) Mkdir(path string, mode uint32) int {
if w.denied() {
return -eROFS
}
parent, name, ref, status := w.resolveParent(path)
parent, name, status := w.resolveParent(path)
if status != fuse.OK {
return toErrno(status)
}
defer ref.release()
in := &fuse.MkdirIn{InHeader: w.caller(parent), Mode: mode}
var out fuse.EntryOut
status = w.wfs.Mkdir(never, in, name, &out)
if status == fuse.OK {
// Mkdir hands back an EntryOut, and nothing on this side will forget it.
w.forget(out.NodeId)
w.purgeWithParent(path, false)
// The new directory is about to be filled; cache it, reference and all.
w.paths.insert(cacheKey(path), out.NodeId, out.Attr)
}
return toErrno(status)
}
@@ -340,60 +401,72 @@ func (w *WinFS) Rmdir(path string) int {
if w.denied() {
return -eROFS
}
parent, name, ref, status := w.resolveParent(path)
parent, name, status := w.resolveParent(path)
if status != fuse.OK {
return toErrno(status)
}
defer ref.release()
return toErrno(w.wfs.Rmdir(never, ptr(w.caller(parent)), name))
status = w.wfs.Rmdir(never, ptr(w.caller(parent)), name)
if status == fuse.OK {
w.purgeWithParent(path, true)
}
return toErrno(status)
}
func (w *WinFS) Unlink(path string) int {
if w.denied() {
return -eROFS
}
parent, name, ref, status := w.resolveParent(path)
parent, name, status := w.resolveParent(path)
if status != fuse.OK {
return toErrno(status)
}
defer ref.release()
return toErrno(w.wfs.Unlink(never, ptr(w.caller(parent)), name))
status = w.wfs.Unlink(never, ptr(w.caller(parent)), name)
if status == fuse.OK {
w.purgeWithParent(path, false)
}
return toErrno(status)
}
func (w *WinFS) Rename(oldpath string, newpath string) int {
if w.denied() {
return -eROFS
}
oldParent, oldName, oldRef, status := w.resolveParent(oldpath)
oldParent, oldName, status := w.resolveParent(oldpath)
if status != fuse.OK {
return toErrno(status)
}
defer oldRef.release()
newParent, newName, newRef, status := w.resolveParent(newpath)
newParent, newName, status := w.resolveParent(newpath)
if status != fuse.OK {
return toErrno(status)
}
defer newRef.release()
in := &fuse.RenameIn{InHeader: w.caller(oldParent), Newdir: newParent}
return toErrno(w.wfs.Rename(never, in, oldName, newName))
status = w.wfs.Rename(never, in, oldName, newName)
if status == fuse.OK {
// Directories carry their subtree's cached paths with them, and the
// rename may have replaced an entry at the destination.
w.purgeWithParent(oldpath, true)
w.purgeWithParent(newpath, true)
}
return toErrno(status)
}
func (w *WinFS) Create(path string, flags int, mode uint32) (int, uint64) {
if w.denied() {
return -eROFS, noHandle
}
parent, name, ref, status := w.resolveParent(path)
parent, name, status := w.resolveParent(path)
if status != fuse.OK {
glog.Errorf("create %s: resolving the parent directory: %v", path, status)
return toErrno(status), noHandle
}
defer ref.release()
in := &fuse.CreateIn{InHeader: w.caller(parent), Flags: translateOpenFlags(flags), Mode: mode}
var out fuse.CreateOut
if status := w.wfs.Create(never, in, name, &out); status != fuse.OK {
glog.Errorf("create %s in inode %d: %v", name, parent, status)
return toErrno(status), noHandle
}
// Whatever the cache held for this path was replaced by the create.
w.purgeWithParent(path, false)
// Create grants a reference on the new inode; hold it for as long as the
// handle lives and give it back in Release.
w.retain(w.fileInodes, out.Fh, out.NodeId)
@@ -401,19 +474,18 @@ func (w *WinFS) Create(path string, flags int, mode uint32) (int, uint64) {
}
func (w *WinFS) Open(path string, flags int) (int, uint64) {
inode, ref, status := w.resolve(path)
inode, status := w.resolveAndSteal(path)
if status != fuse.OK {
logResolveFailure("open", path, status)
return toErrno(status), noHandle
}
defer ref.release()
in := &fuse.OpenIn{InHeader: w.caller(inode), Flags: translateOpenFlags(flags)}
var out fuse.OpenOut
if status := w.wfs.Open(never, in, &out); status != fuse.OK {
glog.Errorf("open %s inode %d: %v", path, inode, status)
w.forget(inode)
return toErrno(status), noHandle
}
ref.keepLast()
w.retain(w.fileInodes, out.Fh, inode)
return 0, out.Fh
}
@@ -469,13 +541,11 @@ func (w *WinFS) Truncate(path string, size int64, fh uint64) int {
}
inode := w.inodeForHandle(w.fileInodes, fh)
if inode == 0 {
var ref *lookupRef
var status fuse.Status
inode, ref, status = w.resolve(path)
inode, status = w.resolve(path)
if status != fuse.OK {
return toErrno(status)
}
defer ref.release()
}
in := &fuse.SetAttrIn{}
in.NodeId = inode
@@ -486,35 +556,41 @@ func (w *WinFS) Truncate(path string, size int64, fh uint64) int {
in.Fh = fh
}
var out fuse.AttrOut
return toErrno(w.wfs.SetAttr(never, in, &out))
status := w.wfs.SetAttr(never, in, &out)
if status == fuse.OK {
w.paths.purge(cacheKey(path), false)
}
return toErrno(status)
}
func (w *WinFS) Chmod(path string, mode uint32) int {
if w.denied() {
return -eROFS
}
inode, ref, status := w.resolve(path)
inode, status := w.resolve(path)
if status != fuse.OK {
return toErrno(status)
}
defer ref.release()
in := &fuse.SetAttrIn{}
in.NodeId = inode
in.Valid = fuse.FATTR_MODE
in.Mode = mode
var out fuse.AttrOut
return toErrno(w.wfs.SetAttr(never, in, &out))
status = w.wfs.SetAttr(never, in, &out)
if status == fuse.OK {
w.paths.purge(cacheKey(path), false)
}
return toErrno(status)
}
func (w *WinFS) Utimens(path string, tmsp []cgofuse.Timespec) int {
if w.denied() {
return -eROFS
}
inode, ref, status := w.resolve(path)
inode, status := w.resolve(path)
if status != fuse.OK {
return toErrno(status)
}
defer ref.release()
in := &fuse.SetAttrIn{}
in.NodeId = inode
if len(tmsp) < 2 {
@@ -536,7 +612,11 @@ func (w *WinFS) Utimens(path string, tmsp []cgofuse.Timespec) int {
}
}
var out fuse.AttrOut
return toErrno(w.wfs.SetAttr(never, in, &out))
status = w.wfs.SetAttr(never, in, &out)
if status == fuse.OK {
w.paths.purge(cacheKey(path), false)
}
return toErrno(status)
}
// applyTimespec reports whether a timestamp should be written. UTIME_OMIT asks
@@ -574,22 +654,45 @@ func (w *WinFS) Release(path string, fh uint64) int {
if fh == noHandle {
return 0
}
// The handle's view of the file is the freshest there is, and the close
// has flushed by now, so read it before the handle goes away: the next
// operation on the path — a stat right after a copy, most often — is then
// served from the cache instead of walking to the filer again.
var attr *fuse.Attr
if inode := w.inodeForHandle(w.fileInodes, fh); inode != 0 {
in := &fuse.GetAttrIn{InHeader: w.caller(inode)}
in.Fh_ = fh
in.Flags_ = fuse.FUSE_GETATTR_FH
var out fuse.AttrOut
if w.wfs.GetAttr(never, in, &out) == fuse.OK {
attr = &out.Attr
}
}
w.wfs.Release(never, &fuse.ReleaseIn{Fh: fh})
w.forget(w.releaseRetained(w.fileInodes, fh))
if inode := w.releaseRetained(w.fileInodes, fh); inode != 0 {
// The reference moves from the handle to the cache — but only if the
// path still names this inode: WinFsp reports the path the handle
// opened with, and after a delete-on-close or a rename caching it
// would resurrect an entry that is gone.
if key := cacheKey(path); attr != nil && w.stillNames(key, inode) {
w.paths.insert(key, inode, *attr)
} else {
w.forget(inode)
}
}
return 0
}
func (w *WinFS) Opendir(path string) (int, uint64) {
inode, ref, status := w.resolve(path)
inode, status := w.resolveAndSteal(path)
if status != fuse.OK {
return toErrno(status), noHandle
}
defer ref.release()
var out fuse.OpenOut
if status := w.wfs.OpenDir(never, &fuse.OpenIn{InHeader: w.caller(inode)}, &out); status != fuse.OK {
w.forget(inode)
return toErrno(status), noHandle
}
ref.keepLast()
w.retain(w.dirInodes, out.Fh, inode)
return 0, out.Fh
}
@@ -657,13 +760,11 @@ func (s *readdirSink) TakesLookupRef() bool { return false }
func (w *WinFS) Readdir(path string, fill func(name string, stat *cgofuse.Stat_t, ofst int64) bool, ofst int64, fh uint64) int {
inode := w.inodeForHandle(w.dirInodes, fh)
if inode == 0 {
var ref *lookupRef
var status fuse.Status
inode, ref, status = w.resolve(path)
inode, status = w.resolve(path)
if status != fuse.OK {
return toErrno(status)
}
defer ref.release()
}
offset := uint64(ofst)
for {
@@ -719,11 +820,10 @@ func (w *WinFS) Readlink(path string) (int, string) {
if len(splitPath(path)) == 0 {
return -eNOSYS, ""
}
inode, ref, status := w.resolve(path)
inode, status := w.resolve(path)
if status != fuse.OK {
return toErrno(status), ""
}
defer ref.release()
target, status := w.wfs.Readlink(never, ptr(w.caller(inode)))
if status != fuse.OK {
return toErrno(status), ""
weed/mount/winfsp/host_windows.go
+38 -10
View File
@@ -4,6 +4,7 @@ import (
"fmt"
"strconv"
"strings"
"time"
cgofuse "github.com/winfsp/cgofuse/fuse"
@@ -21,9 +22,11 @@ type Options struct {
Uid uint32
Gid uint32
// AttrTimeout bounds how long WinFsp caches attributes, in seconds.
// Nothing invalidates its cache, so this is the only coherence knob.
AttrTimeout float64
// CacheTimeout bounds how long resolved paths and attributes may be
// served from the adapter's cache, and how long WinFsp may serve cached
// directory listings. Metadata events shorten it by purging; nothing
// else invalidates these caches.
CacheTimeout time.Duration
// ReadOnly rejects every modification. WinFsp has no "ro" option — it
// discards the flag and leaves the volume writable — so the refusal has
@@ -33,28 +36,31 @@ type Options struct {
// Debug turns on cgofuse's operation trace.
Debug bool
// ExtraOptions are passed through to WinFsp as -o arguments.
// ExtraOptions are passed through to WinFsp as -o arguments, after the
// defaults, so they win when they name the same option.
ExtraOptions []string
}
// Host is a WinFsp mount that has not been started yet.
type Host struct {
host *cgofuse.FileSystemHost
fs *WinFS
options Options
}
// New wires wfs up to WinFsp. Nothing is mounted until Serve.
func New(wfs *mount.WFS, options Options) *Host {
host := cgofuse.NewFileSystemHost(NewWinFS(wfs, options.Uid, options.Gid, options.ReadOnly))
fs := NewWinFS(wfs, options.Uid, options.Gid, options.ReadOnly, options.CacheTimeout)
host := cgofuse.NewFileSystemHost(fs)
host.SetCapReaddirPlus(true)
host.SetUseIno(true)
return &Host{host: host, options: options}
return &Host{host: host, fs: fs, options: options}
}
// Notify wires the mount's metadata events to Windows. Called once before
// Serve; the host has to exist first, which is why it is not done in New.
func (h *Host) Notify(wfs *mount.WFS) {
n := &notifier{host: h.host, mountRoot: wfs.MountRoot()}
n := &notifier{host: h.host, fs: h.fs, mountRoot: wfs.MountRoot()}
wfs.SetEntryChangeListener(n.notify)
}
@@ -66,10 +72,32 @@ func (h *Host) Serve(mountPoint string) error {
"-o", "volname=" + h.volumeName(),
"-o", "uid=-1",
"-o", "gid=-1",
// Only an infinite FileInfoTimeout lets the Windows cache manager
// cache file data; at any finite value every application read and
// write is a synchronous trip into this process at whatever size the
// application issued. Remote changes stay visible because every
// applied metadata event goes through Notify, which purges the file's
// cached pages along with its attributes.
//
// KeepFileCache is deliberately absent: it would keep the cache alive
// past cleanup, deferring the close — and with it the flush that
// persists a written file — until Windows reclaims the memory.
"-o", "FileInfoTimeout=-1",
// FlushOnCleanup is absent for the same reason: it makes every
// handle's cleanup flush, and those flushes race the unlinks of
// delete-on-close. The flush stays at close, which WinFsp runs after
// CloseHandle has returned; the mount persists entries eagerly at
// create instead, so nothing that reads through the filer depends on
// when the flush runs.
}
if h.options.AttrTimeout > 0 {
timeout := strconv.FormatFloat(h.options.AttrTimeout, 'f', -1, 64)
opts = append(opts, "-o", "attr_timeout="+timeout, "-o", "entry_timeout="+timeout)
if h.options.CacheTimeout > 0 {
ms := strconv.FormatInt(h.options.CacheTimeout.Milliseconds(), 10)
// These would silently inherit the infinite FileInfoTimeout.
opts = append(opts,
"-o", "DirInfoTimeout="+ms,
"-o", "VolumeInfoTimeout="+ms,
"-o", "EaTimeout="+ms,
)
}
if h.options.Debug {
opts = append(opts, "-d")
weed/mount/winfsp/notify_windows.go
+5 -1
View File
@@ -14,11 +14,13 @@ import (
// stays invisible until the user refreshes by hand.
type notifier struct {
host *cgofuse.FileSystemHost
fs *WinFS
mountRoot util.FullPath
}
// notify reports one applied event. Windows wants the path relative to the
// mount, and an action describing what happened to it.
// mount, and an action describing what happened to it; the adapter's own path
// cache is out of date for the same path.
func (n *notifier) notify(invalidation meta_cache.EntryInvalidation) {
if n.host == nil {
return
@@ -27,6 +29,8 @@ func (n *notifier) notify(invalidation meta_cache.EntryInvalidation) {
// describing the new, so reporting RenamedTo here as well would send the
// destination twice — and as a create even when a directory moved.
if path, ok := relativeToMount(n.mountRoot, invalidation.Path); ok {
isDirectory := invalidation.WasDirectory || invalidation.Entry.GetIsDirectory()
n.fs.invalidatePath(path, isDirectory)
n.send(path, n.action(invalidation))
}
}
weed/mount/winfsp/pathcache.go
+154
View File
@@ -0,0 +1,154 @@
package winfsp
import (
"strings"
"sync"
"time"
"github.com/seaweedfs/go-fuse/v2/fuse"
)
// pathCache stands in for the dentry and attribute caches the kernel provides
// on the unix mounts. WinFsp addresses every operation by path, so without it
// each operation walks the whole path through Lookup again, and in a directory
// the filer has not listed yet every one of those lookups is a filer round
// trip.
//
// The cache owns one lookup reference per entry, the way the kernel holds one
// until it sends FORGET. An evicted reference sits out one sweep in the
// graveyard before it is returned, so an operation that resolved just before
// the eviction is not left holding a reclaimed inode.
type pathCache struct {
ttl time.Duration
forget func(inode uint64)
mu sync.Mutex
entries map[string]*pathCacheEntry
graveyard []uint64
lastSweep time.Time
}
type pathCacheEntry struct {
inode uint64
attr fuse.Attr
expires time.Time
}
// maxCachedPaths bounds the references parked here. Overflow clears the whole
// cache rather than tracking recency: entries expire within ttl anyway, so
// exact eviction order buys nothing.
const maxCachedPaths = 64 << 10
func newPathCache(ttl time.Duration, forget func(inode uint64)) *pathCache {
return &pathCache{
ttl: ttl,
forget: forget,
entries: map[string]*pathCacheEntry{},
lastSweep: time.Now(),
}
}
// cacheKey canonicalises a WinFsp path. The root maps to "", which is never
// cached: its inode is fixed and holds no reference.
func cacheKey(path string) string {
return strings.Join(splitPath(path), "/")
}
// lookup reports the cached inode and attributes for key. The entry stays in
// the cache; the inode remains valid at least one sweep past its expiry.
func (c *pathCache) lookup(key string) (inode uint64, attr fuse.Attr, ok bool) {
if key == "" {
return 0, fuse.Attr{}, false
}
c.mu.Lock()
defer c.mu.Unlock()
entry, found := c.entries[key]
if !found || time.Now().After(entry.expires) {
return 0, fuse.Attr{}, false
}
return entry.inode, entry.attr, true
}
// insert takes ownership of one lookup reference on inode.
func (c *pathCache) insert(key string, inode uint64, attr fuse.Attr) {
if key == "" {
c.forget(inode)
return
}
var pending []uint64
c.mu.Lock()
if existing, found := c.entries[key]; found {
c.graveyard = append(c.graveyard, existing.inode)
} else if len(c.entries) >= maxCachedPaths {
for _, entry := range c.entries {
c.graveyard = append(c.graveyard, entry.inode)
}
c.entries = map[string]*pathCacheEntry{}
}
c.entries[key] = &pathCacheEntry{inode: inode, attr: attr, expires: time.Now().Add(c.ttl)}
pending = c.sweepLocked()
c.mu.Unlock()
c.forgetAll(pending)
}
// steal removes key and hands its reference to the caller.
func (c *pathCache) steal(key string) (inode uint64, ok bool) {
c.mu.Lock()
defer c.mu.Unlock()
entry, found := c.entries[key]
if !found {
return 0, false
}
delete(c.entries, key)
return entry.inode, true
}
// purge drops key, and everything under it when prefix is set, which a rename
// or removal of a directory needs: the children's cached paths name entries
// that are no longer there.
func (c *pathCache) purge(key string, prefix bool) {
var pending []uint64
c.mu.Lock()
if entry, found := c.entries[key]; found {
c.graveyard = append(c.graveyard, entry.inode)
delete(c.entries, key)
}
if prefix {
under := key + "/"
for k, entry := range c.entries {
if key == "" || strings.HasPrefix(k, under) {
c.graveyard = append(c.graveyard, entry.inode)
delete(c.entries, k)
}
}
}
pending = c.sweepLocked()
c.mu.Unlock()
c.forgetAll(pending)
}
// sweepLocked returns the previous graveyard for the caller to forget outside
// the lock, and moves expired entries into the next one. Sweeps run at most
// once per ttl, so a reference rests here for at least one full ttl.
func (c *pathCache) sweepLocked() []uint64 {
now := time.Now()
if now.Sub(c.lastSweep) < c.ttl {
return nil
}
c.lastSweep = now
pending := c.graveyard
c.graveyard = nil
for key, entry := range c.entries {
if now.After(entry.expires) {
c.graveyard = append(c.graveyard, entry.inode)
delete(c.entries, key)
}
}
return pending
}
func (c *pathCache) forgetAll(inodes []uint64) {
for _, inode := range inodes {
c.forget(inode)
}
}
weed/mount/winfsp/pathcache_test.go
+157
View File
@@ -0,0 +1,157 @@
package winfsp
import (
"sync"
"testing"
"time"
"github.com/seaweedfs/go-fuse/v2/fuse"
)
type forgetRecorder struct {
mu sync.Mutex
inodes []uint64
}
func (r *forgetRecorder) forget(inode uint64) {
r.mu.Lock()
defer r.mu.Unlock()
r.inodes = append(r.inodes, inode)
}
func (r *forgetRecorder) forgotten() []uint64 {
r.mu.Lock()
defer r.mu.Unlock()
return append([]uint64(nil), r.inodes...)
}
func TestPathCacheLookupAndExpiry(t *testing.T) {
rec := &forgetRecorder{}
c := newPathCache(20*time.Millisecond, rec.forget)
c.insert("a/b", 7, fuse.Attr{Ino: 7, Size: 42})
inode, attr, ok := c.lookup("a/b")
if !ok || inode != 7 || attr.Size != 42 {
t.Fatalf("lookup = %d %v %v, want 7 size 42 true", inode, attr.Size, ok)
}
time.Sleep(30 * time.Millisecond)
if _, _, ok := c.lookup("a/b"); ok {
t.Fatal("expired entry still served")
}
// The reference survives at least one sweep past expiry before it is
// forgotten: one insert moves it to the graveyard, the next returns it.
c.insert("x", 8, fuse.Attr{})
time.Sleep(30 * time.Millisecond)
c.insert("y", 9, fuse.Attr{})
found := false
for _, inode := range rec.forgotten() {
if inode == 7 {
found = true
}
}
if !found {
t.Fatalf("expired reference never forgotten; forgot %v", rec.forgotten())
}
}
func TestPathCacheStealTransfersOwnership(t *testing.T) {
rec := &forgetRecorder{}
c := newPathCache(20*time.Millisecond, rec.forget)
c.insert("a", 5, fuse.Attr{})
inode, ok := c.steal("a")
if !ok || inode != 5 {
t.Fatalf("steal = %d %v, want 5 true", inode, ok)
}
if _, ok := c.steal("a"); ok {
t.Fatal("second steal succeeded")
}
// Drive several sweeps; the stolen reference must never be forgotten.
for i := 0; i < 4; i++ {
time.Sleep(25 * time.Millisecond)
c.insert("churn", uint64(100+i), fuse.Attr{})
}
for _, inode := range rec.forgotten() {
if inode == 5 {
t.Fatal("stolen reference was forgotten by the cache")
}
}
}
func TestPathCacheReplaceReturnsOldReference(t *testing.T) {
rec := &forgetRecorder{}
c := newPathCache(20*time.Millisecond, rec.forget)
c.insert("a", 5, fuse.Attr{})
c.insert("a", 6, fuse.Attr{})
if inode, _, ok := c.lookup("a"); !ok || inode != 6 {
t.Fatalf("lookup after replace = %d %v, want 6 true", inode, ok)
}
for i := 0; i < 3; i++ {
time.Sleep(25 * time.Millisecond)
c.insert("churn", uint64(100+i), fuse.Attr{})
}
found := false
for _, inode := range rec.forgotten() {
if inode == 5 {
found = true
}
if inode == 6 && !found {
t.Fatal("live reference forgotten before the replaced one")
}
}
if !found {
t.Fatalf("replaced reference never forgotten; forgot %v", rec.forgotten())
}
}
func TestPathCachePurgePrefix(t *testing.T) {
rec := &forgetRecorder{}
c := newPathCache(time.Minute, rec.forget)
c.insert("dir", 2, fuse.Attr{})
c.insert("dir/a", 3, fuse.Attr{})
c.insert("dir/a/b", 4, fuse.Attr{})
c.insert("dirt", 5, fuse.Attr{})
c.purge("dir", true)
if _, _, ok := c.lookup("dir"); ok {
t.Fatal("purged key still cached")
}
if _, _, ok := c.lookup("dir/a/b"); ok {
t.Fatal("purged subtree still cached")
}
// A sibling that only shares the name as a string prefix stays.
if _, _, ok := c.lookup("dirt"); !ok {
t.Fatal("sibling was purged with the subtree")
}
}
func TestPathCacheRootNeverCached(t *testing.T) {
rec := &forgetRecorder{}
c := newPathCache(time.Minute, rec.forget)
c.insert("", 9, fuse.Attr{})
if _, _, ok := c.lookup(""); ok {
t.Fatal("root was cached")
}
if got := rec.forgotten(); len(got) != 1 || got[0] != 9 {
t.Fatalf("root reference not returned immediately: %v", got)
}
}
func TestCacheKey(t *testing.T) {
for path, want := range map[string]string{
`/a/b`: "a/b",
`\a\b`: "a/b",
`a//b/`: "a/b",
`/`: "",
``: "",
`/a/./b`: "a/b",
} {
if got := cacheKey(path); got != want {
t.Errorf("cacheKey(%q) = %q, want %q", path, got, want)
}
}
}
weed/mount/winfsp/xattr_windows.go
+4 -8
View File
@@ -12,11 +12,10 @@ import (
const xattrBufferSize = 64 * 1024
func (w *WinFS) Getxattr(path string, name string) (int, []byte) {
inode, ref, status := w.resolve(path)
inode, status := w.resolve(path)
if status != fuse.OK {
return toErrno(status), nil
}
defer ref.release()
dest := make([]byte, xattrBufferSize)
size, status := w.wfs.GetXAttr(never, ptr(w.caller(inode)), name, dest)
@@ -30,11 +29,10 @@ func (w *WinFS) Setxattr(path string, name string, value []byte, flags int) int
if w.denied() {
return -eROFS
}
inode, ref, status := w.resolve(path)
inode, status := w.resolve(path)
if status != fuse.OK {
return toErrno(status)
}
defer ref.release()
// cgofuse and the raw filesystem number XATTR_CREATE and XATTR_REPLACE the
// same, so the flags pass straight through; TestXattrFlagValues pins that.
@@ -48,20 +46,18 @@ func (w *WinFS) Removexattr(path string, name string) int {
if w.denied() {
return -eROFS
}
inode, ref, status := w.resolve(path)
inode, status := w.resolve(path)
if status != fuse.OK {
return toErrno(status)
}
defer ref.release()
return toErrno(w.wfs.RemoveXAttr(never, ptr(w.caller(inode)), name))
}
func (w *WinFS) Listxattr(path string, fill func(name string) bool) int {
inode, ref, status := w.resolve(path)
inode, status := w.resolve(path)
if status != fuse.OK {
return toErrno(status)
}
defer ref.release()
dest := make([]byte, xattrBufferSize)
size, status := w.wfs.ListXAttr(never, ptr(w.caller(inode)), dest)