mpdlib.py

fortran并行计算包

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


import sys, os, signal, popen2, socket, select, inspect

from  cPickle   import  dumps, loads
from  types     import  TupleType
from  traceback import  extract_tb, extract_stack, format_list
from  re        import  sub, split
from  errno     import  EINTR, ECONNRESET, EISCONN, ECONNREFUSED, EPIPE
from  md5       import  new as md5new
from  time      import  sleep
from  random    import  randrange, random

try:
    import pwd
    pwd_module_available = 1
except:
    pwd_module_available = 0
try:
    import grp
    grp_module_available = 1
except:
    grp_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


# some global vars for some utilities
global mpd_my_id, mpd_signum, mpd_my_hostname, mpd_procedures_to_trace
global mpd_cli_app  # for debug during mpich nightly tests
mpd_cli_app = ''
mpd_my_id = ''
mpd_procedures_to_trace = []
mpd_my_hostname = ''
# mpd_signum can be set by mpd_handle_signal to indicate which signal was recently caught;
# this can be useful below to pop out of loops that ordinarily continue after sigs
# NOTE: mpd_handle_signal must be called by the user, e.g. in his own signal handler
mpd_signum = 0
mpd_zc = 0

# For easier debugging, we provide this variable that is used in the
# mpd_print calls.  This makes it a little easier to debug problems involving
# communication with other processes, such as handling EINTR from signals.
global mpd_dbg_level
mpd_dbg_level = 0

def mpd_set_dbg_level(flag):
    global mpd_dbg_level
    mpd_dbg_level = flag

def mpd_set_my_id(myid=''):
    global mpd_my_id
    mpd_my_id = myid

def mpd_get_my_id():
    global mpd_my_id
    return(mpd_my_id)

def mpd_set_cli_app(app):    # for debug during mpich nightly tests
    global mpd_cli_app
    mpd_cli_app = app

def mpd_handle_signal(signum,frame):
    global mpd_signum
    mpd_signum = signum

def mpd_print(*args):
    global mpd_my_id
    if not args[0]:
        return
    stack = extract_stack()
    callingProc = stack[-2][2]
    callingLine = stack[-2][1]
    printLine = '%s (%s %d): ' % (mpd_my_id,callingProc,callingLine)
    for arg in args[1:]:
        printLine = printLine + str(arg)
    # We've seen an EINTR on the flush here
    while 1:
        try:
            print printLine
            break
        except os.error, errinfo:
            if errinfo[0] != EINTR:
                raise os.error, errinfo
    # end of while
    while 1:
        try:
            sys.stdout.flush()
            break
        except os.error, errinfo:
	    if errinfo[0] != EINTR:
                raise os.error, errinfo
    # end of while
    if syslog_module_available:
        syslog.syslog(syslog.LOG_INFO,printLine)

def mpd_print_tb(*args):
    global mpd_my_id
    if not args[0]:
        return
    stack = extract_stack()
    callingProc = stack[-2][2]
    callingLine = stack[-2][1]
    stack = extract_stack()
    stack.reverse()
    stack = stack[1:]
    printLine = '%s (%s %d):' % (mpd_my_id,callingProc,callingLine)
    for arg in args[1:]:
        printLine = printLine + str(arg)
    printLine += '\n  mpdtb:\n'
    for line in format_list(stack):
        line = sub(r'\n.*','',line)
        splitLine = split(',',line)
        splitLine[0] = sub('  File "(.*)"',lambda mo: mo.group(1),splitLine[0])
        splitLine[1] = sub(' line ','',splitLine[1])
        splitLine[2] = sub(' in ','',splitLine[2])
        printLine = printLine + '    %s,  %s,  %s\n' % tuple(splitLine)
    if mpd_cli_app:    # debug mpich apps in nightly tests
        printLine += '    mpd_cli_app=%s\n' % (mpd_cli_app)
        printLine += '    cwd=%s' % (os.getcwd())
    print printLine
    sys.stdout.flush()
    if syslog_module_available:
        syslog.syslog(syslog.LOG_INFO,printLine)

def mpd_uncaught_except_tb(arg1,arg2,arg3):
    global mpd_my_id
    global mpd_cli_id
    if mpd_my_id:
        errstr = '%s: ' % (mpd_my_id)
    else:
        errstr = ''
    errstr += 'mpd_uncaught_except_tb handling:\n'
    errstr += '  %s: %s\n' % (arg1,arg2)
    tb = extract_tb(arg3)
    tb.reverse()
    for tup in tb:
        # errstr += '    file %s  line# %i  procedure %s\n        %s\n' % (tup)
        errstr += '    %s  %i  %s\n        %s\n' % (tup)
    if mpd_cli_app:    # debug mpich apps in nightly tests
        errstr += '    mpd_cli_app=%s\n' % (mpd_cli_app)
        errstr += '    cwd=%s' % (os.getcwd())
    print errstr,
    if syslog_module_available:
        syslog.syslog(syslog.LOG_ERR, errstr)

def mpd_set_procedures_to_trace(procs):
    global mpd_procedures_to_trace
    mpd_procedures_to_trace = procs

