Function to be spawned concurrently,
    consume data keys from input queue, and push the resulting dataframes to output map
329269
def data_worker(**kwargs):
    
    if kwargs is not None:
        if "function" in kwargs:
            function = kwargs["function"]
        else:
            Exception("Invalid arguments, no function specified")
        if "input" in kwargs:
            input_queue = kwargs["input"]
        else:
            Exception("Invalid Arguments, no input queue")
        if "output" in kwargs:
            output_map = kwargs["output"]
        else:
            Exception("Invalid Arguments, no output map")
        if "token" in kwargs:
            argsdict = {"quandl_token": kwargs["token"]}

        else:
            if "Quandl" in function.__module__:
                Exception("Invalid Arguments, no Quandl token")
        if ("source" and "begin" and "end") in kwargs:
            argsdict = {"data_source": kwargs["source"], "begin": kwargs["begin"], "end": kwargs["end"]}
        else:
            if "pandas.io.data" in function.__module__:
                Exception("Invalid Arguments, no pandas data source specified")
        if ("source" in kwargs) and (("begin" and "end") not in kwargs):
            argsdict = {"data_source": kwargs["source"]}
        else:
            if "pandas.io.data" in function.__module__:
                Exception("Invalid Arguments, no pandas data source specified")
    else:
        Exception("Invalid Arguments")

    retries = 5
    while not input_queue.empty():
        data_key = input_queue.get()
        get_data(function, data_key, output_map, retries, argsdict)
216678
def _map_or_starmap_async(function, iterable, args, kwargs, map_or_starmap):
    
    arg_newarg = (("parallel", "pm_parallel"), ("chunksize", "pm_chunksize"),
                  ("pool", "pm_pool"), ("processes", "pm_processes"),
                  ("callback", "pm_callback"),
                  ("error_callback", "pm_error_callback"))
    kwargs = _deprecated_kwargs(kwargs, arg_newarg)
    chunksize = kwargs.pop("pm_chunksize", None)
    callback = kwargs.pop("pm_callback", None)
    error_callback = kwargs.pop("pm_error_callback", None)
    parallel, pool, close_pool = _create_pool(kwargs)
    # Map:
    if parallel:
        func_star = _get_helper_func(map_or_starmap)
        try:
            if sys.version_info[0] == 2:  # does not support error_callback
                result = pool.map_async(func_star,
                                        izip(repeat(function),
                                             iterable,
                                             repeat(list(args)),
                                             repeat(kwargs)),
                                        chunksize, callback)
            else:
                result = pool.map_async(func_star,
                                        izip(repeat(function),
                                             iterable,
                                             repeat(list(args)),
                                             repeat(kwargs)),
                                        chunksize, callback, error_callback)
        finally:
            if close_pool:
                pool.close()
                result = _ParallelAsyncResult(result, pool)
            else:
                result = _ParallelAsyncResult(result)
    else:
        values = _serial_map_or_starmap(function, iterable, args, kwargs,
                                        False, map_or_starmap)
        result = _DummyAsyncResult(values)
    return result
688039
def _create_tensor_dicts(input_queue: Queue,
                         output_queue: Queue,
                         iterator: DataIterator,
                         shuffle: bool,
                         index: int) -> None:
    
    def instances() -> Iterator[Instance]:
        instance = input_queue.get()
        while instance is not None:
            yield instance
            instance = input_queue.get()

    for tensor_dict in iterator(instances(), num_epochs=1, shuffle=shuffle):
        output_queue.put(tensor_dict)

    output_queue.put(index)
