Многопроцессорность и очередь с Dataframe

Рискуя оказаться не в состоянии предоставить полностью функциональный пример, вот что пойдет не так.

Прежде всего, это проблема времени.

Я снова попробовал ваш код с большими фреймами данных (10000 или даже 100000) и начал видеть то же, что и вы. Это означает, что вы видите это поведение, как только размер массивов пересекает определенный порог, который будет зависеть от системы (ЦП?).

Я немного изменил ваш код, чтобы было легче увидеть, что происходит. Во-первых, 5 DataFrames стоят put в очереди без каких-либо time.sleep пользовательских. В функции f_get я добавил счетчик (и time.sleep(0), см. ниже) в цикл (while not q.empty()).

Новый код:

import multiprocessing, time, sys                                                 
import pandas as pd                                                              

NR_ROWS = 10000                                                                  
i = 0                                                                            
def getDf():                                                                     
    global i, NR_ROWS                                                            
    myheader = ["name", "test2", "test3"]                                        
    myrow1 =   [ i,  i+400, i+250]                                               
    df = pd.DataFrame([myrow1]*NR_ROWS, columns = myheader)                      
    i = i+1                                                                      
    return df                                                                    

def f_put(q):                                                                    
    print "f_put start"                                                          
    j = 0                                                                        
    while(j < 5):                                                                
        data = getDf()                                                           
        q.put(data)                                                              
        print "P:", data["name"].iloc[0]                                         
        sys.stdout.flush()                                                       
        j += 1                                                                   

def f_get(q):                                                                    
    print "f_get start"                                                          
    while(1):
        data = pd.DataFrame()                                                    
        loop = 0                                                                 
        while not q.empty():                                                     
            data = q.get()                                                  
            print "get (loop: %s)" %loop
            time.sleep(0)                                         
            loop += 1                                                            
        time.sleep(1.)                                                           

if __name__ == "__main__":                                                       

    q = multiprocessing.Queue()                                                  
    p = multiprocessing.Process(target=f_put, args=(q,))                         
    p.start()                                                                    
    while(1):                                                                    
        f_get(q)                                                                 
    p.join()

Теперь, если вы запустите это для другого количества строк, вы увидите что-то вроде этого:

N=100:

f_get start
f_put start
P: 0
P: 1
P: 2
P: 3
P: 4
get (loop: 0)
get (loop: 1)
get (loop: 2)
get (loop: 3)
get (loop: 4)

N=10000:

f_get start
f_put start
P: 0
P: 1
P: 2
P: 3
P: 4
get (loop: 0)
get (loop: 1)
get (loop: 0)
get (loop: 0)
get (loop: 0)

О чем это нам говорит? Пока DataFrame мало, ваше предположение о том, что процесс put быстрее, чем get, кажется верным, мы можем получить все 5 элементов за один цикл while not q.empty().

Но по мере увеличения количества строк что-то меняется. Условие while q.empty() оценивается как True (очередь пуста) и внешний цикл while(1).

Это может означать, что put теперь медленнее, чем get, и нам нужно подождать. Но если мы установим время сна для всего f_get на что-то вроде 15, мы все равно получим такое же поведение.

С другой стороны, если мы изменим time.sleep(0) во внутреннем цикле q.get() на 1,

while not q.empty():                                                  
    data = q.get()                                                    
    time.sleep(1)                                                     
    print "get (loop: %s)" %loop                                      
    loop += 1

мы получаем это:

f_get start
f_put start
P: 0
P: 1
P: 2
P: 3
P: 4
get (loop: 0)
get (loop: 1)
get (loop: 2)
get (loop: 3)
get (loop: 4)

Это выглядит правильно! А это значит, что на самом деле get делает что-то странное. Кажется, что пока он все еще обрабатывает get, состояние очереди empty, и после того, как get выполнено, доступен следующий элемент.

Я уверен, что для этого есть причина, но я недостаточно знаком с multiprocessing, чтобы понять это.

В зависимости от вашего приложения вы можете просто добавить соответствующий time.sleep во внутренний цикл и посмотреть, достаточно ли этого.

Или, если вы хотите решить эту проблему (вместо использования обходного пути в качестве метода time.sleep), вы можете заглянуть в multiprocessing и найти информацию о блокировке, неблокировке или асинхронная коммуникация — думаю, решение будет найдено там.