sha*_*ams 2 python concurrency multithreading multiprocessing
我有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.如果是这样,我应该使用有关备用数据结构的任何建议吗？
看起来每个项目的计算成本不会超过与将工作分派给另一个线程/进程相关的开销.例如,以下是我在计算机上运行测试应用程序时看到的结果(与结果非常相似):
Version : 2.7.1
Compiler : MSC v.1500 32 bit (Intel)
Platform : Windows-7-6.1.7601-SP1
Processor : Intel64 Family 6 Model 30 Stepping 5, GenuineIntel
Num Threads/Processes: 8 ; Num Items: 16000
mainMultiprocessAndThreaded exec time: 1134.00006294 ms
mainPureMultiprocessing exec time: 917.000055313 ms
mainPureThreaded exec time: 111.000061035 ms
mainSerial exec time: 41.0001277924 ms
Terminating
如果我将正在执行的工作修改为计算成本更高的工作,例如:
def checkPrime(candidate):
i = 0;
for k in xrange(1,10000):
i += k
return i < 5000
然后我看到的结果更符合我的预期:
Version : 2.7.1
Compiler : MSC v.1500 32 bit (Intel)
Platform : Windows-7-6.1.7601-SP1
Processor : Intel64 Family 6 Model 30 Stepping 5, GenuineIntel
Num Threads/Processes: 8 ; Num Items: 16000
mainMultiprocessAndThreaded exec time: 2190.99998474 ms
mainPureMultiprocessing exec time: 2154.99997139 ms
mainPureThreaded exec time: 16170.0000763 ms
mainSerial exec time: 9143.00012589 ms
Terminating
您可能还想看看multiprocessing.Pool.它提供了与您描述的类似的模型(多个工作进程从公共队列中提取作业).对于您的示例,实现可能类似于:
@timeFunc
def mainPool(numThreads, numItems):
jobQueue = ConcurrentQueue()
fillQueueWithWork(jobQueue, numItems)
pool = multiprocessing.Pool(processes=numThreads)
results = []
while True:
dataItem = jobQueue.pop()
if dataItem == None:
break
results.append(pool.apply_async(checkPrime, dataItem))
pool.close()
pool.join()
在我的机器上,通过替代checkPrime实现,我看到了以下结果:
Version : 2.7.1
Compiler : MSC v.1500 32 bit (Intel)
Platform : Windows-7-6.1.7601-SP1
Processor : Intel64 Family 6 Model 30 Stepping 5, GenuineIntel
Num Threads/Processes: 8 ; Num Items: 1600
mainPool exec time: 1530.99989891 ms
Terminating
由于multiprocessing.Pool已经为插入工作提供了安全访问,因此您可能会将ConcurrentQueue您的动态工作直接删除并插入到工作中Pool.