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 """ This file contains instance of the net-rpc server
  23
  24
  25 """
  26 import logging
  27 import select
  28 import socket
  29 import sys
  30 import threading
  31 import traceback
  32
  33 import openerp.netsvc as netsvc
  34 import openerp.tiny_socket as tiny_socket
  35 import openerp.tools as tools
  36
  37 class TinySocketClientThread(threading.Thread, netsvc.OpenERPDispatcher):
  38     def __init__(self, sock, threads):
  39         spn = sock and sock.getpeername()
  40         spn = 'netrpc-client-%s:%s' % spn[0:2]
  41         threading.Thread.__init__(self, name=spn)
  42         self.sock = sock
  43         # Only at the server side, use a big timeout: close the
  44         # clients connection when they're idle for 20min.
  45         self.sock.settimeout(1200)
  46         self.threads = threads
  47
  48     def run(self):
  49         self.running = True
  50         try:
  51             ts = tiny_socket.mysocket(self.sock)
  52         except Exception:
  53             self.threads.remove(self)
  54             self.running = False
  55             return False
  56
  57         while self.running:
  58             try:
  59                 msg = ts.myreceive()
  60                 result = self.dispatch(msg[0], msg[1], msg[2:])
  61                 ts.mysend(result)
  62             except socket.timeout:
  63                 #terminate this channel because other endpoint is gone
  64                 break
  65             except netsvc.OpenERPDispatcherException, e:
  66                 try:
  67                     new_e = Exception(tools.exception_to_unicode(e.exception)) # avoid problems of pickeling
  68                     logging.getLogger('web-services').debug("netrpc: rpc-dispatching exception", exc_info=True)
  69                     ts.mysend(new_e, exception=True, traceback=e.traceback)
  70                 except Exception:
  71                     #terminate this channel if we can't properly send back the error
  72                     logging.getLogger('web-services').exception("netrpc: cannot deliver exception message to client")
  73                     break
  74             except Exception, e:
  75                 try:
  76                     tb = getattr(e, 'traceback', sys.exc_info())
  77                     tb_s = "".join(traceback.format_exception(*tb))
  78                     logging.getLogger('web-services').debug("netrpc: communication-level exception", exc_info=True)
  79                     ts.mysend(e, exception=True, traceback=tb_s)
  80                     break
  81                 except Exception, ex:
  82                     #terminate this channel if we can't properly send back the error
  83                     logging.getLogger('web-services').exception("netrpc: cannot deliver exception message to client")
  84                     break
  85
  86         netsvc.close_socket(self.sock)
  87         self.sock = None
  88         self.threads.remove(self)
  89         self.running = False
  90         return True
  91
  92     def stop(self):
  93         self.running = False
  94
  95
  96 class TinySocketServerThread(threading.Thread,netsvc.Server):
  97     def __init__(self, interface, port, secure=False):
  98         threading.Thread.__init__(self, name="NetRPCDaemon-%d"%port)
  99         netsvc.Server.__init__(self)
 100         self.__port = port
 101         self.__interface = interface
 102         self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 103         self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 104         self.socket.bind((self.__interface, self.__port))
 105         self.socket.listen(5)
 106         self.threads = []
 107         netsvc.Logger().notifyChannel("web-services", netsvc.LOG_INFO,
 108                          "starting NET-RPC service at %s port %d" % (interface or '0.0.0.0', port,))
 109
 110     def run(self):
 111         try:
 112             self.running = True
 113             while self.running:
 114                 fd_sets = select.select([self.socket], [], [], self._busywait_timeout)
 115                 if not fd_sets[0]:
 116                     continue
 117                 (clientsocket, address) = self.socket.accept()
 118                 ct = TinySocketClientThread(clientsocket, self.threads)
 119                 clientsocket = None
 120                 self.threads.append(ct)
 121                 ct.start()
 122                 lt = len(self.threads)
 123                 if (lt > 10) and (lt % 10 == 0):
 124                      # Not many threads should be serving at the same time, so log
 125                      # their abuse.
 126                      netsvc.Logger().notifyChannel("web-services", netsvc.LOG_DEBUG,
 127                         "Netrpc: %d threads" % len(self.threads))
 128             self.socket.close()
 129         except Exception, e:
 130             logging.getLogger('web-services').warning("Netrpc: closing because of exception %s" % str(e))
 131             self.socket.close()
 132             return False
 133
 134     def stop(self):
 135         self.running = False
 136         for t in self.threads:
 137             t.stop()
 138         self._close_socket()
 139
 140     def stats(self):
 141         res = "Net-RPC: " + ( (self.running and "running") or  "stopped")
 142         i = 0
 143         for t in self.threads:
 144             i += 1
 145             res += "\nNet-RPC #%d: %s " % (i, t.name)
 146             if t.isAlive():
 147                 res += "running"
 148             else:
 149                 res += "finished"
 150             if t.sock:
 151                 res += ", socket"
 152         return res
 153
 154 netrpcd = None
 155
 156 def init_servers():
 157     global netrpcd
 158     if tools.config.get('netrpc', False):
 159         netrpcd = TinySocketServerThread(
 160             tools.config.get('netrpc_interface', ''),
 161             int(tools.config.get('netrpc_port', 8070)))