#include "Python.h"

#include "opcode.h"

#include "pycore_code.h" // _PyCodeConstructor

#include "pycore_function.h" // _PyFunction_ClearCodeByVersion()

#include "pycore_hashtable.h" // _Py_hashtable_t

#include "pycore_index_pool.h" // _PyIndexPool_Fini()

#include "pycore_initconfig.h" // _PyStatus_OK()

#include "pycore_interp.h" // PyInterpreterState.co_extra_freefuncs

#include "pycore_interpframe.h" // FRAME_SPECIALS_SIZE

#include "pycore_opcode_metadata.h" // _PyOpcode_Caches

#include "pycore_opcode_utils.h" // RESUME_AT_FUNC_START

#include "pycore_optimizer.h" // _Py_ExecutorDetach

#include "pycore_pymem.h" // _PyMem_FreeDelayed()

#include "pycore_pystate.h" // _PyInterpreterState_GET()

#include "pycore_setobject.h" // _PySet_NextEntry()

#include "pycore_tuple.h" // _PyTuple_ITEMS()

#include "pycore_unicodeobject.h" // _PyUnicode_InternImmortal()

#include "pycore_uniqueid.h" // _PyObject_AssignUniqueId()

#include "pycore_weakref.h" // FT_CLEAR_WEAKREFS()

#include "clinic/codeobject.c.h"

#include <stdbool.h>

#define INITIAL_SPECIALIZED_CODE_SIZE 16

static const char *

code_event_name(PyCodeEvent event) {

switch (event) {

#define CASE(op) \

case PY_CODE_EVENT_##op: \

return "PY_CODE_EVENT_" #op;

PY_FOREACH_CODE_EVENT(CASE)

#undef CASE

}

Py_UNREACHABLE();

}

static void

notify_code_watchers(PyCodeEvent event, PyCodeObject *co)

{

assert(Py_REFCNT(co) > 0);

PyInterpreterState *interp = _PyInterpreterState_GET();

assert(interp->_initialized);

uint8_t bits = interp->active_code_watchers;

int i = 0;

while (bits) {

assert(i < CODE_MAX_WATCHERS);

if (bits & 1) {

PyCode_WatchCallback cb = interp->code_watchers[i];

// callback must be non-null if the watcher bit is set

assert(cb != NULL);

if (cb(event, co) < 0) {

PyErr_FormatUnraisable(

"Exception ignored in %s watcher callback for %R",

code_event_name(event), co);

}

}

i++;

bits >>= 1;

}

}

int

PyCode_AddWatcher(PyCode_WatchCallback callback)

{

PyInterpreterState *interp = _PyInterpreterState_GET();

assert(interp->_initialized);

for (int i = 0; i < CODE_MAX_WATCHERS; i++) {

if (!interp->code_watchers[i]) {

interp->code_watchers[i] = callback;

interp->active_code_watchers |= (1 << i);

return i;

}

}

PyErr_SetString(PyExc_RuntimeError, "no more code watcher IDs available");

return -1;

}

static inline int

validate_watcher_id(PyInterpreterState *interp, int watcher_id)

{

if (watcher_id < 0 || watcher_id >= CODE_MAX_WATCHERS) {

PyErr_Format(PyExc_ValueError, "Invalid code watcher ID %d", watcher_id);

return -1;

}

if (!interp->code_watchers[watcher_id]) {

PyErr_Format(PyExc_ValueError, "No code watcher set for ID %d", watcher_id);

return -1;

}

return 0;

}

int

PyCode_ClearWatcher(int watcher_id)

{

PyInterpreterState *interp = _PyInterpreterState_GET();

assert(interp->_initialized);

if (validate_watcher_id(interp, watcher_id) < 0) {

return -1;

}

interp->code_watchers[watcher_id] = NULL;

interp->active_code_watchers &= ~(1 << watcher_id);

return 0;

}

/******************

* generic helpers

******************/

#define _PyCodeObject_CAST(op) (assert(PyCode_Check(op)), (PyCodeObject *)(op))

static int

should_intern_string(PyObject *o)

{

#ifdef Py_GIL_DISABLED

// The free-threaded build interns (and immortalizes) all string constants

return 1;

#else

// compute if s matches [a-zA-Z0-9_]

const unsigned char *s, *e;

if (!PyUnicode_IS_ASCII(o))

return 0;

s = PyUnicode_1BYTE_DATA(o);

e = s + PyUnicode_GET_LENGTH(o);

for (; s != e; s++) {

if (!Py_ISALNUM(*s) && *s != '_')

return 0;

}

return 1;

#endif

}

#ifdef Py_GIL_DISABLED

static PyObject *intern_one_constant(PyObject *op);

// gh-130851: In the free threading build, we intern and immortalize most

// constants, except code objects. However, users can generate code objects

// with arbitrary co_consts. We don't want to immortalize or intern unexpected

// constants or tuples/sets containing unexpected constants.

static int

should_immortalize_constant(PyObject *v)

{

// Only immortalize containers if we've already immortalized all their

// elements.

if (PyTuple_CheckExact(v)) {

for (Py_ssize_t i = PyTuple_GET_SIZE(v); --i >= 0; ) {

if (!_Py_IsImmortal(PyTuple_GET_ITEM(v, i))) {

return 0;

}

}

return 1;

}

else if (PyFrozenSet_CheckExact(v)) {

PyObject *item;

Py_hash_t hash;

Py_ssize_t pos = 0;

while (_PySet_NextEntry(v, &pos, &item, &hash)) {

if (!_Py_IsImmortal(item)) {

return 0;

}

}

return 1;

}

else if (PySlice_Check(v)) {

PySliceObject *slice = (PySliceObject *)v;

return (_Py_IsImmortal(slice->start) &&

_Py_IsImmortal(slice->stop) &&

_Py_IsImmortal(slice->step));

}

return (PyLong_CheckExact(v) || PyFloat_CheckExact(v) ||

PyComplex_Check(v) || PyBytes_CheckExact(v));

}

#endif

static int

intern_strings(PyObject *tuple)

{

PyInterpreterState *interp = _PyInterpreterState_GET();

Py_ssize_t i;

for (i = PyTuple_GET_SIZE(tuple); --i >= 0; ) {

PyObject *v = PyTuple_GET_ITEM(tuple, i);

if (v == NULL || !PyUnicode_CheckExact(v)) {

PyErr_SetString(PyExc_SystemError,

"non-string found in code slot");

return -1;

}

_PyUnicode_InternImmortal(interp, &_PyTuple_ITEMS(tuple)[i]);

}

return 0;

}

/* Intern constants. In the default build, this interns selected string

constants. In the free-threaded build, this also interns non-string

constants. */

static int

intern_constants(PyObject *tuple, int *modified)

{

PyInterpreterState *interp = _PyInterpreterState_GET();

for (Py_ssize_t i = PyTuple_GET_SIZE(tuple); --i >= 0; ) {

PyObject *v = PyTuple_GET_ITEM(tuple, i);

if (PyUnicode_CheckExact(v)) {

if (should_intern_string(v)) {

PyObject *w = v;

_PyUnicode_InternMortal(interp, &v);

if (w != v) {

PyTuple_SET_ITEM(tuple, i, v);

if (modified) {

*modified = 1;

}

}

}

}

else if (PyTuple_CheckExact(v)) {

if (intern_constants(v, NULL) < 0) {

return -1;

}

}

else if (PyFrozenSet_CheckExact(v)) {

PyObject *w = v;

PyObject *tmp = PySequence_Tuple(v);

if (tmp == NULL) {

return -1;

}

int tmp_modified = 0;

if (intern_constants(tmp, &tmp_modified) < 0) {

Py_DECREF(tmp);

return -1;

}

if (tmp_modified) {

v = PyFrozenSet_New(tmp);

if (v == NULL) {

Py_DECREF(tmp);

return -1;

}

PyTuple_SET_ITEM(tuple, i, v);

Py_DECREF(w);

if (modified) {

*modified = 1;

}

}

Py_DECREF(tmp);

}

#ifdef Py_GIL_DISABLED

else if (PySlice_Check(v)) {

PySliceObject *slice = (PySliceObject *)v;

PyObject *tmp = PyTuple_New(3);

if (tmp == NULL) {

return -1;

}

PyTuple_SET_ITEM(tmp, 0, Py_NewRef(slice->start));

PyTuple_SET_ITEM(tmp, 1, Py_NewRef(slice->stop));

PyTuple_SET_ITEM(tmp, 2, Py_NewRef(slice->step));

int tmp_modified = 0;

if (intern_constants(tmp, &tmp_modified) < 0) {

Py_DECREF(tmp);

return -1;

}

if (tmp_modified) {

v = PySlice_New(PyTuple_GET_ITEM(tmp, 0),

PyTuple_GET_ITEM(tmp, 1),

PyTuple_GET_ITEM(tmp, 2));

if (v == NULL) {

Py_DECREF(tmp);

return -1;

}

PyTuple_SET_ITEM(tuple, i, v);

Py_DECREF(slice);

if (modified) {

*modified = 1;

}

}

Py_DECREF(tmp);

}

// Intern non-string constants in the free-threaded build

_PyThreadStateImpl *tstate = (_PyThreadStateImpl *)_PyThreadState_GET();

if (!_Py_IsImmortal(v) && !PyUnicode_CheckExact(v) &&

should_immortalize_constant(v) &&

!tstate->suppress_co_const_immortalization)

{

PyObject *interned = intern_one_constant(v);

if (interned == NULL) {

return -1;

}

else if (interned != v) {

PyTuple_SET_ITEM(tuple, i, interned);

Py_SETREF(v, interned);

if (modified) {

*modified = 1;

}

}

}

#endif

}

return 0;

}

