/* Type object implementation */

#include "Python.h"

#include "pycore_abstract.h" // _PySequence_IterSearch()

#include "pycore_call.h" // _PyObject_VectorcallTstate()

#include "pycore_code.h" // CO_FAST_FREE

#include "pycore_descrobject.h" // _PyMember_GetOffset()

#include "pycore_dict.h" // _PyDict_KeysSize()

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

#include "pycore_interpframe.h" // _PyInterpreterFrame

#include "pycore_lock.h" // _PySeqLock_*

#include "pycore_long.h" // _PyLong_IsNegative(), _PyLong_GetOne()

#include "pycore_memoryobject.h" // _PyMemoryView_FromBufferProc()

#include "pycore_modsupport.h" // _PyArg_NoKwnames()

#include "pycore_moduleobject.h" // _PyModule_GetDef()

#include "pycore_object.h" // _PyType_HasFeature()

#include "pycore_object_alloc.h" // _PyObject_MallocWithType()

#include "pycore_pyatomic_ft_wrappers.h"

#include "pycore_pyerrors.h" // _PyErr_Occurred()

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

#include "pycore_slots.h" // _PySlotIterator_Init

#include "pycore_symtable.h" // _Py_Mangle()

#include "pycore_tuple.h" // _PyTuple_FromPair

#include "pycore_typeobject.h" // struct type_cache

#include "pycore_unicodeobject.h" // _PyUnicode_Copy

#include "pycore_unionobject.h" // _Py_union_type_or

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

#include "pycore_cell.h" // PyCell_GetRef()

#include "pycore_stats.h"

#include "opcode.h" // MAKE_CELL

#include <stddef.h> // ptrdiff_t

/*[clinic input]

class type "PyTypeObject *" "&PyType_Type"

class object "PyObject *" "&PyBaseObject_Type"

[clinic start generated code]*/

/*[clinic end generated code: output=da39a3ee5e6b4b0d input=4b94608d231c434b]*/

#include "clinic/typeobject.c.h"

/* Support type attribute lookup cache */

/* The cache can keep references to the names alive for longer than

they normally would. This is why the maximum size is limited to

MCACHE_MAX_ATTR_SIZE, since it might be a problem if very large

strings are used as attribute names. */

#define MCACHE_MAX_ATTR_SIZE 100

#define MCACHE_HASH(version, name_hash) \

(((unsigned int)(version) ^ (unsigned int)(name_hash)) \

& ((1 << MCACHE_SIZE_EXP) - 1))

#define MCACHE_HASH_METHOD(type, name) \

MCACHE_HASH(FT_ATOMIC_LOAD_UINT_RELAXED((type)->tp_version_tag), \

((Py_ssize_t)(name)) >> 3)

#define MCACHE_CACHEABLE_NAME(name) \

(PyUnicode_CheckExact(name) && \

(PyUnicode_GET_LENGTH(name) <= MCACHE_MAX_ATTR_SIZE))

#define NEXT_VERSION_TAG(interp) \

(interp)->types.next_version_tag

#ifdef Py_GIL_DISABLED

// There's a global lock for types that ensures that tp_version_tag and

// _spec_cache are correctly updated if the type is modified. It also protects

// tp_mro, tp_bases, and tp_base. This avoids having to take additional locks

// while doing various subclass processing which may result in odd behaviors

// w.r.t. running with the GIL as the outer type lock could be released and

// reacquired during a subclass update if there's contention on the subclass

// lock.

//

// Note that this lock does not protect updates of other type slots or the

// tp_flags member. Instead, we either ensure those updates are done before

// the type has been revealed to other threads or we only do those updates

// while the stop-the-world mechanism is active. The slots and flags are read

// in many places without holding a lock and without atomics.

#define TYPE_LOCK &_PyInterpreterState_GET()->types.mutex

#define BEGIN_TYPE_LOCK() Py_BEGIN_CRITICAL_SECTION_MUTEX(TYPE_LOCK)

#define END_TYPE_LOCK() Py_END_CRITICAL_SECTION()

#define BEGIN_TYPE_DICT_LOCK(d) \

