openerp/service/netrpc_server.py

   1 # -*- coding: utf-8 -*-
   2
   3 #
   4 # Copyright P. Christeas <p_christ@hol.gr> 2008,2009
   5 #    Copyright (C) 2004-2009 Tiny SPRL (<http://tiny.be>).
   6 #
   7 #    This program is free software: you can redistribute it and/or modify
   8 #    it under the terms of the GNU Affero General Public License as
   9 #    published by the Free Software Foundation, either version 3 of the
  10 #    License, or (at your option) any later version.
  11 #
  12 #    This program is distributed in the hope that it will be useful,
  13 #    but WITHOUT ANY WARRANTY; without even the implied warranty of
  14 #    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15 #    GNU Affero General Public License for more details.
  16 #
  17 #    You should have received a copy of the GNU Affero General Public License
  18 #    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  19 #
  20 ##############################################################################
  21
  22 #.apidoc title: NET-RPC Server
  23
  24 """ This file contains instance of the net-rpc server
  25
  26
  27 """
  28 import logging
  29 import select
  30 import socket
  31 import sys
  32 import threading
  33 import traceback
  34
  35 import openerp.netsvc as netsvc
  36 import openerp.tiny_socket as tiny_socket
  37 import openerp.tools as tools
  38
  39 class TinySocketClientThread(threading.Thread):
  40     def __init__(self, sock, threads):
  41         spn = sock and sock.getpeername()
  42         spn = 'netrpc-client-%s:%s' % spn[0:2]
  43         threading.Thread.__init__(self, name=spn)
  44         self.sock = sock
  45         # Only at the server side, use a big timeout: close the
  46         # clients connection when they're idle for 20min.
  47         self.sock.settimeout(1200)
  48         self.threads = threads
  49
  50     def run(self):
  51         self.running = True
  52         try:
  53             ts = tiny_socket.mysocket(self.sock)
  54         except Exception:
  55             self.threads.remove(self)
  56             self.running = False
  57             return False
  58
  59         while self.running:
  60             try:
  61                 msg = ts.myreceive()
  62                 auth = getattr(self, 'auth_provider', None)
  63                 result = netsvc.dispatch_rpc(msg[0], msg[1], msg[2:], auth)
  64                 ts.mysend(result)
  65             except socket.timeout:
  66                 #terminate this channel because other endpoint is gone
  67                 break
  68             except netsvc.OpenERPDispatcherException, e:
  69                 try:
  70                     new_e = Exception(tools.exception_to_unicode(e.exception)) # avoid problems of pickeling
  71                     logging.getLogger('web-services').debug("netrpc: rpc-dispatching exception", exc_info=True)
  72                     ts.mysend(new_e, exception=True, traceback=e.traceback)
  73                 except Exception:
  74                     #terminate this channel if we can't properly send back the error
  75                     logging.getLogger('web-services').exception("netrpc: cannot deliver exception message to client")
  76                     break
  77             except Exception, e:
  78                 try:
  79                     tb = getattr(e, 'traceback', sys.exc_info())
  80                     tb_s = "".join(traceback.format_exception(*tb))
  81                     logging.getLogger('web-services').debug("netrpc: communication-level exception", exc_info=True)
  82                     ts.mysend(e, exception=True, traceback=tb_s)
  83                     break
  84                 except Exception, ex:
  85                     #terminate this channel if we can't properly send back the error
  86                     logging.getLogger('web-services').exception("netrpc: cannot deliver exception message to client")
  87                     break
  88
  89         netsvc.close_socket(self.sock)
  90         self.sock = None
  91         self.threads.remove(self)
  92         self.running = False
  93         return True
  94
  95     def stop(self):
  96         self.running = False
  97
  98
  99 class TinySocketServerThread(threading.Thread,netsvc.Server):
 100     def __init__(self, interface, port, secure=False):
 101         threading.Thread.__init__(self, name="NetRPCDaemon-%d"%port)
 102         netsvc.Server.__init__(self)
 103         self.__port = port
 104         self.__interface = interface
 105         self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 106         self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 107         self.socket.bind((self.__interface, self.__port))
 108         self.socket.listen(5)
 109         self.threads = []
 110         netsvc.Logger().notifyChannel("web-services", netsvc.LOG_INFO,
 111                          "starting NET-RPC service at %s port %d" % (interface or '0.0.0.0', port,))
 112
 113     def run(self):
 114         try:
 115             self.running = True
 116             while self.running:
 117                 fd_sets = select.select([self.socket], [], [], self._busywait_timeout)
 118                 if not fd_sets[0]:
 119                     continue
 120                 (clientsocket, address) = self.socket.accept()
 121                 ct = TinySocketClientThread(clientsocket, self.threads)
 122                 clientsocket = None
 123                 self.threads.append(ct)
 124                 ct.start()
 125                 lt = len(self.threads)
 126                 if (lt > 10) and (lt % 10 == 0):
 127                      # Not many threads should be serving at the same time, so log
 128                      # their abuse.
 129                      netsvc.Logger().notifyChannel("web-services", netsvc.LOG_DEBUG,
 130                         "Netrpc: %d threads" % len(self.threads))
 131             self.socket.close()
 132         except Exception, e:
 133             logging.getLogger('web-services').warning("Netrpc: closing because of exception %s" % str(e))
 134             self.socket.close()
 135             return False
 136
 137     def stop(self):
 138         self.running = False
 139         for t in self.threads:
 140             t.stop()
 141         self._close_socket()
 142
 143     def stats(self):
 144         res = "Net-RPC: " + ( (self.running and "running") or  "stopped")
 145         i = 0
 146         for t in self.threads:
 147             i += 1
 148             res += "\nNet-RPC #%d: %s " % (i, t.name)
 149             if t.isAlive():
 150                 res += "running"
 151             else:
 152                 res += "finished"
 153             if t.sock:
 154                 res += ", socket"
 155         return res
 156
 157 netrpcd = None
 158
 159 def init_servers():
 160     global netrpcd
 161     if tools.config.get('netrpc', False):
 162         netrpcd = TinySocketServerThread(
 163             tools.config.get('netrpc_interface', ''),
 164             int(tools.config.get('netrpc_port', 8070)))