Function test_dali_proxy_error_propagation

dali/test/python/test_dali_proxy.py:350–390

()

@attr("pytorch")
def test_dali_proxy_error_propagation():
    from nvidia.dali.plugin.pytorch.experimental import proxy as dali_proxy
    import torchvision.datasets as datasets

    batch_size = 4
    num_threads = 3
    device_id = 0
    nworkers = 2

    @pipeline_def
    def pipe_with_error():
        images = fn.external_source(name="images", no_copy=True)
        error_anchor = types.Constant(np.array([-10], dtype=np.float32))
        return fn.crop(
            images, crop=(224, 224), crop_pos_x=error_anchor, out_of_bounds_policy="error"
        )

    pipe = pipe_with_error(
        batch_size=batch_size,
        num_threads=num_threads,
        device_id=device_id,
        prefetch_queue_depth=3,
    )
    with dali_proxy.DALIServer(pipe) as dali_server:

        dataset = datasets.ImageFolder(jpeg, transform=dali_server.proxy)
        loader = dali_proxy.DataLoader(
            dali_server,
            dataset,
            batch_size=batch_size,
            num_workers=nworkers,
        )

        err_msg = "Critical error in pipeline:*Anchor for dimension 1*is out of range*"
        with assert_raises(RuntimeError, glob=err_msg):
            next(iter(loader))

    # For some reason if we don't do this in this test, we see some ignored exception
    # messages in the next test
    pipe._shutdown()
    del pipe


@attr("pytorch")