model.rb in mixed_gauge-1.2.0.beta1

- old
+ new

@@ -22,19 +22,29 @@
     included do
       class_attribute :cluster_routing, instance_writer: false
       class_attribute :shard_repository, instance_writer: false
       class_attribute :distkey, instance_writer: false
       class_attribute :replication_mapping, instance_writer: false
+      class_attribute :service, instance_writer: false
     end
 
     module ClassMethods
       # The cluster config must be defined before `use_cluster`.
       # @param [Symbol] name A cluster name which is set by MixedGauge.configure
       def use_cluster(name)
         config = MixedGauge.config.fetch_cluster_config(name)
         self.cluster_routing = MixedGauge::Routing.new(config)
         self.shard_repository = MixedGauge::ShardRepository.new(config, self)
+        thread_size = (shard_repository.all.size * 100)
+        self.service = Expeditor::Service.new(
+          executor: Concurrent::ThreadPoolExecutor.new(
+            min_threads: thread_size,
+            max_threads: thread_size,
+            max_queue: shard_repository.all.size,
+            fallback_policy: :abort,
+          )
+        )
         self.abstract_class = true
       end
 
       # Distkey is a column. mixed_gauge hashes that value and determine which
       # shard to store.
@@ -120,10 +130,10 @@
 
       # @return [Mixedgauge::AllShardsInParallel]
       # @example
       #   User.all_shards_in_parallel.map {|m| m.where.find_by(name: 'Alice') }.compact
       def all_shards_in_parallel
-        AllShardsInParallel.new(all_shards)
+        AllShardsInParallel.new(all_shards, service: service)
       end
       alias_method :parallel, :all_shards_in_parallel
 
       # See example definitions in `spec/models.rb`.
       # @param [Symbol] A role name of target cluster.