/* Return a shallow copy of a tuple that is

guaranteed to contain exact strings, by converting string subclasses

to exact strings and complaining if a non-string is found. */

static PyObject*

validate_and_copy_tuple(PyObject *tup)

{

PyObject *newtuple;

PyObject *item;

Py_ssize_t i, len;

len = PyTuple_GET_SIZE(tup);

newtuple = PyTuple_New(len);

if (newtuple == NULL)

return NULL;

for (i = 0; i < len; i++) {

item = PyTuple_GET_ITEM(tup, i);

if (PyUnicode_CheckExact(item)) {

Py_INCREF(item);

}

else if (!PyUnicode_Check(item)) {

PyErr_Format(

PyExc_TypeError,

"name tuples must contain only "

"strings, not '%.500s'",

Py_TYPE(item)->tp_name);

Py_DECREF(newtuple);

return NULL;

}

else {

item = _PyUnicode_Copy(item);

if (item == NULL) {

Py_DECREF(newtuple);

return NULL;

}

}

PyTuple_SET_ITEM(newtuple, i, item);

}

return newtuple;

}

static int

init_co_cached(PyCodeObject *self)

{

_PyCoCached *cached = FT_ATOMIC_LOAD_PTR(self->_co_cached);

if (cached != NULL) {

return 0;

}

Py_BEGIN_CRITICAL_SECTION(self);

cached = self->_co_cached;

if (cached == NULL) {

cached = PyMem_New(_PyCoCached, 1);

if (cached == NULL) {

PyErr_NoMemory();

}

else {

cached->_co_code = NULL;

cached->_co_cellvars = NULL;

cached->_co_freevars = NULL;

cached->_co_varnames = NULL;

FT_ATOMIC_STORE_PTR(self->_co_cached, cached);

}

}

Py_END_CRITICAL_SECTION();

return cached != NULL ? 0 : -1;

}

/******************

* _PyCode_New()

******************/

// This is also used in compile.c.

void

_Py_set_localsplus_info(int offset, PyObject *name, _PyLocals_Kind kind,

PyObject *names, PyObject *kinds)

{

PyTuple_SET_ITEM(names, offset, Py_NewRef(name));

_PyLocals_SetKind(kinds, offset, kind);

}

static void

get_localsplus_counts(PyObject *names, PyObject *kinds,

int *pnlocals, int *pncellvars,

int *pnfreevars)

