mpd.py

fortran并行计算包

                    except:
                        pass
                    self.streamHandler.del_handler(self.conSock)
                    self.conSock.close()
                    self.conSock = 0
                    return
                self.streamHandler.set_handler(self.conSock,self.handle_console_input)
                self.conSock.beingChallenged = 1
                self.conSock.name = 'console'
                randNum = randrange(1,10000)
                randVal = sock.secretword + str(randNum)
                self.conSock.expectedResponse = md5new(randVal).digest()
                self.conSock.send_dict_msg({'cmd' : 'con_challenge', 'randnum' : randNum })
                self.conSock.realUsername = mpd_get_my_username()
            self.streamHandler.set_handler(self.conSock,self.handle_console_input)
            self.conSock.name = 'console'
        else:
            return  ## postpone it; hope the other one frees up soon
    def handle_console_input(self,sock):
        msg = self.conSock.recv_dict_msg()
        if not msg:
            mpd_print(0000, 'console has disappeared; closing it')
            self.streamHandler.del_handler(self.conSock)
            self.conSock.close()
            self.conSock = 0
            return
        if not msg.has_key('cmd'):
            mpd_print(1, 'console sent bad msg :%s:' % msg)
            try:  # try to let console know
                self.conSock.send_dict_msg({ 'cmd':'invalid_msg_received_from_you' })
            except:
                pass
            self.streamHandler.del_handler(self.conSock)
            self.conSock.close()
            self.conSock = 0
            return
        if self.conSock.beingChallenged  and  msg['cmd'] != 'con_challenge_response':
            mpd_print(1, 'console did not respond to con_challenge; msg=:%s:' % msg)
            try:  # try to let console know
                self.conSock.send_dict_msg({ 'cmd':'expected_con_challenge_response' })
            except:
                pass
            self.streamHandler.del_handler(self.conSock)
            self.conSock.close()
            self.conSock = 0
            return
        if msg['cmd'] == 'con_challenge_response':
            self.conSock.beingChallenged = 0
            self.conSock.realUsername = msg['realusername']
            if not msg.has_key('response'):
                try:  # try to let console know
                    self.conSock.send_dict_msg({ 'cmd':'missing_response_in_msg' })
                except:
                    pass
                self.streamHandler.del_handler(self.conSock)
                self.conSock.close()
                self.conSock = 0
                return
            elif msg['response'] != self.conSock.expectedResponse:
                try:  # try to let console know
                    self.conSock.send_dict_msg({ 'cmd':'invalid_response' })
                except:
                    pass
                self.streamHandler.del_handler(self.conSock)
                self.conSock.close()
                self.conSock = 0
                return
            self.conSock.send_dict_msg({ 'cmd':'valid_response' })
        elif msg['cmd'] == 'mpdrun':
            # permit anyone to run but use THEIR own username
            #   thus, override any username specified by the user
            if self.conSock.realUsername != 'root':
                msg['username'] = self.conSock.realUsername
                msg['users'] = { (0,msg['nprocs']-1) : self.conSock.realUsername }
            #
            msg['mpdid_mpdrun_start'] = self.myId
            msg['nstarted_on_this_loop'] = 0
            msg['first_loop'] = 1
            msg['ringsize'] = 0
            msg['ring_ncpus'] = 0
            if msg.has_key('try_1st_locally'):
                self.do_mpdrun(msg)
            else:
                self.ring.rhsSock.send_dict_msg(msg)
            # send ack after job is going
        elif msg['cmd'] == 'get_mpdrun_values':
            msgToSend = { 'cmd' : 'response_get_mpdrun_values',
	                  'mpd_version' : mpd_version(),
	                  'mpd_ifhn' : self.myIfhn }
            self.conSock.send_dict_msg(msgToSend)
        elif msg['cmd'] == 'mpdtrace':
            msgToSend = { 'cmd'     : 'mpdtrace_info',
                          'dest'    : self.myId,
                          'id'      : self.myId,
                          'ifhn'    : self.myIfhn,
                          'lhsport' : '%s' % (self.ring.lhsPort),
                          'lhsifhn' : '%s' % (self.ring.lhsIfhn),
                          'rhsport' : '%s' % (self.ring.rhsPort),
                          'rhsifhn' : '%s' % (self.ring.rhsIfhn) }
            self.ring.rhsSock.send_dict_msg(msgToSend)
            msgToSend = { 'cmd'  : 'mpdtrace_trailer', 'dest' : self.myId }
            self.ring.rhsSock.send_dict_msg(msgToSend)
            # do not send an ack to console now; will send trace info later
        elif msg['cmd'] == 'mpdallexit':
            if self.conSock.realUsername != self.myRealUsername:
                msgToSend = { 'cmd':'invalid_username_to_make_this_request' }
                self.conSock.send_dict_msg(msgToSend)
                self.streamHandler.del_handler(self.conSock)
                self.conSock.close()
                self.conSock = 0
                return
            # self.allExiting = 1  # doesn't really help here
            self.ring.rhsSock.send_dict_msg( {'cmd' : 'mpdallexit', 'src' : self.myId} )
            self.conSock.send_dict_msg( {'cmd' : 'mpdallexit_ack'} )
        elif msg['cmd'] == 'mpdexit':
            if self.conSock.realUsername != self.myRealUsername:
                msgToSend = { 'cmd':'invalid_username_to_make_this_request' }
                self.conSock.send_dict_msg(msgToSend)
                self.streamHandler.del_handler(self.conSock)
                self.conSock.close()
                self.conSock = 0
                return
            if msg['mpdid'] == 'localmpd':
                msg['mpdid'] = self.myId
            self.ring.rhsSock.send_dict_msg( {'cmd' : 'mpdexit', 'src' : self.myId,
                                              'done' : 0, 'dest' : msg['mpdid']} )
        elif msg['cmd'] == 'mpdringtest':
            msg['src'] = self.myId
            self.ring.rhsSock.send_dict_msg(msg)
            # do not send an ack to console now; will send ringtest info later
        elif msg['cmd'] == 'mpdlistjobs':
            msgToSend = { 'cmd'  : 'local_mpdid', 'id' : self.myId }
            self.conSock.send_dict_msg(msgToSend)
            for jobid in self.activeJobs.keys():
                for manPid in self.activeJobs[jobid]:
                    msgToSend = { 'cmd' : 'mpdlistjobs_info',
                                  'dest' : self.myId,
                                  'jobid' : jobid,
                                  'username' : self.activeJobs[jobid][manPid]['username'],
                                  'host' : self.myHost,
                                  'ifhn' : self.myIfhn,
                                  'clipid' : str(self.activeJobs[jobid][manPid]['clipid']),
                                  'sid' : str(manPid),  # may chg to actual sid later
                                  'pgm'  : self.activeJobs[jobid][manPid]['pgm'],
                                  'rank' : self.activeJobs[jobid][manPid]['rank'] }
                    self.conSock.send_dict_msg(msgToSend)
            msgToSend = { 'cmd'  : 'mpdlistjobs_trailer', 'dest' : self.myId }
            self.ring.rhsSock.send_dict_msg(msgToSend)
            # do not send an ack to console now; will send listjobs info later
        elif msg['cmd'] == 'mpdkilljob':
            # permit anyone to kill but use THEIR own username
            #   thus, override any username specified by the user
            if self.conSock.realUsername != 'root':
                msg['username'] = self.conSock.realUsername
            msg['src'] = self.myId
            msg['handled'] = 0
            if msg['mpdid'] == '':
                msg['mpdid'] = self.myId
            self.ring.rhsSock.send_dict_msg(msg)
            # send ack to console after I get this msg back and do the kill myself
        elif msg['cmd'] == 'mpdsigjob':
            # permit anyone to sig but use THEIR own username
            #   thus, override any username specified by the user
            if self.conSock.realUsername != 'root':
                msg['username'] = self.conSock.realUsername
            msg['src'] = self.myId
            msg['handled'] = 0
            if msg['mpdid'] == '':
                msg['mpdid'] = self.myId
            self.ring.rhsSock.send_dict_msg(msg)
            # send ack to console after I get this msg back
        elif msg['cmd'] == 'verify_hosts_in_ring':
            msgToSend = { 'cmd'  : 'verify_hosts_in_ring', 'dest' : self.myId,
                          'host_list' : msg['host_list'] }
            self.ring.rhsSock.send_dict_msg(msgToSend)
            # do not send an ack to console now; will send trace info later
        else:
            msgToSend = { 'cmd' : 'invalid_msg_received_from_you' }
            self.conSock.send_dict_msg(msgToSend)
            badMsg = 'invalid msg received from console: %s' % (str(msg))
            mpd_print(1, badMsg)
            if syslog_module_available:
                syslog.syslog(syslog.LOG_ERR,badMsg)
    def handle_man_input(self,sock):
        msg = sock.recv_dict_msg()
        if not msg:
            for jobid in self.activeJobs.keys():
                deleted = 0
                for manPid in self.activeJobs[jobid]:
                    if sock == self.activeJobs[jobid][manPid]['socktoman']:
			mpd_print(mpd_dbg_level,\
                                  "Deleting %s %d" % (str(jobid),manPid))
                        del self.activeJobs[jobid][manPid]
                        if len(self.activeJobs[jobid]) == 0:
                            del self.activeJobs[jobid]
                        deleted = 1
                        break
                if deleted:
                    break
            self.streamHandler.del_handler(sock)
            sock.close()
            return
        if not msg.has_key('cmd'):
            mpd_print(1, 'INVALID msg for man request msg=:%s:' % (msg) )
            msgToSend = { 'cmd' : 'invalid_msg' }
            sock.send_dict_msg(msgToSend)
            self.streamHandler.del_handler(sock)
            sock.close()
            return
	# Who asks, and why?  
        # We have a failure that deletes the spawnerManPid from the
	# activeJobs[jobid] variable.   The temporary work-around is
        # to ignore this request if the target process is no longer 
	# in the activeJobs table.
        if msg['cmd'] == 'client_info':
            jobid = msg['jobid']
            manPid = msg['manpid']
            self.activeJobs[jobid][manPid]['clipid'] = msg['clipid']
            if msg['spawner_manpid']  and  msg['rank'] == 0:
                if msg['spawner_mpd'] == self.myId:
                    spawnerManPid = msg['spawner_manpid']
		    mpd_print(mpd_dbg_level,\
                       "About to check %s:%s" % (str(jobid),str(spawnerManPid)))

                    if not self.activeJobs[jobid].has_key(spawnerManPid):
                        mpd_print(0,"Missing %d in %s" % (spawnerManPid,str(jobid)))
                    elif not self.activeJobs[jobid][spawnerManPid].has_key('socktoman'):
                        mpd_print(0,"Missing socktoman!")
                    else:
                        spawnerManSock = self.activeJobs[jobid][spawnerManPid]['socktoman']
                        msgToSend = { 'cmd' : 'spawn_done_by_mpd', 'rc' : 0, 'reason' : '' }
                        spawnerManSock.send_dict_msg(msgToSend)
                else:
                    self.ring.rhsSock.send_dict_msg(msg)
        elif msg['cmd'] == 'spawn':
            msg['mpdid_mpdrun_start'] = self.myId
            msg['spawner_mpd'] = self.myId
            msg['nstarted_on_this_loop'] = 0
            msg['first_loop'] = 1
            msg['jobalias'] = ''
            msg['stdin_dest'] = '0'
            msg['ringsize'] = 0
            msg['ring_ncpus'] = 0
            msg['gdb'] = 0
            msg['gdba'] = ''
            msg['totalview'] = 0
            msg['ifhns'] = {}
            self.spawnQ.append(msg)
        elif msg['cmd'] == 'publish_name':
            self.pmi_published_names[msg['service']] = msg['port']
            msgToSend = { 'cmd' : 'publish_result', 'info' : 'ok' }
            sock.send_dict_msg(msgToSend)
        elif msg['cmd'] == 'lookup_name':
            if self.pmi_published_names.has_key(msg['service']):
                msgToSend = { 'cmd' : 'lookup_result', 'info' : 'ok',
                              'port' : self.pmi_published_names[msg['service']] }
                sock.send_dict_msg(msgToSend)
            else:
                msg['cmd'] = 'pmi_lookup_name'    # add pmi_
                msg['src'] = self.myId
                msg['port'] = 0    # invalid
                self.ring.rhsSock.send_dict_msg(msg)
        elif msg['cmd'] == 'unpublish_name':
            if self.pmi_published_names.has_key(msg['service']):
                del self.pmi_published_names[msg['service']]
                msgToSend = { 'cmd' : 'unpublish_result', 'info' : 'ok' }
                sock.send_dict_msg(msgToSend)
            else: