Skip to content

kraken.std.sccache

kraken.std.sccache

SccacheManager dataclass

Source code in kraken/std/sccache.py 
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
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
@dataclasses.dataclass
class SccacheManager:
    cache_config: AzureBlobStorageCache | LocalCache | None
    log_level: str | None = None
    log_file: Path | None = None
    bin: Path | None = None

    def __post_init__(self) -> None:
        self._proc: sp.Popen[Any] | None = None

    def is_running(self) -> bool:
        return self._proc is not None and self._proc.returncode is None

    def get_cache_location(self) -> str:
        stats = self.stats()
        match = re.search(r"^Cache\s+location\s+(.*)$", stats, re.I | re.M)
        assert match is not None, f"Could not determine cache location from sccache stats output: {stats!r}"
        return match.group(1)

    def start(self) -> None:
        """Start the Sccache server."""

        if self.is_running():
            assert self._proc is not None
            raise RuntimeError(f"Sccache is already running (pid: {self._proc.pid})")

        command = [str(self.bin) if self.bin else "sccache"]
        env = {}
        if self.cache_config:
            env.update(self.cache_config.to_env())
        if self.log_level is not None:
            env["SCCACHE_LOG"] = self.log_level
        if self.log_file is not None:
            env["SCCACHE_ERROR_LOG"] = str(self.log_file.absolute())
        env["SCCACHE_START_SERVER"] = "1"
        self._proc = sp.Popen(command, env={**os.environ, **env})

    def stats(self) -> str:
        command = [str(self.bin) if self.bin else "sccache", "-s"]
        return sp.check_output(command).decode()

    def stop(self, show_stats: bool = False) -> None:
        """Stop the Sccache server."""

        if not self.is_running():
            return
        assert self._proc is not None

        env = {"SCCACHE_NO_DAEMON": "1"}
        command = [str(self.bin) if self.bin else "sccache", "--stop-server"]
        sp.check_call(command, env={**os.environ, **env}, stdout=None if show_stats else sp.DEVNULL)

        self._proc.wait(10)
        if self.is_running():
            self._proc.terminate()
start 
start() -> None

Start the Sccache server.

Source code in kraken/std/sccache.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def start(self) -> None:
    """Start the Sccache server."""

    if self.is_running():
        assert self._proc is not None
        raise RuntimeError(f"Sccache is already running (pid: {self._proc.pid})")

    command = [str(self.bin) if self.bin else "sccache"]
    env = {}
    if self.cache_config:
        env.update(self.cache_config.to_env())
    if self.log_level is not None:
        env["SCCACHE_LOG"] = self.log_level
    if self.log_file is not None:
        env["SCCACHE_ERROR_LOG"] = str(self.log_file.absolute())
    env["SCCACHE_START_SERVER"] = "1"
    self._proc = sp.Popen(command, env={**os.environ, **env})
stop 
stop(show_stats: bool = False) -> None

Stop the Sccache server.

Source code in kraken/std/sccache.py 
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def stop(self, show_stats: bool = False) -> None:
    """Stop the Sccache server."""

    if not self.is_running():
        return
    assert self._proc is not None

    env = {"SCCACHE_NO_DAEMON": "1"}
    command = [str(self.bin) if self.bin else "sccache", "--stop-server"]
    sp.check_call(command, env={**os.environ, **env}, stdout=None if show_stats else sp.DEVNULL)

    self._proc.wait(10)
    if self.is_running():
        self._proc.terminate()

SccacheTask

Bases: BackgroundTask

This task ensures that an Sccache server is running for all its dependant tasks.

Source code in kraken/std/sccache.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
class SccacheTask(BackgroundTask):
    """This task ensures that an Sccache server is running for all its dependant tasks."""

    description = "Start sccache in the background."
    manager: Property[SccacheManager]
    fail_fast: Property[bool] = Property.default(False)

    def start_background_task(self, exit_stack: contextlib.ExitStack) -> TaskStatus:
        manager = self.manager.get()
        if not manager.is_running():
            manager.start()
        exit_stack.callback(lambda: manager.stop(show_stats=True))
        return TaskStatus.started(manager.get_cache_location())

sccache 

sccache(
    manager: SccacheManager,
    *,
    name: str = "sccache",
    group: str | None = None,
    project: Project | None = None
) -> SccacheTask

Creates a background task that starts the sccache server.

Source code in kraken/std/sccache.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
def sccache(
    manager: SccacheManager,
    *,
    name: str = "sccache",
    group: str | None = None,
    project: Project | None = None,
) -> SccacheTask:
    """Creates a background task that starts the sccache server."""

    project = project or Project.current()
    task = project.task(name, SccacheTask, group=group)
    task.manager = manager
    return task