2

我有以下导致 Nuke 挂起的代码。基本上，我想做的是从文件系统中获取文件和文件夹的列表，并且我试图通过并行处理来加速它。这在 Nuke 之外非常有效，但正如我之前所说，在 Nuke 中运行它会导致 Nuke 挂起。有没有更好的方法来做到这一点，这将导致 Nuke 不挂起？最好，我想通过 Python 的标准库或与平台无关的包来解决这个问题。但是，如果没有办法做到这一点，那我就可以了。最坏的情况，我将不得不回到不使用并行处理并找到其他优化。

此外，当我在 Nuke 中运行此代码时，我在控制台中收到以下错误：

Unknown units in -c from multiprocessing.forking import main; main()

编码：

#!/bin/env python

import multiprocessing
import os

CPU_COUNT = multiprocessing.cpu_count()


def _threaded_master(root):
    in_queue = multiprocessing.JoinableQueue()
    folder_queue = multiprocessing.JoinableQueue()
    file_queue = multiprocessing.JoinableQueue()

    in_queue.put(root)

    for _ in xrange(CPU_COUNT):
        multiprocessing.Process(target=_threaded_slave, args=(in_queue, folder_queue, file_queue)).start()

    in_queue.join()

    return {"folders": folder_queue, "files": file_queue}


def _threaded_slave(in_queue, folder_queue, file_queue):
    while True:
        path_item = in_queue.get()

        if os.path.isdir(path_item):
            for item in os.listdir(path_item):
                path = os.path.join(path_item, item)
                in_queue.put(path)

        in_queue.task_done()


if __name__ == "__main__":
    print _threaded_master(r"/path/to/root")
4

2 回答 2

2

这是我使用多个线程扫描大型目录树的代码。

我最初编写代码是为了使用 good old multiprocessing.Pool()，因为它非常简单，并且可以为您提供函数的结果。不需要输入和输出队列。另一个区别是它使用进程而不是线程，这有一些权衡。

有一个很大的Pool缺点：它假设您有一个要处理的静态项目列表。

因此，我按照您的原始示例重写了代码：要处理的目录的输入/输出队列和输出队列。调用者必须显式地从输出队列中获取项目。

为了咧嘴笑，我与老好人进行了时间比较os.walk()……至少在我的机器上，传统的解决方案更快。这两种解决方案产生了完全不同数量的文件，我无法解释。

玩得开心！

资源

#!/bin/env python

import multiprocessing, threading, time
import logging, os, Queue, sys

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)-4s %(levelname)s %(threadName)s %(message)s", 
    datefmt="%H:%M:%S",
    stream=sys.stderr,
)

def scan_dir(topdir):
    try:
        for name in os.listdir(topdir):
            path = os.path.join(topdir, name)
            yield (path, os.path.isdir(path))
    except OSError:
        logging.error('uhoh: %s', topdir)

def scan_dir_queue(inqueue, outqueue):
    logging.info('start')
    while True:
        try:
            dir_item = inqueue.get_nowait()
        except Queue.Empty:
            break

        res = list( scan_dir(dir_item) )
        logging.debug('- %d paths', len(res))
        for path,isdir in res:
            outqueue.put( (path,isdir) )
            if isdir:
                inqueue.put(path)
    logging.info('done')

def thread_master(root):
    dir_queue = Queue.Queue() # pylint: disable=E1101
    dir_queue.put(root)
    result_queue = Queue.Queue()

    threads = [
        threading.Thread(
            target=scan_dir_queue, args=[dir_queue, result_queue]
        )
        for _ in range(multiprocessing.cpu_count())
    ]

    for th in threads:
        th.start()
    for th in threads:
        th.join()
    return result_queue.queue

if __name__ == "__main__":
    topdir = os.path.expanduser('~')

    start = time.time()
    res = thread_master(topdir)
    print 'threaded:', time.time() - start
    print len(res), 'paths'

    def mywalk(topdir):
        for (dirpath, _dirnames, filenames) in os.walk(topdir):
            for name in filenames:
                yield os.path.join(dirpath, name)
    start = time.time()
    res = list(mywalk(topdir))
    print 'os.walk:', time.time() - start
    print len(res), 'paths'

输出

11:56:35 INFO Thread-1 start
11:56:35 INFO Thread-2 start
11:56:35 INFO Thread-3 start
11:56:35 INFO Thread-4 start
11:56:35 INFO Thread-2 done
11:56:35 INFO Thread-3 done
11:56:35 INFO Thread-4 done
11:56:42 INFO Thread-1 done
threaded: 6.49218010902
299230 paths
os.walk: 1.6940600872
175741 paths
于 2014-08-27T19:49:09.283 回答
1

这是一个可以参考的链接：https ://learn.foundry.com/nuke/developers/63/pythondevguide/threading.html

值得注意的是其中提到的警告： nuke.executeInMainThread和nuke.executeInMainThreadWithResult应始终从子线程运行。如果从主线程中运行，它们会冻结 NUKE。

所以，产生一个新的子线程，并在那里做你的事情。

于 2019-06-04T01:20:15.203 回答