Skip to content

API

Easy process freeze & thaw

cli

Command-line entry point for pdum-criu utilities.

doctor()

Run environment diagnostics and report missing prerequisites.

Notes
  • Always exits on non-Linux hosts because CRIU requires Linux kernel support.
  • Prints a checklist showing which requirements passed or failed.

Raises:

Type Description
Exit

Raised when the platform is unsupported or after reporting failed checks.
Source code in src/pdum/criu/cli.py 
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
@app.command("doctor")
def doctor() -> None:
    """Run environment diagnostics and report missing prerequisites.

    Notes
    -----
    - Always exits on non-Linux hosts because CRIU requires Linux kernel support.
    - Prints a checklist showing which requirements passed or failed.

    Raises
    ------
    typer.Exit
        Raised when the platform is unsupported or after reporting failed checks.
    """
    if not sys.platform.startswith("linux"):
        console.print(
            f"[bold red]pdum-criu doctor only supports Linux hosts (detected: {sys.platform}).[/]\n"
            "CRIU depends on Linux kernel checkpoint/restore features."
        )
        raise typer.Exit(code=1)

    console.print("[bold cyan]Running environment diagnostics...[/]")

    all_ok = True
    results = utils.doctor_check_results(verbose=True)
    for label, ok, message in results:
        if ok:
            console.print(f"[bold green]✓ {label}[/]")
        else:
            console.print(f"[bold red]✗ {label}[/]")
            if message:
                console.print(f"    {message}")
            if label == "sudo closefrom_override":
                console.print(
                    "[bold yellow]Tip:[/] Run `sudo visudo` and add either `Defaults    closefrom_override` "
                    "or `Defaults:YOURUSER    closefrom_override`."
                )
        all_ok = all_ok and ok

    if all_ok:
        console.print("[bold green]All doctor checks passed![/]")
    else:
        console.print("[bold yellow]Resolve the failed checks above before continuing.[/]")

main_callback(ctx)

Fallback handler when no top-level subcommand is invoked.

Parameters:

Name Type Description Default
ctx Context

Current Typer context used to detect whether a subcommand was chosen.

 required

Raises:

Type Description
Exit

Raised to stop execution after printing the autogenerated help text.
Source code in src/pdum/criu/cli.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
@app.callback(invoke_without_command=True)
def main_callback(ctx: typer.Context) -> None:
    """Fallback handler when no top-level subcommand is invoked.

    Parameters
    ----------
    ctx : typer.Context
        Current Typer context used to detect whether a subcommand was chosen.

    Raises
    ------
    typer.Exit
        Raised to stop execution after printing the autogenerated help text.
    """
    if ctx.invoked_subcommand is None:
        typer.echo(ctx.get_help())
        raise typer.Exit()

shell_beam(images_dir=typer.Option(None, '--dir', '-d', help='Optional directory for the CRIU image set (defaults to a temp dir).'), pid=typer.Option(None, '--pid', help='PID to beam.'), pgrep=typer.Option(None, '--pgrep', help='pgrep pattern to resolve the PID.'), log_file=typer.Option(None, '--log-file', '-l', help='Optional log file override for the freeze phase.'), verbosity=typer.Option(4, '--verbosity', '-v', min=0, max=4, help='CRIU verbosity level.'), leave_running=typer.Option(True, '--leave-running/--no-leave-running', help='Keep the target process running after dump completes.'), cleanup=typer.Option(True, '--cleanup/--no-cleanup', help='Remove the image directory once the beamed process exits.'), show_command=typer.Option(True, '--show-command/--hide-command', help='Print each CRIU command before executing it.'), pidfile=typer.Option(None, '--pidfile', help='Path where CRIU writes the restored PID (defaults to a temp file in the image dir).'))

Freeze a shell and immediately thaw it ("beam" workflow).

Parameters:

Name Type Description Default
images_dir Path

Directory for CRIU artifacts. A temporary directory is created when omitted.

 Option(None, '--dir', '-d', help='Optional directory for the CRIU image set (defaults to a temp dir).')
pid int

Explicit PID to beam. Mutually exclusive with pgrep.

 Option(None, '--pid', help='PID to beam.')
pgrep str

pgrep expression used to determine the PID when pid is not provided.

 Option(None, '--pgrep', help='pgrep pattern to resolve the PID.')
log_file Path

Override for the freeze-phase log file.

 Option(None, '--log-file', '-l', help='Optional log file override for the freeze phase.')
verbosity int

CRIU verbosity level between 0 and 4.

 Option(4, '--verbosity', '-v', min=0, max=4, help='CRIU verbosity level.')
leave_running bool

If True, keep the original process alive after the dump completes.

 Option(True, '--leave-running/--no-leave-running', help='Keep the target process running after dump completes.')
cleanup bool

Remove the image directory after the beamed process exits.

 Option(True, '--cleanup/--no-cleanup', help='Remove the image directory once the beamed process exits.')
show_command bool

When True print each CRIU command before running it.

 Option(True, '--show-command/--hide-command', help='Print each CRIU command before executing it.')
pidfile Path

Location where CRIU writes the restored PID. Defaults to a temp file in images_dir.

 Option(None, '--pidfile', help='Path where CRIU writes the restored PID (defaults to a temp file in the image dir).')

Raises:

Type Description
Exit

Raised when freezing or restoring fails, or if prerequisites are missing.
Source code in src/pdum/criu/cli.py 
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
@shell_app.command("beam")
def shell_beam(
    images_dir: Optional[Path] = typer.Option(
        None,
        "--dir",
        "-d",
        help="Optional directory for the CRIU image set (defaults to a temp dir).",
    ),
    pid: Optional[int] = typer.Option(None, "--pid", help="PID to beam."),
    pgrep: Optional[str] = typer.Option(None, "--pgrep", help="pgrep pattern to resolve the PID."),
    log_file: Optional[Path] = typer.Option(
        None,
        "--log-file",
        "-l",
        help="Optional log file override for the freeze phase.",
    ),
    verbosity: int = typer.Option(4, "--verbosity", "-v", min=0, max=4, help="CRIU verbosity level."),
    leave_running: bool = typer.Option(
        True,
        "--leave-running/--no-leave-running",
        help="Keep the target process running after dump completes.",
    ),
    cleanup: bool = typer.Option(
        True,
        "--cleanup/--no-cleanup",
        help="Remove the image directory once the beamed process exits.",
    ),
    show_command: bool = typer.Option(
        True,
        "--show-command/--hide-command",
        help="Print each CRIU command before executing it.",
    ),
    pidfile: Optional[Path] = typer.Option(
        None,
        "--pidfile",
        help="Path where CRIU writes the restored PID (defaults to a temp file in the image dir).",
    ),
) -> None:
    """Freeze a shell and immediately thaw it (\"beam\" workflow).

    Parameters
    ----------
    images_dir : Path, optional
        Directory for CRIU artifacts. A temporary directory is created when omitted.
    pid : int, optional
        Explicit PID to beam. Mutually exclusive with ``pgrep``.
    pgrep : str, optional
        ``pgrep`` expression used to determine the PID when ``pid`` is not provided.
    log_file : Path, optional
        Override for the freeze-phase log file.
    verbosity : int
        CRIU verbosity level between 0 and 4.
    leave_running : bool
        If ``True``, keep the original process alive after the dump completes.
    cleanup : bool
        Remove the image directory after the beamed process exits.
    show_command : bool
        When ``True`` print each CRIU command before running it.
    pidfile : Path, optional
        Location where CRIU writes the restored PID. Defaults to a temp file in ``images_dir``.

    Raises
    ------
    typer.Exit
        Raised when freezing or restoring fails, or if prerequisites are missing.
    """

    try:
        utils.ensure_linux()
    except RuntimeError as exc:
        console.print(f"[bold red]{exc}[/]")
        raise typer.Exit(code=1)

    target_pid = _resolve_pid_option(pid, pgrep)
    _require(utils.ensure_sudo, verbose=True, raise_=True)
    _require(utils.ensure_pgrep, verbose=True, raise_=True)
    criu_path = _require(utils.ensure_criu, verbose=True, raise_=True)

    sudo_path = _require(utils.resolve_command, "sudo")

    if images_dir is None:
        images_dir = Path(tempfile.mkdtemp(prefix="pdum-criu-beam-"))
        console.print(f"[bold cyan]Beam images directory:[/] {images_dir}")
    else:
        images_dir = _prepare_dir(images_dir)

    log_path = _resolve_log_path(log_file, images_dir, f"freeze.{target_pid}.log")
    pidfile_path, pidfile_is_temp = _resolve_pidfile_option(pidfile, images_dir, prefix="beam-restore")

    command = _build_criu_dump_command(
        sudo_path,
        criu_path,
        images_dir,
        target_pid,
        log_path,
        verbosity,
        leave_running,
    )

    console.print(f"[bold cyan]Freezing PID {target_pid} before beam[/]")
    exit_code = _run_command(command, show=show_command)
    if exit_code != 0:
        tail = utils.tail_file(log_path, lines=10)
        console.print(f"[bold red]Beam freeze failed (exit {exit_code}).[/]")
        if tail:
            console.print("[bold yellow]Log tail:[/]")
            console.print(tail)
        raise typer.Exit(code=exit_code)

    _record_freeze_metadata(images_dir, target_pid)
    console.print(f"[bold cyan]Thawing beam image from {images_dir}[/]")
    restore_result = _execute_restore(images_dir, show_command=show_command, pidfile=pidfile_path)
    if restore_result.exit_code != 0:
        console.print(f"[bold red]Beam restore failed (exit {restore_result.exit_code}).[/]")
        tail = utils.tail_file(restore_result.log_path, lines=10)
        if tail:
            console.print("[bold yellow]Log tail:[/]")
            console.print(tail)
        _maybe_report_vscode_from_metadata(images_dir, fallback_pid=target_pid)
        if pidfile_is_temp:
            _safe_unlink(pidfile_path)
        raise typer.Exit(code=restore_result.exit_code)

    restored_pid = _read_pidfile(restore_result.pidfile)
    if restored_pid is not None:
        console.print(
            f"[bold green]Beam complete.[/] Restore PID: {restored_pid}  Log: {restore_result.log_path}"
        )
    else:
        console.print(f"[bold green]Beam complete.[/] Restore log: {restore_result.log_path}")
        console.print(
            f"[bold yellow]Warning:[/] Unable to read PID from {restore_result.pidfile}."
        )

    if cleanup:
        if restored_pid is None:
            console.print(
                "[bold yellow]Note:[/] Cleanup will run when this CLI exits because the restored PID is unknown."
            )
        watcher_pid = restored_pid if restored_pid is not None else os.getpid()
        utils.spawn_directory_cleanup(images_dir, watcher_pid)

    if pidfile_is_temp:
        _safe_unlink(restore_result.pidfile)
    else:
        console.print(f"[bold cyan]PID file:[/] {restore_result.pidfile}")

shell_callback(ctx)

Display shell helper usage when no subcommand is provided.

Parameters:

Name Type Description Default
ctx Context

Current Typer context describing the shell command invocation.

 required

Raises:

Type Description
Exit

Raised to stop execution after emitting help text.
Source code in src/pdum/criu/cli.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
@shell_app.callback(invoke_without_command=True)
def shell_callback(ctx: typer.Context) -> None:
    """Display shell helper usage when no subcommand is provided.

    Parameters
    ----------
    ctx : typer.Context
        Current Typer context describing the shell command invocation.

    Raises
    ------
    typer.Exit
        Raised to stop execution after emitting help text.
    """
    if ctx.invoked_subcommand is None:
        typer.echo(ctx.get_help())
        raise typer.Exit()

shell_freeze(images_dir=typer.Option(..., '--dir', '-d', help='Directory that will contain the CRIU image set.'), pid=typer.Option(None, '--pid', help='PID to freeze.'), pgrep=typer.Option(None, '--pgrep', help='pgrep pattern to resolve the PID.'), log_file=typer.Option(None, '--log-file', '-l', help='Optional log file override (default: freeze.<pid>.log inside the image dir).'), verbosity=typer.Option(4, '--verbosity', '-v', min=0, max=4, help='CRIU verbosity level (0-4).'), leave_running=typer.Option(True, '--leave-running/--no-leave-running', help='Keep the target process running after dump completes.'), show_command=typer.Option(True, '--show-command/--hide-command', help='Print the CRIU command before executing it.'), show_tail=typer.Option(True, '--show-tail/--hide-tail', help='Display the last five lines of the freeze log after success.'), validate_tty=typer.Option(True, '--validate-tty/--no-validate-tty', help='Fail fast if the process is using an unsupported terminal (e.g., VS Code).'))

Checkpoint a running shell/job into a CRIU image directory.

Parameters:

Name Type Description Default
images_dir Path

Target directory that will contain the CRIU image set (created if needed).

 Option(..., '--dir', '-d', help='Directory that will contain the CRIU image set.')
pid int

Explicit PID to freeze. Mutually exclusive with pgrep.

 Option(None, '--pid', help='PID to freeze.')
pgrep str

pgrep expression used to resolve the PID when pid is not provided.

 Option(None, '--pgrep', help='pgrep pattern to resolve the PID.')
log_file Path

Override for the CRIU log (defaults to freeze.<pid>.log under images_dir).

 Option(None, '--log-file', '-l', help='Optional log file override (default: freeze.<pid>.log inside the image dir).')
verbosity int

CRIU verbosity level between 0 and 4.

 Option(4, '--verbosity', '-v', min=0, max=4, help='CRIU verbosity level (0-4).')
leave_running bool

Whether to keep the target process alive after the dump completes.

 Option(True, '--leave-running/--no-leave-running', help='Keep the target process running after dump completes.')
show_command bool

When True print the CRIU command prior to execution.

 Option(True, '--show-command/--hide-command', help='Print the CRIU command before executing it.')
show_tail bool

When True print the last five log lines on success.

 Option(True, '--show-tail/--hide-tail', help='Display the last five lines of the freeze log after success.')
validate_tty bool

Fail fast if the process is attached to an unsupported TTY (e.g., VS Code).

 Option(True, '--validate-tty/--no-validate-tty', help='Fail fast if the process is using an unsupported terminal (e.g., VS Code).')

Raises:

Type Description
Exit

Raised when validation fails or CRIU exits with a non-zero status.
Source code in src/pdum/criu/cli.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
@shell_app.command("freeze")
def shell_freeze(
    images_dir: Path = typer.Option(..., "--dir", "-d", help="Directory that will contain the CRIU image set."),
    pid: Optional[int] = typer.Option(None, "--pid", help="PID to freeze."),
    pgrep: Optional[str] = typer.Option(None, "--pgrep", help="pgrep pattern to resolve the PID."),
    log_file: Optional[Path] = typer.Option(
        None,
        "--log-file",
        "-l",
        help="Optional log file override (default: freeze.<pid>.log inside the image dir).",
    ),
    verbosity: int = typer.Option(4, "--verbosity", "-v", min=0, max=4, help="CRIU verbosity level (0-4)."),
    leave_running: bool = typer.Option(
        True,
        "--leave-running/--no-leave-running",
        help="Keep the target process running after dump completes.",
    ),
    show_command: bool = typer.Option(
        True,
        "--show-command/--hide-command",
        help="Print the CRIU command before executing it.",
    ),
    show_tail: bool = typer.Option(
        True,
        "--show-tail/--hide-tail",
        help="Display the last five lines of the freeze log after success.",
    ),
    validate_tty: bool = typer.Option(
        True,
        "--validate-tty/--no-validate-tty",
        help="Fail fast if the process is using an unsupported terminal (e.g., VS Code).",
    ),
) -> None:
    """Checkpoint a running shell/job into a CRIU image directory.

    Parameters
    ----------
    images_dir : Path
        Target directory that will contain the CRIU image set (created if needed).
    pid : int, optional
        Explicit PID to freeze. Mutually exclusive with ``pgrep``.
    pgrep : str, optional
        ``pgrep`` expression used to resolve the PID when ``pid`` is not provided.
    log_file : Path, optional
        Override for the CRIU log (defaults to ``freeze.<pid>.log`` under ``images_dir``).
    verbosity : int
        CRIU verbosity level between 0 and 4.
    leave_running : bool
        Whether to keep the target process alive after the dump completes.
    show_command : bool
        When ``True`` print the CRIU command prior to execution.
    show_tail : bool
        When ``True`` print the last five log lines on success.
    validate_tty : bool
        Fail fast if the process is attached to an unsupported TTY (e.g., VS Code).

    Raises
    ------
    typer.Exit
        Raised when validation fails or CRIU exits with a non-zero status.
    """

    try:
        utils.ensure_linux()
    except RuntimeError as exc:
        console.print(f"[bold red]{exc}[/]")
        raise typer.Exit(code=1)

    target_pid = _resolve_pid_option(pid, pgrep)
    _require(utils.ensure_sudo, verbose=True, raise_=True)
    _require(utils.ensure_pgrep, verbose=True, raise_=True)
    criu_path = _require(utils.ensure_criu, verbose=True, raise_=True)

    sudo_path = _require(utils.resolve_command, "sudo")
    images_dir = _prepare_dir(images_dir)
    log_path = _resolve_log_path(log_file, images_dir, f"freeze.{target_pid}.log")

    if validate_tty:
        ok, reason = _tty_is_supported(target_pid)
        if not ok:
            console.print(
                "[bold red]Cannot freeze target:[/] "
                f"{reason}\n"
                "[bold yellow]Hint:[/] Use a real terminal (tmux/screen/gnome-terminal) "
                "or detach via `setsid`/`script`. VS Code's integrated terminal is unsupported. "
                "Override with [bold cyan]--no-validate-tty[/] only if you plan to thaw inside VS Code."
            )
            raise typer.Exit(code=1)

    command = _build_criu_dump_command(
        sudo_path,
        criu_path,
        images_dir,
        target_pid,
        log_path,
        verbosity,
        leave_running,
    )

    console.print(f"[bold cyan]Freezing PID {target_pid} into {images_dir}[/]")
    exit_code = _run_command(command, show=show_command)
    if exit_code == 0:
        console.print(f"[bold green]Freeze complete.[/] Log: {log_path}")
        _record_freeze_metadata(images_dir, target_pid)
        if show_tail:
            _print_log_tail(sudo_path, log_path, lines=5)
        return

    tail = utils.tail_file(log_path, lines=10)
    console.print(f"[bold red]Freeze failed (exit {exit_code}).[/]")
    if tail:
        console.print("[bold yellow]Log tail:[/]")
        console.print(tail)
    raise typer.Exit(code=exit_code)

shell_thaw(images_dir=typer.Option(..., '--dir', '-d', help='CRIU image directory to restore.'), show_command=typer.Option(True, '--show-command/--hide-command', help='Print the CRIU command before executing it.'), pidfile=typer.Option(None, '--pidfile', help='Path where CRIU writes the restored PID (defaults to a temp file in the image dir).'))

Restore a shell/job from a CRIU image directory.

Parameters:

Name Type Description Default
images_dir Path

Directory that contains the CRIU image set to restore.

 Option(..., '--dir', '-d', help='CRIU image directory to restore.')
show_command bool

When True print the CRIU restore command before executing it.

 Option(True, '--show-command/--hide-command', help='Print the CRIU command before executing it.')
pidfile Path

Location where CRIU writes the restored PID. Defaults to a temporary file inside images_dir.

 Option(None, '--pidfile', help='Path where CRIU writes the restored PID (defaults to a temp file in the image dir).')

Raises:

Type Description
Exit

Raised when prerequisites are missing or CRIU restoration fails.
Source code in src/pdum/criu/cli.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
@shell_app.command("thaw")
def shell_thaw(
    images_dir: Path = typer.Option(..., "--dir", "-d", help="CRIU image directory to restore."),
    show_command: bool = typer.Option(
        True,
        "--show-command/--hide-command",
        help="Print the CRIU command before executing it.",
    ),
    pidfile: Optional[Path] = typer.Option(
        None,
        "--pidfile",
        help="Path where CRIU writes the restored PID (defaults to a temp file in the image dir).",
    ),
) -> None:
    """Restore a shell/job from a CRIU image directory.

    Parameters
    ----------
    images_dir : Path
        Directory that contains the CRIU image set to restore.
    show_command : bool
        When ``True`` print the CRIU restore command before executing it.
    pidfile : Path, optional
        Location where CRIU writes the restored PID. Defaults to a temporary file inside ``images_dir``.

    Raises
    ------
    typer.Exit
        Raised when prerequisites are missing or CRIU restoration fails.
    """

    try:
        utils.ensure_linux()
    except RuntimeError as exc:
        console.print(f"[bold red]{exc}[/]")
        raise typer.Exit(code=1)

    images_dir = images_dir.expanduser().resolve()
    if not images_dir.exists():
        console.print(f"[bold red]Image directory does not exist:[/] {images_dir}")
        raise typer.Exit(code=1)

    pidfile_path, pidfile_is_temp = _resolve_pidfile_option(pidfile, images_dir, prefix="restore")
    result = _execute_restore(images_dir, show_command=show_command, pidfile=pidfile_path)
    if result.exit_code == 0:
        restored_pid = _read_pidfile(result.pidfile)
        if restored_pid is not None:
            console.print(f"[bold green]Restore complete.[/] PID: {restored_pid}  Log: {result.log_path}")
        else:
            console.print(f"[bold green]Restore complete.[/] Log: {result.log_path}")
            console.print(
                f"[bold yellow]Warning:[/] Unable to read PID from {result.pidfile}."
            )
        if pidfile_is_temp:
            _safe_unlink(result.pidfile)
        else:
            console.print(f"[bold cyan]PID file:[/] {result.pidfile}")
        return

    console.print(f"[bold red]Restore failed (exit {result.exit_code}).[/]")
    tail = utils.tail_file(result.log_path, lines=10)
    if tail:
        console.print("[bold yellow]Log tail:[/]")
        console.print(tail)
    _maybe_report_vscode_from_metadata(images_dir)
    if pidfile_is_temp:
        _safe_unlink(pidfile_path)
    raise typer.Exit(code=result.exit_code)

version_command()

Display the CLI version banner.

Notes

This command always succeeds and prints the current pdum-criu package version.

Source code in src/pdum/criu/cli.py 
85
86
87
88
89
90
91
92
93
94
95
96
@app.command("version")
def version_command() -> None:
    """Display the CLI version banner.

    Notes
    -----
    This command always succeeds and prints the current `pdum-criu` package version.
    """
    console.print(
        "[bold green]pdum-criu[/] CLI\n"
        f"[bold cyan]version:[/] {__version__}"
    )

goblins

Utility APIs for freezing and thawing goblin processes.

AsyncGoblinProcess dataclass

Async counterpart returned by :func:thaw_async.

Parameters:

Name Type Description Default
helper_pid int | None

PID of the helper process (criu or criu-ns).

 required
stdin StreamWriter

Writable pipe towards the goblin's stdin.

 required
stdout StreamReader

Readers for stdout/stderr respectively.

 required
stderr StreamReader

Readers for stdout/stderr respectively.

 required
images_dir Path

Directory containing the image set.

 required
log_path Path

CRIU restore log path.

 required
pidfile Path

File CRIU uses to publish the restored PID.

 required
Source code in src/pdum/criu/goblins/__init__.py 
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
@dataclass
class AsyncGoblinProcess:
    """Async counterpart returned by :func:`thaw_async`.

    Parameters
    ----------
    helper_pid : int | None
        PID of the helper process (``criu`` or ``criu-ns``).
    stdin : asyncio.StreamWriter
        Writable pipe towards the goblin's stdin.
    stdout, stderr : asyncio.StreamReader
        Readers for stdout/stderr respectively.
    images_dir : Path
        Directory containing the image set.
    log_path : Path
        CRIU restore log path.
    pidfile : Path
        File CRIU uses to publish the restored PID.
    """

    helper_pid: int | None
    stdin: asyncio.StreamWriter
    stdout: asyncio.StreamReader
    stderr: asyncio.StreamReader
    images_dir: Path
    log_path: Path
    pidfile: Path

    async def read_pidfile(self) -> int:
        """Asynchronously return the PID recorded by CRIU."""

        data = await _read_file_async(self.pidfile)
        return int(data.decode("utf-8").strip())

    async def close(self) -> None:
        self.stdin.close()
        try:
            await self.stdin.wait_closed()
        except Exception:
            pass

read_pidfile() async

Asynchronously return the PID recorded by CRIU.

Source code in src/pdum/criu/goblins/__init__.py 
263
264
265
266
267
async def read_pidfile(self) -> int:
    """Asynchronously return the PID recorded by CRIU."""

    data = await _read_file_async(self.pidfile)
    return int(data.decode("utf-8").strip())

GoblinProcess dataclass

Synchronous handle returned by :func:thaw.

Parameters:

Name Type Description Default
helper_pid int | None

PID of the helper process (criu or criu-ns) coordinating the restore.

 required
stdin BufferedWriter

Binary file objects connected to the goblin's stdio pipes.

 required
stdout BufferedWriter

Binary file objects connected to the goblin's stdio pipes.

 required
stderr BufferedWriter

Binary file objects connected to the goblin's stdio pipes.

 required
images_dir Path

Directory that contains the CRIU image set used for this restore.

 required
log_path Path

Path to the CRIU restore log file.

 required
pidfile Path

Path CRIU populates with the restored process PID.

 required
Source code in src/pdum/criu/goblins/__init__.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
@dataclass
class GoblinProcess:
    """Synchronous handle returned by :func:`thaw`.

    Parameters
    ----------
    helper_pid : int | None
        PID of the helper process (``criu`` or ``criu-ns``) coordinating the restore.
    stdin, stdout, stderr :
        Binary file objects connected to the goblin's stdio pipes.
    images_dir : Path
        Directory that contains the CRIU image set used for this restore.
    log_path : Path
        Path to the CRIU restore log file.
    pidfile : Path
        Path CRIU populates with the restored process PID.
    """

    helper_pid: int | None
    stdin: io.BufferedWriter
    stdout: io.BufferedReader
    stderr: io.BufferedReader
    images_dir: Path
    log_path: Path
    pidfile: Path

    def read_pidfile(self) -> int:
        """Return the PID recorded by CRIU."""

        return int(self.pidfile.read_text().strip())

    def terminate(self, sig: int = signal.SIGTERM) -> None:
        os.kill(self.read_pidfile(), sig)

    def close(self) -> None:
        for stream in (self.stdin, self.stdout, self.stderr):
            try:
                stream.close()
            except Exception:
                pass

