6ca60969b6893268680d4386e2f22cdc3bc3c405,src/fonduer/utils/udf.py,UDFRunner,apply_mt,#UDFRunner#Any#Any#,86

Before Change

            docs_added += 1

        while any([udf.is_alive() for udf in self.udfs]) and count_parsed < total_count:
            if docs_added < total_count:
                in_queue.put(next(xs_generator))
                docs_added += 1
            if docs_added == total_count:
                in_queue.put(UDF.QUEUE_CLOSED)
                docs_added += 1

After Change

        // Create a Queue to feed documents to parsers
        manager = Manager()
        in_queue = manager.Queue()

        // Use an output queue to track multiprocess progress
        out_queue = JoinableQueue()

        total_count = len(xs)

        // 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
        pool = Pool(parallelism)
        in_tuples = ((in_queue, x) for x in xs)
        pool.map_async(func=async_fill_input_queue, iterable=in_tuples)

        count_parsed = 0
        while count_parsed < total_count:

Instances