diff --git a/.github/workflows/make-check.yml b/.github/workflows/make-check.yml index 5ef4a7662e..d77354ee7b 100644 --- a/.github/workflows/make-check.yml +++ b/.github/workflows/make-check.yml @@ -6,6 +6,10 @@ on: pull_request: branches: [ main ] +concurrency: + group: build-${{ github.ref }} + cancel-in-progress: ${{ github.event_name == 'pull_request' }} + jobs: build: @@ -20,8 +24,8 @@ jobs: - name: configure run: ./configure - name: make - run: make + run: make -j2 - name: make tests - run: make tests + run: make tests -j2 - name: make check run: make check AUTOTEST="--retry-once --jobs 8" diff --git a/.github/workflows/warmctl-release.yml b/.github/workflows/warmctl-release.yml new file mode 100644 index 0000000000..e8c8029f6b --- /dev/null +++ b/.github/workflows/warmctl-release.yml @@ -0,0 +1,128 @@ +name: warmctl release + +on: + pull_request: + branches: [main, warmctl] + push: + tags: ['v*-warmctl.*'] + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: warmctl-release-${{ github.ref }} + cancel-in-progress: ${{ github.event_name == 'pull_request' }} + +jobs: + linux: + strategy: + fail-fast: false + matrix: + include: + - runner: ubuntu-24.04 + arch: amd64 + - runner: ubuntu-24.04-arm + arch: arm64 + runs-on: ${{ matrix.runner }} + container: + image: debian:bookworm-slim@sha256:88200866dfff7ea7f5cbcb6ec7c8a701889efe6fe859fe64d6990e4b07ea4171 + # Restored process trees are adopted by PID 1; reap them in this test container. + options: --init + timeout-minutes: 30 + steps: + - name: Install native build dependencies + run: apt-get update && apt-get install -y --no-install-recommends build-essential python3 git ca-certificates gzip zstd + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + fetch-depth: 0 + persist-credentials: false + - name: Build + run: | + ./configure --prefix=/opt/warmctl/dmtcp --docdir=/opt/warmctl/dmtcp/share/doc/dmtcp + make -j2 + make tests -j2 + - name: Check memory readers and checkpoint bytes + run: python3 test/restore-memory-check.py --root . --output "$RUNNER_TEMP/memory-check" + - name: Existing process, thread and memory regressions + run: DMTCP_RESTART_HUGEPAGES=1 DMTCP_RESTART_READERS=16 python3 test/autotest.py --retry-once --jobs 2 file1 file3 mmap1 mmap-noreserve zeropages zeropages-pread pthread1 forkexec sysv-shm1 modify-env + - name: Package installed runtime and source + env: + TAG: ${{ startsWith(github.ref, 'refs/tags/') && github.ref_name || 'snapshot' }} + ARCH: ${{ matrix.arch }} + run: | + make install DESTDIR="$RUNNER_TEMP/stage" + make -C plugin/fdkeep install DESTDIR="$RUNNER_TEMP/stage" + install -m 0644 COPYING.LESSER "$RUNNER_TEMP/stage/opt/warmctl/dmtcp/share/doc/dmtcp/" + mkdir -p dist + printf 'release=%s\ncommit=%s\nupstream_version=%s\nos=debian-12\narch=%s\nprefix=/opt/warmctl/dmtcp\n' \ + "$TAG" "$GITHUB_SHA" "$(cat VERSION)" "$ARCH" > "$RUNNER_TEMP/stage/opt/warmctl/dmtcp/BUILDINFO" + ldd --version | head -1 >> "$RUNNER_TEMP/stage/opt/warmctl/dmtcp/BUILDINFO" + tar -czf "dist/dmtcp-$TAG-linux-$ARCH-debian12.tar.gz" -C "$RUNNER_TEMP/stage/opt/warmctl" dmtcp + if [ "$ARCH" = amd64 ]; then + git -c safe.directory="$GITHUB_WORKSPACE" archive --format=tar.gz --prefix="dmtcp-$TAG/" HEAD > "dist/dmtcp-$TAG-source.tar.gz" + fi + (cd dist && sha256sum *.tar.gz > "dmtcp-$TAG-linux-$ARCH-checksums.txt") + - name: Verify packaged runtime + run: | + mkdir -p "$RUNNER_TEMP/relocated" + tar -xzf dist/*-linux-${{ matrix.arch }}-debian12.tar.gz -C "$RUNNER_TEMP/relocated" + "$RUNNER_TEMP/relocated/dmtcp/bin/dmtcp_launch" --version + python3 test/restore-memory-check.py --root "$RUNNER_TEMP/relocated/dmtcp" \ + --plugin "$RUNNER_TEMP/relocated/dmtcp/lib/libdmtcp_fdkeep.so" --output "$RUNNER_TEMP/package-check" + - name: Archive test logs + if: always() + run: | + mkdir -p dist + python3 - <<'PY' + import os + from pathlib import Path + import tarfile + root = Path(os.environ['RUNNER_TEMP']) + with tarfile.open('dist/test-logs-${{ matrix.arch }}.tgz', 'w:gz') as archive: + for name in ('memory-check', 'package-check'): + for log in (root / name).rglob('*.log'): + archive.add(log, arcname=str(log.relative_to(root))) + PY + - uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 + if: always() + with: + name: dmtcp-linux-${{ matrix.arch }} + path: dist/ + if-no-files-found: error + retention-days: 14 + + publish: + if: startsWith(github.ref, 'refs/tags/') + needs: linux + runs-on: ubuntu-24.04 + permissions: + contents: write + id-token: write + attestations: write + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + - uses: actions/download-artifact@9000827ccba6bdab643e8b6fd33ac0654aef8333 # v8.0.2 + with: + pattern: dmtcp-linux-* + merge-multiple: true + path: artifacts + - name: Combined checksums + env: + TAG: ${{ github.ref_name }} + run: | + cd artifacts + rm *-linux-*-checksums.txt + sha256sum *.tar.gz > "dmtcp-$TAG-checksums.txt" + - uses: actions/attest-build-provenance@4d101475d8b20a2381f78447822ac1eab6504dd8 # v4.2.2 + with: + subject-path: artifacts/*.tar.gz + - name: Publish fork prerelease + env: + GH_TOKEN: ${{ github.token }} + TAG: ${{ github.ref_name }} + run: | + mapfile -t assets < <(find artifacts -type f \( -name '*.tar.gz' -o -name '*checksums.txt' \)) + gh release create "$TAG" "${assets[@]}" --verify-tag --prerelease --title "DMTCP $TAG" --notes-file doc/warmctl-release-notes.md diff --git a/RELEASING.md b/RELEASING.md new file mode 100644 index 0000000000..6d0297af88 --- /dev/null +++ b/RELEASING.md @@ -0,0 +1,46 @@ +# TensorRails DMTCP releases + +The fork has its own tags, `v5.0.0-warmctl.N`, independently of warmctl's +CLI versions. `VERSION` remains the upstream checkpoint ABI version (5.0); +the tag and full commit in the installed `BUILDINFO` identify the fork build. +Do not change the upstream checkpoint signature just to number a fork release. + +Merge a reviewed PR into `main`, with `Fixes #N` for its performance issue. +The PR workflow builds natively in pinned Debian 12 containers on x86-64 and ARM64, runs the memory-reader and +process regressions, packages an installed runtime, and tests that package. +After the checks pass, tag the merge commit and push the tag: + +```sh +git tag -a v5.0.0-warmctl.1 MERGE_COMMIT -m v5.0.0-warmctl.1 +git push origin v5.0.0-warmctl.1 +``` + +The workflow reruns the checks and publishes a prerelease with source, +Debian 12 x86-64 and ARM64 runtimes, SHA-256 checksums and GitHub build provenance. +Release notes come from `doc/warmctl-release-notes.md`. Before each later +release, update those notes in the PR. Keep the historical `warmctl` branch +fast-forwarded to the released commit for consumers still using that branch. + +The binary archive contains `dmtcp/`, to extract under `/opt/warmctl`. +It includes the separately licensed warmctl fdkeep plugin; cuInterpose is a +separate GPU dependency. fdkeep is built from plugin/fdkeep and tested on both +release architectures. The archive requires Debian 12-compatible glibc +(2.36 or newer) and libstdc++ on the matching architecture; it is not a +portable Linux binary. DMTCP runs on Linux, not macOS or Windows. Parallel +pread helpers currently support x86-64 only; ARM64 uses the ordinary reader +with optional huge pages and is CPU-tested, not GPU-qualified by this release. +The package tests also bake and restore from a relocated installation path. +warmctl bundles these released packages and uses the same assets in its runtime +Dockerfile, verifying source-pinned SHA-256 checksums. For an older runtime +base, build the tagged source and fdkeep inside that base instead. + +```sh +sha256sum --ignore-missing -c dmtcp-v5.0.0-warmctl.1-checksums.txt +gh attestation verify dmtcp-v5.0.0-warmctl.1-linux-amd64-debian12.tar.gz \ + --repo TensorRails/dmtcp +``` + +Donor and receiver must use byte-identical mapped DMTCP libraries and runtime +files. A new release requires a fresh bake and the same resulting runtime +image on receivers. Never overwrite `libdmtcp.so` underneath a live engine or +an old snapshot and assume compatibility. Preserve that snapshot's old image. diff --git a/doc/h200-restart-performance.md b/doc/h200-restart-performance.md new file mode 100644 index 0000000000..5f952d68c0 --- /dev/null +++ b/doc/h200-restart-performance.md @@ -0,0 +1,75 @@ +# H200 local restart measurements, 2026-11-01 + +Model: `nvidia/Qwen3.8-Flash-Next-NVFP4`, one H200. A local 135.43 GiB +snapshot, with file cache hot, on a Xeon Platinum 8558 host; cgroup CPU quota +20.4 CPUs and RAM limit 233.76 GiB. GPU weights occupy 76.50 GiB. This +measures restart through HTTP readiness and the first streaming token, not +pod provisioning, runtime installation or snapshot transfer. + +All modes used the same snapshot and H200. Earlier observations of 52.8 s +were on another host with a subsequently lost snapshot and are not an A/B +baseline for these measurements. + +| Mode | Runs to health (seconds) | Median | +| --- | --- | --- | +| Ordinary reader | 77.36, 82.14 | 79.75 | +| Huge pages, one reader | 35.95, 37.46 | 36.71 | +| Private mmap, unaligned image (fallback) | 79.24, 75.98 | 77.61 | +| Private mmap, aligned copy | 77.02, 52.83 | 64.93 | +| Ordinary reader, aligned copy | 83.12 | 83.12 | +| Huge pages, four readers | 34.27, 26.58 | 30.42 | +| Huge pages, eight readers | 38.73, 27.03 | 32.88 | +| Huge pages, eight readers after cache preparation | 23.45, 23.77 | 23.61 | +| Huge pages, sixteen readers after cache preparation | 20.21, 23.43, 20.27, 21.39 | **20.83** | +| One-reader huge-page control after cache preparation | 44.97 | 44.97 | + +Cache preparation evicted only unused experimental aligned-image and CPU +benchmark pages with `POSIX_FADV_DONTNEED`. It did not evict the primary +snapshot, model or host-wide page cache. `memory.current` fell from 163.61 +to 135.07 GiB. The four fast runs peaked at 213.08 GiB, with no additional +direct reclaim scans. This preparation and the small sample sizes matter: +these are experiments, not a p95 or a <=20 s guarantee. The final binary, +including helper parent-death protection, produced the last two fast runs. +Median streaming first token from start across the four: **20.92 s**. + +One complete before/after pair: + +| Phase | Ordinary | Huge pages + 16 readers | +| --- | ---: | ---: | +| Setup and filesystem | 3.09 s | 3.33 s | +| DMTCP | 54.09 s | 4.04 s | +| CUDA | 13.52 s | 6.73 s | +| Wake and reload weights | 5.62 s | 5.12 s | +| KV, health, canary | 0.99 s | 0.99 s | +| External health polling | 0.05 s | 0.07 s | +| Total | 77.36 s | 20.27 s | + +The slow worker's memory pass took 53.61 s, with 14,154,801 minor faults, +zero major faults and zero disk `read_bytes`. A single 47.6846 GiB anonymous +payload took 47.05 s. Its size matches the model's CPU PLE/Engram table, +allowing for allocation rounding; direct tensor-pointer attribution was +not performed. Other processes finished in 0.08–3.80 s and waited at the +barrier; adding their barrier waits would double-count elapsed time. + +THP reduced that worker's minor faults about 49-fold. Sixteen readers loaded +the dominant region in 1.27–1.33 s and the whole worker in 3.56–3.73 s. +Private mmap reached 2.68–2.84 s for DMTCP but 39.81–60.66 s for CUDA. +Observed anonymous RSS growth is consistent with COW during CUDA host-memory +registration, though the driver's internal function was not traced. + +## Correctness and limits + +All 20 GPU runs matched weight fingerprints, returned streaming READY and +a valid forced echo tool call. All 640 short smoke requests succeeded. +Numerical equivalence is not proven: identical donor requests *before* +checkpoint, seed 42 and temperature 0, already differed in token sequence +and by up to 1.15547 logprob on the five-token common prefix. Both reported +zero cached tokens. The canary tolerance stayed 0.001; variability before +checkpoint does not prove absence of restoration bugs. + +CPU tests verified every word of dense/sparse mappings, dirty-page writes, +second checkpoint/restore, source-image digests, uneven parallel chunks, +parent cancellation and private dirty bytes after file EOF. CI repeats +these checks against both the build and the packaged runtime. The fork +switches remain opt-in; CUDA mmap remains experimental and disabled in +the recommended configuration. diff --git a/doc/restart-memory-profiling.md b/doc/restart-memory-profiling.md new file mode 100644 index 0000000000..e0a71887ce --- /dev/null +++ b/doc/restart-memory-profiling.md @@ -0,0 +1,77 @@ +# Restart memory profiling experiments + +These opt-in switches diagnose large CPU process images. Configure DMTCP +normally. The default reader and checkpoint format stay unchanged. + +| Environment variable | Value | Effect | +| --- | --- | --- | +| `DMTCP_RESTART_PROFILE` | `1` | Log memory restoration, large/slow regions, page faults, the restart barrier and plugin phase. | +| `DMTCP_RESTART_HUGEPAGES` | `1` | Advise huge pages for eligible large private anonymous mappings before loading their contents. | +| `DMTCP_RESTART_MMAP` | `1` | Map eligible raw anonymous payloads of at least 64 MiB directly from the image using private mappings. | +| `DMTCP_RESTART_READERS` | `2`–`8`, `16` | On x86-64, copy raw regular-file payloads of at least 1 GiB with parallel `pread` helpers. Default is one reader. | + +Use one optimization at a time for A/B measurements. Example: + +```sh +DMTCP_RESTART_PROFILE=1 DMTCP_RESTART_HUGEPAGES=1 warmctl up SNAPSHOT --mode in-place +DMTCP_RESTART_PROFILE=1 DMTCP_RESTART_MMAP=1 warmctl up SNAPSHOT --mode in-place +``` + +The profiling fields in `RestoreInfo` are transient. Modified `libdmtcp.so` +must match the library present at checkpoint creation: rebuild the runtime and +bake a new snapshot before testing its post-restart instrumentation. The MTCP +memory reader uses raw syscalls and stores its counters in relocated state. + +`phase=memory` reports microseconds spent in the memory pass and payload reads, +anonymous mmap setup time, bytes read or file-mapped in KiB, and minor/major +faults. `pid` is the real process PID; `virtual_pid` is the checkpoint PID. +`phase=area` reports large or slow payloads, mmap eligibility, offset alignment +and mmap errors. An eligible payload with nonzero `offset_mod` falls back to +reading. `restart_barrier`, +`restart_directory`, `restart_plugins`, and `restart_maps_and_resume` are +sequential post-memory phases, identified by DMTCP's virtual PID. Barrier wait +times across processes overlap and must not be added to the critical path. + +Huge-page advice is best effort and leaves zero runs unpopulated. Actual THP +availability, NUMA placement, pressure and pinning determine its effectiveness. + +The mmap experiment replaces only a region created by the restore itself, +including bounded children of an anonymous parent region. It leaves stacks, +small payloads, original file-backed regions and compressed/pipe inputs on the +ordinary reader. Private mappings protect the source image against writes. +Keep the source image immutable for the entire restored process lifetime. +Do not truncate or rewrite it while mapped. + +Older raw checkpoints can have unaligned payloads after a partial file-backed +mapping. The writer now saves the complete final page of regular files, +bounded by the mapping. This also fixes loss of private dirty bytes between +EOF and the end of that page. Existing readers honor the rounded +`mmapFileSize`; no format version changes. `test/restore-partial-file.c` +reproduces the old dirty-tail loss and verifies the corrected writer. +This writer correction affects newly created snapshots, not existing ones. + +Mmap shifts faults to first access. Measure first useful work, CUDA restoration, +and steady-state inference, not just the memory-pass duration. CUDA host-memory +registration may force private copies and remove some or all of the benefit. +Neither experiment is qualified for GPU production solely by CPU tests. + +Parallel readers can be combined with huge-page advice. Their temporary stacks +are unmapped and every helper is reaped before moving to another memory area or +resuming the application. Helpers share the address space but do not use libc, +TLS or pthreads. Failed helper creation falls back to reading its range in the +parent; failed reads fall back to the ordinary reader after all helpers exit. +Helpers arm a parent-death signal and recheck their parent, so cancellation +does not leave them holding the restore's address space. +`parallel_kib` identifies actual use. `child_minflt` and `child_majflt` are the +helpers' faults, separate from the parent's `minflt` and `majflt`. + +`test/restore-memory-bench.c` checkpoints dense or sparse anonymous memory, +verifies every word after restart, and optionally writes to every page and +checkpoints/restores again. The investigation runner compares both modes on the +same images, explicitly warms their file cache, and checks source SHA-256 for +mmap runs. Its wall time includes full data verification and process exit. + +During instrumentation, `dmtcp_get_restart_env()` was found to bound its scan +by `sizeof(char *)`, which examined only the first environment entry. The fix +uses the actual read length and requires an exact name followed by `=`. The +benchmark verifies a later variable whose longer prefixed name occurs first. diff --git a/doc/warmctl-release-notes.md b/doc/warmctl-release-notes.md new file mode 100644 index 0000000000..2d9d86610c --- /dev/null +++ b/doc/warmctl-release-notes.md @@ -0,0 +1,50 @@ +# v5.0.0-warmctl.1 + +First versioned TensorRails fork release, based on DMTCP 5.0. It includes the +existing warmctl branch's socket, shared-memory, dlopen, fcntl64 and checkpoint +storage fixes, plus the large-memory restart changes below. + +- Opt-in transparent huge-page advice reduces first-touch page-fault overhead. +- Opt-in 2–8 or 16 parallel `pread` readers restore raw payloads of at least + 1 GiB on x86-64. Helpers are reaped before resumption, die with the parent, + and use raw syscalls; failed reads/creation fall back to ordinary reading. +- Restart profiling reports per-process memory reads, faults, mappings, + barriers and plugin time. +- Restart environment lookup scans all entries and matches exact names. +- The checkpoint writer preserves private dirty bytes in a regular file's + last partial page. The checkpoint format is unchanged. +- Experimental private mmap of raw images remains opt-in and is **not + recommended for CUDA restores**: measured work moved into the CUDA phase. + +On one H200, Qwen3.8-Flash-Next-NVFP4, same local snapshot with hot file cache: +ordinary restores took 77.36/82.14 s; huge pages + 16 readers took +20.21/23.43/20.27/21.39 s (median 20.83 s to health, 20.92 s to first streaming +token). In the illustrated pair, DMTCP fell from 54.09 to 4.04 s. This is not +a 20-second SLA or a measurement of provisioning/network transfer. See +[the measured phase breakdown](https://github.com/TensorRails/dmtcp/blob/v5.0.0-warmctl.1/doc/h200-restart-performance.md). + +All 20 GPU restores matched weight fingerprints and passed streaming/tool +smoke tests; 640 short requests completed without errors. Numerical canary +equivalence remains unqualified: the donor already varied before checkpoint +(maximum common-prefix logprob difference 1.15547 at tolerance 0.001). + +Enable the measured reader configuration with: + +```sh +DMTCP_RESTART_HUGEPAGES=1 DMTCP_RESTART_READERS=16 DMTCP_RESTART_MMAP=0 \ + warmctl up SNAPSHOT --mode in-place +``` + +The switches are opt-in in DMTCP itself. THP and reader count depend on RAM, +CPU and storage; compressed streams use ordinary reading. The binary asset +targets Debian 12-compatible x86-64 or ARM64 systems (glibc >= 2.36) and includes the warmctl fdkeep plugin. +The plugin now lives in this fork and has native syscall implementations and +CPU descriptor-restoration tests on both architectures. +Both architectures run native CPU checkpoint/restore tests; the H200 results +apply to x86-64. ARM64 currently uses the ordinary reader with optional THP, +and GPU serving is not qualified on ARM64. These are Linux packages, not +macOS/Windows binaries. Use identical +runtime libraries for bake and restore; bake new snapshots after upgrading. + +ARM64 checkpoint creation keeps the upstream gzip-disable safeguard. CI also +checks restoration of an externally compressed owned fixture on both platforms. diff --git a/include/procmapsarea.h b/include/procmapsarea.h index eee324d5d1..36be768d62 100644 --- a/include/procmapsarea.h +++ b/include/procmapsarea.h @@ -66,7 +66,11 @@ typedef char *VA; /* VA = virtual address */ typedef enum ProcMapsAreaProperties { DMTCP_ZERO_PAGE = 0x0001, DMTCP_ZERO_PAGE_PARENT_HEADER = 0x0002, - DMTCP_ZERO_PAGE_CHILD_HEADER = 0x0004 + DMTCP_ZERO_PAGE_CHILD_HEADER = 0x0004, + // The area is the file's own pages (a private mapping nothing wrote): the + // image holds no data for it, and the restart maps the file again after + // checking it is the one checkpointed (fileSize, fileMtime). + DMTCP_FILE_MAPPED = 0x0008 } ProcMapsAreaProperties; typedef union ProcMapsArea { @@ -112,6 +116,12 @@ typedef union ProcMapsArea { uint64_t properties; char name[FILENAMESIZE]; + + // DMTCP_FILE_MAPPED: the file at checkpoint time. After the name, in + // what was padding, so that images without them still read. + uint64_t fileSize; + int64_t fileMtimeSec; + int64_t fileMtimeNsec; }; char _padding[4096]; } ProcMapsArea; diff --git a/include/util.h b/include/util.h index fd30dbefca..eb4f5a8459 100644 --- a/include/util.h +++ b/include/util.h @@ -262,6 +262,13 @@ bool areZeroPages(void *addr, size_t numPages); bool scanOccupiedRangeBatch(uintptr_t start, uintptr_t end, uintptr_t *size_scanned); +// True if [start, end), a private file mapping, has a page that is no longer +// the file's: written through copy-on-write (a library's RELRO after +// relocation, a text page a debugger patched) or swapped out. Such an area's +// contents are not the file's, so the checkpoint must save them. Reports +// true when the pagemap cannot be inspected. +bool areaHasPrivatePages(uintptr_t start, uintptr_t end); + // Linux 6.13+; defined for build environments whose headers predate them. #ifndef MADV_GUARD_INSTALL # define MADV_GUARD_INSTALL 102 diff --git a/plugin/fdkeep/LICENSE b/plugin/fdkeep/LICENSE new file mode 100644 index 0000000000..d645695673 --- /dev/null +++ b/plugin/fdkeep/LICENSE @@ -0,0 +1,202 @@ + + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + + END OF TERMS AND CONDITIONS + + APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + + Copyright [yyyy] [name of copyright owner] + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. diff --git a/plugin/fdkeep/Makefile b/plugin/fdkeep/Makefile new file mode 100644 index 0000000000..2f4b9b9361 --- /dev/null +++ b/plugin/fdkeep/Makefile @@ -0,0 +1,19 @@ +# This optional warmctl plugin is built and shipped by the fork release. +CC ?= cc +ROOT ?= ../.. +PREFIX ?= /opt/warmctl/dmtcp +CFLAGS ?= -O2 -Wall + +.PHONY: all install clean +all: libdmtcp_fdkeep.so + +libdmtcp_fdkeep.so: fdkeep.c raw-syscalls.h + $(CC) $(CFLAGS) -shared -fPIC -I$(ROOT)/include -o $@ fdkeep.c + +install: all + install -d $(DESTDIR)$(PREFIX)/lib $(DESTDIR)$(PREFIX)/share/doc/dmtcp/fdkeep + install -m 0755 libdmtcp_fdkeep.so $(DESTDIR)$(PREFIX)/lib/ + install -m 0644 LICENSE README.md $(DESTDIR)$(PREFIX)/share/doc/dmtcp/fdkeep/ + +clean: + rm -f libdmtcp_fdkeep.so diff --git a/plugin/fdkeep/README.md b/plugin/fdkeep/README.md new file mode 100644 index 0000000000..6fc78fce10 --- /dev/null +++ b/plugin/fdkeep/README.md @@ -0,0 +1,19 @@ +# warmctl fdkeep plugin + +Imported from `TensorRails/warmctl` at `7e082c0`, under its Apache-2.0 +license (LICENSE). This fork now owns and versions the plugin together with +the DMTCP runtime, so warmctl consumes a complete released native package. + +The plugin preserves driver eventfds, real pipes, thread names, Unix listeners +and epoll watches missed by DMTCP wrappers. It is loaded explicitly, never +enabled for arbitrary DMTCP applications by default. + +```sh +make -C plugin/fdkeep +make -C plugin/fdkeep install PREFIX=/opt/warmctl/dmtcp +``` + +Linux x86-64 and ARM64 use native raw syscalls. CI verifies an untracked +eventfd, pipe contents, Unix listener and epoll watch through checkpoint and +restore. H200 GPU evidence is x86-64 only; ARM64 GPU support, including +cuInterpose, is not qualified by the CPU package tests. diff --git a/plugin/fdkeep/fdkeep.c b/plugin/fdkeep/fdkeep.c new file mode 100644 index 0000000000..45abec5a3c --- /dev/null +++ b/plugin/fdkeep/fdkeep.c @@ -0,0 +1,504 @@ +// SPDX-License-Identifier: Apache-2.0 +// Imported from TensorRails/warmctl; see README.md and LICENSE. +// DMTCP plugin: keep eventfds and pipes that DMTCP did not track. +// The CUDA driver creates the eventfds and pipes its helper threads poll during +// cuInit; DMTCP drops them on restart, the helper threads see POLLNVAL and exit, +// and cuda-checkpoint can no longer find the restore thread. Before a checkpoint +// we record every eventfd and pipe end; on restart we recreate the ones that are +// missing at the same fd numbers, with the same flags, counter and pipe bytes. +// Listening Unix sockets bound to a path are kept the same way: cuInterpose +// makes its control socket with raw syscalls, which DMTCP never sees. +// +// Some calls bypass libc with raw syscalls, where DMTCP's wrappers must not see them (its pipe() returns a +// socketpair, its listen() takes a lock the checkpoint thread holds). +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include "dmtcp.h" + +#include "raw-syscalls.h" + +#define MAXFD 512 +#define MAXWATCH 4096 + +enum kind { NONE, EVENTFD, PIPE, LISTEN }; + +struct rec { + enum kind kind; + int flags; // file status flags from fdinfo (octal) + int cloexec; + unsigned long long count; // eventfd counter + unsigned long inode; // pipe inode, pairs read and write ends + int len; // pipe bytes pending (read end only) + char *data; + char *path; // a listening Unix socket's path + int type; // its socket type + mode_t mode; // and the path's permissions +}; + +static struct rec recs[MAXFD]; + +static void logf_(const char *fmt, ...) +{ + char buf[512]; + va_list ap; + va_start(ap, fmt); + int n = vsnprintf(buf, sizeof buf, fmt, ap); + va_end(ap); + int fd = open("/tmp/fdkeep.log", O_WRONLY | O_CREAT | O_APPEND | O_CLOEXEC, 0644); + if (fd >= 0) { + if (write(fd, buf, n) < 0) {} + close(fd); + } +} + +static int fdinfo(int fd, int *flags, unsigned long long *count) +{ + char path[64], line[256]; + snprintf(path, sizeof path, "/proc/self/fdinfo/%d", fd); + FILE *f = fopen(path, "re"); + if (!f) return -1; + while (fgets(line, sizeof line, f)) { + if (sscanf(line, "flags: %o", flags) == 1) continue; + sscanf(line, "eventfd-count: %llx", count); + } + fclose(f); + return 0; +} + +static int nwatches, lostwatches; +static void record_epoll(int epfd); + +// A listening Unix socket's path, type and permissions, kept in case DMTCP does not restore it. +static void record_unix(int fd, struct rec *r) +{ + int domain = 0; + socklen_t len = sizeof domain; + struct sockaddr_un a; + socklen_t alen = sizeof a; + struct stat st; + if (getsockopt(fd, SOL_SOCKET, SO_DOMAIN, &domain, &len) != 0 || domain != AF_UNIX) return; + len = sizeof r->type; + if (getsockopt(fd, SOL_SOCKET, SO_TYPE, &r->type, &len) != 0) return; + memset(&a, 0, sizeof a); + // An abstract name starts with a NUL and has no file to bind again. + if (getsockname(fd, (struct sockaddr *)&a, &alen) != 0 || a.sun_path[0] == 0) return; + a.sun_path[sizeof a.sun_path - 1] = 0; + if (lstat(a.sun_path, &st) != 0 || !S_ISSOCK(st.st_mode)) return; + r->path = strdup(a.sun_path); + r->mode = st.st_mode & 07777; +} + +static void record(void) +{ + for (int fd = 0; fd < MAXFD; fd++) { + free(recs[fd].data); + free(recs[fd].path); + } + memset(recs, 0, sizeof recs); + nwatches = lostwatches = 0; + int past = 0; // fds of a kind we keep, numbered past MAXFD: not kept + DIR *d = opendir("/proc/self/fd"); + if (!d) return; + int self = dirfd(d); + struct dirent *e; + while ((e = readdir(d))) { + int fd = atoi(e->d_name); + if (e->d_name[0] == '.' || fd == self) continue; + char path[64], target[128]; + snprintf(path, sizeof path, "/proc/self/fd/%d", fd); + ssize_t n = readlink(path, target, sizeof target - 1); + if (n <= 0) continue; + target[n] = 0; + if (fd >= MAXFD) { + // DMTCP's own descriptors (its coordinator's socket, a pipe: 821 and 828) are its to restore. + if (dmtcp_is_protected_fd(fd)) continue; + if (strncmp(target, "anon_inode:[event", 17) == 0 || strncmp(target, "pipe:", 5) == 0 || + strncmp(target, "socket:", 7) == 0) + past++; + continue; + } + struct rec *r = &recs[fd]; + if (strcmp(target, "anon_inode:[eventfd]") == 0) { + r->kind = EVENTFD; + } else if (sscanf(target, "pipe:[%lu]", &r->inode) == 1) { + r->kind = PIPE; + } else if (strcmp(target, "anon_inode:[eventpoll]") == 0) { + record_epoll(fd); + continue; + } else if (strncmp(target, "socket:", 7) == 0) { + // DMTCP sometimes misses listen() and restores the socket bound but not listening. + int on = 0; + socklen_t len = sizeof on; + if (getsockopt(fd, SOL_SOCKET, SO_ACCEPTCONN, &on, &len) != 0 || !on) continue; + r->kind = LISTEN; + record_unix(fd, r); + fdinfo(fd, &r->flags, &r->count); + r->cloexec = (fcntl(fd, F_GETFD) & FD_CLOEXEC) != 0; + continue; + } else { + continue; + } + fdinfo(fd, &r->flags, &r->count); + r->cloexec = (fcntl(fd, F_GETFD) & FD_CLOEXEC) != 0; + if (r->kind == PIPE && (r->flags & O_ACCMODE) == O_RDONLY) { + int avail = 0; + if (ioctl(fd, FIONREAD, &avail) == 0 && avail > 0) { + // Take the bytes out to save them, then put them back for the live process. + r->data = malloc(avail); + int fl = fcntl(fd, F_GETFL); + fcntl(fd, F_SETFL, fl | O_NONBLOCK); + r->len = read(fd, r->data, avail); + fcntl(fd, F_SETFL, fl); + if (r->len < 0) r->len = 0; + for (int w = 0; w < MAXFD && r->len > 0; w++) { + if (recs[w].kind == PIPE && recs[w].inode == r->inode && (recs[w].flags & O_ACCMODE) == O_WRONLY) { + if (write(w, r->data, r->len) < 0) {} + break; + } + } + } + } + } + closedir(d); + int ne = 0, np = 0, nl = 0; + for (int fd = 0; fd < MAXFD; fd++) { + if (recs[fd].kind == EVENTFD) ne++; + if (recs[fd].kind == PIPE) np++; + if (recs[fd].kind == LISTEN) nl++; + } + logf_("pid %d precheckpoint: %d eventfds, %d pipe ends, %d listening sockets, %d epoll watches\n", getpid(), ne, + np, nl, nwatches); + if (past > 0 || lostwatches > 0) + logf_("pid %d precheckpoint: NOT KEPT: %d eventfds, pipes, epolls or sockets past fd %d, %d epoll watches past %d\n", + getpid(), past, MAXFD, lostwatches, MAXWATCH); +} + +static long raw_pipe2(int *p, int flags) +{ + return fdkeep_raw4(SYS_pipe2, (long)p, flags, 0, 0); +} + +// Epoll watches. A forked child shares its parent's epoll fds and sockets, and DMTCP refills a shared +// epoll from the watch list of whichever process restores it: when that is the child, the watches the +// parent added after the fork (uvloop's listening socket in vLLM's API server) are gone, and the +// server never accepts. Record the kernel's own list, keeping only watches whose fd in this process +// is still the watched file, and add back whatever is missing after restart. +struct watch { + int epfd, tfd; + unsigned int events; + unsigned long long data; +}; +static struct watch watches[MAXWATCH]; + +static void record_epoll(int epfd) +{ + char path[64], line[256]; + snprintf(path, sizeof path, "/proc/self/fdinfo/%d", epfd); + FILE *f = fopen(path, "re"); + if (!f) return; + while (fgets(line, sizeof line, f)) { + int tfd; + unsigned int events; + unsigned long long data; + unsigned long ino; + if (sscanf(line, "tfd: %d events: %x data: %llx pos:%*d ino:%lx", &tfd, &events, &data, &ino) != 4) continue; + struct stat st; + if (fstat(tfd, &st) != 0 || st.st_ino != ino) continue; // registered through another process's fd table + if (nwatches == MAXWATCH) { + lostwatches++; + continue; + } + watches[nwatches++] = (struct watch){epfd, tfd, events, data}; + } + fclose(f); +} + +static long raw_epoll_ctl(int epfd, int op, int fd, struct epoll_event *ev) +{ + return fdkeep_raw4(SYS_epoll_ctl, epfd, op, fd, (long)ev); +} + +static void rewatch(void) +{ + int added = 0, failed = 0; + for (int i = 0; i < nwatches; i++) { + struct epoll_event ev = {.events = watches[i].events, .data.u64 = watches[i].data}; + long r = raw_epoll_ctl(watches[i].epfd, EPOLL_CTL_ADD, watches[i].tfd, &ev); + if (r == 0) added++; + else if (r != -EEXIST) failed++; + } + logf_("pid %d restart: %d of %d epoll watches were missing and added back, %d failed\n", getpid(), added, nwatches, failed); +} + +// DMTCP's listen() wrapper takes a lock the checkpoint thread may hold while threads resume. +static long raw_listen(int fd, int backlog) +{ + return fdkeep_raw4(SYS_listen, fd, backlog, 0, 0); +} + +static long raw3(long nr, long a, long b, long c) +{ + return fdkeep_raw4(nr, a, b, c, 0); +} + +// Move a fresh fd above every slot we may fill, so placing one never clobbers another. +static int high(int fd) +{ + int h = fcntl(fd, F_DUPFD_CLOEXEC, MAXFD + 100); + close(fd); + return h; +} + +static void place(int src, int fd, const struct rec *r) +{ + dup2(src, fd); + fcntl(fd, F_SETFL, r->flags & (O_NONBLOCK | O_APPEND)); + fcntl(fd, F_SETFD, r->cloexec ? FD_CLOEXEC : 0); +} + +// Is fd open and of the recorded kind? DMTCP restores the pipes it tracked as +// socketpairs, and cuda-checkpoint wakes the restore thread by opening +// /proc/PID/fd/N of its pipe, which fails with ENXIO on a socket. +static int same_kind(int fd, enum kind kind) +{ + char path[64], target[128]; + snprintf(path, sizeof path, "/proc/self/fd/%d", fd); + ssize_t n = readlink(path, target, sizeof target - 1); + if (n <= 0) return 0; + target[n] = 0; + if (kind == EVENTFD) return strcmp(target, "anon_inode:[eventfd]") == 0; + return strncmp(target, "pipe:", 5) == 0; +} + +// Bind a Unix listener DMTCP did not restore again at its path, as its process made it (cuInterpose: +// nonblocking, the path 0600 before listen). The path left by the process checkpointed is removed only +// when it is a socket of ours that nothing listens on. +static int relisten(int fd, const struct rec *r) +{ + struct sockaddr_un a = {.sun_family = AF_UNIX}; + struct stat st; + strncpy(a.sun_path, r->path, sizeof a.sun_path - 1); + if (lstat(r->path, &st) == 0) { + long probe = raw3(SYS_socket, AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0); + long c = probe < 0 ? probe : raw3(SYS_connect, probe, (long)&a, sizeof a); + if (probe >= 0) close((int)probe); + if (!S_ISSOCK(st.st_mode) || st.st_uid != geteuid() || c != -ECONNREFUSED) return -1; + if (unlink(r->path) != 0) return -1; + } + long s = raw3(SYS_socket, AF_UNIX, r->type | SOCK_CLOEXEC, 0); + if (s < 0) return -1; + int h = high((int)s); + if (h < 0 || raw3(SYS_bind, h, (long)&a, sizeof a) != 0 || chmod(r->path, r->mode) != 0 || + raw_listen(h, SOMAXCONN) != 0) { + if (h >= 0) close(h); + return -1; + } + place(h, fd, r); + close(h); + return 0; +} + +static void recreate(void) +{ + int made = 0, kept = 0; + // A pipe is redone whole when either end is missing or not a pipe. + for (int fd = 0; fd < MAXFD; fd++) { + struct rec *r = &recs[fd]; + if (r->kind != PIPE || r->inode == 0 || same_kind(fd, PIPE)) continue; + for (int g = 0; g < MAXFD; g++) { + if (recs[g].kind == PIPE && recs[g].inode == r->inode && fcntl(g, F_GETFD) != -1) { + close(g); + } + } + } + for (int fd = 0; fd < MAXFD; fd++) { + struct rec *r = &recs[fd]; + if (r->kind == NONE) continue; + if (r->kind == LISTEN) { + int on = 0; + socklen_t len = sizeof on; + if (getsockopt(fd, SOL_SOCKET, SO_ACCEPTCONN, &on, &len) == 0) { + if (!on) { + logf_(" fd %d was listening, restored without listen: listen -> %ld\n", fd, raw_listen(fd, 4096)); + made++; + } + } else if (r->path && fcntl(fd, F_GETFD) == -1) { + int ok = relisten(fd, r); + logf_(" fd %d listened on %s, not restored: %s\n", fd, r->path, ok == 0 ? "bound again" : "FAILED"); + if (ok == 0) made++; + } + continue; + } + if (fcntl(fd, F_GETFD) != -1) { // DMTCP restored it, or the slot is taken + if (r->kind == EVENTFD || r->inode != 0) kept++; + continue; + } + if (r->kind == EVENTFD) { + int e = high(eventfd(r->count, 0)); + place(e, fd, r); + close(e); + made++; + continue; + } + // Pipe: create once per inode, then place both ends wherever they lived. + if (r->inode == 0) continue; + unsigned long ino = r->inode; + int p[2]; + // DMTCP's pipe() and syscall() wrappers return a socketpair; we need a real pipe. + if (raw_pipe2(p, 0) != 0) continue; + p[0] = high(p[0]); + p[1] = high(p[1]); + for (int g = fd; g < MAXFD; g++) { + struct rec *q = &recs[g]; + if (q->kind != PIPE || q->inode != ino || fcntl(g, F_GETFD) != -1) continue; + int rd = (q->flags & O_ACCMODE) == O_RDONLY; + if (rd && q->len > 0 && write(p[1], q->data, q->len) < 0) {} + place(rd ? p[0] : p[1], g, q); + q->inode = 0; // done + made++; + } + close(p[0]); + close(p[1]); + } + logf_("pid %d restart: recreated %d fds, %d already open\n", getpid(), made, kept); + for (int fd = 0; fd < MAXFD; fd++) { + if (recs[fd].kind == PIPE && !same_kind(fd, PIPE)) logf_(" fd %d is not a pipe after fixup\n", fd); + } +} + +// Thread names. DMTCP restarts every thread as "DMTCP:" cut to 15 bytes, +// and cuda-checkpoint finds the driver's restore thread by its name +// (cuda). Save the names before a checkpoint and put them back after. +#define MAXTHREADS 256 +static char names[MAXTHREADS][16]; +static int nnames; + +static int readcomm(const char *tid, char *out) +{ + char path[64]; + snprintf(path, sizeof path, "/proc/self/task/%s/comm", tid); + int fd = open(path, O_RDONLY | O_CLOEXEC); + if (fd < 0) return -1; + int n = read(fd, out, 15); + close(fd); + if (n <= 0) return -1; + out[n] = 0; + out[strcspn(out, "\n")] = 0; + return 0; +} + +static void savenames(void) +{ + nnames = 0; + int past = 0; + DIR *d = opendir("/proc/self/task"); + if (!d) return; + struct dirent *e; + while ((e = readdir(d))) { + if (e->d_name[0] == '.') continue; + if (nnames == MAXTHREADS) { + past++; + continue; + } + if (readcomm(e->d_name, names[nnames]) == 0) nnames++; + } + closedir(d); + if (past > 0) logf_("pid %d precheckpoint: NOT KEPT: the names of %d threads past %d\n", getpid(), past, MAXTHREADS); +} + +static void restorenames(void) +{ + char used[MAXTHREADS] = {0}; + int fixed = 0; + DIR *d = opendir("/proc/self/task"); + if (!d) return; + struct dirent *e; + while ((e = readdir(d))) { + char cur[16]; + if (e->d_name[0] == '.' || readcomm(e->d_name, cur) != 0) continue; + if (strncmp(cur, "DMTCP:", 6) != 0) continue; + const char *stem = cur + 6; + for (int i = 0; i < nnames; i++) { + if (used[i] || strncmp(names[i], stem, strlen(stem)) != 0) continue; + used[i] = 1; + char path[64]; + snprintf(path, sizeof path, "/proc/self/task/%.20s/comm", e->d_name); // a TID: digits + int fd = open(path, O_WRONLY | O_CLOEXEC); + if (fd >= 0) { + if (write(fd, names[i], strlen(names[i])) > 0) fixed++; + close(fd); + } + break; + } + } + closedir(d); + logf_("pid %d restart: renamed %d of %d threads\n", getpid(), fixed, nnames); +} + +// RESTART runs plugins in reverse order, so ours runs before DMTCP's file and +// socket plugins, which then dup their socketpairs over our pipes. THREAD_RESUME +// comes after them and before any thread returns to application code: the first +// thread there does the fixup, the rest wait for it. +static volatile int restarted, fixing, fixed; + +static void hook(DmtcpEvent_t event, DmtcpEventData_t *data) +{ + (void)data; + switch (event) { + case DMTCP_EVENT_INIT: + case DMTCP_EVENT_ATFORK_CHILD: + // Yama ptrace_scope=1 lets only ancestors attach; with FDKEEP_PTRACER_ANY set, gdb can attach to a + // process stuck in a checkpoint from any shell. + if (getenv("FDKEEP_PTRACER_ANY")) prctl(PR_SET_PTRACER, PR_SET_PTRACER_ANY, 0, 0, 0); + break; + case DMTCP_EVENT_PRECHECKPOINT: + record(); + savenames(); + break; + case DMTCP_EVENT_RESTART: + fixing = fixed = 0; + __atomic_store_n(&restarted, 1, __ATOMIC_SEQ_CST); + break; + case DMTCP_EVENT_THREAD_RESUME: + if (!__atomic_load_n(&restarted, __ATOMIC_SEQ_CST)) break; + if (__atomic_exchange_n(&fixing, 1, __ATOMIC_SEQ_CST) == 0) { + recreate(); + rewatch(); + restorenames(); + __atomic_store_n(&restarted, 0, __ATOMIC_SEQ_CST); + __atomic_store_n(&fixed, 1, __ATOMIC_SEQ_CST); + } else { + while (!__atomic_load_n(&fixed, __ATOMIC_SEQ_CST)) usleep(100); + } + break; + default: + break; + } +} + +DmtcpPluginDescriptor_t fdkeep_plugin = { + DMTCP_PLUGIN_API_VERSION, + DMTCP_PACKAGE_VERSION, + "fdkeep", + "warmctl", + "", + "Keep what DMTCP loses: CUDA's eventfds and pipes, thread names, listening sockets, epoll watches", + hook +}; + +DMTCP_DECL_PLUGIN(fdkeep_plugin); diff --git a/plugin/fdkeep/raw-syscalls.h b/plugin/fdkeep/raw-syscalls.h new file mode 100644 index 0000000000..4d94e926f0 --- /dev/null +++ b/plugin/fdkeep/raw-syscalls.h @@ -0,0 +1,28 @@ +/* SPDX-License-Identifier: Apache-2.0 */ +#ifndef FDKEEP_RAW_SYSCALLS_H +#define FDKEEP_RAW_SYSCALLS_H + +/* Bypass libc/DMTCP wrappers; return the kernel's negative errno directly. */ +static inline long fdkeep_raw4(long nr, long a, long b, long c, long d) +{ +#if defined(__x86_64__) + long ret; + register long r10 __asm__("r10") = d; + __asm__ volatile("syscall" : "=a"(ret) + : "0"(nr), "D"(a), "S"(b), "d"(c), "r"(r10) + : "rcx", "r11", "memory"); + return ret; +#elif defined(__aarch64__) + register long x8 __asm__("x8") = nr; + register long x0 __asm__("x0") = a; + register long x1 __asm__("x1") = b; + register long x2 __asm__("x2") = c; + register long x3 __asm__("x3") = d; + __asm__ volatile("svc 0" : "+r"(x0) + : "r"(x8), "r"(x1), "r"(x2), "r"(x3) : "memory", "cc"); + return x0; +#else +#error "fdkeep supports Linux x86-64 and ARM64" +#endif +} +#endif diff --git a/src/ckptserializer.cpp b/src/ckptserializer.cpp index 577b71c873..3f5f42c1fe 100644 --- a/src/ckptserializer.cpp +++ b/src/ckptserializer.cpp @@ -246,6 +246,25 @@ open_ckpt_to_write_gz(int fd, int pipe_fds[2], char *gzip_path) return open_ckpt_to_write(fd, pipe_fds, gzip_args); } +// zstd at its fastest level on every core: about 2.8x on a vLLM process +// image, and decompressed at 1 GB/s or more by `zstd -d` on restart. +static int +open_ckpt_to_write_zstd(int fd, int pipe_fds[2], char *zstd_path) +{ + char *zstd_args[] = { + const_cast("zstd"), + const_cast("-1"), + const_cast("-T0"), + const_cast("-q"), + const_cast("-c"), + const_cast("-"), + NULL + }; + zstd_args[0] = zstd_path; + TRACE("open_ckpt_to_write_zstd\n"); + return open_ckpt_to_write(fd, pipe_fds, zstd_args); +} + static int perform_open_ckpt_image_fd(const char *tempCkptFilename, bool *use_compression, @@ -265,31 +284,46 @@ perform_open_ckpt_image_fd(const char *tempCkptFilename, return fd; #endif // ifdef FAST_RST_VIA_MMAP - /* 2. Test if using GZIP compression */ + /* 2. Test if using compression: zstd (DMTCP_ZSTD=1, off by default) wins + * over gzip (DMTCP_GZIP, on by default). */ + int use_zstd_compression = 0; int use_gzip_compression = 0; int use_deltacompression = 0; - char *gzip_cmd = const_cast("gzip"); + char zstd_path[PATH_MAX]; char gzip_path[PATH_MAX]; - use_gzip_compression = test_use_compression(const_cast("GZIP"), - gzip_cmd, gzip_path, 1); + use_zstd_compression = test_use_compression(const_cast("ZSTD"), + const_cast("zstd"), + zstd_path, 0); + if (!use_zstd_compression) { + use_gzip_compression = test_use_compression(const_cast("GZIP"), + const_cast("gzip"), + gzip_path, 1); + } /* 3. We now have the information to pipe to gzip, or directly to fd. * We do it this way, so that gzip will be direct child of forked process * when using forked checkpointing. */ - if (use_deltacompression || use_gzip_compression) { /* fork compr. process */ + if (use_deltacompression || use_gzip_compression || + use_zstd_compression) { /* fork compr. process */ /* 3a. Open pipe */ int pipe_fds[2]; if (_real_pipe(pipe_fds) == -1) { WARN_ERRNO(false, "error creating compression pipe; compression will not be " "used"); - use_gzip_compression = use_deltacompression = 0; + use_gzip_compression = use_deltacompression = use_zstd_compression = 0; } /* 3b. Fork compressor child */ - if (use_gzip_compression) {/* fork a gzip process */ + if (use_zstd_compression) { /* fork a zstd process */ + *use_compression = true; + fd = open_ckpt_to_write_zstd(fd, pipe_fds, zstd_path); + if (pipe_fds[0] == -1) { + *use_compression = false; + } + } else if (use_gzip_compression) {/* fork a gzip process */ *use_compression = true; fd = open_ckpt_to_write_gz(fd, pipe_fds, gzip_path); if (pipe_fds[0] == -1) { diff --git a/src/constants.h b/src/constants.h index d2049540d5..0a659d5577 100644 --- a/src/constants.h +++ b/src/constants.h @@ -143,6 +143,11 @@ // reports) and always uses ioctl(PAGEMAP_SCAN). #define ENV_VAR_DISABLE_PAGEMAP_SCAN "DMTCP_DISABLE_PAGEMAP_SCAN" +// Set to "1" to save the pages of every file mapping in the image, as before +// DMTCP_FILE_MAPPED, instead of mapping unmodified ones from their file on +// restart. +#define ENV_VAR_DISABLE_FILE_MAPPED "DMTCP_DISABLE_FILE_MAPPED" + // this list should be kept up to date with all "protected" environment vars #define ENV_VARS_ALL \ ENV_VAR_NAME_HOST, \ @@ -181,6 +186,7 @@ ENV_VAR_REAL_PPID, \ ENV_VAR_FSGSBASE_ENABLED, \ ENV_VAR_DISABLE_PAGEMAP_SCAN, \ + ENV_VAR_DISABLE_FILE_MAPPED, \ ENV_VAR_LOG_FILE #define DMTCP_RESTART_CMD "dmtcp_restart" diff --git a/src/dmtcpplugin.cpp b/src/dmtcpplugin.cpp index 82447c56bd..ae2ecd5e2f 100644 --- a/src/dmtcpplugin.cpp +++ b/src/dmtcpplugin.cpp @@ -237,10 +237,16 @@ dmtcp_get_uniquepid_str(void) return ProcessInfo::instance().upidStr().c_str(); } +// A forked child's upid is null until its first checkpoint (resetOnFork); +// getUpid() then takes the child's own, which UniquePid reset at the fork. +// Read directly, every connection a child created before its first +// checkpoint was named 0-0-0(n), with n counted on from the parent's: two +// children's connections got the same names, and the restart waited forever +// for a connection the other child had taken. EXTERNC DmtcpUniqueProcessId dmtcp_get_uniquepid(void) { - return ProcessInfo::instance().upid; + return ProcessInfo::instance().getUpid().upid(); } EXTERNC DmtcpUniqueProcessId @@ -385,9 +391,12 @@ dmtcp_get_restart_env(const char *name, // IN char *start_ptr = env_buf; // iterate over the flattened list of name-value pairs - while (start_ptr - env_buf < (int)sizeof(env_buf)) { + // env_buf is a pointer, not an array. sizeof(env_buf) only inspected + // the first environment entry on 64-bit hosts. Bound the scan by the + // bytes read, and require the complete variable name (not its prefix). + while (start_ptr - env_buf < count) { pos = NULL; - if (strncmp(start_ptr, name, namelen) == 0) { + if (strncmp(start_ptr, name, namelen) == 0 && start_ptr[namelen] == '=') { if ((pos = strchr(start_ptr, '='))) { strncpy(value, pos + 1, maxvaluelen); if (strlen(pos + 1) >= maxvaluelen) { diff --git a/src/dmtcprestartinternal.cpp b/src/dmtcprestartinternal.cpp index f6a7d697c4..c15e0d1a37 100644 --- a/src/dmtcprestartinternal.cpp +++ b/src/dmtcprestartinternal.cpp @@ -24,6 +24,7 @@ #include #include #include +#include #include #include #include @@ -648,6 +649,17 @@ openCkptFileToRead(const string &filename) const_cast("-"), NULL }; + const char *zstd_path = "zstd"; + static const char *zstd_args[] = { + const_cast("zstd"), + const_cast("-d"), + const_cast("-q"), + const_cast("-c"), + const_cast("-"), + NULL + }; + static const unsigned char zstd_magic[] = { 0x28, 0xb5, 0x2f, 0xfd }; + unsigned char magic[sizeof zstd_magic] = {}; pid_t cpid; @@ -670,17 +682,19 @@ openCkptFileToRead(const string &filename) ASSERT_ERRNO(lseek(fd, 0, SEEK_SET) != -1, "failed to rewind checkpoint file before magic read: path={}", filename); - ASSERT_EQ(static_cast(1), - Util::readAll(fd, &fc, 1), - "failed to read checkpoint magic byte: path={}", + ASSERT_EQ(static_cast(sizeof magic), + Util::readAll(fd, magic, sizeof magic), + "failed to read checkpoint magic bytes: path={}", filename.c_str()); ASSERT_ERRNO(lseek(fd, 0, SEEK_SET) != -1, "failed to rewind checkpoint file after magic read: path={}", filename); + fc = static_cast(magic[0]); + bool zstd = memcmp(magic, zstd_magic, sizeof magic) == 0; - if (fc == GZIP_FIRST) { - decomp_path = gzip_path; - decomp_args = gzip_args; + if (fc == GZIP_FIRST || zstd) { + decomp_path = zstd ? zstd_path : gzip_path; + decomp_args = zstd ? zstd_args : gzip_args; ASSERT_ERRNO(pipe(fds) != -1, "creating gunzip pipe for checkpoint file: {}", diff --git a/src/dmtcpworker.cpp b/src/dmtcpworker.cpp index 065954e73a..362876e3f4 100644 --- a/src/dmtcpworker.cpp +++ b/src/dmtcpworker.cpp @@ -23,6 +23,7 @@ #include #include #include +#include #include "../jalib/jconvert.h" #include "../jalib/jfilesystem.h" #include "../jalib/jsocket.h" @@ -550,13 +551,35 @@ DmtcpWorker::postCheckpoint() void DmtcpWorker::postRestart() { + char profileValue[8] = {}; + bool profile = dmtcp_get_restart_env("DMTCP_RESTART_PROFILE", profileValue, + sizeof(profileValue)) == RESTART_ENV_SUCCESS && + strcmp(profileValue, "1") == 0; + auto profileTime = []() -> uint64_t { + struct timespec ts; + clock_gettime(CLOCK_MONOTONIC, &ts); + return (uint64_t) ts.tv_sec * 1000000 + ts.tv_nsec / 1000; + }; + uint64_t phaseStart = profile ? profileTime() : 0; + auto profilePhase = [&](const char *phase) { + if (profile) { + uint64_t now = profileTime(); + dprintf(STDERR_FILENO, "DMTCP_PROFILE pid=%d phase=%s total_us=%llu\n", + getpid(), phase, (unsigned long long) (now - phaseStart)); + phaseStart = profileTime(); + } + }; TRACE("begin postRestart()"); WorkerState::setCurrentState(WorkerState::RESTARTING); TRACE("Waiting for Restart barrier"); CoordinatorAPI::waitForBarrier("DMT:Restart"); + profilePhase("restart_barrier"); + ProcessInfo::instance().restartCkptDir(); + profilePhase("restart_directory"); PluginManager::eventHook(DMTCP_EVENT_RESTART); + profilePhase("restart_plugins"); TRACE("got resume message after restart"); @@ -575,4 +598,5 @@ DmtcpWorker::postRestart() TRACE("Informing coordinator of RUNNING status: process={}", UniquePid::ThisProcess()); CoordinatorAPI::sendMsgToCoordinator(DMT_WORKER_RESUMING); + profilePhase("restart_maps_and_resume"); } diff --git a/src/mtcp/mtcp_restart.c b/src/mtcp/mtcp_restart.c index cba2eb854e..f8667fc341 100644 --- a/src/mtcp/mtcp_restart.c +++ b/src/mtcp/mtcp_restart.c @@ -51,6 +51,9 @@ #include #include #include +#include +#include +#include #include #include @@ -175,7 +178,8 @@ mtcp_simulateread(RestoreInfo *rinfo) } if ((area.properties & DMTCP_ZERO_PAGE) == 0 && - (area.properties & DMTCP_ZERO_PAGE_PARENT_HEADER) == 0) { + (area.properties & DMTCP_ZERO_PAGE_PARENT_HEADER) == 0 && + (area.properties & DMTCP_FILE_MAPPED) == 0) { off_t seekLen = area.size; if (!(area.flags & MAP_ANONYMOUS) && area.mmapFileSize > 0) { @@ -196,7 +200,8 @@ mtcp_simulateread(RestoreInfo *rinfo) ((area.flags & MAP_SHARED) ? 's' : ((area.flags & MAP_PRIVATE) ? 'p' : '-')), - ((area.flags & MAP_ANONYMOUS) ? "Anon" : " "), + ((area.flags & MAP_ANONYMOUS) ? "Anon" + : (area.properties & DMTCP_FILE_MAPPED) ? "File" : " "), // area.offset, area.devmajor, area.devminor, area.inodenum, area.name); @@ -223,6 +228,16 @@ mtcp_restart_process_args(int argc, char *argv[], char **environ, void (*restore rinfo.use_gdb = 0; rinfo.dprintfEnabled = mtcp_parse_dprintf_env(environ); + char *profile = mtcp_getenv("DMTCP_RESTART_PROFILE", environ); + rinfo.profileRestart = profile && mtcp_strcmp(profile, "1") == 0; + char *huge = mtcp_getenv("DMTCP_RESTART_HUGEPAGES", environ); + rinfo.restartHugePages = huge && mtcp_strcmp(huge, "1") == 0; + char *mapImage = mtcp_getenv("DMTCP_RESTART_MMAP", environ); + rinfo.restartMmap = mapImage && mtcp_strcmp(mapImage, "1") == 0; + char *readers = mtcp_getenv("DMTCP_RESTART_READERS", environ); + rinfo.restartReaders = readers && readers[0] >= '2' && readers[0] <= '8' && + readers[1] == '\0' ? readers[0] - '0' : 1; + if (readers && mtcp_strcmp(readers, "16") == 0) rinfo.restartReaders = 16; char *restart_pause_str = mtcp_getenv("DMTCP_RESTART_PAUSE", environ); if (restart_pause_str == NULL) { @@ -826,14 +841,198 @@ mtcp_restore_vdso_vvar(RestoreInfo *rinfo) * file. MAP_ANONYMOUS is not currently POSIX.) * **************************************************************************/ +// Raw syscalls only: libc and the original globals are unavailable here. +static uint64_t +restart_time_us(void) +{ + int mtcp_sys_errno; + struct timespec ts; + if (mtcp_inline_syscall(clock_gettime, 2, CLOCK_MONOTONIC, &ts) < 0) { + return 0; + } + return (uint64_t) ts.tv_sec * 1000000 + ts.tv_nsec / 1000; +} + +#if defined(__x86_64__) +// These short-lived helpers share only the address space. They have separate +// stacks and fd tables, use raw syscalls only, and are reaped before restoring +// another region or resuming DMTCP. No libc, TLS, pthreads, or persistent pool. +// The assembly trampoline is necessary: a raw clone returns on a NEW stack. +// Returning into a normal C caller in the child would use the wrong frame. +extern long mtcp_read_clone(int (*fn)(void *), void *stack, void *arg) + __attribute__((visibility("hidden"))); +__asm__(".text\n" + ".type mtcp_read_clone,@function\n" + ".hidden mtcp_read_clone\n" + "mtcp_read_clone:\n" + "and $-16,%rsi\n" + "sub $16,%rsi\n" + "mov %rdi,(%rsi)\n" + "mov %rdx,8(%rsi)\n" + "mov $273,%edi\n" // CLONE_VM | SIGCHLD on Linux x86-64 + "xor %edx,%edx\n" + "xor %r10d,%r10d\n" + "xor %r8d,%r8d\n" + "mov $56,%eax\n" // __NR_clone on Linux x86-64 + "syscall\n" + "test %rax,%rax\n" + "jnz 1f\n" + "pop %rax\n" + "pop %rdi\n" + "call *%rax\n" + "mov %eax,%edi\n" + "mov $60,%eax\n" // __NR_exit, not exit_group + "syscall\n" + "ud2\n" + "1: ret\n" + ".size mtcp_read_clone,.-mtcp_read_clone\n"); + +typedef struct { + int fd; + char *address; + size_t size; + off_t offset; + volatile int result; + pid_t parent; +} RestartReadJob; + +static int +restart_read_job(void *raw) +{ + int mtcp_sys_errno; + RestartReadJob *job = (RestartReadJob *) raw; + size_t done = 0; + while (done < job->size) { + size_t chunk = job->size - done; + if (chunk > 64 * 1024 * 1024) chunk = 64 * 1024 * 1024; + long n = mtcp_inline_syscall(pread64, 4, job->fd, job->address + done, + chunk, job->offset + done); + if (n < 0 && mtcp_sys_errno == EINTR) continue; + if (n <= 0) return job->result = 1; + done += n; + } + return job->result = 0; +} + +static int +restart_read_child(void *raw) +{ + int mtcp_sys_errno; + RestartReadJob *job = (RestartReadJob *) raw; + // A canceled restore must not leave a helper holding its address space. + // Recheck the parent after arming the signal to close the creation race. + if (mtcp_inline_syscall(prctl, 5, PR_SET_PDEATHSIG, SIGKILL, 0, 0, 0) < 0 || + mtcp_sys_getppid() != job->parent) return job->result = 1; + return restart_read_job(raw); +} +#endif + +static int +restart_parallel_read(RestoreInfo *rinfo, int fd, void *address, size_t size) +{ +#if defined(__x86_64__) + int mtcp_sys_errno; + if (rinfo->restartReaders <= 1 || !rinfo->restartImageRegular || + size < 1024UL * 1024 * 1024) return 0; + off_t offset = mtcp_sys_lseek(fd, 0, SEEK_CUR); + if (offset < 0) return 0; + const size_t stackSize = 64 * 1024; + int readers = rinfo->restartReaders; + char *stacks = mtcp_sys_mmap(NULL, stackSize * (readers - 1), + PROT_READ | PROT_WRITE, + MAP_PRIVATE | MAP_ANONYMOUS, -1, 0); + if (stacks == MAP_FAILED) return 0; + RestartReadJob jobs[16]; + long pids[16]; + const size_t huge = 2 * 1024 * 1024; + size_t stride = ((size + readers - 1) / readers + huge - 1) & ~(huge - 1); + pid_t parent = mtcp_sys_getpid(); + for (int i = 0; i < readers; i++) { + size_t start = stride * i; + size_t length = size > start ? size - start : 0; + if (length > stride) length = stride; + jobs[i] = (RestartReadJob){fd, (char *) address + start, + length, offset + start, -1, parent}; + pids[i] = i + 1 < readers ? + mtcp_read_clone(restart_read_child, stacks + stackSize * (i + 1), &jobs[i]) : -1; + } + // Failed clone calls fall back to doing that helper's range in the parent. + for (int i = 0; i < readers; i++) { + if (pids[i] < 0) restart_read_job(&jobs[i]); + } + int ok = 1; + for (int i = 0; i < readers; i++) { + if (pids[i] > 0) { + int status = 0; + struct rusage usage; + long waited; + do { + waited = mtcp_sys_wait4(pids[i], &status, 0, &usage); + } while (waited < 0 && mtcp_sys_errno == EINTR); + // Never unmap a stack or resume while a helper might still be alive. + MTCP_ASSERT(waited == pids[i]); + if (status != 0) ok = 0; + rinfo->profileChildMinorFaults += usage.ru_minflt; + rinfo->profileChildMajorFaults += usage.ru_majflt; + } + if (jobs[i].result != 0) ok = 0; + } + MTCP_ASSERT(mtcp_sys_munmap(stacks, stackSize * (readers - 1)) == 0); + if (!ok) return 0; // pread did not move fd: the ordinary reader can retry. + MTCP_ASSERT(mtcp_sys_lseek(fd, offset + size, SEEK_SET) == offset + size); + rinfo->profileParallelBytes += size; + return 1; +#else + return 0; +#endif +} + void mtcp_readmemoryareas(RestoreInfo *rinfo, int fd, VA endOfStack) { + int mtcp_sys_errno; + struct rusage before, after; + uint64_t start = 0; + int usage_ok = 0; + rinfo->profileReadBytes = rinfo->profileReadUs = rinfo->profileMapUs = 0; + rinfo->profileAreas = rinfo->profileHugeAdvice = rinfo->profileHugeFailed = 0; + rinfo->profileFilemapBytes = rinfo->profileFilemapUs = 0; + rinfo->profileParallelBytes = rinfo->profileChildMinorFaults = 0; + rinfo->profileChildMajorFaults = 0; + rinfo->mmapParentStart = rinfo->mmapParentEnd = 0; + struct stat imageStat; + rinfo->restartImageRegular = (rinfo->restartMmap || rinfo->restartReaders > 1) && + mtcp_sys_fstat(fd, &imageStat) == 0 && S_ISREG(imageStat.st_mode); + if (rinfo->profileRestart) { + start = restart_time_us(); + usage_ok = mtcp_inline_syscall(getrusage, 2, RUSAGE_SELF, &before) == 0; + } while (1) { if (read_one_memory_area(rinfo, fd, endOfStack) == -1) { break; /* error */ } } + if (rinfo->profileRestart) { + uint64_t elapsed = restart_time_us() - start; + usage_ok = usage_ok && + mtcp_inline_syscall(getrusage, 2, RUSAGE_SELF, &after) == 0; + MTCP_PRINTF("DMTCP_PROFILE pid=%d virtual_pid=%d phase=memory total_us=%u read_us=%u " + "map_us=%u read_kib=%u areas=%u minflt=%u majflt=%u " + "usage_ok=%u huge_advice=%u huge_failed=%u " + "filemap_kib=%u filemap_us=%u readers=%u parallel_kib=%u " + "child_minflt=%u child_majflt=%u\n", + mtcp_sys_getpid(), rinfo->ckptHdr.pid, (unsigned) elapsed, + (unsigned) rinfo->profileReadUs, (unsigned) rinfo->profileMapUs, + (unsigned) (rinfo->profileReadBytes / 1024), rinfo->profileAreas, + usage_ok ? (unsigned) (after.ru_minflt - before.ru_minflt) : 0, + usage_ok ? (unsigned) (after.ru_majflt - before.ru_majflt) : 0, + usage_ok, rinfo->profileHugeAdvice, rinfo->profileHugeFailed, + (unsigned) (rinfo->profileFilemapBytes / 1024), + (unsigned) rinfo->profileFilemapUs, rinfo->restartReaders, + (unsigned) (rinfo->profileParallelBytes / 1024), + (unsigned) rinfo->profileChildMinorFaults, + (unsigned) rinfo->profileChildMajorFaults); + } #if defined(__arm__) || defined(__aarch64__) /* On ARM, with gzip enabled, we sometimes see SEGFAULT without this. @@ -844,6 +1043,44 @@ mtcp_readmemoryareas(RestoreInfo *rinfo, int fd, VA endOfStack) #endif /* if defined(__arm__) || defined(__aarch64__) */ } +// Maps again a file area the checkpoint did not save (DMTCP_FILE_MAPPED). +// The file must be the one checkpointed, same size and mtime: another would +// silently put other code or data at these addresses, so the restart stops. +NO_OPTIMIZE +static void +restore_file_mapped_area(RestoreInfo *rinfo, Area *area) +{ + int mtcp_sys_errno; + struct stat st; + + int fd = mtcp_sys_open(area->name, O_RDONLY, 0); + if (fd < 0) { + MTCP_PRINTF("cannot open %s, which the image maps at %p: error %d\n", + area->name, area->addr, mtcp_sys_errno); + mtcp_abort(); + } + if (mtcp_sys_fstat(fd, &st) < 0) { + MTCP_PRINTF("cannot stat %s: error %d\n", area->name, mtcp_sys_errno); + mtcp_abort(); + } + if ((uint64_t)st.st_size != area->fileSize || + st.st_mtim.tv_sec != area->fileMtimeSec || + st.st_mtim.tv_nsec != area->fileMtimeNsec) { + MTCP_PRINTF("%s is not the file checkpointed: size %p mtime %p.%p, " + "checkpointed size %p mtime %p.%p\n", area->name, + (void *)st.st_size, (void *)st.st_mtim.tv_sec, + (void *)st.st_mtim.tv_nsec, (void *)area->fileSize, + (void *)area->fileMtimeSec, (void *)area->fileMtimeNsec); + mtcp_abort(); + } + DPRINTF(rinfo, "mapping %p bytes at %p from %s + %p\n", + area->size, area->addr, area->name, area->offset); + void *addr = mmap_fixed_noreplace(rinfo, area->addr, area->size, area->prot, + MAP_PRIVATE, fd, area->offset); + MTCP_ASSERT(addr == area->addr); + mtcp_sys_close(fd); +} + NO_OPTIMIZE static int read_one_memory_area(RestoreInfo *rinfo, int fd, VA endOfStack) @@ -859,6 +1096,12 @@ read_one_memory_area(RestoreInfo *rinfo, int fd, VA endOfStack) if (area.addr == NULL) { return -1; } + rinfo->profileAreas++; + + if (area.properties & DMTCP_FILE_MAPPED) { + restore_file_mapped_area(rinfo, &area); + return 0; + } if (area.name[0] && mtcp_strstr(area.name, "[heap]") && mtcp_sys_brk(NULL) != area.addr + area.size) { @@ -1011,12 +1254,33 @@ read_one_memory_area(RestoreInfo *rinfo, int fd, VA endOfStack) } MTCP_ASSERT(imagefd == -1); // anonymous memory region } + uint64_t mapStart = rinfo->profileRestart ? restart_time_us() : 0; mmappedat = mmap_fixed_noreplace(rinfo, area.addr, area.size, area.prot | PROT_WRITE, mmapFlags, imagefd, area.offset); MTCP_ASSERT(mmappedat == area.addr); + if (area.properties & DMTCP_ZERO_PAGE_PARENT_HEADER) { + rinfo->mmapParentStart = (uint64_t) area.addr; + rinfo->mmapParentEnd = (uint64_t) area.endAddr; + } + if (rinfo->profileRestart) { + rinfo->profileMapUs += restart_time_us() - mapStart; + } + + // Opt-in experiment: first-touch allocation in large private anonymous + // regions. No MAP_POPULATE, no eager touching of zero runs, no change to + // addresses or checkpoint bytes. Unsupported advice leaves the old path. + if (rinfo->restartHugePages && (mmapFlags & MAP_ANONYMOUS) && + (mmapFlags & MAP_PRIVATE) && !(mmapFlags & MAP_GROWSDOWN) && + area.size >= 2 * 1024 * 1024) { + rinfo->profileHugeAdvice++; + if (mtcp_inline_syscall(madvise, 3, area.addr, area.size, + MADV_HUGEPAGE) < 0) { + rinfo->profileHugeFailed++; + } + } #if 0 /* @@ -1043,12 +1307,70 @@ read_one_memory_area(RestoreInfo *rinfo, int fd, VA endOfStack) */ /* ANALYZE THE CONDITION FOR DOING mmapfile MORE CAREFULLY. */ + uint64_t readStart = rinfo->profileRestart ? restart_time_us() : 0; + size_t readSize = area.size; if (area.mmapFileSize > 0 && area.name[0] == '/') { DPRINTF(rinfo, "restoring memory region %p of %p bytes at %p\n", area.mmapFileSize, area.size, area.addr); - mtcp_readfile(fd, area.addr, area.mmapFileSize); - } else { - mtcp_readfile(fd, area.addr, area.size); + readSize = area.mmapFileSize; + } + int mappedImage = 0; + int mmapEligible = 0; + int imageOffsetMod = -1; + int mmapError = 0; + // Experimental mmap path for LARGE raw anonymous payloads only. The + // parent header has already reserved these addresses, including zero + // children. Replace only our own reservation, never an unrelated VMA. + // MAP_PRIVATE preserves the immutable snapshot when the process writes. + // Leave pipes/compression, stacks, partial file-backed payloads and small + // mappings on the ordinary reader. Do not confuse a parent header with + // payload data (the old compile-time FAST_RST_VIA_MMAP path does). + int child = area.properties & DMTCP_ZERO_PAGE_CHILD_HEADER; + int owned = !child || + ((uint64_t) area.addr >= rinfo->mmapParentStart && + (uint64_t) area.endAddr <= rinfo->mmapParentEnd); + if (rinfo->restartMmap && rinfo->restartImageRegular && owned && + (area.flags & MAP_ANONYMOUS) && (area.flags & MAP_PRIVATE) && + !(area.flags & MAP_GROWSDOWN) && area.name[0] != '/' && + readSize == area.size && readSize >= 64 * 1024 * 1024) { + mmapEligible = 1; + off_t offset = mtcp_sys_lseek(fd, 0, SEEK_CUR); + if (offset >= 0) imageOffsetMod = offset % MTCP_PAGE_SIZE; + if (offset >= 0 && offset % MTCP_PAGE_SIZE == 0) { + void *mapped = mtcp_sys_mmap(area.addr, readSize, area.prot, + MAP_PRIVATE | MAP_FIXED, fd, offset); + if (mapped != MAP_FAILED) { + MTCP_ASSERT(mapped == area.addr); + MTCP_ASSERT(mtcp_sys_lseek(fd, readSize, SEEK_CUR) >= 0); + mappedImage = 1; + } else { + mmapError = mtcp_sys_errno; + } + } + } + int parallelRead = !mappedImage && + restart_parallel_read(rinfo, fd, area.addr, readSize); + if (!mappedImage && !parallelRead) { + mtcp_readfile(fd, area.addr, readSize); + } + if (rinfo->profileRestart) { + uint64_t us = restart_time_us() - readStart; + if (mappedImage) { + rinfo->profileFilemapUs += us; + rinfo->profileFilemapBytes += readSize; + } else { + rinfo->profileReadUs += us; + rinfo->profileReadBytes += readSize; + } + if (readSize >= 64 * 1024 * 1024 || us >= 10000) { + MTCP_PRINTF("DMTCP_PROFILE pid=%d virtual_pid=%d phase=area addr=%p read_kib=%u " + "read_us=%u flags=%x properties=%x mode=%s " + "mmap_eligible=%u offset_mod=%d mmap_errno=%d name=%s\n", + mtcp_sys_getpid(), rinfo->ckptHdr.pid, area.addr, (unsigned) (readSize / 1024), + (unsigned) us, area.flags, area.properties, + mappedImage ? "mmap" : parallelRead ? "pread" : "read", mmapEligible, + imageOffsetMod, mmapError, area.name); + } } if (!(area.prot & PROT_WRITE)) { diff --git a/src/mtcp/mtcp_restart.h b/src/mtcp/mtcp_restart.h index bdd707000a..24ed7f591e 100644 --- a/src/mtcp/mtcp_restart.h +++ b/src/mtcp/mtcp_restart.h @@ -134,6 +134,27 @@ typedef struct RestoreInfo { DmtcpCkptHeader ckptHdr; char ckptImage[PATH_MAX]; + + // Restart-only diagnostics/options; not part of the checkpoint format. + // Keep these in the relocated RestoreInfo, never in unmapped globals. + int profileRestart; + int restartHugePages; + uint64_t profileReadBytes; + uint64_t profileReadUs; + uint64_t profileMapUs; + unsigned profileAreas; + unsigned profileHugeAdvice; + unsigned profileHugeFailed; + int restartMmap; + int restartImageRegular; + uint64_t mmapParentStart; + uint64_t mmapParentEnd; + uint64_t profileFilemapBytes; + uint64_t profileFilemapUs; + int restartReaders; + uint64_t profileParallelBytes; + uint64_t profileChildMinorFaults; + uint64_t profileChildMajorFaults; } RestoreInfo; // int mtcp_restart_process_args(RestoreInfo *rinfo, int argc, char *argv[], char **environ); diff --git a/src/mtcp/mtcp_sys.h b/src/mtcp/mtcp_sys.h index 12a0aac318..09b9068c58 100644 --- a/src/mtcp/mtcp_sys.h +++ b/src/mtcp/mtcp_sys.h @@ -253,6 +253,9 @@ struct linux_dirent { # define mtcp_sys_open2(args ...) mtcp_sys_open(args, 0777) # define mtcp_sys_ftruncate(args ...) mtcp_inline_syscall(ftruncate, 2, args) # define mtcp_sys_close(args ...) mtcp_inline_syscall(close, 1, args) +// fstat() through newfstatat(), which every architecture has. +# define mtcp_sys_fstat(fd, statbuf) \ + mtcp_inline_syscall(newfstatat, 4, fd, "", statbuf, AT_EMPTY_PATH) # if defined(__aarch64__) # define mtcp_sys_access(path, mode) \ mtcp_inline_syscall(faccessat, 4, AT_FDCWD, path, mode, AT_EACCESS) diff --git a/src/plugin/connectionlist.h b/src/plugin/connectionlist.h index f0d5b078ce..c710718c33 100644 --- a/src/plugin/connectionlist.h +++ b/src/plugin/connectionlist.h @@ -67,6 +67,22 @@ class ConnectionList void erase(ConnectionIdentifier &key); Connection *getConnection(const ConnectionIdentifier &id); Connection *getConnection(int fd); + + // Calls fn with fd's connection, or NULL, with the table locked, so that + // no other thread deletes the connection meanwhile: add() drops the + // entry of an fd that was closed past the close wrapper (raw syscalls, + // glibc's internal close) when the number comes back, and processClose() + // drops a closed one. Callers hold the WrapperLock, so that the + // checkpoint never suspends them with the table locked. fn must not call + // back into the table. + template + void withConnection(int fd, Fn fn) + { + _lock_tbl(); + fn(_fdToCon.contains(fd) ? _fdToCon[fd] : NULL); + _unlock_tbl(); + } + void processClose(int fd); void processDup(int oldfd, int newfd); void list(); diff --git a/src/plugin/dl/dlwrappers.cpp b/src/plugin/dl/dlwrappers.cpp index 281dce500d..0dab616cd8 100644 --- a/src/plugin/dl/dlwrappers.cpp +++ b/src/plugin/dl/dlwrappers.cpp @@ -171,6 +171,28 @@ dlopen_try_paths(const char *filename, int flags, const char *paths) return NULL; } +// A filename with a slash is a path, which dlopen(3) opens without a search. glibc +// expands $ORIGIN in it to the directory of the object that called dlopen, which +// here would be this library: expand it for the wrapper's caller instead. A +// library that loads its own parts as "$ORIGIN/libpart.so" (NVIDIA's cuInterpose +// does) otherwise finds nothing under DMTCP. +static void * +dlopen_path(const char *filename, int flags, void *caller) +{ + if (strstr(filename, "$ORIGIN") == NULL && + strstr(filename, "${ORIGIN}") == NULL) { + return _real_dlopen(filename, flags); + } + + Dl_info info; + ASSERT_NE(0, dladdr(caller, &info)); + char dirname[4096]; + jalib::Filesystem::DirName(dirname, info.dli_fname); + char path[4096]; + copyDynamicPath(path, sizeof(path), filename, dirname, "dlopen filename"); + return _real_dlopen(path, flags); +} + static void * dmtcp_dlopen_with_search_policy(const char *filename, int flags, void *caller) { @@ -178,6 +200,10 @@ dmtcp_dlopen_with_search_policy(const char *filename, int flags, void *caller) return _real_dlopen(filename, flags); } + if (strchr(filename, '/') != NULL) { + return dlopen_path(filename, flags, caller); + } + char rpath[4096]; char runpath[4096]; void *ret = NULL; diff --git a/src/plugin/file/fileconnection.cpp b/src/plugin/file/fileconnection.cpp index c020d3ae24..bfac6fd035 100644 --- a/src/plugin/file/fileconnection.cpp +++ b/src/plugin/file/fileconnection.cpp @@ -527,6 +527,13 @@ FileConnection::postRestart() TRACE("Restoring file connection: con_id={} path={}", id().toString(), _path); + // The copy was saved next to the image, which may have moved since the + // checkpoint: look for it next to the image this process restarts from. + string moved = string(dmtcp_get_ckpt_files_subdir()) + "/" + + jalib::Filesystem::BaseName(_savedFilePath); + if (jalib::Filesystem::FileExists(moved)) { + _savedFilePath = moved; + } ASSERT(jalib::Filesystem::FileExists(_savedFilePath), "Unable to find checkpointed copy of file: saved={} path={}", _savedFilePath, _path); diff --git a/src/plugin/file/fileconnlist.cpp b/src/plugin/file/fileconnlist.cpp index 7614c1e2f0..dad5c2928d 100644 --- a/src/plugin/file/fileconnlist.cpp +++ b/src/plugin/file/fileconnlist.cpp @@ -370,8 +370,11 @@ FileConnList::resume(bool isRestart) if (isRestart) { // Now unlink the files that we created as a side-effect of restoreShmArea. + // Processes that share an area all recreated the same file in refill, so + // every one but the first finds it already unlinked here. for (size_t i = 0; i < missingUnlinkedShmFiles.size(); i++) { - WARN_NE(-1, unlink(missingUnlinkedShmFiles[i].name), + int rc = unlink(missingUnlinkedShmFiles[i].name); + WARN_ERRNO(rc == 0 || errno == ENOENT, "The file was unlinked at checkpoint, but unlinking it after restart " "failed: path={}", missingUnlinkedShmFiles[i].name); diff --git a/src/plugin/socket/connectionrewirer.cpp b/src/plugin/socket/connectionrewirer.cpp index d9fb70934d..92ad6ed11a 100644 --- a/src/plugin/socket/connectionrewirer.cpp +++ b/src/plugin/socket/connectionrewirer.cpp @@ -130,16 +130,27 @@ ConnectionRewirer::doReconnect() Connection *con = i->second; struct RemoteAddr &remoteAddr = _remoteInfo[id]; int fd = con->getFds()[0]; + + // A socket created with SOCK_NONBLOCK would return EINPROGRESS: connect + // and send the identifier blocking, then put its flags back. + int flags = _real_fcntl(fd, F_GETFL, NULL); + ASSERT_NE(-1, flags, "fcntl(F_GETFL) failed: fd={}", fd); + if (flags & O_NONBLOCK) { + markSocketBlocking(fd); + } errno = 0; - ASSERT_NE(-1, - _real_connect(fd, - (sockaddr *)&remoteAddr.addr, - remoteAddr.len), - "failed to restore connection: fd={} host={} pid={} time={} " - "con_id={}", - fd, id.hostid(), id.pid(), id.time(), id.conId()); + ASSERT_ERRNO(_real_connect(fd, + (sockaddr *)&remoteAddr.addr, + remoteAddr.len) != -1, + "failed to restore connection: fd={} family={} host={} " + "pid={} time={} con_id={}", + fd, remoteAddr.addr.ss_family, id.hostid(), id.pid(), + id.time(), id.conId()); Util::writeAll(fd, &id, sizeof id); + if (flags & O_NONBLOCK) { + markSocketNonBlocking(fd); + } checkForPendingIncoming(PROTECTED_RESTORE_IP4_SOCK_FD, &_pendingIP4Incoming); @@ -309,6 +320,16 @@ ConnectionRewirer::openRestoreSocket(bool hasIPv4Sock, } } +int +ConnectionRewirer::restoreDomain(int domain) +{ +#ifdef ENABLE_IP6_SUPPORT + return domain; +#else // ifdef ENABLE_IP6_SUPPORT + return domain == AF_INET6 ? AF_INET : domain; +#endif // ifdef ENABLE_IP6_SUPPORT +} + void ConnectionRewirer::registerIncoming(const ConnectionIdentifier &local, Connection *con, @@ -319,14 +340,11 @@ ConnectionRewirer::registerIncoming(const ConnectionIdentifier &local, SocketConnection *sc = dynamic_cast(con); ASSERT_NOT_NULL(sc); + domain = restoreDomain(domain); if (domain == AF_INET) { _pendingIP4Incoming[local] = con; } else if (domain == AF_INET6) { -#ifdef ENABLE_IP6_SUPPORT _pendingIP6Incoming[local] = con; -#else // ifdef ENABLE_IP6_SUPPORT - _pendingIP4Incoming[local] = con; -#endif // ifdef ENABLE_IP6_SUPPORT } else if (domain == AF_UNIX) { // Route AF_UNIX by base type: stream vs seqpacket if (sc->baseType() == SOCK_SEQPACKET) { @@ -371,6 +389,10 @@ ConnectionRewirer::registerNSData(void *addr, iterator i; ASSERT_NOT_NULL(theRewirer); + // A peer would connect to a zeroed address and fail far from the cause. + ASSERT(conList->empty() || ((sockaddr *)addr)->sa_family != AF_UNSPEC, + "no restore listener for {} incoming connections", + conList->size()); for (i = conList->begin(); i != conList->end(); ++i) { const ConnectionIdentifier &id = i->first; string addrStr = dmtcp::base64::encode((const char*) addr, addrLen); diff --git a/src/plugin/socket/connectionrewirer.h b/src/plugin/socket/connectionrewirer.h index 7c67160ccd..2f3ce07bfc 100644 --- a/src/plugin/socket/connectionrewirer.h +++ b/src/plugin/socket/connectionrewirer.h @@ -46,6 +46,10 @@ class ConnectionRewirer static ConnectionRewirer &instance(); static void destroy(); + // The address family whose restore listener rewires a socket of this + // domain: without IPv6 support, IPv6 connections go through IPv4. + static int restoreDomain(int domain); + void openRestoreSocket(bool hasIPv4, bool hasIPv6, bool hasUNIX); void registerIncoming(const ConnectionIdentifier &local, Connection *con, diff --git a/src/plugin/socket/socketconnection.cpp b/src/plugin/socket/socketconnection.cpp index 44feb18e8c..ae8d6d0b80 100644 --- a/src/plugin/socket/socketconnection.cpp +++ b/src/plugin/socket/socketconnection.cpp @@ -140,6 +140,37 @@ SocketConnection::addSetsockopt(int level, } } +// Options that change what bind() accepts must be set before it: IPV6_V6ONLY +// decides whether an IPv6 socket also takes IPv4, and SO_REUSEADDR and +// SO_REUSEPORT whether the address may be bound while connections of the +// checkpointed process linger in TIME_WAIT (the usual case right after the +// original was killed). The rest come back after bind, in the order of +// restoreSocketOptions(). +void +SocketConnection::restoreBindOptions(int fd) +{ + for (const auto &lvl : _sockOptions) { + for (const auto &opt : lvl.second) { + bool beforeBind = + (lvl.first == SOL_SOCKET && + (opt.first == SO_REUSEADDR || opt.first == SO_REUSEPORT)) || + (lvl.first == IPPROTO_IPV6 && opt.first == IPV6_V6ONLY && + _sockDomain == AF_INET6); + if (!beforeBind) { + continue; + } + TRACE("Restoring socket option before bind: fd={} level={} option={}", + fd, lvl.first, opt.first); + int ret = _real_setsockopt(fd, lvl.first, opt.first, + opt.second.data(), opt.second.size()); + ASSERT_ERRNO(ret == 0, + "restoring socket option before bind failed: fd={} " + "level={} option={}", + fd, lvl.first, opt.first); + } + } +} + void SocketConnection::restoreSocketOptions(vector &fds) { @@ -801,46 +832,19 @@ TcpConnection::postRestart() * --Kapil */ - if (_sockDomain == AF_INET6) { - TRACE("Restoring IPv6 socket options before bind: fd={}", _fds[0]); - typedef map > >::iterator levelIterator; - typedef map >::iterator optionIterator; - - for (levelIterator lvl = _sockOptions.begin(); - lvl != _sockOptions.end(); ++lvl) { - if (lvl->first == IPPROTO_IPV6) { - for (optionIterator opt = lvl->second.begin(); - opt != lvl->second.end(); ++opt) { - if (opt->first == IPV6_V6ONLY) { - if (really_verbose) { - TRACE("Restoring socket option: fd={} option={} size={}", - _fds[0], opt->first, opt->second.size()); - } - int ret = _real_setsockopt(_fds[0], lvl->first, opt->first, - opt->second.data(), - opt->second.size()); - ASSERT_ERRNO(ret == 0, - "Restoring IPV6_V6ONLY setsockopt failed: fd={} " - "level={} option={} value={} size={}", - _fds[0], lvl->first, opt->first, - opt->second.data(), opt->second.size()); - } - } - } - } - } + restoreBindOptions(_fds[0]); if (really_verbose) { TRACE("Binding recreated TCP socket: con_id={} fd={}", id().toString(), _fds[0]); } errno = 0; - WARN_NE(-1, _real_bind(_fds[0], - (sockaddr *)&_bindAddr, - _bindAddrlen), - "Bind failed: fd={} con_id={} addrlen={}", - _fds[0], id().conId(), _bindAddrlen); + WARN_ERRNO(_real_bind(_fds[0], (sockaddr *)&_bindAddr, _bindAddrlen) != -1, + "Bind failed: fd={} con_id={} family={} port={} addrlen={}", + _fds[0], id().conId(), _bindAddr.ss_family, + _sockDomain == AF_UNIX + ? 0 : ntohs(((sockaddr_in *)&_bindAddr)->sin_port), + _bindAddrlen); if (_type == TCP_BIND) { break; } diff --git a/src/plugin/socket/socketconnection.h b/src/plugin/socket/socketconnection.h index f1bf721f04..aa923e7eca 100644 --- a/src/plugin/socket/socketconnection.h +++ b/src/plugin/socket/socketconnection.h @@ -57,6 +57,7 @@ class SocketConnection ConnectionIdentifier remote); void addSetsockopt(int level, int option, const void *value, int len); void restoreSocketOptions(vector &fds); + void restoreBindOptions(int fd); void serialize(jalib::JBinarySerializer &o); int sockDomain() const { return _sockDomain; } int sockType() const { return _sockType; } diff --git a/src/plugin/socket/socketconnlist.cpp b/src/plugin/socket/socketconnlist.cpp index 5b1b4cbdd6..9d92d66c42 100644 --- a/src/plugin/socket/socketconnlist.cpp +++ b/src/plugin/socket/socketconnlist.cpp @@ -190,7 +190,8 @@ SocketConnList::preCkpt() for (iterator i = begin(); i != end(); ++i) { Connection *con = i->second; if (con->hasLock() && con->conType() == Connection::TCP) { - int domain = ((TcpConnection *)con)->sockDomain(); + int domain = ConnectionRewirer::restoreDomain( + ((TcpConnection *)con)->sockDomain()); if (domain == AF_INET) { _hasIPv4Sock = true; } else if (domain == AF_INET6) { diff --git a/src/plugin/socket/socketwrappers.cpp b/src/plugin/socket/socketwrappers.cpp index 3f4d0d0ead..120bd6ed45 100644 --- a/src/plugin/socket/socketwrappers.cpp +++ b/src/plugin/socket/socketwrappers.cpp @@ -93,14 +93,15 @@ connect(int sockfd, const struct sockaddr *serv_addr, socklen_t addrlen) if ((ret != -1 || errno == EINPROGRESS) && dmtcp_is_running_state() && !_doNotProcessSockets) { - SocketConnection *con = - dynamic_cast(SocketConnList::instance().getConnection( - sockfd)); - if (con == NULL) { - TRACE("Called connect() on untracked socket type: fd={}", sockfd); - } else { - con->onConnect(serv_addr, addrlen, (ret == -1 && errno == EINPROGRESS)); - } + bool inProgress = ret == -1 && errno == EINPROGRESS; + SocketConnList::instance().withConnection(sockfd, [&](Connection *c) { + SocketConnection *con = dynamic_cast(c); + if (con == NULL) { + TRACE("Called connect() on untracked socket type: fd={}", sockfd); + } else { + con->onConnect(serv_addr, addrlen, inProgress); + } + }); } errno = savedErrno; @@ -117,14 +118,14 @@ bind(int sockfd, const struct sockaddr *my_addr, socklen_t addrlen) WrapperLock wrapperLock; int ret = _real_bind(sockfd, my_addr, addrlen); if (ret != -1 && dmtcp_is_running_state() && !_doNotProcessSockets) { - SocketConnection *con = - dynamic_cast(SocketConnList::instance().getConnection( - sockfd)); - if (con == NULL) { - TRACE("Called bind() on untracked socket type: fd={}", sockfd); - } else { - con->onBind((struct sockaddr *)my_addr, addrlen); - } + SocketConnList::instance().withConnection(sockfd, [&](Connection *c) { + SocketConnection *con = dynamic_cast(c); + if (con == NULL) { + TRACE("Called bind() on untracked socket type: fd={}", sockfd); + } else { + con->onBind((struct sockaddr *)my_addr, addrlen); + } + }); } return ret; } @@ -139,14 +140,14 @@ listen(int sockfd, int backlog) WrapperLock wrapperLock; int ret = _real_listen(sockfd, backlog); if (ret != -1 && dmtcp_is_running_state() && !_doNotProcessSockets) { - SocketConnection *con = - dynamic_cast(SocketConnList::instance().getConnection( - sockfd)); - if (con == NULL) { - TRACE("Called listen() on untracked socket type: fd={}", sockfd); - } else { - con->onListen(backlog); - } + SocketConnList::instance().withConnection(sockfd, [&](Connection *c) { + SocketConnection *con = dynamic_cast(c); + if (con == NULL) { + TRACE("Called listen() on untracked socket type: fd={}", sockfd); + } else { + con->onListen(backlog); + } + }); } return ret; } @@ -261,20 +262,24 @@ setsockopt(int sockfd, return _real_setsockopt(sockfd, level, optname, optval, optlen); } + // Threads that set options while another closes and reopens descriptors + // (cuInterpose's, after a restart) must not reach a connection the table + // drops meanwhile: the lock keeps the checkpoint off, withConnection the + // table's own changes. + WrapperLock wrapperLock; int ret = _real_setsockopt(sockfd, level, optname, optval, optlen); if (ret != -1 && dmtcp_is_running_state() && !_doNotProcessSockets) { TRACE("Recording socket option: fd={} level={} option={} size={}", sockfd, level, optname, optlen); - SocketConnection *con = - dynamic_cast(SocketConnList::instance().getConnection( - sockfd)); - if (con == NULL) { - TRACE("Called setsockopt() on untracked socket type: fd={}", sockfd); - return ret; - } else { - con->addSetsockopt(level, optname, optval, optlen); - } + SocketConnList::instance().withConnection(sockfd, [&](Connection *c) { + SocketConnection *con = dynamic_cast(c); + if (con == NULL) { + TRACE("Called setsockopt() on untracked socket type: fd={}", sockfd); + } else { + con->addSetsockopt(level, optname, optval, optlen); + } + }); } return ret; } diff --git a/src/plugin/svipc/sysvipcwrappers.cpp b/src/plugin/svipc/sysvipcwrappers.cpp index 71011896a4..dea66c7548 100644 --- a/src/plugin/svipc/sysvipcwrappers.cpp +++ b/src/plugin/svipc/sysvipcwrappers.cpp @@ -258,6 +258,21 @@ shmget(key_t key, size_t size, int shmflg) return virtId; } +// A segment created through a path that bypasses shmget's wrapper has no +// virtual id: UCX's memory hooks call the original shmget through +// dlsym(RTLD_NEXT) to test shmat events, then remove the segment with the +// wrapped shmctl. Its id is the kernel's; use it as is, untracked. +static int +realShmId(int shmid) +{ + int realShmid = VIRTUAL_TO_REAL_SHM_ID(shmid); + if (realShmid == -1) { + TRACE("SysV shm id not issued by DMTCP, passed through: shmid={}", shmid); + return shmid; + } + return realShmid; +} + extern "C" void *shmat(int shmid, const void *shmaddr, int shmflg) { @@ -266,9 +281,7 @@ void *shmat(int shmid, const void *shmaddr, int shmflg) } WrapperLock wrapperLock; - int realShmid = VIRTUAL_TO_REAL_SHM_ID(shmid); - ASSERT(realShmid != -1, "shmat virtual id has no real mapping: shmid={}", - shmid); + int realShmid = realShmId(shmid); void *ret = _real_shmat(realShmid, shmaddr, shmflg); #ifdef __arm__ @@ -390,9 +403,7 @@ shmctl(int shmid, int cmd, struct shmid_ds *buf) break; } - int realShmid = VIRTUAL_TO_REAL_SHM_ID(shmid); - ASSERT(realShmid != -1, "shmctl virtual id has no real mapping: shmid={}", - shmid); + int realShmid = realShmId(shmid); int ret = _real_shmctl(realShmid, cmd, buf); if (ret != -1 && buf != NULL && cmd == IPC_STAT) { buf->shm_cpid = dmtcp_pid_real_to_virtual(buf->shm_cpid); diff --git a/src/processinfo.cpp b/src/processinfo.cpp index b056724e10..67d731e3ee 100644 --- a/src/processinfo.cpp +++ b/src/processinfo.cpp @@ -583,6 +583,19 @@ restoreArgvEnvBoundsBestEffort(MemRegion argBounds, MemRegion envBounds) } } +// The checkpoint directory becomes the one this process restarts from, which +// may not be where it was checkpointed (images copied to another host or +// directory). It runs before any plugin's restart hook, so that the plugins +// find what was saved next to the image (dmtcp_get_ckpt_files_subdir()). +void +ProcessInfo::restartCkptDir() +{ + string ckptDir = jalib::Filesystem::GetDeviceName(PROTECTED_CKPT_DIR_FD); + ASSERT(ckptDir.length() > 0, "restart checkpoint directory is empty"); + _real_close(PROTECTED_CKPT_DIR_FD); + updateCkptDirFileSubdir(ckptDir); +} + void ProcessInfo::restart() { @@ -592,12 +605,6 @@ ProcessInfo::restart() restoreArgvEnvBoundsBestEffort(argBounds, envBounds); - // Update the ckptDir - string ckptDir = jalib::Filesystem::GetDeviceName(PROTECTED_CKPT_DIR_FD); - ASSERT(ckptDir.length() > 0, "restart checkpoint directory is empty"); - _real_close(PROTECTED_CKPT_DIR_FD); - updateCkptDirFileSubdir(ckptDir); - if (_launchCWD != _ckptCWD) { string rpath = ""; size_t llen = _launchCWD.length(); diff --git a/src/processinfo.h b/src/processinfo.h index 1be12e7a39..6b52e0ee63 100644 --- a/src/processinfo.h +++ b/src/processinfo.h @@ -46,6 +46,7 @@ class ProcessInfo : public DmtcpCkptHeader void resetOnFork(); void preCkpt(); void restart(); + void restartCkptDir(); void restoreProcessGroupInfo(); void restoreHeap(); void growStack(); @@ -87,7 +88,7 @@ class ProcessInfo : public DmtcpCkptHeader const string &upidStr() { if (_upidStr.empty()) { - _upidStr = UniquePid(upid).toString(); + _upidStr = getUpid().toString(); } return _upidStr; } diff --git a/src/util_misc.cpp b/src/util_misc.cpp index 7d8c2c0753..89bf12e294 100644 --- a/src/util_misc.cpp +++ b/src/util_misc.cpp @@ -65,6 +65,7 @@ struct pm_scan_arg { uint64_t return_mask; }; +# define PAGE_IS_FILE (1 << 2) # define PAGE_IS_PRESENT (1 << 3) # define PAGE_IS_SWAPPED (1 << 4) # define PAGEMAP_SCAN _IOWR('f', 16, struct pm_scan_arg) @@ -835,6 +836,74 @@ Util::scanOccupiedRangeBatch(uintptr_t start, uintptr_t end, return !leading_occupied; } +// See the declaration in util.h. Allocation-free, so it is safe to call while +// writing a checkpoint image. +bool +Util::areaHasPrivatePages(uintptr_t start, uintptr_t end) +{ + const size_t page_size = pageSize(); + const uintptr_t pgMask = ~(uintptr_t)(page_size - 1); + const uintptr_t startAddr = start & pgMask; + const uintptr_t endAddr = (end + page_size - 1) & pgMask; + + if (startAddr >= endAddr) { + return false; + } + int fd = _real_open("/proc/self/pagemap", O_RDONLY); + if (fd < 0) { + return true; // can't inspect: the caller writes the data + } + + if (!readBooleanEnv(ENV_VAR_DISABLE_PAGEMAP_SCAN, false)) { + // A page that is present or swapped and not a file page is a private copy. + struct page_region region; + struct pm_scan_arg arg; + memset(&arg, 0, sizeof(arg)); + arg.size = sizeof(arg); + arg.start = startAddr; + arg.end = endAddr; + arg.vec = (uintptr_t)®ion; + arg.vec_len = 1; + arg.category_inverted = PAGE_IS_FILE; + arg.category_mask = PAGE_IS_FILE; + arg.category_anyof_mask = PAGE_IS_PRESENT | PAGE_IS_SWAPPED; + arg.return_mask = PAGE_IS_PRESENT | PAGE_IS_SWAPPED; + + long ret = ioctl(fd, PAGEMAP_SCAN, &arg); + if (ret >= 0) { + _real_close(fd); + // A walk that stopped short of the end decided nothing about the rest. + return ret > 0 || (arg.walk_end != 0 && arg.walk_end < endAddr); + } + } + + // Fallback: pagemap bit 63 is present, 62 swapped, 61 a file page. + uint64_t entries[PAGEMAP_BATCH_SIZE]; + uintptr_t current_addr = startAddr; + while (current_addr < endAddr) { + size_t pages = MIN((endAddr - current_addr) / page_size, + (size_t)PAGEMAP_BATCH_SIZE); + off_t offset = (off_t)(current_addr / page_size) * sizeof(uint64_t); + ssize_t bytes_read = pread(fd, entries, pages * sizeof(uint64_t), offset); + if (bytes_read <= 0 || bytes_read % sizeof(uint64_t) != 0) { + _real_close(fd); + return true; + } + for (size_t i = 0; i < bytes_read / sizeof(uint64_t); i++) { + const bool present = (entries[i] >> 63) & 1; + const bool swapped = (entries[i] >> 62) & 1; + const bool file = (entries[i] >> 61) & 1; + if (swapped || (present && !file)) { + _real_close(fd); + return true; + } + } + current_addr += (bytes_read / sizeof(uint64_t)) * page_size; + } + _real_close(fd); + return false; +} + // See the declaration in util.h. Allocation-free, so it is safe to call while // writing a checkpoint image. bool diff --git a/src/wrappers.cpp b/src/wrappers.cpp index 5a96714e2a..c755eb23d8 100644 --- a/src/wrappers.cpp +++ b/src/wrappers.cpp @@ -794,6 +794,25 @@ fcntl(int fd, int cmd, ...) return res; } +#ifdef __LP64__ +// glibc 2.28 and later compile fcntl into fcntl64 when _FILE_OFFSET_BITS is +// 64, as CPython is built. Without this wrapper a duplicate made through it +// (os.dup is F_DUPFD_CLOEXEC) was a descriptor DMTCP did not know: the child +// of multiprocessing's spawn keeps its parent sentinel only as such a +// duplicate, and the parent then drained its end of the pipe forever at +// checkpoint. On 64-bit targets fcntl64 is fcntl; like glibc's, this takes +// the argument as a pointer-sized value whatever the command. +extern "C" int +fcntl64(int fd, int cmd, ...) +{ + va_list ap; + va_start(ap, cmd); + void *arg = va_arg(ap, void *); + va_end(ap); + return fcntl(fd, cmd, arg); +} +#endif + extern "C" FILE * tmpfile() { diff --git a/src/writeckpt.cpp b/src/writeckpt.cpp index b1936cd217..7718cc6068 100644 --- a/src/writeckpt.cpp +++ b/src/writeckpt.cpp @@ -445,6 +445,31 @@ mtcp_write_anonymous_pages(int fd, Area area, bool residencyScanSafe) } } +// True if the area is the file's own pages, which the restart can map from +// the file instead of the image: a private mapping that cannot be written, +// within the file's pages, and with no page copied on write (a library's +// RELRO is read-only too, but relocated) or swapped out. Most of a process +// that loaded torch and the CUDA libraries is such text and read-only data. +static bool +fileMappable(const Area &area, const struct stat &st, bool statValid) +{ + if (!statValid || !S_ISREG(st.st_mode) || !(area.flags & MAP_PRIVATE) || + (area.flags & MAP_SHARED) || (area.prot & PROT_WRITE) || + Util::strEndsWith(area.name, DELETED_FILE_SUFFIX) || + Util::readBooleanEnv(ENV_VAR_DISABLE_FILE_MAPPED, false)) { + return false; + } + // Past the file's last page a mapping faults with SIGBUS: only saved data + // can stand in there. + const uint64_t page = Util::pageSize(); + const uint64_t filePages = ((uint64_t)st.st_size + page - 1) / page * page; + if ((uint64_t)area.offset + area.size > filePages) { + return false; + } + return !Util::areaHasPrivatePages((uintptr_t)area.addr, + (uintptr_t)area.endAddr); +} + static void writememoryarea(int fd, Area area) { @@ -679,6 +704,15 @@ writememoryarea(int fd, Area area) // Handle non-existing files (deleted file: not residency-safe, so the // content scan runs regardless of residencyScanSafe). mtcp_write_anonymous_pages(fd, area, residencyScanSafe); + } else if (fileMappable(area, areaStat, areaStatValid)) { + // No data: mtcp_restart maps the file again, once it checked it is this one. + area.properties |= DMTCP_FILE_MAPPED; + area.fileSize = areaStat.st_size; + area.fileMtimeSec = areaStat.st_mtim.tv_sec; + area.fileMtimeNsec = areaStat.st_mtim.tv_nsec; + TRACE("file area mapped on restart, not saved: addr={} size={} name={}", + (void *)area.addr, area.size, area.name); + writeAreaHeader(fd, &area); } else { ASSERT(strlen(area.name) > 0, "checkpoint file-backed area has an empty path: addr={} size={}", @@ -697,6 +731,17 @@ writememoryarea(int fd, Area area) off_t availBytes = areaStat.st_size - area.offset; if (availBytes > 0) { area.mmapFileSize = availBytes; + // A regular file's final partial page is readable up to its page + // boundary; only whole pages beyond EOF fault. Save that complete + // page, including any private dirty tail. Besides preserving those + // bytes, this keeps subsequent raw checkpoint payloads page-aligned + // so mtcp_restart can mmap them. Existing readers already honor + // mmapFileSize, so no checkpoint-format change is needed. + if (S_ISREG(areaStat.st_mode)) { + area.mmapFileSize = (availBytes + MTCP_PAGE_SIZE - 1) & MTCP_PAGE_MASK; + ASSERT_LE(area.mmapFileSize, area.size, + "rounded final file page exceeds mapping"); + } } else { area.mmapFileSize = 0; } diff --git a/test/Makefile.in b/test/Makefile.in index b8c281aae6..f1e97caeea 100644 --- a/test/Makefile.in +++ b/test/Makefile.in @@ -318,6 +318,14 @@ libdlopen-lib2.so: dlopen1.c dlopen1: dlopen1.c libdlopen-lib1.so libdlopen-lib2.so ${CC} $(CFLAGS) -o $@ $< -ldl +# dlopen-origin: libdlopen-origin-a.so loads libdlopen-origin-b.so through $ORIGIN +libdlopen-origin-a.so: dlopen-origin.c + ${CC} ${CFLAGS} -shared -fPIC -DLIB_A -o $@ $< -ldl +libdlopen-origin-b.so: dlopen-origin.c + ${CC} ${CFLAGS} -shared -fPIC -DLIB_B -o $@ $< +dlopen-origin: dlopen-origin.c libdlopen-origin-a.so libdlopen-origin-b.so + ${CC} $(CFLAGS) -o $@ $< -ldl + # dlopen2 will dlopen/dlclose libdlopen-lib[34].so libdlopen-lib3.so: dlopen2.cpp ${CXX} ${CXXFLAGS} -shared -fPIC -DLIB3 -o $@ $< diff --git a/test/autotest-parity.md b/test/autotest-parity.md index f60f829188..1a12af0e2f 100644 --- a/test/autotest-parity.md +++ b/test/autotest-parity.md @@ -141,3 +141,14 @@ When moving a deferred test into the new harness, record: - host-sensitive skip reason - artifact needs, especially PTY, MPI, or external-service logs - whether the test passes on `main` before treating it as a branch gate +# TensorRails fork regressions + +These tests were added by the shipped warmctl branch and have no old-harness +equivalent. They are part of the authoritative registry and run in `make check`. + +| Test | Coverage | +| --- | --- | +| `dlopen-origin` | Resolve `$ORIGIN` for the original dlopen caller. | +| `fcntl64-dup` | Track CPython-style fcntl64 descriptor duplication. | +| `sem-unlinked-shared` | Restore shared semaphores whose files were already unlinked. | +| `setsockopt-race` | Keep a socket connection locked while updating its options. | diff --git a/test/autotest.py b/test/autotest.py index 42cce453a9..b20de294fb 100755 --- a/test/autotest.py +++ b/test/autotest.py @@ -2012,6 +2012,10 @@ def _build_tests(self) -> List[TestSpec]: TestSpec("timer1", 1, ["./test/timer1"]), TestSpec("clock", 1, ["./test/clock"]), TestSpec("dlopen1", 1, ["./test/dlopen1"]), + TestSpec("dlopen-origin", 1, ["./test/dlopen-origin"]), + TestSpec("fcntl64-dup", 2, ["./test/fcntl64-dup"]), + TestSpec("sem-unlinked-shared", 2, ["./test/sem-unlinked-shared"]), + TestSpec("setsockopt-race", 1, ["./test/setsockopt-race"]), TestSpec("dmtcp5", 2, ["./test/dmtcp5"]), TestSpec("syscall-tester", 1, ["--checkpoint-open-files ./test/syscall-tester"], diff --git a/test/dlopen-origin.c b/test/dlopen-origin.c new file mode 100644 index 0000000000..7cce562569 --- /dev/null +++ b/test/dlopen-origin.c @@ -0,0 +1,79 @@ +/* libdlopen-origin-a.so loads its sibling as "$ORIGIN/libdlopen-origin-b.so", as a + * library that ships its parts beside itself does. glibc expands $ORIGIN for the + * object that called dlopen, which DMTCP's dlopen wrapper must keep. + * To compile, use -DLIB_A or -DLIB_B for the libraries, neither for the executable. + */ + +#if defined(LIB_A) +#include +#include + +int +load_b(void) +{ + void *handle = dlopen("$ORIGIN/libdlopen-origin-b.so", RTLD_NOW); + if (handle == NULL) { + fprintf(stderr, "dlopen $ORIGIN/libdlopen-origin-b.so failed: %s\n", + dlerror()); + return -1; + } + int (*value)(void) = (int (*)(void))dlsym(handle, "value"); + int result = value != NULL ? value() : -1; + dlclose(handle); + return result; +} + +#elif defined(LIB_B) +int +value(void) +{ + return 42; +} + +#else +#include +#include +#include +#include +#include + +int +main(int argc, char *argv[]) +{ + // chdir to DMTCP root dir + char *char1 = strrchr(argv[0], '/'); + *char1 = '\0'; + char *char2 = strrchr(argv[0], '/'); + *char2 = '\0'; + int rc = chdir(argv[0]); + if (rc != 0) {fprintf(stderr, "Failed to chdir to %s\n", argv[0]); exit(1);} + *char1 = *char2 = '/'; + + void *a = dlopen("test/libdlopen-origin-a.so", RTLD_NOW); + if (a == NULL) { + fprintf(stderr, "dlopen failed: %s\n", dlerror()); + exit(1); + } + int (*load_b)(void) = (int (*)(void))dlsym(a, "load_b"); + if (load_b == NULL) { + fprintf(stderr, "dlsym load_b failed: %s\n", dlerror()); + exit(1); + } + + int cnt = 0; + printf("0: "); fflush(stdout); + while (1) { + if (load_b() != 42) { + exit(1); + } + if (++cnt % 1000 == 0) { + printf("."); + if (cnt % 50000 == 0) { + printf("\n%d: ", cnt / 50000); + } + fflush(stdout); + } + } + return 0; +} +#endif diff --git a/test/fcntl64-dup.c b/test/fcntl64-dup.c new file mode 100644 index 0000000000..7459751197 --- /dev/null +++ b/test/fcntl64-dup.c @@ -0,0 +1,59 @@ +/* A child that keeps a pipe only through an F_DUPFD_CLOEXEC duplicate, as the + * child of CPython's multiprocessing spawn keeps its parent sentinel (os.dup). + * With _FILE_OFFSET_BITS=64, glibc 2.28 and later compile fcntl into fcntl64, + * which DMTCP must see: otherwise the parent's checkpoint drains its end of + * the pipe forever. + */ +#define _FILE_OFFSET_BITS 64 +#define _GNU_SOURCE +#include +#include +#include +#include +#include + +int +main(int argc, char *argv[]) +{ + if (argc == 2) { + // The exec'd child: argv[1] is the pipe's read end. + int fd = atoi(argv[1]); + int dup = fcntl(fd, F_DUPFD_CLOEXEC, 0); + if (dup < 0) { + perror("fcntl(F_DUPFD_CLOEXEC)"); + exit(1); + } + close(fd); + while (1) { + sleep(1); + } + } + + int p[2]; + if (pipe(p) != 0) { + perror("pipe"); + exit(1); + } + pid_t pid = fork(); + if (pid == 0) { + close(p[1]); + char arg[16]; + snprintf(arg, sizeof arg, "%d", p[0]); + execl(argv[0], argv[0], arg, (char *)NULL); + perror("execl"); + exit(1); + } + // The parent keeps the write end and writes nothing, as spawn's does. + close(p[0]); + int cnt = 0; + printf("0: "); + fflush(stdout); + while (1) { + usleep(100000); + if (++cnt % 10 == 0) { + printf("."); + fflush(stdout); + } + } + return 0; +} diff --git a/test/restore-fdkeep.c b/test/restore-fdkeep.c new file mode 100644 index 0000000000..230053bd7b --- /dev/null +++ b/test/restore-fdkeep.c @@ -0,0 +1,53 @@ +/* SPDX-License-Identifier: Apache-2.0 + * Real checkpoint/restore of descriptors created outside DMTCP wrappers. */ +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include "dmtcp.h" +#include "../plugin/fdkeep/raw-syscalls.h" + +#define RAW(n,a,b,c,d) fdkeep_raw4(SYS_##n,(long)(a),(long)(b),(long)(c),(long)(d)) +#define CHECK(x) do { if (!(x)) { fprintf(stderr,"FDKEEP_FAILED line=%d\n",__LINE__); return 2; } } while (0) + +int main(void) +{ + int event = RAW(eventfd2, 7, EFD_CLOEXEC | EFD_NONBLOCK, 0, 0); + int pipes[2]; + CHECK(event >= 0 && RAW(pipe2, pipes, O_CLOEXEC | O_NONBLOCK, 0, 0) == 0); + CHECK(RAW(write, pipes[1], "fdkeep", 6, 0) == 6); + int ep = epoll_create1(EPOLL_CLOEXEC); + struct epoll_event watch = {.events = EPOLLIN, .data.u64 = 0xfeed1234}; + CHECK(ep >= 0 && RAW(epoll_ctl, ep, EPOLL_CTL_ADD, pipes[0], &watch) == 0); + struct sockaddr_un addr = {.sun_family = AF_UNIX}; + char path[] = "/tmp/dmtcp-fdkeep-XXXXXX"; + int tmp = mkstemp(path); + CHECK(tmp >= 0); + close(tmp); unlink(path); + strcpy(addr.sun_path, path); + int listener = RAW(socket, AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0, 0); + CHECK(listener >= 0 && RAW(bind, listener, &addr, sizeof(addr), 0) == 0); + CHECK(RAW(listen, listener, 1, 0, 0) == 0); + int result = dmtcp_checkpoint(); + if (result == DMTCP_AFTER_CHECKPOINT) return 0; + CHECK(result == DMTCP_AFTER_RESTART); + uint64_t counter = 0; + CHECK(RAW(read, event, &counter, sizeof(counter), 0) == sizeof(counter) && counter == 7); + struct epoll_event ready; + CHECK(epoll_wait(ep, &ready, 1, 1000) == 1 && ready.data.u64 == watch.data.u64); + char data[6]; + CHECK(RAW(read, pipes[0], data, sizeof(data), 0) == sizeof(data) && !memcmp(data, "fdkeep", 6)); + int client = RAW(socket, AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0, 0); + CHECK(client >= 0 && RAW(connect, client, &addr, sizeof(addr), 0) == 0); + unlink(path); + puts("FDKEEP_VERIFIED"); + return 0; +} diff --git a/test/restore-memory-bench.c b/test/restore-memory-bench.c new file mode 100644 index 0000000000..8f3c91febf --- /dev/null +++ b/test/restore-memory-bench.c @@ -0,0 +1,108 @@ +/* Large anonymous-memory restart benchmark. Not a GPU performance claim. + * gcc -O2 -Iinclude test/restore-memory-bench.c -o test/restore-memory-bench + * Launch under dmtcp_launch --no-gzip; it checkpoints once and exits. + * On restart it reports resumption, verifies EVERY word, and exits again. + */ +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include +#include +#include +#include "dmtcp.h" +#pragma weak dmtcp_get_restart_env +#pragma weak dmtcp_set_ckpt_dir + +static double now(void) +{ + struct timespec ts; + clock_gettime(CLOCK_MONOTONIC, &ts); + return ts.tv_sec + ts.tv_nsec * 1e-9; +} + +static uint64_t word(size_t i) +{ + return (i * UINT64_C(0x9e3779b97f4a7c15)) ^ UINT64_C(0xa5725a2985c912e7); +} + +int main(int argc, char **argv) +{ + size_t mib = argc > 1 ? strtoull(argv[1], NULL, 10) : 256; + int sparse = argc > 2 && strcmp(argv[2], "sparse") == 0; + int cow = argc > 3 && strcmp(argv[3], "cow") == 0; + const size_t stripeWords = 64 * 1024 * 1024 / 8; + if (!mib || mib > 65536) return 2; + size_t bytes = mib * 1024 * 1024; + uint64_t *p = mmap(NULL, bytes, PROT_READ | PROT_WRITE, + MAP_PRIVATE | MAP_ANONYMOUS, -1, 0); + if (p == MAP_FAILED) { perror("mmap"); return 2; } + /* Match the baseline's ordinary anonymous pages at checkpoint creation. */ + if (madvise(p, bytes, MADV_NOHUGEPAGE)) { perror("madvise"); return 2; } + for (size_t i = 0; i < bytes / 8; ++i) { + if (!sparse || (i / stripeWords) % 2 == 0) p[i] = word(i); + } + printf("BENCH_CHECKPOINT mib=%zu sparse=%d\n", mib, sparse); + fflush(stdout); + int result = dmtcp_checkpoint(); + if (result == DMTCP_NOT_PRESENT) return 3; + if (result == DMTCP_AFTER_CHECKPOINT) return 0; + if (result != DMTCP_AFTER_RESTART) return 4; + char receiver[32]; + if (!dmtcp_get_restart_env || + dmtcp_get_restart_env("DMTCP_BENCH_RESTART_ENV", receiver, + sizeof(receiver)) != RESTART_ENV_SUCCESS || + strcmp(receiver, "receiver") != 0) { + fprintf(stderr, "BENCH_RESTART_ENV_FAILED\n"); + return 6; + } + double start = now(); + struct rusage before, after; + getrusage(RUSAGE_SELF, &before); + printf("BENCH_RESUMED monotonic=%.6f mib=%zu\n", start, mib); + fflush(stdout); + for (size_t i = 0; i < bytes / 8; ++i) { + uint64_t expected = sparse && (i / stripeWords) % 2 ? 0 : word(i); + if (p[i] != expected) { + fprintf(stderr, "BENCH_CORRUPT word=%zu got=%" PRIx64 " expected=%" PRIx64 "\n", + i, p[i], expected); + return 5; + } + } + getrusage(RUSAGE_SELF, &after); + printf("BENCH_VERIFIED seconds=%.6f minflt=%ld majflt=%ld\n", now() - start, + after.ru_minflt - before.ru_minflt, after.ru_majflt - before.ru_majflt); + FILE *f = fopen("/proc/self/smaps_rollup", "r"); + if (f) { + char line[256]; + while (fgets(line, sizeof(line), f)) { + if (strstr(line, "AnonHugePages:") || strstr(line, "Rss:")) fputs(line, stdout); + } + fclose(f); + } + fflush(stdout); + if (cow) { + /* Dirty one word per page, checkpoint again, then verify the dirty AND + * untouched words after another restore. The first image must stay intact. */ + for (size_t i = 0; i < bytes / 8; i += 4096 / 8) p[i] ^= UINT64_C(0xfedcba9876543210); + char second[4096]; + if (!dmtcp_set_ckpt_dir || + dmtcp_get_restart_env("DMTCP_BENCH_SECOND_DIR", second, sizeof(second)) != RESTART_ENV_SUCCESS) + return 7; + dmtcp_set_ckpt_dir(second); + int secondResult = dmtcp_checkpoint(); + if (secondResult == DMTCP_AFTER_CHECKPOINT) return 0; + if (secondResult != DMTCP_AFTER_RESTART) return 8; + for (size_t i = 0; i < bytes / 8; ++i) { + uint64_t expected = sparse && (i / stripeWords) % 2 ? 0 : word(i); + if (i % (4096 / 8) == 0) expected ^= UINT64_C(0xfedcba9876543210); + if (p[i] != expected) return 9; + } + puts("BENCH_SECOND_VERIFIED"); + fflush(stdout); + } + munmap(p, bytes); + return 0; +} diff --git a/test/restore-memory-check.py b/test/restore-memory-check.py new file mode 100644 index 0000000000..d2adec5d62 --- /dev/null +++ b/test/restore-memory-check.py @@ -0,0 +1,172 @@ +#!/usr/bin/env python3 +"""Real checkpoint/restore regressions for the opt-in memory readers (Linux). + +Run after make, with --root pointing to the build or installed prefix. +Only this runner's coordinators/processes are stopped. Logs survive failures. +""" +import argparse +import gzip +import hashlib +import os +import platform +from pathlib import Path +import re +import shutil +import subprocess +import time + + +def run(root, output, command, env): + port_file = output.with_suffix('.port') + port_file.unlink(missing_ok=True) + port = None + with output.with_suffix('.coord.log').open('w') as log: + coord = subprocess.Popen( + [str(root / 'bin/dmtcp_coordinator'), '--port', '0', '--port-file', + str(port_file), '--exit-on-last'], stdout=log, stderr=subprocess.STDOUT, + stdin=subprocess.DEVNULL, start_new_session=True, env=env) + try: + deadline = time.monotonic() + 15 + while not port_file.exists(): + if coord.poll() is not None or time.monotonic() > deadline: + raise RuntimeError('Coordinator failed: ' + str(output)) + time.sleep(.02) + port = port_file.read_text().strip() + with output.open('w') as child_log: + result = subprocess.run( + [str(root / 'bin' / command[0]), '--join-coordinator', + '--coord-port', port, *command[1:]], + stdout=child_log, stderr=subprocess.STDOUT, env=env, + stdin=subprocess.DEVNULL, start_new_session=True, timeout=180) + text = output.read_text() + if result.returncode: + raise RuntimeError(f'{output}: exit {result.returncode}\n{text[-6000:]}') + return text + finally: + if coord.poll() is None and port: + subprocess.run([str(root / 'bin/dmtcp_command'), '--coord-port', + port, '--quit'], env=env, timeout=10, + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + try: + coord.wait(timeout=10) + except subprocess.TimeoutExpired: + coord.kill() + coord.wait() + + +def images(directory): + result = sorted(directory.glob('ckpt_*.dmtcp')) + if not result: + raise RuntimeError('No images in ' + str(directory)) + return list(map(str, result)) + + +def digests(files): + result = [] + for name in files: + h = hashlib.sha256() + with open(name, 'rb') as f: + for data in iter(lambda: f.read(8 * 1024 * 1024), b''): + h.update(data) + result.append(h.hexdigest()) + return result + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument('--root', type=Path, required=True) + parser.add_argument('--output', type=Path, required=True) + parser.add_argument('--plugin', type=Path) + args = parser.parse_args() + root, output = args.root.resolve(), args.output.resolve() + output.mkdir(parents=True, exist_ok=False) + source = Path(__file__).resolve().parent + for fixture in ('restore-memory-bench', 'restore-partial-file', 'restore-fdkeep'): + subprocess.run(['gcc', '-O2', '-I' + str(source.parent / 'include'), + str(source / (fixture + '.c')), '-o', str(output / fixture)], check=True) + # Last entries exercise both the old scan-length bug and prefix matching. + env = dict(os.environ) + if args.plugin: + env['DMTCP_PLUGIN'] = str(args.plugin.resolve()) + env.update(DMTCP_RESTART_PROFILE='1', DMTCP_RESTART_MMAP='0', + DMTCP_RESTART_READERS='1', DMTCP_RESTART_HUGEPAGES='0', + DMTCP_BENCH_RESTART_ENV_SUFFIX='wrong', DMTCP_BENCH_RESTART_ENV='donor') + # 1025 MiB crosses the parallel-reader threshold with an uneven tail. + cases = [('ordinary', 257, False, 1, False, False, False), + ('parallel2', 1025, False, 2, False, False, False), + ('parallel4', 1025, False, 4, True, False, False), + ('parallel8', 1025, False, 8, True, False, False), + ('parallel16', 1025, False, 16, True, False, False), + ('sparse', 257, True, 16, True, False, False), + ('mmap', 257, False, 1, False, True, False), + ('mmap-sparse', 257, True, 1, False, True, False), + ('gzip-fallback', 1025, False, 16, True, True, True)] + for name, mib, sparse, readers, huge, mapped, compressed in cases: + case = output / name + case.mkdir() + first, second, tmp = [case / n for n in ('first', 'second', 'tmp')] + for d in (first, second, tmp): + d.mkdir() + env.update(DMTCP_TMPDIR=str(tmp), DMTCP_BENCH_SECOND_DIR=str(second), + DMTCP_BENCH_RESTART_ENV='donor', DMTCP_GZIP='0') + run(root, case / 'checkpoint.log', ['dmtcp_launch', '--no-gzip', '--ckptdir', str(first), + str(output / 'restore-memory-bench'), str(mib), + 'sparse' if sparse else 'dense', 'cow'], env) + files = images(first) + if compressed: + # ARM64's launcher deliberately disables gzip at checkpoint due to + # an upstream SIGRETURN limitation. Compress our owned raw fixture + # externally to test the restart pipe/fallback on both platforms. + for name in files: + zipped = name + '.gz.tmp' + with open(name, 'rb') as src, gzip.open(zipped, 'wb', compresslevel=1) as dst: + shutil.copyfileobj(src, dst, 8 * 1024 * 1024) + os.replace(zipped, name) + with open(name, 'rb') as f: + assert f.read(2) == b'\x1f\x8b', 'gzip case must use a compressed image' + before = digests(files) + env.update(DMTCP_BENCH_RESTART_ENV='receiver', DMTCP_RESTART_READERS=str(readers), + DMTCP_RESTART_HUGEPAGES=str(int(huge)), DMTCP_RESTART_MMAP=str(int(mapped))) + text = run(root, case / 'restart.log', ['dmtcp_restart', *files], env) + assert 'BENCH_VERIFIED' in text, text[-6000:] + field = 'filemap_kib' if mapped else 'parallel_kib' + values = list(map(int, re.findall(r'\b' + field + r'=(\d+)', text))) + parallel = platform.machine() == 'x86_64' and readers > 1 and mib >= 1024 and not sparse + if not compressed and (mapped or parallel): + assert any(values), f'{name}: optimized reader was not exercised' + if platform.machine() != 'x86_64' and not mapped: + assert values and not any(values), 'unsupported architecture must use ordinary reader' + if compressed: + values = list(map(int, re.findall(r'\b(?:filemap_kib|parallel_kib)=(\d+)', text))) + assert values and not any(values), 'compressed stream must use ordinary reader' + text = run(root, case / 'second-restart.log', ['dmtcp_restart', *images(second)], env) + assert 'BENCH_SECOND_VERIFIED' in text, text[-6000:] + assert before == digests(files), 'source image changed' + for d in (first, second, tmp): + shutil.rmtree(d) + print(name + ': PASS (all words, dirty pages, second checkpoint, source SHA-256)', flush=True) + case = output / 'partial-file' + case.mkdir() + env.update(DMTCP_TMPDIR=str(case), DMTCP_RESTART_MMAP='0', DMTCP_GZIP='0') + run(root, case / 'checkpoint.log', ['dmtcp_launch', '--no-gzip', '--ckptdir', + str(case), str(output / 'restore-partial-file')], env) + text = run(root, case / 'restart.log', ['dmtcp_restart', *images(case)], env) + assert 'PARTIAL_FILE_VERIFIED' in text, text[-6000:] + for f in case.glob('ckpt_*.dmtcp'): + f.unlink() + print('partial-file: PASS (private dirty bytes past EOF)', flush=True) + if args.plugin: + case = output / 'fdkeep' + case.mkdir() + env.update(DMTCP_TMPDIR=str(case), DMTCP_RESTART_MMAP='0', DMTCP_GZIP='0') + run(root, case / 'checkpoint.log', ['dmtcp_launch', '--no-gzip', '--ckptdir', + str(case), str(output / 'restore-fdkeep')], env) + text = run(root, case / 'restart.log', ['dmtcp_restart', *images(case)], env) + assert 'FDKEEP_VERIFIED' in text, text[-6000:] + for f in case.glob('ckpt_*.dmtcp'): + f.unlink() + print('fdkeep: PASS (eventfd, pipe bytes, Unix listener, epoll watch)', flush=True) + + +if __name__ == '__main__': + main() diff --git a/test/restore-partial-file.c b/test/restore-partial-file.c new file mode 100644 index 0000000000..e3c477d3a5 --- /dev/null +++ b/test/restore-partial-file.c @@ -0,0 +1,37 @@ +/* Verify the private dirty tail of a regular file's final partial page. + * gcc -O2 -Iinclude test/restore-partial-file.c -o test/restore-partial-file + * Run under dmtcp_launch --no-gzip, then dmtcp_restart its image. + */ +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include +#include "dmtcp.h" + +int main(void) +{ + char path[] = "/tmp/dmtcp-partial-file-XXXXXX"; + int fd = mkstemp(path); + if (fd < 0 || ftruncate(fd, 3001)) return 2; + unsigned char *p = mmap(NULL, 8192, PROT_READ | PROT_WRITE, MAP_PRIVATE, fd, 0); + if (p == MAP_FAILED) return 3; + close(fd); + for (size_t i = 0; i < 4096; i++) p[i] = (i * 17 + 39) & 255; + if (mprotect(p, 8192, PROT_READ)) return 4; + int result = dmtcp_checkpoint(); + if (result == DMTCP_AFTER_CHECKPOINT) return 0; + if (result != DMTCP_AFTER_RESTART) return 5; + for (size_t i = 0; i < 4096; i++) { + if (p[i] != ((i * 17 + 39) & 255)) { + fprintf(stderr, "PARTIAL_FILE_CORRUPT byte=%zu\n", i); + return 6; + } + } + puts("PARTIAL_FILE_VERIFIED"); + munmap(p, 8192); + unlink(path); + return 0; +} diff --git a/test/sem-unlinked-shared.c b/test/sem-unlinked-shared.c new file mode 100644 index 0000000000..86d628cb67 --- /dev/null +++ b/test/sem-unlinked-shared.c @@ -0,0 +1,65 @@ +/* Two processes that share named semaphores unlinked at once, as CPython's + * multiprocessing creates its locks: each maps the deleted /dev/shm file and + * holds no descriptor for it. A restart must recreate one file that both map + * again; with a file each, a post would no longer reach the other process and + * its wait would time out. + */ +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include +#include + +static sem_t * +open_unlinked(const char *role) +{ + char name[64]; + snprintf(name, sizeof name, "/dmtcp-sem-%s-%d", role, getpid()); + sem_t *s = sem_open(name, O_CREAT | O_EXCL, 0600, 0); + if (s == SEM_FAILED) { + perror("sem_open"); + exit(1); + } + sem_unlink(name); + return s; +} + +static void +wait_for(sem_t *s) +{ + struct timespec t; + clock_gettime(CLOCK_REALTIME, &t); + t.tv_sec += 10; + while (sem_timedwait(s, &t) != 0) { + if (errno != EINTR) { + perror("sem_timedwait"); + exit(1); + } + } +} + +int +main() +{ + sem_t *ping = open_unlinked("ping"); + sem_t *pong = open_unlinked("pong"); + if (fork() == 0) { + while (1) { + wait_for(ping); + sem_post(pong); + } + } + int cnt = 0; + while (1) { + sem_post(ping); + wait_for(pong); + if (++cnt % 1000 == 0) { + printf("%d ", cnt / 1000); + fflush(stdout); + usleep(100000); + } + } +} diff --git a/test/setsockopt-race.c b/test/setsockopt-race.c new file mode 100644 index 0000000000..ea45e78fa9 --- /dev/null +++ b/test/setsockopt-race.c @@ -0,0 +1,47 @@ +/* One thread sets a socket option while another closes the descriptor past + * DMTCP's close wrapper (a raw syscall, as cuInterpose's Rust code closes + * its sockets) and creates a socket at the same number, so that DMTCP drops + * the old connection. setsockopt recorded the option on the connection it + * had looked up, which the other thread could free meanwhile: the process + * crashed in std::map, as a vLLM worker did right after a restart. + */ +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include + +static volatile int fd = -1; + +static void * +churn(void *arg) +{ + (void)arg; + while (1) { + syscall(SYS_close, fd); + fd = socket(AF_UNIX, SOCK_STREAM, 0); + } + return NULL; +} + +int +main() +{ + fd = socket(AF_UNIX, SOCK_STREAM, 0); + pthread_t t; + if (pthread_create(&t, NULL, churn, NULL) != 0) { + perror("pthread_create"); + return 1; + } + struct timeval tv = { 1, 0 }; + unsigned long n = 0; + while (1) { + setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof tv); + if (++n % 1000000 == 0) { + printf("%lu ", n / 1000000); + fflush(stdout); + } + } +}