Python dis 模块，Instruction() 实例源码

def POP_JUMP_IF_TRUE(ctx: _VSContext, instruction: dis.Instruction):
    """
    Jumps to the specified instruction if True-y is on the top of the stack.
    """

    i = ctx.pop()
    if not i:
        # Falsey, stay where we are.
        return ctx

    # Jump, again.
    ctx.instruction_pointer = get_instruction_index_by_offset(ctx, instruction)

    return ctx


# endregion


# region Stubs
# Instructions that do nothing currently.

def __init__(self, do_context_switching=True):
        self.current_instruction: dis.Instruction = None
        self.current_context: _VSContext = None

        self.do_context_switching: bool = do_context_switching

def __run_natively(self, context: _VSContext, instruction: dis.Instruction):
        """
        Invokes a function natively.
        """
        # Get the number of arguments to pop off of the stack.
        number_of_args = instruction.arg
        args = []
        for x in range(0, number_of_args):
            # Pop each argument off of the stack.
            args.append(context.stack.pop())

        args = reversed(args)

        # Now pop the function, which is underneath all the others.
        fn = context.stack.pop()
        if not callable(fn):
            safe_raise(context, TypeError("'{}' object is not callable".format(fn)))
            return

        # Run the function.
        try:
            result = fn(*args)
        except BaseException as e:
            safe_raise(context, e)
            return

        return result

def LOAD_GLOBAL(ctx: _VSContext, instruction: dis.Instruction):
    """
    Loads a global from `ctx.__globals__`.
    """
    name = ctx.co_names[instruction.arg]
    try:
        item = ctx.get_global(name)
    except KeyError:
        # todo: safe_raise
        return safe_raise(ctx, NameError("name '{}' is not defined".format(name)))

    ctx.push(item)
    return ctx

def LOAD_CONST(ctx: _VSContext, instruction: dis.Instruction):
    """
    Loads a const from `ctx.co_consts`.
    """
    ctx.push(ctx.co_consts[instruction.arg])
    return ctx

def LOAD_FAST(ctx: _VSContext, instruction: dis.Instruction):
    """
    Loads from VARNAMES.
    """
    item = ctx.varnames[instruction.arg]
    if item == NO_RESULT:
        safe_raise(ctx, NameError("name '{}' is not defined".format(ctx.co_varnames[instruction.arg])))
        return ctx
    ctx.push(item)
    return ctx

def POP_TOP(ctx: _VSContext, instruction: dis.Instruction):
    """
    Pops off the top of the stack.
    """
    ctx.pop()
    return ctx

def DUP_TOP(ctx: _VSContext, instruction: dis.Instruction):
    """
    Duplicates the top-most item on the stack.
    """
    item = ctx.pop()
    ctx.push(item)
    ctx.push(item)
    return ctx

def STORE_FAST(ctx: _VSContext, instruction: dis.Instruction):
    """
    Stores data in co_varnames.
    """
    ctx.varnames[instruction.arg] = ctx.pop()
    return ctx

def STORE_NAME(ctx: _VSContext, instruction: dis.Instruction):
    ctx.names[instruction.arg] = ctx.pop()
    return ctx

def RETURN_VALUE(ctx: _VSContext, instruction: dis.Instruction):
    """
    Returns a value.

    This will set the state of the context.
    """
    ctx._result = ctx.pop()
    ctx.state = VSCtxState.FINISHED

    ctx._handling_exception = False

    return ctx

def JUMP_FORWARD(ctx: _VSContext, instruction: dis.Instruction):
    """
    Jumps forward to the specified instruction.
    """
    ctx.instruction_pointer = get_instruction_index_by_offset(ctx, instruction)

    return ctx

def POP_JUMP_IF_FALSE(ctx: _VSContext, instruction: dis.Instruction):
    """
    Jumps to the specified instruction if False-y is on the top of the stack.
    """
    i = ctx.pop()
    if i:
        # Truthy, don't jump.
        return ctx

    # Jump!
    ctx.instruction_pointer = get_instruction_index_by_offset(ctx, instruction)

    return ctx

def POP_BLOCK(ctx: _VSContext, instruction: dis.Instruction):
    return ctx


# endregion

# region Exception handling
# Exception handling.
# These are all part of the Vanstein bootleg exception system.

def SETUP_EXCEPT(ctx: _VSContext, instruction: dis.Instruction):
    """
    Sets a context up for an except.
    """
    # Update the exception pointer with the calculated offset.
    # This is where we will jump to if an error is encountered.
    ctx.exc_next_pointer = get_instruction_index_by_offset(ctx, instruction)

    return ctx

def RAISE_VARARGS(ctx: _VSContext, instruction: dis.Instruction):
    """
    Raises an exception to either the current scope or the outer scope.
    """
    # This is relatively simple.
    # We ignore the argc == 3, and pretend it's argc == 2
    argc = instruction.arg
    if argc == 3:
        # fuck you
        ctx.pop()
        argc = 2

    if argc == 2:
        # FROM exception is Top of stack now.
        fr = ctx.pop()
        # The real exception is top of stack now.
        exc = ctx.pop()
        exc.__cause__ = fr

    elif argc == 1:
        exc = ctx.pop()
    else:
        # Bare raise.
        exc = ctx._exception_state

    # Inject the exception.
    safe_raise(ctx, exc)
    # Raise the exception.

    return exc