Py_BEGIN_CRITICAL_SECTION2_MUTEX(TYPE_LOCK, &_PyObject_CAST(d)->ob_mutex)

#define END_TYPE_DICT_LOCK() Py_END_CRITICAL_SECTION2()

#if !defined(NDEBUG) || defined(Py_DEBUG)

// Return true if the world is currently stopped.

static bool

types_world_is_stopped(void)

{

PyInterpreterState *interp = _PyInterpreterState_GET();

return interp->stoptheworld.world_stopped;

}

#endif

// Checks that the type has not yet been revealed (exposed) to other

// threads. The _Py_TYPE_REVEALED_FLAG flag is set by type_new() and

// PyType_FromMetaclass() to indicate that a newly initialized type might be

// revealed. We only have ob_flags on 64-bit platforms.

#if SIZEOF_VOID_P > 4

#define TYPE_IS_REVEALED(tp) ((((PyObject *)(tp))->ob_flags & _Py_TYPE_REVEALED_FLAG) != 0)

#else

#define TYPE_IS_REVEALED(tp) 0

#endif

#ifdef Py_DEBUG

#define ASSERT_TYPE_LOCK_HELD() \

if (!types_world_is_stopped()) { _Py_CRITICAL_SECTION_ASSERT_MUTEX_LOCKED(TYPE_LOCK); }

// Checks if we can safely update type slots or tp_flags.

#define ASSERT_WORLD_STOPPED_OR_NEW_TYPE(tp) \

assert(!TYPE_IS_REVEALED(tp) || types_world_is_stopped())

#define ASSERT_NEW_TYPE_OR_LOCKED(tp) \

if (TYPE_IS_REVEALED(tp)) { ASSERT_TYPE_LOCK_HELD(); }

#else

#define ASSERT_TYPE_LOCK_HELD()

#define ASSERT_WORLD_STOPPED_OR_NEW_TYPE(tp)

#define ASSERT_NEW_TYPE_OR_LOCKED(tp)

#endif

static void

types_stop_world(void)

{

assert(!types_world_is_stopped());

PyInterpreterState *interp = _PyInterpreterState_GET();

_PyEval_StopTheWorld(interp);

assert(types_world_is_stopped());

}

static void

types_start_world(void)

{

assert(types_world_is_stopped());

PyInterpreterState *interp = _PyInterpreterState_GET();

_PyEval_StartTheWorld(interp);

assert(!types_world_is_stopped());

}

// This is used to temporarily prevent the TYPE_LOCK from being suspended

// when held by the topmost critical section.

static void

type_lock_prevent_release(void)

{

PyThreadState *tstate = _PyThreadState_GET();

uintptr_t *tagptr = &tstate->critical_section;

PyCriticalSection *c = (PyCriticalSection *)(*tagptr & ~_Py_CRITICAL_SECTION_MASK);

if (!(*tagptr & _Py_CRITICAL_SECTION_TWO_MUTEXES)) {

assert(c->_cs_mutex == TYPE_LOCK);

c->_cs_mutex = NULL;

}

else {

PyCriticalSection2 *c2 = (PyCriticalSection2 *)c;

if (c->_cs_mutex == TYPE_LOCK) {

c->_cs_mutex = c2->_cs_mutex2;

c2->_cs_mutex2 = NULL;

} else {

assert(c2->_cs_mutex2 == TYPE_LOCK);

c2->_cs_mutex2 = NULL;

}

}

}

static void

type_lock_allow_release(void)

{

PyThreadState *tstate = _PyThreadState_GET();

uintptr_t *tagptr = &tstate->critical_section;

PyCriticalSection *c = (PyCriticalSection *)(*tagptr & ~_Py_CRITICAL_SECTION_MASK);

if (!(*tagptr & _Py_CRITICAL_SECTION_TWO_MUTEXES)) {

assert(c->_cs_mutex == NULL);

c->_cs_mutex = TYPE_LOCK;

}

else {

PyCriticalSection2 *c2 = (PyCriticalSection2 *)c;

assert(c2->_cs_mutex2 == NULL);

c2->_cs_mutex2 = TYPE_LOCK;

}

}

#else

