mirror of
https://github.com/trezor/trezor-firmware.git
synced 2024-12-02 20:48:30 +00:00
docs(core): add and modify docs to context and cache
[no changelog]
This commit is contained in:
parent
2eab963862
commit
01cf58f2a1
@ -11,6 +11,16 @@ if TYPE_CHECKING:
|
||||
|
||||
|
||||
def stored(key: int) -> Callable[[ByteFunc[P]], ByteFunc[P]]:
|
||||
"""
|
||||
Caches the result of a function call based on the given key.
|
||||
|
||||
- If the key is already present in the cache, the cached value is returned
|
||||
directly without invoking the decorated function.
|
||||
|
||||
- If the key is not present in the cache, the decorated function is executed,
|
||||
and its result is stored in the cache before being returned to the caller.
|
||||
"""
|
||||
|
||||
def decorator(func: ByteFunc[P]) -> ByteFunc[P]:
|
||||
|
||||
def wrapper(*args: P.args, **kwargs: P.kwargs) -> bytes:
|
||||
@ -26,6 +36,17 @@ def stored(key: int) -> Callable[[ByteFunc[P]], ByteFunc[P]]:
|
||||
|
||||
|
||||
def stored_async(key: int) -> Callable[[AsyncByteFunc[P]], AsyncByteFunc[P]]:
|
||||
"""
|
||||
Caches the result of an async function call based on the given key.
|
||||
|
||||
- If the key is already present in the cache, the cached value is returned
|
||||
directly without invoking the decorated asynchronous function.
|
||||
|
||||
- If the key is not present in the cache, the decorated asynchronous function
|
||||
is executed, and its result is stored in the cache before being returned
|
||||
to the caller.
|
||||
"""
|
||||
|
||||
def decorator(func: AsyncByteFunc[P]) -> AsyncByteFunc[P]:
|
||||
async def wrapper(*args: P.args, **kwargs: P.kwargs) -> bytes:
|
||||
value = context.cache_get(key)
|
||||
|
@ -4,7 +4,6 @@ import gc
|
||||
from storage import cache_codec
|
||||
from storage.cache_common import SESSIONLESS_FLAG, SessionlessCache
|
||||
|
||||
|
||||
# Cache initialization
|
||||
_SESSIONLESS_CACHE = SessionlessCache()
|
||||
_PROTOCOL_CACHE = cache_codec
|
||||
@ -15,6 +14,9 @@ gc.collect()
|
||||
|
||||
|
||||
def clear_all() -> None:
|
||||
"""
|
||||
Clears all data from both the protocol cache and the sessionless cache.
|
||||
"""
|
||||
global autolock_last_touch
|
||||
autolock_last_touch = None
|
||||
_SESSIONLESS_CACHE.clear()
|
||||
@ -22,6 +24,13 @@ def clear_all() -> None:
|
||||
|
||||
|
||||
def get_int_all_sessions(key: int) -> builtins.set[int]:
|
||||
"""
|
||||
Returns set of int values associated with a given key from all relevant sessions.
|
||||
|
||||
If the key has the `SESSIONLESS_FLAG` set, the values are retrieved
|
||||
from the sessionless cache. Otherwise, the values are fetched
|
||||
from the protocol cache.
|
||||
"""
|
||||
if key & SESSIONLESS_FLAG:
|
||||
values = builtins.set()
|
||||
encoded = _SESSIONLESS_CACHE.get(key)
|
||||
|
@ -16,6 +16,11 @@ SESSION_ID_LENGTH = const(32)
|
||||
|
||||
|
||||
class SessionCache(DataCache):
|
||||
"""
|
||||
A cache for storing values that depend on seed derivation
|
||||
or are specific to a `protocol_v1` session.
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.session_id = bytearray(SESSION_ID_LENGTH)
|
||||
if utils.BITCOIN_ONLY:
|
||||
|
@ -36,6 +36,11 @@ class InvalidSessionError(Exception):
|
||||
|
||||
|
||||
class DataCache:
|
||||
"""
|
||||
A single unit of cache storage, designed to store common-type
|
||||
values efficiently in bytearrays in a sequential manner.
|
||||
"""
|
||||
|
||||
fields: Sequence[int] # field sizes
|
||||
|
||||
def __init__(self) -> None:
|
||||
@ -111,6 +116,11 @@ class DataCache:
|
||||
|
||||
|
||||
class SessionlessCache(DataCache):
|
||||
"""
|
||||
A cache for values that are independent of both
|
||||
passphrase seed derivation and the active session.
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.fields = (
|
||||
64, # APP_COMMON_SEED_WITHOUT_PASSPHRASE
|
||||
|
@ -129,6 +129,7 @@ if __debug__:
|
||||
mem_info(True)
|
||||
|
||||
def get_bytes_as_str(a: bytes) -> str:
|
||||
"""Converts the provided bytes to a hexadecimal string (decoded as`utf-8`)."""
|
||||
return hexlify(a).decode("utf-8")
|
||||
|
||||
|
||||
|
@ -16,11 +16,7 @@ if TYPE_CHECKING:
|
||||
|
||||
|
||||
class CodecContext(Context):
|
||||
"""Wire context.
|
||||
|
||||
Represents USB communication inside a particular session on a particular interface
|
||||
(i.e., wire, debug, single BT connection, etc.)
|
||||
"""
|
||||
""" "Wire context" for `protocol_v1`."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
@ -39,12 +35,6 @@ class CodecContext(Context):
|
||||
expected_types: Container[int],
|
||||
expected_type: type[protobuf.MessageType] | None = None,
|
||||
) -> protobuf.MessageType:
|
||||
"""Read a message from the wire.
|
||||
|
||||
The read message must be of one of the types specified in `expected_types`.
|
||||
If only a single type is expected, it can be passed as `expected_type`,
|
||||
to save on having to decode the type code into a protobuf class.
|
||||
"""
|
||||
if __debug__:
|
||||
log.debug(
|
||||
__name__,
|
||||
@ -78,7 +68,6 @@ class CodecContext(Context):
|
||||
return wrap_protobuf_load(msg.data, expected_type)
|
||||
|
||||
async def write(self, msg: protobuf.MessageType) -> None:
|
||||
"""Write a message to the wire."""
|
||||
if __debug__:
|
||||
log.debug(
|
||||
__name__,
|
||||
|
@ -47,7 +47,7 @@ def wrap_protobuf_load(
|
||||
|
||||
|
||||
async def handle_single_message(ctx: Context, msg: Message) -> bool:
|
||||
"""Handle a message that was loaded from USB by the caller.
|
||||
"""Handle a message that was loaded from a WireInterface by the caller.
|
||||
|
||||
Find the appropriate handler, run it and write its result on the wire. In case
|
||||
a problem is encountered at any point, write the appropriate error on the wire.
|
||||
|
@ -13,6 +13,11 @@ if TYPE_CHECKING:
|
||||
|
||||
|
||||
class Message:
|
||||
"""
|
||||
Encapsulates protobuf encoded message, where
|
||||
- `type` is the `WIRE_TYPE` of the message
|
||||
- `data` is the protobuf encoded message
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
@ -24,6 +29,13 @@ class Message:
|
||||
|
||||
|
||||
class Context:
|
||||
"""Wire context.
|
||||
|
||||
Represents communication between the Trezor device and a host within
|
||||
a specific session over a particular interface (i.e., wire, debug,
|
||||
single Bluetooth connection, etc.).
|
||||
"""
|
||||
|
||||
channel_id: bytes
|
||||
|
||||
def __init__(self, iface: WireInterface, channel_id: bytes | None = None) -> None:
|
||||
@ -47,15 +59,25 @@ class Context:
|
||||
self,
|
||||
expected_types: Container[int],
|
||||
expected_type: type[protobuf.MessageType] | None = None,
|
||||
) -> protobuf.MessageType: ...
|
||||
) -> protobuf.MessageType:
|
||||
"""Read a message from the wire.
|
||||
|
||||
async def write(self, msg: protobuf.MessageType) -> None: ...
|
||||
The read message must be of one of the types specified in `expected_types`.
|
||||
If only a single type is expected, it can be passed as `expected_type`,
|
||||
to save on having to decode the type code into a protobuf class.
|
||||
"""
|
||||
...
|
||||
|
||||
async def write(self, msg: protobuf.MessageType) -> None:
|
||||
"""Write a message to the wire."""
|
||||
...
|
||||
|
||||
async def call(
|
||||
self,
|
||||
msg: protobuf.MessageType,
|
||||
expected_type: type[LoadedMessageType],
|
||||
) -> LoadedMessageType:
|
||||
"""Write a message to the wire, then await and return the response message."""
|
||||
assert expected_type.MESSAGE_WIRE_TYPE is not None
|
||||
|
||||
await self.write(msg)
|
||||
@ -63,10 +85,13 @@ class Context:
|
||||
return await self.read((expected_type.MESSAGE_WIRE_TYPE,), expected_type)
|
||||
|
||||
def release(self) -> None:
|
||||
"""Release resources used by the context, eg. clear context cache."""
|
||||
pass
|
||||
|
||||
@property
|
||||
def cache(self) -> DataCache: ...
|
||||
def cache(self) -> DataCache:
|
||||
"""Access to the backing cache of the context, if the context has any."""
|
||||
...
|
||||
|
||||
|
||||
class WireError(Exception):
|
||||
|
Loading…
Reference in New Issue
Block a user