Run the UDF multi-threaded using python multiprocessing
549092
def _apply_mt(self, doc_loader, parallelism, **kwargs):
        """"""
        if not Meta.postgres:
            raise ValueError("Fonduer must use PostgreSQL as a database backend.")

        def fill_input_queue(in_queue, doc_loader, terminal_signal):
            for doc in doc_loader:
                in_queue.put(doc)
            in_queue.put(terminal_signal)

        # Create an input queue to feed documents to UDF workers
        manager = Manager()
        in_queue = manager.Queue()
        # Use an output queue to track multiprocess progress
        out_queue = JoinableQueue()

        total_count = len(doc_loader)

        # Start UDF Processes
        for i in range(parallelism):
            udf = self.udf_class(
                in_queue=in_queue,
                out_queue=out_queue,
                worker_id=i,
                **self.udf_init_kwargs,
            )
            udf.apply_kwargs = kwargs
            self.udfs.append(udf)

        # Start the UDF processes, and then join on their completion
        for udf in self.udfs:
            udf.start()

        # Fill input queue with documents
        terminal_signal = UDF.QUEUE_CLOSED
        in_queue_filler = Process(
            target=fill_input_queue, args=(in_queue, doc_loader, terminal_signal)
        )
        in_queue_filler.start()

        count_parsed = 0
        while count_parsed < total_count:
            y = out_queue.get()
            # Update progress bar whenever an item has been processed
            if y == UDF.TASK_DONE:
                count_parsed += 1
                if self.pb is not None:
                    self.pb.update(1)
            else:
                raise ValueError("Got non-sentinal output.")

        in_queue_filler.join()
        in_queue.put(UDF.QUEUE_CLOSED)

        for udf in self.udfs:
            udf.join()

        # Terminate and flush the processes
        for udf in self.udfs:
            udf.terminate()
        self.udfs = []
388556
def _run_pipeline(self, multiprocessing):
        """"""
        if not self._flag_m:
            raise UnsupportedCall(f"Extractor(engine={self._exeng})' has no attribute '_run_pipline'")

        if not self._flag_q:
            self._flag_q = True
            warnings.warn("'Extractor(engine=pipeline)' does not support output; "
                          f"'fout={self._ofnm}' ignored", AttributeWarning, stacklevel=stacklevel())

        self._frnum = 1                                                 # frame number (revised)
        self._expkg = multiprocessing                                   # multiprocessing module
        self._mpprc = list()                                            # multiprocessing process list
        self._mpfdp = collections.defaultdict(multiprocessing.Queue)    # multiprocessing file pointer

        self._mpmng = multiprocessing.Manager()                         # multiprocessing manager
        self._mpkit = self._mpmng.Namespace()                           # multiprocessing work kit

        self._mpkit.counter = 0                                         # work count (on duty)
        self._mpkit.pool = 1                                            # work pool (ready)
        self._mpkit.current = 1                                         # current frame number
        self._mpkit.eof = False                                         # EOF flag
        self._mpkit.frames = dict()                                     # frame storage
        self._mpkit.trace = self._trace                                 # flow tracer
        self._mpkit.reassembly = copy.deepcopy(self._reasm)             # reassembly buffers

        # preparation
        self.record_header()
        self._mpfdp[0].put(self._gbhdr.length)

        # extraction
        while True:
            # check EOF
            if self._mpkit.eof:
                self._update_eof()
                break

            # check counter
            if self._mpkit.pool and self._mpkit.counter < CPU_CNT:
                # update file offset
                self._ifile.seek(self._mpfdp.pop(self._frnum-1).get(), os.SEEK_SET)

                # create worker
                # print(self._frnum, 'start')
                proc = multiprocessing.Process(
                    target=self._pipeline_read_frame,
                    kwargs={'mpkit': self._mpkit, 'mpfdp': self._mpfdp[self._frnum]}
                )

                # update status
                self._mpkit.pool -= 1
                self._mpkit.counter += 1

                # start and record
                proc.start()
                self._frnum += 1
                self._mpprc.append(proc)

            # check buffer
            if len(self._mpprc) >= CPU_CNT:
                [proc.join() for proc in self._mpprc[:-4]]
                del self._mpprc[:-4]
477503
def parallelize_func(iterable, func, chunksz=1, n_jobs=16, *args, **kwargs):
    
    chunker = func
    chunks = more_itertools.chunked(iterable, chunksz)
    chunks_results = Parallel(n_jobs=n_jobs, verbose=50)(
        delayed(chunker)(chunk, *args, **kwargs) for chunk in chunks)
    results = more_itertools.flatten(chunks_results)
    return list(results)
273233
def _multiprocessing_to_asyncio(in_queue, out_queue, loop):
    

    while True:
        value = in_queue.get()
        loop.call_soon_threadsafe(out_queue.put_nowait, value)
815984
def multi_p_run(tot_num, _func, worker, params, n_process):
    
    from multiprocessing import Process, Queue
    out_q = Queue()
    procs = []

    split_num = split_seq(list(range(0, tot_num)), n_process)

    print(tot_num, ">>", split_num)

    split_len = len(split_num)
    if n_process > split_len:
        n_process = split_len

    for i in range(n_process):
        _p = Process(target=_func,
                     args=(worker, split_num[i][0], split_num[i][1],
                           params, out_q))
        _p.daemon = True
        procs.append(_p)
        _p.start()

    try:
        result = []
        for i in range(n_process):
            result.append(out_q.get())
        for i in procs:
            i.join()
    except KeyboardInterrupt:
        print('Killing all the children in the pool.')
        for i in procs:
            i.terminate()
            i.join()
        return -1

    while not out_q.empty():
        print(out_q.get(block=False))

    return result
798595
def runner(view, parallel, dirs, config):
    
    def run(fn_name, items):
        setpath.prepend_bcbiopath()
        out = []
        fn, fn_name = (fn_name, fn_name.__name__) if callable(fn_name) else (_get_ipython_fn(fn_name, parallel), fn_name)
        items = [x for x in items if x is not None]
        items = diagnostics.track_parallel(items, fn_name)
        logger.info("ipython: %s" % fn_name)
        if len(items) > 0:
            items = [config_utils.add_cores_to_config(x, parallel["cores_per_job"], parallel) for x in items]
            if "wrapper" in parallel:
                wrap_parallel = {k: v for k, v in parallel.items() if k in set(["fresources"])}
                items = [[fn_name] + parallel.get("wrapper_args", []) + [wrap_parallel] + list(x) for x in items]
            items = zip_args([args for args in items])
            for data in view.map_sync(fn, items, track=False):
                if data:
                    out.extend(unzip_args(data))
        return out
    return run
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)
995652
def parmap(f, X, nprocs=multiprocessing.cpu_count()):
    
    q_in = multiprocessing.Queue(1)
    q_out = multiprocessing.Queue()

    proc = [multiprocessing.Process(target=fun, args=(f, q_in, q_out))
            for _ in range(nprocs)]
    for p in proc:
        p.daemon = True
        p.start()

    sent = [q_in.put((i, x)) for i, x in enumerate(X)]
    [q_in.put((None, None)) for _ in range(nprocs)]
    res = [q_out.get() for _ in range(len(sent))]

    [p.join() for p in proc]

    return [x for i, x in sorted(res)]
1137690
def run_multicore(fn, items, config, parallel=None):
    
    if len(items) == 0:
        return []
    if parallel is None or "num_jobs" not in parallel:
        if parallel is None:
            parallel = {"type": "local", "cores": config["algorithm"].get("num_cores", 1)}
        sysinfo = system.get_info({}, parallel)
        parallel = resources.calculate(parallel, items, sysinfo, config,
                                       parallel.get("multiplier", 1),
                                       max_multicore=int(parallel.get("max_multicore", sysinfo["cores"])))
    items = [config_utils.add_cores_to_config(x, parallel["cores_per_job"]) for x in items]
    if joblib is None:
        raise ImportError("Need joblib for multiprocessing parallelization")
    out = []
    for data in joblib.Parallel(parallel["num_jobs"], batch_size=1, backend="multiprocessing")(joblib.delayed(fn)(*x) for x in items):
        if data:
            out.extend(data)
    return out
