Repository: tensorflow
Issue: tf.data.Dataset prefetch not fetching data asynchronously
Labels: stat:awaiting tensorflower,type:bug,comp:data,type:performance,TF 2.11

<details><summary>Click to expand!</summary> 
 
 ### Issue Type

Bug

### Have you reproduced the bug with TF nightly?

No

### Source

source

### Tensorflow Version

2.11

### Custom Code

Yes

### OS Platform and Distribution

Debian/Linux 11

### Mobile device

_No response_

### Python version

3.7

### Bazel version

_No response_

### GCC/Compiler version

_No response_

### CUDA/cuDNN version

_No response_

### GPU model and memory

_No response_

### Current Behaviour?

After implementing a data pipeline using tf.data.Dataset to pull image data from Google Cloud Storage, TensorBoard profiler shows that the GPU compute and CPU prefetch are running synchronously. I used data.Dataset.AUTOTUNE to determine the appropriate prefetch batch size. Monitoring GPU usage while the model is running confirms this with the GPU at 0% utilization to actually computing something for about a 2:1 ratio, which is reflected in the profiler. CPU usage when monitored does not appear to max out.

I expected the prefetch to occur concurrently with GPU processing as described in the data.Dataset documentation and tutorials.

![ch](https://github.com/tensorflow/tensorflow/assets/79778984/e96ad312-12b0-4bbb-b06e-f4e4976714b3)

![cp](https://github.com/tensorflow/tensorflow/assets/79778984/f52959d1-23fd-45ed-ba57-5a532afd0972)

![gp](https://github.com/tensorflow/tensorflow/assets/79778984/2b2e15b6-2cf7-40d8-89b1-98889f151863)


### Standalone code to reproduce the issue

```shell
os.environ["CUDA_VISIBLE_DEVICES"] = "0"
os.environ['TF_GPU_ALLOCATOR'] = "cuda_malloc_async"
config = ConfigProto()
config.gpu_options.allow_growth = True
session = InteractiveSession(config=config)

def get_label(file_path):
    parts = tf.strings.split(file_path, os.path.sep)
    one_hot = parts[-2] == class_names
    return tf.argmax(one_hot)

def decode_img(img):
    img = tf.io.decode_image(img, channels=3, expand_animations = False)
    img = tf.image.resize(img, [244, 244])
    img = tf.cast(img, tf.float32)
    return img

def process_path(file_path):
    label = get_label(file_path)
    img = tf.io.read_file(file_path)
    img = decode_img(img)
    return img, label

def configure_for_performance(ds):
    ds = ds.batch(128)
    ds = ds.prefetch(buffer_size=tf.data.AUTOTUNE)
    return ds

files = tf.data.Dataset.list_files((data_dir + '/*/*.png'), shuffle=False)
files = files.shuffle(image_count, reshuffle_each_iteration=False)

val_size = int(image_count * 0.2)

train_files = files.skip(val_size)
val_files = files.take(val_size)

train_ds = train_files.interleave(lambda x: tf.data.Dataset.from_tensor_slices([x]), cycle_length=4, num_parallel_calls=tf.data.AUTOTUNE)
train_ds = train_ds.map(process_path, num_parallel_calls=tf.data.AUTOTUNE)

val_ds = val_files.interleave(lambda x: tf.data.Dataset.from_tensor_slices([x]), cycle_length=4, num_parallel_calls=tf.data.AUTOTUNE)
val_ds = val_ds.map(process_path, num_parallel_calls=tf.data.AUTOTUNE)

train_ds = configure_for_performance(train_ds)
val_ds = configure_for_performance(val_ds)
```


### Relevant log output

_No response_</details>