[Python] FlightStreamReader not stop iteration when peer writer closed while local FlightStreamWriter not
- Dominant language
- C++
- Stars
- 17.1k
- Forks
- 4.3k
- Avg merge
- 3d 13h
- Merged PRs (30d)
- 88
Description
### Describe the bug, including details regarding any error messages, version, and platform.
version: pyarrow 12.0.1
python: 3.8.17
OS: MacOS 12.6.7
The following code is randomly deadlocked.
```python
import concurrent.futures
import pyarrow as pa
import pyarrow.flight
def write(writer, prefix):
schema = pa.schema([("id", pa.utf8())])
writer.begin(schema)
for i in range(10):
print(f"{prefix} write {i}")
writer.write(
pa.record_batch([[prefix]], schema)
)
writer.close()
def read(reader, prefix):
i = 0
for rb in reader.to_reader():
print(f"{prefix} {i} [{rb.to_pylist()}]")
i += 1
print(f"{prefix} in-read done")
class ScratchServer(pa.flight.FlightServerBase):
def do_exchange(self, context, descriptor, reader: pa.flight.MetadataRecordBatchReader,
writer: pa.flight.MetadataRecordBatchWriter):
print("do exchange")
with concurrent.futures.ThreadPoolExecutor() as executor:
write_task = executor.submit(write, writer, "server")
read_task = executor.submit(read, reader, "server")
write_task.add_done_callback(lambda _: print("server write done"))
read_task.add_done_callback(lambda _: print("server read done"))
print("do exchange done")
if __name__ == "__main__":
server = ScratchServer(location="grpc://127.0.0.1:12345")
print(">>>>>>")
client = pa.flight.FlightClient(location="grpc://127.0.0.1:12345")
writer, reader = client.do_exchange(pa.flight.FlightDescriptor.for_command(b""))
with concurrent.futures.ThreadPoolExecutor() as executor:
write_task = executor.submit(write, writer, "client")
read_task = executor.submit(read, reader, "client")
write_task.add_done_callback(lambda _: print("client write done"))
read_task.add_done_callback(lambda _: print("client read done"))
server.shutdown()
```
When a writer print "done", the peer reader should print "done" as well.
The snippet code may stop as expected sometims, but it may stuck when one side write has done, but the reader stuck forever.
```
>>>>>>
do exchange
client write 0
server write 0
server write 1
client 0 [[{'id': 'server'}]]
server write 2
client 1 [[{'id': 'server'}]]
server write 3
client 2 [[{'id': 'server'}]]server write 4
server write 5
client 3 [[{'id': 'server'}]]
client 4 [[{'id': 'server'}]]
client 5 [[{'id': 'server'}]]
server write 6
server write 7
client 6 [[{'id': 'server'}]]
server write 8
client 7 [[{'id': 'server'}]]
server write 9
client write 1
client 8 [[{'id': 'server'}]]
server 0 [[{'id': 'client'}]]
server write doneclient write 2
client 9 [[{'id': 'server'}]]
server 1 [[{'id': 'client'}]]
client write 3
server 2 [[{'id': 'client'}]]
client write 4
server 3 [[{'id': 'client'}]]
client write 5
server 4 [[{'id': 'client'}]]
client write 6
server 5 [[{'id': 'client'}]]
client write 7
client write 8
server 6 [[{'id': 'client'}]]
client write 9
server 7 [[{'id': 'client'}]]
server 8 [[{'id': 'client'}]]
server 9 [[{'id': 'client'}]]
```
### Component(s)
Python
Contributor guide
Assessment
This issue has not been assessed yet.