def _apply_mt(self, doc_loader, parallelism, **kwargs):
if not Meta.postgres:<EOL><INDENT>raise ValueError("<STR_LIT>")<EOL><DEDENT>def fill_input_queue(in_queue, doc_loader, terminal_signal):<EOL><INDENT>for doc in doc_loader:<EOL><INDENT>in_queue.put(doc)<EOL><DEDENT>in_queue.put(terminal_signal)<EOL><DEDENT>manager = Manager()<EOL>in_queue = manager.Queue()<EOL>out_queue = JoinableQueue()<EOL>total_count = len(doc_loader)<EOL>for i in range(parallelism):<EOL><INDENT>udf = self.udf_class(<EOL>in_queue=in_queue,<EOL>out_queue=out_queue,<EOL>worker_id=i,<EOL>**self.udf_init_kwargs,<EOL>)<EOL>udf.apply_kwargs = kwargs<EOL>self.udfs.append(udf)<EOL><DEDENT>for udf in self.udfs:<EOL><INDENT>udf.start()<EOL><DEDENT>terminal_signal = UDF.QUEUE_CLOSED<EOL>in_queue_filler = Process(<EOL>target=fill_input_queue, args=(in_queue, doc_loader, terminal_signal)<EOL>)<EOL>in_queue_filler.start()<EOL>count_parsed = <NUM_LIT:0><EOL>while count_parsed < total_count:<EOL><INDENT>y = out_queue.get()<EOL>if y == UDF.TASK_DONE:<EOL><INDENT>count_parsed += <NUM_LIT:1><EOL>if self.pb is not None:<EOL><INDENT>self.pb.update(<NUM_LIT:1>)<EOL><DEDENT><DEDENT>else:<EOL><INDENT>raise ValueError("<STR_LIT>")<EOL><DEDENT><DEDENT>in_queue_filler.join()<EOL>in_queue.put(UDF.QUEUE_CLOSED)<EOL>for udf in self.udfs:<EOL><INDENT>udf.join()<EOL><DEDENT>for udf in self.udfs:<EOL><INDENT>udf.terminate()<EOL><DEDENT>self.udfs = []<EOL>
Run the UDF multi-threaded using python multiprocessing