gdal_test_vsimem_threads.py

from osgeo import gdal
import joblib
from itertools import cycle, product, chain
from more_itertools import ichunked
import time
import os

SRC = 'RS2_OK76385_PK678064_DK606753_F2N_20080419_142127_HH_HV_SLC/RS2_OK76385_PK678064_DK606753_F2N_20080419_142127_HH_HV_SLC'

def block_windows(src, xsize, ysize):
     height, width = src.RasterYSize, src.RasterXSize
     nrows, rmod = divmod(height, ysize)
     ncols, cmod = divmod(width, xsize)
     offsets = product(range(0, width, xsize), range(0, height, ysize))
     sizes = product(chain([xsize]*ncols, [cmod]), chain([ysize]*nrows, [rmod]))
     for (xoff, yoff), (xsize, ysize) in zip(offsets, sizes):
         yield xoff, yoff, xsize, ysize

def worker_process(offsets_chunk):
    dataset_pool = tuple(gdal.Open(SRC) for _ in range(4))
    with joblib.parallel_backend('threading', n_jobs=4):
        with joblib.Parallel() as p:
            p(joblib.delayed(worker_thread)(ds, xoff, yoff, xsize, ysize) for ds, (xoff, yoff, xsize, ysize) in zip(cycle(dataset_pool), offsets_chunk))

def worker_thread(dataset, xoff, yoff, xsize, ysize):
    return dataset.ReadAsArray(xoff, yoff, xsize, ysize)

if __name__ == "__main__":
    os.environ['VRT_SHARED_SOURCE'] = '0'
    template_src = gdal.Open(SRC)
    offset_gen = list(block_windows(template_src, 512, 512))
    start = time.time()
    with joblib.Parallel(n_jobs=4, prefer='processes') as p:
        p(joblib.delayed(worker_process)(chunk) for chunk in ichunked(offset_gen, len(offset_gen) // 4))
    end = time.time()
    print('execution time:', (end-start))