read_pidfile()

Return the PID recorded by CRIU.

Source code in src/pdum/criu/goblins/__init__.py 
219
220
221
222
def read_pidfile(self) -> int:
    """Return the PID recorded by CRIU."""

    return int(self.pidfile.read_text().strip())

freeze(pid, images_dir, *, leave_running=True, log_path=None, verbosity=4, extra_args=None, shell_job=True)

Checkpoint a goblin process into the specified image directory.

Parameters:

Name Type Description Default
pid int

PID of the goblin process to dump.

 required
images_dir str | Path

Directory that will store the CRIU image set.

 required
leave_running bool

Whether to keep the goblin running after the dump completes. Defaults to True.

 True
log_path str | Path

Optional path for CRIU's log file. Defaults to images_dir / f"goblin-freeze.{pid}.log".

 None
verbosity int

CRIU verbosity level (0-4). Defaults to 4 to aid troubleshooting.

 4
extra_args Iterable[str]

Additional CRIU arguments to append verbatim.

 None
shell_job bool

Whether to include --shell-job. Disable when the target already runs detached from any controlling TTY. Defaults to True.

 True

Returns:

Type Description
Path

Path to the CRIU log file for the dump operation.

Raises:

Type Description
RuntimeError

If CRIU fails to dump the process.
ValueError

If pid is not positive.
Source code in src/pdum/criu/goblins/__init__.py 
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def freeze(
    pid: int,
    images_dir: str | Path,
    *,
    leave_running: bool = True,
    log_path: str | Path | None = None,
    verbosity: int = 4,
    extra_args: Iterable[str] | None = None,
    shell_job: bool = True,
) -> Path:
    """Checkpoint a goblin process into the specified image directory.

    Parameters
    ----------
    pid : int
        PID of the goblin process to dump.
    images_dir : str | Path
        Directory that will store the CRIU image set.
    leave_running : bool, optional
        Whether to keep the goblin running after the dump completes. Defaults to True.
    log_path : str | Path, optional
        Optional path for CRIU's log file. Defaults to ``images_dir / f"goblin-freeze.{pid}.log"``.
    verbosity : int, optional
        CRIU verbosity level (0-4). Defaults to 4 to aid troubleshooting.
    extra_args : Iterable[str], optional
        Additional CRIU arguments to append verbatim.
    shell_job : bool, optional
        Whether to include ``--shell-job``. Disable when the target already runs
        detached from any controlling TTY. Defaults to True.

    Returns
    -------
    Path
        Path to the CRIU log file for the dump operation.

    Raises
    ------
    RuntimeError
        If CRIU fails to dump the process.
    ValueError
        If ``pid`` is not positive.
    """

    if pid <= 0:
        raise ValueError("PID must be a positive integer")

    logger.info("Freezing goblin pid %s into %s", pid, images_dir)

    context = _build_freeze_context(
        pid,
        images_dir,
        leave_running=leave_running,
        log_path=log_path,
        verbosity=verbosity,
        extra_args=extra_args,
        shell_job=shell_job,
    )

    logger.debug("Running command: %s", shlex.join(context.command))

    result = subprocess.run(
        context.command,
        check=False,
        stdin=subprocess.DEVNULL,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        start_new_session=True,
    )
    _ensure_log_readable(context.log_path)
    _handle_freeze_result(result.returncode, context.log_path)

    _record_freeze_metadata(context.images_dir, pid, context.pipe_ids)

    logger.info("Goblin pid %s frozen successfully.", pid)
    return context.log_path

freeze_async(pid, images_dir, *, leave_running=True, log_path=None, verbosity=4, extra_args=None, shell_job=True) async

Asynchronously checkpoint a goblin process with CRIU.

Parameters:

Name Type Description Default
pid int

PID of the goblin process to dump.

 required
images_dir str | Path

Directory that will store the CRIU image set.

 required
leave_running bool

Keep the goblin alive after dumping. Defaults to True.

 True
log_path str | Path

Path for CRIU's log file. Defaults to images_dir / f"goblin-freeze.{pid}.log".

 None
verbosity int

CRIU verbosity level (0-4). Defaults to 4.

 4
extra_args Iterable[str]

Additional CRIU arguments to append verbatim.

 None
shell_job bool

Whether to pass --shell-job to CRIU. Defaults to True.

 True

Returns:

Type Description
Path

Path to the CRIU log file for the dump operation.

Raises:

Type Description
RuntimeError

If CRIU fails to dump the process.
ValueError

If pid is not positive.
Source code in src/pdum/criu/goblins/__init__.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
async def freeze_async(
    pid: int,
    images_dir: str | Path,
    *,
    leave_running: bool = True,
    log_path: str | Path | None = None,
    verbosity: int = 4,
    extra_args: Iterable[str] | None = None,
    shell_job: bool = True,
) -> Path:
    """Asynchronously checkpoint a goblin process with CRIU.

    Parameters
    ----------
    pid : int
        PID of the goblin process to dump.
    images_dir : str | Path
        Directory that will store the CRIU image set.
    leave_running : bool, optional
        Keep the goblin alive after dumping. Defaults to True.
    log_path : str | Path, optional
        Path for CRIU's log file. Defaults to ``images_dir / f"goblin-freeze.{pid}.log"``.
    verbosity : int, optional
        CRIU verbosity level (0-4). Defaults to 4.
    extra_args : Iterable[str], optional
        Additional CRIU arguments to append verbatim.
    shell_job : bool, optional
        Whether to pass ``--shell-job`` to CRIU. Defaults to True.

    Returns
    -------
    Path
        Path to the CRIU log file for the dump operation.

    Raises
    ------
    RuntimeError
        If CRIU fails to dump the process.
    ValueError
        If ``pid`` is not positive.
    """

    context = _build_freeze_context(
        pid,
        images_dir,
        leave_running=leave_running,
        log_path=log_path,
        verbosity=verbosity,
        extra_args=extra_args,
        shell_job=shell_job,
    )

    logger.debug("Running command (async): %s", shlex.join(context.command))
    logger.info("Freezing goblin pid %s into %s (async)", pid, images_dir)

    process = await asyncio.create_subprocess_exec(
        *context.command,
        stdin=asyncio.subprocess.DEVNULL,
        stdout=asyncio.subprocess.DEVNULL,
        stderr=asyncio.subprocess.DEVNULL,
        start_new_session=True,
    )
    returncode = await process.wait()
    _ensure_log_readable(context.log_path)
    _handle_freeze_result(returncode, context.log_path)

    _record_freeze_metadata(context.images_dir, pid, context.pipe_ids)
    logger.info("Goblin pid %s frozen successfully (async).", pid)
    return context.log_path

