openerp/service/workers.py

   1 #-----------------------------------------------------------
   2 # Threaded, Gevent and Prefork Servers
   3 #-----------------------------------------------------------
   4 import datetime
   5 import errno
   6 import fcntl
   7 import logging
   8 import os
   9 import psutil
  10 import random
  11 import resource
  12 import select
  13 import signal
  14 import socket
  15 import sys
  16 import threading
  17 import time
  18 import traceback
  19 import subprocess
  20 import os.path
  21 import platform
  22
  23 import wsgi_server
  24
  25 import werkzeug.serving
  26 try:
  27     from setproctitle import setproctitle
  28 except ImportError:
  29     setproctitle = lambda x: None
  30
  31 import openerp
  32 import openerp.tools.config as config
  33 from openerp.release import nt_service_name
  34 from openerp.tools.misc import stripped_sys_argv
  35
  36 _logger = logging.getLogger(__name__)
  37
  38 SLEEP_INTERVAL = 60 # 1 min
  39
  40 #----------------------------------------------------------
  41 # Common
  42 #----------------------------------------------------------
  43
  44 class CommonServer(object):
  45     def __init__(self, app):
  46         # TODO Change the xmlrpc_* options to http_*
  47         self.app = app
  48         # config
  49         self.interface = config['xmlrpc_interface'] or '0.0.0.0'
  50         self.port = config['xmlrpc_port']
  51         # runtime
  52         self.pid = os.getpid()
  53
  54     def dumpstacks(self):
  55         """ Signal handler: dump a stack trace for each existing thread."""
  56         # code from http://stackoverflow.com/questions/132058/getting-stack-trace-from-a-running-python-application#answer-2569696
  57         # modified for python 2.5 compatibility
  58         threads_info = dict([(th.ident, {'name': th.name,
  59                                         'uid': getattr(th,'uid','n/a')})
  60                                     for th in threading.enumerate()])
  61         code = []
  62         for threadId, stack in sys._current_frames().items():
  63             thread_info = threads_info.get(threadId)
  64             code.append("\n# Thread: %s (id:%s) (uid:%s)" % \
  65                         (thread_info and thread_info['name'] or 'n/a',
  66                          threadId,
  67                          thread_info and thread_info['uid'] or 'n/a'))
  68             for filename, lineno, name, line in traceback.extract_stack(stack):
  69                 code.append('File: "%s", line %d, in %s' % (filename, lineno, name))
  70                 if line:
  71                     code.append("  %s" % (line.strip()))
  72         _logger.info("\n".join(code))
  73
  74     def close_socket(self, sock):
  75         """ Closes a socket instance cleanly
  76         :param sock: the network socket to close
  77         :type sock: socket.socket
  78         """
  79         try:
  80             sock.shutdown(socket.SHUT_RDWR)
  81         except socket.error, e:
  82             # On OSX, socket shutdowns both sides if any side closes it
  83             # causing an error 57 'Socket is not connected' on shutdown
  84             # of the other side (or something), see
  85             # http://bugs.python.org/issue4397
  86             # note: stdlib fixed test, not behavior
  87             if e.errno != errno.ENOTCONN or platform.system() not in ['Darwin', 'Windows']:
  88                 raise
  89         sock.close()
  90
  91 #----------------------------------------------------------
  92 # Threaded
  93 #----------------------------------------------------------
  94
  95 class ThreadedServer(CommonServer):
  96     def __init__(self, app):
  97         super(ThreadedServer, self).__init__(app)
  98         self.main_thread_id = threading.currentThread().ident
  99         # Variable keeping track of the number of calls to the signal handler defined
 100         # below. This variable is monitored by ``quit_on_signals()``.
 101         self.quit_signals_received = 0
 102
 103         #self.socket = None
 104         #self.queue = []
 105         self.httpd = None
 106
 107     def signal_handler(self, sig, frame):
 108         if sig in [signal.SIGINT,signal.SIGTERM]:
 109             # shutdown on kill -INT or -TERM
 110             self.quit_signals_received += 1
 111             if self.quit_signals_received > 1:
 112                 # logging.shutdown was already called at this point.
 113                 sys.stderr.write("Forced shutdown.\n")
 114                 os._exit(0)
 115         elif sig == signal.SIGHUP:
 116             # restart on kill -HUP
 117             openerp.phoenix = True
 118             self.quit_signals_received += 1
 119         elif sig == signal.SIGQUIT:
 120             # dump stacks on kill -3
 121             self.dumpstacks()
 122
 123     def cron_thread(self, number):
 124         while True:
 125             time.sleep(SLEEP_INTERVAL + number) # Steve Reich timing style
 126             registries = openerp.modules.registry.RegistryManager.registries
 127             _logger.debug('cron%d polling for jobs', number)
 128             for db_name, registry in registries.items():
 129                 while True and registry.ready:
 130                     acquired = openerp.addons.base.ir.ir_cron.ir_cron._acquire_job(db_name)
 131                     if not acquired:
 132                         break
 133
 134     def cron_spawn(self):
 135         """ Start the above runner function in a daemon thread.
 136
 137         The thread is a typical daemon thread: it will never quit and must be
 138         terminated when the main process exits - with no consequence (the processing
 139         threads it spawns are not marked daemon).
 140
 141         """
 142         # Force call to strptime just before starting the cron thread
 143         # to prevent time.strptime AttributeError within the thread.
 144         # See: http://bugs.python.org/issue7980
 145         datetime.datetime.strptime('2012-01-01', '%Y-%m-%d')
 146         for i in range(openerp.tools.config['max_cron_threads']):
 147             def target():
 148                 self.cron_thread(i)
 149             t = threading.Thread(target=target, name="openerp.service.cron.cron%d" % i)
 150             t.setDaemon(True)
 151             t.start()
 152             _logger.debug("cron%d started!" % i)
 153
 154     def http_thread(self):
 155         self.httpd = werkzeug.serving.make_server(self.interface, self.port, self.app, threaded=True)
 156         self.httpd.serve_forever()
 157
 158     def http_spawn(self):
 159         threading.Thread(target=self.http_thread).start()
 160         _logger.info('HTTP service (werkzeug) running on %s:%s', self.interface, self.port)
 161
 162     def start(self):
 163         _logger.debug("Setting signal handlers")
 164         if os.name == 'posix':
 165             signal.signal(signal.SIGINT, self.signal_handler)
 166             signal.signal(signal.SIGTERM, self.signal_handler)
 167             signal.signal(signal.SIGCHLD, self.signal_handler)
 168             signal.signal(signal.SIGHUP, self.signal_handler)
 169             signal.signal(signal.SIGQUIT, self.signal_handler)
 170         elif os.name == 'nt':
 171             import win32api
 172             win32api.SetConsoleCtrlHandler(lambda sig: signal_handler(sig, None), 1)
 173         self.cron_spawn()
 174         self.http_spawn()
 175
 176     def stop(self):
 177         """ Shutdown the WSGI server. Wait for non deamon threads.
 178         """
 179         _logger.info("Initiating shutdown")
 180         _logger.info("Hit CTRL-C again or send a second signal to force the shutdown.")
 181
 182         self.httpd.shutdown()
 183         self.close_socket(self.httpd.socket)
 184
 185         # Manually join() all threads before calling sys.exit() to allow a second signal
 186         # to trigger _force_quit() in case some non-daemon threads won't exit cleanly.
 187         # threading.Thread.join() should not mask signals (at least in python 2.5).
 188         me = threading.currentThread()
 189         _logger.debug('current thread: %r', me)
 190         for thread in threading.enumerate():
 191             _logger.debug('process %r (%r)', thread, thread.isDaemon())
 192             if thread != me and not thread.isDaemon() and thread.ident != main_thread_id:
 193                 while thread.isAlive():
 194                     _logger.debug('join and sleep')
 195                     # Need a busyloop here as thread.join() masks signals
 196                     # and would prevent the forced shutdown.
 197                     thread.join(0.05)
 198                     time.sleep(0.05)
 199
 200         _logger.debug('--')
 201         openerp.modules.registry.RegistryManager.delete_all()
 202         logging.shutdown()
 203
 204     def run(self):
 205         """ Start the http server and the cron thread then wait for a signal.
 206
 207         The first SIGINT or SIGTERM signal will initiate a graceful shutdown while
 208         a second one if any will force an immediate exit.
 209         """
 210         self.start()
 211
 212         # Wait for a first signal to be handled. (time.sleep will be interrupted
 213         # by the signal handler.) The try/except is for the win32 case.
 214         try:
 215             while self.quit_signals_received == 0:
 216                 time.sleep(60)
 217         except KeyboardInterrupt:
 218             pass
 219
 220         self.stop()
 221
 222 #----------------------------------------------------------
 223 # Gevent
 224 #----------------------------------------------------------
 225
 226 class GeventServer(CommonServer):
 227     def __init__(self, app):
 228         super(GeventServer, self).__init__(app)
 229         self.port = config['longpolling_port']
 230         self.httpd = None
 231
 232     def watch_parent(self, beat=4):
 233         import gevent
 234         ppid = os.getppid()
 235         while True:
 236             if ppid != os.getppid():
 237                 pid = os.getpid()
 238                 _logger.info("LongPolling (%s) Parent changed", pid)
 239                 # suicide !!
 240                 os.kill(pid, signal.SIGTERM)
 241                 return
 242             gevent.sleep(beat)
 243
 244     def start(self):
 245         import gevent
 246         from gevent.wsgi import WSGIServer
 247         gevent.spawn(self.watch_parent)
 248         self.httpd = WSGIServer((self.interface, self.port), self.app)
 249         _logger.info('Evented Service (longpolling) running on %s:%s', self.interface, self.port)
 250         self.httpd.serve_forever()
 251
 252     def stop(self):
 253         import gevent
 254         self.httpd.stop()
 255         gevent.shutdown()
 256
 257     def run(self):
 258         self.start()
 259         self.stop()
 260
 261 #----------------------------------------------------------
 262 # Prefork
 263 #----------------------------------------------------------
 264
 265 class Multicorn(CommonServer):
 266     """ Multiprocessing inspired by (g)unicorn.
 267     Multicorn currently uses accept(2) as dispatching method between workers
 268     but we plan to replace it by a more intelligent dispatcher to will parse
 269     the first HTTP request line.
 270     """
 271     def __init__(self, app):
 272         # config
 273         self.address = (config['xmlrpc_interface'] or '0.0.0.0', config['xmlrpc_port'])
 274         self.population = config['workers']
 275         self.timeout = config['limit_time_real']
 276         self.limit_request = config['limit_request']
 277         # working vars
 278         self.beat = 4
 279         self.app = app
 280         self.pid = os.getpid()
 281         self.socket = None
 282         self.workers_http = {}
 283         self.workers_cron = {}
 284         self.workers = {}
 285         self.generation = 0
 286         self.queue = []
 287         self.long_polling_pid = None
 288
 289     def pipe_new(self):
 290         pipe = os.pipe()
 291         for fd in pipe:
 292             # non_blocking
 293             flags = fcntl.fcntl(fd, fcntl.F_GETFL) | os.O_NONBLOCK
 294             fcntl.fcntl(fd, fcntl.F_SETFL, flags)
 295             # close_on_exec
 296             flags = fcntl.fcntl(fd, fcntl.F_GETFD) | fcntl.FD_CLOEXEC
 297             fcntl.fcntl(fd, fcntl.F_SETFD, flags)
 298         return pipe
 299
 300     def pipe_ping(self, pipe):
 301         try:
 302             os.write(pipe[1], '.')
 303         except IOError, e:
 304             if e.errno not in [errno.EAGAIN, errno.EINTR]:
 305                 raise
 306
 307     def signal_handler(self, sig, frame):
 308         if len(self.queue) < 5 or sig == signal.SIGCHLD:
 309             self.queue.append(sig)
 310             self.pipe_ping(self.pipe)
 311         else:
 312             _logger.warn("Dropping signal: %s", sig)
 313
 314     def worker_spawn(self, klass, workers_registry):
 315         self.generation += 1
 316         worker = klass(self)
 317         pid = os.fork()
 318         if pid != 0:
 319             worker.pid = pid
 320             self.workers[pid] = worker
 321             workers_registry[pid] = worker
 322             return worker
 323         else:
 324             worker.run()
 325             sys.exit(0)
 326
 327     def long_polling_spawn(self):
 328         nargs = stripped_sys_argv('--pidfile','--workers')
 329         cmd = nargs[0]
 330         cmd = os.path.join(os.path.dirname(cmd), "openerp-gevent")
 331         nargs[0] = cmd
 332         popen = subprocess.Popen(nargs)
 333         self.long_polling_pid = popen.pid
 334
 335     def worker_pop(self, pid):
 336         if pid in self.workers:
 337             _logger.debug("Worker (%s) unregistered",pid)
 338             try:
 339                 self.workers_http.pop(pid,None)
 340                 self.workers_cron.pop(pid,None)
 341                 u = self.workers.pop(pid)
 342                 u.close()
 343             except OSError:
 344                 return
 345
 346     def worker_kill(self, pid, sig):
 347         try:
 348             os.kill(pid, sig)
 349         except OSError, e:
 350             if e.errno == errno.ESRCH:
 351                 self.worker_pop(pid)
 352
 353     def process_signals(self):
 354         while len(self.queue):
 355             sig = self.queue.pop(0)
 356             if sig in [signal.SIGINT,signal.SIGTERM]:
 357                 raise KeyboardInterrupt
 358             elif sig == signal.SIGHUP:
 359                 # restart on kill -HUP
 360                 openerp.phoenix = True
 361                 raise KeyboardInterrupt
 362             elif sig == signal.SIGQUIT:
 363                 # dump stacks on kill -3
 364                 self.dumpstacks()
 365
 366     def process_zombie(self):
 367         # reap dead workers
 368         while 1:
 369             try:
 370                 wpid, status = os.waitpid(-1, os.WNOHANG)
 371                 if not wpid:
 372                     break
 373                 if (status >> 8) == 3:
 374                     msg = "Critial worker error (%s)"
 375                     _logger.critical(msg, wpid)
 376                     raise Exception(msg % wpid)
 377                 self.worker_pop(wpid)
 378             except OSError, e:
 379                 if e.errno == errno.ECHILD:
 380                     break
 381                 raise
 382
 383     def process_timeout(self):
 384         now = time.time()
 385         for (pid, worker) in self.workers.items():
 386             if (worker.watchdog_timeout is not None) and \
 387                 (now - worker.watchdog_time >= worker.watchdog_timeout):
 388                 _logger.error("Worker (%s) timeout", pid)
 389                 self.worker_kill(pid, signal.SIGKILL)
 390
 391     def process_spawn(self):
 392         while len(self.workers_http) < self.population:
 393             self.worker_spawn(WorkerHTTP, self.workers_http)
 394         while len(self.workers_cron) < config['max_cron_threads']:
 395             self.worker_spawn(WorkerCron, self.workers_cron)
 396         if not self.long_polling_pid:
 397             self.long_polling_spawn()
 398
 399     def sleep(self):
 400         try:
 401             # map of fd -> worker
 402             fds = dict([(w.watchdog_pipe[0],w) for k,w in self.workers.items()])
 403             fd_in = fds.keys() + [self.pipe[0]]
 404             # check for ping or internal wakeups
 405             ready = select.select(fd_in, [], [], self.beat)
 406             # update worker watchdogs
 407             for fd in ready[0]:
 408                 if fd in fds:
 409                     fds[fd].watchdog_time = time.time()
 410                 try:
 411                     # empty pipe
 412                     while os.read(fd, 1):
 413                         pass
 414                 except OSError, e:
 415                     if e.errno not in [errno.EAGAIN]:
 416                         raise
 417         except select.error, e:
 418             if e[0] not in [errno.EINTR]:
 419                 raise
 420
 421     def start(self):
 422         # wakeup pipe, python doesnt throw EINTR when a syscall is interrupted
 423         # by a signal simulating a pseudo SA_RESTART. We write to a pipe in the
 424         # signal handler to overcome this behaviour
 425         self.pipe = self.pipe_new()
 426         # set signal
 427         signal.signal(signal.SIGINT, self.signal_handler)
 428         signal.signal(signal.SIGTERM, self.signal_handler)
 429         signal.signal(signal.SIGCHLD, self.signal_handler)
 430         # listen to socket
 431         self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 432         self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 433         self.socket.setblocking(0)
 434         self.socket.bind(self.address)
 435         self.socket.listen(8*self.population)
 436
 437     def stop(self, graceful=True):
 438         if self.long_polling_pid is not None:
 439             self.worker_kill(self.long_polling_pid, signal.SIGKILL)     # FIXME make longpolling process handle SIGTERM correctly
 440             self.long_polling_pid = None
 441         if graceful:
 442             _logger.info("Stopping gracefully")
 443             limit = time.time() + self.timeout
 444             for pid in self.workers.keys():
 445                 self.worker_kill(pid, signal.SIGTERM)
 446             while self.workers and time.time() < limit:
 447                 self.process_zombie()
 448                 time.sleep(0.1)
 449         else:
 450             _logger.info("Stopping forcefully")
 451         for pid in self.workers.keys():
 452             self.worker_kill(pid, signal.SIGTERM)
 453         self.socket.close()
 454
 455     def run(self):
 456         self.start()
 457         _logger.debug("Multiprocess starting")
 458         while 1:
 459             try:
 460                 #_logger.debug("Multiprocess beat (%s)",time.time())
 461                 self.process_signals()
 462                 self.process_zombie()
 463                 self.process_timeout()
 464                 self.process_spawn()
 465                 self.sleep()
 466             except KeyboardInterrupt:
 467                 _logger.debug("Multiprocess clean stop")
 468                 self.stop()
 469                 break
 470             except Exception,e:
 471                 _logger.exception(e)
 472                 self.stop(False)
 473                 sys.exit(-1)
 474
 475 class Worker(object):
 476     """ Workers """
 477     def __init__(self, multi):
 478         self.multi = multi
 479         self.watchdog_time = time.time()
 480         self.watchdog_pipe = multi.pipe_new()
 481         # Can be set to None if no watchdog is desired.
 482         self.watchdog_timeout = multi.timeout
 483         self.ppid = os.getpid()
 484         self.pid = None
 485         self.alive = True
 486         # should we rename into lifetime ?
 487         self.request_max = multi.limit_request
 488         self.request_count = 0
 489
 490     def setproctitle(self, title=""):
 491         setproctitle('openerp: %s %s %s' % (self.__class__.__name__, self.pid, title))
 492
 493     def close(self):
 494         os.close(self.watchdog_pipe[0])
 495         os.close(self.watchdog_pipe[1])
 496
 497     def signal_handler(self, sig, frame):
 498         self.alive = False
 499
 500     def sleep(self):
 501         try:
 502             ret = select.select([self.multi.socket], [], [], self.multi.beat)
 503         except select.error, e:
 504             if e[0] not in [errno.EINTR]:
 505                 raise
 506
 507     def process_limit(self):
 508         # If our parent changed sucide
 509         if self.ppid != os.getppid():
 510             _logger.info("Worker (%s) Parent changed", self.pid)
 511             self.alive = False
 512         # check for lifetime
 513         if self.request_count >= self.request_max:
 514             _logger.info("Worker (%d) max request (%s) reached.", self.pid, self.request_count)
 515             self.alive = False
 516         # Reset the worker if it consumes too much memory (e.g. caused by a memory leak).
 517         rss, vms = psutil.Process(os.getpid()).get_memory_info()
 518         if vms > config['limit_memory_soft']:
 519             _logger.info('Worker (%d) virtual memory limit (%s) reached.', self.pid, vms)
 520             self.alive = False # Commit suicide after the request.
 521
 522         # VMS and RLIMIT_AS are the same thing: virtual memory, a.k.a. address space
 523         soft, hard = resource.getrlimit(resource.RLIMIT_AS)
 524         resource.setrlimit(resource.RLIMIT_AS, (config['limit_memory_hard'], hard))
 525
 526         # SIGXCPU (exceeded CPU time) signal handler will raise an exception.
 527         r = resource.getrusage(resource.RUSAGE_SELF)
 528         cpu_time = r.ru_utime + r.ru_stime
 529         def time_expired(n, stack):
 530             _logger.info('Worker (%d) CPU time limit (%s) reached.', config['limit_time_cpu'])
 531             # We dont suicide in such case
 532             raise Exception('CPU time limit exceeded.')
 533         signal.signal(signal.SIGXCPU, time_expired)
 534         soft, hard = resource.getrlimit(resource.RLIMIT_CPU)
 535         resource.setrlimit(resource.RLIMIT_CPU, (cpu_time + config['limit_time_cpu'], hard))
 536
 537     def process_work(self):
 538         pass
 539
 540     def start(self):
 541         self.pid = os.getpid()
 542         self.setproctitle()
 543         _logger.info("Worker %s (%s) alive", self.__class__.__name__, self.pid)
 544         # Reseed the random number generator
 545         random.seed()
 546         # Prevent fd inherientence close_on_exec
 547         flags = fcntl.fcntl(self.multi.socket, fcntl.F_GETFD) | fcntl.FD_CLOEXEC
 548         fcntl.fcntl(self.multi.socket, fcntl.F_SETFD, flags)
 549         # reset blocking status
 550         self.multi.socket.setblocking(0)
 551         signal.signal(signal.SIGINT, self.signal_handler)
 552         signal.signal(signal.SIGTERM, signal.SIG_DFL)
 553         signal.signal(signal.SIGCHLD, signal.SIG_DFL)
 554
 555     def stop(self):
 556         pass
 557
 558     def run(self):
 559         try:
 560             self.start()
 561             while self.alive:
 562                 self.process_limit()
 563                 self.multi.pipe_ping(self.watchdog_pipe)
 564                 self.sleep()
 565                 self.process_work()
 566             _logger.info("Worker (%s) exiting. request_count: %s.", self.pid, self.request_count)
 567             self.stop()
 568         except Exception,e:
 569             _logger.exception("Worker (%s) Exception occured, exiting..." % self.pid)
 570             # should we use 3 to abort everything ?
 571             sys.exit(1)
 572
 573 class WorkerHTTP(Worker):
 574     """ HTTP Request workers """
 575     def process_request(self, client, addr):
 576         client.setblocking(1)
 577         client.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
 578         # Prevent fd inherientence close_on_exec
 579         flags = fcntl.fcntl(client, fcntl.F_GETFD) | fcntl.FD_CLOEXEC
 580         fcntl.fcntl(client, fcntl.F_SETFD, flags)
 581         # do request using WorkerBaseWSGIServer monkey patched with socket
 582         self.server.socket = client
 583         # tolerate broken pipe when the http client closes the socket before
 584         # receiving the full reply
 585         try:
 586             self.server.process_request(client,addr)
 587         except IOError, e:
 588             if e.errno != errno.EPIPE:
 589                 raise
 590         self.request_count += 1
 591
 592     def process_work(self):
 593         try:
 594             client, addr = self.multi.socket.accept()
 595             self.process_request(client, addr)
 596         except socket.error, e:
 597             if e[0] not in (errno.EAGAIN, errno.ECONNABORTED):
 598                 raise
 599
 600     def start(self):
 601         Worker.start(self)
 602         self.server = WorkerBaseWSGIServer(self.multi.app)
 603
 604 class WorkerBaseWSGIServer(werkzeug.serving.BaseWSGIServer):
 605     """ werkzeug WSGI Server patched to allow using an external listen socket
 606     """
 607     def __init__(self, app):
 608         werkzeug.serving.BaseWSGIServer.__init__(self, "1", "1", app)
 609     def server_bind(self):
 610         # we dont bind beause we use the listen socket of Multicorn#socket
 611         # instead we close the socket
 612         if self.socket:
 613             self.socket.close()
 614     def server_activate(self):
 615         # dont listen as we use Multicorn#socket
 616         pass
 617
 618 class WorkerCron(Worker):
 619     """ Cron workers """
 620
 621     def __init__(self, multi):
 622         super(WorkerCron, self).__init__(multi)
 623         # process_work() below process a single database per call.
 624         # The variable db_index is keeping track of the next database to
 625         # process.
 626         self.db_index = 0
 627
 628     def sleep(self):
 629         # Really sleep once all the databases have been processed.
 630         if self.db_index == 0:
 631             interval = SLEEP_INTERVAL + self.pid % 10 # chorus effect
 632             time.sleep(interval)
 633
 634     def _db_list(self):
 635         if config['db_name']:
 636             db_names = config['db_name'].split(',')
 637         else:
 638             db_names = openerp.service.db.exp_list(True)
 639         return db_names
 640
 641     def process_work(self):
 642         rpc_request = logging.getLogger('openerp.netsvc.rpc.request')
 643         rpc_request_flag = rpc_request.isEnabledFor(logging.DEBUG)
 644         _logger.debug("WorkerCron (%s) polling for jobs", self.pid)
 645         db_names = self._db_list()
 646         if len(db_names):
 647             self.db_index = (self.db_index + 1) % len(db_names)
 648             db_name = db_names[self.db_index]
 649             self.setproctitle(db_name)
 650             if rpc_request_flag:
 651                 start_time = time.time()
 652                 start_rss, start_vms = psutil.Process(os.getpid()).get_memory_info()
 653
 654             import openerp.addons.base as base
 655             base.ir.ir_cron.ir_cron._acquire_job(db_name)
 656             openerp.modules.registry.RegistryManager.delete(db_name)
 657
 658             # dont keep cursors in multi database mode
 659             if len(db_names) > 1:
 660                 openerp.sql_db.close_db(db_name)
 661             if rpc_request_flag:
 662                 end_time = time.time()
 663                 end_rss, end_vms = psutil.Process(os.getpid()).get_memory_info()
 664                 logline = '%s time:%.3fs mem: %sk -> %sk (diff: %sk)' % (db_name, end_time - start_time, start_vms / 1024, end_vms / 1024, (end_vms - start_vms)/1024)
 665                 _logger.debug("WorkerCron (%s) %s", self.pid, logline)
 666
 667             self.request_count += 1
 668             if self.request_count >= self.request_max and self.request_max < len(db_names):
 669                 _logger.error("There are more dabatases to process than allowed "
 670                     "by the `limit_request` configuration variable: %s more.",
 671                     len(db_names) - self.request_max)
 672         else:
 673             self.db_index = 0
 674
 675     def start(self):
 676         os.nice(10)     # mommy always told me to be nice with others...
 677         Worker.start(self)
 678         self.multi.socket.close()
 679
 680         # chorus effect: make cron workers do not all start at first database
 681         mct = config['max_cron_threads']
 682         p = float(self.pid % mct) / mct
 683         self.db_index = int(len(self._db_list()) * p)
 684
 685 #----------------------------------------------------------
 686 # start/stop public api
 687 #----------------------------------------------------------
 688
 689 server = None
 690
 691 def load_server_wide_modules():
 692     for m in openerp.conf.server_wide_modules:
 693         try:
 694             openerp.modules.module.load_openerp_module(m)
 695         except Exception:
 696             msg = ''
 697             if m == 'web':
 698                 msg = """
 699 The `web` module is provided by the addons found in the `openerp-web` project.
 700 Maybe you forgot to add those addons in your addons_path configuration."""
 701             _logger.exception('Failed to load server-wide module `%s`.%s', m, msg)
 702
 703 def _reexec():
 704     """reexecute openerp-server process with (nearly) the same arguments"""
 705     if openerp.tools.osutil.is_running_as_nt_service():
 706         subprocess.call('net stop {0} && net start {0}'.format(nt_service_name), shell=True)
 707     exe = os.path.basename(sys.executable)
 708     args = stripped_sys_argv()
 709     if not args or args[0] != exe:
 710         args.insert(0, exe)
 711     os.execv(sys.executable, args)
 712
 713 def start():
 714     """ Start the openerp http server and cron processor.
 715     """
 716     load_server_wide_modules()
 717     if config['workers']:
 718         openerp.multi_process = True
 719         server = Multicorn(openerp.service.wsgi_server.application)
 720     elif openerp.evented:
 721         server = GeventServer(openerp.service.wsgi_server.application)
 722     else:
 723         server = ThreadedServer(openerp.service.wsgi_server.application)
 724     server.run()
 725
 726     # like the legend of the phoenix, all ends with beginnings
 727     if getattr(openerp, 'phoenix', False):
 728         _reexec()
 729     sys.exit(0)
 730
 731 def restart_server():
 732     """ Restart the server
 733     """
 734     if os.name == 'nt':
 735         # run in a thread to let the current thread return response to the caller.
 736         threading.Thread(target=_reexec).start()
 737     else:
 738         os.kill(server.pid, signal.SIGHUP)
 739
 740
 741 # vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4: