import sys

import types

from collections import defaultdict

from dataclasses import dataclass

from typing import (

Any,

Dict,

Generator,

Iterable,

Iterator,

List,

Optional,

Set,

SupportsIndex,

Tuple,

TypeVar,

Union,

overload,

)

# alias to keep the 'bytecode' variable free

import bytecode as _bytecode

from bytecode.concrete import ConcreteInstr

from bytecode.flags import CompilerFlags

from bytecode.instr import UNSET, Instr, Label, SetLineno, TryBegin, TryEnd

T = TypeVar("T", bound="BasicBlock")

U = TypeVar("U", bound="ControlFlowGraph")

class BasicBlock(_bytecode._InstrList[Union[Instr, SetLineno, TryBegin, TryEnd]]):

def __init__(

self,

instructions: Optional[

Iterable[Union[Instr, SetLineno, TryBegin, TryEnd]]

] = None,

) -> None:

# a BasicBlock object, or None

self.next_block: Optional["BasicBlock"] = None

if instructions:

super().__init__(instructions)

def __iter__(self) -> Iterator[Union[Instr, SetLineno, TryBegin, TryEnd]]:

index = 0

while index < len(self):

instr = self[index]

index += 1

if not isinstance(instr, (SetLineno, Instr, TryBegin, TryEnd)):

raise ValueError(

"BasicBlock must only contain SetLineno and Instr objects, "

"but %s was found" % instr.__class__.__name__

)

if isinstance(instr, Instr) and instr.has_jump():

if index < len(self) and any(

isinstance(self[i], Instr) for i in range(index, len(self))

):

raise ValueError(

"Only the last instruction of a basic " "block can be a jump"

)

if not isinstance(instr.arg, BasicBlock):

raise ValueError(

"Jump target must a BasicBlock, got %s",

type(instr.arg).__name__,

)

if isinstance(instr, TryBegin):

if not isinstance(instr.target, BasicBlock):

raise ValueError(

"TryBegin target must a BasicBlock, got %s",

type(instr.target).__name__,

)

yield instr

@overload

def __getitem__(

self, index: SupportsIndex

) -> Union[Instr, SetLineno, TryBegin, TryEnd]:

...

@overload

def __getitem__(self: T, index: slice) -> T:

...

def __getitem__(self, index):

value = super().__getitem__(index)

if isinstance(index, slice):

value = type(self)(value)

value.next_block = self.next_block

return value

def get_last_non_artificial_instruction(self) -> Optional[Instr]:

for instr in reversed(self):

if isinstance(instr, Instr):

return instr

return None

def copy(self: T) -> T:

new = type(self)(super().copy())

new.next_block = self.next_block

return new

def legalize(self, first_lineno: int) -> int:

"""Check that all the element of the list are valid and remove SetLineno."""

lineno_pos = []

set_lineno = None

current_lineno = first_lineno

for pos, instr in enumerate(self):

if isinstance(instr, SetLineno):

set_lineno = current_lineno = instr.lineno

lineno_pos.append(pos)

continue

if isinstance(instr, (TryBegin, TryEnd)):

continue

if set_lineno is not None:

instr.lineno = set_lineno

elif instr.lineno is UNSET:

instr.lineno = current_lineno

elif instr.lineno is not None:

current_lineno = instr.lineno

for i in reversed(lineno_pos):

del self[i]

return current_lineno

def get_jump(self) -> Optional["BasicBlock"]:

if not self:

return None

last_instr = self.get_last_non_artificial_instruction()

if last_instr is None or not last_instr.has_jump():

return None

target_block = last_instr.arg

assert isinstance(target_block, BasicBlock)

return target_block

def get_trailing_try_end(self, index: int):

while index + 1 < len(self):

if isinstance(b := self[index + 1], TryEnd):

return b

index += 1

return None

def _update_size(pre_delta, post_delta, size, maxsize, minsize):

size += pre_delta

if size < 0:

msg = "Failed to compute stacksize, got negative size"

raise RuntimeError(msg)

size += post_delta

maxsize = max(maxsize, size)

minsize = min(minsize, size)

return size, maxsize, minsize

# We can never have nested TryBegin, so we can simply update the min stack size

