Casting in multiprocessing hang forever
I don't think this is something we support.

I have a feeling you'll run into this whenever TF tries to use multithreading at all.  I doubt it has anything to do with `cast` or `convert_to_tensor` specifically.  It's likely _any_ tensorflow operation above a certain size.

What are you trying to accomplish? TensorFlow has state.  Once the TF context is initialized, threadpools are created, and these threads go into a sleep pattern, waiting on a condition variable for new work.  When you spawn a new process, you're copying the state, but this corrupts the underlying mutexes, and the threads never wake - so it hangs.

The `multiprocessing` package in general will only work correctly with stateless functions.

Internally, both `convert_to_tensor` and `cast` (and many other ops) are already multithreaded and SIMD vectorized, so you won't gain anything by trying to use a process pool here - simply call the bare `tf.convert_to_tensor`.  If you do want multiple processes, they'll need to have distinct TF contexts, and you won't be able to share data between them.

 Thanks for the reply. I apologize if it is missing some context, I wanted to put the minimal reproducible example.

A bit of context, I have many JSON files containing some annotations, I create one process for each of them that writes a tfrecord file (write as soon as SerializeToStringt is created). I choose that over the main process read them all and send one annotation to each process that sends back a SerializeToString that the main process then writes every time it has reached a certain number because the memory grows too much. But I am open to suggestions if you have some?

See the snippet below for a general idea (note: if you remove `tf.data.TFRecordDataset` after the pool the code does not hang anymore; you can replace it with any tf operation like `tf.ones((2, 2))` and the problem is there again)

```python
import multiprocessing

import tensorflow as tf


def _bytes_feature(value):
    if isinstance(value, type(tf.constant(0))):
        value = value.numpy()
    return tf.train.Feature(bytes_list=tf.train.BytesList(value=[value]))


def preprocess_image(image, height, width):
    return tf.cast(tf.image.resize(image, size=(height, width)), tf.uint8)


def write_tf_records(name):
    with tf.io.TFRecordWriter(f"/tmp/{name}.tfrecords") as writer:
        # load a json in COCO format then read each filename and open the image
        for image in [[[[0 for _ in range(3)] for _ in range(700)] for _ in range(700)]]:
            writer.write(
                tf.train.Example(
                    features=tf.train.Features(
                        feature={
                            "image": _bytes_feature(
                                tf.io.serialize_tensor(preprocess_image(image, 800, 800))
                            )
                        }
                    )
                ).SerializeToString()
            )


def test_cast_success():
    with multiprocessing.Pool(processes=1) as p:
        p.map(func=write_tf_records, iterable=["file1"])
    dataset = tf.data.TFRecordDataset(["/tmp/file1.tfrecords"])


def test_cast_hang_forever():
    with multiprocessing.Pool(processes=1) as p:
        p.map(func=write_tf_records, iterable=["file2"])
    dataset = tf.data.TFRecordDataset(["/tmp/file1.tfrecords"])


test_cast_success()
test_cast_hang_forever()
```
 Right, so as described in my previous reply you won't be able to use any TensorFlow commands within the multiprocess pool.  It's just not compatible with how TensorFlow works, or any package that creates and handles its own threads with persistent threadpools.  The `test_cast_success` works because there is no TF context the first time you spawn a new process (it's created on the first `tf` call).  Then in the main process, the context is created when you call `tf.data.TFRecordDataset`.  The `test_cast_hang_forever` call then breaks on creation of the new process, because the context exists and is copied to the new process, but this breaks all the internal state.

Your best bet is to stick to `tf.data` input pipelines.  Maybe something like [this](https://github.com/tensorflow/models/blob/master/research/object_detection/dataset_tools/create_coco_tf_record.py) can help you?  Or you just do this processing once, single-threaded. thanks @cantonios, I will take a look at your link Are you satisfied with the resolution of your issue?
<a href="https://docs.google.com/forms/d/e/1FAIpQLSfaP12TRhd9xSxjXZjcZFNXPGk4kc1-qMdv3gc6bEP90vY1ew/viewform?entry.85265664=Yes&entry.2137816233=https://github.com/tensorflow/tensorflow/issues/60660">Yes</a>
<a href="https://docs.google.com/forms/d/e/1FAIpQLSfaP12TRhd9xSxjXZjcZFNXPGk4kc1-qMdv3gc6bEP90vY1ew/viewform?entry.85265664=No&entry.2137816233=https://github.com/tensorflow/tensorflow/issues/60660">No</a>

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

Bug

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

Yes

### Source

source

### Tensorflow Version

tf>=2.8

### Custom Code

Yes

### OS Platform and Distribution

Linux Ubuntu 20.04, 22.04

### Mobile device

_No response_

### Python version

3.8

### Bazel version

_No response_

### GCC/Compiler version

_No response_

### CUDA/cuDNN version

None

### GPU model and memory

_No response_

### Current Behaviour?

When casting an array above a certain size in multiprocessing without GPU, the second call just hangs forever (either using `convert_to_tensor` or `cast`).

### Standalone code to reproduce the issue

```shell
import multiprocessing

import tensorflow as tf


def func(x):
    return tf.convert_to_tensor([[[0 for _ in range(700)] for _ in range(700)] for _ in range(3)], tf.uint8)


def test_cast_success():
    with multiprocessing.Pool(processes=1) as p:
        p.map(func=func, iterable=[1])


def test_cast_hang_forever():
    with multiprocessing.Pool(processes=1) as p:
        p.map(func=func, iterable=[1])


test_cast_success()
test_cast_hang_forever()
```


### Relevant log output

```shell
KeyboardInterrupt                         Traceback (most recent call last)
Cell In[3], line 21
     17         p.map(func=func, iterable=[1])
     20 test_cast_success()
---> 21 test_cast_hang_forever()

Cell In[3], line 17, in test_cast_hang_forever()
     15 def test_cast_hang_forever():
     16     with multiprocessing.Pool(processes=1) as p:
---> 17         p.map(func=func, iterable=[1])

File ~/.pyenv/versions/3.8.16/lib/python3.8/multiprocessing/pool.py:364, in Pool.map(self, func, iterable, chunksize)
    359 def map(self, func, iterable, chunksize=None):
    360     '''
    361     Apply `func` to each element in `iterable`, collecting the results
    362     in a list that is returned.
    363     '''
--> 364     return self._map_async(func, iterable, mapstar, chunksize).get()

File ~/.pyenv/versions/3.8.16/lib/python3.8/multiprocessing/pool.py:765, in ApplyResult.get(self, timeout)
    764 def get(self, timeout=None):
--> 765     self.wait(timeout)
    766     if not self.ready():
    767         raise TimeoutError

File ~/.pyenv/versions/3.8.16/lib/python3.8/multiprocessing/pool.py:762, in ApplyResult.wait(self, timeout)
    761 def wait(self, timeout=None):
--> 762     self._event.wait(timeout)

File ~/.pyenv/versions/3.8.16/lib/python3.8/threading.py:558, in Event.wait(self, timeout)
    556 signaled = self._flag
    557 if not signaled:
--> 558     signaled = self._cond.wait(timeout)
    559 return signaled

File ~/.pyenv/versions/3.8.16/lib/python3.8/threading.py:302, in Condition.wait(self, timeout)
    300 try:    # restore state no matter what (e.g., KeyboardInterrupt)
    301     if timeout is None:
--> 302         waiter.acquire()
    303         gotit = True
    304     else:

KeyboardInterrupt:
```
</details>