【发布时间】：2012-01-22 15:50:11
【问题描述】：
我有 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。如果是这样，对我应该使用的替代数据结构有什么建议吗？
【问题讨论】：
-
据我所知，python 多线程不是为了加快执行速度。 （我可能完全错了，所以请纠正我）。 understanding GIL
-
@reclosedev 我在示例中看到的不同之处在于预先提供了目标的所有输入，而不是重复调用 queue.get()。同样，每个进程只有一次对 result.get() 的调用。
-
@c-ram 是的，我希望多处理变体能够实现一些加速，但即使是这些变体也更慢。
标签： python multithreading concurrency multiprocessing