mpd.py

fortran并行计算包

#!/usr/bin/env python
#
#   (C) 2001 by Argonne National Laboratory.
#       See COPYRIGHT in top-level directory.
#

"""
usage: mpd [--host=<host> --port=<portnum>] [--noconsole]
           [--trace] [--echo] [--daemon] [--bulletproof] --ncpus=<ncpus>
           [--ifhn=<interface-hostname>] [--listenport=<listenport>]
           [--pid=<pidfilename>] [-zc] [--debug]

Some long parameter names may be abbreviated to their first letters by using
  only one hyphen and no equal sign:
     mpd -h donner -p 4268 -n
  is equivalent to
     mpd --host=magpie --port=4268 --noconsole

--host and --port must be specified together; they tell the new mpd where
  to enter an existing ring;  if they are omitted, the new mpd forms a
  stand-alone ring that other mpds may enter later
--noconsole is useful for running 2 mpds on the same machine; only one of
  them will accept mpd commands
--trace yields lots of traces thru mpd routines; currently too verbose
--debug turns on some debugging prints; currently not verbose enough
--echo causes the mpd echo its listener port by which other mpds may connect
--daemon causes mpd to run backgrounded, with no controlling tty
--bulletproof says to turn bulletproofing on (experimental)
--ncpus indicates how many cpus are on the local host; used for starting processes
--ifhn specifies an alternate interface hostname for the host this mpd is running on;
  e.g. may be used to specify the alias for an interface other than default
--listenport specifies a port for this mpd to listen on; by default it will
  acquire one from the system
--conlistenport specifies a port for this mpd to listen on for console
  connections (only used when employing inet socket for console); by default it
  will acquire one from the system
--pid=filename writes the mpd pid into the specified file, or --pid alone
  writes it into /var/run/mpd.pid
-zc is a purely EXPERIMENTAL option right now used to investigate zeroconf
  networking; it can be used to allow mpds to discover each other locally
  using multicast DNS; its usage may change over time
  Currently, -zc is specified like this:  -zc N
  where N specifies a 'level' in a startup set of mpds.  The first mpd in a ring
  must have 1 and it will establish a ring of one mpd.  Subsequent mpds can specify
  -zc 2 and will hook into the ring via the one at level 1.  Except for level 1, new
  mpds enter the ring via an mpd at level-1.

A file named .mpd.conf file must be present in the user's home directory
  with read and write access only for the user, and must contain at least
  a line with MPD_SECRETWORD=<secretword>

To run mpd as root, install it while root and instead of a .mpd.conf file
use mpd.conf (no leading dot) in the /etc directory.' 
"""
from  time    import  ctime
from  mpdlib  import  mpd_version
__author__ = "Ralph Butler and Rusty Lusk"
__date__ = ctime()
__version__ = "$Revision: 1.160 $"
__version__ += "  " + str(mpd_version())
__credits__ = ""


import sys, os, signal, socket, stat

from  re          import  sub
from  atexit      import  register
from  cPickle     import  dumps
from  types       import  ClassType
from  random      import  seed, randrange, random
from  time        import  sleep
from  md5         import  new as md5new
from  mpdlib      import  mpd_set_my_id, mpd_check_python_version, mpd_sockpair, \
                          mpd_print, mpd_get_my_username, mpd_close_zc, \
                          mpd_get_groups_for_username, mpd_uncaught_except_tb, \
                          mpd_set_procedures_to_trace, mpd_trace_calls, \
                          mpd_dbg_level, mpd_set_dbg_level, \
                          MPDSock, MPDListenSock, MPDConListenSock, \
                          MPDStreamHandler, MPDRing, MPDParmDB
from  mpdman      import  MPDMan

try:
    import pwd
    pwd_module_available = 1
except:
    pwd_module_available = 0
try:
    import  syslog
    syslog_module_available = 1
except:
    syslog_module_available = 0
try:
    import  subprocess
    subprocess_module_available = 1
except:
    subprocess_module_available = 0


def sigchld_handler(signum,frame):
    done = 0
    while not done:
        try:
            (pid,status) = os.waitpid(-1,os.WNOHANG)
            if pid == 0:    # no existing child process is finished
                done = 1
        except:    # no more child processes to be waited for
            done = 1
            