947039
def run(input=sys.stdin, output=sys.stdout):
    
    """
    functions = []
    environments = dict()

    def _writejson(obj):
        obj = json.encode(obj)
        if isinstance(obj, unicode):
            obj = obj.encode('utf-8')
        output.write(obj)
        output.write('\n')
        output.flush()

    def _log(message):
        if not isinstance(message, basestring):
            message = json.encode(message)
        _writejson({'log': message})

    def reset(config=None):
        del functions[:]
        return True

    def add_fun(string):
        string = BOM_UTF8 + string.encode('utf-8')
        try:
            function = _compile(string, "map_compilation_error", "map")
        except CompileError as e:
            return e.args[0]
        functions.append(function)
        return True

    def map_doc(doc):
        results = []
        for function in functions:
            try:
                results.append([[key, value] for key, value in function(doc)])
            except Exception, e:
                log.error('runtime error in map function: %s', e,
                          exc_info=True)
                results.append([])
                _log(traceback.format_exc())
        return results

    def reduce(*cmd, **kwargs):
        code = BOM_UTF8 + cmd[0][0].encode('utf-8')
        args = cmd[1]

        try:
            function = _compile(code, "reduce_compilation_error", "reduce")
        except CompileError as e:
            return e.args[0]

        rereduce = kwargs.get('rereduce', False)
        results = []
        if rereduce:
            keys = None
            vals = args
        else:
            if args:
                keys, vals = zip(*args)
            else:
                keys = []
                vals = []
        if function.func_code.co_argcount == 3:
            results = function(keys, vals, rereduce)
        else:
            results = function(keys, vals)
        return [True, [results]]

    def rereduce(*cmd):
        # Note: weird kwargs is for Python 2.5 compat
        return reduce(*cmd, **{'rereduce': True})

    def ddoc_new(name, doc):
        env = dict()
        # key in design doc -> expected function name
        method_mapping = dict(filters='filter', views='map')
        for key, section in doc.iteritems():
            if key not in method_mapping:
                continue
            env[key] = dict()
            id_error = "%s_compilation_error" % (key)
            for f_name, func_str in section.iteritems():
                if isinstance(func_str, dict):
                    env[key][f_name] = dict()
                    for expected_name, code in func_str.iteritems():
                        try:
                            env[key][f_name][expected_name] = _compile(
                                code, id_error, expected_name)
                        except CompileError as e:
                            return e.args[0]
                else:
                    try:
                        expected_name = method_mapping[key]
                        env[key][f_name] = _compile(func_str, id_error,
                                                    expected_name)
                    except CompileError as e:
                        return e.args[0]
        environments[name] = env
        return True

    def ddoc_exec(ddoc_name, ref, args):
        e = environments[ddoc_name]
        for r in ref:
            e = e[r]
        function = e
        h = handlers[ref[0]]

        return h(*args, **{"function": function})

    def ddoc(*args):
        args = list(args)
        n = args.pop(0)
        if n == 'new':
            return ddoc_new(*args)
        return ddoc_exec(n, args.pop(0), args[0])

    def filter(rows, request, dbinfo=None, function=None):
        request['db'] = dbinfo
        results = []

        for row in rows:
            r = function(row, request)
            if r is True:
                results.append(True)
            else:
                results.append(False)

        return [True, results]

    def filter_view(rows, function=None):
        results = []

        for row in rows:
            r = function(row)
            try:
                r.next()
                results.append(True)
            except StopIteration:
                results.append(False)

        return [True, results]

    def _compile(func_str, error_id, f_name):
        globals_ = {}

        if func_str in ('_count', '_sum', '_stats'):
            # These are builtins which are not ment to be complited
            # by the python server
            return func_str

        try:
            exec func_str in {'log': _log}, globals_
        except Exception, e:
            log.error('runtime error in filter function: %s', e,
                      exc_info=True)
            raise CompileError({'error': {
                'id': error_id,
                'reason': e.args[0]}})
        err = {'error': {
            'id': error_id,
            'reason': ('string must eval to a function named %(f_name)s '
                       '(ex: "def %(f_name)s(doc, request): return True")' %
                       dict(f_name=f_name))}}
        if f_name not in globals_:
            raise CompileError(err)
        function = globals_.pop(f_name)
        if type(function) is not FunctionType:
            raise CompileError(err)
        # merge in remaining globals so that they are available in the function
        function.func_globals.update(globals_)
        return function


    handlers = {'reset': reset, 'add_fun': add_fun, 'map_doc': map_doc,
                'reduce': reduce, 'rereduce': rereduce, 'ddoc': ddoc,
                'filters': filter, 'views': filter_view}

    try:
        while True:
            line = input.readline()
            if not line:
                break
            try:
                cmd = json.decode(line)
                log.debug('Processing %r', cmd)
            except ValueError, e:
                log.error('Error: %s', e, exc_info=True)
                return 1
            else:
                retval = handlers[cmd[0]](*cmd[1:])
                log.debug('Returning  %r', retval)
                _writejson(retval)
    except KeyboardInterrupt:
        return 0
    except Exception, e:
        log.error('Error: %s', e, exc_info=True)
        return 1
606491
def serving_input_fn(self, hparams, decode_hparams=None, use_tpu=False):
    """"""
    mode = tf.estimator.ModeKeys.PREDICT
    serialized_example = tf.placeholder(
        dtype=tf.string, shape=[None], name="serialized_example")
    dataset = tf.data.Dataset.from_tensor_slices(serialized_example)
    dataset = dataset.map(self.decode_example)
    dataset = dataset.map(lambda ex: self.preprocess_example(ex, mode, hparams))
    dataset = dataset.map(data_reader.cast_ints_to_int32)

    if use_tpu:
      padded_shapes = data_reader.pad_for_tpu(dataset.output_shapes, hparams,
                                              hparams.max_length)
      batch_size = 1 if not decode_hparams else getattr(decode_hparams,
                                                        "batch_size", 1)
      dataset = dataset.padded_batch(
          batch_size, padded_shapes, drop_remainder=False)
      dataset = dataset.map(
          functools.partial(data_reader.pad_batch, batch_multiple=batch_size))
    else:
      dataset = dataset.padded_batch(
          tf.shape(serialized_example, out_type=tf.int64)[0],
          dataset.output_shapes)

    dataset = dataset.map(data_reader.standardize_shapes)
    features = tf.data.experimental.get_single_element(dataset)

    if self.has_inputs:
      features.pop("targets", None)

    return tf.estimator.export.ServingInputReceiver(
        features=features, receiver_tensors=serialized_example)
1038289
def consume_keys_asynchronous_threads(self):
        
        print("\nLooking up " + self.input_queue.qsize().__str__() + " keys from " + self.source_name + "\n")
        jobs = multiprocessing.cpu_count()*4 if (multiprocessing.cpu_count()*4 < self.input_queue.qsize()) \
            else self.input_queue.qsize()

        pool = ThreadPool(jobs)

        for x in range(jobs):
            pool.apply(self.data_worker, [], self.worker_args)

        pool.close()
        pool.join()
1374912
def pool(data, batch_size, key, batch_size_fn=lambda new, count, sofar: count,
         random_shuffler=None, shuffle=False, sort_within_batch=False):
    
    if random_shuffler is None:
        random_shuffler = random.shuffle
    for p in batch(data, batch_size * 100, batch_size_fn):
        p_batch = batch(sorted(p, key=key), batch_size, batch_size_fn) \
            if sort_within_batch \
            else batch(p, batch_size, batch_size_fn)
        if shuffle:
            for b in random_shuffler(list(p_batch)):
                yield b
        else:
            for b in list(p_batch):
                yield b
523804
def _thread_worker_fn(samples, batchify_fn, dataset):
    """"""
    if isinstance(samples[0], (list, tuple)):
        batch = [batchify_fn([dataset[i] for i in shard]) for shard in samples]
    else:
        batch = batchify_fn([dataset[i] for i in samples])
    return batch
545763
function processRowBuffer()
  {
    // get the row to add to the read queue
    var row = rowBuffer[rowIndex++];

    // if we just read the last row in the row buffer, clear the row buffer and
    // reset the row index so that we load the next chunk in the result stream
    // when _read() is called
    if (rowIndex === rowBuffer.length)
    {
      rowBuffer = null;
      rowIndex = 0;
    }

    // initialize the columns and column-related maps if necessary
    if (!columns)
    {
      columns = statement.getColumns();
    }
    if (!mapColumnIdToExtractFnName)
    {
      mapColumnIdToExtractFnName =
          buildMapColumnExtractFnNames(columns, fetchAsString);
    }

    // add the next row to the read queue
    process.nextTick(function()
    {
      self.push(externalizeRow(row, columns, mapColumnIdToExtractFnName));
    });
  }
862782
def sg_producer_func(func):
    
    """
    @wraps(func)
    def wrapper(**kwargs):
        r"""Manages arguments of `tf.sg_opt`.

        Args:
          **kwargs:
            source: A source queue list to enqueue
            dtypes: Input data types of each tensor
            out_dtypes: Output data types of each tensor ( If None, same as dtypes )
            capacity: Queue capacity. Default is 32.
            num_threads: Number of threads. Default is 1.
        """

        # default option
        opt = tf.sg_opt(kwargs) + tf.sg_opt(dtypes=[tf.sg_floatx], capacity=32, num_threads=1)

        # source queue list check
        assert opt.source is not None, 'source is mandatory.'
        if type(opt.source) is not list and type(opt.source) is not tuple:
            opt.source = [opt.source]
        if type(opt.dtypes) is not list and type(opt.dtypes) is not tuple:
            opt.dtypes = [opt.dtypes]
        # default out_dtypes
        if opt.out_dtypes is None:
            opt.out_dtypes = opt.dtypes
        if type(opt.out_dtypes) is not list and type(opt.out_dtypes) is not tuple:
            opt.out_dtypes = [opt.out_dtypes]
        assert len(opt.source) == len(opt.dtypes), 'Source and dtypes should have same length.'

        # enqueue function
        def enqueue_func(sess, op):
            # read data from source queue
            data = func(sess.run(opt.source))
            # create feeder dict
            feed_dict = {}
            for ph, col in zip(placeholders, data):
                feed_dict[ph] = col
            # run session
            sess.run(op, feed_dict=feed_dict)

        # create place holder list
        placeholders = []
        for dtype in opt.dtypes:
            placeholders.append(tf.placeholder(dtype=dtype))

        # create FIFO queue
        queue = tf.FIFOQueue(opt.capacity, dtypes=opt.out_dtypes)

        # enqueue operation
        enqueue_op = queue.enqueue(placeholders)

        # create queue runner
        runner = _FuncQueueRunner(enqueue_func, queue, [enqueue_op] * opt.num_threads)

        # register to global collection
        tf.train.add_queue_runner(runner)

        # return de-queue operation
        return queue.dequeue()

    return wrapper