{

int nlocals = 0;

int ncellvars = 0;

int nfreevars = 0;

Py_ssize_t nlocalsplus = PyTuple_GET_SIZE(names);

for (int i = 0; i < nlocalsplus; i++) {

_PyLocals_Kind kind = _PyLocals_GetKind(kinds, i);

if (kind & CO_FAST_LOCAL) {

nlocals += 1;

if (kind & CO_FAST_CELL) {

ncellvars += 1;

}

}

else if (kind & CO_FAST_CELL) {

ncellvars += 1;

}

else if (kind & CO_FAST_FREE) {

nfreevars += 1;

}

}

if (pnlocals != NULL) {

*pnlocals = nlocals;

}

if (pncellvars != NULL) {

*pncellvars = ncellvars;

}

if (pnfreevars != NULL) {

*pnfreevars = nfreevars;

}

}

static PyObject *

get_localsplus_names(PyCodeObject *co, _PyLocals_Kind kind, int num)

{

PyObject *names = PyTuple_New(num);

if (names == NULL) {

return NULL;

}

int index = 0;

for (int offset = 0; offset < co->co_nlocalsplus; offset++) {

_PyLocals_Kind k = _PyLocals_GetKind(co->co_localspluskinds, offset);

if ((k & kind) == 0) {

continue;

}

assert(index < num);

PyObject *name = PyTuple_GET_ITEM(co->co_localsplusnames, offset);

PyTuple_SET_ITEM(names, index, Py_NewRef(name));

index += 1;

}

assert(index == num);

return names;

}

int

_PyCode_Validate(struct _PyCodeConstructor *con)

{

/* Check argument types */

if (con->argcount < con->posonlyargcount || con->posonlyargcount < 0 ||

con->kwonlyargcount < 0 ||

con->stacksize < 0 || con->flags < 0 ||

con->code == NULL || !PyBytes_Check(con->code) ||

con->consts == NULL || !PyTuple_Check(con->consts) ||

con->names == NULL || !PyTuple_Check(con->names) ||

con->localsplusnames == NULL || !PyTuple_Check(con->localsplusnames) ||

con->localspluskinds == NULL || !PyBytes_Check(con->localspluskinds) ||

PyTuple_GET_SIZE(con->localsplusnames)

!= PyBytes_GET_SIZE(con->localspluskinds) ||

con->name == NULL || !PyUnicode_Check(con->name) ||

con->qualname == NULL || !PyUnicode_Check(con->qualname) ||

con->filename == NULL || !PyUnicode_Check(con->filename) ||

con->linetable == NULL || !PyBytes_Check(con->linetable) ||

con->exceptiontable == NULL || !PyBytes_Check(con->exceptiontable)

) {

PyErr_BadInternalCall();

return -1;

}

/* Make sure that code is indexable with an int, this is

a long running assumption in ceval.c and many parts of

the interpreter. */

if (PyBytes_GET_SIZE(con->code) > INT_MAX) {

PyErr_SetString(PyExc_OverflowError,

"code: co_code larger than INT_MAX");

return -1;

}

if (PyBytes_GET_SIZE(con->code) % sizeof(_Py_CODEUNIT) != 0 ||

!_Py_IS_ALIGNED(PyBytes_AS_STRING(con->code), sizeof(_Py_CODEUNIT))

) {

PyErr_SetString(PyExc_ValueError, "code: co_code is malformed");

return -1;

}

/* Ensure that the co_varnames has enough names to cover the arg counts.

* Note that totalargs = nlocals - nplainlocals. We check nplainlocals

* here to avoid the possibility of overflow (however remote). */

int nlocals;

get_localsplus_counts(con->localsplusnames, con->localspluskinds,

&nlocals, NULL, NULL);

int nplainlocals = nlocals -

con->argcount -

con->kwonlyargcount -

((con->flags & CO_VARARGS) != 0) -

((con->flags & CO_VARKEYWORDS) != 0);

if (nplainlocals < 0) {

PyErr_SetString(PyExc_ValueError, "code: co_varnames is too small");

return -1;

}

return 0;

}

extern void

_PyCode_Quicken(_Py_CODEUNIT *instructions, Py_ssize_t size, int enable_counters, int flags);

#ifdef Py_GIL_DISABLED

static _PyCodeArray * _PyCodeArray_New(Py_ssize_t size);

#endif

static int

init_code(PyCodeObject *co, struct _PyCodeConstructor *con)

