ruby on rails - Redis+ActionController:: Los hilos en vivo no están muriendo

ruby-on-rails multithreading

Question

Antecedentes: hemos creado una función de chat en una de nuestras aplicaciones Rails existentes. Estamos utilizando el nuevo módulo ActionController::Live y ejecutando Puma (con Nginx en producción) y suscribiéndonos a los mensajes a través de Redis. Estamos utilizando el lado del cliente de EventSource para establecer la conexión de forma asincrónica.

Resumen del problema: los hilos nunca se mueren cuando se termina la conexión.

Por ejemplo, si el usuario se va, cierra el navegador o incluso va a una página diferente dentro de la aplicación, se genera un nuevo hilo (como se esperaba), pero el anterior continúa vivo.

El problema, como lo veo ahora, es que cuando ocurre cualquiera de estas situaciones, el servidor no tiene forma de saber si la conexión en el extremo del navegador finaliza, hasta que algo intenta escribir en esta secuencia interrumpida, lo que nunca ocurriría una vez que el navegador se ha alejado de la página original.

Este problema parece estar documentado en github , y preguntas similares se hacen en StackOverflow aquí (la misma pregunta es bastante exacta) y aquí (en cuanto a obtener el número de hilos activos) .

La única solución que he podido encontrar, basada en estas publicaciones, es implementar un tipo de subproceso / conexión de póker. Intentar escribir en una conexión interrumpida genera un IOError que puedo capturar y cerrar la conexión correctamente, lo que permite que el hilo muera. Este es el código de controlador para esa solución:

def events
  response.headers["Content-Type"] = "text/event-stream"

  stream_error = false; # used by flusher thread to determine when to stop

  redis = Redis.new

  # Subscribe to our events
  redis.subscribe("message.create", "message.user_list_update") do |on| 
    on.message do |event, data| # when message is received, write to stream
      response.stream.write("messageType: ''#{event}'', data: #{data}/n/n")
    end

    # This is the monitor / connection poker thread
    # Periodically poke the connection by attempting to write to the stream
    flusher_thread = Thread.new do
      while !stream_error
        $redis.publish "message.create", "flusher_test"
        sleep 2.seconds
      end
    end
  end 

  rescue IOError
    logger.info "Stream closed"
    stream_error = true;
  ensure
    logger.info "Events action is quitting redis and closing stream!"
    redis.quit
    response.stream.close
end

(Nota: el método de events parece bloquearse en la invocación del método de subscribe . Todo lo demás (la transmisión) funciona correctamente, así que asumo que esto es normal).

(Otra nota: el concepto de hilo de enjuague tiene más sentido como un proceso de fondo de larga ejecución, un poco como un colector de hilo de basura. El problema con mi implementación anterior es que se genera un nuevo hilo para cada conexión, lo cual es inútil. intentar implementar este concepto debería hacerlo más como un único proceso, no tanto como lo he esbozado. Actualizaré esta publicación cuando lo vuelva a implementar correctamente como un proceso de fondo único).

La desventaja de esta solución es que solo hemos retrasado o disminuido el problema, no completamente resuelto. Todavía tenemos 2 hilos por usuario, además de otras solicitudes como ajax, lo que parece terrible desde una perspectiva de escalado; parece completamente inalcanzable y poco práctico para un sistema más grande con muchas conexiones simultáneas posibles.

Siento que me estoy perdiendo algo vital; Me resulta un tanto difícil de creer que Rails tiene una característica que está tan obviamente rota sin implementar un corrector de conexión personalizado como lo he hecho.

Pregunta: ¿Cómo permitimos que las conexiones / subprocesos mueran sin implementar algo cursi como un "enlace de conexión" o un recolector de subprocesos de basura?

Como siempre, avíseme si he olvidado algo.

Actualización Solo para agregar un poco de información adicional: Huetsch en github publicó este comentario señalando que SSE se basa en TCP, que normalmente envía un paquete FIN cuando la conexión está cerrada, dejando que el otro extremo (servidor en este caso) sepa que Es seguro cerrar la conexión. Huetsch señala que el navegador no está enviando ese paquete (¿quizás un error en la biblioteca de EventSource ?), O que Rails no lo está EventSource o está haciendo algo con él (definitivamente es un error en Rails, si ese es el caso). La búsqueda continúa ...