# when we encounter one and use the number we have when we encounter the TryEnd

@dataclass

class _StackSizeComputationStorage:

"""Common storage shared by the computers involved in computing CFG stack usage."""

#: Should we check that all stack operation are "safe" i.e. occurs while there

#: is a sufficient number of items on the stack.

check_pre_and_post: bool

#: Id the blocks for which an analysis is under progress to avoid getting stuck

#: in recursions.

seen_blocks: Set[int]

#: Sizes and exception handling status with which the analysis of the block

#: has been performed. Used to avoid running multiple times equivalent analysis.

blocks_startsizes: Dict[int, Set[Tuple[int, Optional[bool]]]]

#: Track the encountered TryBegin pseudo-instruction to update their target

#: depth at the end of the calculation.

try_begins: List[TryBegin]

#: Stacksize that should be used for exception blocks. This is the smallest size

#: with which this block was reached which is the only size that can be safely

#: restored.

exception_block_startsize: Dict[int, int]

#: Largest stack size used in an exception block. We record the size corresponding

#: to the smallest start size for the block since the interpreter enforces that

#: we start with this size.

exception_block_maxsize: Dict[int, int]

class _StackSizeComputer:

"""Helper computing the stack usage for a single block."""

#: Common storage shared by all helpers involved in the stack size computation

common: _StackSizeComputationStorage

#: Block this helper is running the computation for.

block: BasicBlock

#: Current stack usage.

size: int

#: Maximal stack usage.

maxsize: int

#: Minimal stack usage. This value is only relevant in between a TryBegin/TryEnd

#: pair and determine the startsize for the exception handling block associated

#: with the try begin.

minsize: int

#: Flag indicating if the block analyzed is an exception handler (i.e. a target

#: of a TryBegin).

exception_handler: Optional[bool]

#: TryBegin that was encountered before jumping to this block and for which

#: no try end was met yet.

pending_try_begin: Optional[TryBegin]

def __init__(

self,

common: _StackSizeComputationStorage,

block: BasicBlock,

size: int,

maxsize: int,

minsize: int,

exception_handler: Optional[bool],

pending_try_begin: Optional[TryBegin],

) -> None:

self.common = common

self.block = block

self.size = size

self.maxsize = maxsize

self.minsize = minsize

self.exception_handler = exception_handler

self.pending_try_begin = pending_try_begin

self._current_try_begin = pending_try_begin

def run(self) -> Generator[Union["_StackSizeComputer", int], int, None]:

"""Iterate over the block instructions to compute stack usage."""

# Blocks are not hashable but in this particular context we know we won't be

# modifying blocks in place so we can safely use their id as hash rather than

# making them generally hashable which would be weird since they are list

# subclasses

block_id = id(self.block)

# If the block is currently being visited (seen = True) or

# it was visited previously with parameters that makes the computation

# irrelevant return the maxsize.

fingerprint = (self.size, self.exception_handler)

if id(self.block) in self.common.seen_blocks or (

not self._is_stacksize_computation_relevant(block_id, fingerprint)

):

yield self.maxsize

# Prevent recursive visit of block if two blocks are nested (jump from one

# to the other).

self.common.seen_blocks.add(block_id)

# Track which size has been used to run an analysis to avoid re-running multiple

# times the same calculation.

self.common.blocks_startsizes[block_id].add(fingerprint)

# If this block is an exception handler reached through the exception table

# we will push some extra objects on the stack before processing start.

if self.exception_handler is not None:

self._update_size(0, 1 + self.exception_handler)

# True is used to indicated that push_lasti is True, leading to pushing

# an extra object on the stack.

for i, instr in enumerate(self.block):

# Ignore SetLineno

if isinstance(instr, (SetLineno)):

continue

# When we encounter a TryBegin, we:

# - store it as the current TryBegin (since TryBegin cannot be nested)

# - record its existence to remember to update its stack size when

# the computation ends

# - update the minsize to the current size value since we need to

# know the minimal stack usage between the TryBegin/TryEnd pair to

# set the startsize of the exception handling block

#

# This approach does not require any special handling for with statements.

if isinstance(instr, TryBegin):

assert self._current_try_begin is None

self.common.try_begins.append(instr)

self._current_try_begin = instr

self.minsize = self.size

continue

elif isinstance(instr, TryEnd):

# When we encounter a TryEnd we can start the computation for the

# exception block using the minimum stack size encountered since

# the TryBegin matching this TryEnd.

# TryBegin cannot be nested so a TryEnd should always match the

# current try begin. However inside the CFG some blocks may

# start with a TryEnd relevant only when reaching this block

# through a particular jump. So we are lenient here.

if instr.entry is not self._current_try_begin:

continue

# Compute the stack usage of the exception handler

assert isinstance(instr.entry.target, BasicBlock)

yield from self._compute_exception_handler_stack_usage(

instr.entry.target,

instr.entry.push_lasti,

)

self._current_try_begin = None

continue

# For instructions with a jump first compute the stacksize required when the

# jump is taken.

if instr.has_jump():

effect = (

instr.pre_and_post_stack_effect(jump=True)

if self.common.check_pre_and_post

else (instr.stack_effect(jump=True), 0)

)

taken_size, maxsize, minsize = _update_size(

*effect, self.size, self.maxsize, self.minsize

)

# Yield the parameters required to compute the stacksize required

# by the block to which the jump points to and resume when we now

# the maxsize.

assert isinstance(instr.arg, BasicBlock)

maxsize = yield _StackSizeComputer(

self.common,

instr.arg,

taken_size,

maxsize,

minsize,

None,

# Do not propagate the TryBegin if a final instruction is followed

# by a TryEnd.

None

if instr.is_final() and self.block.get_trailing_try_end(i)

else self._current_try_begin,

)

# Update the maximum used size by the usage implied by the following

# the jump

self.maxsize = max(self.maxsize, maxsize)

# For unconditional jumps abort early since the other instruction will

# never be seen.

if instr.is_uncond_jump():

# Check for TryEnd after the final instruction which is possible

# TryEnd being only pseudo instructions

if te := self.block.get_trailing_try_end(i):

# TryBegin cannot be nested

assert te.entry is self._current_try_begin

assert isinstance(te.entry.target, BasicBlock)

yield from self._compute_exception_handler_stack_usage(

te.entry.target,

te.entry.push_lasti,

)

self.common.seen_blocks.remove(id(self.block))

yield self.maxsize

# jump=False: non-taken path of jumps, or any non-jump

effect = (

instr.pre_and_post_stack_effect(jump=False)

if self.common.check_pre_and_post

else (instr.stack_effect(jump=False), 0)

)

self._update_size(*effect)

# Instruction is final (return, raise, ...) so any following instruction

# in the block is dead code.

if instr.is_final():

# Check for TryEnd after the final instruction which is possible

# TryEnd being only pseudo instructions.

if te := self.block.get_trailing_try_end(i):

assert isinstance(te.entry.target, BasicBlock)

yield from self._compute_exception_handler_stack_usage(

te.entry.target,

te.entry.push_lasti,

)

self.common.seen_blocks.remove(id(self.block))

yield self.maxsize

if self.block.next_block:

self.maxsize = yield _StackSizeComputer(

self.common,

self.block.next_block,

self.size,

self.maxsize,

self.minsize,

None,

self._current_try_begin,

)

self.common.seen_blocks.remove(id(self.block))

yield self.maxsize

# --- Private API

_current_try_begin: Optional[TryBegin]

def _update_size(self, pre_delta: int, post_delta: int) -> None:

size, maxsize, minsize = _update_size(

pre_delta, post_delta, self.size, self.maxsize, self.minsize

)

self.size = size

self.minsize = minsize

self.maxsize = maxsize

def _compute_exception_handler_stack_usage(

self, block: BasicBlock, push_lasti: bool

) -> Generator[Union["_StackSizeComputer", int], int, None]:

b_id = id(block)

if self.minsize < self.common.exception_block_startsize[b_id]:

block_size = yield _StackSizeComputer(

self.common,

block,

self.minsize,

self.maxsize,

self.minsize,

push_lasti,

None,

)

# The entry cannot be smaller than abs(stc.minimal_entry_size) as otherwise

# we an underflow would have occured.

self.common.exception_block_startsize[b_id] = self.minsize

self.common.exception_block_maxsize[b_id] = block_size

def _is_stacksize_computation_relevant(

self, block_id: int, fingerprint: Tuple[int, Optional[bool]]

) -> bool:

if sys.version_info >= (3, 11):

# The computation is relevant if the block was not visited previously

# with the same starting size and exception handler status than the

# one in use

return fingerprint not in self.common.blocks_startsizes[block_id]

else:

# The computation is relevant if the block was only visited with smaller

# starting sizes than the one in use

if sizes := self.common.blocks_startsizes[block_id]:

return fingerprint[0] > max(f[0] for f in sizes)

else:

return True

class ControlFlowGraph(_bytecode.BaseBytecode):

def __init__(self) -> None:

super().__init__()

self._blocks: List[BasicBlock] = []

self._block_index: Dict[int, int] = {}

self.argnames: List[str] = []

self.add_block()

def legalize(self) -> None:

"""Legalize all blocks."""

current_lineno = self.first_lineno

for block in self._blocks:

current_lineno = block.legalize(current_lineno)

def get_block_index(self, block: BasicBlock) -> int:

try:

return self._block_index[id(block)]

except KeyError:

raise ValueError("the block is not part of this bytecode")

def _add_block(self, block: BasicBlock) -> None:

block_index = len(self._blocks)

self._blocks.append(block)

self._block_index[id(block)] = block_index

def add_block(

self, instructions: Optional[Iterable[Union[Instr, SetLineno]]] = None

) -> BasicBlock:

block = BasicBlock(instructions)

self._add_block(block)

return block

def compute_stacksize(

self,

*,

check_pre_and_post: bool = True,

compute_exception_stack_depths: bool = True,

) -> int:

"""Compute the stack size by iterating through the blocks

The implementation make use of a generator function to avoid issue with

deeply nested recursions.

"""

# In the absence of any block return 0

if not self:

return 0

# Create the common storage for the calculation

common = _StackSizeComputationStorage(

check_pre_and_post,

seen_blocks=set(),

blocks_startsizes={id(b): set() for b in self},

exception_block_startsize=dict.fromkeys([id(b) for b in self], 32768),

exception_block_maxsize=dict.fromkeys([id(b) for b in self], -32768),

try_begins=[],

)

# Starting with Python 3.10, generator and coroutines start with one object

# on the stack (None, anything is an error).

initial_stack_size = 0

if sys.version_info >= (3, 10) and self.flags & (

CompilerFlags.GENERATOR

| CompilerFlags.COROUTINE

| CompilerFlags.ASYNC_GENERATOR

):

initial_stack_size = 1

# Create a generator/coroutine responsible of dealing with the first block

coro = _StackSizeComputer(

common, self[0], initial_stack_size, 0, 0, None, None

).run()

# Create a list of generator that have not yet been exhausted

coroutines: List[Generator[Union[_StackSizeComputer, int], int, None]] = []

push_coroutine = coroutines.append

pop_coroutine = coroutines.pop

args = None

try:

while True:

# Mypy does not seem to honor the fact that one must send None

# to a brand new generator irrespective of its send type.

args = coro.send(None) # type: ignore

# Consume the stored generators as long as they return a simple

# integer that is to be used to resume the last stored generator.

while isinstance(args, int):

coro = pop_coroutine()

args = coro.send(args)

# Otherwise we enter a new block and we store the generator under

# use and create a new one to process the new block

push_coroutine(coro)

coro = args.run()

except IndexError:

# The exception occurs when all the generators have been exhausted

# in which case the last yielded value is the stacksize.

assert args is not None and isinstance(args, int)

# Exception handling block size is reported separately since we need

# to report only the stack usage for the smallest start size for the

# block

args = max(args, *common.exception_block_maxsize.values())

# Check if there is dead code that may contain TryBegin/TryEnd pairs.

# For any such pair we set a huge size (the exception table format does not

# mandate a maximum value). We do so so that if the pair is fused with

# another it does not alter the computed size.

for block in self:

if not common.blocks_startsizes[id(block)]:

for i in block:

if isinstance(i, TryBegin) and i.stack_depth is UNSET:

i.stack_depth = 32768

# If requested update the TryBegin stack size

if compute_exception_stack_depths:

