fix: reset a concurrent-worker instance when a fork is detected, but the pooled model is not production ready and is not used yet (#264)

Author: supercaracal

lib/redis_client/cluster/concurrent_worker/pooled.rb

@@ -1,23 +1,23 @@
 # frozen_string_literal: true
 
+require 'redis_client/pid_cache'
+
 class RedisClient
   class Cluster
     module ConcurrentWorker
-      # This class is just an experimental implementation. There are some bugs for race condition.
+      # This class is just an experimental implementation.
       # Ruby VM allocates 1 MB memory as a stack for a thread.
       # It is a fixed size but we can modify the size with some environment variables.
       # So it consumes memory 1 MB multiplied a number of workers.
       class Pooled
         def initialize
-          @q = Queue.new
-          size = ::RedisClient::Cluster::ConcurrentWorker::MAX_WORKERS
-          size = size.positive? ? size : 5
-          @workers = Array.new(size)
+          setup
         end
 
         def new_group(size:)
           raise ArgumentError, "size must be positive: #{size} given" unless size.positive?
 
+          reset if @pid != ::RedisClient::PIDCache.pid
           ensure_workers if @workers.first.nil?
           ::RedisClient::Cluster::ConcurrentWorker::Group.new(worker: self, size: size)
         end
@@ -31,6 +31,7 @@ def close
           @workers.each { |t| t&.exit }
           @workers.clear
           @q.close
+          @pid = nil
           nil
         end
 
@@ -47,6 +48,19 @@ def spawn_worker
             loop { q.pop.exec }
           end
         end
+
+        def setup
+          @q = Queue.new
+          size = ::RedisClient::Cluster::ConcurrentWorker::MAX_WORKERS
+          size = size.positive? ? size : 5
+          @workers = Array.new(size)
+          @pid = ::RedisClient::PIDCache.pid
+        end
+
+        def reset
+          close
+          setup
+        end
       end
     end
   end