代码之家  ›  专栏  ›  技术社区  ›  Sander

多个python进程之间的os.chdir

  •  2
  • Sander  · 技术社区  · 12 年前

    我有一个复杂的python管道(我无法更改代码),调用多个其他脚本和其他可执行文件。重点是,要运行8000多个目录,进行一些科学分析,需要很长时间。因此,我使用多处理模块编写了一个简单的包装器(可能不是最有效的,但似乎有效)。

    from os import path, listdir, mkdir, system
    from os.path import join as osjoin, exists, isfile
    from GffTools import Gene, Element, Transcript
    from GffTools import read as gread, write as gwrite, sort as gsort
    from re import match
    from multiprocessing import JoinableQueue, Process
    from sys import argv, exit
    
    # some absolute paths
    inbase = "/.../abfgp_in"
    outbase = "/.../abfgp_out"
    abfgp_cmd = "python /.../abfgp-2.rev/abfgp.py"
    refGff = "/.../B0510_manual_reindexed_noSeq.gff"
    
    # the Queue
    Q = JoinableQueue()
    i = 0
    
    # define number of processes
    try: num_p = int(argv[1])
    except ValueError: exit("Wrong CPU argument")
    
    # This is the function calling the abfgp.py script, which in its turn calls alot of third party software
    def abfgp(id_, pid):
        out = osjoin(outbase, id_)
        if not exists(out): mkdir(out)
    
        # logfile
        log = osjoin(outbase, "log_process_%s" %(pid))
        try:
            # call the script
            system("%s --dna %s --multifasta %s --target %s -o %s -q >>%s" %(abfgp_cmd, osjoin(inbase, id_, id_ +".dna.fa"), osjoin(inbase, id_, "informants.mfa"), id_, out, log))
        except:
            print "ABFGP FAILED"
            return
    
    # parse the output
    def extractGff(id_):
       # code not relevant
    
    
    # function called by multiple processes, using the Queue
    def run(Q, pid):
        while not Q.empty():
            try:
                d = Q.get()             
                print "%s\t=>>\t%s" %(str(i-Q.qsize()), d)          
                abfgp(d, pid)
                Q.task_done()
            except KeyboardInterrupt:
                exit("Interrupted Child")
    
    # list of directories
    genedirs = [d for d in listdir(inbase)]
    genes = gread(refGff)
    for d in genedirs:
        i += 1
        indir = osjoin(inbase, d)
        outdir = osjoin(outbase, d)
        Q.put(d)
    
    # this loop creates the multiple processes
    procs = []
    for pid in range(num_p):
        try:
            p = Process(target=run, args=(Q, pid+1))
            p.daemon = True
            procs.append(p) 
            p.start()
        except KeyboardInterrupt:
            print "Aborting start of child processes"
            for x in procs:
                x.terminate()
            exit("Interrupted")     
    
    try:
        for p in procs:
            p.join()
    except:
        print "Terminating child processes"
        for x in procs:
            x.terminate()
        exit("Interrupted")
    
    print "Parsing output..."
    for d in genedirs: extractGff(d)
    

    现在的问题是,abfgp.py使用os.chdir函数,这似乎会中断并行处理。我收到了很多错误,指出一些(输入/输出)文件/目录无法读取/写入。尽管我通过os.system()调用脚本,但我认为从中派生单独的进程可以防止这种情况。

    如何解决这些chdir干扰?

    编辑:我可能会用正确的目录将os.system()更改为subprocess.Popen(cwd=“…”)。我希望这会有所不同。

    谢谢

    2 回复  |  直到 12 年前
        1
  •  1
  •   User    12 年前

    编辑2

    不要使用 os.system() 使用 subprocess.call()

    system("%s --dna %s --multifasta %s --target %s -o %s -q >>%s" %(abfgp_cmd, osjoin(inbase, id_, id_ +".dna.fa"), osjoin(inbase, id_, "informants.mfa"), id_, out, log))
    

    将转换为

    subprocess.call((abfgp_cmd, '--dna', osjoin(inbase, id_, id_ +".dna.fa"), '--multifasta', osjoin(inbase, id_, "informants.mfa"), '--target', id_, '-o', out, '-q')) # without log.
    

    编辑1 我认为问题是多处理使用模块名来序列化函数和类。

    这意味着如果你 import module 模块所在位置 ./module.py 你做的事情 os.chdir('./dir') 现在你需要 from .. import module .

    子进程继承父进程的文件夹。这可能是个问题。

    解决

    1. 确保导入所有模块(在子进程中),然后更改目录
    2. 插入原件 os.getcwd() 到 sys.path 以启用从原始目录导入。这必须在从本地目录调用任何函数之前完成。
    3. 将您使用的所有函数放在始终可以导入的目录中。这个 site-packages 可能是这样的目录。然后你可以做一些类似的事情 导入模块 module.main() 开始你的工作。
    4. 这是我做的一个黑客,因为我知道泡菜是如何工作的。仅当其他尝试失败时使用此选项。 脚本打印:

      serialized # the function runD is serialized
      string executed # before the function is loaded the code is executed
      loaded # now the function run is deserialized
      run # run is called
      

      在你的情况下,你会这样做:

      runD = evalBeforeDeserialize('__import__("sys").path.append({})'.format(repr(os.getcwd())), run)
      p = Process(target=runD, args=(Q, pid+1))
      

      这是脚本:

      # functions that you need
      
      class R(object):
          def __init__(self, call, *args):
      
              self.ret = (call, args)
          def __reduce__(self):
              return self.ret
          def __call__(self, *args, **kw):
              raise NotImplementedError('this should never be called')
      
      class evalBeforeDeserialize(object):
          def __init__(self, string, function):
              self.function = function
              self.string = string
          def __reduce__(self):
              return R(getattr, tuple, '__getitem__'), \
                       ((R(eval, self.string), self.function), -1)
      
      # code to show how it works        
      
      def printing():
          print('string executed')
      
      def run():
          print('run')
      
      runD = evalBeforeDeserialize('__import__("__main__").printing()', run)
      
      import pickle
      
      s = pickle.dumps(runD)
      print('serialized')
      run2 = pickle.loads(s)
      print('loaded')
      run2()
      

    如果这些不起作用,请反馈。

        2
  •  0
  •   Chris Johnson user3351229    12 年前

    您可以确定 os 不可更改程序正在使用的库;然后创建定制版本 chdir 在那个库中,它可以满足您的需要--防止目录更改,记录它,等等。如果定制的行为需要仅针对单个程序,则可以使用 inspect 模块来识别调用者,并以特定的方式为该调用者定制行为。

    如果你真的不能改变现有的程序,你的选择是有限的;但是,如果您可以选择更改它导入的库,那么这样做可能是避免不希望出现的行为的一种侵入性最小的方法。

    更改标准库时,通常需要注意。