不能grok python多處理

Question

我需要爲我的數據庫的每個元素運行一個函數。

當我嘗試以下方法：

from multiprocessing import Pool 
from pymongo import Connection 

def foo(): 
... 


connection1 = Connection('127.0.0.1', 27017) 
db1 = connection1.data 

my_pool = Pool(6) 
my_pool.map(foo, db1.index.find()) 

，我發現了以下錯誤：

Job 1, 'python myscript.py ' terminated by signal SIGKILL (Forced quit)

這是，我認爲，造成db1.index.find()吃所有可用的RAM，而試圖返回數以百萬計的數據庫元素...

我該如何修改我的代碼才能正常工作？

一些日誌會在這裏：

dmesg | tail -500 | grep memory 
[177886.768927] Out of memory: Kill process 3063 (python) score 683 or sacrifice child 
[177891.001379] [<ffffffff8110e51a>] out_of_memory+0xfa/0x250 
[177891.021362] Out of memory: Kill process 3063 (python) score 684 or sacrifice child 
[177891.025399] [<ffffffff8110e51a>] out_of_memory+0xfa/0x250 

實際以下功能：

def create_barrel(item): 
    connection = Connection('127.0.0.1', 27017) 
    db = connection.data 
    print db.index.count() 
    barrel = [] 
    fls = [] 
    if 'name' in item.keys(): 
     barrel.append(WhitespaceTokenizer().tokenize(item['name'])) 
     name = item['name'] 
    elif 'name.utf-8' in item.keys(): 
     barrel.append(WhitespaceTokenizer().tokenize(item['name.utf-8'])) 
     name = item['name.utf-8'] 
    else: 
     print item.keys() 
    if 'files' in item.keys(): 
     for file in item['files']: 
      if 'path' in file.keys(): 
       barrel.append(WhitespaceTokenizer().tokenize(" ".join(file['path']))) 
       fls.append(("\\".join(file['path']),file['length'])) 
      elif 'path.utf-8' in file.keys(): 
       barrel.append(WhitespaceTokenizer().tokenize(" ".join(file['path.utf-8']))) 
       fls.append(("\\".join(file['path.utf-8']),file['length'])) 
      else: 
       print file 
       barrel.append(WhitespaceTokenizer().tokenize(file)) 
    if len(fls) < 1: 
     fls.append((name,item['length'])) 
    barrel = sum(barrel,[]) 
    for s in barrel: 
     vs = re.findall("\d[\d|\.]*\d", s) #versions i.e. numbes such as 4.2.7500 
    b0 = [] 
    for s in barrel: 
     b0.append(re.split("[" + string.punctuation + "]", s)) 
    b1 = filter(lambda x: x not in string.punctuation, sum(b0,[])) 
    flag = True 
    while flag: 
     bb = [] 
     flag = False 
     for bt in b1: 
      if bt[0] in string.punctuation: 
       bb.append(bt[1:]) 
       flag = True 
      elif bt[-1] in string.punctuation: 
       bb.append(bt[:-1]) 
       flag = True 
      else: 
       bb.append(bt) 
     b1 = bb 
    b2 = b1 + barrel + vs 
    b3 = list(set(b2)) 
    b4 = map(lambda x: x.lower(), b3) 
    b_final = {} 
    b_final['_id'] = item['_id'] 
    b_final['tags'] = b4 
    b_final['name'] = name 
    b_final['files'] = fls 
    print db.barrels.insert(b_final) 

我已經注意到有趣的事情。然後我按Ctrl + C停止的過程，我發現了以下內容：

python index2barrel.py 
Traceback (most recent call last): 
    File "index2barrel.py", line 83, in <module> 
    my_pool.map(create_barrel, db1.index.find, 6) 
    File "/usr/lib/python2.7/multiprocessing/pool.py", line 227, in map 
    return self.map_async(func, iterable, chunksize).get() 
    File "/usr/lib/python2.7/multiprocessing/pool.py", line 280, in map_async 
    iterable = list(iterable) 
TypeError: 'instancemethod' object is not iterable 

我的意思是，爲什麼多是試圖事端轉換到列表中？這不是問題的根源嗎？

從堆棧跟蹤：

brk(0x231ccf000)      = 0x231ccf000 
futex(0x1abb150, FUTEX_WAKE_PRIVATE, 1) = 1 
sendto(3, "+\0\0\0\260\263\355\356\0\0\0\0\325\7\0\0\0\0\0\0data.index\0\0"..., 43, 0, NULL, 0) = 43 
recvfrom(3, "Some text from my database."..., 491663, 0, NULL, NULL) = 491663 
... [manymany times] 
brk(0x2320d5000)      = 0x2320d5000 
.... manymany times 

將上述樣品進入並進入在strace的輸出，由於某種原因strace的-o日誌文件蟒myscript.py 不會暫停。它只吃所有可用的RAM並寫入日誌文件。

UPDATE。使用imap而不是map解決了我的問題。

Answer 1

由於find()操作返回光標的地圖功能，因爲你說當你做這個運行沒有問題 for item in db1.index.find(): create_barrel(item) 它看起來像功能是OK。

你可以嘗試限制遊標中返回結果的數量，看看這是否有幫助？我覺得語法是：

db1.index.find().limit(100) 

如果你可以試試這個，看看它是否有助於它可能有助於獲取問題的原因。編輯1：我認爲你是通過使用map函數去錯誤的方式 - 我認爲你應該在mongo python驅動程序中使用map_reduce - 這樣map函數將由mongod進程執行。

Answer 2

map()函數給出了給定函數塊中的項。默認情況下此CHUNKSIZE計算如下（link to source）：

chunksize, extra = divmod(len(iterable), len(self._pool) * 4) 

這可能導致過大的塊大小，你的情況，並讓內存耗盡的過程。嘗試手動設置塊的大小是這樣的：

my_pool.map(foo, db1.index.find(), 100) 

編輯：你也應該考慮重新使用的數據庫連接，使用後關閉它們。現在您爲每個項目創建新的數據庫連接，並且您不要爲它們調用close()。

編輯2：同時檢查是否while循環進入無限循環（將解釋症狀）。

EDIT3：根據你添加的回溯函數，map函數試圖將光標轉換成列表，導致所有的項目一次被提取。發生這種情況是因爲它想要查找集合中有多少物品。這是pool.py的map()部分代碼：

if not hasattr(iterable, '__len__'): 
    iterable = list(iterable) 

你可以試試這個，避免轉換到列表：

cursor = db1.index.find() 
cursor.__len__ = cursor.count() 
my_pool.map(foo, cursor)