Method test_multiprocess

tables/tests/test_basics.py:2274–2321

(self)

@common.unittest.skipIf(not common.blosc_avail, "Blosc not available")
class BloscSubprocess(common.PyTablesTestCase):
    def test_multiprocess(self):
        # Create a relatively large table with Blosc level 9 (large blocks)
        h5fname = tempfile.mktemp(prefix="multiproc-blosc9-", suffix=".h5")
        try:
            size = 300_000
            sa = np.fromiter(
                ((i, i**2, i // 3) for i in range(size)), "i4,i8,f8"
            )
            with tb.open_file(h5fname, "w") as h5file:
                h5file.create_table(
                    h5file.root,
                    "table",
                    sa,
                    filters=tb.Filters(complevel=9, complib="blosc"),
                    chunkshape=(size // 3,),
                )

            if common.verbose:
                print("**** Running from main process:")
            _worker(h5fname)

            if common.verbose:
                print("**** Running from subprocess:")

            try:
                qout = mp.Queue()
            except OSError:
                print("Permission denied due to /dev/shm settings")
            else:
                ps = mp.Process(
                    target=_worker,
                    args=(
                        h5fname,
                        qout,
                    ),
                )
                ps.daemon = True
                ps.start()

                result = qout.get()
                if common.verbose:
                    print(result)

                ps.join()
                # Avoid warnings with later tests forking subprocesses.
                ps.terminate()
        finally:
            Path(h5fname).unlink()


class HDF5ErrorHandling(common.PyTablesTestCase):