def mpd_trace_calls(frame,event,args):
    global mpd_my_id, mpd_procedures_to_trace
    if frame.f_code.co_name not in mpd_procedures_to_trace:
        return None
    args_info = apply(inspect.formatargvalues,inspect.getargvalues(frame))
    # Be VERY careful here; under AIX, it looked like EINTR is 
    # possible within print (!).  
    while (1):
        try:
            print '%s: ENTER %s in %s at line %d; ARGS=%s' % \
          (mpd_my_id,frame.f_code.co_name,frame.f_code.co_filename,frame.f_lineno,args_info)
            break
        except os.error, errinfo:
            if errinfo[0] != EINTR:
                raise os.error, errinfo
    # end of while
    return mpd_trace_returns

def mpd_trace_returns(frame,event,args):
    global mpd_my_id
    if event == 'return':
        # Be VERY careful here; under AIX, it looked like EINTR is 
        # possible within print (!).  
        while (1):
            try:
                print '%s: EXIT %s at line %d ' % (mpd_my_id,frame.f_code.co_name,frame.f_lineno)
                break
            except os.error, errinfo:
                if errinfo[0] != EINTR:
                    raise os.error, errinfo
        # end of while
        return None
    else:
        return mpd_trace_returns

def mpd_sockpair():
    sock1 = MPDSock()
    rc = sock1.sock.bind(('',0))
    rc = sock1.sock.listen(5)
    port1 = sock1.sock.getsockname()[1]
    sock2 = MPDSock()
    #
    # We have encountered situations where the connection fails; as this is
    # a connection to this process, we retry a few times in that case 
    # (seen on AIX)
    #
    try:
        connAttempts = 0
        while (1):
            try:
                rc = sock2.sock.connect(('localhost',port1))
                break
            except socket.error, errinfo:
                # In some cases, connect will return EINTR and then on the
                # next iteration, returns EISCONN.
                if errinfo[0] == EISCONN:
                    break
                if errinfo[0] == ECONNREFUSED and connAttempts < 10:
                    mpd_print(mpd_dbg_level,"Retrying on connection refused")
                    connAttempts += 1
                    sleep(random())
                elif errinfo[0] != EINTR:
                    mpd_print(1,"connect %d %s" % (errinfo[0],errinfo[1]))
                    raise socket.error, errinfo
	# End of the while
    except socket.error, errinfo:
        # we have seen at least one machine that needs it this way
        # We've seen a failure here; it could be EINPROGRESS, EALREADY, 
        # or EADDRINUSE.  In that case, we may need to do something else
	mpd_print(1,"connect error with %d %s" % (errinfo[0],errinfo[1]))
        # Should this only attempt on ECONNREFUSED, ENETUNREACH, EADDRNOTAVAIL
        # FIXME: Does this need a try/except?
        while 1:
            try:  
                rc = sock2.sock.connect(('',port1))
                break
            except socket.error, errinfo:
                if errinfo[0] == EISCONN:
                    break
                elif errinfo[0] != EINTR:
                    mpd_print(1,"connect %d %s" % (errinfo[0],errinfo[1]))
                    raise socket.error, errinfo
        # end of while
    # Accept can fail on EINTR, so we handle that here
    while (1):
        try:
            (sock3,addr) = sock1.sock.accept()
            break
        except socket.error, errinfo:
            if errinfo[0] != EINTR:
                mpd_print(1,"connect %d %s" % (errinfo[0],errinfo[1]))
                raise socket.error, errinfo
    # end of while
    sock3 = MPDSock(sock=sock3)
    sock1.close()
    return (sock2,sock3)

def mpd_which(execName,user_path=None):
    if not user_path:
        if os.environ.has_key('PATH'):
            user_path = os.environ['PATH']
        else:
            return ''
    for d in user_path.split(os.pathsep):
        fpn = os.path.join(d,execName)
        if os.path.isdir(fpn):  # follows symlinks; dirs can have execute permission
            continue
        if os.access(fpn,os.X_OK):    # NOTE access works based on real uid (not euid)
            return fpn
    return ''

def mpd_check_python_version():
    # version_info: (major,minor,micro,releaselevel,serial)
    if (sys.version_info[0] < 2)  or  \
       (sys.version_info[0] == 2 and sys.version_info[1] < 2):
        return sys.version_info
    return 0

def mpd_version():
    return (1,0,1,'July, 2006 release')  # major, minor, micro, special

def mpd_get_my_username():
    if pwd_module_available:
        username = pwd.getpwuid(os.getuid())[0]    # favor this over env
    elif os.environ.has_key('USER'):
        username = os.environ['USER']
    elif os.environ.has_key('USERNAME'):
        username = os.environ['USERNAME']
    else:
        username = 'unknown_username'
    return username

def mpd_get_ranks_in_binary_tree(myRank,nprocs):
    if myRank == 0:
        parent = -1;
    else:   
        parent = (myRank - 1) / 2; 
    lchild = (myRank * 2) + 1
    if lchild > (nprocs - 1):
        lchild = -1;
    rchild = (myRank * 2) + 2
    if rchild > (nprocs - 1):
        rchild = -1;
    return (parent,lchild,rchild)

def mpd_same_ips(host1,host2):    # hosts may be names or IPs
    try: