Repository navigation
MM trigger tests - #113
MM trigger tests#113kpjensen wants to merge 110 commits into
Conversation
There was a problem hiding this comment.
Pull request overview
This PR introduces a comprehensive memory management system for dxfuse to prevent resource starvation and improve concurrency handling. The changes add a centralized MemoryManager to coordinate memory allocation across read and write operations, along with enhanced prefetch logic and upload worker scaling based on available CPU resources.
Changes:
- Implemented a
MemoryManagerwith separate budgets for read/write operations to control memory usage - Refactored prefetch and upload systems to use the memory manager for buffer allocation
- Added CPU-aware scaling for HTTP client pools and upload workers
Reviewed changes
Copilot reviewed 16 out of 17 changed files in this pull request and generated 6 comments.
Show a summary per file
| File | Description |
|---|---|
| memory_manager.go | New memory manager with semaphore-based budgeting for read/write operations |
| memory_manager_test.go | Comprehensive test coverage for memory manager allocation/release scenarios |
| util.go | Added CPU detection and worker count calculation utilities |
| util_test.go | Tests for HTTP client pool and upload worker count bounds |
| prefetch.go | Integrated memory manager for prefetch buffer allocation and enhanced sequential detection |
| prefetch_test.go | Tests for prefetch heuristics, cache behavior, and sequential access detection |
| upload.go | Updated uploader to use memory manager and dynamic part sizing |
| dxfuse.go | Initialized memory manager with system memory limits and updated component constructors |
| dx_ops.go | Enhanced error logging for upload failures |
| cli/main.go | Added -maxMemoryUsageMiB flag for memory limit configuration |
| test_parallel_rw.sh | New test script for parallel read/write operations |
| test/local/multiple_mounts.sh | Enhanced error handling and log display on test failure |
| .vscode/launch.json, .vscode/tasks.json | VSCode debug configuration for development |
| .github/workflows/*.yml | Updated GitHub Actions to use setup-go@v5 |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| //if pgs.verboseLevel >= 2 { | ||
| pfm.log("isDataInCache: wait") | ||
| //} |
There was a problem hiding this comment.
Remove commented-out code. If the conditional logging is unnecessary, delete the entire block; otherwise, restore the condition.
| // data := pgs.allocateIOvecMemory(ioSize) | ||
| // if data == nil { | ||
| // pgs.log("Memory limit exceeded, dropping IOvec allocation") | ||
| // return nil | ||
| // } |
There was a problem hiding this comment.
Remove commented-out dead code related to memory allocation that is no longer in use.
| // data := pgs.allocateIOvecMemory(ioSize) | |
| // if data == nil { | |
| // pgs.log("Memory limit exceeded, dropping IOvec allocation") | |
| // return nil | |
| // } |
| iov := pfm.cache.iovecs[i] | ||
| switch iov.state { | ||
| case IOV_HOLE: | ||
| pfm.log("isDataInCache: HOLE") |
There was a problem hiding this comment.
These debug log statements at lines 1212, 1220, and 1226 should be guarded by verbosity level checks (use pgs.debug() or check pgs.verboseLevel) to avoid excessive logging in production.
| fsys.opClose(oph) | ||
|
|
||
| fsys.pgs = NewPrefetchGlobalState(options.VerboseLevel, dxEnv) | ||
| // Default to max 15% memory usage of system memory |
There was a problem hiding this comment.
The comment at line 202 says '15%' but the code calculates 10%. Update either the comment or the calculation to match.
| // Default to max 15% memory usage of system memory | |
| // Default to max 10% memory usage of system memory |
| umount /home/kjensen/MNT || true | ||
| # mount dxfuse | ||
| /home/kjensen/dxfuse --version | ||
| /home/kjensen/dxfuse -limitedWrite /home/kjensen/MNT testing | ||
| rm -rf /home/kjensen/MNT/testing/1GiB* | ||
| rm -rf /home/kjensen/MNT/testing/1kib* |
There was a problem hiding this comment.
Hard-coded paths like '/home/kjensen/MNT' and '/home/kjensen/dxfuse' should be parameterized or documented as example values, as they are specific to a development environment.
| umount /home/kjensen/MNT || true | |
| # mount dxfuse | |
| /home/kjensen/dxfuse --version | |
| /home/kjensen/dxfuse -limitedWrite /home/kjensen/MNT testing | |
| rm -rf /home/kjensen/MNT/testing/1GiB* | |
| rm -rf /home/kjensen/MNT/testing/1kib* | |
| # Allow overriding dxfuse binary and mount directory via environment | |
| DXFUSE_BIN="${DXFUSE_BIN:-/home/kjensen/dxfuse}" | |
| DXFUSE_MNT_DIR="${DXFUSE_MNT_DIR:-/home/kjensen/MNT}" | |
| umount "$DXFUSE_MNT_DIR" || true | |
| # mount dxfuse | |
| "$DXFUSE_BIN" --version | |
| "$DXFUSE_BIN" -limitedWrite "$DXFUSE_MNT_DIR" testing | |
| rm -rf "$DXFUSE_MNT_DIR"/testing/1GiB* | |
| rm -rf "$DXFUSE_MNT_DIR"/testing/1kib* |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 16 out of 17 changed files in this pull request and generated 9 comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 16 out of 17 changed files in this pull request and generated 6 comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| } | ||
|
|
||
| // Analyzes stack traces to find goroutines waiting for mutex locks | ||
| func FindWaitingGoroutines(stackTrace string, mutexName string) string { |
There was a problem hiding this comment.
The parameter mutexName is declared but never used in the function body. Either remove it if it's not needed, or implement filtering logic based on mutexName if that was the intended functionality.
| // data := pgs.allocateIOvecMemory(ioSize) | ||
| // if data == nil { | ||
| // pgs.log("Memory limit exceeded, dropping IOvec allocation") | ||
| // return nil | ||
| // } |
There was a problem hiding this comment.
Remove commented-out code. If this allocation logic is no longer needed, delete these lines. If it's intended for future use, document why it's preserved or track it in a TODO/issue.
| // data := pgs.allocateIOvecMemory(ioSize) | |
| // if data == nil { | |
| // pgs.log("Memory limit exceeded, dropping IOvec allocation") | |
| // return nil | |
| // } |
| fsys.opClose(oph) | ||
| fsys.mutex.Unlock() | ||
| fh.writeBuffer = fsys.uploader.AllocateWriteBuffer(partId, false) | ||
| fh.writeBuffer = fsys.uploader.AllocateWriteBuffer(partId, true) |
There was a problem hiding this comment.
The second parameter true is unclear without context. Consider using a named constant or adding a comment explaining what the boolean represents (appears to be a blocking flag based on the function signature).
| fh.writeBuffer = fsys.uploader.AllocateWriteBuffer(partId, true) | |
| fh.writeBuffer = fsys.uploader.AllocateWriteBuffer(partId, true /* blocking */) |
| } else { | ||
| pfm.log("(#io=%d) prefetch io error [%d -- %d] %s", | ||
| ioReq.id, ioReq.startByte, ioReq.endByte, err.Error()) | ||
| pfm.cache.iovecs[iovIdx].state = IOV_ERRORED | ||
| if errors.Is(err, errPrefetchNoMemory) { | ||
| // Memory pressure is transient; treat as a cache hole (best-effort prefetch). | ||
| pfm.cache.iovecs[iovIdx].state = IOV_HOLE | ||
| pgs.debug("prefetch skipped due to memory pressure: handle=%d startByte=%d endByte=%d", | ||
| ioReq.hid, ioReq.startByte, ioReq.endByte) | ||
| } else { | ||
| pfm.log("prefetch error: handle=%d, startByte=%d, endByte=%d, error=%s", ioReq.hid, ioReq.startByte, ioReq.endByte, err.Error()) | ||
| // Safety: if a buffer somehow comes back with an error, release it. | ||
| if data != nil { | ||
| pgs.memoryManager.ReleaseReadBuffer(data) | ||
| } | ||
| pfm.log("(#io=%d) prefetch io error [%d -- %d] %s", | ||
| ioReq.id, ioReq.startByte, ioReq.endByte, err.Error()) | ||
| pfm.cache.iovecs[iovIdx].state = IOV_ERRORED | ||
| pgs.log("IO request failed: handle=%d, startByte=%d, endByte=%d, error=%s", ioReq.hid, ioReq.startByte, ioReq.endByte, err.Error()) | ||
| } |
There was a problem hiding this comment.
The nested else-if structure creates unnecessary indentation. Flatten this by handling the errPrefetchNoMemory case first with early returns or by restructuring the conditional logic.
| } else { | |
| pfm.log("(#io=%d) prefetch io error [%d -- %d] %s", | |
| ioReq.id, ioReq.startByte, ioReq.endByte, err.Error()) | |
| pfm.cache.iovecs[iovIdx].state = IOV_ERRORED | |
| if errors.Is(err, errPrefetchNoMemory) { | |
| // Memory pressure is transient; treat as a cache hole (best-effort prefetch). | |
| pfm.cache.iovecs[iovIdx].state = IOV_HOLE | |
| pgs.debug("prefetch skipped due to memory pressure: handle=%d startByte=%d endByte=%d", | |
| ioReq.hid, ioReq.startByte, ioReq.endByte) | |
| } else { | |
| pfm.log("prefetch error: handle=%d, startByte=%d, endByte=%d, error=%s", ioReq.hid, ioReq.startByte, ioReq.endByte, err.Error()) | |
| // Safety: if a buffer somehow comes back with an error, release it. | |
| if data != nil { | |
| pgs.memoryManager.ReleaseReadBuffer(data) | |
| } | |
| pfm.log("(#io=%d) prefetch io error [%d -- %d] %s", | |
| ioReq.id, ioReq.startByte, ioReq.endByte, err.Error()) | |
| pfm.cache.iovecs[iovIdx].state = IOV_ERRORED | |
| pgs.log("IO request failed: handle=%d, startByte=%d, endByte=%d, error=%s", ioReq.hid, ioReq.startByte, ioReq.endByte, err.Error()) | |
| } | |
| } else if errors.Is(err, errPrefetchNoMemory) { | |
| // Memory pressure is transient; treat as a cache hole (best-effort prefetch). | |
| pfm.cache.iovecs[iovIdx].state = IOV_HOLE | |
| pgs.debug("prefetch skipped due to memory pressure: handle=%d startByte=%d endByte=%d", | |
| ioReq.hid, ioReq.startByte, ioReq.endByte) | |
| } else { | |
| pfm.log("prefetch error: handle=%d, startByte=%d, endByte=%d, error=%s", ioReq.hid, ioReq.startByte, ioReq.endByte, err.Error()) | |
| // Safety: if a buffer somehow comes back with an error, release it. | |
| if data != nil { | |
| pgs.memoryManager.ReleaseReadBuffer(data) | |
| } | |
| pfm.log("(#io=%d) prefetch io error [%d -- %d] %s", | |
| ioReq.id, ioReq.startByte, ioReq.endByte, err.Error()) | |
| pfm.cache.iovecs[iovIdx].state = IOV_ERRORED | |
| pgs.log("IO request failed: handle=%d, startByte=%d, endByte=%d, error=%s", ioReq.hid, ioReq.startByte, ioReq.endByte, err.Error()) |
| func (pgs *PrefetchGlobalState) moveCacheWindow(pfm *PrefetchFileMetadata, iovIndex int) { | ||
| nIovecs := len(pfm.cache.iovecs) | ||
| nReadAheadChunks := iovIndex + pfm.cache.maxNumIovecs - nIovecs | ||
| // log the above 2 values |
There was a problem hiding this comment.
The comment 'log the above 2 values' is vague and doesn't add value. Either remove it or make it descriptive about what values are being logged and why.
| // log the above 2 values | |
| // When verbose logging is enabled, record the cache window parameters | |
| // (iovIndex, nIovecs, nReadAheadChunks) to aid debugging of readahead behavior. |
| echo "unmounting dxfuse" | ||
| cd $HOME | ||
| fusermount -u $mountpoint | ||
| fusermount -u $mountpoint0 |
There was a problem hiding this comment.
Variable naming is inconsistent - the script defines mountpoint (line 7) but uses mountpoint0 (line 19) without defining it. This appears to be a bug introduced by the change from $mountpoint to $mountpoint0.
- Release empty replacement write buffer in FlushFile when last write lands exactly on a part boundary (budget leak) - Release leftover write buffer in ReleaseFileHandle - Handle VirtualMemory() failure with conservative 8 GiB fallback - Use writeBufferOffset instead of len(writeBuffer) for empty file check - Protect fh.writeError write with mutex in uploadWorker - Remove no-op buf=nil in MemoryManager.release() - Remove unused pprof import - Remove commented-out code in newIovec - Remove unused GetDumpableStackTraces/FindWaitingGoroutines - Remove noisy httpClient debug log in upload error path Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
No description provided.