class MPD(object):
    def __init__(self):
        self.myHost = socket.gethostname()
        try:
            hostinfo = socket.gethostbyname_ex(self.myHost)
            self.myIfhn = hostinfo[2][0]    # chgd below when I get the real value
        except:
            print 'mpd failed: gethostbyname_ex failed for %s' % (self.myHost)
            sys.exit(-1)
    def run(self):
        if syslog_module_available:
            syslog.openlog("mpd",0,syslog.LOG_DAEMON)
            syslog.syslog(syslog.LOG_INFO,"mpd starting; no mpdid yet")
        sys.excepthook = mpd_uncaught_except_tb
        self.spawnQ = []
        self.spawnInProgress = 0
        self.parmdb = MPDParmDB(orderedSources=['cmdline','xml','env','rcfile','thispgm'])
        self.parmsToOverride = {
                                 'MPD_SECRETWORD'       :  '',
                                 'MPD_MY_IFHN'          :  self.myIfhn,
                                 'MPD_ENTRY_IFHN'       :  '',
                                 'MPD_ENTRY_PORT'       :  0,
                                 'MPD_NCPUS'            :  1,
                                 'MPD_LISTEN_PORT'      :  0,
                                 'MPD_TRACE_FLAG'       :  0,
                                 'MPD_CONSOLE_FLAG'     :  1,
                                 'MPD_ECHO_PORT_FLAG'   :  0,
                                 'MPD_DAEMON_FLAG'      :  0,
                                 'MPD_BULLETPROOF_FLAG' :  0,
                                 'MPD_PID_FILENAME'     :  '',
                                 'MPD_ZC'               :  0,
                                 'MPD_LOGFILE_TRUNC_SZ' :  4000000,  # -1 -> don't trunc
                                 'MPD_PORT_RANGE'       :  0,
                               }
        for (k,v) in self.parmsToOverride.items():
            self.parmdb[('thispgm',k)] = v
        self.get_parms_from_cmdline()
        self.parmdb.get_parms_from_rcfile(self.parmsToOverride,errIfMissingFile=1)
        self.parmdb.get_parms_from_env(self.parmsToOverride)
        self.myIfhn = self.parmdb['MPD_MY_IFHN']    # variable for convenience
        self.myPid = os.getpid()
        if self.parmdb['MPD_PORT_RANGE']:
            os.environ['MPICH_PORT_RANGE'] = self.parmdb['MPD_PORT_RANGE']
        self.listenSock = MPDListenSock(name='ring_listen_sock',
                                        port=self.parmdb['MPD_LISTEN_PORT'])
        self.parmdb[('thispgm','MPD_LISTEN_PORT')] = self.listenSock.sock.getsockname()[1]
        self.myId = '%s_%d' % (self.myHost,self.parmdb['MPD_LISTEN_PORT'])
        mpd_set_my_id(myid=self.myId)
        self.streamHandler = MPDStreamHandler()
        self.ring = MPDRing(streamHandler=self.streamHandler,
                            secretword=self.parmdb['MPD_SECRETWORD'],
                            listenSock=self.listenSock,
                            myIfhn=self.myIfhn,
                            entryIfhn=self.parmdb['MPD_ENTRY_IFHN'],
                            entryPort=self.parmdb['MPD_ENTRY_PORT'],
                            zcMyLevel=self.parmdb['MPD_ZC'])
        # setup tracing if requested via args
        if self.parmdb['MPD_TRACE_FLAG']:
            proceduresToTrace = []
            import inspect
            symbolsAndTypes = globals().items() + \
                              inspect.getmembers(self) + \
                              inspect.getmembers(self.ring) + \
                              inspect.getmembers(self.streamHandler)
            for (symbol,symtype) in symbolsAndTypes:
                if symbol == '__init__':  # a problem to trace
                    continue
                if inspect.isfunction(symtype)  or  inspect.ismethod(symtype):
                    # print symbol
                    proceduresToTrace.append(symbol)
            mpd_set_procedures_to_trace(proceduresToTrace)
            sys.settrace(mpd_trace_calls)
        if syslog_module_available:
            syslog.syslog(syslog.LOG_INFO,"mpd has mpdid=%s (port=%d)" % \
                          (self.myId,self.parmdb['MPD_LISTEN_PORT']) )
        vinfo = mpd_check_python_version()
        if vinfo:
            print "mpd: your python version must be >= 2.2 ; current version is:", vinfo
            sys.exit(-1)
        os.close(0)
        if self.parmdb['MPD_ECHO_PORT_FLAG']:    # do this before becoming a daemon
            # print self.parmdb['MPD_LISTEN_PORT']
            print "mpd_port=%d" % self.parmdb['MPD_LISTEN_PORT']
            sys.stdout.flush()
            ##### NEXT 2 for debugging
            ## print >>sys.stderr, self.parmdb['MPD_LISTEN_PORT']
            ## sys.stderr.flush()
        self.myRealUsername = mpd_get_my_username()
        self.currRingSize = 1    # default
        self.currRingNCPUs = 1   # default
        if os.environ.has_key('MPD_CON_EXT'):
            self.conExt = '_'  + os.environ['MPD_CON_EXT']
        else:
            self.conExt = ''
        self.logFilename = '/tmp/mpd2.logfile_' + mpd_get_my_username() + self.conExt
        if self.parmdb['MPD_PID_FILENAME']:  # may overwrite it below
            pidFile = open(self.parmdb['MPD_PID_FILENAME'],'w')
            print >>pidFile, "%d" % (os.getpid())
            pidFile.close()

        self.conListenSock = 0    # don't want one when I do cleanup for forked daemon procs
        if self.parmdb['MPD_DAEMON_FLAG']:      # see if I should become a daemon with no controlling tty
            rc = os.fork()
            if rc != 0:   # parent exits; child in background
                sys.exit(0)
            os.setsid()  # become session leader; no controlling tty
            signal.signal(signal.SIGHUP,signal.SIG_IGN)  # make sure no sighup when leader ends
            ## leader exits; svr4: make sure do not get another controlling tty
            rc = os.fork()
            if rc != 0:
                sys.exit(0)
            if self.parmdb['MPD_PID_FILENAME']:  # overwrite one above before chg usmask
                pidFile = open(self.parmdb['MPD_PID_FILENAME'],'w')
                print >>pidFile, "%d" % (os.getpid())
                pidFile.close()
            os.chdir("/")  # free up filesys for umount
            os.umask(0)
            try:    os.unlink(self.logFilename)
            except: pass
            logFileFD = os.open(self.logFilename,os.O_CREAT|os.O_WRONLY|os.O_EXCL,0600)
            self.logFile = os.fdopen(logFileFD,'w',0)
            sys.stdout = self.logFile
            sys.stderr = self.logFile
            print >>sys.stdout, 'logfile for mpd with pid %d' % os.getpid()
            sys.stdout.flush()
            os.dup2(self.logFile.fileno(),sys.__stdout__.fileno())
            os.dup2(self.logFile.fileno(),sys.__stderr__.fileno())
        if self.parmdb['MPD_CONSOLE_FLAG']:
            self.conListenSock = MPDConListenSock(secretword=self.parmdb['MPD_SECRETWORD'])
            self.streamHandler.set_handler(self.conListenSock,
                                           self.handle_console_connection)
        register(self.cleanup)
        seed()
        self.nextJobInt    = 1
        self.activeJobs    = {}
        self.conSock       = 0
        self.allExiting    = 0    # for mpdallexit (for first loop for graceful exit)
        self.exiting       = 0    # for mpdexit or mpdallexit
        self.kvs_cntr      = 0    # for mpdman
        self.pulse_cntr    = 0
        rc = self.ring.enter_ring(lhsHandler=self.handle_lhs_input,
                                  rhsHandler=self.handle_rhs_input)
        if rc < 0:
            mpd_print(1,"failed to enter ring")
            sys.exit(-1)
        self.pmi_published_names = {}
        if hasattr(signal,'SIGCHLD'):
            signal.signal(signal.SIGCHLD,sigchld_handler)
        if not self.parmdb['MPD_BULLETPROOF_FLAG']:
            #    import profile ; profile.run('self.runmainloop()')
            self.runmainloop()
        else:
            try:
                from threading import Thread
            except:
                print '*** mpd terminating'
                print '    bulletproof option must be able to import threading-Thread'
                sys.exit(-1)
            # may use SIG_IGN on all but SIGCHLD and SIGHUP (handled above)
            while 1: