State Trie
Table of Contents
Introduction
The state trie is the structure responsible for storing .fork_types.Account objects.
Module Contents
Classes
Leaf node in the Merkle Trie |
|
Extension node in the Merkle Trie |
|
Branch node in the Merkle Trie |
|
The Merkle Trie. |
Functions
Encodes a Merkle Trie node into its RLP form. The RLP will then be |
|
Encode a Node for storage in the Merkle Trie. |
|
Create a copy of trie. Since only frozen objects may be stored in tries, |
|
Stores an item in a Merkle Trie. |
|
Gets an item from the Merkle Trie. |
|
Find the longest common prefix of two sequences. |
|
Compresses nibble-list into a standard byte array with a flag. |
|
Converts a Bytes into to a sequence of nibbles (bytes with value < 16). |
|
Prepares the trie for root calculation. Removes values that are empty, |
|
Computes the root of a modified merkle patricia trie (MPT). |
|
Structural composition function. |
Attributes
Module Details
EMPTY_TRIE_ROOT
- EMPTY_TRIE_ROOT
EMPTY_TRIE_ROOT = Root(
hex_to_bytes(
"56e81f171bcc55a6ff8345e692c0f86e5b48e01b996cadc001622fb5e363b421"
)
)
Node
- Node
Node = Union[Account, Bytes, Transaction, Receipt, Uint, U256, None]
K
- K
K = TypeVar("K", bound=Bytes)
V
- V
V = TypeVar(
"V",
Optional[Account],
Optional[Bytes],
Bytes,
Optional[Transaction],
Optional[Receipt],
Uint,
U256,
)
LeafNode
Leaf node in the Merkle Trie
ExtensionNode
Extension node in the Merkle Trie
BranchNode
Branch node in the Merkle Trie
InternalNode
- InternalNode
InternalNode = Union[LeafNode, ExtensionNode, BranchNode]
encode_internal_node
- encode_internal_node(node: Optional[InternalNode]) → ethereum.rlp.RLP
Encodes a Merkle Trie node into its RLP form. The RLP will then be serialized into a Bytes and hashed unless it is less that 32 bytes when serialized.
This function also accepts None, representing the absence of a node, which is encoded to b””.
- Parameters
node (Optional[InternalNode]) – The node to encode.
- Returns
encoded – The node encoded as RLP.
- Return type
rlp.RLP
def encode_internal_node(node: Optional[InternalNode]) -> rlp.RLP:
unencoded: rlp.RLP
if node is None:
unencoded = b""
elif isinstance(node, LeafNode):
unencoded = (
nibble_list_to_compact(node.rest_of_key, True),
node.value,
)
elif isinstance(node, ExtensionNode):
unencoded = (
nibble_list_to_compact(node.key_segment, False),
node.subnode,
)
elif isinstance(node, BranchNode):
unencoded = node.subnodes + [node.value]
else:
raise AssertionError(f"Invalid internal node type {type(node)}!")
encoded = rlp.encode(unencoded)
if len(encoded) < 32:
return unencoded
else:
return keccak256(encoded)
encode_node
- encode_node(node: Node, storage_root: Optional[ethereum.base_types.Bytes] = None) → ethereum.base_types.Bytes
Encode a Node for storage in the Merkle Trie.
Currently mostly an unimplemented stub.
def encode_node(node: Node, storage_root: Optional[Bytes] = None) -> Bytes:
if isinstance(node, Account):
assert storage_root is not None
return encode_account(node, storage_root)
elif isinstance(node, (Transaction, Receipt, U256)):
return rlp.encode(cast(rlp.RLP, node))
elif isinstance(node, Bytes):
return node
else:
raise AssertionError(
f"encoding for {type(node)} is not currently implemented"
)
Trie
The Merkle Trie.
copy_trie
- copy_trie(trie: Trie[K, V]) → Trie[K, V]
Create a copy of trie. Since only frozen objects may be stored in tries, the contents are reused.
- Parameters
trie (Trie) – Trie to copy.
- Returns
new_trie – A copy of the trie.
- Return type
Trie[K, V]
def copy_trie(trie: Trie[K, V]) -> Trie[K, V]:
return Trie(trie.secured, trie.default, copy.copy(trie._data))
trie_set
- trie_set(trie: Trie[K, V], key: K, value: V) → None
Stores an item in a Merkle Trie.
This method deletes the key if value == trie.default, because the Merkle Trie represents the default value by omitting it from the trie.
- Parameters
trie (Trie) – Trie to store in.
key (Bytes) – Key to lookup.
value (V) – Node to insert at key.
def trie_set(trie: Trie[K, V], key: K, value: V) -> None:
if value == trie.default:
if key in trie._data:
del trie._data[key]
else:
trie._data[key] = value
trie_get
- trie_get(trie: Trie[K, V], key: K) → V
Gets an item from the Merkle Trie.
This method returns trie.default if the key is missing.
- Parameters
trie – Trie to lookup in.
key – Key to lookup.
- Returns
node – Node at key in the trie.
- Return type
V
def trie_get(trie: Trie[K, V], key: K) -> V:
return trie._data.get(key, trie.default)
common_prefix_length
- common_prefix_length(a: Sequence, b: Sequence) → int
Find the longest common prefix of two sequences.
def common_prefix_length(a: Sequence, b: Sequence) -> int:
for i in range(len(a)):
if i >= len(b) or a[i] != b[i]:
return i
return len(a)
nibble_list_to_compact
- nibble_list_to_compact(x: ethereum.base_types.Bytes, is_leaf: bool) → ethereum.base_types.Bytes
Compresses nibble-list into a standard byte array with a flag.
A nibble-list is a list of byte values no greater than 15. The flag is encoded in high nibble of the highest byte. The flag nibble can be broken down into two two-bit flags.
Highest nibble:
+---+---+----------+--------+ | _ | _ | is_leaf | parity | +---+---+----------+--------+ 3 2 1 0
The lowest bit of the nibble encodes the parity of the length of the remaining nibbles – 0 when even and 1 when odd. The second lowest bit is used to distinguish leaf and extension nodes. The other two bits are not used.
- Parameters
x – Array of nibbles.
is_leaf – True if this is part of a leaf node, or false if it is an extension node.
- Returns
compressed – Compact byte array.
- Return type
bytearray
def nibble_list_to_compact(x: Bytes, is_leaf: bool) -> Bytes:
compact = bytearray()
if len(x) % 2 == 0: # ie even length
compact.append(16 * (2 * is_leaf))
for i in range(0, len(x), 2):
compact.append(16 * x[i] + x[i + 1])
else:
compact.append(16 * ((2 * is_leaf) + 1) + x[0])
for i in range(1, len(x), 2):
compact.append(16 * x[i] + x[i + 1])
return Bytes(compact)
bytes_to_nibble_list
- bytes_to_nibble_list(bytes_: ethereum.base_types.Bytes) → ethereum.base_types.Bytes
Converts a Bytes into to a sequence of nibbles (bytes with value < 16).
- Parameters
bytes – The Bytes to convert.
- Returns
nibble_list – The Bytes in nibble-list format.
- Return type
Bytes
def bytes_to_nibble_list(bytes_: Bytes) -> Bytes:
nibble_list = bytearray(2 * len(bytes_))
for byte_index, byte in enumerate(bytes_):
nibble_list[byte_index * 2] = (byte & 0xF0) >> 4
nibble_list[byte_index * 2 + 1] = byte & 0x0F
return Bytes(nibble_list)
_prepare_trie
- _prepare_trie(trie: Trie[K, V], get_storage_root: Callable[[ethereum.frontier.fork_types.Address], ethereum.frontier.fork_types.Root] = None) → Mapping[ethereum.base_types.Bytes, ethereum.base_types.Bytes]
Prepares the trie for root calculation. Removes values that are empty, hashes the keys (if secured == True) and encodes all the nodes.
- Parameters
trie – The Trie to prepare.
get_storage_root – Function to get the storage root of an account. Needed to encode Account objects.
- Returns
out – Object with keys mapped to nibble-byte form.
- Return type
Mapping[ethereum.base_types.Bytes, Node]
def _prepare_trie(
trie: Trie[K, V],
get_storage_root: Callable[[Address], Root] = None,
) -> Mapping[Bytes, Bytes]:
mapped: MutableMapping[Bytes, Bytes] = {}
for (preimage, value) in trie._data.items():
if isinstance(value, Account):
assert get_storage_root is not None
address = Address(preimage)
encoded_value = encode_node(value, get_storage_root(address))
else:
encoded_value = encode_node(value)
# Empty values are represented by their absence
ensure(encoded_value != b"", AssertionError)
key: Bytes
if trie.secured:
# "secure" tries hash keys once before construction
key = keccak256(preimage)
else:
key = preimage
mapped[bytes_to_nibble_list(key)] = encoded_value
return mapped
root
- root(trie: Trie[K, V], get_storage_root: Callable[[ethereum.frontier.fork_types.Address], ethereum.frontier.fork_types.Root] = None) → ethereum.frontier.fork_types.Root
Computes the root of a modified merkle patricia trie (MPT).
- Parameters
trie – Trie to get the root of.
get_storage_root – Function to get the storage root of an account. Needed to encode Account objects.
- Returns
root – MPT root of the underlying key-value pairs.
- Return type
.fork_types.Root
def root(
trie: Trie[K, V],
get_storage_root: Callable[[Address], Root] = None,
) -> Root:
obj = _prepare_trie(trie, get_storage_root)
root_node = encode_internal_node(patricialize(obj, Uint(0)))
if len(rlp.encode(root_node)) < 32:
return keccak256(rlp.encode(root_node))
else:
assert isinstance(root_node, Bytes)
return Root(root_node)
patricialize
- patricialize(obj: Mapping[ethereum.base_types.Bytes, ethereum.base_types.Bytes], level: ethereum.base_types.Uint) → Optional[InternalNode]
Structural composition function.
Used to recursively patricialize and merkleize a dictionary. Includes memoization of the tree structure and hashes.
- Parameters
obj – Underlying trie key-value pairs, with keys in nibble-list format.
level – Current trie level.
- Returns
node – Root node of obj.
- Return type
ethereum.base_types.Bytes
def patricialize(
obj: Mapping[Bytes, Bytes], level: Uint
) -> Optional[InternalNode]:
if len(obj) == 0:
return None
arbitrary_key = next(iter(obj))
# if leaf node
if len(obj) == 1:
leaf = LeafNode(arbitrary_key[level:], obj[arbitrary_key])
return leaf
# prepare for extension node check by finding max j such that all keys in
# obj have the same key[i:j]
substring = arbitrary_key[level:]
prefix_length = len(substring)
for key in obj:
prefix_length = min(
prefix_length, common_prefix_length(substring, key[level:])
)
# finished searching, found another key at the current level
if prefix_length == 0:
break
# if extension node
if prefix_length > 0:
prefix = arbitrary_key[level : level + prefix_length]
return ExtensionNode(
prefix,
encode_internal_node(patricialize(obj, level + prefix_length)),
)
branches: List[MutableMapping[Bytes, Bytes]] = []
for _ in range(16):
branches.append({})
value = b""
for key in obj:
if len(key) == level:
# shouldn't ever have an account or receipt in an internal node
if isinstance(obj[key], (Account, Receipt, Uint)):
raise AssertionError
value = obj[key]
else:
branches[key[level]][key] = obj[key]
return BranchNode(
[
encode_internal_node(patricialize(branches[k], level + 1))
for k in range(16)
],
value,
)