mpdlib.py

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

#!/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
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 = 0
# 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

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)
    print printLine
    sys.stdout.flush()
    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' % (mpd_cli_app)
    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' % (mpd_cli_app)
    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))
    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)
    return mpd_trace_returns

def mpd_trace_returns(frame,event,args):
    global mpd_my_id
    if event == 'return':
        print '%s: EXIT %s at line %d ' % (mpd_my_id,frame.f_code.co_name,frame.f_lineno)
        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()
    try:
        rc = sock2.sock.connect(('localhost',port1))
    except:
        # we have seen at least one machine that needs it this way
        rc = sock2.sock.connect(('',port1))
    (sock3,addr) = sock1.sock.accept()
    sock3 = MPDSock(sock=sock3)
    sock1.close()
    return (sock2,sock3)

def mpd_which(execName):
    for d in os.environ['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,0,'May, 2005 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 = 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:
        ips1 = socket.gethostbyname_ex(host1)[2]    # may fail if invalid host
        ips2 = socket.gethostbyname_ex(host2)[2]    # may fail if invalid host
    except:
        return 0
    for ip1 in ips1:
        for ip2 in ips2:
            if ip1 == ip2:
                return 1
    return 0

def mpd_read_nbytes(fd,nbytes):
    global mpd_signum
    rv = 0
    while 1:
        try:
            rv = os.read(fd,nbytes)
            break
        except os.error, errinfo:
            if errinfo[0] == EINTR:
                if mpd_signum == signal.SIGINT  or  mpd_signum == signal.SIGALRM:
                    break
                else:
                    continue
            elif errinfo[0] == ECONNRESET:   # connection reset (treat as eof)
                break
            else:
                mpd_print(1, 'read error: %s' % os.strerror(errinfo[0]))
                break
        except KeyboardInterrupt, errinfo:
            break
        except Exception, errinfo:
            mpd_print(1, 'other error after read %s :%s:' % ( errinfo.__class__, errinfo) )
            break
    return rv

def mpd_get_groups_for_username(username):
    if pwd_module_available  and  grp_module_available:
        userGroups = [pwd.getpwnam(username)[3]]  # default group for the user
        allGroups = grp.getgrall();
        for group in allGroups:
            if username in group[3]  and  group[2] not in userGroups:
                userGroups.append(group[2])
    else:
        userGroups = []
    return userGroups


class MPDSock(object):
    def __init__(self,family=socket.AF_INET,socktype=socket.SOCK_STREAM,proto=0,
                 sock=None,name=''):
        if sock:
            self.sock = sock
        else:
            self.sock = socket.socket(family=family,type=socktype,proto=proto)
        self.name = name
        self.type = socktype
        self.family = family
        ## used this when inherited from socket.socket (only works with py 2.3+)
        ## socket.socket.__init__(self,family=family,type=socktype,proto=proto,_sock=sock)
    def close(self):
        self.sock.close()
    def sendall(self,data):
        self.sock.sendall(data)
    def getsockname(self):
        return self.sock.getsockname()
    def fileno(self):
        return self.sock.fileno()
    def connect(self,*args):
        self.sock.connect(*args)
    def accept(self,name='accepter'):
        global mpd_signum
        newsock = 0
        newaddr = 0
        while 1:
            try:
                mpd_signum = 0
                (newsock,newaddr) = self.sock.accept()
                break
            except socket.error, errinfo:
                if errinfo[0] == EINTR:   # sigchld, sigint, etc.
                    if mpd_signum == signal.SIGINT  or  mpd_signum == signal.SIGALRM:
                        break
                    else:
                        continue
                elif errinfo[0] == ECONNRESET:   # connection reset (treat as eof)
                    break
                else:
                    print '%s: accept error: %s' % (mpd_my_id,os.strerror(errinfo[0]))
                    break