🔒 multiprocessing "sometimes locking" issue

[cloud flint]

I'm sorry, i have to divide the messag edue to the character limit

def argos_target_worker(task_queue: multiprocessing.Queue, result_queue):
    while True:
        try:
            task = task_queue.get(timeout=5)  # Short timeout to detect queue exhaustion
            if task is None:  # Sentinel to stop
                break
            task.name = argos_fill_multi_lang_name(task.name)
            result_queue.put(task)
        except Empty:
            break

[rose marten]

@cloud flint

Python help channel opened

Remember to:

• Ask your Python question, not if you can ask or if there's an expert who can help.
• Show a code sample as text (rather than a screenshot) and the error message, if you've got one.
• Explain what you expect to happen and what actually happens.

:warning: Do not pip install anything that isn't related to your question, especially if asked to over DMs.

Closes after a period of inactivity, or when you send !close.

[cloud flint]

def argos_group_translate(logger: Logger, o_set: set, num_workers=6) -> set:
    # num_workers = 5 # num_workers or multiprocessing.cpu_count() # // 2
    if not o_set:
        return o_set
    task_queue = multiprocessing.Queue()
    r_queue = multiprocessing.Queue()
    results = set()
    logger.info(f'Called translation for {len(o_set)} objects...')
    # Read file and feed workers
    amt = 0
    for o in o_set:
        if not o.name.invalid_lang_name and any([n is None for n in [o.name.name_en, o.name.name_es, o.name.name_it,
                                                                     o.name.name_fr, o.name.name_de]]):
            task_queue.put(o)
            amt += 1
        else:
            results.add(o)
    if amt == 0:
        return o_set
    logger.info(f'Starting translation for {amt} {o.__class__.__name__} items using {num_workers} workers.')
    workers = [multiprocessing.Process(target=argos_target_worker, args=(task_queue, r_queue))
               for _ in range(num_workers)]
    for p in workers:
        p.start()

    expected_results = amt  # Number of tasks we expect back
    collected = 0

    while collected < expected_results:
        try:
            if collected % 1000 == 0:
                logger.debug(f'Current tally: {collected}/{expected_results}')
            result = r_queue.get(timeout=10)
            if result is not None:  # Explicitly skip None (shouldn’t happen now, but safety)
                results.add(result)
                collected += 1
            else:
                logger.warn("Main: Received unexpected None from r_queue")
        except Empty:
            logger.warn(f"Main: Timeout waiting for result after 10s, collected {collected}/{expected_results}")
            break
    # Send stop signals to workers after collecting results
    for _ in range(len(workers)):
        task_queue.put(None)
    # Wait for workers to finish
    for p in workers:
        p.join()
    return results

Sometimes this works fine, the last times i ran it the

if collected % 1000 == 0:
    logger.debug(f'Current tally: {collected}/{expected_results}')

didn't spit anything for a long time, so i assume its either pycharm or there's some lock condition that sometimes triggers. I'm not well versed on how to work with multiple processes

[rose marten]

@cloud flint

Python help channel closed for inactivity

This help channel has been closed. Feel free to create a new post in #1035199133436354600. To maximize your chances of getting a response, check out this guide on asking good questions.