15

Preciso fazer uma série de operações envolvendo arquivos binários (não dá pra usar BD), e preciso garantir que elas terminem com sucesso mesmo em caso de falha no meio da operação. Para isso, não vejo saída a não ser implementar um sistema de journaling manualmente. Comecei a escrever um código, mas não estou seguro se está correto (i.e. se não existem casos em que uma falha no meio da tarefa pode colocá-lo em um estado inconsistente - inclusive falha na própria escrita do journal).

Existe algo pronto nesse sentido? E se não existe, há algum problema na minha tentativa de solução abaixo?

def nova_tarefa(args):
    refazer_journal() # Refaz o que ficou inacabado da última vez (se houver)
    with open('args.json', 'w') as f:
        json.dump(args, f) # Prepara as novas tarefas
    with open('journal.txt', 'w') as f:
        f.write('comecar\n') # Coloca as novas tarefas no journal
    refazer_journal() # Faz as novas tarefas

def refazer_journal():
    try:
        with open('journal.txt', 'r') as f:
            passos = [x.strip() for x in f.readlines() if x.strip()]
    except:
        if not os.path.exists('journal.txt'):
            passos = []
        else:
            raise
    if not passos: # Se não há nada inacabado, termina
        return
    with open('args.json', 'r') as f:
        args = json.load(f)

    # Realiza as tarefas que ainda não foram marcadas como concluídas
    for indice,tarefa in enumerate(args['tarefas']):
        if len(passos) <= indice+1:
            realizar_tarefa_indepotente(tarefa)
            with open('journal.txt', 'a') as f:
                f.write('\ntarefa concluida\n')

    with open('journal.txt', 'w') as f:
        pass # Tudo foi feito com sucesso: esvazia o journal

(Nota: o que está escrito no journal - comecar, tarefa concluida - não é importante, se apenas uma letra que seja for escrita na linha já se considera que o passo foi bem-sucedido)

Código para teste (comigo funcionou sempre - inclusive editando o journal.txt para introduzir uma série de erros):

def realizar_tarefa_indepotente(tarefa):
    if 'falhar' in tarefa:
        raise Exception('Tarefa falhou!')
    print 'Realizando tarefa: ' + tarefa['nome']

tarefa_ok = {'tarefas':[{'nome':'foo'}, {'nome':'bar'}, {'nome':'baz'}]}
falha_3 = {'tarefas':[{'nome':'foo'}, {'nome':'bar'}, {'nome':'baz','falhar':True}]}
falha_2 = {'tarefas':[{'nome':'foo'}, {'nome':'bar','falhar':True}, {'nome':'baz'}]}
falha_1 = {'tarefas':[{'nome':'foo','falhar':True}, {'nome':'bar'}, {'nome':'baz'}]}
falha_1_3 = {'tarefas':[{'nome':'foo','falhar':True}, {'nome':'bar'}, {'nome':'baz','falhar':True}]}
3
  • Nota: é do meu entendimento que revisão de código (code review) faz parte do nosso foco (é on topic) atualmente. Se alguém não concorda, favor se manifestar no meta.
    – mgibsonbr
    Commented 14/02/2014 às 5:55
  • Não quis colocar isso como resposta, já que é só um link, mas acho interessante documentar. Essa solução te satisfaz? Commented 14/02/2014 às 12:38
  • 1
    @FelipeAvelar A maioria dos recursos que achei ao buscar por "python journal" se referia não a journaling [no contexto de sistema de arquivos] mas à implementação de alguma espécie de "diário"... Creio que esse link também se trata disso - já que é um plugin para o plone (sistema de CMS).
    – mgibsonbr
    Commented 14/02/2014 às 22:40

4 Respostas 4

7

Escrevi um artigo que descreve como fazer isto: http://epx.com.br/artigos/arqtrans.php e inclusive um cidadão fez uma implementação em Python baseado no artigo: https://pypi.python.org/pypi/acidfile/1.2.0

A técnica básica é escrever dois arquivos no mínimo. Primeiro um, depois o outro em modo síncrono - conforme foi descrito em outras respostas.

O "tempero" do esquema é adicionar timestamp e um somatório (hash) nos dois arquivos. Dessa forma podemos saber qual é o arquivo mais recente - pelo timestamp - podendo verificar se o arquivo está íntegro (pelo hash).

O sistema de arquivos deve implementar journaling pelo menos dos meta-dados (geralmente é o caso, journaling dos dados custa performance). Isto garante que pelo menos a pasta e o arquivo sejam legíveis. Se os dados estiverem corrompidos você detecta pelo hash.

2
  • Exatamente o que eu estava procurando! Melhor do que reinventar a roda... Vou testar essa implementação que você citou, mas já dei uma olhada nos fontes e me parece que está tudo ok.
    – mgibsonbr
    Commented 15/02/2014 às 0:00
  • Acabei por utilizar sua solução: agora eu salvo o próprio args.json como um ACIDFile - assegurando de só atualizá-lo após me certificar que a tarefa anterior completou com sucesso (ver resposta do @utluiz). Simplificou bastante o código.
    – mgibsonbr
    Commented 15/02/2014 às 2:43
