Skip to content

Commit f082cee

Browse files
ankitmaster08ankitmaster08
committed
fix(msfs): address review findings in the S3 write path
Four defects found in review of the write path, each with a regression test where the failure is observable. DoSetAttr built its reply from backend.uid/gid, but backend is deliberately nil when backendNonce is 0. Only the size branch rejected that case, so a mode-only or mtime-only request on the mount root faulted. With no deferred unlock the fault left globals held, so a chmod on the mount point wedged every later FUSE callback; the added test hangs rather than fails without the fix. uid/gid now come from an inodeType switch, matching DoGetAttr. writeFile assigned the eTag-recovery HeadObject into the named return, so a failed HEAD reported a committed multipart upload as failed. The inode stayed dirty with a stale eTag, the object re-uploaded, and fsync returned EIO for a write that had succeeded. UploadPartCopy escaped x-amz-copy-source with url.QueryEscape, which emits "+" for a space where S3 expects %20, and escapes the bucket/key separator. AIStore and GCS accepted readonly: false while their writeFile returns "not implemented", so writes were buffered and failed at flush, after close(2) had already returned success under the flush_on_close: false default. Both now reject a writable config at mount, as PSEUDO already did. Also release the deferred-write budget when an inode disappears mid-commit. Only that branch leaks: prepareSmallWriteCommitLocked copies the counter to the control without clearing it, so every branch that leaves a live inode still has an owner for those bytes. Left unreleased they accumulate until shouldPromoteDeferredLocked always trips and every new object goes multipart, silently disabling the single-PUT path for small files. globals_lock.go is regenerated: the new test lock sites need registering, and the embedded line numbers shift.
1 parent 0a47c4f commit f082cee

7 files changed

Lines changed: 296 additions & 147 deletions

File tree

‎multi-storage-file-system/backend_s3.go‎

Lines changed: 11 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -521,7 +521,6 @@ func (s3Context *s3ContextStruct) writeFile(writeFileInput *writeFileInputStruct
521521
multipartThreshold = backend.multiPartCacheLineThreshold * globals.config.cacheLineSize
522522
s3PutObjectInput *s3.PutObjectInput
523523
s3PutObjectOutput *s3.PutObjectOutput
524-
s3HeadObjectOutput *s3.HeadObjectOutput
525524
)
526525

