Skip to content

Instantly share code, notes, and snippets.

@brandtbucher
Last active June 15, 2023 01:21
Show Gist options
  • Save brandtbucher/e26972c919cfe7afaf8b3bc96d1f0e3f to your computer and use it in GitHub Desktop.
Save brandtbucher/e26972c919cfe7afaf8b3bc96d1f0e3f to your computer and use it in GitHub Desktop.
A few examples of thread-safety bugs in nogil-3.12 (debug build).
"""A few examples of thread-safety bugs in nogil-3.12 (debug build)."""
import dis
import threading
import types
# You may need to increase these values (depending on your machine):
WRITERS = 1 << 1
ITEMS = 1 << 10
class _Crasher:
# True if the specialization is in the items (not the _read function):
CHECK_ITEMS = False
@staticmethod
def _read(items):
raise NotImplementedError("_read")
@staticmethod
def _write(items):
raise NotImplementedError("_write")
@staticmethod
def _get_items():
raise NotImplementedError("_get_items")
@classmethod
def _check(cls, f):
for instruction in dis.get_instructions(f, adaptive=True):
if instruction.opname == cls.__name__:
return
raise RuntimeError(f"Missing {cls.__name__} in {f}!")
@classmethod
def crash(cls):
# This might need a few dozen loops in some cases:
while True:
print(".", end="", flush=True)
items = cls._get_items()
writers = []
for _ in range(WRITERS):
writer = threading.Thread(target=cls._write, args=[items])
writers.append(writer)
# Specialize:
for _ in range(2):
cls._read(items)
if cls.CHECK_ITEMS:
for item in items:
cls._check(item)
else:
cls._check(cls._read)
# Run:
for thread in writers:
thread.start()
cls._read(items) # BOOM!
for thread in writers:
thread.join()
class BINARY_SUBSCR_GETITEM(_Crasher):
@staticmethod
def _read(items):
for item in items:
try:
item[None]
except TypeError:
pass
@staticmethod
def _write(items):
for item in items:
try:
del item.__getitem__
except AttributeError:
pass
type(item).__getitem__ = lambda self, item: None
@staticmethod
def _get_items():
class C:
__getitem__ = lambda self, item: None
items = []
for _ in range(ITEMS):
item = C()
items.append(item)
return items
class BINARY_SUBSCR_LIST_INT(_Crasher):
@staticmethod
def _read(items):
for item in items:
try:
item[0]
except IndexError:
pass
@staticmethod
def _write(items):
for item in items:
item.clear()
item.append(None)
@staticmethod
def _get_items():
items = []
for _ in range(ITEMS):
item = [None]
items.append(item)
return items
class FOR_ITER_GEN(_Crasher):
@staticmethod
def _read(items):
for item in items:
try:
for _ in item:
break
except ValueError:
pass
@staticmethod
def _write(items):
for item in items:
try:
for _ in item:
break
except ValueError:
pass
@staticmethod
def _get_items():
def g():
yield
yield
items = []
for _ in range(ITEMS):
item = g()
items.append(item)
return items
class FOR_ITER_LIST(_Crasher):
@staticmethod
def _read(items):
for item in items:
for item in item:
break
@staticmethod
def _write(items):
for item in items:
item.clear()
item.append(None)
@staticmethod
def _get_items():
items = []
for _ in range(ITEMS):
item = [None]
items.append(item)
return items
class LOAD_ATTR_CLASS(_Crasher):
@staticmethod
def _read(items):
for item in items:
try:
item.a
except AttributeError:
pass
@staticmethod
def _write(items):
for item in items:
try:
del item.a
except AttributeError:
pass
item.a = object()
@staticmethod
def _get_items():
class C:
a = object()
items = []
for _ in range(ITEMS):
item = C
items.append(item)
return items
class LOAD_ATTR_GETATTRIBUTE_OVERRIDDEN(_Crasher):
@staticmethod
def _read(items):
for item in items:
try:
item.a
except AttributeError:
pass
@staticmethod
def _write(items):
for item in items:
try:
del item.__getattribute__
except AttributeError:
pass
type(item).__getattribute__ = lambda self, name: None
@staticmethod
def _get_items():
class C:
__getattribute__ = lambda self, name: None
items = []
for _ in range(ITEMS):
item = C()
items.append(item)
return items
class LOAD_ATTR_INSTANCE_VALUE(_Crasher):
@staticmethod
def _read(items):
for item in items:
item.a
@staticmethod
def _write(items):
for item in items:
item.__dict__[None] = None
@staticmethod
def _get_items():
class C:
pass
items = []
for _ in range(ITEMS):
item = C()
item.a = None
items.append(item)
return items
class LOAD_ATTR_METHOD_LAZY_DICT(_Crasher):
@staticmethod
def _read(items):
for item in items:
try:
item.m()
except AttributeError:
pass
@staticmethod
def _write(items):
for item in items:
try:
del item.m
except AttributeError:
pass
type(item).m = lambda self: None
@staticmethod
def _get_items():
class C(Exception):
m = lambda self: None
items = []
for _ in range(ITEMS):
item = C()
items.append(item)
return items
class LOAD_ATTR_METHOD_NO_DICT(_Crasher):
@staticmethod
def _read(items):
for item in items:
try:
item.m()
except AttributeError:
pass
@staticmethod
def _write(items):
for item in items:
try:
del item.m
except AttributeError:
pass
type(item).m = lambda self: None
@staticmethod
def _get_items():
class C:
__slots__ = ()
m = lambda self: None
items = []
for _ in range(ITEMS):
item = C()
items.append(item)
return items
class LOAD_ATTR_METHOD_WITH_VALUES(_Crasher):
@staticmethod
def _read(items):
for item in items:
try:
item.m()
except AttributeError:
pass
@staticmethod
def _write(items):
for item in items:
try:
del item.m
except AttributeError:
pass
type(item).m = lambda self: None
@staticmethod
def _get_items():
class C:
m = lambda self: None
items = []
for _ in range(ITEMS):
item = C()
items.append(item)
return items
class LOAD_ATTR_MODULE(_Crasher):
@staticmethod
def _read(items):
for item in items:
try:
item.__name__
except AttributeError:
pass
@staticmethod
def _write(items):
for item in items:
d = item.__dict__.copy()
item.__dict__.clear()
item.__dict__.update(d)
@staticmethod
def _get_items():
items = []
for _ in range(ITEMS):
item = types.ModuleType("<item>")
items.append(item)
return items
class LOAD_ATTR_PROPERTY(_Crasher):
@staticmethod
def _read(items):
for item in items:
try:
item.a
except AttributeError:
pass
@staticmethod
def _write(items):
for item in items:
try:
del type(item).a
except AttributeError:
pass
type(item).a = property(lambda self: None)
@staticmethod
def _get_items():
class C:
a = property(lambda self: None)
items = []
for _ in range(ITEMS):
item = C()
items.append(item)
return items
class LOAD_ATTR_WITH_HINT(_Crasher):
@staticmethod
def _read(items):
for item in items:
item.a
@staticmethod
def _write(items):
for item in items:
item.__dict__[None] = None
@staticmethod
def _get_items():
class C:
pass
items = []
for _ in range(ITEMS):
item = C()
item.__dict__
item.a = None
items.append(item)
return items
class LOAD_GLOBAL_MODULE(_Crasher):
CHECK_ITEMS = True
@staticmethod
def _read(items):
for item in items:
item()
@staticmethod
def _write(items):
for item in items:
item.__globals__[None] = None
@staticmethod
def _get_items():
items = []
for _ in range(ITEMS):
item = eval("lambda: x", {"x": None})
items.append(item)
return items
class STORE_ATTR_INSTANCE_VALUE(_Crasher):
@staticmethod
def _read(items):
for item in items:
item.a = None
@staticmethod
def _write(items):
for item in items:
item.__dict__[None] = None
@staticmethod
def _get_items():
class C:
pass
items = []
for _ in range(ITEMS):
item = C()
items.append(item)
return items
class STORE_ATTR_WITH_HINT(_Crasher):
@staticmethod
def _read(items):
for item in items:
item.a = None
@staticmethod
def _write(items):
for item in items:
item.__dict__[None] = None
@staticmethod
def _get_items():
class C:
pass
items = []
for _ in range(ITEMS):
item = C()
item.__dict__
items.append(item)
return items
class STORE_SUBSCR_LIST_INT(_Crasher):
@staticmethod
def _read(items):
for item in items:
try:
item[0] = None
except IndexError:
pass
@staticmethod
def _write(items):
for item in items:
item.clear()
item.append(None)
@staticmethod
def _get_items():
items = []
for _ in range(ITEMS):
item = [None]
items.append(item)
return items
class UNPACK_SEQUENCE_LIST(_Crasher):
@staticmethod
def _read(items):
for item in items:
try:
[_] = item
except ValueError:
pass
@staticmethod
def _write(items):
for item in items:
item.clear()
item.append(None)
@staticmethod
def _get_items():
items = []
for _ in range(ITEMS):
item = [None]
items.append(item)
return items
# TODO: Other specializations that *appear* thread-unsafe:
# - CALL_PY_EXACT_ARGS
# - CALL_PY_WITH_DEFAULTS
# - LOAD_GLOBAL_BUILTIN
# Uncomment one of the following lines to crash the interpreter:
# BINARY_SUBSCR_GETITEM.crash()
# BINARY_SUBSCR_LIST_INT.crash()
# FOR_ITER_GEN.crash()
# FOR_ITER_LIST.crash()
# LOAD_ATTR_CLASS.crash()
# LOAD_ATTR_GETATTRIBUTE_OVERRIDDEN.crash()
# LOAD_ATTR_INSTANCE_VALUE.crash()
# LOAD_ATTR_METHOD_LAZY_DICT.crash()
# LOAD_ATTR_METHOD_NO_DICT.crash()
# LOAD_ATTR_METHOD_WITH_VALUES.crash()
# LOAD_ATTR_MODULE.crash()
# LOAD_ATTR_PROPERTY.crash()
# LOAD_ATTR_WITH_HINT.crash()
# LOAD_GLOBAL_MODULE.crash()
# STORE_ATTR_INSTANCE_VALUE.crash()
# STORE_ATTR_WITH_HINT.crash()
# STORE_SUBSCR_LIST_INT.crash()
# UNPACK_SEQUENCE_LIST.crash()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment