Python: GIL is not released during blocking read from ADIOS2 stream

[janEbert]

Describe the bug
When waiting to read from an openpmd_api.Series.read_iterations that was created from an ADIOS2 stream, the GIL is never released.
This creates a race condition when asynchronously reading via threading.Thread and adding to a queue.Queue: if data is not consumed fast enough and a read out of the stream ever starts to block because the Queue is full, the Python process cannot continue execution because the GIL is forever kept on the openPMD side.

Using multiprocessing.{Process,Queue} is a way around this problem.

Pseudocode

from queue import Queue
from threading import Thread
import time

import openpmd_api

def read_queued(stream, queue):
    for x in iter(stream.read_iterations()):
        queue.put(x)

stream = openpmd_api.Series('stream.sst', openpmd_api.Access_Type.read_only)
queue = Queue(maxsize=3)

reader_thread = Thread(target=read_queued, args=(stream, queue))
reader_thread.start()

for i in range(20):
    queue.get()
    # This duration must be larger than what reading a data point from the stream takes.
    time.sleep(5)
print('I'll never be reached!')
reader_thread.join()

Expected behavior
If the read blocks while waiting for stream data, the GIL needs to be released. (Related Python doc and PyBind11 doc.)

Software Environment

• version of openPMD-api: 0.15.0-dev
• installed openPMD-api via: from source
• operating system: Rocky Linux 8
• machine: JUWELS Booster
• name and version of Python implementation: CPython 3.9.6
• version of ADIOS2: 3d498e5c423ec3830264b1cfdf7c0182dd0bddc7
• name and version of MPI: ParaStationMPI 5.5.0

Additional context
Sorry for no example, don't know how to create a minimal ADIOS2 stream.

[franzpoeschel]

Notes to myself for fixing this:

• Direct fix https://pybind11.readthedocs.io/en/stable/advanced/misc.html#global-interpreter-lock-gil
• Maybe use timeout param in ADIOS2
• More idiomatic fix via async for https://github.com/pybind/pybind11/blob/master/tests/test_async.cpp https://github.com/pybind/pybind11/blob/master/tests/test_async.py (if possible somehow, maybe in combo with timeout)
• https://github.com/pybind/pybind11/pull/1842/files https://peps.python.org/pep-0492/

[franzpoeschel]

I've tested your example with #1381, replacing examples/10_streaming_write.py with the following:

#!/usr/bin/env python
import json
import sys

import openpmd_api as io

# pass-through for ADIOS2 engine parameters
# https://adios2.readthedocs.io/en/latest/engines/engines.html
config = {'adios2': {'engine': {}, 'dataset': {}}}
config['adios2']['engine'] = {'parameters': {'Threads': '4'}}
config['adios2']['dataset'] = {'operators': [{'type': 'bzip2'}]}


from queue import Queue
from threading import Thread
import time

def read_queued(stream, queue):
    print("READ QUEUED")
    for x in stream.read_iterations():
        print("SEEING ITERATION", x.iteration_index)
        queue.put(x)
        print("ENQUEUED ITERATION", x.iteration_index)

print("OPENING STREAM")
stream = io.Series('simData.sst', io.Access_Type.read_only, json.dumps(config))
print("OPENED STREAM")
queue = Queue(maxsize=3)

reader_thread = Thread(target=read_queued, args=(stream, queue))
reader_thread.start()

for i in range(6):
    print("Trying to get iteration", i)
    queue.get()
    print("Got iteration", i)
    # This duration must be larger than what reading a data point from the stream takes.
    time.sleep(5)

print('Ill never be reached!')
reader_thread.join()

The result is:

> bin/10_streaming_write.py & bin/10_streaming_read.py 
[3] 2159352
Warning: ADIOS2 backend does not support compression method bzip2. Continuing without compression.
Original error: [Wed Mar 01 16:48:11 2023] [ADIOS2 EXCEPTION] <Operator> <OperatorFactory> <MakeOperator> : ADIOS2 didn't compile with bzip2 library, operator not added

OPENING STREAM
Warning: ADIOS2 backend does not support compression method bzip2. Continuing without compression.
Original error: [Wed Mar 01 16:48:11 2023] [ADIOS2 EXCEPTION] <Operator> <OperatorFactory> <MakeOperator> : ADIOS2 didn't compile with bzip2 library, operator not added

OPENED STREAM
READ QUEUED
Trying to get iteration 0
SEEING ITERATION 1
ENQUEUED ITERATION 1
Got iteration 0
SEEING ITERATION 2
ENQUEUED ITERATION 2
SEEING ITERATION 3
ENQUEUED ITERATION 3
SEEING ITERATION 4
ENQUEUED ITERATION 4
SEEING ITERATION 5
Trying to get iteration 1
Got iteration 1
ENQUEUED ITERATION 5
SEEING ITERATION 6
Trying to get iteration 2
Got iteration 2
ENQUEUED ITERATION 6
SEEING ITERATION 7
Trying to get iteration 3
Got iteration 3
ENQUEUED ITERATION 7
SEEING ITERATION 8
Trying to get iteration 4
Got iteration 4
ENQUEUED ITERATION 8
SEEING ITERATION 9
Trying to get iteration 5
Got iteration 5
ENQUEUED ITERATION 9
Ill never be reached!
[3]-  Done                    bin/10_streaming_write.py

Without the PR:

> bin/10_streaming_write.py & bin/10_streaming_read.py 
[4] 2161146
Warning: ADIOS2 backend does not support compression method bzip2. Continuing without compression.
Original error: [Wed Mar 01 16:52:13 2023] [ADIOS2 EXCEPTION] <Operator> <OperatorFactory> <MakeOperator> : ADIOS2 didn't compile with bzip2 library, operator not added

OPENING STREAM
Warning: ADIOS2 backend does not support compression method bzip2. Continuing without compression.
Original error: [Wed Mar 01 16:52:13 2023] [ADIOS2 EXCEPTION] <Operator> <OperatorFactory> <MakeOperator> : ADIOS2 didn't compile with bzip2 library, operator not added

OPENED STREAM
READ QUEUED
SEEING ITERATION 0
ENQUEUED ITERATION 0
Trying to get iteration 0
SEEING ITERATION 1
ENQUEUED ITERATION 1
Got iteration 0
SEEING ITERATION 2
ENQUEUED ITERATION 2
SEEING ITERATION 3
ENQUEUED ITERATION 3
SEEING ITERATION 4
Trying to get iteration 1
Got iteration 1
ENQUEUED ITERATION 4
SEEING ITERATION 5
Trying to get iteration 2
Got iteration 2
ENQUEUED ITERATION 5
SEEING ITERATION 6
Trying to get iteration 3
Got iteration 3
ENQUEUED ITERATION 6
SEEING ITERATION 7
Trying to get iteration 4
Got iteration 4
ENQUEUED ITERATION 7
SEEING ITERATION 8
Trying to get iteration 5
Got iteration 5
ENQUEUED ITERATION 8
SEEING ITERATION 9
Ill never be reached!
^CTraceback (most recent call last):
  File "/home/franzpoeschel/singularity_build/nbuild/bin/10_streaming_read.py", line 41, in <module>
    reader_thread.join()
  File "/nix/store/iw1vmh509hcbby8dbpsaanbri4zsq7dj-python3-3.10.10/lib/python3.10/threading.py", line 1096, in join
    self._wait_for_tstate_lock()
  File "/nix/store/iw1vmh509hcbby8dbpsaanbri4zsq7dj-python3-3.10.10/lib/python3.10/threading.py", line 1116, in _wait_for_tstate_lock
    if lock.acquire(block, timeout):
KeyboardInterrupt

[3]   Done                    bin/10_streaming_write.py
Writer 0 (0x95d9b0): Got an unexpected connection close event

(The print is still reached, but the thread is never joined)

Do take care when using this pattern, loading data from a time step must take place during the time that the time step's loop iteration is active. When receiving queue.get(), the iteration will generally be closed already.