527526
if writeFileInput.forceSinglePut || multipartThreshold == 0 || writeFileInput.size <= multipartThreshold || writeFileInput.readerAt == nil {
@@ -556,11 +555,13 @@ func (s3Context *s3ContextStruct) writeFile(writeFileInput *writeFileInputStruct
556555
}
557556

558557
if writeFileOutput.eTag == "" {
559-
s3HeadObjectOutput, err = s3Context.s3Client.HeadObject(context.Background(), &s3.HeadObjectInput{
558+
// The upload already committed. This HEAD only recovers the eTag, so its
559+
// failure must not surface as a failed write.
560+
s3HeadObjectOutput, headErr := s3Context.s3Client.HeadObject(context.Background(), &s3.HeadObjectInput{
560561
Bucket: aws.String(backend.bucketContainerName),
561562
Key: aws.String(fullFilePath),
562563
})
563-
if err == nil && s3HeadObjectOutput.ETag != nil {
564+
if headErr == nil && s3HeadObjectOutput.ETag != nil {
564565
writeFileOutput.eTag = trimS3ETag(*s3HeadObjectOutput.ETag)
565566
}
566567
}
@@ -751,11 +752,13 @@ func (s3Context *s3ContextStruct) writeFileOverlay(inode *inodeStruct) (writeFil
751752

752753
if !inode.writeState.truncateAtOpen && offset+length <= inode.sizeInBackend && !inode.writeState.hasDirtyOverlap(offset, length) {
753754
copyOutput, copyErr := s3Context.s3Client.UploadPartCopy(context.Background(), &s3.UploadPartCopyInput{
754-
Bucket: aws.String(backend.bucketContainerName),
755-
Key: aws.String(fullFilePath),
756-
UploadId: aws.String(uploadID),
757-
PartNumber: aws.Int32(partNumber),
758-
CopySource: aws.String(url.QueryEscape(backend.bucketContainerName + "/" + fullFilePath)),
755+
Bucket: aws.String(backend.bucketContainerName),
756+
Key: aws.String(fullFilePath),
757+
UploadId: aws.String(uploadID),
758+
PartNumber: aws.Int32(partNumber),
759+
// QueryEscape would emit "+" for a space, which S3 percent-decodes
760+
// back to "+", and would escape the bucket/key separator.
761+
CopySource: aws.String((&url.URL{Path: backend.bucketContainerName + "/" + fullFilePath}).EscapedPath()),
759762
CopySourceRange: aws.String(fmt.Sprintf("bytes=%d-%d", offset, offset+length-1)),
760763
})
761764
if copyErr != nil {

‎multi-storage-file-system/config.go‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1357,6 +1357,11 @@ func checkConfigFile() (err error) {
13571357

13581358
switch backendAsStructNew.backendType {
13591359
case "AIStore":
1360+
if !backendAsStructNew.readOnly {
1361+
err = fmt.Errorf("backends[%v (\"%s\")] specified as backend_type \"AIStore\" must be readonly", backendsAsInterfaceSliceIndex, backendAsStructNew.dirName)
1362+
return
1363+
}
1364+
13601365
backendConfigAIStoreAsInterface, ok = backendAsMap["AIStore"]
13611366
if ok {
13621367
backendConfigAIStoreAsMap, ok = backendConfigAIStoreAsInterface.(map[string]interface{})
@@ -1421,6 +1426,11 @@ func checkConfigFile() (err error) {
14211426

14221427
backendAsStructNew.backendTypeSpecifics = backendConfigAIStoreAsStruct
14231428
case "GCS":
1429+
if !backendAsStructNew.readOnly {
1430+
err = fmt.Errorf("backends[%v (\"%s\")] specified as backend_type \"GCS\" must be readonly", backendsAsInterfaceSliceIndex, backendAsStructNew.dirName)
1431+
return
1432+
}
1433+
14241434
backendConfigGCSAsInterface, ok = backendAsMap["GCS"]
14251435
if ok {
14261436
backendConfigGCSAsMap, ok = backendConfigGCSAsInterface.(map[string]interface{})

‎multi-storage-file-system/fission.go‎

Lines changed: 50 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -415,14 +415,16 @@ func (*globalsStruct) DoSetAttr(inHeader *fission.InHeader, setAttrIn *fission.S
415415
attrValidSec uint64
416416
backend *backendStruct
417417
err error
418+
gid uint32
418419
mTimeNSec uint32
419420
mTimeSec uint64
420421
ok bool
421422
thisInode *inodeStruct
422423
timeNow = time.Now()
424+
uid uint32
423425
)
424426

425-
globalsLock("fission.go:425:2:(*globalsStruct).DoSetAttr")
427+
globalsLock("fission.go:427:2:(*globalsStruct).DoSetAttr")
426428

427429
thisInode, ok = globals.inodeMap.get(inHeader.NodeID)
428430
if !ok {
@@ -496,6 +498,21 @@ func (*globalsStruct) DoSetAttr(inHeader *fission.InHeader, setAttrIn *fission.S
496498

497499
thisInode.touch(nil)
498500
thisInode.updateParentDirEntryLocked()
501+
502+
// The mount root has no backend, and a request that does not set size never
503+
// reaches the check above that rejects a nil one.
504+
switch thisInode.inodeType {
505+
case FUSERootDir:
506+
uid = uint32(globals.config.uid)
507+
gid = uint32(globals.config.gid)
508+
case FileObject, BackendRootDir, PseudoDir:
509+
uid = uint32(backend.uid)
510+
gid = uint32(backend.gid)
511+
default:
512+
dumpStack()
513+
globals.logger.Fatalf("[FATAL] unrecognized inodeType (%v)", thisInode.inodeType)
514+
}
515+
499516
attrValidSec, attrValidNSec = timeDurationToAttrDuration(globals.config.entryAttrTTL)
500517
mTimeSec, mTimeNSec = timeTimeToAttrTime(thisInode.mTime)
501518
setAttrOut = &fission.SetAttrOut{
@@ -512,8 +529,8 @@ func (*globalsStruct) DoSetAttr(inHeader *fission.InHeader, setAttrIn *fission.S
512529
MTimeNSec: mTimeNSec,
513530
CTimeNSec: mTimeNSec,
514531
Mode: thisInode.mode,
515-
UID: uint32(backend.uid),
516-
GID: uint32(backend.gid),
532+
UID: uid,
533+
GID: gid,
517534
RDev: 0,
518535
Padding: 0,
519536
},
@@ -563,7 +580,7 @@ func (*globalsStruct) DoMkDir(inHeader *fission.InHeader, mkDirIn *fission.MkDir
563580

564581
defer func() {
565582
latency = time.Since(startTime).Seconds()
566-
globalsLock("fission.go:566:3:funcLit@564")
583+
globalsLock("fission.go:583:3:funcLit@581")
567584
if errno == 0 {
568585
globals.fissionMetrics.MkDirSuccesses.Inc()
569586
globals.fissionMetrics.MkDirSuccessLatencies.Observe(latency)
@@ -582,7 +599,7 @@ func (*globalsStruct) DoMkDir(inHeader *fission.InHeader, mkDirIn *fission.MkDir
582599
globalsUnlock()
583600
}()
584601

585-
globalsLock("fission.go:585:2:(*globalsStruct).DoMkDir")
602+
globalsLock("fission.go:602:2:(*globalsStruct).DoMkDir")
586603

587604
parentInode, ok = globals.inodeMap.get(inHeader.NodeID)
588605
if !ok {
@@ -687,7 +704,7 @@ func (*globalsStruct) DoUnlink(inHeader *fission.InHeader, unlinkIn *fission.Unl
687704
// Record metrics on function exit
688705
defer func() {
689706
latency = time.Since(startTime).Seconds()
690-
globalsLock("fission.go:690:3:funcLit@688")
707+
globalsLock("fission.go:707:3:funcLit@705")
691708
if errno == 0 {
692709
globals.fissionMetrics.UnlinkSuccesses.Inc()
693710
globals.fissionMetrics.UnlinkSuccessLatencies.Observe(latency)
@@ -706,7 +723,7 @@ func (*globalsStruct) DoUnlink(inHeader *fission.InHeader, unlinkIn *fission.Unl
706723
globalsUnlock()
707724
}()
708725

709-
globalsLock("fission.go:709:2:(*globalsStruct).DoUnlink")
726+
globalsLock("fission.go:726:2:(*globalsStruct).DoUnlink")
710727

711728
parentInode, ok = globals.inodeMap.get(inHeader.NodeID)
712729
if !ok {
@@ -809,7 +826,7 @@ func (*globalsStruct) DoRmDir(inHeader *fission.InHeader, rmDirIn *fission.RmDir
809826

810827
defer func() {
811828
latency = time.Since(startTime).Seconds()
812-
globalsLock("fission.go:812:3:funcLit@810")
829+
globalsLock("fission.go:829:3:funcLit@827")
813830
if errno == 0 {
814831
globals.fissionMetrics.RmDirSuccesses.Inc()
815832
globals.fissionMetrics.RmDirSuccessLatencies.Observe(latency)
@@ -828,7 +845,7 @@ func (*globalsStruct) DoRmDir(inHeader *fission.InHeader, rmDirIn *fission.RmDir
828845
globalsUnlock()
829846
}()
830847

831-
globalsLock("fission.go:831:2:(*globalsStruct).DoRmDir")
848+
globalsLock("fission.go:848:2:(*globalsStruct).DoRmDir")
832849

833850
parentInode, ok = globals.inodeMap.get(inHeader.NodeID)
834851
if !ok {
@@ -976,7 +993,7 @@ func (*globalsStruct) DoOpen(inHeader *fission.InHeader, openIn *fission.OpenIn)
976993

977994
defer func() {
978995
latency = time.Since(startTime).Seconds()
979-
globalsLock("fission.go:979:3:funcLit@977")
996+
globalsLock("fission.go:996:3:funcLit@994")
980997
if errno == 0 {
981998
globals.fissionMetrics.OpenSuccesses.Inc()
982999
globals.fissionMetrics.OpenSuccessLatencies.Observe(latency)
@@ -995,7 +1012,7 @@ func (*globalsStruct) DoOpen(inHeader *fission.InHeader, openIn *fission.OpenIn)
9951012
globalsUnlock()
9961013
}()
9971014

998-
globalsLock("fission.go:998:2:(*globalsStruct).DoOpen")
1015+
globalsLock("fission.go:1015:2:(*globalsStruct).DoOpen")
9991016

10001017
inode, ok = globals.inodeMap.get(inHeader.NodeID)
10011018
if !ok {
@@ -1203,7 +1220,7 @@ func (*globalsStruct) DoRead(inHeader *fission.InHeader, readIn *fission.ReadIn)
12031220
}
12041221

12051222
for len(readOut.Data) < cap(readOut.Data) {
1206-
globalsLock("fission.go:1206:3:(*globalsStruct).DoRead")
1223+
globalsLock("fission.go:1223:3:(*globalsStruct).DoRead")
12071224

12081225
inode, ok = globals.inodeMap.get(inHeader.NodeID)
12091226
if !ok {
@@ -1301,7 +1318,7 @@ func (*globalsStruct) DoRead(inHeader *fission.InHeader, readIn *fission.ReadIn)
13011318

13021319
dataCacheLineNumbers, _ = allocateDataCacheLines(1 + uint64(len(prefetchCacheLineNumbers)))
13031320

1304-
globalsLock("fission.go:1304:4:(*globalsStruct).DoRead")
1321+
globalsLock("fission.go:1321:4:(*globalsStruct).DoRead")
13051322

13061323
inode, ok = globals.inodeMap.get(inHeader.NodeID)
13071324
if !ok {
@@ -1565,7 +1582,7 @@ func (*globalsStruct) DoWrite(inHeader *fission.InHeader, writeIn *fission.Write
15651582
ok bool
15661583
)
15671584

1568-
globalsLock("fission.go:1568:2:(*globalsStruct).DoWrite")
1585+
globalsLock("fission.go:1585:2:(*globalsStruct).DoWrite")
15691586

15701587
inode, ok = globals.inodeMap.get(inHeader.NodeID)
15711588
if !ok {
@@ -1633,7 +1650,7 @@ func (*globalsStruct) DoWrite(inHeader *fission.InHeader, writeIn *fission.Write
16331650

16341651
// `DoStatFS` implements the package fission callback to fetch statistics about this FUSE file system.
16351652
func (*globalsStruct) DoStatFS(inHeader *fission.InHeader) (statFSOut *fission.StatFSOut, errno syscall.Errno) {
1636-
globalsLock("fission.go:1636:2:(*globalsStruct).DoStatFS")
1653+
globalsLock("fission.go:1653:2:(*globalsStruct).DoStatFS")
16371654

16381655
statFSOut = &fission.StatFSOut{
16391656
KStatFS: fission.KStatFS{
@@ -1673,7 +1690,7 @@ func (*globalsStruct) DoRelease(inHeader *fission.InHeader, releaseIn *fission.R
16731690

16741691
defer func() {
16751692
latency = time.Since(startTime).Seconds()
1676-
globalsLock("fission.go:1676:3:funcLit@1674")
1693+
globalsLock("fission.go:1693:3:funcLit@1691")
16771694
if errno == 0 {
16781695
globals.fissionMetrics.ReleaseSuccesses.Inc()
16791696
globals.fissionMetrics.ReleaseSuccessLatencies.Observe(latency)
@@ -1692,7 +1709,7 @@ func (*globalsStruct) DoRelease(inHeader *fission.InHeader, releaseIn *fission.R
16921709
globalsUnlock()
16931710
}()
16941711

1695-
globalsLock("fission.go:1695:2:(*globalsStruct).DoRelease")
1712+
globalsLock("fission.go:1712:2:(*globalsStruct).DoRelease")
16961713

16971714
inode, ok = globals.inodeMap.get(inHeader.NodeID)
16981715
if !ok {
@@ -1805,7 +1822,7 @@ func (*globalsStruct) DoFSync(inHeader *fission.InHeader, fSyncIn *fission.FSync
18051822
ok bool
18061823
)
18071824

1808-
globalsLock("fission.go:1808:2:(*globalsStruct).DoFSync")
1825+
globalsLock("fission.go:1825:2:(*globalsStruct).DoFSync")
18091826
inode, ok = globals.inodeMap.get(inHeader.NodeID)
18101827
if !ok {
18111828
globalsUnlock()
@@ -1888,7 +1905,7 @@ func (*globalsStruct) DoFlush(inHeader *fission.InHeader, flushIn *fission.Flush
18881905
ok bool
18891906
)
18901907

1891-
globalsLock("fission.go:1891:2:(*globalsStruct).DoFlush")
1908+
globalsLock("fission.go:1908:2:(*globalsStruct).DoFlush")
18921909
inode, ok = globals.inodeMap.get(inHeader.NodeID)
18931910
if !ok {
18941911
globalsUnlock()
@@ -1942,7 +1959,7 @@ func (*globalsStruct) DoOpenDir(inHeader *fission.InHeader, openDirIn *fission.O
19421959

19431960
defer func() {
19441961
latency = time.Since(startTime).Seconds()
1945-
globalsLock("fission.go:1945:3:funcLit@1943")
1962+
globalsLock("fission.go:1962:3:funcLit@1960")
19461963
if errno == 0 {
19471964
globals.fissionMetrics.OpenDirSuccesses.Inc()
19481965
globals.fissionMetrics.OpenDirSuccessLatencies.Observe(latency)
@@ -1961,7 +1978,7 @@ func (*globalsStruct) DoOpenDir(inHeader *fission.InHeader, openDirIn *fission.O
19611978
globalsUnlock()
19621979
}()
19631980

1964-
globalsLock("fission.go:1964:2:(*globalsStruct).DoOpenDir")
1981+
globalsLock("fission.go:1981:2:(*globalsStruct).DoOpenDir")
19651982

19661983
inode, ok = globals.inodeMap.get(inHeader.NodeID)
19671984
if !ok {
@@ -2102,7 +2119,7 @@ func (*globalsStruct) DoReadDir(inHeader *fission.InHeader, readDirIn *fission.R
21022119
}
21032120

21042121
latency = time.Since(startTime).Seconds()
2105-
globalsLock("fission.go:2105:3:funcLit@2098")
2122+
globalsLock("fission.go:2122:3:funcLit@2115")
21062123
if errno == 0 {
21072124
globals.fissionMetrics.ReadDirSuccesses.Inc()
21082125
globals.fissionMetrics.ReadDirSuccessLatencies.Observe(latency)
@@ -2140,7 +2157,7 @@ func (*globalsStruct) DoReadDir(inHeader *fission.InHeader, readDirIn *fission.R
21402157
curReadDirOutSize = 0
21412158
curOffset = readDirIn.Offset
21422159

2143-
globalsLock("fission.go:2143:2:(*globalsStruct).DoReadDir")
2160+
globalsLock("fission.go:2160:2:(*globalsStruct).DoReadDir")
21442161

21452162
Restart:
21462163

@@ -2263,7 +2280,7 @@ Restart:
22632280

22642281
listDirectoryOutput, err = listDirectoryWrapper(backend.context, listDirectoryInput)
22652282

2266-
globalsLock("fission.go:2266:4:(*globalsStruct).DoReadDir")
2283+
globalsLock("fission.go:2283:4:(*globalsStruct).DoReadDir")
22672284

22682285
fh.listDirectoryInProgress = false
22692286

@@ -2379,7 +2396,7 @@ func (*globalsStruct) DoReleaseDir(inHeader *fission.InHeader, releaseDirIn *fis
23792396

23802397
defer func() {
23812398
latency = time.Since(startTime).Seconds()
2382-
globalsLock("fission.go:2382:3:funcLit@2380")
2399+
globalsLock("fission.go:2399:3:funcLit@2397")
23832400
if errno == 0 {
23842401
globals.fissionMetrics.ReleaseDirSuccesses.Inc()
23852402
globals.fissionMetrics.ReleaseDirSuccessLatencies.Observe(latency)
@@ -2398,7 +2415,7 @@ func (*globalsStruct) DoReleaseDir(inHeader *fission.InHeader, releaseDirIn *fis
23982415
globalsUnlock()
23992416
}()
24002417

2401-
globalsLock("fission.go:2401:2:(*globalsStruct).DoReleaseDir")
2418+
globalsLock("fission.go:2418:2:(*globalsStruct).DoReleaseDir")
24022419

24032420
inode, ok = globals.inodeMap.get(inHeader.NodeID)
24042421
if !ok {
@@ -2511,7 +2528,7 @@ func (*globalsStruct) DoCreate(inHeader *fission.InHeader, createIn *fission.Cre
25112528

25122529
defer func() {
25132530
latency = time.Since(startTime).Seconds()
2514-
globalsLock("fission.go:2514:3:funcLit@2512")
2531+
globalsLock("fission.go:2531:3:funcLit@2529")
25152532
if errno == 0 {
25162533
globals.fissionMetrics.CreateSuccesses.Inc()
25172534
globals.fissionMetrics.CreateSuccessLatencies.Observe(latency)
@@ -2530,7 +2547,7 @@ func (*globalsStruct) DoCreate(inHeader *fission.InHeader, createIn *fission.Cre
25302547
globalsUnlock()
25312548
}()
25322549

2533-
globalsLock("fission.go:2533:2:(*globalsStruct).DoCreate")
2550+
globalsLock("fission.go:2550:2:(*globalsStruct).DoCreate")
25342551

25352552
parentInode, ok = globals.inodeMap.get(inHeader.NodeID)
25362553
if !ok {
@@ -2787,7 +2804,7 @@ func (*globalsStruct) DoReadDirPlus(inHeader *fission.InHeader, readDirPlusIn *f
27872804
}
27882805

27892806
latency = time.Since(startTime).Seconds()
2790-
globalsLock("fission.go:2790:3:funcLit@2783")
2807+
globalsLock("fission.go:2807:3:funcLit@2800")
27912808
if errno == 0 {
27922809
globals.fissionMetrics.ReadDirPlusSuccesses.Inc()
27932810
globals.fissionMetrics.ReadDirPlusSuccessLatencies.Observe(latency)
@@ -2827,7 +2844,7 @@ func (*globalsStruct) DoReadDirPlus(inHeader *fission.InHeader, readDirPlusIn *f
28272844

28282845
entryAttrValidSec, entryAttrValidNSec = timeDurationToAttrDuration(globals.config.entryAttrTTL)
28292846

2830-
globalsLock("fission.go:2830:2:(*globalsStruct).DoReadDirPlus")
2847+
globalsLock("fission.go:2847:2:(*globalsStruct).DoReadDirPlus")
28312848

28322849
Restart:
28332850

@@ -3111,7 +3128,7 @@ Restart:
31113128

31123129
listDirectoryOutput, err = listDirectoryWrapper(backend.context, listDirectoryInput)
31133130

3114-
globalsLock("fission.go:3114:4:(*globalsStruct).DoReadDirPlus")
3131+
globalsLock("fission.go:3131:4:(*globalsStruct).DoReadDirPlus")
31153132

31163133
fh.listDirectoryInProgress = false
31173134

@@ -3248,7 +3265,7 @@ func (*globalsStruct) DoStatX(inHeader *fission.InHeader, statXIn *fission.StatX
32483265

32493266
defer func() {
32503267
latency = time.Since(startTime).Seconds()
3251-
globalsLock("fission.go:3251:3:funcLit@3249")
3268+
globalsLock("fission.go:3268:3:funcLit@3266")
32523269
if errno == 0 {
32533270
globals.fissionMetrics.StatXSuccesses.Inc()
32543271
globals.fissionMetrics.StatXSuccessLatencies.Observe(latency)
@@ -3267,7 +3284,7 @@ func (*globalsStruct) DoStatX(inHeader *fission.InHeader, statXIn *fission.StatX
32673284
globalsUnlock()
32683285
}()
32693286

3270-
globalsLock("fission.go:3270:2:(*globalsStruct).DoStatX")
3287+
globalsLock("fission.go:3287:2:(*globalsStruct).DoStatX")
32713288

32723289
thisInode, ok = globals.inodeMap.get(inHeader.NodeID)
32733290
if !ok {

0 commit comments

Comments
 (0)