MultithreaderMapRunner.
439022
private synchronized RecordWriter getRecordWriter(String namedOutput,
                                                    String baseFileName,
                                                    final Reporter reporter)
    throws IOException {
    RecordWriter writer = recordWriters.get(baseFileName);
    if (writer == null) {
      if (countersEnabled && reporter == null) {
        throw new IllegalArgumentException(
          "Counters are enabled, Reporter cannot be NULL");
      }
      JobConf jobConf = new JobConf(conf);
      jobConf.set(InternalFileOutputFormat.CONFIG_NAMED_OUTPUT, namedOutput);
      FileSystem fs = FileSystem.get(conf);
      writer =
        outputFormat.getRecordWriter(fs, jobConf, baseFileName, reporter);

      if (countersEnabled) {
        if (reporter == null) {
          throw new IllegalArgumentException(
            "Counters are enabled, Reporter cannot be NULL");
        }
        writer = new RecordWriterWithCounter(writer, baseFileName, reporter);
      }

      recordWriters.put(baseFileName, writer);
    }
    return writer;
  }
462891
public final void streamBufferWithGroups(Iterator<IN1> iterator1, Iterator<IN2> iterator2, Collector<OUT> c) {
		SingleElementPushBackIterator<IN1> i1 = new SingleElementPushBackIterator<>(iterator1);
		SingleElementPushBackIterator<IN2> i2 = new SingleElementPushBackIterator<>(iterator2);
		try {
			int size;
			if (i1.hasNext() || i2.hasNext()) {
				while (true) {
					int sig = in.readInt();
					switch (sig) {
						case SIGNAL_BUFFER_REQUEST_G0:
							if (i1.hasNext()) {
								size = sender.sendBuffer1(i1);
								sendWriteNotification(size, i1.hasNext());
							}
							break;
						case SIGNAL_BUFFER_REQUEST_G1:
							if (i2.hasNext()) {
								size = sender.sendBuffer2(i2);
								sendWriteNotification(size, i2.hasNext());
							}
							break;
						case SIGNAL_FINISHED:
							return;
						case SIGNAL_ERROR:
							try {
								outPrinter.join();
							} catch (InterruptedException e) {
								outPrinter.interrupt();
							}
							try {
								errorPrinter.join();
							} catch (InterruptedException e) {
								errorPrinter.interrupt();
							}
							throw new RuntimeException(
								"External process for task " + function.getRuntimeContext().getTaskName() + " terminated prematurely due to an error." + msg);
						default:
							receiver.collectBuffer(c, sig);
							sendReadConfirmation();
							break;
					}
				}
			}
		} catch (SocketTimeoutException ignored) {
			throw new RuntimeException("External process for task " + function.getRuntimeContext().getTaskName() + " stopped responding." + msg);
		} catch (Exception e) {
			throw new RuntimeException("Critical failure for task " + function.getRuntimeContext().getTaskName() + ". " + msg.get(), e);
		}
	}
912861
def sequential_apply(task, args, concurrent_tasks=cpu_count * 3,
                     weight=lambda item: 1, key=lambda item: 'Unspecified'):
    
    chunks = split_in_blocks(args[0], concurrent_tasks or 1, weight, key)
    task_args = [(ch,) + args[1:] for ch in chunks]
    return itertools.starmap(task, task_args)
443472
def during(f):
    """"""
    def decorator(g):
        @wraps(g)
        def h(*args, **kargs):
            tf = Thread(target=f, args=args, kwargs=kargs)
            tf.start()
            r = g(*args, **kargs)
            tf.join()
            return r
        return h
    return decorator
353785
@Nullable
    private ThreadPoolExecutor unwrapThreadPoolExecutor(ExecutorService executor, Class<?> wrapper) {
        try {
            Field e = wrapper.getDeclaredField("e");
            e.setAccessible(true);
            return (ThreadPoolExecutor) e.get(executorService);
        } catch (NoSuchFieldException | IllegalAccessException e) {
            // Do nothing. We simply can't get to the underlying ThreadPoolExecutor.
        }
        return null;
    }