{

int nlocalsplus = (int)PyTuple_GET_SIZE(con->localsplusnames);

int nlocals, ncellvars, nfreevars;

get_localsplus_counts(con->localsplusnames, con->localspluskinds,

&nlocals, &ncellvars, &nfreevars);

if (con->stacksize == 0) {

con->stacksize = 1;

}

PyInterpreterState *interp = _PyInterpreterState_GET();

co->co_filename = Py_NewRef(con->filename);

co->co_name = Py_NewRef(con->name);

co->co_qualname = Py_NewRef(con->qualname);

_PyUnicode_InternMortal(interp, &co->co_filename);

_PyUnicode_InternMortal(interp, &co->co_name);

_PyUnicode_InternMortal(interp, &co->co_qualname);

co->co_flags = con->flags;

co->co_firstlineno = con->firstlineno;

co->co_linetable = Py_NewRef(con->linetable);

co->co_consts = Py_NewRef(con->consts);

co->co_names = Py_NewRef(con->names);

co->co_localsplusnames = Py_NewRef(con->localsplusnames);

co->co_localspluskinds = Py_NewRef(con->localspluskinds);

co->co_argcount = con->argcount;

co->co_posonlyargcount = con->posonlyargcount;

co->co_kwonlyargcount = con->kwonlyargcount;

co->co_stacksize = con->stacksize;

co->co_exceptiontable = Py_NewRef(con->exceptiontable);

/* derived values */

co->co_nlocalsplus = nlocalsplus;

co->co_nlocals = nlocals;

co->co_framesize = nlocalsplus + con->stacksize + FRAME_SPECIALS_SIZE;

co->co_ncellvars = ncellvars;

co->co_nfreevars = nfreevars;

FT_MUTEX_LOCK(&interp->func_state.mutex);

co->co_version = interp->func_state.next_version;

if (interp->func_state.next_version != 0) {

interp->func_state.next_version++;

}

FT_MUTEX_UNLOCK(&interp->func_state.mutex);

co->_co_monitoring = NULL;

co->_co_instrumentation_version = 0;

/* not set */

co->co_weakreflist = NULL;

co->co_extra = NULL;

co->_co_cached = NULL;

co->co_executors = NULL;

memcpy(_PyCode_CODE(co), PyBytes_AS_STRING(con->code),

PyBytes_GET_SIZE(con->code));

#ifdef Py_GIL_DISABLED

co->co_tlbc = _PyCodeArray_New(INITIAL_SPECIALIZED_CODE_SIZE);

if (co->co_tlbc == NULL) {

return -1;

}

co->co_tlbc->entries[0] = co->co_code_adaptive;

#endif

int entry_point = 0;

while (entry_point < Py_SIZE(co)) {

if (_PyCode_CODE(co)[entry_point].op.code == RESUME &&

(_PyCode_CODE(co)[entry_point].op.arg & RESUME_OPARG_LOCATION_MASK) != RESUME_AT_GEN_EXPR_START

) {

break;

}

entry_point++;

}

co->_co_firsttraceable = entry_point;

#ifdef Py_GIL_DISABLED

int enable_counters = interp->config.tlbc_enabled && interp->opt_config.specialization_enabled;

_PyCode_Quicken(_PyCode_CODE(co), Py_SIZE(co), enable_counters, co->co_flags);

#else

_PyCode_Quicken(_PyCode_CODE(co), Py_SIZE(co), interp->opt_config.specialization_enabled, co->co_flags);

#endif

notify_code_watchers(PY_CODE_EVENT_CREATE, co);

return 0;

}

static int

scan_varint(const uint8_t *ptr)

{

unsigned int read = *ptr++;

unsigned int val = read & 63;

unsigned int shift = 0;

while (read & 64) {

read = *ptr++;

shift += 6;

val |= (read & 63) << shift;

}

return val;

}

static int

scan_signed_varint(const uint8_t *ptr)

{

unsigned int uval = scan_varint(ptr);

if (uval & 1) {

return -(int)(uval >> 1);

}

else {

return uval >> 1;

}

}

static int

get_line_delta(const uint8_t *ptr)

