RustPython/Lib/_android_support.py at main · RustPython/RustPython · GitHub

History

192 lines (155 loc) · 7.24 KB

Raw

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

import io

import sys

from threading import RLock

from time import sleep, time

# The maximum length of a log message in bytes, including the level marker and

# tag, is defined as LOGGER_ENTRY_MAX_PAYLOAD at

# https://cs.android.com/android/platform/superproject/+/android-14.0.0_r1:system/logging/liblog/include/log/log.h;l=71.

# Messages longer than this will be truncated by logcat. This limit has already

# been reduced at least once in the history of Android (from 4076 to 4068 between

# API level 23 and 26), so leave some headroom.

MAX_BYTES_PER_WRITE = 4000

# UTF-8 uses a maximum of 4 bytes per character, so limiting text writes to this

# size ensures that we can always avoid exceeding MAX_BYTES_PER_WRITE.

# However, if the actual number of bytes per character is smaller than that,

# then we may still join multiple consecutive text writes into binary

# writes containing a larger number of characters.

MAX_CHARS_PER_WRITE = MAX_BYTES_PER_WRITE // 4

# When embedded in an app on current versions of Android, there's no easy way to

# monitor the C-level stdout and stderr. The testbed comes with a .c file to

# redirect them to the system log using a pipe, but that wouldn't be convenient

# or appropriate for all apps. So we redirect at the Python level instead.

def init_streams(android_log_write, stdout_prio, stderr_prio):

if sys.executable:

return # Not embedded in an app.

global logcat

logcat = Logcat(android_log_write)

sys.stdout = TextLogStream(stdout_prio, "python.stdout", sys.stdout)

sys.stderr = TextLogStream(stderr_prio, "python.stderr", sys.stderr)

class TextLogStream(io.TextIOWrapper):

def __init__(self, prio, tag, original=None, **kwargs):

# Respect the -u option.

if original:

kwargs.setdefault("write_through", original.write_through)

fileno = original.fileno()

else:

fileno = None

# The default is surrogateescape for stdout and backslashreplace for

# stderr, but in the context of an Android log, readability is more

# important than reversibility.

kwargs.setdefault("encoding", "UTF-8")

kwargs.setdefault("errors", "backslashreplace")

super().__init__(BinaryLogStream(prio, tag, fileno), **kwargs)

self._lock = RLock()

self._pending_bytes = []

self._pending_bytes_count = 0

def __repr__(self):

return f"<TextLogStream {self.buffer.tag!r}>"

def write(self, s):

if not isinstance(s, str):

raise TypeError(

f"write() argument must be str, not {type(s).__name__}")

# In case `s` is a str subclass that writes itself to stdout or stderr

# when we call its methods, convert it to an actual str.

s = str.__str__(s)

# We want to emit one log message per line wherever possible, so split

# the string into lines first. Note that "".splitlines() == [], so

# nothing will be logged for an empty string.

with self._lock:

for line in s.splitlines(keepends=True):

while line:

chunk = line[:MAX_CHARS_PER_WRITE]

line = line[MAX_CHARS_PER_WRITE:]

self._write_chunk(chunk)

return len(s)

# The size and behavior of TextIOWrapper's buffer is not part of its public

# API, so we handle buffering ourselves to avoid truncation.

def _write_chunk(self, s):

b = s.encode(self.encoding, self.errors)

if self._pending_bytes_count + len(b) > MAX_BYTES_PER_WRITE:

self.flush()

self._pending_bytes.append(b)

self._pending_bytes_count += len(b)

if (

self.write_through

or b.endswith(b"\n")

or self._pending_bytes_count > MAX_BYTES_PER_WRITE

self.flush()

def flush(self):

with self._lock:

self.buffer.write(b"".join(self._pending_bytes))

self._pending_bytes.clear()

self._pending_bytes_count = 0

# Since this is a line-based logging system, line buffering cannot be turned

# off, i.e. a newline always causes a flush.

@property

def line_buffering(self):

return True

class BinaryLogStream(io.RawIOBase):

def __init__(self, prio, tag, fileno=None):

self.prio = prio

self.tag = tag

self._fileno = fileno

def __repr__(self):

return f"<BinaryLogStream {self.tag!r}>"

def writable(self):

return True

def write(self, b):

if type(b) is not bytes:

try:

b = bytes(memoryview(b))

except TypeError:

raise TypeError(

f"write() argument must be bytes-like, not {type(b).__name__}"

) from None

# Writing an empty string to the stream should have no effect.

if b:

logcat.write(self.prio, self.tag, b)

return len(b)

# This is needed by the test suite --timeout option, which uses faulthandler.

def fileno(self):

if self._fileno is None:

raise io.UnsupportedOperation("fileno")

return self._fileno

# When a large volume of data is written to logcat at once, e.g. when a test

# module fails in --verbose3 mode, there's a risk of overflowing logcat's own

# buffer and losing messages. We avoid this by imposing a rate limit using the

# token bucket algorithm, based on a conservative estimate of how fast `adb

# logcat` can consume data.

MAX_BYTES_PER_SECOND = 1024 * 1024

# The logcat buffer size of a device can be determined by running `logcat -g`.

# We set the token bucket size to half of the buffer size of our current minimum

# API level, because other things on the system will be producing messages as

# well.

BUCKET_SIZE = 128 * 1024

# https://cs.android.com/android/platform/superproject/+/android-14.0.0_r1:system/logging/liblog/include/log/log_read.h;l=39

PER_MESSAGE_OVERHEAD = 28

class Logcat:

def __init__(self, android_log_write):

self.android_log_write = android_log_write

self._lock = RLock()

self._bucket_level = 0

self._prev_write_time = time()

def write(self, prio, tag, message):

# Encode null bytes using "modified UTF-8" to avoid them truncating the

# message.

message = message.replace(b"\x00", b"\xc0\x80")

# On API level 30 and higher, Logcat will strip any number of leading

# newlines. This is visible in all `logcat` modes, even --binary. Work

# around this by adding a leading space, which shouldn't make any

# difference to the log's usability.

if message.startswith(b"\n"):

message = b" " + message

with self._lock:

now = time()

self._bucket_level += (

(now - self._prev_write_time) * MAX_BYTES_PER_SECOND)

# If the bucket level is still below zero, the clock must have gone

# backwards, so reset it to zero and continue.

self._bucket_level = max(0, min(self._bucket_level, BUCKET_SIZE))

self._prev_write_time = now

self._bucket_level -= PER_MESSAGE_OVERHEAD + len(tag) + len(message)

if self._bucket_level < 0:

sleep(-self._bucket_level / MAX_BYTES_PER_SECOND)

self.android_log_write(prio, tag, message)

RustPython/Lib/_android_support.py at main · RustPython/RustPython · GitHub

Uh oh!

Expand file tree

Breadcrumbs

_android_support.py

Latest commit

History

Breadcrumbs

_android_support.py

File metadata and controls