Grafos de tarefas e métodos de agendamento

Programação Paralela com Dask em Python

James Fulton

Climate Informatics Researcher

Visualizando um grafo de tarefas

# Create 2 delayed objects
delayed_num1 = delayed(my_square_function)(3)
delayed_num2 = delayed(my_square_function)(4)

# Add them
result = delayed_num1 + delayed_num2


# Plot the task graph
result.visualize()

Um diagrama mostrando as etapas para calcular o resultado. my-square-function roda duas vezes e as duas saídas são passadas para uma função add, que retorna uma saída.

Grafo de tarefas com sobreposição

delayed_intermediate = delayed(my_square_function)(3)

# These two results both use delayed_intermediate_result
delayed_result1 = delayed_intermediate - 5
delayed_result2 = delayed_intermediate + 4

Grafo de tarefas com sobreposição

delayed_result1.visualize()

Um diagrama mostrando o grafo de tarefas do resultado 1.

delayed_result2.visualize()

Um diagrama mostrando o grafo de tarefas do resultado 2.

Grafo de tarefas com sobreposição

# Plot the task graph
dask.visualize(delayed_result1, delayed_result2)

Um grafo de tarefas mostrando que o resultado 1 e o resultado 2 compartilham um resultado intermediário.

Multithreading vs. processamento paralelo

Movendo dados

Processamento paralelo

Processos têm seu próprio espaço de RAM

Multithreading

Threads usam o mesmo espaço de RAM

Multithreading vs. processamento paralelo

# Run a sum on two big arrays
sum1 = delayed(np.sum)(big_array1)
sum2 = delayed(np.sum)(big_array2)

# Compute using processes
dask.compute(sum1, sum2)

Lento com processamento paralelo

O diagrama mostra que os dois arrays que se originam em um processo Python devem ser enviados para dois outros processos Python.

Multithreading vs. processamento paralelo

# Run a sum on two big arrays
sum1 = delayed(np.sum)(big_array1)
sum2 = delayed(np.sum)(big_array2)

# Compute using threads
dask.compute(sum1, sum2)

Rápido com multithreading

O diagrama mostra que os dois arrays não precisam ser copiados.

O GIL

Global interpreter lock - apenas uma thread pode ler o script Python por vez

def sum_to_n(n):
    """Sums numbers from 0 to n"""
    total = 0
    for i in range(n+1):
        total += i
    return total

Multithreading não ajuda aqui
Processamento paralelo ajuda

sum1 = delayed(sum_to_n)(1000)
sum2 = delayed(sum_to_n)(1000)

Tempos de exemplo - GIL

Três gráficos de Gantt mostrando os tempos para rodar uma função Python simples 16 vezes. Entre três métodos de agendamento, processos foi o mais rápido.

Funções que liberam o GIL

Ex.: a função pd.read_csv() libera o GIL

df1 = delayed(pd.read_csv)('file1.csv')
df2 = delayed(pd.read_csv)('file2.csv')

Tempos de exemplo - Carregando dados

Resumo

Threads

Iniciam muito rápido
Compartilham a memória com a sessão principal
Sem transferência de memória
Limitadas pelo GIL, que permite uma thread ler o código por vez

Processos

Demoram e consomem memória para iniciar
Têm pools de memória separados
Transferência de dados muito lenta entre si e com a sessão principal do Python
Cada um tem seu próprio GIL, então não precisam se revezar para ler o código

Vamos praticar!

Programação Paralela com Dask em Python