for tb in common.try_begins:

size = common.exception_block_startsize[id(tb.target)]

assert size >= 0

tb.stack_depth = size

return args

def __repr__(self) -> str:

return "<ControlFlowGraph block#=%s>" % len(self._blocks)

# Helper to obtain a flat list of instr, which does not refer to block at

# anymore. Used for comparison of different CFG.

def _get_instructions(

self,

) -> List:

instructions: List = []

try_begins: Dict[TryBegin, int] = {}

for block in self:

for index, instr in enumerate(block):

if isinstance(instr, TryBegin):

assert isinstance(instr.target, BasicBlock)

try_begins.setdefault(instr, len(try_begins))

instructions.append(

(

"TryBegin",

try_begins[instr],

self.get_block_index(instr.target),

instr.push_lasti,

)

)

elif isinstance(instr, TryEnd):

instructions.append(("TryEnd", try_begins[instr.entry]))

elif isinstance(instr, Instr) and (

instr.has_jump() or instr.is_final()

):

if instr.has_jump():

target_block = instr.arg

assert isinstance(target_block, BasicBlock)

# We use a concrete instr here to be able to use an integer as

# argument rather than a Label. This is fine for comparison

# purposes which is our sole goal here.

c_instr = ConcreteInstr(

instr.name,

self.get_block_index(target_block),

location=instr.location,

)

instructions.append(c_instr)

else:

instructions.append(instr)

if te := block.get_trailing_try_end(index):

instructions.append(("TryEnd", try_begins[te.entry]))

break

else:

instructions.append(instr)

return instructions

def __eq__(self, other: Any) -> bool:

if type(self) != type(other):

return False

if self.argnames != other.argnames:

return False

instrs1 = self._get_instructions()

instrs2 = other._get_instructions()

if instrs1 != instrs2:

return False

# FIXME: compare block.next_block

return super().__eq__(other)

def __len__(self) -> int:

return len(self._blocks)

def __iter__(self) -> Iterator[BasicBlock]:

return iter(self._blocks)

@overload

def __getitem__(self, index: Union[int, BasicBlock]) -> BasicBlock:

...

@overload

def __getitem__(self: U, index: slice) -> U:

...

def __getitem__(self, index):

if isinstance(index, BasicBlock):

index = self.get_block_index(index)

return self._blocks[index]

def __delitem__(self, index: Union[int, BasicBlock]) -> None:

if isinstance(index, BasicBlock):

index = self.get_block_index(index)

block = self._blocks[index]

del self._blocks[index]

del self._block_index[id(block)]

for index in range(index, len(self)):

block = self._blocks[index]

self._block_index[id(block)] -= 1

def split_block(self, block: BasicBlock, index: int) -> BasicBlock:

if not isinstance(block, BasicBlock):

raise TypeError("expected block")

block_index = self.get_block_index(block)

if index < 0:

raise ValueError("index must be positive")

block = self._blocks[block_index]

if index == 0:

return block

if index > len(block):

raise ValueError("index out of the block")

instructions = block[index:]

if not instructions:

if block_index + 1 < len(self):

return self[block_index + 1]

del block[index:]

block2 = BasicBlock(instructions)

block.next_block = block2

for block in self[block_index + 1 :]:

self._block_index[id(block)] += 1

self._blocks.insert(block_index + 1, block2)

self._block_index[id(block2)] = block_index + 1

return block2

def get_dead_blocks(self) -> List[BasicBlock]:

if not self:

return []

seen_block_ids = set()

stack = [self[0]]

while stack:

block = stack.pop()

if id(block) in seen_block_ids:

continue

seen_block_ids.add(id(block))

for i in block:

if isinstance(i, Instr) and isinstance(i.arg, BasicBlock):

stack.append(i.arg)

elif isinstance(i, TryBegin):

assert isinstance(i.target, BasicBlock)

stack.append(i.target)

return [b for b in self if id(b) not in seen_block_ids]

@staticmethod

def from_bytecode(bytecode: _bytecode.Bytecode) -> "ControlFlowGraph":

# label => instruction index

label_to_block_index = {}

jumps = []

try_end_locations = {}

for index, instr in enumerate(bytecode):

if isinstance(instr, Label):

label_to_block_index[instr] = index

elif isinstance(instr, Instr) and isinstance(instr.arg, Label):

jumps.append((index, instr.arg))

elif isinstance(instr, TryBegin):

assert isinstance(instr.target, Label)

jumps.append((index, instr.target))

elif isinstance(instr, TryEnd):

try_end_locations[instr.entry] = index

# Figure out on which index block targeted by a label start

block_starts = {}

for target_index, target_label in jumps:

target_index = label_to_block_index[target_label]

block_starts[target_index] = target_label

bytecode_blocks = ControlFlowGraph()

bytecode_blocks._copy_attr_from(bytecode)

bytecode_blocks.argnames = list(bytecode.argnames)

# copy instructions, convert labels to block labels

block = bytecode_blocks[0]

labels = {}

jumping_instrs: List[Instr] = []

# Map input TryBegin to CFG TryBegins (split across blocks may yield multiple

# TryBegin from a single in the bytecode).

try_begins: Dict[TryBegin, list[TryBegin]] = {}

# Storage for TryEnds that need to be inserted at the beginning of a block.

# We use a list because the same block can be reached through several paths

# with different active TryBegins

add_try_end: Dict[Label, List[TryEnd]] = defaultdict(list)

# Track the currently active try begin

active_try_begin: Optional[TryBegin] = None

try_begin_inserted_in_block = False

last_instr: Optional[Instr] = None

for index, instr in enumerate(bytecode):

# Reference to the current block if we create a new one in the following.

old_block: BasicBlock | None = None

# First we determine if we need to create a new block:

# - by checking the current instruction index

if index in block_starts:

old_label = block_starts[index]

# Create a new block if the last created one is not empty

# (of real instructions)

if index != 0 and (li := block.get_last_non_artificial_instruction()):

old_block = block

new_block = bytecode_blocks.add_block()

# If the last non artificial instruction is not final connect

# this block to the next.

if not li.is_final():

block.next_block = new_block

block = new_block

if old_label is not None:

labels[old_label] = block

# - by inspecting the last instr

elif block.get_last_non_artificial_instruction() and last_instr is not None:

# The last instruction is final but we did not create a block

# -> sounds like a block of dead code but we preserve it

if last_instr.is_final():

old_block = block

block = bytecode_blocks.add_block()

# We are dealing with a conditional jump

elif last_instr.has_jump():

assert isinstance(last_instr.arg, Label)

old_block = block

new_block = bytecode_blocks.add_block()

block.next_block = new_block

block = new_block

# If we created a new block, we check:

# - if the current instruction is a TryEnd and if the last instruction

# is final in which case we insert the TryEnd in the old block.

# - if we have a currently active TryBegin for which we may need to

# create a TryEnd in the previous block and a new TryBegin in the

# new one because the blocks are not connected.

if old_block is not None:

temp = try_begin_inserted_in_block

try_begin_inserted_in_block = False

if old_block is not None and last_instr is not None:

# The last instruction is final, if the current instruction is a

# TryEnd insert it in the same block and move to the next instruction

if last_instr.is_final() and isinstance(instr, TryEnd):

assert active_try_begin

nte = instr.copy()

nte.entry = try_begins[active_try_begin][-1]

old_block.append(nte)

active_try_begin = None

continue

# If we have an active TryBegin and last_instr is:

elif active_try_begin is not None:

# - a jump whose target is beyond the TryEnd of the active

# TryBegin: we remember TryEnd should be prepended to the

# target block.

if (

last_instr.has_jump()

and active_try_begin in try_end_locations

and (

# last_instr is a jump so arg is a Label

label_to_block_index[last_instr.arg] # type: ignore

>= try_end_locations[active_try_begin]

)

):

assert isinstance(last_instr.arg, Label)

add_try_end[last_instr.arg].append(

TryEnd(try_begins[active_try_begin][-1])

)

# - final and the try begin originate from the current block:

# we insert a TryEnd in the old block and a new TryBegin in

# the new one since the blocks are disconnected.

if last_instr.is_final() and temp:

old_block.append(TryEnd(try_begins[active_try_begin][-1]))

new_tb = TryBegin(

active_try_begin.target, active_try_begin.push_lasti

)

