mpdlib.py

fortran并行计算包

                else:    # assume python2.3 or later
                    oldTimeout = socket.getdefaulttimeout()
                    socket.setdefaulttimeout(8)
                try:
                    self.sock.connect((conIfhn,conPort))
                except Exception, errmsg:
                    mpd_print(1,"failed to connect to host %s port %d" % \
                              (conIfhn,conPort) )
                    self.sock.close()
                    self.sock = 0
                if hasattr(signal,'alarm'):
                    signal.alarm(oldAlarmTime)
                else:    # assume python2.3 or later
                    socket.setdefaulttimeout(oldTimeout)
                if not self.sock:
                    print '%s: cannot connect to local mpd (%s); possible causes:' % \
                          (mpd_my_id,self.conFilename)
                    print '  1. no mpd is running on this host'
                    print '  2. an mpd is running but was started without a "console" (-n option)'
                    print 'In case 1, you can start an mpd on this host with:'
                    print '    mpd &'
                    print 'and you will be able to run jobs just on this host.'
                    print 'For more details on starting mpds on a set of hosts, see'
                    print 'the MPICH2 Installation Guide.'
                    sys.exit(-1)
                msgToSend = { 'cmd' : 'con_init' }
                self.sock.send_dict_msg(msgToSend)
                msg = self.sock.recv_dict_msg()
                if not msg:
                    mpd_print(1,'expected con_challenge from mpd; got eof')
                    sys.exit(-1)
                if msg['cmd'] != 'con_challenge':
                    mpd_print(1,'expected con_challenge from mpd; got msg=:%s:' % (msg) )
                    sys.exit(-1)
                randVal = self.secretword + str(msg['randnum'])
                response = md5new(randVal).digest()
                msgToSend = { 'cmd' : 'con_challenge_response', 'response' : response,
                              'realusername' : mpd_get_my_username() }
                self.sock.send_dict_msg(msgToSend)
                msg = self.sock.recv_dict_msg()
                if not msg  or  msg['cmd'] != 'valid_response':
                    mpd_print(1,'expected valid_response from mpd; got msg=:%s:' % (msg) )
                    sys.exit(-1)
        if not self.sock:
            print '%s: cannot connect to local mpd (%s); possible causes:' % \
                  (mpd_my_id,self.conFilename)
            print '  1. no mpd is running on this host'
            print '  2. an mpd is running but was started without a "console" (-n option)'
            print 'In case 1, you can start an mpd on this host with:'
            print '    mpd &'
            print 'and you will be able to run jobs just on this host.'
            print 'For more details on starting mpds on a set of hosts, see'
            print 'the MPICH2 Installation Guide.'
            sys.exit(-1)

