mpd.py

mpi并行计算的c++代码 可用vc或gcc编译通过 可以用来搭建并行计算试验环境

#!/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>]

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
--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

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.141 $"
__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_get_groups_for_username, mpd_uncaught_except_tb, \
                          mpd_set_procedures_to_trace, mpd_trace_calls, \
                          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.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_LOGFILE_TRUNC_SZ' :  4000000,  # -1 -> don't trunc
                               }
        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()
        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'])
        # 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']
            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
        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:
                mpdtid = Thread(target=self.runmainloop)
                mpdtid.start()
                # signals must be handled in main thread; thus we permit timeout of join
                while mpdtid.isAlive():
                    mpdtid.join(2)   # come out sometimes and handle signals
                if self.exiting:
                    break
                if self.conSock:
                    msgToSend = { 'cmd' : 'restarting_mpd' }
                    self.conSock,msgToSend.send_dict_msg(msgToSend)
                    self.streamHandler.del_handler(self.conSock)
                    self.conSock.close()
                    self.conSock = 0
    def runmainloop(self):
        # Main Loop
        while 1:
            rv = self.streamHandler.handle_active_streams(timeout=8.0)
            if rv[0] < 0:
                if type(rv[1]) == ClassType  and  rv[1] == KeyboardInterrupt: # ^C
                    sys.exit(-1)