#define BEGIN_TYPE_LOCK()

#define END_TYPE_LOCK()

#define BEGIN_TYPE_DICT_LOCK(d)

#define END_TYPE_DICT_LOCK()

#define ASSERT_TYPE_LOCK_HELD()

#define TYPE_IS_REVEALED(tp) 0

#define ASSERT_WORLD_STOPPED_OR_NEW_TYPE(tp)

#define ASSERT_NEW_TYPE_OR_LOCKED(tp)

#define types_world_is_stopped() 1

#define types_stop_world()

#define types_start_world()

#define type_lock_prevent_release()

#define type_lock_allow_release()

#endif

#define PyTypeObject_CAST(op) ((PyTypeObject *)(op))

static void

slot_bf_releasebuffer(PyObject *self, Py_buffer *buffer);

static void

releasebuffer_call_python(PyObject *self, Py_buffer *buffer);

static PyObject *

slot_tp_new(PyTypeObject *type, PyObject *args, PyObject *kwds);

static int

slot_tp_setattro(PyObject *self, PyObject *name, PyObject *value);

static PyObject *

slot_tp_call(PyObject *self, PyObject *args, PyObject *kwds);

static inline PyTypeObject *

type_from_ref(PyObject *ref)

{

PyObject *obj = _PyWeakref_GET_REF(ref);

if (obj == NULL) {

return NULL;

}

return _PyType_CAST(obj);

}

/* helpers for static builtin types */

#ifndef NDEBUG

static inline int

managed_static_type_index_is_set(PyTypeObject *self)

{

return self->tp_subclasses != NULL;

}

#endif

static inline size_t

managed_static_type_index_get(PyTypeObject *self)

{

assert(managed_static_type_index_is_set(self));

/* We store a 1-based index so 0 can mean "not initialized". */

return (size_t)self->tp_subclasses - 1;

}

static inline void

managed_static_type_index_set(PyTypeObject *self, size_t index)

{

assert(index < _Py_MAX_MANAGED_STATIC_BUILTIN_TYPES);

/* We store a 1-based index so 0 can mean "not initialized". */

self->tp_subclasses = (PyObject *)(index + 1);

}

static inline void

managed_static_type_index_clear(PyTypeObject *self)

{

self->tp_subclasses = NULL;

}

static PyTypeObject *

static_ext_type_lookup(PyInterpreterState *interp, size_t index,

int64_t *p_interp_count)

{

assert(interp->runtime == &_PyRuntime);

assert(index < _Py_MAX_MANAGED_STATIC_EXT_TYPES);

size_t full_index = index + _Py_MAX_MANAGED_STATIC_BUILTIN_TYPES;

int64_t interp_count = _Py_atomic_load_int64(

&_PyRuntime.types.managed_static.types[full_index].interp_count);

assert((interp_count == 0) ==

(_PyRuntime.types.managed_static.types[full_index].type == NULL));

*p_interp_count = interp_count;

PyTypeObject *type = interp->types.for_extensions.initialized[index].type;

if (type == NULL) {

return NULL;

}

assert(!interp->types.for_extensions.initialized[index].isbuiltin);

assert(type == _PyRuntime.types.managed_static.types[full_index].type);

assert(managed_static_type_index_is_set(type));

return type;

}

static managed_static_type_state *

managed_static_type_state_get(PyInterpreterState *interp, PyTypeObject *self)

{

// It's probably a builtin type.

size_t index = managed_static_type_index_get(self);

managed_static_type_state *state =

&(interp->types.builtins.initialized[index]);

if (state->type == self) {

return state;

}

if (index > _Py_MAX_MANAGED_STATIC_EXT_TYPES) {

return state;

}

return &(interp->types.for_extensions.initialized[index]);

}

/* For static types we store some state in an array on each interpreter. */

managed_static_type_state *

_PyStaticType_GetState(PyInterpreterState *interp, PyTypeObject *self)

{

assert(self->tp_flags & _Py_TPFLAGS_STATIC_BUILTIN);

return managed_static_type_state_get(interp, self);

}

/* Set the type's per-interpreter state. */

static void

managed_static_type_state_init(PyInterpreterState *interp, PyTypeObject *self,

int isbuiltin, int initial)

{

assert(interp->runtime == &_PyRuntime);

size_t index;

if (initial) {

assert(!managed_static_type_index_is_set(self));

if (isbuiltin) {

index = interp->types.builtins.num_initialized;

assert(index < _Py_MAX_MANAGED_STATIC_BUILTIN_TYPES);

}

else {

PyMutex_Lock(&interp->types.mutex);

index = interp->types.for_extensions.next_index;

interp->types.for_extensions.next_index++;

PyMutex_Unlock(&interp->types.mutex);

assert(index < _Py_MAX_MANAGED_STATIC_EXT_TYPES);

}

managed_static_type_index_set(self, index);

}

else {

index = managed_static_type_index_get(self);

if (isbuiltin) {

assert(index == interp->types.builtins.num_initialized);

assert(index < _Py_MAX_MANAGED_STATIC_BUILTIN_TYPES);

}

else {

assert(index < _Py_MAX_MANAGED_STATIC_EXT_TYPES);

}

}

size_t full_index = isbuiltin

? index

: index + _Py_MAX_MANAGED_STATIC_BUILTIN_TYPES;

assert((initial == 1) ==

(_Py_atomic_load_int64(&_PyRuntime.types.managed_static.types[full_index].interp_count) == 0));

(void)_Py_atomic_add_int64(

&_PyRuntime.types.managed_static.types[full_index].interp_count, 1);

if (initial) {

assert(_PyRuntime.types.managed_static.types[full_index].type == NULL);

_PyRuntime.types.managed_static.types[full_index].type = self;

}

else {

assert(_PyRuntime.types.managed_static.types[full_index].type == self);

}

managed_static_type_state *state = isbuiltin

? &(interp->types.builtins.initialized[index])

: &(interp->types.for_extensions.initialized[index]);

/* It should only be called once for each builtin type per interpreter. */

assert(state->type == NULL);

state->type = self;

state->isbuiltin = isbuiltin;

/* state->tp_subclasses is left NULL until init_subclasses() sets it. */

/* state->tp_weaklist is left NULL until insert_head() or insert_after()

(in weakrefobject.c) sets it. */

if (isbuiltin) {

interp->types.builtins.num_initialized++;

}

else {

interp->types.for_extensions.num_initialized++;

}

}

/* Reset the type's per-interpreter state.

This basically undoes what managed_static_type_state_init() did. */

static void

managed_static_type_state_clear(PyInterpreterState *interp, PyTypeObject *self,

int isbuiltin, int final)

{

size_t index = managed_static_type_index_get(self);

size_t full_index = isbuiltin

? index

: index + _Py_MAX_MANAGED_STATIC_BUILTIN_TYPES;

managed_static_type_state *state = isbuiltin

? &(interp->types.builtins.initialized[index])

: &(interp->types.for_extensions.initialized[index]);

assert(state != NULL);

assert(_Py_atomic_load_int64(&_PyRuntime.types.managed_static.types[full_index].interp_count) > 0);

assert(_PyRuntime.types.managed_static.types[full_index].type == state->type);

assert(state->type != NULL);

state->type = NULL;

assert(state->tp_weaklist == NULL); // It was already cleared out.

(void)_Py_atomic_add_int64(

&_PyRuntime.types.managed_static.types[full_index].interp_count, -1);

if (final) {

assert(!_Py_atomic_load_int64(&_PyRuntime.types.managed_static.types[full_index].interp_count));

_PyRuntime.types.managed_static.types[full_index].type = NULL;

managed_static_type_index_clear(self);

}

if (isbuiltin) {

assert(interp->types.builtins.num_initialized > 0);

interp->types.builtins.num_initialized--;

}

else {

PyMutex_Lock(&interp->types.mutex);

assert(interp->types.for_extensions.num_initialized > 0);

interp->types.for_extensions.num_initialized--;

if (interp->types.for_extensions.num_initialized == 0) {

interp->types.for_extensions.next_index = 0;

}

PyMutex_Unlock(&interp->types.mutex);

}

}

PyObject *