class MPDParmDB(dict):
    def __init__(self,orderedSources=[]):
        dict.__init__(self)
        self.orderedSources = orderedSources
        self.db = {}
        for src in orderedSources:  # highest to lowest
            self.db[src] = {}
    def __setitem__(self,sk_tup,val):
        if type(sk_tup) != TupleType  or  len(sk_tup) != 2:
            mpd_print_tb(1,"must use a 2-tuple as key in a parm db; invalid: %s" % (sk_tup) )
            sys.exit(-1)
        s,k = sk_tup
        for src in self.orderedSources:
            if src == s:
                self.db[src][k] = val
                break
        else:
            mpd_print_tb(1,"invalid src specified for insert into parm db; src=%s" % (src) )
            sys.exit(-1)
    def __getitem__(self,key):
        for src in self.orderedSources:
            if self.db[src].has_key(key):
                return self.db[src][key]
        raise KeyError, "key %s not found in parm db" % (key)
    def has_key(self,key):
        for src in self.orderedSources:
            if self.db[src].has_key(key):
                return 1
        return 0
    def printall(self):
        print "MPDRUN's PARMDB; values from all sources:"
        for src in self.orderedSources:
            print '  %s (source)' % (src)
            for key in self.db[src].keys():
                print '    %s = %s' % (key,self.db[src][key])
    def printdef(self):
        print "MPDRUN's PARMDB; default values only:"
        printed = {}
        for src in self.orderedSources:
            for key in self.db[src]:
                if not printed.has_key(key):
                    printed[key] = 1
                    print '  %s  %s = %s' % (src,key,self.db[src][key])
    def get_parms_from_env(self,parmsToOverride):
        for k in parmsToOverride.keys():
            if os.environ.has_key(k):
                self[('env',k)] = os.environ[k]
    def get_parms_from_rcfile(self,parmsToOverride,errIfMissingFile=0):
        if os.environ.has_key('MPD_CONF_FILE'):
            parmsRCFilename = os.environ['MPD_CONF_FILE']
        elif hasattr(os,'getuid')  and  os.getuid() == 0:    # if ROOT
            parmsRCFilename = os.path.abspath('/etc/mpd.conf')
        elif os.environ.has_key('HOME'):
            parmsRCFilename = os.path.join(os.environ['HOME'],'.mpd.conf')
        elif os.environ.has_key('HOMEPATH'):    # e.g. win32
            parmsRCFilename = os.path.join(os.environ['HOMEPATH'],'.mpd.conf')
        else:
            print 'unable to find mpd.conf file'
            sys.exit(-1)
        if sys.platform == 'win32':
            mode = 0x80   # fake it
        else:
            try:
                mode = os.stat(parmsRCFilename)[0]
            except:
                mode = ''
	# sometimes a missing file is OK, e.g. when user running with root's mpd
        if not mode  and  not errIfMissingFile:
            return
        if not mode:
            print 'configuration file %s not found' % (parmsRCFilename)
            print 'A file named .mpd.conf file must be present in the user\'s home'
            print 'directory (/etc/mpd.conf if root) with read and write access'
            print 'only for the user, and must contain at least a line with:'
            print 'MPD_SECRETWORD=<secretword>'
            print 'One way to safely create this file is to do the following:'
            print '  cd $HOME'
            print '  touch .mpd.conf'
            print '  chmod 600 .mpd.conf'
            print 'and then use an editor to insert a line like'
            print '  MPD_SECRETWORD=mr45-j9z'
            print 'into the file.  (Of course use some other secret word than mr45-j9z.)' 
            sys.exit(-1)
        if  (mode & 0x3f):
            print 'configuration file %s is accessible by others' % (parmsRCFilename)
            print 'change permissions to allow read and write access only by you'
            sys.exit(-1)
        parmsRCFile = open(parmsRCFilename)
        for line in parmsRCFile:
            lineWithoutComments = line.split('#')[0]    # will at least be ''
            lineWithoutComments = lineWithoutComments.strip()
            if not lineWithoutComments:
                continue
            splitLine = lineWithoutComments.split('=')
            if not splitLine[0]:    # ['']
                print 'warning: unrecognized (null) key in %s' % (parmsRCFilename)
                continue
            if len(splitLine) == 2:
                (k,v) = splitLine
                origKey = k
                if k == 'secretword':    # for bkwd-compat
                    k = 'MPD_SECRETWORD'
                if k in parmsToOverride.keys():
                    if k != 'MPD_SECRETWORD'  and  v.isdigit():
                        v = int(v)
                    self[('rcfile',k)] = v
            else:
                mpd_print(1, 'line in mpd conf is not key=val pair; line=:%s:' % (line) )

