1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
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
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
|
import time
from typing import Iterator, Optional
from .exceptions import SysCallError
from .general import SysCommand, SysCommandWorker, locate_binary
from .installer import Installer
from .output import error
from .storage import storage
class Boot:
def __init__(self, installation: Installer):
self.instance = installation
self.container_name = 'archinstall'
self.session: Optional[SysCommandWorker] = None
self.ready = False
def __enter__(self) -> 'Boot':
if (existing_session := storage.get('active_boot', None)) and existing_session.instance != self.instance:
raise KeyError("Archinstall only supports booting up one instance, and a active session is already active and it is not this one.")
if existing_session:
self.session = existing_session.session
self.ready = existing_session.ready
else:
# '-P' or --console=pipe could help us not having to do a bunch
# of os.write() calls, but instead use pipes (stdin, stdout and stderr) as usual.
self.session = SysCommandWorker([
'/usr/bin/systemd-nspawn',
'-D', str(self.instance.target),
'--timezone=off',
'-b',
'--no-pager',
'--machine', self.container_name
])
if not self.ready and self.session:
while self.session.is_alive():
if b' login:' in self.session:
self.ready = True
break
storage['active_boot'] = self
return self
def __exit__(self, *args :str, **kwargs :str) -> None:
# b''.join(sys_command('sync')) # No need to, since the underlying fs() object will call sync.
# TODO: https://stackoverflow.com/questions/28157929/how-to-safely-handle-an-exception-inside-a-context-manager
if len(args) >= 2 and args[1]:
error(
args[1],
f"The error above occurred in a temporary boot-up of the installation {self.instance}"
)
shutdown = None
shutdown_exit_code: Optional[int] = -1
try:
shutdown = SysCommand(f'systemd-run --machine={self.container_name} --pty shutdown now')
except SysCallError as err:
shutdown_exit_code = err.exit_code
if self.session:
while self.session.is_alive():
time.sleep(0.25)
if shutdown and shutdown.exit_code:
shutdown_exit_code = shutdown.exit_code
if self.session and (self.session.exit_code == 0 or shutdown_exit_code == 0):
storage['active_boot'] = None
else:
session_exit_code = self.session.exit_code if self.session else -1
raise SysCallError(
f"Could not shut down temporary boot of {self.instance}: {session_exit_code}/{shutdown_exit_code}",
exit_code=next(filter(bool, [session_exit_code, shutdown_exit_code]))
)
def __iter__(self) -> Iterator[bytes]:
if self.session:
for value in self.session:
yield value
def __contains__(self, key: bytes) -> bool:
if self.session is None:
return False
return key in self.session
def is_alive(self) -> bool:
if self.session is None:
return False
return self.session.is_alive()
def SysCommand(self, cmd: list, *args, **kwargs) -> SysCommand:
if cmd[0][0] != '/' and cmd[0][:2] != './':
# This check is also done in SysCommand & SysCommandWorker.
# However, that check is done for `machinectl` and not for our chroot command.
# So this wrapper for SysCommand will do this additionally.
cmd[0] = locate_binary(cmd[0])
return SysCommand(["systemd-run", f"--machine={self.container_name}", "--pty", *cmd], *args, **kwargs)
def SysCommandWorker(self, cmd: list, *args, **kwargs) -> SysCommandWorker:
if cmd[0][0] != '/' and cmd[0][:2] != './':
cmd[0] = locate_binary(cmd[0])
return SysCommandWorker(["systemd-run", f"--machine={self.container_name}", "--pty", *cmd], *args, **kwargs)
|