Python Forum

I have 2functions

runPA(), it will call a product function to run 10seconds and generate a temp file in /tmp then automatically delete the temp file after 10seconds

getCapFileName(), it searches /tmp to find temp file name which generated by runPA(),write its name to a file

How can i launch above 2 functions at the same time in python?

tried using threading when running the code in linux it goes to defunct state eventually

it's hard to debug it in linux it's RHEL7.5,python3.7
appreciate any ideas, thanks!

def test_UNV28366(self):
    import re
    import threading
    import pdb
    import time

    pid=os.getpid()
    event =threading.Event()                
    forBreak=False
    self.uvorud()

    def getCapFileName():
        gPid=os.getpid() 
        sleep(1)
        fcFile=""
        U2TEMP="C:\\U2\\UV\\XDEMO"
        for root, dirs, files in os.walk(u2py.U2TEMP):
            for name in files:
                if re.match(r'u2pyruncapture\-\d+-\d+$', name):
                    sname=name.split("-")
                    #pdb.set_trace()
                    if int(sname[1]) == pid:   
                        #pdb.set_trace()                 
                        fcFile=name
                        print(fcFile)
                        #'''
            break

        return fcFile

    def runPA():
        rPid=os.getpid()
        #sleep(0.8)
        #if sysStr == 'Linux':        
        if TESTDB == 'uv':
            fVoc = u2py.File("VOC")                         
            dMpa = u2py.DynArray("")
            dMpa.insert(1, "PA")
            dMpa.insert(2, "DISPLAY before sleep")
            dMpa.insert(3, "SLEEP 10")
            dMpa.insert(4, "DISPLAY after sleep")
            fVoc.write("MPA", dMpa)
            fVoc.close()               

            cPa = u2py.Command("MPA")
            cPa.run(capture=True)  

    class MyThread(threading.Thread):
        def __init__(self, func, name='',daemon=None):
            threading.Thread.__init__(self)
            self.name = name
            self.func = func
            #self.result = self.func(*self.args)
            self.result = self.func()

        def run(self):
            print("\n","run at ",self.name,time.ctime())
            #self.func()
            if self.name=="t1":
                runPA()
            if self.name=="t2":
                getCapFileName()                    
            #print_time(self.name, 5, self.counter)
            #pdb.set_trace()
            print("Exiting ",self.name,time.ctime())    

        def get_result(self):
            try:
                return self.result
            except Exception:
                return None        
        '''
    #pdb.set_trace()
        '''
        #'''

    t1=MyThread(runPA,"t1")
    t2=MyThread(getCapFileName,"t2")
    t1.start()
    t2.start()        
    t1.join()
    t2.join()
    try:
        cFile=t2.get_result()
        print("cFile: ",cFile)
    except ValueError:
        print("cFile is None!")              

    self.replaceStdStreams()
    pdb.set_trace()
    self.repstdout.write(cFile)    
    self.repstdout.write("\n")
    self.restoreStdStreams()        

    assert(self.search_in_testoutput(cFile))

the output is below:

Quote:#python3 -m unittest -v u2py_command_test.TestCommandClass.test_UNV28366
test_UNV28366 (u2py_command_test.TestCommandClass) ...

run at t1 Thu Aug 29 18:48:27 2019

run at t2 Thu Aug 29 18:48:27 2019

u2pyruncapture-67277-722095

Exiting t2 Thu Aug 29 18:48:37 2019

Exiting t1 Thu Aug 29 18:48:37 2019

the script hung after print "Exiting t1 Thu Aug 29 18:48:37 2019"

without writing "u2pyruncapture-67277-722095" to file