A máquina de eventos protege o ciclo de vida

Isenção de responsabilidade: Este artigo descreve uma solução não óbvia para um problema não óbvio. Antes de correr ovos colocá-lo em prática, recomendo ler o artigo até o fim e pensar duas vezes.

Olá pessoal! Ao trabalhar com código, geralmente precisamos lidar com o estado . Um desses casos é o ciclo de vida dos objetos. Gerenciar um objeto com vários estados possíveis pode ser uma tarefa não trivial. Adicione a execução assíncrona aqui e a tarefa é complicada por uma ordem de magnitude. Existe uma solução eficaz e natural. Neste artigo, falarei sobre a máquina de eventos e como implementá-la no Go.

Por que gerenciar o estado?

Para começar, vamos definir o próprio conceito. O exemplo mais simples de um estado: arquivos e várias conexões. Você não pode simplesmente pegar e ler um arquivo. Ele deve primeiro ser aberto e, no final de preferência não se esqueça de fechar. Acontece que a ação atual depende do resultado da ação anterior: a leitura depende da abertura. O resultado salvo é o estado.

O principal problema com o estado é a complexidade. Qualquer estado complica automaticamente o código. Você precisa armazenar os resultados das ações na memória e adicionar várias verificações à lógica. É por isso que as arquiteturas sem estado são tão atraentes para os programadores - ninguém quer problemas dificuldades. Se os resultados de suas ações não afetarem a lógica de execução, você não precisará de um estado.

No entanto, há uma propriedade que faz você contar com as dificuldades. Um estado exige que você siga uma ordem específica de ações. Em geral, essas situações devem ser evitadas, mas isso nem sempre é possível. Um exemplo é o ciclo de vida dos objetos do programa. Graças ao bom gerenciamento de estado, é possível obter um comportamento previsível de objetos com um ciclo de vida complexo.

Agora vamos descobrir como fazer isso legal .

Automático como forma de resolver problemas

Quando as pessoas falam sobre estados, as máquinas de estados finitos vêm imediatamente à mente. É lógico, porque um autômato é a maneira mais natural de gerenciar um estado.

Não vou me aprofundar na teoria dos autômatos : há informações mais do que suficientes na Internet.

Se você procurar exemplos de máquinas de estado finito para o Go, certamente encontrará um lexer de Rob Pike . Um ótimo exemplo de autômato no qual os dados processados ​​são o alfabeto de entrada. Isso significa que as transições de estado são causadas pelo texto que o lexer processa. Solução elegante para um problema específico.

O principal a entender é que um autômato é uma solução para um problema estritamente específico. Portanto, antes de considerá-lo um remédio para todos os problemas, você deve entender completamente a tarefa. Especificamente, a entidade que você deseja controlar:

• estados - ciclo de vida;
• eventos - o que exatamente causa a transição para cada estado;
• resultado do trabalho - dados de saída;
• modo de execução (síncrono / assíncrono);
• principais casos de uso.

O lexer é bonito, mas apenas muda de estado devido aos dados que ele próprio processa. Mas e a situação em que o usuário chama transições? É aqui que a máquina do evento pode ajudar.

Exemplo real

Para deixar mais claro, analisarei um exemplo da biblioteca phono .

Para uma imersão completa no contexto, você pode ler o artigo introdutório . Isso não é necessário para este tópico, mas ajudará a entender melhor o que estamos gerenciando.

E o que estamos gerenciando?

phono é baseado no pipeline DSP. Consiste em três estágios de processamento. Cada estágio pode incluir de um a vários componentes:

1. pipe.Pump (bomba inglesa) é um estágio obrigatório de recebimento de som, sempre apenas um componente.
2. pipe.Processor (manipulador inglês) - um estágio opcional do processamento de som, de 0 a N componentes.
3. pipe.Sink (pia inglesa) - um estágio obrigatório de transmissão de som, de 1 a N componentes.

Na verdade, gerenciaremos o ciclo de vida do transportador.

Ciclo de vida

É assim que o diagrama de estado do pipe.Pipe parece.

Itálico indica transições causadas pela lógica de execução interna. Negrito - transições causadas por eventos. O diagrama mostra que os estados são divididos em 2 tipos:

• estados inativos - ready e paused , você só pode pular deles por evento
• estados ativos - running e pausing , transições por evento e devido à lógica de execução

Antes de uma análise detalhada do código, um exemplo claro do uso de todos os estados:

 // PlayWav  .wav    portaudio  -. func PlayWav(wavFile string) error { bufferSize := phono.BufferSize(512) //      w, err := wav.NewPump(wavFile, bufferSize) //  wav pump if err != nil { return err } pa := portaudio.NewSink( //  portaudio sink bufferSize, w.WavSampleRate(), w.WavNumChannels(), ) p := pipe.New( //  pipe.Pipe    ready w.WavSampleRate(), pipe.WithPump(w), pipe.WithSinks(pa), ) p.Run() //    running   p.Run() errc := p.Pause() //    pausing   p.Pause() err = pipe.Wait(errc) //     paused if err != nil { return err } errc = p.Resume() //    running   p.Resume() err = pipe.Wait(errc) //     ready if err != nil { return err } return pipe.Wait(p.Close()) //      } 

