Farm: Deadlock in DataSilo._get_dataset when using docker

[trifle]

UPDATE/SOLUTION: PSA FOR POSTERITY
If you have a deadlock running FARM in a docker container, make sure you are running the container with --ipc=host to increase shared memory.
SOLUTION END

Describe the bug
There seems to be a multiprocessing-related deadlock in DataSilo._get_dataset, which transforms tsv lines into dicts into datasets, chunkwise. Reading a moderately-sized training set (~18k docs) stalls with zero CPU activity after around 2/3 of the data.

Error message
This is a rather unhelpful trace just like so many multiprocessing deadlocks:

Process ForkPoolWorker-23:
Process ForkPoolWorker-20:
Process ForkPoolWorker-22:
Process ForkPoolWorker-21:
Process ForkPoolWorker-19:
Process ForkPoolWorker-17:
Process ForkPoolWorker-18:
Traceback (most recent call last):
  File "/usr/lib/python3.6/multiprocessing/process.py", line 258, in _bootstrap
    self.run()
  File "/usr/lib/python3.6/multiprocessing/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/lib/python3.6/multiprocessing/pool.py", line 108, in worker
    task = get()
  File "/usr/lib/python3.6/multiprocessing/queues.py", line 334, in get
    with self._rlock:
  File "/usr/lib/python3.6/multiprocessing/synchronize.py", line 95, in __enter__
    return self._semlock.__enter__()
KeyboardInterrupt
Traceback (most recent call last):
  File "/usr/lib/python3.6/multiprocessing/process.py", line 258, in _bootstrap
    self.run()
  File "/usr/lib/python3.6/multiprocessing/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/lib/python3.6/multiprocessing/pool.py", line 108, in worker
    task = get()
  File "/usr/lib/python3.6/multiprocessing/queues.py", line 334, in get
    with self._rlock:
  File "/usr/lib/python3.6/multiprocessing/synchronize.py", line 95, in __enter__
    return self._semlock.__enter__()
Traceback (most recent call last):
Traceback (most recent call last):
KeyboardInterrupt
Traceback (most recent call last):
  File "/usr/lib/python3.6/multiprocessing/process.py", line 258, in _bootstrap
    self.run()
  File "/usr/lib/python3.6/multiprocessing/process.py", line 258, in _bootstrap
    self.run()
  File "/usr/lib/python3.6/multiprocessing/process.py", line 258, in _bootstrap
    self.run()
  File "/usr/lib/python3.6/multiprocessing/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/lib/python3.6/multiprocessing/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/lib/python3.6/multiprocessing/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/lib/python3.6/multiprocessing/pool.py", line 108, in worker
    task = get()
  File "/usr/lib/python3.6/multiprocessing/pool.py", line 108, in worker
    task = get()
  File "/usr/lib/python3.6/multiprocessing/pool.py", line 108, in worker
    task = get()
  File "/usr/lib/python3.6/multiprocessing/queues.py", line 334, in get
    with self._rlock:
  File "/usr/lib/python3.6/multiprocessing/queues.py", line 334, in get
    with self._rlock:
  File "/usr/lib/python3.6/multiprocessing/queues.py", line 334, in get
    with self._rlock:
  File "/usr/lib/python3.6/multiprocessing/synchronize.py", line 95, in __enter__
    return self._semlock.__enter__()
  File "/usr/lib/python3.6/multiprocessing/synchronize.py", line 95, in __enter__
    return self._semlock.__enter__()
  File "/usr/lib/python3.6/multiprocessing/synchronize.py", line 95, in __enter__
    return self._semlock.__enter__()
KeyboardInterrupt
KeyboardInterrupt
KeyboardInterrupt
Traceback (most recent call last):
Traceback (most recent call last):
  File "/usr/lib/python3.6/multiprocessing/process.py", line 258, in _bootstrap
    self.run()
  File "/usr/lib/python3.6/multiprocessing/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/lib/python3.6/multiprocessing/process.py", line 258, in _bootstrap
    self.run()
  File "/usr/lib/python3.6/multiprocessing/pool.py", line 108, in worker
    task = get()
  File "/usr/lib/python3.6/multiprocessing/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/lib/python3.6/multiprocessing/queues.py", line 334, in get
    with self._rlock:
  File "/usr/lib/python3.6/multiprocessing/pool.py", line 108, in worker
    task = get()
  File "/usr/lib/python3.6/multiprocessing/synchronize.py", line 95, in __enter__
    return self._semlock.__enter__()
  File "/usr/lib/python3.6/multiprocessing/queues.py", line 335, in get
    res = self._reader.recv_bytes()
  File "/usr/lib/python3.6/multiprocessing/connection.py", line 216, in recv_bytes
    buf = self._recv_bytes(maxlength)
  File "/usr/lib/python3.6/multiprocessing/connection.py", line 407, in _recv_bytes
    buf = self._recv(4)
KeyboardInterrupt
  File "/usr/lib/python3.6/multiprocessing/connection.py", line 379, in _recv
    chunk = read(handle, remaining)
KeyboardInterrupt
 60%|██████████████████████████████████████████████████████▋                                    | 10752/17908 [04:04<02:42, 44.04 Dicts/s]
Process ForkPoolWorker-24:
Traceback (most recent call last):
  File "/usr/lib/python3.6/multiprocessing/process.py", line 258, in _bootstrap
    self.run()
  File "/usr/lib/python3.6/multiprocessing/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/lib/python3.6/multiprocessing/pool.py", line 108, in worker
    task = get()
  File "/usr/lib/python3.6/multiprocessing/queues.py", line 334, in get
    with self._rlock:
  File "/usr/lib/python3.6/multiprocessing/synchronize.py", line 95, in __enter__
    return self._semlock.__enter__()
KeyboardInterrupt

Expected behavior
Dataset loading should successfully finish after processing the last chunk.

Additional context
I have tested that manually loading the data works:

# This works
train_dicts = processor.file_to_dicts("train.tsv")
train_dataset, tensor_names = processor.dataset_from_dicts(dicts=train_dicts)

Loading a subset of the first 10k docs works, too.

One hunch is that grouper(dicts, multiprocessing_chunk_size) under some condition produces a pathological chunk size.