699804
def run(self, target, args=()):
        
        run_event = threading.Event()
        run_event.set()
        thread = threading.Thread(target=target, args=args + (run_event, ))
        self.thread_pool.append(thread)
        self.run_events.append(run_event)
        thread.start()
1005024
def _init_threading(self, function, params={}, num_threads=10):
        
        q = Queue(maxsize=0) # TODO: find something appropriate
        for i in range(num_threads):
            worker = Thread(target=function, args=(q, params))
            worker.setDaemon(True)
            worker.start()
        return q
987016
def threadpooled(
    func: typing.Callable[..., typing.Union["typing.Awaitable[typing.Any]", typing.Any]],
    *,
    loop_getter: None = None,
    loop_getter_need_context: bool = False,
) -> typing.Callable[..., "concurrent.futures.Future[typing.Any]"]:
    """"""
19607
def threadpooled(
    func: typing.Callable[..., typing.Union["typing.Awaitable[typing.Any]", typing.Any]],
    *,
    loop_getter: typing.Union[typing.Callable[..., asyncio.AbstractEventLoop], asyncio.AbstractEventLoop],
    loop_getter_need_context: bool = False,
) -> typing.Callable[..., "asyncio.Task[typing.Any]"]:
    """"""
120617
def run(self):
        '''
            
        '''

        # If we are chaining after another process, wait for it to complete.
        #   We use a flag here instead of joining the thread for various reasons
        chainAfter = self.chainAfter
        if chainAfter is not None:
            chainPollTime = self.backgroundIOPriority.chainPollTime
            while chainAfter.finished is False:
                time.sleep(chainPollTime)


        # Pull class data into locals
        fileObj = self.fileObj
        bandwidthPct = self.backgroundIOPriority.bandwidthPct
        bandwidthPctDec = bandwidthPct / 100.0


        # Number of blocks, total
        numBlocks = len(self.remainingData)
        # Bytes written
        dataWritten = 0

        # Mark that we have started writing data
        self.startedWriting = True

        # Create a conditional lambda for flushing. I'd rather just only support flushable streams, but
        #   some unfortunatly just aren't. This should be cheaper than testing with hasattr at each iteration
        if hasattr(fileObj, 'flush'):
            doFlush = lambda obj : obj.flush()
        else:
            doFlush = lambda obj : 1


        # numChunksRateSmoothing - How often we stop for a short bit to be gracious to other running tasks.
        #  float for division below
        numChunksRateSmoothing = float(self.backgroundIOPriority.numChunksRateSmoothing)

        # i will be the counter from 1 to numChunksRateSmoothing, and then reset
        i = 1

        # We start with using max bandwidth until we hit #numChunksRateSmoothing , at which case we recalculate
        #  sleepTime. We sleep after every block written to maintain a desired average throughput based on
        #  bandwidthPct
        sleepTime = 0

        # Before represents the "start" time. When we sleep, we will increment this value
        #  such that [ delta = (after - before) ] only accounts for time we've spent writing,
        #  not in charity.
        before = time.time()

        # timeSlept - Amount of time slept, which must be subtracted from total time spend
        #  to get an accurate picture of throughput.
        timeSlept = 0

        # firstPass - Mark the first pass through, so we can get a rough calculation
        #  of speed from the first write, and recalculate after #numChunksRateSmoothing
        firstPass = True

        if bandwidthPct == 100:
            shouldRecalculate = lambda i, numChunksRateSmoothing, firstPass : False
        else:
            shouldRecalculate = lambda i, numChunksRateSmoothing, firstPass : firstPass or i == numChunksRateSmoothing
            

        while len(self.remainingData) > 0:

            # pop, write, flush
            nextData = self.remainingData.popleft()
            fileObj.write(nextData)
            doFlush(fileObj)
            
            dataWritten += len(nextData)
            if sleepTime:
                sleepBefore = time.time()

                time.sleep(sleepTime)

                sleepAfter = time.time()
                timeSlept += (sleepAfter - sleepBefore)
 
            if shouldRecalculate(i, numChunksRateSmoothing, firstPass) is True:
                # if not sleeptime, we are on first 
                # We've completed a full period, time for charity
                after = time.time()

                delta = after - before - timeSlept

                rate = dataWritten / delta

