macOS – multiprocessingのQueue.qsize() NotImplementedErrorを解決する!

環境

Python 3.9
MacOS 13.0

やりたいこと

サブプロセスを複数生成してデータを並列処理し、キューに結果をputしていく
別のサブプロセスから順次結果を取り出して利用する

キューの文字列に"all done"を見つけたら処理を終了する

def pull_queue(queue:Manager):
while True:
    if queue.qsize() != 0:
        result=queue.get(True)
        print(result)
        if(result=="all done"):
            break
    time.sleep(0.03)

発生した例外

Traceback (most recent call last):
  File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/multiprocessing/process.py", line 315, in _bootstrap
    self.run()
  File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "/Users/***/***/***/***.py", line 12, in show_queue
    if queue.qsize() != 0:
  File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/multiprocessing/queues.py", line 126, in qsize
    return self._maxsize - self._sem._semlock._get_value()
NotImplementedError

原因

Mac OSXだとqsize()はNotImplementedErrorを出す実装になっているため。

def qsize(self):
        # Raises NotImplementedError on Mac OSX because of broken sem_getvalue()
        return self._maxsize - self._sem._semlock._get_value()

回避策

qsize()の実装されているManager関数経由のQueueクラスを使う。

    #q=multiprocessing.Queue()
    q=Manager().Queue()

実行結果

qsize()が使えるようになった。

...
durian:31.92
blueberry:5.32
kiwifruit:1.596
all done