Papers
Large-scale Incremental Processing Using Distributed Transactions and Notifications
Daniel Peng, Frank Dabek · 9th USENIX Symposium on Operating Systems Design and Implementation (OSDI) 2010, 2010
O problema: atualizar um repositório gigante sem reprocessá-lo inteiro
O artigo parte da tarefa de manter o índice de busca da web enquanto páginas novas são rastreadas. O sistema de indexação precisa preservar invariantes: se o mesmo conteúdo aparece em várias URLs, só a de maior PageRank entra no índice, e os links são invertidos para que o texto âncora seja associado à página apontada, inclusive através de duplicatas. Antes do trabalho descrito, isso era feito por uma série de operações MapReduce sobre todo o repositório. Em lote, as invariantes são fáceis de manter, porque cada etapa termina para todos os documentos antes de a próxima começar. O custo é que rodar os MapReduces só sobre as páginas novas não basta, já que há links entre elas e o resto da web; é preciso reprocessar tudo, o que descarta o trabalho de execuções anteriores e faz a latência ser proporcional ao tamanho do repositório, não ao tamanho da atualização. Guardar tudo num banco de dados com transações também não serve, porque os SGBDs da época não comportam o volume: o sistema de indexação do Google armazena dezenas de petabytes em milhares de máquinas. Já sistemas de armazenamento distribuído como o Bigtable escalam, mas não oferecem ferramentas para manter invariantes sob atualizações concorrentes. O Percolator ocupa essa lacuna com duas abstrações: transações ACID sobre um repositório de acesso aleatório e observers, trechos de código disparados quando uma coluna muda. Os autores dizem que ele não substitui as soluções existentes na maioria das tarefas: o que não se divide em pequenas atualizações (ordenar um arquivo) fica melhor no MapReduce, e quem não precisa de consistência forte pode usar só o Bigtable.
Arquitetura: uma camada sobre o Bigtable e o GFS
O Percolator é uma biblioteca cliente construída sobre o Bigtable. Em cada máquina do cluster rodam três binários: um worker do Percolator, um tablet server do Bigtable e um chunkserver do GFS. Os observers são ligados ao worker, que varre o Bigtable atrás de colunas alteradas (as notificações) e chama o observer correspondente como uma função no próprio processo. Os observers fazem transações enviando leituras e escritas aos tablet servers, que por sua vez falam com os chunkservers. O sistema depende ainda de dois serviços pequenos: o oráculo de timestamps, que fornece timestamps estritamente crescentes, e um serviço leve de locks, usado pelos workers para tornar mais eficiente a busca por notificações pendentes. Para o programador, o repositório é um punhado de tabelas; cada tabela é uma coleção de células indexadas por linha e coluna, e cada célula guarda um array de bytes sem interpretação. Internamente, para permitir isolamento de snapshot, cada célula é uma série de valores indexados por timestamp. Como o Bigtable já oferece transações atômicas apenas dentro de uma linha, o desafio do Percolator é entregar o que falta: transações entre várias linhas e o framework de observers. Duas decisões de projeto ficam claras. Primeiro, o sistema foi desenhado para escala enorme e sem exigir latência extremamente baixa, o que permite limpar de forma preguiçosa os locks deixados por transações em máquinas que falharam, um atraso que pode adiar o commit em dezenas de segundos. Segundo, não há um local central de gerência de transações, nem detector global de deadlock: isso aumenta a latência de transações em conflito, mas deixa o sistema escalar para milhares de máquinas.
Transações com isolamento de snapshot e commit em duas fases
O Percolator oferece transações entre linhas e tabelas com semântica ACID de isolamento de snapshot. Cada transação lê em um timestamp inicial, obtido do oráculo, e escreve em um timestamp de commit posterior. Se duas transações concorrentes escrevem na mesma célula, no máximo uma confirma. Isso não é serializabilidade: o isolamento de snapshot está sujeito a write skew. A vantagem, dizem os autores, é que as leituras ficam mais eficientes, pois qualquer timestamp representa um snapshot consistente e ler uma célula não exige adquirir locks. Como qualquer nó pode modificar o Bigtable diretamente, não existe um ponto onde interceptar o tráfego e conceder locks, então o Percolator mantém os locks explicitamente, em colunas especiais do próprio Bigtable, e eles precisam sobreviver a falhas de máquina. Cada coluna do Percolator vira três colunas do Bigtable: data, lock e write. As escritas feitas com Set() ficam em buffer até o commit, que é um commit em duas fases coordenado pelo cliente. Na fase de prewrite, a transação tenta travar todas as células escritas: aborta se vir um registro write posterior ao seu timestamp inicial (conflito de escrita) ou qualquer lock, em qualquer timestamp; senão grava o dado e o lock. Na segunda fase, obtém o timestamp de commit e, célula a célula, começando pela primária, troca o lock por um registro write que aponta para o timestamp onde está o dado. Um Get() verifica antes se há lock no intervalo visível do snapshot e, se houver, espera a liberação.
Falhas de cliente, lock primário e o oráculo de timestamps
Se um cliente cai no meio do commit, ficam locks para trás que travariam transações futuras. O Percolator limpa isso de forma preguiçosa: uma transação A que encontra um lock conflitante deixado por B pode concluir que B falhou e apagar seus locks. Como A nunca tem certeza total, é preciso evitar a corrida entre A limpando e um B não realmente falho confirmando. A solução é escolher uma célula de cada transação como ponto de sincronização, cujo lock é o lock primário; a localização do primário é gravada em todos os outros locks. Tanto o commit quanto a limpeza precisam modificar o primário dentro de uma transação de linha do Bigtable, então só um dos dois tem sucesso. Se o cliente cai depois do ponto de commit (já existe ao menos um registro write), o lock é avançado (roll forward): quem encontra o lock inspeciona o primário e, se ele virou um registro write, a transação confirmou. Caso contrário, faz rollback, o que é seguro porque o primário sempre é confirmado primeiro. Para estimar se um worker morreu, usa-se um token no Chubby, apagado automaticamente quando o processo sai, mais o horário gravado no lock, para pegar workers vivos porém parados. O oráculo entrega timestamps em ordem estritamente crescente e é contatado duas vezes por transação. Ele aloca faixas gravando em armazenamento estável o maior valor alocado, atende pedidos da memória e, após reiniciar, salta para frente, nunca para trás. Cada worker agrupa pedidos num único RPC pendente, e o oráculo serve cerca de 2 milhões de timestamps por segundo numa única máquina.
Notificações e observers: estruturar a computação incremental
Transações mantêm invariantes, mas o usuário também precisa de um modo de disparar o trabalho. No Percolator, o programador escreve observers: cada um registra uma função e um conjunto de colunas, e o sistema a invoca depois que dados são escritos em uma dessas colunas em qualquer linha. Uma aplicação é uma cadeia de observers, cada um criando trabalho para os seguintes ao escrever na tabela; no sistema de indexação, um MapReduce carrega os documentos por transações de carga, que disparam o processador do documento, que dispara o agrupamento, e assim por diante. Ao contrário de triggers de banco de dados, as notificações não servem para manter invariantes: o observer roda em transação separada da escrita que o disparou, então as duas não são atômicas. A garantia dada é que no máximo uma transação de observer confirma para cada mudança de coluna observada; já várias escritas podem causar uma única invocação, o que se chama colapso de mensagens. Para isso, cada coluna observada tem uma coluna de confirmação (ack) com o último timestamp inicial em que o observer rodou. Para achar células sujas entre trilhões, o Percolator mantém uma coluna notify, uma coluna simples do Bigtable sem propriedades transacionais, que serve só de dica para o varredor conferir a coluna ack e decidir se o observer deve rodar, num locality group separado; os workers a varrem a partir de pontos aleatórios, pegando antes um lock do serviço leve para não duplicar trabalho. Houve ainda um ajuste: as threads de varredura se agrupavam em poucas regiões (efeito de pelotão), e passaram a escolher nova posição aleatória ao detectar que varriam a mesma linha de outra thread.
Resultados e trade-offs da conversão do indexador
O Percolator foi construído para criar o índice base do Google, antes feito por uma série de 100 MapReduces, em que cada documento passava 2 a 3 dias sendo indexado. O sistema baseado nele, chamado Caffeine, rastreia o mesmo número de documentos, mas passa cada um pelo Percolator assim que é rastreado: a mediana é mais de 100 vezes mais rápida, e o resumo do artigo diz que a idade média dos documentos nos resultados caiu 50%. O preço é o consumo de recursos: o Caffeine usa cerca de duas vezes mais recursos para a mesma taxa de rastreamento, e a coleção de documentos era, na época do artigo, 3 vezes maior que a do sistema anterior. Em um benchmark sintético com 240 máquinas e um repositório de um bilhão de documentos, com 1% do repositório rastreado por hora, o MapReduce leva cerca de 30 minutos em média para agrupar um documento, enquanto o Percolator leva cerca de 2 segundos, uns 1000 vezes menos. Com taxas altas de rastreamento o MapReduce passa a ser melhor, porque ler dados em sequência é mais barato por byte que fazer consultas aleatórias; naquela configuração o Percolator saturava em 40% por hora. Num microbenchmark de escrita de uma única célula, a sobrecarga sobre o Bigtable puro é de cerca de quatro vezes. Num benchmark baseado no TPC-E, o desempenho escala quase linearmente de 11 a 15.000 núcleos, chegando a 11.200 transações por segundo, mas com estimativa de cerca de 30 vezes mais CPU por transação que o sistema comercial de referência, o custo das camadas.
Veja as ideias em ação
Pratique na escola de System Design os desenhos que usam este paper.