_PyStaticType_GetBuiltins(void)

{

PyInterpreterState *interp = _PyInterpreterState_GET();

Py_ssize_t count = (Py_ssize_t)interp->types.builtins.num_initialized;

assert(count <= _Py_MAX_MANAGED_STATIC_BUILTIN_TYPES);

PyObject *results = PyList_New(count);

if (results == NULL) {

return NULL;

}

for (Py_ssize_t i = 0; i < count; i++) {

PyTypeObject *cls = interp->types.builtins.initialized[i].type;

assert(cls != NULL);

assert(interp->types.builtins.initialized[i].isbuiltin);

PyList_SET_ITEM(results, i, Py_NewRef((PyObject *)cls));

}

return results;

}

// Also see _PyStaticType_InitBuiltin() and _PyStaticType_FiniBuiltin().

/* end static builtin helpers */

static void

type_set_flags(PyTypeObject *tp, unsigned long flags)

{

ASSERT_WORLD_STOPPED_OR_NEW_TYPE(tp);

tp->tp_flags = flags;

}

static void

type_set_flags_with_mask(PyTypeObject *tp, unsigned long mask, unsigned long flags)

{

ASSERT_WORLD_STOPPED_OR_NEW_TYPE(tp);

unsigned long new_flags = (tp->tp_flags & ~mask) | flags;

type_set_flags(tp, new_flags);

}

static void

type_add_flags(PyTypeObject *tp, unsigned long flag)

{

type_set_flags(tp, tp->tp_flags | flag);

}

static void

type_clear_flags(PyTypeObject *tp, unsigned long flag)

{

type_set_flags(tp, tp->tp_flags & ~flag);

}

static inline void

start_readying(PyTypeObject *type)

{

if (type->tp_flags & _Py_TPFLAGS_STATIC_BUILTIN) {

PyInterpreterState *interp = _PyInterpreterState_GET();

managed_static_type_state *state = managed_static_type_state_get(interp, type);

assert(state != NULL);

assert(!state->readying);

state->readying = 1;

return;

}

assert((type->tp_flags & Py_TPFLAGS_READYING) == 0);

type_add_flags(type, Py_TPFLAGS_READYING);

}

static inline void

stop_readying(PyTypeObject *type)

{

if (type->tp_flags & _Py_TPFLAGS_STATIC_BUILTIN) {

PyInterpreterState *interp = _PyInterpreterState_GET();

managed_static_type_state *state = managed_static_type_state_get(interp, type);

assert(state != NULL);

assert(state->readying);

state->readying = 0;

return;

}

assert(type->tp_flags & Py_TPFLAGS_READYING);

type_clear_flags(type, Py_TPFLAGS_READYING);

}

static inline int

is_readying(PyTypeObject *type)

{

if (type->tp_flags & _Py_TPFLAGS_STATIC_BUILTIN) {

PyInterpreterState *interp = _PyInterpreterState_GET();

managed_static_type_state *state = managed_static_type_state_get(interp, type);

assert(state != NULL);

return state->readying;

}

return (type->tp_flags & Py_TPFLAGS_READYING) != 0;

}

/* accessors for objects stored on PyTypeObject */

static inline PyObject *

lookup_tp_dict(PyTypeObject *self)

{

if (self->tp_flags & _Py_TPFLAGS_STATIC_BUILTIN) {

PyInterpreterState *interp = _PyInterpreterState_GET();

managed_static_type_state *state = _PyStaticType_GetState(interp, self);

assert(state != NULL);

return state->tp_dict;

}

return self->tp_dict;

}

PyObject *

_PyType_GetDict(PyTypeObject *self)

{

/* It returns a borrowed reference. */

return lookup_tp_dict(self);

}

PyObject *

PyType_GetDict(PyTypeObject *self)

{

PyObject *dict = lookup_tp_dict(self);

return _Py_XNewRef(dict);

}

static inline void

set_tp_dict(PyTypeObject *self, PyObject *dict)

{

if (self->tp_flags & _Py_TPFLAGS_STATIC_BUILTIN) {

PyInterpreterState *interp = _PyInterpreterState_GET();

managed_static_type_state *state = _PyStaticType_GetState(interp, self);

assert(state != NULL);

state->tp_dict = dict;

return;

}

self->tp_dict = dict;

}