199934
def _process_batch(self, param, dataframe):
        
        now = time.time()
        if param.eval_metric is not None:
            metrics = dict(param.eval_metric.get_name_value())
            param.eval_metric.reset()
        else:
            metrics = {}
        # #11504
        try:
            speed = self.frequent / (now - self.last_time)
        except ZeroDivisionError:
            speed = float('inf')
        metrics['batches_per_sec'] = speed * self.batch_size
        metrics['records_per_sec'] = speed
        metrics['elapsed'] = self.elapsed()
        metrics['minibatch_count'] = param.nbatch
        metrics['epoch'] = param.epoch
        self.append_metrics(metrics, dataframe)
        self.last_time = now
550260
def _spawn_threads(self):
        '''
        
        '''
        for devname,pdev in self._pcaps.items():
            t = threading.Thread(target=LLNetReal._low_level_dispatch, args=(pdev, devname, self._pktqueue))
            t.start()
            self._threads.append(t)
228118
def run_to_queue(self, queue, conn, options=None, dsquery=None):
    """"""
    try:
      multiquery = self._maybe_multi_query()
      if multiquery is not None:
        yield multiquery.run_to_queue(queue, conn, options=options)
        return

      if dsquery is None:
        dsquery = self._get_query(conn)
      rpc = dsquery.run_async(conn, options)
      while rpc is not None:
        batch = yield rpc
        if (batch.skipped_results and
            datastore_query.FetchOptions.offset(options)):
          offset = options.offset - batch.skipped_results
          options = datastore_query.FetchOptions(offset=offset, config=options)
        rpc = batch.next_batch_async(options)
        for i, result in enumerate(batch.results):
          queue.putq((batch, i, result))
      queue.complete()

    except GeneratorExit:
      raise
    except Exception:
      if not queue.done():
        _, e, tb = sys.exc_info()
        queue.set_exception(e, tb)
      raise
390023
def _multi_rpush_pipeline(self, pipe, queue, values, bulk_size=0):
        ''' 
        '''
        cont = 0
        for value in values:
            pipe.rpush(queue, value)
            if bulk_size != 0 and cont % bulk_size == 0:
                pipe.execute()
920064
def Run(self):
    """"""
    count = 0
    for count, input_data in enumerate(self.GetInput()):
      if count % 2000 == 0:
        logging.debug("%d processed.", count)
      args = (input_data, self.out_queue, self.token)
      self.thread_pool.AddTask(
          target=self.IterFunction, args=args, name=self.THREAD_POOL_NAME)

    while count >= 0:
      try:
        # We only use the timeout to wait if we got to the end of the Queue but
        # didn't process everything yet.
        out = self.out_queue.get(timeout=self.QUEUE_TIMEOUT, block=True)
        if out:
          yield out
          count -= 1
      except queue.Empty:
        break

    # Join and stop to clean up the threadpool.
    self.thread_pool.Stop(join_timeout=THREADPOOL_JOIN_TIMEOUT)
696389
def _distribute_save_datasets(self, scenes_iter, client, batch_size=1, **kwargs):
        """"""
        def load_data(q):
            idx = 0
            while True:
                future_list = q.get()
                if future_list is None:
                    break

                # save_datasets shouldn't be returning anything
                for future in future_list:
                    future.result()
                    log.info("Finished saving %d scenes", idx)
                    idx += 1
                q.task_done()

        input_q = Queue(batch_size if batch_size is not None else 1)
        load_thread = Thread(target=load_data, args=(input_q,))
        load_thread.start()

        for scene in scenes_iter:
            delayed = scene.save_datasets(compute=False, **kwargs)
            if isinstance(delayed, (list, tuple)) and len(delayed) == 2:
                # TODO Make this work for (source, target) datasets
                # given a target, source combination
                raise NotImplementedError("Distributed save_datasets does not support writers "
                                          "that return (source, target) combinations at this time. Use "
                                          "the non-distributed save_datasets instead.")
            future = client.compute(delayed)
            input_q.put(future)
        input_q.put(None)

        log.debug("Waiting for child thread to get saved results...")
        load_thread.join()
        log.debug("Child thread died successfully")
354697
def processpool_map(task, args, message, concurrency, batchsize=1, nargs=None):
    
    njobs = get_njobs(nargs, args)
    show_progress = bool(message)
    batches = grouper(batchsize, tupleise(args))
    def batched_task(*batch):
        return [task(*job) for job in batch]

    if show_progress:
        message += ' (PP:{}w:{}b)'.format(concurrency, batchsize)
        pbar = setup_progressbar(message, njobs, simple_progress=True)
        pbar.start()
    
    q_in   = multiprocessing.Queue()  # Should I limit either queue size? Limiting in-queue
    q_out  = multiprocessing.Queue()  # increases time taken to send jobs, makes pbar less useful

    proc = [multiprocessing.Process(target=fun, args=(batched_task, q_in, q_out)) for _ in range(concurrency)]
    for p in proc:
        p.daemon = True
        p.start()
    sent = [q_in.put((i, x)) for (i, x) in enumerate(batches)]
    [q_in.put((None, None)) for _ in range(concurrency)]
    res = []
    completed_count = 0
    for _ in range(len(sent)):
        result = get_from_queue(q_out)
        res.append(result)
        completed_count += len(result[1])
        if show_progress:
            pbar.update(completed_count)

    [p.join() for p in proc]
    if show_progress:
        pbar.finish()

    return flatten_list([x for (i, x) in sorted(res)])