# endregion

def MAKE_FUNCTION(ctx: _VSContext, instruction: dis.Instruction):
    """
    Called to create a new function.

    This assumes a name is on TOS, and that a code object is on TOS2.
    """

def get_instructions(code):
        """
        Iterator parsing the bytecode into easy-usable minimal emulation of
        Python 3.4 `dis.Instruction` instances.
        """

        # shortcuts
        HAVE_ARGUMENT = dis.HAVE_ARGUMENT
        EXTENDED_ARG = dis.EXTENDED_ARG

        class Instruction:
            # Minimal emulation of Python 3.4 dis.Instruction
            def __init__(self, opcode, oparg):
                self.opname = dis.opname[opcode]
                self.arg = oparg
                # opcode, argval, argrepr, offset, is_jump_target and
                # starts_line are not used by our code, so we leave them away
                # here.

        code = code.co_code
        extended_arg = 0
        i = 0
        n = len(code)
        while i < n:
            c = code[i]
            i = i + 1
            op = _cOrd(c)
            if op >= HAVE_ARGUMENT:
                oparg = _cOrd(code[i]) + _cOrd(code[i + 1]) * 256 + extended_arg
                extended_arg = 0
                i += 2
                if op == EXTENDED_ARG:
                    extended_arg = oparg*65536
            else:
                oparg = None
            yield Instruction(op, oparg)


#FIXME: Leverage this rather than magic numbers below.

def _step(self):
        """
        Moves one step forward in the event loop.

        This will collect the next task from the deque, and run it.

        This is an internal function and should not be called.
        """
        if not self._running:
            raise RuntimeError("Loop is not running.")

        if not self.running_tasks:
            # TODO: Handle loop wind-down.
            return

        next_task = self.running_tasks.popleft()
        assert isinstance(next_task, _VSContext)
        if next_task.state is VSCtxState.SUSPENDED:
            # It hasn't reached a wake-up call yet, so re-add it to the end of the deque.
            self.running_tasks.append(next_task)
            return

        if next_task.state == VSCtxState.RUNNING:
            # No need to use safe_raise here.
            # This will never raise a VS-handled exception, because it's a native invoke function.
            raise RuntimeError("Current task state is RUNNING - this should never happen!")

        if next_task.state is VSCtxState.PENDING:
            # It's newly created, or otherwise ready. Continue execution.
            # This should automatically pop or push it as appropriate.
            try:
                return self._start_execution(next_task)
            except NotImplementedError as e:
                print("Fatal error in Vanstein:")
                print("Instruction '{}' is not implemented yet.".format(
                    self.bytecode_engine.current_instruction.opname))
                print("Function disassembly:")
                dis.dis(next_task)
                print("Current context: {}".format(self.bytecode_engine.current_context))
                print("Current instruction: {}".format(self.bytecode_engine.current_instruction))
                raise
            except BaseException as e:
                print("Fatal error in Vanstein:")
                traceback.print_exc(file=sys.stdout)
                print("Function disassembly:")
                dis.dis(next_task)
                print("Current context: {}".format(self.bytecode_engine.current_context))
                print("Current instruction: {}".format(self.bytecode_engine.current_instruction))
                raise

        if next_task.state is VSCtxState.FINISHED:
            # Hopefully, we never have to see this.
            warnings.warn("Reached FINISHED task in event loop...")
            return next_task

def get_instruction_index_by_offset(ctx: _VSContext, instruction: dis.Instruction) -> int:
    """
    Returns the index of an instruction (i.e ctx.instructions[I]) when given an offset to jump to.

    This is useful for when implementing an operator such as JUMP_FORWARD or SETUP_*.

    :param ctx: The context in which this is currently executing.
    :param instruction: The instruction to use.
    :return: The instruction index.
    """

    if PY36:
        # In CPython 3.6 and above, bytecode instructions are 2 bytes wide, all the time.
        # That means the offset can be divided by two and added to the instruction pointer.
        return ctx.instruction_pointer + (instruction.arg / 2)
    else:
        # This is a bit trickier.

        # However, we can use some known constants.
        # Opcodes with a code more than 90 have arguments, which means that they are length 3 (opcode + argv[0, 1])
        # We can loop over the instructions, check their opcode, and then check if the opcode is more than 90.
        # If it is, add 3 to our local raw pointer, because it has an argument, and 1 to the offset.
        # If it isn't, then only add 1 to our local raw pointer, but 1 to the offset again.
        # Once we've looped over these, we can then check if our local pointer is equal to the arg (NOT ARGVAL) of
        # the current instruction.
        # When it is, that means the instruction we're looking for is the current one, and we set the exception
        # pointer to our offset.

        # We have to special case arg = 1 on 3.5 and below.
        if instruction.arg == 1:
            return ctx.instruction_pointer + 1

        local_pointer = 0
        offset = 0
        for ins in ctx.instructions[ctx.instruction_pointer:]:
            if ins.opcode >= dis.HAVE_ARGUMENT:
                local_pointer += 3
            else:
                local_pointer += 1

            offset += 1

            if local_pointer == instruction.arg:
                # We've reached the right instruction.
                break
        else:
            raise SystemError("Could not find instruction at '{}'".format(instruction.argval))

        # The new instruction pointer is the current one + the offset we just calculated.
        return ctx.instruction_pointer + offset