static inline void

clear_tp_dict(PyTypeObject *self)

{

if (self->tp_flags & _Py_TPFLAGS_STATIC_BUILTIN) {

PyInterpreterState *interp = _PyInterpreterState_GET();

managed_static_type_state *state = _PyStaticType_GetState(interp, self);

assert(state != NULL);

Py_CLEAR(state->tp_dict);

return;

}

Py_CLEAR(self->tp_dict);

}

static inline PyObject *

lookup_tp_bases(PyTypeObject *self)

{

return self->tp_bases;

}

PyObject *

_PyType_GetBases(PyTypeObject *self)

{

PyObject *res;

BEGIN_TYPE_LOCK();

res = lookup_tp_bases(self);

Py_INCREF(res);

END_TYPE_LOCK();

return res;

}

static inline void

set_tp_bases(PyTypeObject *self, PyObject *bases, int initial)

{

assert(PyTuple_Check(bases));

ASSERT_NEW_TYPE_OR_LOCKED(self);

if (self->tp_flags & _Py_TPFLAGS_STATIC_BUILTIN) {

// XXX tp_bases can probably be statically allocated for each

// static builtin type.

assert(PyTuple_CheckExact(bases));

assert(initial);

assert(self->tp_bases == NULL);

if (PyTuple_GET_SIZE(bases) == 0) {

assert(self->tp_base == NULL);

}

else {

assert(PyTuple_GET_SIZE(bases) == 1);

assert(PyTuple_GET_ITEM(bases, 0) == (PyObject *)self->tp_base);

assert(self->tp_base->tp_flags & _Py_TPFLAGS_STATIC_BUILTIN);

assert(_Py_IsImmortal(self->tp_base));

}

_Py_SetImmortal(bases);

}

self->tp_bases = bases;

}

static inline void

clear_tp_bases(PyTypeObject *self, int final)

{

if (self->tp_flags & _Py_TPFLAGS_STATIC_BUILTIN) {

if (final) {

if (self->tp_bases != NULL) {

if (PyTuple_GET_SIZE(self->tp_bases) == 0) {

Py_CLEAR(self->tp_bases);

}

else {

assert(_Py_IsImmortal(self->tp_bases));

_Py_ClearImmortal(self->tp_bases);

}

}

}

return;

}

Py_CLEAR(self->tp_bases);

}

static inline PyObject *

lookup_tp_mro(PyTypeObject *self)

{

return self->tp_mro;

}

static inline void

set_tp_mro(PyTypeObject *self, PyObject *mro, int initial)

{

if (mro != NULL) {

assert(PyTuple_CheckExact(mro));

if (self->tp_flags & _Py_TPFLAGS_STATIC_BUILTIN) {

// XXX tp_mro can probably be statically allocated for each

// static builtin type.

assert(initial);

assert(self->tp_mro == NULL);

/* Other checks are done via set_tp_bases. */

_Py_SetImmortal(mro);

}

else {

PyUnstable_Object_EnableDeferredRefcount(mro);

}

}

if (!initial) {

type_lock_prevent_release();

types_stop_world();

}

self->tp_mro = mro;

if (!initial) {

types_start_world();

type_lock_allow_release();

}

}

static inline void

clear_tp_mro(PyTypeObject *self, int final)

{

if (self->tp_flags & _Py_TPFLAGS_STATIC_BUILTIN) {

if (final) {

if (self->tp_mro != NULL) {

if (PyTuple_GET_SIZE(self->tp_mro) == 0) {

Py_CLEAR(self->tp_mro);

}

else {

assert(_Py_IsImmortal(self->tp_mro));

_Py_ClearImmortal(self->tp_mro);

}

}

}

return;

}

Py_CLEAR(self->tp_mro);

}

static PyObject *

init_tp_subclasses(PyTypeObject *self)

