pengyuthon/odoo_source
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[MERGE] netsvc, netrpc cleanup and wizard kill

Browse Source 
bzr revid: al@openerp.com-20121209190948-ggt4tv0qbernwr6k
This commit is contained in:
Antony Lesuisse
2012-12-09 20:09:48 +01:00
parent 66af21a896 ae45df93a4
commit 1e1d99b6ea
11 changed files with 133 additions and 391 deletions
Download Patch File Download Diff File Expand all files Collapse all files
openerp/__init__.py
-1
View File
@@ -40,7 +40,6 @@ import service
import sql_db
import test
import tools
import wizard
import workflow
# backward compatilbility
# TODO: This is for the web addons, can be removed later.
openerp/modules/registry.py
+1 -9
View File
@@ -240,15 +240,7 @@ class RegistryManager(object):
@classmethod
def delete(cls, db_name):
"""Delete the registry linked to a given database.
This also cleans the associated caches. For good measure this also
cancels the associated cron job. But please note that the cron job can
be running and take some time before ending, and that you should not
remove a registry if it can still be used by some thread. So it might
be necessary to call yourself openerp.cron.Agent.cancel(db_name) and
and join (i.e. wait for) the thread.
"""
"""Delete the registry linked to a given database. """
with cls.registries_lock:
if db_name in cls.registries:
cls.registries[db_name].clear_caches()
openerp/netsvc.py
+3 -100
View File
@@ -40,23 +40,6 @@ import openerp
_logger = logging.getLogger(__name__)
def close_socket(sock):
""" Closes a socket instance cleanly
:param sock: the network socket to close
:type sock: socket.socket
"""
try:
sock.shutdown(socket.SHUT_RDWR)
except socket.error, e:
# On OSX, socket shutdowns both sides if any side closes it
# causing an error 57 'Socket is not connected' on shutdown
# of the other side (or something), see
# http://bugs.python.org/issue4397
# note: stdlib fixed test, not behavior
if e.errno != errno.ENOTCONN or platform.system() not in ['Darwin', 'Windows']:
raise
sock.close()
#.apidoc title: Common Services: netsvc
@@ -67,9 +50,9 @@ def abort_response(dummy_1, description, dummy_2, details):
raise openerp.osv.osv.except_osv(description, details)
class Service(object):
""" Base class for *Local* services
Functionality here is trusted, no authentication.
""" Base class for Local services
Functionality here is trusted, no authentication.
Workflow engine and reports subclass this.
"""
_services = {}
def __init__(self, name):
@@ -145,7 +128,6 @@ class ColoredFormatter(DBFormatter):
record.levelname = COLOR_PATTERN % (30 + fg_color, 40 + bg_color, record.levelname)
return DBFormatter.format(self, record)
def init_logger():
from tools.translate import resetlocale
resetlocale()
@@ -246,85 +228,6 @@ def init_alternative_logger():
logger.addHandler(handler)
logger.setLevel(logging.ERROR)
class Server:
""" Generic interface for all servers with an event loop etc.
Override this to impement http, net-rpc etc. servers.
Servers here must have threaded behaviour. start() must not block,
there is no run().
"""
__is_started = False
__servers = []
__starter_threads = []
# we don't want blocking server calls (think select()) to
# wait forever and possibly prevent exiting the process,
# but instead we want a form of polling/busy_wait pattern, where
# _server_timeout should be used as the default timeout for
# all I/O blocking operations
_busywait_timeout = 0.5
def __init__(self):
Server.__servers.append(self)
if Server.__is_started:
# raise Exception('All instances of servers must be inited before the startAll()')
# Since the startAll() won't be called again, allow this server to
# init and then start it after 1sec (hopefully). Register that
# timer thread in a list, so that we can abort the start if quitAll
# is called in the meantime
t = threading.Timer(1.0, self._late_start)
t.name = 'Late start timer for %s' % str(self.__class__)
Server.__starter_threads.append(t)
t.start()
def start(self):
_logger.debug("called stub Server.start")
def _late_start(self):
self.start()
for thr in Server.__starter_threads:
if thr.finished.is_set():
Server.__starter_threads.remove(thr)
def stop(self):
_logger.debug("called stub Server.stop")
def stats(self):
""" This function should return statistics about the server """
return "%s: No statistics" % str(self.__class__)
@classmethod
def startAll(cls):
if cls.__is_started:
return
_logger.info("Starting %d services" % len(cls.__servers))
for srv in cls.__servers:
srv.start()
cls.__is_started = True
@classmethod
def quitAll(cls):
if not cls.__is_started:
return
_logger.info("Stopping %d services" % len(cls.__servers))
for thr in cls.__starter_threads:
if not thr.finished.is_set():
thr.cancel()
cls.__starter_threads.remove(thr)
for srv in cls.__servers:
srv.stop()
cls.__is_started = False
@classmethod
def allStats(cls):
res = ["Servers %s" % ('stopped', 'started')[cls.__is_started]]
res.extend(srv.stats() for srv in cls.__servers)
return '\n'.join(res)
def _close_socket(self):
close_socket(self.socket)
def replace_request_password(args):
# password is always 3rd argument in a request, we replace it in RPC logs
# so it's easier to forward logs for diagnostics/debugging purposes...
openerp/service/__init__.py
+14 -16
View File
@@ -28,12 +28,12 @@ import cron
import netrpc_server
import web_services
import web_services
import wsgi_server
import openerp.modules
import openerp.netsvc
import openerp.osv
import openerp.tools
import openerp.service.wsgi_server
#.apidoc title: RPC Services
@@ -69,34 +69,32 @@ def start_internal():
return
openerp.netsvc.init_logger()
openerp.modules.loading.open_openerp_namespace()
# Instantiate local services (this is a legacy design).
openerp.osv.osv.start_object_proxy()
# Export (for RPC) services.
web_services.start_web_services()
web_services.start_service()
load_server_wide_modules()
start_internal_done = True
def start_services():
""" Start all services including http, netrpc and cron """
start_internal()
# Initialize the HTTP stack.
netrpc_server.init_servers()
# Start the main cron thread.
cron.start_master_thread()
# Start the top-level servers threads (normally HTTP, HTTPS, and NETRPC).
openerp.netsvc.Server.startAll()
# Initialize the NETRPC server.
netrpc_server.start_service()
# Start the WSGI server.
openerp.service.wsgi_server.start_server()
wsgi_server.start_service()
# Start the main cron thread.
cron.start_service()
def stop_services():
""" Stop all services. """
# stop scheduling new jobs; we will have to wait for the jobs to complete below
openerp.netsvc.Server.quitAll()
openerp.service.wsgi_server.stop_server()
# stop services
cron.stop_service()
netrpc_server.stop_service()
wsgi_server.stop_service()
_logger.info("Initiating shutdown")
_logger.info("Hit CTRL-C again or send a second signal to force the shutdown.")
logging.shutdown()
openerp/service/cron.py
+5 -2
View File
@@ -52,7 +52,7 @@ def cron_runner(number):
if not acquired:
break
def start_master_thread():
def start_service():
""" Start the above runner function in a daemon thread.
The thread is a typical daemon thread: it will never quit and must be
@@ -63,9 +63,12 @@ def start_master_thread():
for i in range(openerp.tools.config['max_cron_threads']):
def target():
cron_runner(i)
t = threading.Thread(target=target, name="openerp.service.cron.cron_runner%d" % i)
t = threading.Thread(target=target, name="openerp.service.cron.cron%d" % i)
t.setDaemon(True)
t.start()
_logger.debug("cron%d started!" % i)
def stop_service():
pass
# vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4:
openerp/service/http_server.py
-2
View File
@@ -176,6 +176,4 @@ class OpenERPAuthProvider(AuthProvider):
self.auth_tries += 1
raise AuthRequiredExc(atype='Basic', realm=self.realm)
#eof
# vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4:
openerp/service/netrpc_server.py
+104 -6
View File
@@ -38,6 +38,103 @@ import openerp.tools as tools
_logger = logging.getLogger(__name__)
def close_socket(sock):
""" Closes a socket instance cleanly
:param sock: the network socket to close
:type sock: socket.socket
"""
try:
sock.shutdown(socket.SHUT_RDWR)
except socket.error, e:
# On OSX, socket shutdowns both sides if any side closes it
# causing an error 57 'Socket is not connected' on shutdown
# of the other side (or something), see
# http://bugs.python.org/issue4397
# note: stdlib fixed test, not behavior
if e.errno != errno.ENOTCONN or platform.system() not in ['Darwin', 'Windows']:
raise
sock.close()
class Server:
""" Generic interface for all servers with an event loop etc.
Override this to impement http, net-rpc etc. servers.
Servers here must have threaded behaviour. start() must not block,
there is no run().
"""
__is_started = False
__servers = []
__starter_threads = []
# we don't want blocking server calls (think select()) to
# wait forever and possibly prevent exiting the process,
# but instead we want a form of polling/busy_wait pattern, where
# _server_timeout should be used as the default timeout for
# all I/O blocking operations
_busywait_timeout = 0.5
def __init__(self):
Server.__servers.append(self)
if Server.__is_started:
# raise Exception('All instances of servers must be inited before the startAll()')
# Since the startAll() won't be called again, allow this server to
# init and then start it after 1sec (hopefully). Register that
# timer thread in a list, so that we can abort the start if quitAll
# is called in the meantime
t = threading.Timer(1.0, self._late_start)
t.name = 'Late start timer for %s' % str(self.__class__)
Server.__starter_threads.append(t)
t.start()
def start(self):
_logger.debug("called stub Server.start")
def _late_start(self):
self.start()
for thr in Server.__starter_threads:
if thr.finished.is_set():
Server.__starter_threads.remove(thr)
def stop(self):
_logger.debug("called stub Server.stop")
def stats(self):
""" This function should return statistics about the server """
return "%s: No statistics" % str(self.__class__)
@classmethod
def startAll(cls):
if cls.__is_started:
return
_logger.info("Starting %d services" % len(cls.__servers))
for srv in cls.__servers:
srv.start()
cls.__is_started = True
@classmethod
def quitAll(cls):
if not cls.__is_started:
return
_logger.info("Stopping %d services" % len(cls.__servers))
for thr in cls.__starter_threads:
if not thr.finished.is_set():
thr.cancel()
cls.__starter_threads.remove(thr)
for srv in cls.__servers:
srv.stop()
cls.__is_started = False
@classmethod
def allStats(cls):
res = ["Servers %s" % ('stopped', 'started')[cls.__is_started]]
res.extend(srv.stats() for srv in cls.__servers)
return '\n'.join(res)
def _close_socket(self):
close_socket(self.socket)
class TinySocketClientThread(threading.Thread):
def __init__(self, sock, threads):
spn = sock and sock.getpeername()
@@ -99,10 +196,10 @@ def netrpc_handle_exception_legacy(e):
return 'AccessDenied ' + str(e)
return openerp.tools.exception_to_unicode(e)
class TinySocketServerThread(threading.Thread,netsvc.Server):
class TinySocketServerThread(threading.Thread,Server):
def __init__(self, interface, port, secure=False):
threading.Thread.__init__(self, name="NetRPCDaemon-%d"%port)
netsvc.Server.__init__(self)
Server.__init__(self)
self.__port = port
self.__interface = interface
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
@@ -157,11 +254,12 @@ class TinySocketServerThread(threading.Thread,netsvc.Server):
netrpcd = None
def init_servers():
def start_service():
global netrpcd
if tools.config.get('netrpc', False):
netrpcd = TinySocketServerThread(
tools.config.get('netrpc_interface', ''),
int(tools.config.get('netrpc_port', 8070)))
netrpcd = TinySocketServerThread(tools.config.get('netrpc_interface', ''), int(tools.config.get('netrpc_port', 8070)))
def stop_service():
Server.quitAll()
# vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4:
openerp/service/web_services.py
+1 -60
View File
@@ -605,63 +605,6 @@ class objects_proxy(netsvc.ExportService):
return res
#
# Wizard ID: 1
# - None = end of wizard
#
# Wizard Type: 'form'
# - form
# - print
#
# Wizard datas: {}
# TODO: change local request to OSE request/reply pattern
#
class wizard(netsvc.ExportService):
def __init__(self, name='wizard'):
netsvc.ExportService.__init__(self,name)
self.id = 0
self.wiz_datas = {}
self.wiz_name = {}
self.wiz_uid = {}
def dispatch(self, method, params):
(db, uid, passwd ) = params[0:3]
threading.current_thread().uid = uid
params = params[3:]
if method not in ['execute','create']:
raise KeyError("Method not supported %s" % method)
security.check(db,uid,passwd)
fn = getattr(self, 'exp_'+method)
res = fn(db, uid, *params)
return res
def _execute(self, db, uid, wiz_id, datas, action, context):
self.wiz_datas[wiz_id].update(datas)
wiz = netsvc.LocalService('wizard.'+self.wiz_name[wiz_id])
return wiz.execute(db, uid, self.wiz_datas[wiz_id], action, context)
def exp_create(self, db, uid, wiz_name, datas=None):
if not datas:
datas={}
#FIXME: this is not thread-safe
self.id += 1
self.wiz_datas[self.id] = {}
self.wiz_name[self.id] = wiz_name
self.wiz_uid[self.id] = uid
return self.id
def exp_execute(self, db, uid, wiz_id, datas, action='init', context=None):
if not context:
context={}
if wiz_id in self.wiz_uid:
if self.wiz_uid[wiz_id] == uid:
return self._execute(db, uid, wiz_id, datas, action, context)
else:
raise openerp.exceptions.AccessDenied()
else:
raise openerp.exceptions.Warning('Wizard not found.')
#
# TODO: set a maximum report number per user to avoid DOS attacks
#
@@ -798,13 +741,11 @@ class report_spool(netsvc.ExportService):
raise Exception, 'ReportNotFound'
def start_web_services():
def start_service():
db()
common()
objects_proxy()
wizard()
report_spool()
# vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4:
openerp/service/wsgi_server.py
+2 -2
View File
@@ -428,14 +428,14 @@ def serve():
_logger.info('HTTP service (werkzeug) running on %s:%s', interface, port)
httpd.serve_forever()
def start_server():
def start_service():
""" Call serve() in its own thread.
The WSGI server can be shutdown with stop_server() below.
"""
threading.Thread(target=serve).start()
def stop_server():
def stop_service():
""" Initiate the shutdown of the WSGI server.
The server is supposed to have been started by start_server() above.
openerp/tools/config.py
+3 -2
View File
@@ -146,8 +146,9 @@ class configmanager(object):
help="specify the TCP IP address for the NETRPC protocol")
group.add_option("--netrpc-port", dest="netrpc_port", my_default=8070,
help="specify the TCP port for the NETRPC protocol", type="int")
group.add_option("--no-netrpc", dest="netrpc", action="store_false", my_default=True,
help="disable the NETRPC protocol")
# Needed a few day for runbot and saas
group.add_option("--no-netrpc", dest="netrpc", action="store_false", my_default=False, help="disable the NETRPC protocol")
group.add_option("--netrpc", dest="netrpc", action="store_true", my_default=False, help="enable the NETRPC protocol")
parser.add_option_group(group)
# WEB
openerp/wizard/__init__.py
-191
View File
@@ -1,191 +0,0 @@
# -*- coding: utf-8 -*-
##############################################################################
#
# OpenERP, Open Source Management Solution
# Copyright (C) 2004-2009 Tiny SPRL (<http://tiny.be>).
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
##############################################################################
import copy
import logging
import openerp.netsvc as netsvc
from openerp.tools.misc import UpdateableStr, UpdateableDict
from openerp.tools.translate import translate
from lxml import etree
import openerp.pooler as pooler
from openerp.osv.osv import except_osv
from openerp.osv.orm import except_orm
_logger = logging.getLogger(__name__)
class except_wizard(Exception):
def __init__(self, name, value):
self.name = name
self.value = value
self.args = (name, value)
class interface(netsvc.Service):
"""
This is the base class used to implement Wizards. This class is deprecated
and `openerp.osv.TransientModel` must be used instead.
"""
states = {}
def __init__(self, name):
assert not self.exists('wizard.'+name), 'The wizard "%s" already exists!' % (name,)
_logger.warning(
"The wizard %s uses the deprecated openerp.wizard.interface class.\n"
"It must use the openerp.osv.TransientModel class instead." % \
name)
super(interface, self).__init__('wizard.'+name)
self.wiz_name = name
def translate_view(self, cr, node, state, lang):
if node.get('string'):
trans = translate(cr, self.wiz_name+','+state, 'wizard_view', lang, node.get('string').encode('utf8'))
if trans:
node.set('string', trans)
for n in node:
self.translate_view(cr, n, state, lang)
def execute_cr(self, cr, uid, data, state='init', context=None):
if not context:
context={}
res = {}
try:
state_def = self.states[state]
result_def = state_def.get('result', {})
actions_res = {}
# iterate through the list of actions defined for this state
for action in state_def.get('actions', []):
# execute them
action_res = action(self, cr, uid, data, context)
assert isinstance(action_res, dict), 'The return value of wizard actions should be a dictionary'
actions_res.update(action_res)
res = copy.copy(result_def)
res['datas'] = actions_res
lang = context.get('lang', False)
if result_def['type'] == 'action':
res['action'] = result_def['action'](self, cr, uid, data, context)
elif result_def['type'] == 'form':
fields = copy.deepcopy(result_def['fields'])
arch = copy.copy(result_def['arch'])
button_list = copy.copy(result_def['state'])
if isinstance(fields, UpdateableDict):
fields = fields.dict
if isinstance(arch, UpdateableStr):
arch = arch.string
# fetch user-set defaut values for the field... shouldn't we pass it the uid?
ir_values_obj = pooler.get_pool(cr.dbname).get('ir.values')
defaults = ir_values_obj.get(cr, uid, 'default', False, [('wizard.'+self.wiz_name, False)])
default_values = dict([(x[1], x[2]) for x in defaults])
for val in fields.keys():
if 'default' in fields[val]:
# execute default method for this field
if callable(fields[val]['default']):
fields[val]['value'] = fields[val]['default'](uid, data, state)
else:
fields[val]['value'] = fields[val]['default']
del fields[val]['default']
else:
# if user has set a default value for the field, use it
if val in default_values:
fields[val]['value'] = default_values[val]
if 'selection' in fields[val]:
if not isinstance(fields[val]['selection'], (tuple, list)):
fields[val] = copy.copy(fields[val])
fields[val]['selection'] = fields[val]['selection'](self, cr, uid, context)
elif lang:
res_name = "%s,%s,%s" % (self.wiz_name, state, val)
trans = lambda x: translate(cr, res_name, 'selection', lang, x) or x
for idx, (key, val2) in enumerate(fields[val]['selection']):
fields[val]['selection'][idx] = (key, trans(val2))
if lang:
# translate fields
for field in fields:
res_name = "%s,%s,%s" % (self.wiz_name, state, field)
trans = translate(cr, res_name, 'wizard_field', lang)
if trans:
fields[field]['string'] = trans
if 'help' in fields[field]:
t = translate(cr, res_name, 'help', lang, fields[field]['help'])
if t:
fields[field]['help'] = t
# translate arch
if not isinstance(arch, UpdateableStr):
doc = etree.XML(arch)
self.translate_view(cr, doc, state, lang)
arch = etree.tostring(doc)
# translate buttons
button_list = list(button_list)
for i, aa in enumerate(button_list):
button_name = aa[0]
trans = translate(cr, self.wiz_name+','+state+','+button_name, 'wizard_button', lang)
if trans:
aa = list(aa)
aa[1] = trans
button_list[i] = aa
res['fields'] = fields
res['arch'] = arch
res['state'] = button_list
elif result_def['type'] == 'choice':
next_state = result_def['next_state'](self, cr, uid, data, context)
return self.execute_cr(cr, uid, data, next_state, context)
except Exception, e:
if isinstance(e, except_wizard) \
or isinstance(e, except_osv) \
or isinstance(e, except_orm):
netsvc.abort_response(2, e.name, 'warning', e.value)
else:
_logger.exception('Exception in call:')
raise
return res
def execute(self, db, uid, data, state='init', context=None):
if not context:
context={}
cr = pooler.get_db(db).cursor()
try:
try:
res = self.execute_cr(cr, uid, data, state, context)
cr.commit()
except Exception:
cr.rollback()
raise
finally:
cr.close()
return res
# vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4: