Create a single-threaded grouper and accumulator.
966732
public static Pair<Grouper<RowBasedKey>, Accumulator<AggregateResult, Row>> createGrouperAccumulatorPair(
      final GroupByQuery query,
      final boolean isInputRaw,
      final Map<String, ValueType> rawInputRowSignature,
      final GroupByQueryConfig config,
      final Supplier<ByteBuffer> bufferSupplier,
      final LimitedTemporaryStorage temporaryStorage,
      final ObjectMapper spillMapper,
      final AggregatorFactory[] aggregatorFactories,
      final int mergeBufferSize
  )
  {
    return createGrouperAccumulatorPair(
        query,
        isInputRaw,
        rawInputRowSignature,
        config,
        bufferSupplier,
        null,
        SINGLE_THREAD_CONCURRENCY_HINT,
        temporaryStorage,
        spillMapper,
        aggregatorFactories,
        null,
        UNKNOWN_THREAD_PRIORITY,
        false,
        UNKNOWN_TIMEOUT,
        mergeBufferSize
    );
  }
28168
def run_in_thread_pool(self, pool_size=None, func=None, *args):
        """"""
        executor = Pool(pool_size)
        return self.loop.run_in_executor(executor, func, *args)
20812
def xgroup_create(self, stream, group_name, latest_id='$', mkstream=False):
        """"""
        args = [b'CREATE', stream, group_name, latest_id]
        if mkstream:
            args.append(b'MKSTREAM')
        fut = self.execute(b'XGROUP', *args)
        return wait_ok(fut)
948546
function DaisyChain(workers, concurrency) {
  var self = this
      , queues = this.queues = new Array(workers.length);

  // create queues and inject unidirectional linked list marker.
  var last;
  for (var i = 0; i < workers.length; i++) {
    var queue = queues[i] = async.queue(workers[i], concurrency);
    last = queue;
  }

  // add hooks for empty and drain
  last.empty = function() {
    if (self.empty) {
      async.all(queues, function(q, cb){ cb(q.length() === 0); }, function(yes) { if (yes) self.empty(); });
    }
  }

  last.drain = function() {
    if (self.drain) {
      async.every(queues, function(q, cb){ cb((q.length()+q.running()) === 0); }, function(yes) { if (yes) self.drain(); });
    }
  }
}
576178
def grow(self, amount):
        """"""
        if self.max > 0:
            budget = max(self.max - len(self._threads), 0)
        else:
            # self.max <= 0 indicates no maximum
            budget = float('inf')

        n_new = min(amount, budget)

        workers = [self._spawn_worker() for i in range(n_new)]
        while not all(worker.ready for worker in workers):
            time.sleep(.1)
        self._threads.extend(workers)
1082165
def grouper(iterable, n):
    
    it = iter(iterable)
    while True:
        chunk = itertools.islice(it, n)
        try:
            first = next(chunk)
        except StopIteration:
            return

        yield itertools.chain([first], chunk)
1199686
private SimpleConsumerThread<T> createAndStartSimpleConsumerThread(
			List<KafkaTopicPartitionState<TopicAndPartition>> seedPartitions,
			Node leader,
			ExceptionProxy errorHandler) throws IOException, ClassNotFoundException {
		// each thread needs its own copy of the deserializer, because the deserializer is
		// not necessarily thread safe
		final KafkaDeserializationSchema<T> clonedDeserializer =
				InstantiationUtil.clone(deserializer, runtimeContext.getUserCodeClassLoader());

		// seed thread with list of fetch partitions (otherwise it would shut down immediately again
		SimpleConsumerThread<T> brokerThread = new SimpleConsumerThread<>(
				this, errorHandler, kafkaConfig, leader, seedPartitions, unassignedPartitionsQueue,
				clonedDeserializer, invalidOffsetBehavior);

		brokerThread.setName(String.format("SimpleConsumer - %s - broker-%s (%s:%d)",
				runtimeContext.getTaskName(), leader.id(), leader.host(), leader.port()));
		brokerThread.setDaemon(true);
		brokerThread.start();

		LOG.info("Starting thread {}", brokerThread.getName());
		return brokerThread;
	}
235727
def queue_runner(concurrency, delay, retry_count, progress)
      queue = ChimpQueue.instance
      queue.max_threads = concurrency
      queue.delay = delay
      queue.retry_count = retry_count
      total_queue_size = queue.size

      puts "Executing..." unless progress or not quiet
      pbar = ProgressBar.new("Executing", 100) if progress
      queue.start

      queue.wait_until_done(@group) do
        pbar.set(((total_queue_size.to_f - queue.size.to_f)/total_queue_size.to_f*100).to_i) if progress
      end

      pbar.finish if progress
    end
226287
def grouper(iterable, n, fillvalue=None):
    
    if isinstance(iterable, int):
        warnings.warn(
            "grouper expects iterable as first parameter",
            DeprecationWarning,
        )
        n, iterable = iterable, n
    args = [iter(iterable)] * n
    return zip_longest(fillvalue=fillvalue, *args)
414554
private void createAccumulators(final OnDemandStatsProducer<S> producer, final Class producerClass, AccumulateWithSubClasses... annotations) {
		for (final AccumulateWithSubClasses annotation : annotations)
			if (annotation != null)
				AccumulatorUtil.getInstance(producerClass).createAccumulator(producer, annotation);
	}
1066694
def threaded(
    name: typing.Callable[..., typing.Any], daemon: bool = False, started: bool = False
) -> typing.Callable[..., threading.Thread]:
    """"""
1326087
def _create_threads(self):
        

        creator = JobCreator(
            self.config,
            self.observers.jobs,
            self.logger
        )
        self.jobs = creator.job_factory()
1249342
def build_from_queue(cls, input_queue, replay_size, batch_size):
    
    return cls(
        lambda: input_queue.dequeue_many(batch_size),
        replay_size,
        batch_size=batch_size)
596590
def _general_multithread(func):
    

    def multithread(templates, stream, *args, **kwargs):
        with pool_boy(ThreadPool, len(stream), **kwargs) as pool:
            return _pool_normxcorr(templates, stream, pool=pool, func=func)

    return multithread
74704
def spawn_thread(self):
        ''
        self.thread = threading.Thread(target=self.run, args=(False,))
        self.thread.daemon = True
        self.thread.start()
998694
def create_batch(cls, size, **kwargs):
        
        return [cls.create(**kwargs) for _ in range(size)]
521862
def run(self):
        """"""
        grouping = Grouping(group_by=lambda x: (x.namespace, x.operation,
                                                x.pattern))
        logfile = self.mloginfo.logfile

        if logfile.start and logfile.end:
            progress_start = self.mloginfo._datetime_to_epoch(logfile.start)
            progress_total = (self.mloginfo._datetime_to_epoch(logfile.end) -
                              progress_start)
        else:
            self.mloginfo.progress_bar_enabled = False

        for i, le in enumerate(logfile):
            # update progress bar every 1000 lines
            if self.mloginfo.progress_bar_enabled and (i % 1000 == 0):
                if le.datetime:
                    progress_curr = self.mloginfo._datetime_to_epoch(le
                                                                     .datetime)
                    if progress_total:
                        (self.mloginfo
                         .update_progress(float(progress_curr -
                                                progress_start) /
                                          progress_total))

            if (le.operation in ['query', 'getmore', 'update', 'remove'] or
                    le.command in ['count', 'findandmodify',
                                   'geonear', 'find']):
                lt = LogTuple(namespace=le.namespace, operation=op_or_cmd(le),
                              pattern=le.pattern, duration=le.duration)
                grouping.add(lt)

        grouping.sort_by_size()

        # clear progress bar again
        if self.mloginfo.progress_bar_enabled:
            self.mloginfo.update_progress(1.0)

        # no queries in the log file
        if len(grouping) < 1:
            print('no queries found.')
            return

        titles = ['namespace', 'operation', 'pattern', 'count', 'min (ms)',
                  'max (ms)', 'mean (ms)', '95%-ile (ms)', 'sum (ms)']
        table_rows = []

        for g in grouping:
            # calculate statistics for this group
            namespace, op, pattern = g

            group_events = [le.duration for le in grouping[g]
                            if le.duration is not None]

            stats = OrderedDict()
            stats['namespace'] = namespace
            stats['operation'] = op
            stats['pattern'] = pattern
            stats['count'] = len(group_events)
            stats['min'] = min(group_events) if group_events else '-'
            stats['max'] = max(group_events) if group_events else '-'
            stats['mean'] = 0
            if np:
                stats['95%'] = (np.percentile(group_events, 95)
                                if group_events else '-')
            else:
                stats['95%'] = 'n/a'
            stats['sum'] = sum(group_events) if group_events else '-'
            stats['mean'] = (stats['sum'] / stats['count']
                             if group_events else '-')

            if self.mloginfo.args['verbose']:
                stats['example'] = grouping[g][0]
                titles.append('example')

            table_rows.append(stats)

        # sort order depending on field names
        reverse = True
        if self.mloginfo.args['sort'] in ['namespace', 'pattern']:
            reverse = False

        table_rows = sorted(table_rows,
                            key=itemgetter(self.mloginfo.args['sort']),
                            reverse=reverse)
        print_table(table_rows, titles, uppercase_headers=False)
        print('')