{

PyObject *subclasses = PyDict_New();

if (subclasses == NULL) {

return NULL;

}

if (self->tp_flags & _Py_TPFLAGS_STATIC_BUILTIN) {

PyInterpreterState *interp = _PyInterpreterState_GET();

managed_static_type_state *state = _PyStaticType_GetState(interp, self);

state->tp_subclasses = subclasses;

return subclasses;

}

self->tp_subclasses = (void *)subclasses;

return subclasses;

}

static void

clear_tp_subclasses(PyTypeObject *self)

{

/* Delete the dictionary to save memory. _PyStaticType_Dealloc()

callers also test if tp_subclasses is NULL to check if a static type

has no subclass. */

if (self->tp_flags & _Py_TPFLAGS_STATIC_BUILTIN) {

PyInterpreterState *interp = _PyInterpreterState_GET();

managed_static_type_state *state = _PyStaticType_GetState(interp, self);

Py_CLEAR(state->tp_subclasses);

return;

}

Py_CLEAR(self->tp_subclasses);

}

static inline PyObject *

lookup_tp_subclasses(PyTypeObject *self)

{

if (self->tp_flags & _Py_TPFLAGS_STATIC_BUILTIN) {

PyInterpreterState *interp = _PyInterpreterState_GET();

managed_static_type_state *state = _PyStaticType_GetState(interp, self);

assert(state != NULL);

return state->tp_subclasses;

}

return (PyObject *)self->tp_subclasses;

}

PyObject *

_PyType_LookupSubclasses(PyTypeObject *self)

{

return lookup_tp_subclasses(self);

}

PyObject *

_PyType_InitSubclasses(PyTypeObject *self)

{

PyObject *existing = lookup_tp_subclasses(self);

if (existing != NULL) {

return existing;

}

return init_tp_subclasses(self);

}

int

_PyType_HasSubclasses(PyTypeObject *self)

{

PyInterpreterState *interp = _PyInterpreterState_GET();

if (self->tp_flags & _Py_TPFLAGS_STATIC_BUILTIN

// XXX _PyStaticType_GetState() should never return NULL.

&& _PyStaticType_GetState(interp, self) == NULL)

{

return 0;

}

if (lookup_tp_subclasses(self) == NULL) {

return 0;

}

return 1;

}

PyObject*

_PyType_GetSubclasses(PyTypeObject *self)

{

PyObject *list = PyList_New(0);

if (list == NULL) {

return NULL;

}

PyObject *subclasses = lookup_tp_subclasses(self); // borrowed ref

if (subclasses == NULL) {

return list;

}

assert(PyDict_CheckExact(subclasses));

// The loop cannot modify tp_subclasses, there is no need

// to hold a strong reference (use a borrowed reference).

Py_ssize_t i = 0;

PyObject *ref; // borrowed ref

while (PyDict_Next(subclasses, &i, NULL, &ref)) {

PyTypeObject *subclass = type_from_ref(ref);

if (subclass == NULL) {

continue;

}

if (PyList_Append(list, _PyObject_CAST(subclass)) < 0) {

Py_DECREF(list);

Py_DECREF(subclass);

return NULL;

}

Py_DECREF(subclass);

}

return list;

}

/* end accessors for objects stored on PyTypeObject */

/*

* finds the beginning of the docstring's introspection signature.

* if present, returns a pointer pointing to the first '('.

* otherwise returns NULL.

*

* doesn't guarantee that the signature is valid, only that it

* has a valid prefix. (the signature must also pass skip_signature.)

*/

static const char *

find_signature(const char *name, const char *doc)

{

const char *dot;

size_t length;

if (!doc)

return NULL;

assert(name != NULL);

/* for dotted names like classes, only use the last component */

dot = strrchr(name, '.');

if (dot)

name = dot + 1;

length = strlen(name);

if (strncmp(doc, name, length))

return NULL;

doc += length;

if (*doc != '(')

return NULL;

return doc;

}

#define SIGNATURE_END_MARKER ")\n--\n\n"

#define SIGNATURE_END_MARKER_LENGTH 6

/*

* skips past the end of the docstring's introspection signature.

* (assumes doc starts with a valid signature prefix.)

*/

static const char *

skip_signature(const char *doc)