Agora, as primeiras coisas primeiro.

Todo o código fonte está disponível no repositório .

Estados e Eventos

Vamos começar com a coisa mais importante.

 // state      . type state interface { listen(*Pipe, target) (state, target) //    transition(*Pipe, eventMessage) (state, error) //   } // idleState  .        . type idleState interface { state } // activeState  .         //   . type activeState interface { state sendMessage(*Pipe) state //    } //  . type ( idleReady struct{} activeRunning struct{} activePausing struct{} idlePaused struct{} ) //  . var ( ready idleReady running activeRunning paused idlePaused pausing activePausing ) 

Graças a tipos separados, as transições também são declaradas separadamente para cada estado. Isso evita a enorme salsichas funções de transição com switch aninhadas. Os próprios estados não contêm nenhum dado ou lógica. Para eles, você pode declarar variáveis ​​no nível do pacote para não fazer isso sempre. A interface de state é necessária para o polimorfismo. Falaremos sobre activeState e idleState pouco mais tarde.

A segunda parte mais importante da nossa máquina são os eventos.

 // event  . type event int //  . const ( run event = iota pause resume push measure cancel ) // target      . type target struct { state idleState //   errc chan error //   ,     } // eventMessage   ,    . type eventMessage struct { event //   params params //   components []string // id  target //      } 

Para entender por que o tipo de target é necessário, considere um exemplo simples. Criamos um novo transportador, está ready . Agora execute-o com p.Run() . O evento de run é enviado para a máquina, o pipeline entra no estado de running . Como descobrir quando o transportador está terminado? É aqui que o tipo de target nos ajudará. Indica o estado de descanso esperado após o evento. Em nosso exemplo, após a conclusão do trabalho, o pipeline entrará novamente no estado ready . A mesma coisa no diagrama:

Agora, mais sobre os tipos de estados. Mais precisamente, sobre as activeState e activeState . Vejamos as funções listen(*Pipe, target) (state, target) para diferentes tipos de estágios:

 // listen     ready. func (s idleReady) listen(p *Pipe, t target) (state, target) { return p.idle(s, t) } // listen     running. func (s activeRunning) listen(p *Pipe, t target) (state, target) { return p.active(s, t) } 

pipe.Pipe possui funções diferentes para aguardar uma transição! O que tem ai?

 // idle     .    . func (p *Pipe) idle(s idleState, t target) (state, target) { if s == t.state || s == ready { t = t.dismiss() //  ,  target } for { var newState state var err error select { case e := <-p.events: //   newState, err = s.transition(p, e) //    if err != nil { e.target.handle(err) } else if e.hasTarget() { t.dismiss() t = e.target } } if s != newState { return newState, t // ,    } } } // active     .     , //   . func (p *Pipe) active(s activeState, t target) (state, target) { for { var newState state var err error select { case e := <-p.events: //   newState, err = s.transition(p, e) //    if err != nil { //  ? e.target.handle(err) // ,    } else if e.hasTarget() { // ,  target t.dismiss() //   t = e.target //   } case <-p.provide: //     newState = s.sendMessage(p) //    case err, ok := <-p.errc: //   if ok { //   ,  interrupt(p.cancel) //   t.handle(err) //    } //    ,  return ready, t //    ready } if s != newState { return newState, t // ,    } } } 

Assim, podemos ouvir diferentes canais em diferentes estados. Por exemplo, isso permite que você não envie mensagens durante uma pausa - simplesmente não ouvimos o canal correspondente.

Construtor e partida da máquina

 // New      . //      ready. func New(sampleRate phono.SampleRate, options ...Option) *Pipe { p := &Pipe{ UID: phono.NewUID(), sampleRate: sampleRate, log: log.GetLogger(), processors: make([]*processRunner, 0), sinks: make([]*sinkRunner, 0), metrics: make(map[string]measurable), params: make(map[string][]phono.ParamFunc), feedback: make(map[string][]phono.ParamFunc), events: make(chan eventMessage, 1), //    cancel: make(chan struct{}), //     provide: make(chan struct{}), consume: make(chan message), } for _, option := range options { //   option(p)() } go p.loop() //    return p } 

Além das opções de inicialização e funcionais , há o início de uma goroutine separada com o ciclo principal. Bem, olhe para ele:

 // loop ,     nil . func (p *Pipe) loop() { var s state = ready //   t := target{} for s != nil { s, t = s.listen(p, t) //      p.log.Debug(fmt.Sprintf("%v is %T", p, s)) } t.dismiss() close(p.events) //    } // listen     ready. func (s idleReady) listen(p *Pipe, t target) (state, target) { return p.idle(s, t) } // transition       . func (s idleReady) transition(p *Pipe, e eventMessage) (state, error) { switch e.event { case cancel: interrupt(p.cancel) return nil, nil case push: e.params.applyTo(p.ID()) p.params = p.params.merge(e.params) return s, nil case measure: for _, id := range e.components { e.params.applyTo(id) } return s, nil case run: if err := p.start(); err != nil { return s, err } return running, nil } return s, ErrInvalidState } 