4
  • Um problema facilmente visível na sua implementação é que a forma que você checa pra ver se o arquivo journal.txt existe após a leitura está sujeita a race conditions, vide Como checar se um arquivo existe usando Python.

  • Além disso, o ideal é, não capturar exceções "naked", e sim definindo o tipo específico de exceção que você espera, p.ex. IOError, OSError, etc. Da mesma forma, quando for raise uma exceção, é bom utilizar uma com um nome específico relacionado ao tipo de erro.

  • Para iterar sobre o dicionário, como uma alternativa, ao invés de iterar sobre o resultado do enumerate(), você pode simplesmente iterar sobre a lista:

    passos = [{'nome':'foo','falhar':True}, {'nome':'bar'}, {'nome':'baz'}]
    for tarefa in passos:
        realizar_tarefa_indepotente(tarefa) 
    
  • A forma que você checa se o valor 'falhar' funciona, mas não é muito explícita. Recomendaria trocar por if 'falhar' in tarefa.items(), se você estiver simplesmente checando se ela existe, ou if tarefa.get('falhar') se você quiser certificar-se que o valor de tarefa['falhar'] é True.

7
  • Uma dúvida sobre o item #1: o try já não faz exatamente isso, solicitando a leitura do arquivo? Deveria então haver um outro bloco try aninhado?
    – utluiz
    Commented 14/02/2014 às 13:18
  • 2
    Não precisa, você pode simplesmente retirar o bloco if já que se ocorrer exceção dentro do try o arquivo provavelmente não existe. Seguindo o bullet seguinte, você poderia checar por exceções diferentes e tratar casos diferentes (p.ex. arquivo não existe, acesso negado, etc). Commented 14/02/2014 às 13:21
  • 2
    Sobre o item #3, acho que a ideia é não executar os passos que já foram executados. Por exemplo, se há 3 passos gravados no journal.txt, as tarefas serão executadas da 4ª em diante. Nesse caso, a não ser que o índice seja usado para alguma outra coisa no código real, talvez pudesse ser feito algo como for tarefa in args['tarefas'][len(passos):]:.
    – utluiz
    Commented 14/02/2014 às 13:36
  • 2
    Ou ainda: já que o seu arquivo journal.txt só guarda o número de tarefas que foram executadas com sucesso, você poderia gravar simplesmente o número, incrementando-o a cada sucesso e utilizando-o ao invés de len(passos) para pular as tarefas que foram executadas com sucesso. Commented 14/02/2014 às 13:44
  • 2
    [sobre o último comentário] isso não é possível, pois há o risco de eu corromper o journal.txt se houver falha durante a manipulação do mesmo. No modo que eu fiz, ou uma linha nova será adicionada ao arquivo (confirmando a tarefa) ou não (exigindo que ela seja refeita depois). [sobre #1 e #2] Boa sugestão, vou fazer! [sobre #3] A sugestão do @utluiz é boa. Eu preciso ignorar tarefas já feitas pois, embora cada tarefa individual seja indepotente no estado correto do sistema, ela não o é se o sistema já mudou de estado (i.e. se a tarefa seguinte já começou).
    – mgibsonbr
    Commented 14/02/2014 às 22:19
3

Garantia da gravação dos arquivos

Meu primeiro pensamento quando se trata de garantir o processamento de arquivos foi o buffer de dados.

Veja o que diz a documentação do file.flush():

Nota: o flush() não necessariamente escreve os dados do arquivo para o disco. Use flush() seguido por os.fsync() para garantir o comportamento. (tradução livre)

Suponho que as tarefas podem ter um volume bem maior de dados do que o journal.txt. Se ocorrer uma queda de energia no final do método refazer_journal(), o journal pode ter sido gravado no disco enquanto os dados das tarefas ainda estão em buffer.

Eu entendo que os dados do journal são enviados depois, mas não creio existir uma garantia de que o buffer será enfileirado sequencialmente pelo sistema operacional e pelo hardware.

4
  • 2
    Boa observação! Vou fazer isso e, pra garantir, também uma verificação dos arquivos recém-escritos para confirmar que está tudo ok antes de confirmar a tarefa. (e se entendi bem, é importante que o os.fsync seja usado também depois de se escrever o journal - para evitar que a tarefa seguinte comece sem que a registro de conclusão da tarefa anterior esteja pronto, certo?)
    – mgibsonbr
    Commented 14/02/2014 às 22:26
  • 2
    @mgibsonbr Sim, acredito que a sincronização deva ser feita após todas as gravações.
    – utluiz
    Commented 14/02/2014 às 23:44
  • 1
    Li ainda algum lugar (acho que foi no Linux System Programming) que o sync deveria ser feito 2 ou 3 vezes, como é o costume dos admins UNIX na linha de comando. Não lembro a razão exata, é algo como: o primeiro sync só inicia a gravação física, mas o segundo sync efetivamente bloqueia antes do primeiro terminar (se ainda não terminou).
    – epx
    Commented 27/02/2014 às 2:52
  • @epx Wow! Aí o nível baixou muito pra mim... rsrs
    – utluiz
    Commented 27/02/2014 às 12:59
2

Crie um journaling baseado em arquivos no filesystem. Sugiro usar uma nomenclatura baseada em epoch... Escrever e apagar itens de uma única vez é sempre mais seguro que manipula-lo milhares de vezes. O próprio journaling pode ser corrompido, já que atomicidade em python é dificil de garantir pra processos longos.

Você deve fazer log-in para responder a esta pergunta.

Esta não é a resposta que você está procurando? Pesquise outras perguntas com a tag .