BERT model from pipeline hangs with multiprocessing pool
Author: Zhenshan-Jin
Pull request: no

## Environment info
<!-- You can run the command `transformers-cli env` and copy-and-paste its output below.
     Don't forget to fill out the missing fields in that output! -->

- `transformers` version: transformers==4.9.1
- Platform: macOS Catalina 10.15.7 (19H2)
- Python version: 3.7.5
- PyTorch version (GPU?): 
- Tensorflow version (GPU?):tensorflow==2.7.0
- Using GPU in script?:No
- Using distributed or parallel set-up in script?:Yes

### Who can help
@sgugger 

Models:

- BERT: @LysandreJik 

Library:

- Pipelines: @Narsil


## Information

The model I am using is Bert. I can run the model without problem without multiprocessing. However, when I try to use multiprocessing and run the model in another process. the process will hang and never return.

## To reproduce

Steps to reproduce the behavior:

helper.py
```python
import multiprocessing
import os
from transformers import DistilBertTokenizer, TFDistilBertForQuestionAnswering, pipeline


qa = pipeline('question-answering',
              model=TFDistilBertForQuestionAnswering.from_pretrained("./bert"),
              tokenizer=DistilBertTokenizer.from_pretrained("./bert"))

def compute(query, context):
    question_list = [{"question": query, "context": context}]
    print(f"start qa: {os.getpid()}")
    result = qa(question_list)
    print("end qa")

    return result


class NoDaemonProcess(multiprocessing.Process):
    @property
    def daemon(self):
        return False

    @daemon.setter
    def daemon(self, value):
        pass

class NoDaemonContext(type(multiprocessing.get_context("fork"))):
    Process = NoDaemonProcess
```

main.py
```python
import multiprocessing
from helper import NoDaemonContext, compute
import os

if __name__ == "__main__":
    print(f"current pid: {os.getpid()}")        
    query = "what is a potato?"
    context = "A potato is a starchy vegetable. A potato is nothing"
    
    mp_context = NoDaemonContext()
    pool = mp_context.Pool(multiprocessing.cpu_count())
    async_res = pool.apply_async(compute, (query, context))
    async_res.wait()
    res = async_res.get()
```

1. run `python main.py`
( I have to use the class NoDaemonContext, otherwise, the multiprocessing module will complain `AssertionError: daemonic processes are not allowed to have children`)

## Expected behavior

the code will hang in the line calling the model
```
result = qa(question_list)
```


Comments:
similar issue: [https://github.com/huggingface/transformers/issues/14919](url)

The error seems to come from Tensorflow itself, here is an example without using `pipeline` whatsoever.

```python
import multiprocessing
import os
from transformers import DistilBertTokenizer, TFDistilBertForQuestionAnswering, pipeline
import tensorflow as tf


model = TFDistilBertForQuestionAnswering.from_pretrained("bert-base-uncased")


def compute(query, context):
    question_list = [{"question": query, "context": context}]
    print(f"start qa: {os.getpid()}")

    tokens = tf.zeros((1, 10), dtype=tf.int32)
    result = model(input_ids=tokens)
    print("end qa")

    return result


class NoDaemonProcess(multiprocessing.Process):
    @property
    def daemon(self):
        return False

    @daemon.setter
    def daemon(self, value):
        pass


class NoDaemonContext(type(multiprocessing.get_context("fork"))):
    Process = NoDaemonProcess


if __name__ == "__main__":
    print(f"current pid: {os.getpid()}")
    query = "what is a potato?"
    context = "A potato is a starchy vegetable. A potato is nothing"

    mp_context = NoDaemonContext()
    pool = mp_context.Pool(multiprocessing.cpu_count())
    async_res = pool.apply_async(compute, (query, context))
    async_res.wait()
```

Other hints that it's tricky to share tf session across processes: https://stackoverflow.com/questions/36610290/tensorflow-and-multiprocessing-passing-sessions

The easiest way to solve ALL your problems related to threading or multiprocessing is to load the  pipeline DIRECTLY on the thread or process instead of trying to share it afterwards. If you are doing GPU processing, you might have different issues, and for CPU processing it will work but the model will be loaded N times (no real way around it with TF I think, also please note that using CPU parallelism this way might not even be a win if TF is able to use all your cores for matrix multiplications, OnnxRuntime does that for instance, just keep measuring.)

Keep in mind that if you are controlling the parallelism yourself, you should deactivate it every else where it might be activated  (otherwise there's a high risk that threads/processes steal work on individual cores and context switches hurt performance instead of helping). For instance, `slow` tokenizers use a threading pool to encode questions, the easy fix is to use `DistilBertTokenizerFast` which will not use old code and work out of the box (no need to handle Daemonic processes vs not daemonic for instance).

Here is the solution closest to your original one (I recommend removing the NoDaemon thing, and using DistilBertTokenizerFast instead if you can though)

```python
import multiprocessing
import os
from transformers import DistilBertTokenizer, TFDistilBertForQuestionAnswering, pipeline


QA = None


def get_qa():
    if QA is not None:
        return QA
    qa = pipeline(
        "question-answering",
        model=TFDistilBertForQuestionAnswering.from_pretrained("bert-base-uncased"),
        tokenizer=DistilBertTokenizer.from_pretrained("bert-base-uncased"),
    )
    return qa


def compute(query, context):
    question_list = [{"question": query, "context": context}]
    print(f"start qa: {os.getpid()}")

    qa = get_qa()

    result = qa(question_list)
    print("end qa", flush=True)

    return result


class NoDaemonProcess(multiprocessing.Process):
    @property
    def daemon(self):
        return False

    @daemon.setter
    def daemon(self, value):
        pass


class NoDaemonContext(type(multiprocessing.get_context("fork"))):
    Process = NoDaemonProcess


if __name__ == "__main__":
    print(f"current pid: {os.getpid()}")
    query = "what is a potato?"
    context = "A potato is a starchy vegetable. A potato is nothing"

    mp_context = NoDaemonContext()
    # compute(query, context=context)
    with mp_context.Pool(multiprocessing.cpu_count()) as pool:
        async_res = pool.apply_async(compute, (query, context))
        print(async_res.get())
        # async_res = pool.apply(compute, (query, context))
        # print(async_res)
```

I will close this in favor of : https://github.com/huggingface/transformers/issues/14919 for further discussion.


@Narsil Thank you so much for your help and I think you are right after reloading the model for every process/thread and the code is working as expected without hanging issue.