O transportador é criado e congelou em antecipação a eventos.

Hora de trabalhar

Ligue para p.Run() !

 // Run   run  . //     pipe.Close  . func (p *Pipe) Run() chan error { runEvent := eventMessage{ event: run, target: target{ state: ready, //    errc: make(chan error, 1), }, } p.events <- runEvent return runEvent.target.errc } // listen     running. func (s activeRunning) listen(p *Pipe, t target) (state, target) { return p.active(s, t) } // transition       . func (s activeRunning) transition(p *Pipe, e eventMessage) (state, error) { switch e.event { case cancel: interrupt(p.cancel) err := Wait(p.errc) return nil, err case measure: e.params.applyTo(p.ID()) p.feedback = p.feedback.merge(e.params) return s, nil case push: e.params.applyTo(p.ID()) p.params = p.params.merge(e.params) return s, nil case pause: return pausing, nil } return s, ErrInvalidState } // sendMessage   . func (s activeRunning) sendMessage(p *Pipe) state { p.consume <- p.newMessage() return s } 

running gera mensagens e é executada até que o pipeline seja concluído.

Pausar

Durante a execução do transportador, podemos fazer uma pausa. Nesse estado, o pipeline não gerará novas mensagens. Para fazer isso, chame o método p.Pause() .

 // Pause   pause  . //     pipe.Close  . func (p *Pipe) Pause() chan error { pauseEvent := eventMessage{ event: pause, target: target{ state: paused, //    errc: make(chan error, 1), }, } p.events <- pauseEvent return pauseEvent.target.errc } // listen     pausing. func (s activePausing) listen(p *Pipe, t target) (state, target) { return p.active(s, t) } // transition       . func (s activePausing) transition(p *Pipe, e eventMessage) (state, error) { switch e.event { case cancel: interrupt(p.cancel) err := Wait(p.errc) return nil, err case measure: e.params.applyTo(p.ID()) p.feedback = p.feedback.merge(e.params) return s, nil case push: e.params.applyTo(p.ID()) p.params = p.params.merge(e.params) return s, nil } return s, ErrInvalidState } // sendMessage   .   -, //      .    //    ,      .  , // ,   , : // 1.     // 2.      func (s activePausing) sendMessage(p *Pipe) state { m := p.newMessage() if len(m.feedback) == 0 { m.feedback = make(map[string][]phono.ParamFunc) } var wg sync.WaitGroup //     wg.Add(len(p.sinks)) //   Sink for _, sink := range p.sinks { param := phono.ReceivedBy(&wg, sink.ID()) // - m.feedback = m.feedback.add(param) } p.consume <- m //   wg.Wait() // ,     return paused } 

Assim que todos os destinatários receberem a mensagem, o pipeline paused estado de paused . Se a mensagem for a última, ocorrerá a transição para o estado ready .

De volta ao trabalho!

Para sair do estado de paused , chame p.Resume() .

 // Resume   resume  . //     pipe.Close  . func (p *Pipe) Resume() chan error { resumeEvent := eventMessage{ event: resume, target: target{ state: ready, errc: make(chan error, 1), }, } p.events <- resumeEvent return resumeEvent.target.errc } // listen     paused. func (s idlePaused) listen(p *Pipe, t target) (state, target) { return p.idle(s, t) } // transition       . func (s idlePaused) transition(p *Pipe, e eventMessage) (state, error) { switch e.event { case cancel: interrupt(p.cancel) err := Wait(p.errc) return nil, err case push: e.params.applyTo(p.ID()) p.params = p.params.merge(e.params) return s, nil case measure: for _, id := range e.components { e.params.applyTo(id) } return s, nil case resume: return running, nil } return s, ErrInvalidState } 

Tudo é trivial aqui, o pipeline novamente entra em estado de running .

Enrolar

O transportador pode ser parado de qualquer estado. Existe p.Close() .

 // Close   cancel  . //      . //    ,   . func (p *Pipe) Close() chan error { resumeEvent := eventMessage{ event: cancel, target: target{ state: nil, //   errc: make(chan error, 1), }, } p.events <- resumeEvent return resumeEvent.target.errc } 

Quem precisa disso?

Não é para todos. Para entender exatamente como gerenciar o estado, você precisa entender sua tarefa. Há exatamente duas circunstâncias em que você pode usar uma máquina assíncrona baseada em eventos:

1. Ciclo de vida complexo - há três ou mais estados com transições não lineares.
2. Execução assíncrona é usada.

Embora a máquina de eventos resolva o problema, é um padrão bastante complicado. Portanto, deve ser usado com muito cuidado e somente após um entendimento completo de todos os prós e contras.

Referências

• phono
• lexer Rob Pike