threading-verification.py 文件源码

def fetch_in_thread(sr, nsa):
    """fetch a sequence in a thread

    """

    def fetch_seq(q, nsa):
        pid, ppid = os.getpid(), os.getppid()
        q.put((pid, ppid, sr[nsa]))

    q = Queue()
    p = Process(target=fetch_seq, args=(q, nsa))
    p.start()
    pid, ppid, seq = q.get()
    p.join()

    assert pid != ppid, "sequence was not fetched from thread"
    return pid, ppid, seq