{

int code = ((*ptr) >> 3) & 15;

switch (code) {

case PY_CODE_LOCATION_INFO_NONE:

return 0;

case PY_CODE_LOCATION_INFO_NO_COLUMNS:

case PY_CODE_LOCATION_INFO_LONG:

return scan_signed_varint(ptr+1);

case PY_CODE_LOCATION_INFO_ONE_LINE0:

return 0;

case PY_CODE_LOCATION_INFO_ONE_LINE1:

return 1;

case PY_CODE_LOCATION_INFO_ONE_LINE2:

return 2;

default:

/* Same line */

return 0;

}

}

static PyObject *

remove_column_info(PyObject *locations)

{

Py_ssize_t offset = 0;

const uint8_t *data = (const uint8_t *)PyBytes_AS_STRING(locations);

PyObject *res = PyBytes_FromStringAndSize(NULL, 32);

if (res == NULL) {

PyErr_NoMemory();

return NULL;

}

uint8_t *output = (uint8_t *)PyBytes_AS_STRING(res);

while (offset < PyBytes_GET_SIZE(locations)) {

Py_ssize_t write_offset = output - (uint8_t *)PyBytes_AS_STRING(res);

if (write_offset + 16 >= PyBytes_GET_SIZE(res)) {

if (_PyBytes_Resize(&res, PyBytes_GET_SIZE(res) * 2) < 0) {

return NULL;

}

output = (uint8_t *)PyBytes_AS_STRING(res) + write_offset;

}

int code = (data[offset] >> 3) & 15;

if (code == PY_CODE_LOCATION_INFO_NONE) {

*output++ = data[offset];

}

else {

int blength = (data[offset] & 7)+1;

output += write_location_entry_start(

output, PY_CODE_LOCATION_INFO_NO_COLUMNS, blength);

int ldelta = get_line_delta(&data[offset]);

output += write_signed_varint(output, ldelta);

}

offset++;

while (offset < PyBytes_GET_SIZE(locations) &&

(data[offset] & 128) == 0) {

offset++;

}

}

Py_ssize_t write_offset = output - (uint8_t *)PyBytes_AS_STRING(res);

if (_PyBytes_Resize(&res, write_offset)) {

return NULL;

}

return res;

}

static int

intern_code_constants(struct _PyCodeConstructor *con)

{

#ifdef Py_GIL_DISABLED

PyInterpreterState *interp = _PyInterpreterState_GET();

struct _py_code_state *state = &interp->code_state;

FT_MUTEX_LOCK(&state->mutex);

#endif

if (intern_strings(con->names) < 0) {

goto error;

}

if (intern_constants(con->consts, NULL) < 0) {

goto error;

}

if (intern_strings(con->localsplusnames) < 0) {

goto error;

}

FT_MUTEX_UNLOCK(&state->mutex);

return 0;

error:

FT_MUTEX_UNLOCK(&state->mutex);

return -1;

}

/* The caller is responsible for ensuring that the given data is valid. */

PyCodeObject *

_PyCode_New(struct _PyCodeConstructor *con)

{

if (intern_code_constants(con) < 0) {

return NULL;

}

PyObject *replacement_locations = NULL;

// Compact the linetable if we are opted out of debug

// ranges.

if (!_Py_GetConfig()->code_debug_ranges) {

replacement_locations = remove_column_info(con->linetable);

if (replacement_locations == NULL) {

return NULL;

}

con->linetable = replacement_locations;

}

Py_ssize_t size = PyBytes_GET_SIZE(con->code) / sizeof(_Py_CODEUNIT);

PyCodeObject *co;

#ifdef Py_GIL_DISABLED

co = PyObject_GC_NewVar(PyCodeObject, &PyCode_Type, size);

#else

co = PyObject_NewVar(PyCodeObject, &PyCode_Type, size);

#endif

if (co == NULL) {

Py_XDECREF(replacement_locations);

PyErr_NoMemory();

return NULL;

}

if (init_code(co, con) < 0) {

Py_DECREF(co);

return NULL;

}

#ifdef Py_GIL_DISABLED

co->_co_unique_id = _PyObject_AssignUniqueId((PyObject *)co);

_PyObject_GC_TRACK(co);

#endif

Py_XDECREF(replacement_locations);

return co;

}

/******************

* the legacy "constructors"

******************/

PyCodeObject *

PyUnstable_Code_NewWithPosOnlyArgs(

int argcount, int posonlyargcount, int kwonlyargcount,

int nlocals, int stacksize, int flags,

PyObject *code, PyObject *consts, PyObject *names,

PyObject *varnames, PyObject *freevars, PyObject *cellvars,

PyObject *filename, PyObject *name,

PyObject *qualname, int firstlineno,

PyObject *linetable,

PyObject *exceptiontable)

{

PyCodeObject *co = NULL;

PyObject *localsplusnames = NULL;

PyObject *localspluskinds = NULL;

if (varnames == NULL || !PyTuple_Check(varnames) ||

cellvars == NULL || !PyTuple_Check(cellvars) ||

freevars == NULL || !PyTuple_Check(freevars)

) {

PyErr_BadInternalCall();

return NULL;

}

// Set the "fast locals plus" info.

int nvarnames = (int)PyTuple_GET_SIZE(varnames);

int ncellvars = (int)PyTuple_GET_SIZE(cellvars);

int nfreevars = (int)PyTuple_GET_SIZE(freevars);

int nlocalsplus = nvarnames + ncellvars + nfreevars;

localsplusnames = PyTuple_New(nlocalsplus);

if (localsplusnames == NULL) {

goto error;

}

localspluskinds = PyBytes_FromStringAndSize(NULL, nlocalsplus);

if (localspluskinds == NULL) {

goto error;

}

int offset = 0;

for (int i = 0; i < nvarnames; i++, offset++) {

PyObject *name = PyTuple_GET_ITEM(varnames, i);

_Py_set_localsplus_info(offset, name, CO_FAST_LOCAL,

localsplusnames, localspluskinds);

}

for (int i = 0; i < ncellvars; i++, offset++) {

PyObject *name = PyTuple_GET_ITEM(cellvars, i);

int argoffset = -1;

for (int j = 0; j < nvarnames; j++) {

int cmp = PyUnicode_Compare(PyTuple_GET_ITEM(varnames, j),

name);

assert(!PyErr_Occurred());

if (cmp == 0) {

argoffset = j;

break;

}

}

if (argoffset >= 0) {

// Merge the localsplus indices.

nlocalsplus -= 1;

offset -= 1;

_PyLocals_Kind kind = _PyLocals_GetKind(localspluskinds, argoffset);

_PyLocals_SetKind(localspluskinds, argoffset, kind | CO_FAST_CELL);

continue;

}

_Py_set_localsplus_info(offset, name, CO_FAST_CELL,

localsplusnames, localspluskinds);

}

for (int i = 0; i < nfreevars; i++, offset++) {

PyObject *name = PyTuple_GET_ITEM(freevars, i);

_Py_set_localsplus_info(offset, name, CO_FAST_FREE,

localsplusnames, localspluskinds);

}

// gh-110543: Make sure the CO_FAST_HIDDEN flag is set correctly.

if (!(flags & CO_OPTIMIZED)) {

Py_ssize_t code_len = PyBytes_GET_SIZE(code);

_Py_CODEUNIT *code_data = (_Py_CODEUNIT *)PyBytes_AS_STRING(code);

Py_ssize_t num_code_units = code_len / sizeof(_Py_CODEUNIT);

int extended_arg = 0;

for (int i = 0; i < num_code_units; i += 1 + _PyOpcode_Caches[code_data[i].op.code]) {

_Py_CODEUNIT *instr = &code_data[i];

uint8_t opcode = instr->op.code;

if (opcode == EXTENDED_ARG) {

extended_arg = extended_arg << 8 | instr->op.arg;

continue;

}

if (opcode == LOAD_FAST_AND_CLEAR) {

int oparg = extended_arg << 8 | instr->op.arg;

if (oparg >= nlocalsplus) {

PyErr_Format(PyExc_ValueError,

"code: LOAD_FAST_AND_CLEAR oparg %d out of range",

oparg);

goto error;

}

_PyLocals_Kind kind = _PyLocals_GetKind(localspluskinds, oparg);

_PyLocals_SetKind(localspluskinds, oparg, kind | CO_FAST_HIDDEN);

}

extended_arg = 0;

}

}

// If any cells were args then nlocalsplus will have shrunk.

if (nlocalsplus != PyTuple_GET_SIZE(localsplusnames)) {

if (_PyTuple_Resize(&localsplusnames, nlocalsplus) < 0

|| _PyBytes_Resize(&localspluskinds, nlocalsplus) < 0) {

goto error;

}

}

struct _PyCodeConstructor con = {

.filename = filename,

.name = name,

.qualname = qualname,

.flags = flags,

.code = code,

.firstlineno = firstlineno,

.linetable = linetable,

.consts = consts,

.names = names,

.localsplusnames = localsplusnames,

.localspluskinds = localspluskinds,

.argcount = argcount,

.posonlyargcount = posonlyargcount,

.kwonlyargcount = kwonlyargcount,

.stacksize = stacksize,

.exceptiontable = exceptiontable,

};

if (_PyCode_Validate(&con) < 0) {

goto error;

}

assert(PyBytes_GET_SIZE(code) % sizeof(_Py_CODEUNIT) == 0);

assert(_Py_IS_ALIGNED(PyBytes_AS_STRING(code), sizeof(_Py_CODEUNIT)));

if (nlocals != PyTuple_GET_SIZE(varnames)) {

PyErr_SetString(PyExc_ValueError,

"code: co_nlocals != len(co_varnames)");

goto error;

}

co = _PyCode_New(&con);

if (co == NULL) {

goto error;

}

error:

Py_XDECREF(localsplusnames);

Py_XDECREF(localspluskinds);

return co;

}

PyCodeObject *

PyUnstable_Code_New(int argcount, int kwonlyargcount,

int nlocals, int stacksize, int flags,

PyObject *code, PyObject *consts, PyObject *names,

PyObject *varnames, PyObject *freevars, PyObject *cellvars,

PyObject *filename, PyObject *name, PyObject *qualname,

int firstlineno,

PyObject *linetable,

PyObject *exceptiontable)

{

return PyCode_NewWithPosOnlyArgs(argcount, 0, kwonlyargcount, nlocals,

stacksize, flags, code, consts, names,

varnames, freevars, cellvars, filename,

name, qualname, firstlineno,

linetable,

exceptiontable);

}

// NOTE: When modifying the construction of PyCode_NewEmpty, please also change

// test.test_code.CodeLocationTest.test_code_new_empty to keep it in sync!

static const uint8_t assert0[8] = {

RESUME, RESUME_AT_FUNC_START,

CACHE, 0,

LOAD_COMMON_CONSTANT, CONSTANT_ASSERTIONERROR,

RAISE_VARARGS, 1

};

static const uint8_t linetable[2] = {

(1 << 7) // New entry.

| (PY_CODE_LOCATION_INFO_NO_COLUMNS << 3)

| (4 - 1), // Four code units.

0, // Offset from co_firstlineno.

};

PyCodeObject *

PyCode_NewEmpty(const char *filename, const char *funcname, int firstlineno)

{

PyObject *nulltuple = NULL;

PyObject *filename_ob = NULL;

PyObject *funcname_ob = NULL;

PyObject *code_ob = NULL;

PyObject *linetable_ob = NULL;

PyCodeObject *result = NULL;

nulltuple = PyTuple_New(0);

if (nulltuple == NULL) {

goto failed;

}

funcname_ob = PyUnicode_FromString(funcname);

if (funcname_ob == NULL) {

goto failed;

}

filename_ob = PyUnicode_DecodeFSDefault(filename);

if (filename_ob == NULL) {

goto failed;

}

code_ob = PyBytes_FromStringAndSize((const char *)assert0, 8);

if (code_ob == NULL) {

goto failed;

}

linetable_ob = PyBytes_FromStringAndSize((const char *)linetable, 2);

if (linetable_ob == NULL) {

goto failed;

}

#define emptystring (PyObject *)&_Py_SINGLETON(bytes_empty)

struct _PyCodeConstructor con = {

.filename = filename_ob,

.name = funcname_ob,

.qualname = funcname_ob,

.code = code_ob,

.firstlineno = firstlineno,

.linetable = linetable_ob,

.consts = nulltuple,

.names = nulltuple,

.localsplusnames = nulltuple,

.localspluskinds = emptystring,

.exceptiontable = emptystring,

.stacksize = 1,

};

result = _PyCode_New(&con);