528155
private Thread[] enumerateThreads() {
        boolean fits = false;
        int inc = 50;
        Thread[] threads = null;
        int nrThreads = 0;
        while (!fits) {
            try {
                threads = new Thread[Thread.activeCount()+inc];
                nrThreads = Thread.enumerate(threads);
                fits = true;
            } catch (ArrayIndexOutOfBoundsException exp) {
                inc += 50;
            }
        }
        // Trim array
        Thread ret[] = new Thread[nrThreads];
        System.arraycopy(threads,0,ret,0,nrThreads);
        return ret;
    }
512026
public static <E> Collection<E> createLazyCollectionFromStream(Supplier<Stream<E>> sequentialStreamSupplier, int size)
  {
    return new AbstractCollection<E>()
    {
      @Override
      public Iterator<E> iterator()
      {
        return sequentialStreamSupplier.get().iterator();
      }

      @Override
      public Spliterator<E> spliterator()
      {
        return sequentialStreamSupplier.get().spliterator();
      }

      @Override
      public Stream<E> stream()
      {
        return sequentialStreamSupplier.get();
      }

      @Override
      public Stream<E> parallelStream()
      {
        return sequentialStreamSupplier.get().parallel();
      }

      @Override
      public int size()
      {
        return size;
      }
    };
  }
1267351
private static ThreadFactory createThreadFactory(ThreadFactory threadFactory, String threadNamePrefix,
			Boolean daemon) {
		if (threadFactory != null) {
			return threadFactory;
		}

		if (threadNamePrefix != null) {
			if (daemon != null) {
				return ThreadPoolUtil.buildThreadFactory(threadNamePrefix, daemon);
			} else {
				return ThreadPoolUtil.buildThreadFactory(threadNamePrefix);
			}
		}

		return Executors.defaultThreadFactory();
	}
940904
public void prime() {
        ThreadInfo[] ti = threading.dumpAllThreads(false, false);
        long[] tids = new long[ti.length];
        int n = 0;
        for(ThreadInfo t:ti) {
            long tid = t.getThreadId();
            String name = t.getThreadName();
            if (threadFilter == null || threadFilter.accept(name)) {
                tids[n++] = tid;
            }
        }
        tids = Arrays.copyOf(tids, n);
        threadSet = tids;
	}
863870
public static <K, V> MutableMultimap<K, V> groupBy(
            Iterable<V> iterable,
            Function<? super V, ? extends K> function,
            int batchSize,
            Executor executor)
    {
        return ParallelIterate.groupBy(iterable, function, SynchronizedPutFastListMultimap.<K, V>newMultimap(), batchSize, executor);
    }
1150703
private static Thread doThread(Runnable runnable, String name, boolean deamon){
        String realname = getThreadName(name);
        Thread t = new Thread(runnable);
        t.setName(realname);
        t.setDaemon(deamon);
        return t;
    }
219637
def multiprocessing_ppmap_worker(in_queue, out_queue, func, fixed_arg):
    
    while True:
        i, var_arg = in_queue.get()
        if i is None:
            break
        out_queue.put((i, func(i, fixed_arg, var_arg)))
516104
def dynamicmap_memoization(callable_obj, streams):
    
    memoization_state = bool(callable_obj._stream_memoization)
    callable_obj._stream_memoization &= not any(s.transient and s._triggering for s in streams)
    try:
        yield
    except:
        raise
    finally:
        callable_obj._stream_memoization = memoization_state
389807
public Map<String, Object> forThread(Thread t) {
        return new HashMap<String, Object>(PER_THREAD_CONTEXTS.getUnchecked(t));
    }
831827
def _run_grid_multithread(self, func, iterables):
        ''' 
        '''
        
        f = lambda x: threading.Thread(target = func,args = (x,))
        threads = map(f, iterables)
        for thread in threads:
            thread.setDaemon(True)
            thread.start()
            thread.join()
825297
def thread_partition_array(x):
    ""
    n_threads = get_threadpool_size()
    if len(x.shape) > 1:
        maxind = x.shape[1]
    else:
        maxind = x.shape[0]
    bounds = np.array(np.linspace(0, maxind, n_threads + 1), dtype='int')
    cmin = bounds[:-1]
    cmax = bounds[1:]
    return cmin, cmax
926575
function writableServerSelector() {
  return function(topologyDescription, servers) {
    return latencyWindowReducer(topologyDescription, servers.filter(s => s.isWritable));
  };
}