block.append(new_tb)

# Add this new TryBegin to the map to properly update

# the target.

try_begins[active_try_begin].append(new_tb)

try_begin_inserted_in_block = True

last_instr = None

if isinstance(instr, Label):

continue

# don't copy SetLineno objects

if isinstance(instr, (Instr, TryBegin, TryEnd)):

new = instr.copy()

if isinstance(instr, TryBegin):

assert active_try_begin is None

active_try_begin = instr

try_begin_inserted_in_block = True

assert isinstance(new, TryBegin)

try_begins[instr] = [new]

elif isinstance(instr, TryEnd):

assert isinstance(new, TryEnd)

new.entry = try_begins[instr.entry][-1]

active_try_begin = None

try_begin_inserted_in_block = False

else:

last_instr = instr

if isinstance(instr.arg, Label):

assert isinstance(new, Instr)

jumping_instrs.append(new)

instr = new

block.append(instr)

# Insert the necessary TryEnds at the beginning of block that were marked

# (if we did not already insert an equivalent TryEnd earlier).

for lab, tes in add_try_end.items():

block = labels[lab]

existing_te_entries = set()

index = 0

# We use a while loop since the block cannot yet be iterated on since

# jumps still use labels instead of blocks

while index < len(block):

i = block[index]

index += 1

if isinstance(i, TryEnd):

existing_te_entries.add(i.entry)

else:

break

for te in tes:

if te.entry not in existing_te_entries:

labels[lab].insert(0, te)

existing_te_entries.add(te.entry)

# Replace labels by block in jumping instructions

for instr in jumping_instrs:

label = instr.arg

assert isinstance(label, Label)

instr.arg = labels[label]

# Replace labels by block in TryBegin

for b_tb, c_tbs in try_begins.items():

label = b_tb.target

assert isinstance(label, Label)

for c_tb in c_tbs:

c_tb.target = labels[label]

return bytecode_blocks

def to_bytecode(self) -> _bytecode.Bytecode:

"""Convert to Bytecode."""

used_blocks = set()

for block in self:

target_block = block.get_jump()

if target_block is not None:

used_blocks.add(id(target_block))

for tb in (i for i in block if isinstance(i, TryBegin)):

used_blocks.add(id(tb.target))

labels = {}

jumps = []

try_begins = {}

seen_try_end: Set[TryBegin] = set()

instructions: List[Union[Instr, Label, TryBegin, TryEnd, SetLineno]] = []

# Track the last seen TryBegin and TryEnd to be able to fuse adjacent

# TryEnd/TryBegin pair which share the same target.

# In each case, we store the value found in the CFG and the value

# inserted in the bytecode.

last_try_begin: tuple[TryBegin, TryBegin] | None = None

last_try_end: tuple[TryEnd, TryEnd] | None = None

for block in self:

if id(block) in used_blocks:

new_label = Label()

labels[id(block)] = new_label

instructions.append(new_label)

for instr in block:

# don't copy SetLineno objects

if isinstance(instr, (Instr, TryBegin, TryEnd)):

new = instr.copy()

if isinstance(instr, TryBegin):

# If due to jumps and split TryBegin, we encounter a TryBegin

# while we still have a TryBegin ensure they can be fused.

if last_try_begin is not None:

cfg_tb, byt_tb = last_try_begin

assert instr.target is cfg_tb.target

assert instr.push_lasti == cfg_tb.push_lasti

byt_tb.stack_depth = min(

byt_tb.stack_depth, instr.stack_depth

)

# If the TryBegin share the target and push_lasti of the

# entry of an adjacent TryEnd, omit the new TryBegin that

# was inserted to allow analysis of the CFG and remove

# the already inserted TryEnd.

if last_try_end is not None:

cfg_te, byt_te = last_try_end

entry = cfg_te.entry

if (

entry.target is instr.target

and entry.push_lasti == instr.push_lasti

):

# If we did not yet compute the required stack depth

# keep the value as UNSET

if entry.stack_depth is UNSET:

assert instr.stack_depth is UNSET

byt_te.entry.stack_depth = UNSET

else:

byt_te.entry.stack_depth = min(

entry.stack_depth, instr.stack_depth