{

while (*doc) {

if ((*doc == *SIGNATURE_END_MARKER) &&

!strncmp(doc, SIGNATURE_END_MARKER, SIGNATURE_END_MARKER_LENGTH))

return doc + SIGNATURE_END_MARKER_LENGTH;

if ((*doc == '\n') && (doc[1] == '\n'))

return NULL;

doc++;

}

return NULL;

}

int

_PyType_CheckConsistency(PyTypeObject *type)

{

#define CHECK(expr) \

do { if (!(expr)) { _PyObject_ASSERT_FAILED_MSG((PyObject *)type, Py_STRINGIFY(expr)); } } while (0)

CHECK(!_PyObject_IsFreed((PyObject *)type));

if (!(type->tp_flags & Py_TPFLAGS_READY)) {

/* don't check static types before PyType_Ready() */

return 1;

}

CHECK(Py_REFCNT(type) >= 1);

CHECK(PyType_Check(type));

CHECK(!is_readying(type));

CHECK(lookup_tp_dict(type) != NULL);

if (type->tp_flags & Py_TPFLAGS_HAVE_GC) {

// bpo-44263: tp_traverse is required if Py_TPFLAGS_HAVE_GC is set.

// Note: tp_clear is optional.

CHECK(type->tp_traverse != NULL);

}

if (type->tp_flags & Py_TPFLAGS_DISALLOW_INSTANTIATION) {

CHECK(type->tp_new == NULL);

CHECK(PyDict_Contains(lookup_tp_dict(type), &_Py_ID(__new__)) == 0);

}

return 1;

#undef CHECK

}

static const char *

_PyType_DocWithoutSignature(const char *name, const char *internal_doc)

{

const char *doc = find_signature(name, internal_doc);

if (doc) {

doc = skip_signature(doc);

if (doc)

return doc;

}

return internal_doc;

}

PyObject *

_PyType_GetDocFromInternalDoc(const char *name, const char *internal_doc)

{

const char *doc = _PyType_DocWithoutSignature(name, internal_doc);

if (!doc || *doc == '\0') {

Py_RETURN_NONE;

}

return PyUnicode_FromString(doc);

}

static const char *

signature_from_flags(int flags)

{

switch (flags & ~METH_COEXIST) {

case METH_NOARGS:

return "($self, /)";

case METH_NOARGS|METH_CLASS:

return "($type, /)";

case METH_NOARGS|METH_STATIC:

return "()";

case METH_O:

return "($self, object, /)";

case METH_O|METH_CLASS:

return "($type, object, /)";

case METH_O|METH_STATIC:

return "(object, /)";

default:

return NULL;

}

}

PyObject *

_PyType_GetTextSignatureFromInternalDoc(const char *name, const char *internal_doc, int flags)

{

const char *start = find_signature(name, internal_doc);

const char *end;

if (start)

end = skip_signature(start);

else

end = NULL;

if (!end) {

start = signature_from_flags(flags);

if (start) {

return PyUnicode_FromString(start);

}

Py_RETURN_NONE;

}

/* back "end" up until it points just past the final ')' */

end -= SIGNATURE_END_MARKER_LENGTH - 1;

assert((end - start) >= 2); /* should be "()" at least */

assert(end[-1] == ')');

assert(end[0] == '\n');

return PyUnicode_FromStringAndSize(start, end - start);

}

static struct type_cache*

get_type_cache(void)

{

PyInterpreterState *interp = _PyInterpreterState_GET();

return &interp->types.type_cache;

}

static void

type_cache_clear(struct type_cache *cache, PyObject *value)

{

for (Py_ssize_t i = 0; i < (1 << MCACHE_SIZE_EXP); i++) {

struct type_cache_entry *entry = &cache->hashtable[i];

#ifdef Py_GIL_DISABLED

_PySeqLock_LockWrite(&entry->sequence);

#endif

entry->version = 0;

Py_XSETREF(entry->name, _Py_XNewRef(value));

entry->value = NULL;

#ifdef Py_GIL_DISABLED

_PySeqLock_UnlockWrite(&entry->sequence);

#endif

}

}

void

_PyType_InitCache(PyInterpreterState *interp)

{