Lock the whole manipulation of the instance __dict__ when accessing attributes
Previously the lock was too fine grained and missed some concurrent access that could result in AttributeError. We implement a slightly changed double-checked locking technique to ensure a fast path when the initialization already occured. Closes #15122
This commit is contained in:
@@ -6,13 +6,14 @@ import datetime as dt
|
||||
import doctest
|
||||
import email.message
|
||||
import os
|
||||
import threading
|
||||
import unittest
|
||||
from copy import deepcopy
|
||||
from decimal import Decimal
|
||||
from io import BytesIO
|
||||
from multiprocessing import Process, set_start_method
|
||||
from multiprocessing.managers import SharedMemoryManager
|
||||
from unittest.mock import patch
|
||||
from unittest.mock import Mock, patch
|
||||
from uuid import uuid4
|
||||
|
||||
import sql
|
||||
@@ -38,7 +39,7 @@ from trytond.tools.domain_inversion import (
|
||||
prepare_reference_domain, simplify, sort, unique_value)
|
||||
from trytond.tools.immutabledict import ImmutableDict
|
||||
from trytond.tools.logging import format_args
|
||||
from trytond.tools.multiprocessing import local
|
||||
from trytond.tools.multiprocessing import _local_impl, local
|
||||
from trytond.tools.string_ import LazyString, StringPartitioned
|
||||
|
||||
try:
|
||||
@@ -1518,27 +1519,80 @@ class ProcessLocalTestCase(TestCase):
|
||||
set_start_method(None, force=True)
|
||||
|
||||
def test_without_fork(self):
|
||||
mydata = local()
|
||||
mydata.number = 42
|
||||
initialize = Mock(wraps=_local_impl.initialize)
|
||||
|
||||
def initialize_wrapper(self, local_obj):
|
||||
initialize(self, local_obj)
|
||||
|
||||
with patch.object(
|
||||
_local_impl, 'initialize', initialize_wrapper):
|
||||
mydata = local()
|
||||
mydata.number = 42
|
||||
initialize.assert_called_once()
|
||||
self.assertEqual(mydata.number, 42)
|
||||
self.assertEqual(mydata.__dict__, {'number': 42})
|
||||
|
||||
def test_fork(self):
|
||||
mydata = local()
|
||||
mydata.number = 42
|
||||
initialize = Mock(wraps=_local_impl.initialize)
|
||||
|
||||
def f(l, num):
|
||||
mydata.number = num
|
||||
l[0] = mydata.number
|
||||
def initialize_wrapper(self, local_obj):
|
||||
initialize(self, local_obj)
|
||||
|
||||
with SharedMemoryManager() as smm:
|
||||
sl = smm.ShareableList([None])
|
||||
p = Process(target=f, args=(sl, 1))
|
||||
p.start()
|
||||
p.join()
|
||||
self.assertEqual(list(sl), [1])
|
||||
with patch.object(
|
||||
_local_impl, 'initialize', initialize_wrapper):
|
||||
mydata = local()
|
||||
mydata.number = 42
|
||||
initialize.assert_called_once()
|
||||
|
||||
self.assertEqual(mydata.number, 42)
|
||||
def f(l, num):
|
||||
mydata.number = num
|
||||
l[0] = mydata.number
|
||||
l[1] = initialize.call_count
|
||||
|
||||
with SharedMemoryManager() as smm:
|
||||
sl = smm.ShareableList([None, None])
|
||||
p = Process(target=f, args=(sl, 1))
|
||||
p.start()
|
||||
p.join()
|
||||
self.assertEqual(list(sl), [1, 2])
|
||||
|
||||
initialize.assert_called_once()
|
||||
self.assertEqual(mydata.number, 42)
|
||||
|
||||
def test_fork_with_threads(self):
|
||||
initialize = Mock(wraps=_local_impl.initialize)
|
||||
|
||||
def initialize_wrapper(self, local_obj):
|
||||
initialize(self, local_obj)
|
||||
|
||||
with patch.object(
|
||||
_local_impl, 'initialize', initialize_wrapper):
|
||||
mydata = local()
|
||||
initialize.assert_not_called()
|
||||
|
||||
def worker_thread():
|
||||
mydata.number = 0
|
||||
|
||||
def f(l):
|
||||
threads = []
|
||||
for _ in range(4):
|
||||
t = threading.Thread(target=worker_thread)
|
||||
t.start()
|
||||
threads.append(t)
|
||||
for t in threads:
|
||||
t.join()
|
||||
|
||||
l[0] = mydata.number
|
||||
l[1] = initialize.call_count
|
||||
|
||||
with SharedMemoryManager() as smm:
|
||||
sl = smm.ShareableList([None, None])
|
||||
p = Process(target=f, args=(sl,))
|
||||
p.start()
|
||||
p.join()
|
||||
self.assertEqual(list(sl), [0, 1])
|
||||
|
||||
initialize.assert_not_called()
|
||||
|
||||
def test_inheritance_setattr(self):
|
||||
class MyLocal(local):
|
||||
|
||||
@@ -3,21 +3,49 @@
|
||||
# this repository contains the full copyright notices and license terms.
|
||||
|
||||
import os
|
||||
from threading import Lock
|
||||
from threading import RLock
|
||||
|
||||
_lock = Lock()
|
||||
_lock = RLock()
|
||||
|
||||
|
||||
class _local_impl:
|
||||
__slots__ = 'dict', 'pid', 'localargs'
|
||||
__slots__ = 'dict', 'pid', 'localargs', 'initialized'
|
||||
|
||||
def __init__(self, localargs):
|
||||
self.localargs = localargs
|
||||
self.reset()
|
||||
|
||||
def reset(self):
|
||||
self.dict = {}
|
||||
self.pid = os.getpid()
|
||||
self.initialized = None
|
||||
|
||||
def initialize(self, local_obj):
|
||||
self.initialized = False
|
||||
self.pid = os.getpid()
|
||||
self.dict = {}
|
||||
object.__setattr__(local_obj, '__dict__', self.dict)
|
||||
args, kwargs = self.localargs
|
||||
init = object.__getattribute__(local_obj, '__init__')
|
||||
init(*args, **kwargs)
|
||||
self.initialized = True
|
||||
|
||||
|
||||
# We'll be using the double-checked locking technique to have a fast path in
|
||||
# the code once the local object has been initialized.
|
||||
# https://en.wikipedia.org/wiki/Double-checked_locking
|
||||
#
|
||||
# But it comes with a twist because we're using a re-entrant lock and a call to
|
||||
# __getattribute__ could trigger a call to __setattr__ through the __init__
|
||||
# call.
|
||||
#
|
||||
# Hence this is why the second check is not exactly the same as the first one.
|
||||
# None means that it's not yet started, False that we're initializing the
|
||||
# object and True that it's fully initialized. We're taking advantage of the
|
||||
# difference between False and None to initialize the object without triggering
|
||||
# an infinite recursion.
|
||||
def _initialize_for_current_process(local_obj):
|
||||
impl = object.__getattribute__(local_obj, '_local_impl')
|
||||
if not impl.initialized or impl.pid != os.getpid():
|
||||
with _lock:
|
||||
if impl.initialized is None or impl.pid != os.getpid():
|
||||
impl.initialize(local_obj)
|
||||
|
||||
|
||||
class local:
|
||||
@@ -30,27 +58,9 @@ class local:
|
||||
return self
|
||||
|
||||
def __getattribute__(self, name):
|
||||
impl = object.__getattribute__(self, '_local_impl')
|
||||
if impl.pid == os.getpid():
|
||||
d = impl.dict
|
||||
else:
|
||||
impl.reset()
|
||||
d = impl.dict
|
||||
args, kwargs = impl.localargs
|
||||
with _lock:
|
||||
self.__init__(*args, **kwargs)
|
||||
object.__setattr__(self, '__dict__', d)
|
||||
_initialize_for_current_process(self)
|
||||
return object.__getattribute__(self, name)
|
||||
|
||||
def __setattr__(self, name, value):
|
||||
impl = object.__getattribute__(self, '_local_impl')
|
||||
if impl.pid == os.getpid():
|
||||
d = impl.dict
|
||||
else:
|
||||
impl.reset()
|
||||
d = impl.dict
|
||||
args, kwargs = impl.localargs
|
||||
with _lock:
|
||||
self.__init__(*args, **kwargs)
|
||||
object.__setattr__(self, '__dict__', d)
|
||||
_initialize_for_current_process(self)
|
||||
return object.__setattr__(self, name, value)
|
||||
|
||||
Reference in New Issue
Block a user