To Reproduce
I'll try and see if I can come up with a synthetic reproducer that doesn't include my data (which I can't give out).

System:

• OS: Ubuntu 18.04 with nvidia-docker2 and a CUDA 10.0 image
• GPU/CPU: GTX 1080 / Xeon 4-core, 120GB RAM
• FARM version: master
  The system is otherwise idle, no file sytem contention, no excessive context switches, plenty of free RAM.

[trifle]

Update: I've further ruled out errors in the textual data by stripping quote chars and force-converting the docs to ascii - neither helps.

[Timoeller]

Hey @trifle thanks for reporting that issue, we recently made changes the way the data processor works and thought we improved it. Seems we did not improve it for all cases.

So lets find out what is going on:
Can you set a breakpoint where it calls the grouper(dicts, multiprocessing_chunk_size) and evaluate the call manually? Does it split the dicts into useful chunks?
Can you also tell the "multiprocessing_chunk_size" and if it is a integer value?

[trifle]

Hi @Timoeller,
thanks for helping out!
I had a brief look at the multiprocessing changeset - it's certainly a great idea in general. The data should be easy to get.

By the way, I haven't had the time to find out - you seem to require a split of at least two chunks to make use of a ConcatDataset. Does this imply any specific split ratio or granularity between the 2+ datasets contained within?

[trifle]

Ok, so I manually split the dataset into two chunks and this fixes the deadlock. Seems to definitely be a multiprocessing issue.

Here's the working hack in farm/data_handler/data_silo.py:

    def _get_dataset(self, filename):
        dicts = self.processor.file_to_dicts(filename)
        #shuffle list of dicts here if we later want to have a random dev set splitted from train set
        if self.processor.train_filename in filename:
            if not self.processor.dev_filename:
                if self.processor.dev_split > 0.0:
                    random.shuffle(dicts)

#     chunk processing/cpu count code here

        print('Processing part one of dicts')
        dataset1, tensor_names = self.processor.dataset_from_dicts(dicts=dicts[:int(len(dicts)/2)])
        print('Processing part two of dicts')
        dataset2, tensor_names = self.processor.dataset_from_dicts(dicts=dicts[int(len(dicts)/2):])
        return ConcatDataset([dataset1, dataset2]), tensor_names

[trifle]

Here are the chunking parameters:

num_cpus = min(mp.cpu_count(), 128) or 1
dicts_per_cpu = np.ceil(len(dicts) / num_cpus)
# automatic adjustment of multiprocessing chunksize
# for small files (containing few dicts) we want small chunksize to ulitize all available cores but never less
# than 2, because we need it to sample another random sentence in LM finetuning
# for large files we want to minimize processor spawning without giving too much data to one process, so we
# clip it at 5k
multiprocessing_chunk_size = int(np.clip((np.ceil(dicts_per_cpu/5)),a_min=2,a_max=5000))
dict_batches_to_process = int(len(dicts) / multiprocessing_chunk_size)
num_cpus_used = min(mp.cpu_count(), 128,  dict_batches_to_process) or 1
print(num_cpus_used, dicts_per_cpu, len(dicts), multiprocessing_chunk_size)

> (8, 2156.0, 17244, 432)

[Timoeller]

No worries, always happy to help.

You are right, splitting away a dev set from train is done by slicing ConcatDataset.
ConcatDataset is a list of small datasets. Unfortunately pytorch doesnt have a split method implemented on ConcatDataset and we had to do it ourselves.

(8, 2156.0, 17244, 432)

That should be fine. How did you set the dev_split param?

[trifle]

I tried both the default (0.1 dev ratio) and supplying my own dev sample via a file. Neither helped.

[Timoeller]

Ok, cool so the splitting is not the issue then.

Did I understand correctly. You manually transformed dataset1, tensor_names = self.processor.dataset_from_dicts(dicts=dicts[:int(len(dicts)/2)]) but without parallelization. Why did you split it into two chunks then? It should also work when calling it on the whole dict - so basically in single process mode.

Anyways, I will have our data engineer look into this as well.

[trifle]

Yes, I serially processed the first and then the second half of the data, and then combined it into a ConcatDataset.

If you don't do that (if your ConcatDataset has only one entry), then you'll get an IndexError later on. FARM tries to access a tensor to convert it to a np array and finds a tuple instead. I forgot the exact details, but it seemed like a nesting issue where ConcatDataset was expected to contain >= 2 entries.

And many thanks for the help! I'm trying to get a minimal reproducible example with dummy data, I'll update when/if I have that.

[trifle]

Okay, here you go! This reproduces the issue once you set the number of dicts to a large value (it hung with 20k).

import multiprocessing as mp
import numpy as np
from contextlib import ExitStack
from itertools import islice
from tqdm import tqdm
from functools import partial

from farm.modeling.tokenization import BertTokenizer
from farm.data_handler.processor import TextClassificationProcessor

max_processes = 128
lang_model = "bert-base-german-cased"

tokenizer = BertTokenizer.from_pretrained(
    pretrained_model_name_or_path=lang_model,
    do_lower_case=False)

label_list = ['a',]
metric = "f1_macro"

processor = TextClassificationProcessor(tokenizer=tokenizer,
                                        max_seq_len=256,
                                        data_dir="GUESS YOU NEED THIS",
                                        train_filename="GUESS YOU NEED THIS",
                                        test_filename="GUESS YOU NEED THIS",
                                        dev_filename="GUESS YOU NEED THIS",
                                        quote_char='"',
                                        label_list=label_list,
                                        metric=metric,
                                        label_column_name="label",
                                        )



def grouper(iterable, n):
    """
    >>> list(grouper('ABCDEFG'), 3)
    [['A', 'B', 'C'], ['D', 'E', 'F'], ['G']]
    """
    iterable = iter(enumerate(iterable))
    return iter(lambda: list(islice(iterable, n)), [])


def _multiproc(chunk, this_processor):
    dicts = [d[1] for d in chunk]
    dataset = this_processor.dataset_from_dicts(dicts=dicts)
    return dataset

dummy_text= 'Verfassung und Verschwörungstheorie. Amerikas Präsident Donald Trump nennt die Gründe für ein Amtenthebungsverfahren einen "Witz". Bis jetzt gab es zwei Amtsanklagen gegen Präsidenten in Amerika, beide jedoch scheiterten – aus gutem Grund. Welche Wege zur Absetzung von Donald Trump wahrscheinlicher sind.'

def test_mp():
    dicts = [{'text': f'{dummy_text}_{i}','text_classification_label': 'a'} for i in range(20000)]
    num_cpus = min(mp.cpu_count(), max_processes) or 1
    dicts_per_cpu = np.ceil(len(dicts) / num_cpus)

    multiprocessing_chunk_size = int(
        np.clip((np.ceil(dicts_per_cpu / 5)), a_min=2, a_max=5000))
    dict_batches_to_process = int(len(dicts) / multiprocessing_chunk_size)
    num_cpus_used = min(mp.cpu_count(), max_processes,
                        dict_batches_to_process) or 1
    print(num_cpus_used, max_processes, multiprocessing_chunk_size, len(dicts))

    with ExitStack() as stack:
        p = stack.enter_context(mp.Pool(processes=num_cpus_used))

        results = p.imap(
            partial(_multiproc, this_processor=processor),
            grouper(dicts, multiprocessing_chunk_size),
            chunksize=1,
        )

        datasets = []
        with tqdm(total=len(dicts), unit=' Dicts') as pbar:
            for dataset, tensor_names in results:
                datasets.append(dataset)
                pbar.update(multiprocessing_chunk_size)

        concat_datasets = ConcatDataset(datasets)
        return concat_datasets, tensor_names

[trifle]

By the way, the docstring for the grouper is wrong (outdated), but that's irrelevant here :)

A fix would be:
"""

list(grouper('ABCDEFG', 3))
[[(0, 'A'), (1, 'B'), (2, 'C')], [(3, 'D'), (4, 'E'), (5, 'F')], [(6, 'G')]]
"""

[Timoeller]

Thanks for isolating the bug!
We will look into it!

By the way, the docstring for the grouper is wrong (outdated), but that's irrelevant here :)

If you want, create a PR, I will merge it into master asap and you are officially a contributor : )