class MPDTest(object):
    def __init__(self):
        pass
    def run(self,cmd='',expIn = '',chkEC=0,expEC=0,chkOut=0,expOut='',ordOut=0,
            grepOut=0, exitOnFail=1):
        rv = {}
        if chkOut and grepOut:
            print "grepOut and chkOut are mutually exclusive"
            sys.exit(-1)
        outLines = []
        if subprocess_module_available:
            import re
            cmd = re.split(r'\s+',cmd)
            runner = subprocess.Popen(cmd,bufsize=0,env=os.environ,close_fds=True,
                                      stdin=subprocess.PIPE,stdout=subprocess.PIPE,
                                      stderr=subprocess.PIPE)
            if expIn:
                runner.stdin.write(expIn)
            runner.stdin.close()
            for line in runner.stdout:
                outLines.append(line[:-1])    # strip newlines
            for line in runner.stderr:
                outLines.append(line[:-1])    # strip newlines
            rv['pid'] = runner.pid
            rv['EC'] = runner.wait()
        elif hasattr(popen2,'Popen4'):    # delete when python2.4+ is common
            runner = popen2.Popen4(cmd)
            if expIn:
                runner.tochild.write(expIn)
            runner.tochild.close()
            for line in runner.fromchild:
                outLines.append(line[:-1])    # strip newlines
            rv['pid'] = runner.pid
            rv['EC'] = runner.wait()
        else:
            mpd_print(1,'can not run with either subprocess or popen2-Popen4')
            sys.exit(-1)
        rv['OUT'] = outLines[:]
        if chkEC  and  expEC != rv['EC']:
            print "bad exit code from test: %s" % (cmd)
            print "   expected exitcode=%d ; got %d" % (expEC,rv['EC'])
            print "output from cmd:"
            for line in outLines:
                print line
            if exitOnFail:
                sys.exit(-1)
        if chkOut:
            orderOK = 1
            expOut = expOut.split('\n')[:-1]  # leave off trailing ''
            for line in outLines[:]:    # copy of outLines
                if line in expOut:
                    if ordOut and line != expOut[0]:
                        orderOK = 0
                        break  # count rest of outLines as bad
                    expOut.remove(line)
                    outLines.remove(line)
            if not orderOK:
                print "lines out of order in output for test: %s" % (cmd)
                for line in outLines:
                    print line
                if exitOnFail:
                    sys.exit(-1)
            if expOut:
                print "some required lines not found in output for test: %s" % (cmd)
                for line in outLines:
                    print line
                if exitOnFail:
                    sys.exit(-1)
            if outLines:
                print "extra lines in output for test: %s" % (cmd)
                for line in outLines:
                    print line
                if exitOnFail:
                    sys.exit(-1)
        elif grepOut:
            foundCnt = 0
            for expLine in expOut:
                for outLine in outLines:
                    if outLine.find(expLine) >= 0:
                        foundCnt += 1
            if foundCnt < len(expOut):
                print "some lines not matched for test: %s" % (cmd)
                for line in outLines:
                     print line
                if exitOnFail:
                    sys.exit(-1)
        return rv

#### experimental code for zeroconf
def mpd_init_zc(ifhn,my_level):
    import threading, Zeroconf
    global mpd_zc
    mpd_zc = Zeroconf.Zeroconf()
    class ListenerForPeers(object):
        def __init__(self):
            mpd_zc.peers = {}
            mpd_zc.peersLock = threading.Lock()
            mpd_zc.peers_available_event = threading.Event()
        def removeService(self, zc, service_type, name):
            mpd_zc.peersLock.acquire()
            del mpd_zc.peers[name]
            print "removed", name ; sys.stdout.flush()
            mpd_zc.peersLock.release()
        def addService(self, zc, service_type, name):
            info = zc.getServiceInfo(service_type, name)
            if info:
                if info.properties['username'] != mpd_get_my_username():
                    return
                mpd_zc.peersLock.acquire()
                mpd_zc.peers[name] = info
                print "added peer:", name, info.properties ; sys.stdout.flush()
                mpd_zc.peersLock.release()
                mpd_zc.peers_available_event.set()
            else:
                print "OOPS NO INFO FOR", name ; sys.stdout.flush()
    service_type = "_mpdzc._tcp.local."
    listenerForPeers = ListenerForPeers()
    browser = Zeroconf.ServiceBrowser(mpd_zc,service_type,listenerForPeers)
    ##  sleep(1.5)  # give browser a chance to find some peers
def mpd_find_zc_peer(peer_level):
    print "finding a peer at level %d..." % (peer_level) ; sys.stdout.flush()
    mpd_zc.peers_available_event.wait(5)
    for (peername,info) in mpd_zc.peers.items():
        if info.properties['mpdid'] == mpd_my_id:
            continue
        if info.properties['level'] != peer_level:
            continue
        peerAddr = str(socket.inet_ntoa(info.getAddress()))
        peerPort = info.getPort()
        return(peerAddr,peerPort)
    return ('',0)
def mpd_register_zc(ifhn,level):
    import Zeroconf
    service_type = "_mpdzc._tcp.local."
    service_ifhn = socket.inet_aton(ifhn)
    service_host = socket.gethostname()
    service_port = int(mpd_my_id.split('_')[1])
    svc = Zeroconf.ServiceInfo(service_type,
                               mpd_my_id + service_type,
                               address = service_ifhn,
                               port = service_port,
                               weight = 0, priority = 0,
                               properties = { 'description': 'mpd',
                                              'mpdid' : mpd_my_id,
                                              'level' : level,