ppserver.py

   1 #!/usr/bin/env python
   2 # Parallel Python Software: http://www.parallelpython.com
   3 # Copyright (c) 2005-2009, Vitalii Vanovschi
   4 # All rights reserved.
   5 # Redistribution and use in source and binary forms, with or without
   6 # modification, are permitted provided that the following conditions are met:
   7 #    * Redistributions of source code must retain the above copyright notice,
   8 #      this list of conditions and the following disclaimer.
   9 #    * Redistributions in binary form must reproduce the above copyright
  10 #      notice, this list of conditions and the following disclaimer in the
  11 #      documentation and/or other materials provided with the distribution.
  12 #    * Neither the name of the author nor the names of its contributors
  13 #      may be used to endorse or promote products derived from this software
  14 #      without specific prior written permission.
  15 #
  16 # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  17 # AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  18 # IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  19 # ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
  20 # LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
  21 # CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
  22 # SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
  23 # INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
  24 # CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
  25 # ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
  26 # THE POSSIBILITY OF SUCH DAMAGE.
  27 """
  28 Parallel Python Software, Network Server
  29
  30 http://www.parallelpython.com - updates, documentation, examples and support
  31 forums
  32 """
  33
  34 import logging
  35 import getopt
  36 import sys
  37 import socket
  38 import thread
  39 import random
  40 import string
  41 import time
  42 import os
  43
  44 import pptransport
  45 import ppauto
  46 from pp import Server
  47
  48
  49 copyright = "Copyright (c) 2005-2009 Vitalii Vanovschi. All rights reserved"
  50 version = "1.5.7"
  51
  52 # compartibility with Python 2.6
  53 try:
  54     import hashlib
  55     sha_new = hashlib.sha1
  56 except ImportError:
  57     import sha
  58     sha_new = sha.new
  59
  60
  61 class _NetworkServer(Server):
  62     """Network Server Class
  63     """
  64
  65     def __init__(self, ncpus="autodetect", interface="0.0.0.0",
  66                 broadcast="255.255.255.255", port=None, secret=None,
  67                 timeout=None, loglevel=logging.WARNING, restart=False,
  68                 proto=0):
  69         Server.__init__(self, ncpus, secret=secret, loglevel=loglevel,
  70                 restart=restart, proto=proto)
  71         self.host = interface
  72         self.bcast = broadcast
  73         if port is not None:
  74             self.port = port
  75         else:
  76             self.port = self.default_port
  77         self.timeout = timeout
  78         self.ncon = 0
  79         self.last_con_time = time.time()
  80         self.ncon_lock = thread.allocate_lock()
  81
  82         logging.debug("Strarting network server interface=%s port=%i"
  83                 % (self.host, self.port))
  84         if self.timeout is not None:
  85             logging.debug("ppserver will exit in %i seconds if no "\
  86                     "connections with clients exist" % (self.timeout))
  87             thread.start_new_thread(self.check_timeout, ())
  88
  89     def ncon_add(self, val):
  90         """Keeps track of the number of connections and time of the last one"""
  91         self.ncon_lock.acquire()
  92         self.ncon += val
  93         self.last_con_time = time.time()
  94         self.ncon_lock.release()
  95
  96     def check_timeout(self):
  97         """Checks if timeout happened and shutdowns server if it did"""
  98         while True:
  99             if self.ncon == 0:
 100                 idle_time = time.time() - self.last_con_time
 101                 if idle_time < self.timeout:
 102                     time.sleep(self.timeout - idle_time)
 103                 else:
 104                     logging.debug("exiting ppserver due to timeout (no client"\
 105                             " connections in last %i sec)", self.timeout)
 106                     os._exit(0)
 107             else:
 108                 time.sleep(self.timeout)
 109
 110     def listen(self):
 111         """Initiates listenting to incoming connections"""
 112         try:
 113             ssocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 114             # following allows ppserver to restart faster on the same port
 115             ssocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 116             ssocket.bind((self.host, self.port))
 117             ssocket.listen(5)
 118         except socket.error:
 119             logging.error("Cannot create socket with port " + str(self.port)
 120                     + " (port is already in use)")
 121
 122         try:
 123             while 1:
 124                 #accept connections from outside
 125                 (csocket, address) = ssocket.accept()
 126                 #now do something with the clientsocket
 127                 #in this case, we'll pretend this is a threaded server
 128                 thread.start_new_thread(self.crun, (csocket, ))
 129         except:
 130             logging.debug("Closing server socket")
 131             ssocket.close()
 132
 133     def crun(self, csocket):
 134         """Authenticates client and handles its jobs"""
 135         mysocket = pptransport.CSocketTransport(csocket)
 136         #send PP version
 137         mysocket.send(version)
 138         #generate a random string
 139         srandom = "".join([random.choice(string.ascii_letters)
 140                 for i in xrange(16)])
 141         mysocket.send(srandom)
 142         answer = sha_new(srandom+self.secret).hexdigest()
 143         cleintanswer = mysocket.receive()
 144         if answer != cleintanswer:
 145             logging.warning("Authentification failed, client host=%s, port=%i"
 146                     % csocket.getpeername())
 147             mysocket.send("FAILED")
 148             csocket.close()
 149             return
 150         else:
 151             mysocket.send("OK")
 152
 153         ctype = mysocket.receive()
 154         logging.debug("Control message received: " + ctype)
 155         self.ncon_add(1)
 156         try:
 157             if ctype == "STAT":
 158                 #reset time at each new connection
 159                 self.get_stats()["local"].time = 0.0
 160                 mysocket.send(str(self.get_ncpus()))
 161                 while 1:
 162                     mysocket.receive()
 163                     mysocket.send(str(self.get_stats()["local"].time))
 164             elif ctype=="EXEC":
 165                 while 1:
 166                     sfunc = mysocket.creceive()
 167                     sargs = mysocket.receive()
 168                     fun = self.insert(sfunc, sargs)
 169                     sresult = fun(True)
 170                     mysocket.send(sresult)
 171         except:
 172             #print sys.excepthook(*sys.exc_info())
 173             logging.debug("Closing client socket")
 174             csocket.close()
 175             self.ncon_add(-1)
 176
 177     def broadcast(self):
 178         """Initiaates auto-discovery mechanism"""
 179         discover = ppauto.Discover(self)
 180         thread.start_new_thread(discover.run,
 181                                 ((self.host, self.port),
 182                                  (self.bcast, self.port)),
 183                                )
 184
 185
 186 def parse_config(file_loc):
 187     """
 188     Parses a config file in a very forgiving way.
 189     """
 190     # If we don't have configobj installed then let the user know and exit
 191     try:
 192         from configobj import ConfigObj
 193     except ImportError, ie:
 194         print >> sys.stderr, "ERROR: You must have configobj installed to use \
 195 configuration files. You can still use command line switches."
 196         sys.exit(1)
 197
 198     if not os.access(file_loc, os.F_OK):
 199         print >> sys.stderr, "ERROR: Can not access %s." % arg
 200         sys.exit(1)
 201
 202     # Load the configuration file
 203     config = ConfigObj(file_loc)
 204     # try each config item and use the result if it exists. If it doesn't
 205     # then simply pass and move along
 206     try:
 207         args['secret'] = config['general'].get('secret')
 208     except:
 209         pass
 210
 211     try:
 212         autodiscovery = config['network'].as_bool('autodiscovery')
 213     except:
 214         pass
 215
 216     try:
 217         args['interface'] = config['network'].get('interface',
 218                                                   default="0.0.0.0")
 219     except:
 220         pass
 221
 222     try:
 223         args['broadcast'] = config['network'].get('broadcast')
 224     except:
 225         pass
 226
 227     try:
 228         args['port'] = config['network'].as_int('port')
 229     except:
 230         pass
 231
 232     try:
 233         args['loglevel'] = config['general'].as_bool('debug')
 234     except:
 235         pass
 236
 237     try:
 238         args['ncpus'] = config['general'].as_int('workers')
 239     except:
 240         pass
 241
 242     try:
 243         args['proto'] = config['general'].as_int('proto')
 244     except:
 245         pass
 246
 247     try:
 248         args['restart'] = config['general'].as_bool('restart')
 249     except:
 250         pass
 251
 252     try:
 253         args['timeout'] = config['network'].as_int('timeout')
 254     except:
 255         pass
 256     # Return a tuple of the args dict and autodiscovery variable
 257     return args, autodiscovery
 258
 259
 260 def print_usage():
 261     """Prints help"""
 262     print "Parallel Python Network Server (pp-" + version + ")"
 263     print "Usage: ppserver.py [-hdar] [-n proto] [-c config_path]"\
 264             " [-i interface] [-b broadcast] [-p port] [-w nworkers]"\
 265             " [-s secret] [-t seconds]"
 266     print
 267     print "Options: "
 268     print "-h                 : this help message"
 269     print "-d                 : debug"
 270     print "-a                 : enable auto-discovery service"
 271     print "-r                 : restart worker process after each"\
 272             " task completion"
 273     print "-n proto           : protocol number for pickle module"
 274     print "-c path            : path to config file"
 275     print "-i interface       : interface to listen"
 276     print "-b broadcast       : broadcast address for auto-discovery service"
 277     print "-p port            : port to listen"
 278     print "-w nworkers        : number of workers to start"
 279     print "-s secret          : secret for authentication"
 280     print "-t seconds         : timeout to exit if no connections with "\
 281             "clients exist"
 282     print
 283     print "Due to the security concerns always use a non-trivial secret key."
 284     print "Secret key set by -s switch will override secret key assigned by"
 285     print "pp_secret variable in .pythonrc.py"
 286     print
 287     print "Please visit http://www.parallelpython.com for extended up-to-date"
 288     print "documentation, examples and support forums"
 289
 290
 291 if __name__ == "__main__":
 292     try:
 293         opts, args = getopt.getopt(sys.argv[1:],
 294             "hdarn:c:b:i:p:w:s:t:", ["help"])
 295     except getopt.GetoptError:
 296         print_usage()
 297         sys.exit(1)
 298
 299     args = {}
 300     autodiscovery = False
 301
 302     for opt, arg in opts:
 303         if opt in ("-h", "--help"):
 304             print_usage()
 305             sys.exit()
 306         elif opt == "-c":
 307             args, autodiscovery = parse_config(arg)
 308         elif opt == "-d":
 309             args["loglevel"] = logging.DEBUG
 310         elif opt == "-i":
 311             args["interface"] = arg
 312         elif opt == "-s":
 313             args["secret"] = arg
 314         elif opt == "-p":
 315             args["port"] = int(arg)
 316         elif opt == "-w":
 317             args["ncpus"] = int(arg)
 318         elif opt == "-a":
 319             autodiscovery = True
 320         elif opt == "-r":
 321             args["restart"] = True
 322         elif opt == "-b":
 323             args["broadcast"] = arg
 324         elif opt == "-n":
 325             args["proto"] = int(arg)
 326         elif opt == "-t":
 327             args["timeout"] = int(arg)
 328
 329     server = _NetworkServer(**args)
 330     if autodiscovery:
 331         server.broadcast()
 332     server.listen()
 333     #have to destroy it here explicitelly otherwise an exception
 334     #comes out in Python 2.4
 335     del server
 336
 337 # Parallel Python Software: http://www.parallelpython.com