Liberando o GIL do python para paralelizar seu código com threads

No Python o Global Interpreter Locker impede que duas threads executem ao mesmo tempo. Uma thread só é executada quando nenhuma outra estiver executando. A solução mais comum para aproveitar todos os cores de uma máquina em python é abandonar threads e utilizar vários processos. No entanto há uma maneira de aproveitar todos os cores com threads no python. Para isso vamos precisar criar uma extensão em C.

Primeiro vamos criar uma extensão que não libere o GIL para podermos comparar e conferir a melhoria de performance depois. O pivô dessa extensão é a seguinte função:

static int reduce_com_gil(int max, int (*f)(int x, int y)) {
   int retorno = 0;
   int i;
   for(i=0; i < max; i++){
       retorno = (*f)(retorno, i);
   }
   return retorno;
}

Essa função faz algo similar ao que um reduce faria, mas indo de 0 ao valor indicado em max. Para cada iteração a função enviada como segundo parâmetro é executada. A idéia é causar um grande processamento para que meu core fique travado. O uso dela está descrito no código abaixo:

static PyObject *antigil_calcular_com_gil(PyObject *self) {
   int valor = reduce_com_gil(100*1000, *antigil_calculos);
   char numero [5000];
   sprintf(numero, "%d", valor );
   return Py_BuildValue("s", numero);
}

Repare que eu passo para a função reduce_com_gil o ponteiro da função antigil_calculos. Essa outra função faz diversos cálculos a cada iteração do reduce. O nome antigil é o nome da extensão de exemplo. O código completo da extensão pode ser visto aqui.

Instalada a extensão, podemos testar a performance da lib com o seguinte código:

import antigil
antigil.calcular_com_gil()


E o seguinte comando:

$ time python teste.py
real 0m2.702s
user 0m2.692s


Mas o que a gente quer é saber como se comportam as threads. No caso o seguinte script abre 4 threads para aproveitar os 4 cores da minha máquina:

import antigil
from threading import Thread

threads = []
for i in xrange(4):
   t = Thread(target=antigil.calcular_com_gil)
   t.start()
   threads.append(t)

for t in threads:
   t.join()


No entando, acaba não aproveitando. Repare que ao executar 4 threads, o GIL age e impede que duas sejam executadas ao mesmo tempo. Dessa forma só um core da máquina é aproveitado. Isso pode ser observado pelo tempo total que é aproximadamente quatro vezes o tempo de execução de uma:

$ time python teste.py
real 0m10.910s
user 0m10.881s


Bom, vamos então à extensão não bloqueante. A chave do sucesso nesse caso são as macros Py_BEGIN_ALLOW_THREADS e Py_END_ALLOW_THREADS que respectivamente liberam o GIL e obtem o GIL de volta. Essas macros estão definidas em Python.h. O código da função reduce ficaria assim:

static int reduce_sem_gil(int max, int (*f)(int x, int y)) {
   int retorno = 0;
   Py_BEGIN_ALLOW_THREADS
   int i;
   for(i=0; i < max; i++){
       retorno = (*f)(retorno, i);
   }
   Py_END_ALLOW_THREADS
   return retorno;
}

Pronto, basta criar a função antigil_calcular_sem_gil similar à antigil_calcular_com_gil, utilizando a função reduce_sem_gil e então podemos repetir o teste. No caso parametrizei o script de teste para que você possa escolher qual função deseja executar:

import antigil
from threading import Thread
import sys

if 'com-gil' in sys.argv:
    calcular = antigil.calcular_com_gil
elif 'sem-gil' in sys.argv:
    calcular = antigil.calcular_sem_gil
else:
    print "Informar com-gil ou sem-gil"
    exit(0)

threads = []
for i in xrange(4):
   t = Thread(target=calcular)
   t.start()
   threads.append(t)

for t in threads:
   t.join()

O resultado é bem animador e mostra bem que o código rodou em paralelo:

$ time python teste.py sem-gil
real 0m3.414s
user 0m13.413s

Como o código utiliza apenas CPU, sem IO algum, ao aumentar para 8 threads, mesmo a versão que libera o GIL dobra de tempo pois só tenho disponível 4 cores na minha máquina. No entanto acredito que se fizer alguma operação de IO entre as macros Py_BEGIN_ALLOW_THREADS e Py_END_ALLOW_THREADS outras threads poderão ser executadas no caminho. Isso eu ainda preciso validar.

Só é preciso ter muito cuidado pois código entre essas macros está em território perigoso. A alteração de váriaveis globais ou ponteiros que podem ser compartilhados entre outras threads podem causar erros inesperados a qualquer momento. Portanto, é importante utilizar somente váriaveis locais e dados copiados.

O código completo da extensão em C antigil pode ser visto aqui.

Cópia exclusiva de coleção

Por serem de díficil identificação, problemas de concorrência acabam indo pra produção sem sequer serem notado em ambiente de testes e desenvolvimento. Erros acabam ocorrendo de maneira intermitente, causando dores de cabeça e acessos de insanidade nos desenvolvedores. Martin Fowler em seu livro Patterns of Enterprise Application Architecture identifica algumas soluções para evitar problemas de concorrência relativos a integridade de dados. Contudo há um outro problema que não está catalogado neste livro: A concorrência de informações na memória.

