-
-
Notifications
You must be signed in to change notification settings - Fork 442
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[CHG] multiprocessing: The Jobs workers are affected by the database …
…being closed when they run in the Cron Worker process. Thus, when OpenERP run in multiprocess, the connector does not start job workers threads. Instead, the new script ``openerp-connector-worker`` should be used. It spawns processes which start the job workers threads themselves. Example of usage: $ PYTHONPATH=/path/to/server /path/to/connector/openerp-connector-worker \ --config /path/to/configfile \ --workers=2 --logfile=/path/to/logfile This is not ideal as soon as we have to ensure that OpenERP AND the script are running. However: it still works normally when OpenERP is not using multiprocessing and this change allow more control on the worker processes (to implement PG's NOTIFY for instance). More details in the nested history of the revision.
- Loading branch information
Showing
6 changed files
with
236 additions
and
62 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,32 @@ | ||
| .. _multiprocessing: | ||
|
|
||
|
|
||
| ###################################### | ||
| Use the connector with multiprocessing | ||
| ###################################### | ||
|
|
||
| When OpenERP is launched with 1 process, the jobs worker will run | ||
| threaded in the same process. | ||
|
|
||
| When OpenERP is launched with multiple processes using the option | ||
| ``--workers``, the jobs workers are not independant processes, however, | ||
| you have to launch them separately with the script | ||
| ``openerp-connector-worker`` located in the connector module. | ||
|
|
||
| It takes the same arguments and configuration file than the OpenERP | ||
| server. | ||
|
|
||
| .. important:: The Python path must contain the path to the OpenERP | ||
| server when ``openerp-connector-worker`` is launched. | ||
|
|
||
| Example:: | ||
|
|
||
| $ PYTHONPATH=/path/to/server connector/openerp-connector-worker --config /path/to/configfile \ | ||
| --workers=2 --logfile=/path/to/logfile | ||
|
|
||
| The 'Enqueue Jobs' scheduled action is useless when multiprocessing is | ||
| used. | ||
|
|
||
| .. note:: The ``openerp-connector-worker`` should not be launched | ||
| alongside OpenERP when the latter does not run in multiprocess | ||
| mode, because the interprocess signaling would not be done. |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,102 @@ | ||
| #!/usr/bin/env python | ||
| import sys | ||
| import logging | ||
| import os | ||
| import signal | ||
| import time | ||
| import threading | ||
| from contextlib import closing | ||
|
|
||
| import openerp | ||
| from openerp.cli import server as servercli | ||
| import openerp.service.workers as workers | ||
| from openerp.modules.registry import RegistryManager | ||
| from openerp.tools import config | ||
|
|
||
| _logger = logging.getLogger('openerp-connector-worker') | ||
|
|
||
|
|
||
| class Multicornnector(workers.Multicorn): | ||
|
|
||
| def __init__(self, app): | ||
| super(Multicornnector, self).__init__(app) | ||
| self.address = ('0.0.0.0', 0) | ||
| self.population = config['workers'] or 1 | ||
| self.workers_connector = {} | ||
|
|
||
| def process_spawn(self): | ||
| while len(self.workers_connector) < self.population: | ||
| self.worker_spawn(WorkerConnector, self.workers_connector) | ||
|
|
||
| def worker_pop(self, pid): | ||
| if pid in self.workers: | ||
| _logger.debug("Worker (%s) unregistered", pid) | ||
| try: | ||
| self.workers_connector.pop(pid, None) | ||
| u = self.workers.pop(pid) | ||
| u.close() | ||
| except OSError: | ||
| return | ||
|
|
||
|
|
||
| class WorkerConnector(workers.Worker): | ||
| """ HTTP Request workers """ | ||
|
|
||
| def __init__(self, multi): | ||
| super(WorkerConnector, self).__init__(multi) | ||
| self.db_index = 0 | ||
|
|
||
| def process_work(self): | ||
| if config['db_name']: | ||
| db_names = config['db_name'].split(',') | ||
| else: | ||
| services = openerp.netsvc.ExportService._services | ||
| if services.get('db'): | ||
| db_names = services['db'].exp_list(True) | ||
| else: | ||
| db_names = [] | ||
| if len(db_names): | ||
| self.db_index = (self.db_index + 1) % len(db_names) | ||
| db_name = db_names[self.db_index] | ||
| self.setproctitle(db_name) | ||
| db = openerp.sql_db.db_connect(db_name) | ||
| threading.current_thread().dbname = db_name | ||
| with closing(db.cursor()) as cr: | ||
| cr.execute("SELECT 1 FROM ir_module_module " | ||
| "WHERE name = %s " | ||
| "AND state = %s", ('connector', 'installed')) | ||
| if cr.fetchone(): | ||
| RegistryManager.check_registry_signaling(db_name) | ||
| registry = openerp.pooler.get_pool(db_name) | ||
| if registry: | ||
| queue_worker = registry['queue.worker'] | ||
| queue_worker.assign_then_enqueue(cr, | ||
| openerp.SUPERUSER_ID, | ||
| max_jobs=50) | ||
| RegistryManager.signal_caches_change(db_name) | ||
| else: | ||
| self.db_index = 0 | ||
|
|
||
| def sleep(self): | ||
| # Really sleep once all the databases have been processed. | ||
| if self.db_index == 0: | ||
| interval = 15 + self.pid % self.multi.population # chorus effect | ||
| time.sleep(interval) | ||
|
|
||
| def start(self): | ||
| workers.Worker.start(self) | ||
| openerp.service.start_internal() | ||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
| args = sys.argv[1:] | ||
| servercli.check_root_user() | ||
| config.parse_config(args) | ||
|
|
||
| servercli.check_postgres_user() | ||
| openerp.netsvc.init_logger() | ||
| servercli.report_configuration() | ||
|
|
||
| openerp.multi_process = True | ||
| openerp.worker_connector = True | ||
| Multicornnector(openerp.service.wsgi_server.application).run() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.