Concurrent reading using multiprocessing

[hadim]

The example from the doc https://cloud.google.com/spanner/docs/reads#read_data_in_parallel works well with multi-threading. That being said in Python I found the performance to be equivalent to without using multithreading (probably because of the GIL).

I tried to use multiprocessing instead but can't make it work and I am not even sure this is possible.

• I first tried to replace the doc example ThreadPoolExecutor with ProcessPoolExecutor. It hangs forever so I guess the snapshot object can't be shared between multi processes.
• I also tried to recreate the session but a partition is only valid within the same session according to the error: details = "Partitioned request was created for a different session."

Here is the code of the latter approach (recreating the session):

spanner_client = spanner.Client(project=project_id)
instance = spanner_client.instance(instance_id)
database = instance.database(database_id)

table = "xxxx"
columns = ("col1", "col2")


def process(batch):
    
    spanner_client = spanner.Client(project=project_id)
    instance = spanner_client.instance(instance_id)
    database = instance.database(database_id)
    
    logger.info(f"Partition: {batch['partition'][:32]}")

    snapshot = database.batch_snapshot()

    row_ct = 0
    for row in tqdm(snapshot.process_read_batch(batch)):
        row_ct += 1
        
    snapshot.close()

    return time.time(), row_ct


snapshot = database.batch_snapshot()

keyset = spanner.KeySet(all_=True)
batches = snapshot.generate_read_batches(table=table, columns=columns, keyset=keyset)
batches = list(batches)

logger.info(f"{len(batches)} batches detected")

start = time.time()
with concurrent.futures.ProcessPoolExecutor() as executor:
    futures = [executor.submit(process, batch) for batch in batches]

    pbar = tqdm(concurrent.futures.as_completed(futures, timeout=3600), total=len(batches))
    for future in pbar:
        finish, row_ct = future.result()
        elapsed = finish - start
        print(f"Completed {row_ct} rows in {elapsed} seconds")

snapshot.close()

Is there is a way to dispatch batched results to multiple processes in Python for high-performance reading? I guess each partition hash should be able to live outside of a snapshot object on the Spanner side. I am not sure this is possible.

[larkee]

Hi there!

Thank you for filing this issue and for your patience!

I first tried to replace the doc example ThreadPoolExecutor with ProcessPoolExecutor. It hangs forever so I guess the snapshot object can't be shared between multi processes.

This is correct. The snapshot object has an indirect reference to the Client object which is not serializable. However, it should be possible to covert the BatchSnapshot into a dict and pass that instead and recreate it in process() like so:

import concurrent
import time
from tqdm import tqdm

from google.cloud import spanner
from google.cloud.spanner_v1.database import BatchSnapshot


spanner_client = spanner.Client(project=project_id)
instance = spanner_client.instance(instance_id)
database = instance.database(database_id)

table = "xxxx"
columns = ("col1", "col2")


def process(snapshot_dict, partition):
    spanner_client = spanner.Client(project=project_id)
    instance = spanner_client.instance(instance_id)
    database = instance.database(database_id)
    
    logger.info(f"Partition: {batch['partition'][:32]}")
    
    snapshot = BatchSnapshot.from_dict(db, snapshot_dict)
    
    row_cnt = 0
    for row in tqdm(snapshot.process_read_batch(partition)):
        row_cnt += 1
    return time.time(), row_cnt
    
 
batch_snapshot = database.batch_snapshot()

keyset = spanner.KeySet(all_=True)
partitions = list(batch_snapshot.generate_read_batches(table=table, columns=columns, keyset=keyset))

logger.info(f"{len(batches)} batches detected")

start = time.time()
with concurrent.futures.ProcessPoolExecutor() as executor:
    futures = [executor.submit(process, batch_snapshot.to_dict(), p) for p in partitions]

    pbar = tqdm(concurrent.futures.as_completed(futures, timeout=3600), total=len(partitions))
    for future in pbar:
        finish, row_ct = future.result()
        elapsed = finish - start
        print(f"Completed {row_ct} rows in {elapsed} seconds")

batch_snapshot.close()

However, even with this code snippet I am still getting the indefinite hanging you mentioned. Given that both batch_snapshot.to_dict() and p are both made up of native Python objects I'm not sure where the issue is. I'm not very familiar with multiprocessing so I'll need more time to look into this further, sorry.

[hadim]

Thanks @larkee for looking into it. I tested your to_dict() approach and ends up with this kind of error:

---------------------------------------------------------------------------
BrokenProcessPool                         Traceback (most recent call last)
<ipython-input-2-4456769017b2> in <module>
     44     pbar = tqdm(concurrent.futures.as_completed(futures, timeout=3600), total=len(partitions))
     45     for future in pbar:
---> 46         results = future.result()
     47 
     48 batch_snapshot.close()

~/local/conda/envs/circus/lib/python3.8/concurrent/futures/_base.py in result(self, timeout)
    430                 raise CancelledError()
    431             elif self._state == FINISHED:
--> 432                 return self.__get_result()
    433 
    434             self._condition.wait(timeout)

~/local/conda/envs/circus/lib/python3.8/concurrent/futures/_base.py in __get_result(self)
    386     def __get_result(self):
    387         if self._exception:
--> 388             raise self._exception
    389         else:
    390             return self._result

BrokenProcessPool: A process in the process pool was terminated abruptly while the future was running or pending.

Note that I am running the code into a Jupyter notebook. I haven't tested much than that so far.

[larkee]

Ahhh, I think I've found the issue. It seems that grpc does not have fork support enabled by default: grpc/grpc#18075

After running export GRPC_ENABLE_FORK_SUPPORT=1 to enable the support, the snippet no longer hangs for me. Let me know if that helps 👍

I'll try spend some time this week comparing the single-thread v multi-thread v multi-process performance.

[hadim]

Indeed GRPC_ENABLE_FORK_SUPPORT seems to do the job and get rid of the BrokenProcessPool: A process in the process pool was terminated abruptly while the future was running or pending. error.

Feel free to close and I'll reopen if I spot something wrong with it.

[larkee]

Thanks @hadim 👍