Speaker diarization and RabbitMQ multiprocessing error
State: closed | Pull request: no
Labels: name: stale

Hello there!

I have create a diarization pipeline and I wanted to implement it into a queue messages system. I am using Pika with RabbitMQ and multiprocessing package. I initialize multiple consumer processes and wait for messages to come into the diarizationInputQueue, diarize the input recordings and put the resulting json in the diarizationOutputQueue. I try to perform parallel diarizations using an extended class from `multiprocessing.Process`

The problem is that whenever I run multiple diarization processes (I run 2 consumers/processes), a job (input recording from the input queue) is consumed multiple times if there is an free process/consumer, not just once. This results in either overriding the other running process or resulting in a connection error, at some point ( the connection is not the real problem).

As it can be seen in this picture, at '07:17,225 job-id 10' is loaded into a consumer. When the other consumer becomes free, it loads 'job-id 10' at 09:17:667 again. 

![jobs](https://user-images.githubusercontent.com/53811655/225154018-06213ab6-850a-4418-b0b7-df2aea526ea7.png)

Another example is when I simply load the models and let each process sleep for a certain amount of time. In this case, after 'job-id:91' sleeps for 201 seconds, before the message is sent to the output queue, another 'job-id:91' process is initiated and pika throws an error.

```
2023-03-15 00:31:00,364 root INFO Message from queue: {'diarization-task': {'job-id': '91', 'diarization-id': '12345', 'file-path': '/home/gpirlogeanu/nemo_rttms/romana_wav/tel6.wav', 'num-speakers': None, 'sleep-time': 201}}
2023-03-15 00:32:24,503 root INFO Diarization ID: 12345 with jobID: 91 has been successfully postprocessed!
2023-03-15 00:32:24,504 pika.adapters.utils.io_services_utils ERROR _AsyncBaseTransport._produce() failed, aborting connection: error=ConnectionResetError(104, 'Connection reset by peer'); sock=<socket.socket fd=15, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('172.18.0.4', 58582)>; Caller's stack:
```

The system works fine when I am using a single consumer, so it is not an error in the scripts. What I have read is: 
``
Pika does not have any notion of threading in the code. If you want to use Pika with threading, make sure you have a Pika connection per thread, created in that thread. It is not safe to share one Pika connection across threads, with one exception: you may call the connection method add_callback_threadsafe from another thread to schedule a callback within an active pika connection.
``

NeMo initializes multiple threads per process and I think it conflicts with the pika library. Did any of you use a similar system or have any idea how I could make it NeMo work with Pika? Can NeMo run without multithreading? If I don't use NeMo, an input message is loaded just once in the current configuration. (I have tested it with sleep commands)

As you can see, a process runs on multiple threads:
![Screenshot from 2023-03-15 00-05-08](https://user-images.githubusercontent.com/53811655/225152474-37e37463-04b3-4644-ab31-9a44ae655a13.png)

My code: (I acknowledge messages "manually" after diarization is done)
``` 
def run(self):
        logging.info('Starting process to consume from Rabbit MQ...')
        while True:
            try:
                self.channel_receive.basic_consume(self.configFile['QUEUE_RECEIVE'], on_message_callback=self.callback)
                self.channel_receive.start_consuming()
            except Exception as e:
                logging.error(
                    "There seem to be connection issues! Error: %s", e)
                logging.error("Trying to establish a new connection...")
                self.createConnectionAndChannels()
                logging.info("Connection recovered!")
                continue
            except KeyboardInterrupt:
                logging.info("You manually stopped the app!")
                break
            time.sleep(5)
```

```
# Send the ACK for the finished task above
        try:
            if self.checkConnectionAndChannels() is False:
                self.createConnectionAndChannels()
            self.channel_receive.basic_ack(delivery_tag=method.delivery_tag)
            logging.info("Sending ACK after successully completed task for Diarization ID: " + diarizationID +  "with jobID: " +
                         jobID)
        except Exception as e:
            logging.error("Error while trying to send positive ack for Diarization ID:" +
                          diarizationID + " with jobID: " + jobID + " " + "Error: %s", e)
            self.channel_receive.basic_nack(delivery_tag=method.delivery_tag)
        logging.info("#######################################################")
        logging.info('\n')
```