ruby on rails - ¿Ocurre la concurrencia incluso cuando solo hay un subproceso en un grupo de subprocesos?

ruby-on-rails multithreading

Question

Estoy usando Rails 5 y Ruby 2.4. ¿Cómo puedo averiguar, o puede decir, mirando a continuación, si hay varios subprocesos ejecutándose al mismo tiempo?

pool = Concurrent::FixedThreadPool.new(1)
promises = links.map do |link|
Concurrent::Promise.execute(executor: pool) do
result = process_link(link)
if result
if result.kind_of?(Array)
result.each do |my_obj|
my_obj.update_attributes({ :a => a })
records_processed = records_processed + my_obj.matches.count
end
else
records_processed = records_processed + result.matches.count
result.update_attributes({ :a => a })
end
end
end
end
promises.map(&:wait).map(&:value!)

Como configuré mi grupo a "1", mi suposición es que no se ejecuta nada simultáneamente, pero sigo recibiendo este error ...

Error during processing: (ActiveRecord::ConnectionTimeoutError) could not obtain a connection from the pool within 5.000 seconds (waited 5.002 seconds); all pooled connections were in use
/Users/nataliab/.rvm/gems/ruby-2.4.0@global/gems/activerecord-5.0.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:202:in `block in wait_poll''
/Users/nataliab/.rvm/gems/ruby-2.4.0@global/gems/activerecord-5.0.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:193:in `loop''
/Users/nataliab/.rvm/gems/ruby-2.4.0@global/gems/activerecord-5.0.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:193:in `wait_poll''
/Users/nataliab/.rvm/gems/ruby-2.4.0@global/gems/activerecord-5.0.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:154:in `internal_poll''
/Users/nataliab/.rvm/gems/ruby-2.4.0@global/gems/activerecord-5.0.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:278:in `internal_poll''
/Users/nataliab/.rvm/gems/ruby-2.4.0@global/gems/activerecord-5.0.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:148:in `block in poll''
/Users/nataliab/.rvm/rubies/ruby-2.4.0/lib/ruby/2.4.0/monitor.rb:214:in `mon_synchronize''
/Users/nataliab/.rvm/gems/ruby-2.4.0@global/gems/activerecord-5.0.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:158:in `synchronize''
/Users/nataliab/.rvm/gems/ruby-2.4.0@global/gems/activerecord-5.0.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:148:in `poll''
/Users/nataliab/.rvm/gems/ruby-2.4.0@global/gems/activerecord-5.0.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:717:in `acquire_connection''
/Users/nataliab/.rvm/gems/ruby-2.4.0@global/gems/activerecord-5.0.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:490:in `checkout''
/Users/nataliab/.rvm/gems/ruby-2.4.0@global/gems/activerecord-5.0.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:364:in `connection''
/Users/nataliab/.rvm/gems/ruby-2.4.0@global/gems/activerecord-5.0.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:883:in `retrieve_connection''
/Users/nataliab/.rvm/gems/ruby-2.4.0@global/gems/activerecord-5.0.1/lib/active_record/connection_handling.rb:128:in `retrieve_connection''

No obtengo el error anterior si modifico mi código para ejecutarlo donde estoy seguro de que no hay concurrencia ...

links.each do |link|
result = process_link(link)
if result
if result.kind_of?(Array)
result.each do |race|
my_obj.update_attributes({ :a => a })
records_processed = records_processed + my_obj.matches.count
end
else
records_processed = records_processed + result.matches.count
result.update_attributes({ :a => a })
end
end
end

Editar: Esta es la configuración de mi base de datos para mi entorno de desarrollo. También tenga en cuenta que todo esto se está ejecutando en la consola de rieles.

development:
adapter: postgresql
encoding: utf8
database: sims
username: postgres
password: password
pool: 5
timeout: 15000
host: 127.0.0.1

Answer 1

Dado que está definiendo el grupo de subprocesos fijos para tener un solo subproceso, asumo que no está logrando ningún tipo de concurrencia. Al observar su error, parece que el único subproceso disponible del grupo estuvo ocupado durante demasiado tiempo y causó la excepción de tiempo de espera de conexión.

Cuando modificó la implementación del código para que no contuviera un grupo de subprocesos, la aplicación fue explícitamente un solo subproceso sin la posibilidad de un tiempo de espera de conexión debido a la espera de subprocesos de un grupo. Intente aumentar el tamaño de su grupo de subprocesos (quizás a 3 o 5) y vea si todavía está recibiendo la misma excepción.

Answer 2

Su suposición de que varios subprocesos deben ejecutarse simultáneamente porque la agrupación de conexiones se está agotando no es correcta. El hecho de que una conexión todavía esté "desprotegida" en el grupo de conexiones no significa que una consulta se esté ejecutando actualmente en la conexión desprotegida en un subproceso, simplemente significa que la conexión del subproceso no se ha verificado de nuevo. El subproceso podría estar inactivo, pero aún aferrarse a una conexión desde el grupo de conexiones, siempre y cuando no haya finalizado explícitamente.

Dado que las Conexiones de ActiveRecord son de subprocesos locales, puede agotar el grupo de conexiones ejecutando consultas de ActiveRecord en varios subprocesos, como lo está haciendo en este caso. (Cada vez que se llama a Concurrent::FixedThreadPool.new(1) , se crea un nuevo subproceso.) Incluso si solo está ejecutando consultas en un solo subproceso a la vez, de forma predeterminada, la conexión se mantendrá abierta en cada subproceso hasta que se terminen.

Para evitar esto, puede verificar las conexiones manualmente después de usarlas, o asegurarse de que sus hilos terminen (se eliminen) para que sus conexiones puedan ser recuperadas por la agrupación.

• Para verificar manualmente las conexiones, consulte la documentación de ConnectionPool para conocer sus opciones. La forma más fácil es envolver su código ActiveRecord en un bloque with_connection :

  Concurrent::Promise.execute(executor: pool) do
  ActiveRecord::Base.connection_pool.with_connection do
  # update_attributes, etc
  end
  end

• Para asegurarse de que todos los subprocesos terminen, llame a #shutdown seguido de #wait_for_termination en el grupo de subprocesos una vez que haya terminado de usarlo:

  values = promises.map(&:value!)
  pool.shutdown
  pool.wait_for_termination

Answer 3

Usted asume que solo hay un hilo es incorrecto. Hay dos: el del grupo de subprocesos y el principal que generó el del grupo de subprocesos.

Podría confundirse al hacer que el subproceso principal esperara y no debería acceder a la base de datos. Eso no significa que todavía no tenga una conexión, por lo que evitará que el otro subproceso la adquiera.

Como regla general, el conjunto de conexiones de la base de datos debe configurarse al menos en el número de subprocesos generados + 1. En este caso - 2.

Código para reproducir fácilmente:

# migration
class CreateFoos < ActiveRecord::Migration[5.0]
def change
create_table :foos do |t|
t.integer :bar
end
end
end
# model
class Foo < ApplicationRecord
end
# rake task
task experiment: :environment do
Foo.create
pool = Concurrent::FixedThreadPool.new(1)
promise =
Concurrent::Promise.execute(executor: pool) do
Foo.first.update_attributes!(bar: rand(-42..42))
end
promise.wait.value!
end

Establezca pool en 1 en su config/database.yml y ejecute la tarea. Obtendrá un error. Establézcalo en 2 - estará bien.

Puede aumentar el número de subprocesos en el grupo y agregar al menos esa cantidad de promesas para el procesamiento. Siempre fallará para el conjunto de conexiones de la base de datos = número de subprocesos en el conjunto de subprocesos y tendrá éxito si agrega uno más en config/database.yml .