Otra actualización usando Wireshark, de hecho puedo ver paquetes FIN que se envían. Es cierto que no tengo mucho conocimiento o experiencia con el nivel de protocolo, sin embargo, por lo que puedo decir, definitivamente detecto que un paquete FIN se envía desde el navegador cuando establezco la conexión SSE usando EventSource desde el navegador, y NO se envía ningún paquete si eliminar esa conexión (lo que significa que no hay SSE). Aunque no estoy muy al tanto de mi conocimiento de TCP, esto parece indicarme que el cliente está terminando la conexión correctamente; quizás esto indica un error en Puma o Rails.

Otra actualización, @JamesBoutcher / boutcheratwest (github), me señaló una discusión en el sitio web de redis con respecto a este tema, específicamente con respecto al hecho de que el método de .(p)subscribe nunca se cierra. El póster de ese sitio señaló lo mismo que hemos descubierto aquí, que el entorno Rails nunca se notifica cuando la conexión del lado del cliente está cerrada y, por lo tanto, no puede ejecutar el método. .(p)unsubscribe . Pregunta sobre un tiempo de espera para el método de .(p)subscribe , que creo que también funcionaría, aunque no estoy seguro de qué método (el póker de conexión que he descrito anteriormente o su sugerencia de tiempo de espera) sería una solución mejor . Idealmente, para la solución de póker de conexión, me gustaría encontrar una manera de determinar si la conexión está cerrada en el otro extremo sin escribir en la transmisión. Como es ahora, como pueden ver, tengo que implementar un código del lado del cliente para manejar mi mensaje de "asentimiento" por separado, lo que creo que es intrincado y tonto como diablos.

Answer 1

Actualmente estoy haciendo una aplicación que gira en torno a ActionController: Live, EventSource y Puma y para aquellos que tienen problemas para cerrar transmisiones y tal, en lugar de rescatar un IOError , en Rails 4.2 necesita rescatar ClientDisconnected . Ejemplo:

def stream
  #Begin is not required
  twitter_client = Twitter::Streaming::Client.new(config_params) do |obj|
    # Do something
  end
rescue ClientDisconnected
  # Do something when disconnected
ensure
  # Do something else to ensure the stream is closed
end

Encontré este útil consejo de esta publicación en el foro (todo el camino en la parte inferior): http://railscasts.com/episodes/401-actioncontroller-live?view=comments

Answer 2

Aquí está la solución con tiempo de espera que saldrá del bloqueo de Redis. (P) suscriba la llamada y elimine la pisada de conexión no utilizada.

class Stream::FixedController < StreamController
  def events
    # Rails reserve a db connection from connection pool for
    # each request, lets put it back into connection pool.
    ActiveRecord::Base.clear_active_connections!

    # Last time of any (except heartbeat) activity on stream
    # it mean last time of any message was send from server to client
    # or time of setting new connection
    @last_active = Time.zone.now

    # Redis (p)subscribe is blocking request so we need do some trick
    # to prevent it freeze request forever.
    redis.psubscribe("messages:*", ''heartbeat'') do |on|
      on.pmessage do |pattern, event, data|
        # capture heartbeat from Redis pub/sub
        if event == ''heartbeat''
          # calculate idle time (in secounds) for this stream connection
          idle_time = (Time.zone.now - @last_active).to_i

          # Now we need to relase connection with Redis.(p)subscribe
          # chanel to allow go of any Exception (like connection closed)
          if idle_time > 4.minutes
            # unsubscribe from Redis because of idle time was to long
            # that''s all - fix in (almost)one line :)
            redis.punsubscribe
          end
        else
          # save time of this (last) activity
          @last_active = Time.zone.now
        end
        # write to stream - even heartbeat - it''s sometimes chance to
        # capture dissconection error before idle_time
        response.stream.write("event: #{event}/ndata: #{data}/n/n")
      end
    end
    # blicking end (no chance to get below this line without unsubscribe)
  rescue IOError
    Logs::Stream.info "Stream closed"
  rescue ClientDisconnected
    Logs::Stream.info "ClientDisconnected"
  rescue ActionController::Live::ClientDisconnected
    Logs::Stream.info "Live::ClientDisconnected"
  ensure
    Logs::Stream.info "Stream ensure close"
    redis.quit
    response.stream.close
  end
end

Tienes que usar rojos. (P) anular la suscripción para finalizar esta llamada de bloqueo. Ninguna excepción puede romper esto.

Mi aplicación simple con información sobre esta solución: https://github.com/piotr-kedziak/redis-subscribe-stream-puma-fix

Answer 3

Aquí hay una solución potencialmente más simple que no usa un latido del corazón. Después de mucha investigación y experimentación, aquí está el código que estoy usando con sinatra + sinatra sse gem (que se debe adaptar fácilmente a Rails 4):

class EventServer < Sinatra::Base
 include Sinatra::SSE
 set :connections, []
 .
 .
 .
 get ''/channel/:channel'' do
 .
 .
 .
  sse_stream do |out|
    settings.connections << out
    out.callback {
      puts ''Client disconnected from sse'';
      settings.connections.delete(out);
    }
  redis.subscribe(channel) do |on|
      on.subscribe do |channel, subscriptions|
        puts "Subscribed to redis ##{channel}/n"
      end
      on.message do |channel, message|
        puts "Message from redis ##{channel}: #{message}/n"
        message = JSON.parse(message)
        .
        .
        .
        if settings.connections.include?(out)
          out.push(message)
        else
          puts ''closing orphaned redis connection''
          redis.unsubscribe
        end
      end
    end
  end
end

La conexión de redis bloquea el mensaje y solo acepta (p) suscribe / (p) borra los comandos. Una vez que se da de baja, la conexión redis ya no está bloqueada y puede ser liberada por el objeto del servidor web que fue instanciado por la solicitud sse inicial. Se borra automáticamente cuando recibe un mensaje en redis y la conexión con el navegador ya no existe en el conjunto de recopilación.

Answer 4

En lugar de enviar un latido del corazón a todos los clientes, podría ser más fácil simplemente establecer un perro guardián para cada conexión. [Gracias a @NeilJewers]

class Stream::FixedController < StreamController
  def events
    # Rails reserve a db connection from connection pool for
    # each request, lets put it back into connection pool.
    ActiveRecord::Base.clear_active_connections!

    redis = Redis.new

    watchdog = Doberman::WatchDog.new(:timeout => 20.seconds)
    watchdog.start

    # Redis (p)subscribe is blocking request so we need do some trick
    # to prevent it freeze request forever.
    redis.psubscribe("messages:*") do |on|
      on.pmessage do |pattern, event, data|
        begin
          # write to stream - even heartbeat - it''s sometimes chance to
          response.stream.write("event: #{event}/ndata: #{data}/n/n")
          watchdog.ping

        rescue Doberman::WatchDog::Timeout => e
          raise ClientDisconnected if response.stream.closed?
          watchdog.ping
        end
      end
    end

  rescue IOError
  rescue ClientDisconnected

  ensure
    response.stream.close
    redis.quit
    watchdog.stop
  end
end

Answer 5

Sobre la base de @James Boutcher, utilicé lo siguiente en Puma agrupado con 2 trabajadores, por lo que tengo solo 1 hilo creado para el latido del corazón en config / initializers / redis.rb:

config / puma.rb

on_worker_boot do |index|
  puts "worker nb #{index.to_s} booting"
  create_heartbeat if index.to_i==0
end

def create_heartbeat
  puts "creating heartbeat"
  $redis||=Redis.new
  heartbeat = Thread.new do
    ActiveRecord::Base.connection_pool.release_connection
    begin
      while true
        hash={event: "heartbeat",data: "heartbeat"}
        $redis.publish("heartbeat",hash.to_json)
        sleep 20.seconds
      end
    ensure
      #no db connection anyway
    end
  end
end

Answer 6

Una solución que acabo de hacer (tomando prestado mucho de @teeg) que parece funcionar bien (aunque no lo haya probado en caso de falla)

config / initializers / redis.rb

$redis = Redis.new(:host => "xxxx.com", :port => 6379)

heartbeat_thread = Thread.new do
  while true
    $redis.publish("heartbeat","thump")
    sleep 30.seconds
  end
end

at_exit do
  # not sure this is needed, but just in case
  heartbeat_thread.kill
  $redis.quit
end

Y luego en mi controlador:

def events
    response.headers["Content-Type"] = "text/event-stream"
    redis = Redis.new(:host => "xxxxxxx.com", :port => 6379)
    logger.info "New stream starting, connecting to redis"
    redis.subscribe([''parse.new'',''heartbeat'']) do |on|
      on.message do |event, data|
        if event == ''parse.new''
          response.stream.write("event: parse/ndata: #{data}/n/n")
        elsif event == ''heartbeat''
          response.stream.write("event: heartbeat/ndata: heartbeat/n/n")
        end
      end
    end
  rescue IOError
    logger.info "Stream closed"
  ensure
    logger.info "Stopping stream thread"
    redis.quit
    response.stream.close
  end