[Timoeller]

Ok, I tested your code locally without docker and it works without deadlock.
So can you change the title of the issue to
"Deadlock in DataSilo._get_dataset when using docker". I guess this will complicate things. @tanaysoni do you have any idea what could have happened?

Another remark (that shouldnt change much): use import torch.multiprocessing as mp instead of multiprocessing

[tanaysoni]

Hi @trifle, thank you for all the inputs on the issue! I am trying to reproduce — can you share the Python version and the requirements file for the environment(particularly torch version)?

Also, is the issue exclusive to Docker? If yes, can you try running Docker with ipc=host as documented in https://github.com/pytorch/pytorch#docker-image.

[trifle]

Thanks for following up so quickly!
I didn't know about docker's shared memory limitations, that's a very good pointer @tanaysoni. Should be quick to test out. Memory pressure seems like a plausible cause for this complete stall without Exceptions.

The torch versions tested were 1.3.0 and 1.2.0, both of which exhibited the issue, by the way.

[trifle]

Yes, this was it! The reproducer runs through without any issue after launching the docker container with --ipc=host.

Thanks a lot, @tanaysoni for the quick help.

Since this issue was basically nothing but a time sink for you (sorry!), I want it to end on some constructive note :)

It seems that the best way of detecting this issue would be to query for the size of /dev/shm/.
The difference is obvious:
host system:

$ df /dev/shm/
Filesystem     1K-blocks  Used Available Use% Mounted on
tmpfs           57702028     0  57702028   0% /dev/shm

Inside a default docker container:

# df /dev/shm/
Filesystem     1K-blocks  Used Available Use% Mounted on
shm                65536     0     65536   0% /dev/shm

Running a check against that could issue a warning.
Although, to be honest, that is probably something that would be worth including in torch upstream rather than here.

[Timoeller]

Since this issue was basically nothing but a time sink for you (sorry!), I want it to end on some constructive note :)

hehe : )
so, interacting with somebody who knows how to code and use Farm is never a time sink but always some knowledge learned. Hope you can use Farm in your future projects.