thaw(images_dir, *, extra_args=None, log_path=None, pidfile=None, shell_job=True, detach=False)

Restore a goblin synchronously and reconnect to its pipes.

Parameters:

Name Type Description Default
images_dir str | Path

Directory containing the CRIU image set to restore.

 required
extra_args Iterable[str]

Additional CRIU restore arguments to append verbatim.

 None
log_path str | Path

Override for the CRIU restore log file. Defaults to images_dir / goblin-thaw.<ts>.log.

 None
pidfile str | Path

Override for the CRIU --pidfile argument. Defaults to images_dir / goblin-thaw.<ts>.pid.

 None
shell_job bool

Whether to run CRIU with --shell-job (required if the target is attached to a TTY). Defaults to True.

 True
detach bool

Whether to pass -d to CRIU and let it run detached from the helper. Defaults to False.

 False

Returns:

Type Description
GoblinProcess

Handle that exposes stdio pipes plus metadata (log path, pidfile, helper pid). Call :meth:GoblinProcess.read_pidfile once CRIU writes the PID file.

Raises:

Type Description
ValueError

If shell_job and detach are both True.
RuntimeError

If CRIU fails to start.
Source code in src/pdum/criu/goblins/__init__.py 
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
def thaw(
    images_dir: str | Path,
    *,
    extra_args: Iterable[str] | None = None,
    log_path: str | Path | None = None,
    pidfile: str | Path | None = None,
    shell_job: bool = True,
    detach: bool = False,
) -> GoblinProcess:
    """Restore a goblin synchronously and reconnect to its pipes.

    Parameters
    ----------
    images_dir : str | Path
        Directory containing the CRIU image set to restore.
    extra_args : Iterable[str], optional
        Additional CRIU restore arguments to append verbatim.
    log_path : str | Path, optional
        Override for the CRIU restore log file. Defaults to ``images_dir / goblin-thaw.<ts>.log``.
    pidfile : str | Path, optional
        Override for the CRIU ``--pidfile`` argument. Defaults to ``images_dir / goblin-thaw.<ts>.pid``.
    shell_job : bool, optional
        Whether to run CRIU with ``--shell-job`` (required if the target is attached to a TTY). Defaults to True.
    detach : bool, optional
        Whether to pass ``-d`` to CRIU and let it run detached from the helper. Defaults to False.

    Returns
    -------
    GoblinProcess
        Handle that exposes stdio pipes plus metadata (log path, pidfile, helper pid).
        Call :meth:`GoblinProcess.read_pidfile` once CRIU writes the PID file.

    Raises
    ------
    ValueError
        If ``shell_job`` and ``detach`` are both True.
    RuntimeError
        If CRIU fails to start.
    """

    if shell_job and detach:
        raise ValueError("detach=True is incompatible with shell_job=True.")

    context = _build_thaw_context(
        images_dir,
        extra_args=extra_args,
        log_path=log_path,
        pidfile=pidfile,
        shell_job=shell_job,
        detach=detach,
    )
    pipes = _prepare_stdio_pipes(context.pipe_ids)

    restore_proc = _launch_criu_restore_sync(context, pipes)
    stdin, stdout, stderr = pipes.build_sync_streams()
    _reap_process_in_background(restore_proc)
    return GoblinProcess(
        helper_pid=restore_proc.pid,
        stdin=stdin,
        stdout=stdout,
        stderr=stderr,
        images_dir=context.images_dir,
        log_path=context.log_path,
        pidfile=context.pidfile,
    )

thaw_async(images_dir, *, extra_args=None, log_path=None, pidfile=None, shell_job=True, detach=False) async

Async variant of :func:thaw that returns asyncio streams.

Parameters:

Name Type Description Default
images_dir str | Path

Directory containing the CRIU image set to restore.

 required
extra_args Iterable[str]

Additional CRIU restore arguments to append verbatim.

 None
log_path str | Path

Override for the CRIU restore log file path.

 None
pidfile str | Path

Override for the CRIU --pidfile argument.

 None
shell_job bool

Whether to include --shell-job in the CRIU command. Defaults to True.

 True
detach bool

Whether to pass -d (detached mode) to CRIU. Defaults to False.

 False

Returns:

Type Description
AsyncGoblinProcess

Handle exposing asyncio streams plus metadata about the restore. Use :meth:AsyncGoblinProcess.read_pidfile after CRIU writes the PID file.

Raises:

Type Description
ValueError

If shell_job and detach are both True.
RuntimeError

If CRIU fails to start.
Source code in src/pdum/criu/goblins/__init__.py 
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
async def thaw_async(
    images_dir: str | Path,
    *,
    extra_args: Iterable[str] | None = None,
    log_path: str | Path | None = None,
    pidfile: str | Path | None = None,
    shell_job: bool = True,
    detach: bool = False,
) -> AsyncGoblinProcess:
    """Async variant of :func:`thaw` that returns asyncio streams.

    Parameters
    ----------
    images_dir : str | Path
        Directory containing the CRIU image set to restore.
    extra_args : Iterable[str], optional
        Additional CRIU restore arguments to append verbatim.
    log_path : str | Path, optional
        Override for the CRIU restore log file path.
    pidfile : str | Path, optional
        Override for the CRIU ``--pidfile`` argument.
    shell_job : bool, optional
        Whether to include ``--shell-job`` in the CRIU command. Defaults to True.
    detach : bool, optional
        Whether to pass ``-d`` (detached mode) to CRIU. Defaults to False.

    Returns
    -------
    AsyncGoblinProcess
        Handle exposing asyncio streams plus metadata about the restore. Use
        :meth:`AsyncGoblinProcess.read_pidfile` after CRIU writes the PID file.

    Raises
    ------
    ValueError
        If ``shell_job`` and ``detach`` are both True.
    RuntimeError
        If CRIU fails to start.
    """

    if shell_job and detach:
        raise ValueError("detach=True is incompatible with shell_job=True.")

    context = _build_thaw_context(
        images_dir,
        extra_args=extra_args,
        log_path=log_path,
        pidfile=pidfile,
        shell_job=shell_job,
        detach=detach,
    )
    pipes = _prepare_stdio_pipes(context.pipe_ids)

    restore_proc = await _launch_criu_restore_async(context, pipes)
    stdin_writer = await _make_writer_from_fd(pipes.parent_stdin_fd)
    stdout_reader = await _make_reader_from_fd(pipes.parent_stdout_fd)
    stderr_reader = await _make_reader_from_fd(pipes.parent_stderr_fd)
    _schedule_async_reap(restore_proc)

    return AsyncGoblinProcess(
        helper_pid=restore_proc.pid,
        stdin=stdin_writer,
        stdout=stdout_reader,
        stderr=stderr_reader,
        images_dir=context.images_dir,
        log_path=context.log_path,
        pidfile=context.pidfile,
    )

