Support pool process with spawn context for cron and worker
When max_tasks_per_child is set, the ProcessPoolExecutor uses a spawn multiprocessing context which starts a new process from scratch with a __name__ being __mp_main__ which prevents the configuration to be restored. So in the ProcessPoolExecutor initializer the configuration is restored using the initial options. Closes #14843 (grafted from cb5c6667ac51818d49ca397347419a40b88768fd) --HG-- branch : 8.0
This commit is contained in:
@@ -13,4 +13,6 @@ if os.path.isdir(DIR):
|
||||
|
||||
from trytond.cli.cron import main # noqa: E402
|
||||
|
||||
main()
|
||||
# Ensure main module can be safely imported by a new interpreter
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
|
||||
@@ -12,4 +12,6 @@ if os.path.isdir(DIR):
|
||||
|
||||
from trytond.cli.worker import main # noqa: E402
|
||||
|
||||
main()
|
||||
# Ensure main module can be safely imported by a new interpreter
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
|
||||
@@ -8,6 +8,8 @@ import time
|
||||
from concurrent import futures
|
||||
from multiprocessing import cpu_count
|
||||
|
||||
import trytond.commandline as commandline
|
||||
import trytond.config as config
|
||||
from trytond.pool import Pool
|
||||
from trytond.transaction import Transaction
|
||||
|
||||
@@ -25,7 +27,7 @@ def run(options):
|
||||
max_workers=processes,
|
||||
mp_context=None,
|
||||
initializer=initializer,
|
||||
initargs=(options.database_names,),
|
||||
initargs=(options,),
|
||||
max_tasks_per_child=options.maxtasksperchild or None,
|
||||
)
|
||||
if sys.version_info < (3, 11):
|
||||
@@ -42,12 +44,15 @@ def run(options):
|
||||
time.sleep(60 - (now.second + now.microsecond / 10**6))
|
||||
|
||||
|
||||
def initializer(database_names, worker=True):
|
||||
def initializer(options, worker=True):
|
||||
# restore configuration when mp context is spawn
|
||||
config.update_etc(options.configfile)
|
||||
commandline.config_log(options)
|
||||
if worker:
|
||||
signal.signal(signal.SIGINT, signal.SIG_IGN)
|
||||
pools = []
|
||||
database_list = Pool.database_list()
|
||||
for database_name in database_names:
|
||||
for database_name in options.database_names:
|
||||
pool = Pool(database_name)
|
||||
if database_name not in database_list:
|
||||
with Transaction().start(database_name, 0, readonly=True):
|
||||
|
||||
@@ -12,7 +12,8 @@ from multiprocessing import cpu_count
|
||||
|
||||
from sql import Flavor
|
||||
|
||||
from trytond import backend, config
|
||||
import trytond.commandline as commandline
|
||||
import trytond.config as config
|
||||
from trytond.exceptions import UserError, UserWarning
|
||||
from trytond.pool import Pool
|
||||
from trytond.status import processing
|
||||
@@ -24,6 +25,7 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
class Queue(object):
|
||||
def __init__(self, database_name, executor):
|
||||
from trytond import backend
|
||||
self.database = backend.Database(database_name).connect()
|
||||
self.connection = self.database.get_connection(autocommit=True)
|
||||
self.executor = executor
|
||||
@@ -57,6 +59,7 @@ def _noop():
|
||||
|
||||
|
||||
def work(options):
|
||||
from trytond import backend
|
||||
Flavor.set(backend.Database.flavor)
|
||||
if not config.getboolean('queue', 'worker', default=False):
|
||||
return
|
||||
@@ -69,7 +72,7 @@ def work(options):
|
||||
max_workers=processes,
|
||||
mp_context=None,
|
||||
initializer=initializer,
|
||||
initargs=(options.database_names,),
|
||||
initargs=(options,),
|
||||
max_tasks_per_child=options.maxtasksperchild or None,
|
||||
)
|
||||
if sys.version_info < (3, 11):
|
||||
@@ -112,12 +115,15 @@ def work(options):
|
||||
queue.get_notifications()
|
||||
|
||||
|
||||
def initializer(database_names, worker=True):
|
||||
def initializer(options, worker=True):
|
||||
# restore configuration when mp context is spawn
|
||||
config.update_etc(options.configfile)
|
||||
commandline.config_log(options)
|
||||
if worker:
|
||||
signal.signal(signal.SIGINT, signal.SIG_IGN)
|
||||
pools = []
|
||||
database_list = Pool.database_list()
|
||||
for database_name in database_names:
|
||||
for database_name in options.database_names:
|
||||
pool = Pool(database_name)
|
||||
if database_name not in database_list:
|
||||
with Transaction().start(database_name, 0, readonly=True):
|
||||
@@ -127,6 +133,7 @@ def initializer(database_names, worker=True):
|
||||
|
||||
|
||||
def run_task(pool, task_id):
|
||||
from trytond import backend
|
||||
if not isinstance(pool, Pool):
|
||||
database_list = Pool.database_list()
|
||||
pool = Pool(pool)
|
||||
|
||||
Reference in New Issue
Block a user