#                if DEBUG is True:
#                    sys.stdout.write('\t  I have written %d bytes in %3.3f seconds and slept %3.3f sec (%4.5f M/s over %3.3fs)\n' %(dataWritten, delta, timeSlept, (rate) / (1024*1024), delta + timeSlept  ))
#                    sys.stdout.flush()

                # Calculate how much time we should give up on each block to other tasks
                sleepTime = delta * (1.00 - bandwidthPctDec)
                sleepTime /= numChunksRateSmoothing

#                if DEBUG is True:
#                    sys.stdout.write('Calculated new sleepTime to be: %f\n' %(sleepTime,))

                timeSlept = 0
                before = time.time()
                i = 0
#            elif DEBUG is True and i == numChunksRateSmoothing:
#                # When bandwidth pct is 100 (prio=1), the above DEBUG will never be hit.
#                after = time.time()
#
#                delta = after - before - timeSlept
#
#                rate = dataWritten / delta
#
#                sys.stdout.write('\t  I have written %d bytes in %3.3f seconds and slept %3.3f sec (%4.5f M/s over %3.3fs)\n' %(dataWritten, delta, timeSlept, (rate) / (1024*1024), delta + timeSlept  ))
#                sys.stdout.flush()
#
#                timeSlept = 0
#                before = time.time()
#                i = 0

            firstPass = False

            i += 1

        if self.closeWhenFinished is True:
            fileObj.close()

        self.finished = True
1029368
def run(self, num_of_processes=multiprocessing.cpu_count()):
        

        # class + wordification on every example of the main table

        p = multiprocessing.Pool(num_of_processes)

        indices = chunks(list(range(len(self.target_table))), num_of_processes)  # )

        for ex_idxs in indices:
            self.resulting_documents.extend(wordify_examples(self.name_to_table, self.connecting_tables, self.context,
                                                             self.index_by_value, self.target_table.name,
                                                             self.word_att_length, ex_idxs))
        p.close()
        p.join()

        for i, ex in enumerate(self.target_table):
            self.resulting_classes.append(ex.get_class())
1086033
static double run(int num_threads) {

    final double membw[] = new double[num_threads];
    Thread[] threads = new Thread[num_threads];
    for (int t=0;t<num_threads;++t) {
      final int thread_num = t;
      threads[t] = new Thread() {
        public void run() {
          MemoryBandwidth l = new MemoryBandwidth();
          membw[thread_num] = l.run_benchmark();
        }
      };
    }
    for (int t=0;t<num_threads;++t) {
      threads[t].start();
    }
    for (int t=0;t<num_threads;++t) {
      try {
        threads[t].join();
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
    return Utils.sum(membw);
  }
292947
def _map_parallel(function, args, n_jobs):
    """"""
    # Following the error checking found in joblib
    multiprocessing = int(os.environ.get('JOBLIB_MULTIPROCESSING', 1)) or None
    if multiprocessing:
        try:
            import multiprocessing
            import multiprocessing.pool
        except ImportError:
            multiprocessing = None
        if sys.platform.startswith("win") and PY2:
            msg = "Multiprocessing is not supported on Windows with Python 2.X. Setting n_jobs=1"
            logger.warning(msg)
            n_jobs = 1
    # 2nd stage: validate that locking is available on the system and
    #            issue a warning if not
    if multiprocessing:
        try:
            _sem = multiprocessing.Semaphore()
            del _sem  # cleanup
        except (ImportError, OSError) as e:
            multiprocessing = None
            logger.warning('{}. _map_parallel will operate in serial mode'.format(e))
    if multiprocessing and int(n_jobs) not in (0, 1):
        if n_jobs == -1:
            n_jobs = None
        try:
            pool = multiprocessing.Pool(processes=n_jobs)
            map_result = pool.map(function, args)
        finally:
            pool.close()
            pool.join()
    else:
        map_result = list(map(function, args))
    return map_result