utils

Helpers for locating CRIU-related executables on the system.

check_sudo_closefrom()

Probe sudo for closefrom_override support without raising.

Source code in src/pdum/criu/utils.py 
368
369
370
371
372
373
374
375
376
377
def check_sudo_closefrom() -> tuple[bool, str | None]:
    """
    Probe ``sudo`` for closefrom_override support without raising.
    """

    try:
        ensure_sudo_closefrom()
    except RuntimeError as exc:
        return False, str(exc)
    return True, None

doctor_check_results(verbose=True)

Run the same checks as the CLI doctor command.

Returns a list of (label, ok, message) tuples.

Source code in src/pdum/criu/utils.py 
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
def doctor_check_results(verbose: bool = True) -> list[tuple[str, bool, str | None]]:
    """Run the same checks as the CLI doctor command.

    Returns a list of ``(label, ok, message)`` tuples.
    """

    checks = [
        ("Password-less sudo", ensure_sudo),
        ("CRIU", ensure_criu),
        ("CRIU-ns", ensure_criu_ns),
        ("pgrep", ensure_pgrep),
    ]

    results: list[tuple[str, bool, str | None]] = []
    for label, checker in checks:
        try:
            ok = bool(checker(verbose=verbose))
            results.append((label, ok, None))
        except Exception as exc:  # pragma: no cover - best-effort reporting
            results.append((label, False, str(exc)))

    closefrom_ok, closefrom_msg = check_sudo_closefrom()
    results.append(("sudo closefrom_override", closefrom_ok, closefrom_msg))
    return results

ensure_criu(*, verbose=True, raise_=False, **kwargs)

Ensure the criu executable is available.

Source code in src/pdum/criu/utils.py 
142
143
144
145
146
147
148
149
150
151
def ensure_criu(*, verbose: bool = True, raise_: bool = False, **kwargs: Any) -> str | None:
    """Ensure the ``criu`` executable is available."""

    raise_flag = _pop_raise_flag(kwargs, raise_)
    return _ensure_tool(
        "criu",
        "Install CRIU on Ubuntu with 'sudo apt update && sudo apt install -y criu'.",
        verbose=verbose,
        raise_flag=raise_flag,
    )

ensure_criu_ns(*, verbose=True, raise_=False, **kwargs)

Ensure the criu-ns helper is available.

Source code in src/pdum/criu/utils.py 
154
155
156
157
158
159
160
161
162
163
def ensure_criu_ns(*, verbose: bool = True, raise_: bool = False, **kwargs: Any) -> str | None:
    """Ensure the ``criu-ns`` helper is available."""

    raise_flag = _pop_raise_flag(kwargs, raise_)
    return _ensure_tool(
        "criu-ns",
        "Install the CRIU tools on Ubuntu with 'sudo apt install -y criu'.",
        verbose=verbose,
        raise_flag=raise_flag,
    )

ensure_linux()

Raise if the current platform is not Linux.

Source code in src/pdum/criu/utils.py 
264
265
266
267
268
def ensure_linux() -> None:
    """Raise if the current platform is not Linux."""

    if not sys.platform.startswith("linux"):
        raise RuntimeError(f"CRIU workflows require Linux (detected {sys.platform}).")

ensure_pgrep(*, verbose=True, raise_=False, **kwargs)

Ensure the pgrep utility is available.

Source code in src/pdum/criu/utils.py 
166
167
168
169
170
171
172
173
174
175
def ensure_pgrep(*, verbose: bool = True, raise_: bool = False, **kwargs: Any) -> str | None:
    """Ensure the ``pgrep`` utility is available."""

    raise_flag = _pop_raise_flag(kwargs, raise_)
    return _ensure_tool(
        "pgrep",
        "Install pgrep via the procps package on Ubuntu: 'sudo apt install -y procps'.",
        verbose=verbose,
        raise_flag=raise_flag,
    )

ensure_sudo(*, verbose=True, raise_=False, **kwargs)

Ensure sudo -n true succeeds on the current system.

Returns:

Type Description
bool

