在Python中使用多进程+线程加速应用遇到问题

Question

我有一个主要依赖CPU的应用程序，想通过使用多进程和多线程来加速，而不是仅仅使用纯线程版本。我写了一个简单的应用程序来测试我的方法，结果让我很惊讶：多进程和多进程加线程的版本表现得比线程版本和串行版本都要差。

在我的应用程序中，有一个工作队列，用来存储所有的工作。线程会一个一个地从队列中取出工作项，然后处理它，处理方式可以是直接处理（线程版本），也可以是把它传递给一个进程。线程需要等结果出来后才能继续处理下一个工作项。我之所以一个一个取工作项，是因为工作是动态的（下面粘贴的原型应用代码中并不是这样），我无法在创建时就把工作分好，分配给每个线程或进程。

我想知道我哪里做错了，以及如何能加速我的应用程序。

这是我在一台16核机器上运行时的执行时间：

Version      : 2.7.2
Compiler     : GCC 4.1.2 20070925 (Red Hat 4.1.2-33)
Platform     : Linux-2.6.24-perfctr-x86_64-with-fedora-8-Werewolf
Processor    : x86_64
Num Threads/Processes: 8 ; Num Items: 16000
mainMultiprocessAndThreaded exec time: 3505.97214699  ms
mainPureMultiprocessing exec time: 2241.89805984  ms
mainPureThreaded exec time: 309.767007828  ms
mainSerial exec time: 52.3412227631  ms
Terminating

这是我使用的代码：

import threading
import multiprocessing
import time
import platform

class ConcurrentQueue:
    def __init__(self):
        self.data = []
        self.lock = threading.Lock()

    def push(self, item):
        self.lock.acquire()
        try:
            self.data.append(item)
        finally:
            self.lock.release()
        return

    def pop(self):
        self.lock.acquire()
        result = None
        try:
            length = len(self.data)
            if length > 0:
                result = self.data.pop()
        finally:
            self.lock.release()
        return result

    def isEmpty(self, item):
        self.lock.acquire()
        result = 0
        try:
            result = len(self.data)
        finally:
            self.lock.release()
        return result != 0


def timeFunc(passedFunc):
    def wrapperFunc(*arg):
        startTime = time.time()
        result = passedFunc(*arg)
        endTime = time.time()
        elapsedTime = (endTime - startTime) * 1000
        print passedFunc.__name__, 'exec time:', elapsedTime, " ms"
        return result
    return wrapperFunc

def checkPrime(candidate):
    # dummy process to do some work
    for k in xrange(3, candidate, 2):
        if candidate % k:
            return False
    return True

def fillQueueWithWork(itemQueue, numItems):
    for item in xrange(numItems, 2 * numItems):
        itemQueue.push(item)


@timeFunc
def mainSerial(numItems):
    jobQueue = ConcurrentQueue()
    fillQueueWithWork(jobQueue, numItems)

    while True:
        dataItem = jobQueue.pop()
        if dataItem is None:
            break
        # do work with dataItem
        result = checkPrime(dataItem)
    return

# Start: Implement a pure threaded version
def pureThreadFunc(jobQueue):
    curThread = threading.currentThread()
    while True:
        dataItem = jobQueue.pop()
        if dataItem is None:
            break
        # do work with dataItem
        result = checkPrime(dataItem)
    return

@timeFunc
def mainPureThreaded(numThreads, numItems):
    jobQueue = ConcurrentQueue()
    fillQueueWithWork(jobQueue, numItems)

    workers = []
    for index in xrange(numThreads):
        loopName = "Thread-" + str(index)
        loopThread = threading.Thread(target=pureThreadFunc, name=loopName, args=(jobQueue, ))
        loopThread.start()
        workers.append(loopThread)

    for worker in workers:
        worker.join()

    return
# End: Implement a pure threaded version

# Start: Implement a pure multiprocessing version
def pureMultiprocessingFunc(jobQueue, resultQueue):
    while True:
        dataItem = jobQueue.get()
        if dataItem is None:
            break
        # do work with dataItem
        result = checkPrime(dataItem)
        resultQueue.put_nowait(result)
    return

@timeFunc
def mainPureMultiprocessing(numProcesses, numItems):
    jobQueue = ConcurrentQueue()
    fillQueueWithWork(jobQueue, numItems)

    workers = []
    queueSize = (numItems/numProcesses) + 10
    for index in xrange(numProcesses):
        jobs = multiprocessing.Queue(queueSize)
        results = multiprocessing.Queue(queueSize)
        loopProcess = multiprocessing.Process(target=pureMultiprocessingFunc, args=(jobs, results, ))
        loopProcess.start()
        workers.append((loopProcess, jobs, results))

    processIndex = 0
    while True:
        dataItem = jobQueue.pop()
        if dataItem is None:
            break
        workers[processIndex][1].put_nowait(dataItem)

        processIndex += 1
        if numProcesses == processIndex:
            processIndex = 0

    for worker in workers:
        worker[1].put_nowait(None)

    for worker in workers:
        worker[0].join()

    return
# End: Implement a pure multiprocessing version

# Start: Implement a threaded+multiprocessing version
def mpFunc(processName, jobQueue, resultQueue):
    while True:
        dataItem = jobQueue.get()
        if dataItem is None:
            break
        result = checkPrime(dataItem)
        resultQueue.put_nowait(result)
    return

def mpThreadFunc(jobQueue):
    curThread = threading.currentThread()
    threadName = curThread.getName()

    jobs = multiprocessing.Queue()
    results = multiprocessing.Queue()

    myProcessName = "Process-" + threadName
    myProcess = multiprocessing.Process(target=mpFunc, args=(myProcessName, jobs, results, ))
    myProcess.start()

    while True:
        dataItem = jobQueue.pop()
        # put item to allow process to start
        jobs.put_nowait(dataItem)
        # terminate loop if work queue is empty
        if dataItem is None:
            break
        # wait to get result from process
        result = results.get()
        # do something with result
    return

@timeFunc
def mainMultiprocessAndThreaded(numThreads, numItems):
    jobQueue = ConcurrentQueue()
    fillQueueWithWork(jobQueue, numItems)

    workers = []
    for index in xrange(numThreads):
        loopName = "Thread-" + str(index)
        loopThread = threading.Thread(target=mpThreadFunc, name=loopName, args=(jobQueue, ))
        loopThread.start()
        workers.append(loopThread)

    for worker in workers:
        worker.join()

    return
# End: Implement a threaded+multiprocessing version

if __name__ == '__main__':

    print 'Version      :', platform.python_version()
    print 'Compiler     :', platform.python_compiler()
    print 'Platform     :', platform.platform()
    print 'Processor    :', platform.processor()

    numThreads = 8
    numItems = 16000 #200000

    print "Num Threads/Processes:", numThreads, "; Num Items:", numItems

    mainMultiprocessAndThreaded(numThreads, numItems)
    mainPureMultiprocessing(numThreads, numItems)
    mainPureThreaded(numThreads, numItems)
    mainSerial(numItems)

    print "Terminating"

补充：我猜测慢的原因之一是Queue.put()在忙等待，而不是释放全局解释器锁（GIL）。如果是这样，有没有建议我应该使用的其他数据结构？

性能优化多线程并发编程数据结构全局解释器锁多进程工作队列 CPU密集型

在Python中使用多进程+线程加速应用遇到问题

2 个回答

撰写回答