Ao guardar objetos em uma coleção que é compartilhada por outras threads, é necessário tomar providências para que não sejam lançadas exceções inesperadas. A medida básica é garantir que os pontos de acesso às coleções compartilhadas devem obter exclusividade sobre seu uso com a diretiva synchronized. Veja abaixo um exemplo de classe que torna o uso de uma lista à prova de erros de concorrência:

public class ListaSegura {
    private List lista = new ArrayList();
    public void adiciona(Object objeto) {
        synchronized (lista) {
            lista.add(objeto);
        }
    }
    public void remove(Object objeto) {
        synchronized (lista) {
            lista.remove(objeto);
        }
    }
    public void listar() {
        synchronized (lista) {
            for (Object o : lista) {
                System.out.println(o);
            }
        }
    }
}

É uma solução simples e que resolve em parte o problema. Contudo na maioria das vezes não podemos obter a exclusividade para iterar sobre uma coleção. No caso mostrado acima isso é possível pois estamos apenas imprimindo o objeto. Mas existem alguns casos em que obter essa exclusividade para a iteração nos traria alguns problemas. Esses casos estão descritos abaixo:

1- Alteração da coleção: No caso de você iterar pela coleção para remover ou adicionar algum objeto a ela. A exclusão ou adição ocorreria no meio da iteração e com isso seria lançada uma exceção de concorrência.Um exemplo simples disso são classes que gerenciam cache e precisam periodicamente remover objetos expirados.

2- Iteração prolongada: Quando a coleção possui objetos demais ou o processo executado durante a iteração é lento, tornando o tempo de exclusividade total muito longo. Isso faria com que o restante do sistema que precisasse utilizar essa coleção ficasse muito tempo aguardando pela exclusividade terminar. Um exemplo comum é a execução de métodos que acessem banco de dados dentro da iteração de um objeto exclusivo.

Para resolver esses dois casos identifiquei o padrão de desenvolvimento Cópia Exclusiva de Coleção. A idéia é obter a exclusividade da lista somente para fazer uma cópia dos itens em outra coleção e então poder iterar sobre esta cópia sem muitas preocupações. Abaixo mostro um exemplo de uma classe Armario que possui muitas Coisas. Se alguma coisa for lixo, ela deve ser removida na execução do método removeLixo.

public class Armario {
    private List coisas = new LinkedList();
    public void removeLixo() {
        List copia = lista();
        for (Coisa coisa : copia) {
            if (coisa.isLixo())
                remove(coisa);
        }
    }    
    public List lista() {
        List copia = null;
        synchronized (coisas) {
            copia = new ArrayList(coisas.size());
            copia.addAll(coisas);
        }
        return copia;
    }
    public void adiciona(Coisa coisa) {
        synchronized (coisas) {
            coisas.add(coisa);
        }
    }
    public void remove(Coisa coisa) {
        synchronized (coisas) {
            coisas.remove(coisa);
        }
    }
}

Com isso o tempo de exclusividade fica restrito ao tempo da cópia para a outra coleção, que ocorre no método lista. Dessa forma temos uma folga no tempo de iteração total e a possibilidade de rearranjar a coleção, adicionando ou removendo itens a ela. É importante ressaltar que o melhor jeito de evitar o calafrio de receber um java.util.ConcurrentModificationException é evitar a utilização de objetos compartilhados entre threads. Quando isso não é possível, o jeito é utilizar um padrão como Cópia Exclusiva de Coleção.

Load alto causado por um HashMap

Recentemente tivemos um problema em produção com nosso sistema. O load do servidor aumentava gradativamente chegando a um ponto em que era necessário reinicia-lo. Para quem não sabe, o load é um cálculo que indica quantas threads estão executando ou tentando executar em um determinado período. Ou seja, este problema de load alto poderia então ser causado por uma grande quantidade de acessos, no entanto, o load não baixava em periodos de pouco acesso, e até mesmo quando tirávamos o servidor do balanceamento.

Para analisar as threads que estavam executando, lançamos mão de um Thread Dump em dois momentos, antes de tirarmos a máquina do balanceamento, e outra uma hora depois de tirá-la. No Java, para obter um Thread Dump basta executar o seguinte comando, onde o processo_id é o pid do processo Java que queremos analisar.

kill -QUIT processo_id

Com o resultado em mãos descobrimos que antes e depois de tirarmos o servidor do balanceamento havia uma thread executando (runnable) travada na linha 325 de um HashMap. Veja a descrição dessa thread abaixo:

"ExecuteThread: '94' for queue: 'default'" daemon prio=5 tid=0x00b525a0 nid=0x2c runnable [0x63bfe000..0x63bffc28]
at java.util.HashMap.get(HashMap.java:325)
(...)

Foi então que descobrimos o problemas. Procuramos no Google por esta linha e encontramos diversas pessoas com o mesmo problema e até mesmo notificações deste possível bug para Sun. Contudo, isto não é um bug. O que acontece é que a classe HashMap não é thread safe, nem mesmo para seus métodos put e get. Portanto, se diversas threads gravam e recuperam valores de um HashMap, pode ocorrer problemas como esse. De fato, se olharmos o código de HashMap, podemos perceber que o método get tem grandes possibilidades de entrar em loop infinito.

A solução recomendade pela Sun é utilizar o wrapper sincronizado para mapas Collections.synchronizedMap(new HashMap()), ou utilizar a classe ConcurrentHashMap da api de concorrência incorporada ao Java 5, a java.util.concurrent. Para quem não pode utilizar o Java 5, pode-se utilizar a versão back-port da java.util.concurrent. No nosso caso, utilizamos essa versão backport.