True if the non-interactive sudo command exits with status 0, otherwise False.
Source code in src/pdum/criu/utils.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def ensure_sudo(*, verbose: bool = True, raise_: bool = False, **kwargs: Any) -> bool:
    """
    Ensure ``sudo -n true`` succeeds on the current system.

    Returns
    -------
    bool
        True if the non-interactive sudo command exits with status 0, otherwise False.
    """

    raise_flag = _pop_raise_flag(kwargs, raise_)

    try:
        sudo_cmd = resolve_command("sudo")
        true_cmd = resolve_command("true")
    except (FileNotFoundError, ValueError) as exc:
        message = f"Unable to locate sudo/true commands: {exc}"
        if verbose:
            logger.warning(message)
        if raise_flag:
            raise RuntimeError(message) from exc
        return False

    try:
        result = subprocess.run(
            [sudo_cmd, "-n", true_cmd],
            check=False,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
    except OSError as exc:
        message = f"Failed to execute sudo: {exc}"
        if verbose:
            logger.warning(message)
        if raise_flag:
            raise RuntimeError(message) from exc
        return False

    if result.returncode == 0:
        return True

    user = os.environ.get("USER", "your-user")
    message = (
        "Password-less sudo is required to continue.\n"
        f"Tip: run 'sudo visudo' and add '{user} ALL=(ALL) NOPASSWD:ALL'."
    )
    if verbose:
        logger.warning(message)

    if raise_flag:
        raise RuntimeError("Password-less sudo is required for pdum-criu operations.")
    return False

ensure_sudo_closefrom()

Verify that sudo supports the -C flag (closefrom_override).

Source code in src/pdum/criu/utils.py 
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
def ensure_sudo_closefrom() -> None:
    """
    Verify that ``sudo`` supports the ``-C`` flag (closefrom_override).
    """

    global _SUDO_CLOSEFROM_SUPPORTED, _SUDO_CLOSEFROM_ERROR

    if _SUDO_CLOSEFROM_SUPPORTED:
        return
    if _SUDO_CLOSEFROM_SUPPORTED is False:
        raise RuntimeError(_SUDO_CLOSEFROM_ERROR or "sudo closefrom_override is not enabled.")

    sudo_path = resolve_command("sudo")
    result = subprocess.run([sudo_path, "-n", "-C", "32", "true"], capture_output=True, text=True)
    if result.returncode == 0:
        _SUDO_CLOSEFROM_SUPPORTED = True
        _SUDO_CLOSEFROM_ERROR = None
        return

    detail = (result.stderr or result.stdout or "sudo rejected -C").strip()
    _SUDO_CLOSEFROM_SUPPORTED = False
    _SUDO_CLOSEFROM_ERROR = (
        "sudo is not configured with closefrom_override; enable it via `sudo visudo` "
        "(add `Defaults    closefrom_override` or `Defaults:YOURUSER    closefrom_override`). "
        f"Original sudo output: {detail}"
    )
    raise RuntimeError(_SUDO_CLOSEFROM_ERROR)

psgrep(query, *, ensure_unique=True)

Locate processes matching the supplied query using pgrep -f.

Parameters:

Name Type Description Default
query str

Pattern passed to pgrep -f. Supports spaces (matches full command line).

 required
ensure_unique bool

When True, ensure exactly one PID is returned; otherwise, return all matches.

 True

Returns:

Type Description
int | list[int]

PID of the unique match, or a list of PIDs when ensure_unique is False.

Raises:

Type Description
ValueError

If query is empty.
RuntimeError

If no processes match, multiple matches are found while ensure_unique is True, or pgrep fails.
Source code in src/pdum/criu/utils.py 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def psgrep(query: str, *, ensure_unique: bool = True) -> int | list[int]:
    """
    Locate processes matching the supplied query using ``pgrep -f``.

    Parameters
    ----------
    query : str
        Pattern passed to ``pgrep -f``. Supports spaces (matches full command line).
    ensure_unique : bool, optional
        When True, ensure exactly one PID is returned; otherwise, return all matches.

    Returns
    -------
    int | list[int]
        PID of the unique match, or a list of PIDs when ``ensure_unique`` is False.

    Raises
    ------
    ValueError
        If ``query`` is empty.
    RuntimeError
        If no processes match, multiple matches are found while ``ensure_unique`` is
        True, or ``pgrep`` fails.
    """

    if not query or query.strip() == "":
        raise ValueError("Query must be a non-empty string.")

    pgrep_cmd = ensure_pgrep(verbose=False, raise_=True)
    result = subprocess.run(
        [pgrep_cmd, "-f", query],
        check=False,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
    )

    if result.returncode == 1:
        raise RuntimeError(f"No processes matched query {query!r}.")
    if result.returncode not in (0, 1):
        stderr = result.stderr.strip()
        raise RuntimeError(f"pgrep failed with exit code {result.returncode}: {stderr}")

    pids = [int(line) for line in result.stdout.splitlines() if line.strip()]
    if not pids:
        raise RuntimeError(f"No processes matched query {query!r}.")

    if ensure_unique:
        if len(pids) > 1:
            raise RuntimeError(
                f"Expected a single process for {query!r}, found {len(pids)} matches: {pids}"
            )
        return pids[0]

    return pids

resolve_command(executable)

Resolve a supported command to a concrete executable path.

The resolver first checks PDUM_CRIU_<EXE> for an override (where <EXE> is the capitalized executable name with non-alphanumerics replaced by underscores) before falling back to shutil.which.

Parameters:

Name Type Description Default
executable str

Default executable name to locate. Can be overridden via environment.

 required

Returns:

Type Description
str

Absolute path to the resolved executable.

Raises:

Type Description
ValueError

If executable is empty.
FileNotFoundError

If the executable cannot be located.
Source code in src/pdum/criu/utils.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def resolve_command(executable: str) -> str:
    """
    Resolve a supported command to a concrete executable path.

    The resolver first checks ``PDUM_CRIU_<EXE>`` for an override (where ``<EXE>``
    is the capitalized executable name with non-alphanumerics replaced by
    underscores) before falling back to ``shutil.which``.

    Parameters
    ----------
    executable : str
        Default executable name to locate. Can be overridden via environment.

    Returns
    -------
    str
        Absolute path to the resolved executable.

    Raises
    ------
    ValueError
        If ``executable`` is empty.
    FileNotFoundError
        If the executable cannot be located.
    """

    if not executable or executable.strip() == "":
        raise ValueError("Executable name must be a non-empty string.")

    default_executable = executable.strip()
    env_var = _env_var_name(default_executable)
    override = os.environ.get(env_var, "").strip()
    candidate = override or default_executable

    resolved = which(candidate)
    if resolved:
        return resolved

    raise FileNotFoundError(
        f"Unable to locate executable for {default_executable!r} "
        f"(checked {candidate!r}, override via {env_var})."
    )

resolve_target_pid(pid, pattern)

Resolve a target PID either directly or via pgrep.

Source code in src/pdum/criu/utils.py 
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
def resolve_target_pid(pid: int | None, pattern: str | None) -> int:
    """Resolve a target PID either directly or via ``pgrep``."""

    if pid is not None and pattern is not None:
        raise ValueError("Specify either --pid or --pgrep, not both.")
    if pid is None and pattern is None:
        raise ValueError("Either --pid or --pgrep is required.")
    if pid is not None:
        if pid <= 0:
            raise ValueError("PID must be a positive integer.")
        return pid
    assert pattern is not None
    resolved = psgrep(pattern, ensure_unique=True)
    if isinstance(resolved, list):
        raise RuntimeError("resolve_target_pid expected a single PID result.")
    return resolved

spawn_directory_cleanup(path, watched_pid)

Spawn a background helper that removes path when watched_pid exits.

Source code in src/pdum/criu/utils.py 
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
def spawn_directory_cleanup(path: Path, watched_pid: int) -> None:
    """
    Spawn a background helper that removes ``path`` when ``watched_pid`` exits.
    """

    script = textwrap.dedent(
        """
        import os
        import shutil
        import sys
        import time

        target = sys.argv[1]
        watched = int(sys.argv[2])

        while True:
            try:
                os.kill(watched, 0)
            except OSError:
                break
            time.sleep(0.5)

        try:
            shutil.rmtree(target)
        except FileNotFoundError:
            pass
        """
    )

    subprocess.Popen(
        [sys.executable, "-c", script, os.fspath(path), str(watched_pid)],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        close_fds=True,
    )

tail_file(path, lines=10)

Return the last lines lines from path.

Source code in src/pdum/criu/utils.py 
289
290
291
292
293
294
295
296
297
298
299
def tail_file(path: Path, lines: int = 10) -> str:
    """Return the last ``lines`` lines from ``path``."""

    if not path.exists():
        return ""

    recent: deque[str] = deque(maxlen=lines)
    with path.open("r", encoding="utf-8", errors="replace") as handle:
        for line in handle:
            recent.append(line.rstrip("\n"))
    return "\n".join(recent)