We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Queue full error - Multi-GPU 1M custom dataset
```
Traceback (most recent call last):
  File "/opt/conda/envs/learn-dev/lib/python3.7/multiprocessing/process.py", line 297, in _bootstrap
    self.run()
  File "/opt/conda/envs/learn-dev/lib/python3.7/multiprocessing/process.py", line 99, in run
    self._target(*self._args, **self._kwargs)
  File "/workspace/OpenNMT-py/train.py", line 127, in batch_producer
    q.put(b, False)
  File "/opt/conda/envs/learn-dev/lib/python3.7/multiprocessing/queues.py", line 83, in put
    raise Full
queue.Full
[2019-06-04 07:32:18,291 INFO] Step 1200/100000; acc:  79.56; ppl:  1.82; xent: 0.60; lr: 1.00000; 33996/13919 tok/s;    401 sec
Traceback (most recent call last):
  File "train.py", line 196, in <module>
    main(opt)
  File "train.py", line 78, in main
    p.join()
  File "/opt/conda/envs/learn-dev/lib/python3.7/multiprocessing/process.py", line 140, in join
    res = self._popen.wait(timeout)
  File "/opt/conda/envs/learn-dev/lib/python3.7/multiprocessing/popen_fork.py", line 48, in wait
    return self.poll(os.WNOHANG if timeout == 0.0 else 0)
  File "/opt/conda/envs/learn-dev/lib/python3.7/multiprocessing/popen_fork.py", line 28, in poll
    pid, sts = os.waitpid(self.pid, flag)
  File "train.py", line 180, in signal_handler
    raise Exception(msg)
Exception: 

-- Tracebacks above this line can probably
                 be ignored --

Traceback (most recent call last):
  File "/workspace/OpenNMT-py/train.py", line 138, in run
    single_main(opt, device_id, batch_queue, semaphore)
  File "/workspace/OpenNMT-py/onmt/train_single.py", line 139, in main
    valid_steps=opt.valid_steps)
  File "/workspace/OpenNMT-py/onmt/trainer.py", line 224, in train
    self._accum_batches(train_iter)):
  File "/workspace/OpenNMT-py/onmt/trainer.py", line 162, in _accum_batches
    for batch in iterator:
  File "/workspace/OpenNMT-py/onmt/train_single.py", line 116, in _train_iter
    batch = batch_queue.get()
  File "/opt/conda/envs/learn-dev/lib/python3.7/multiprocessing/queues.py", line 113, in get
    return _ForkingPickler.loads(res)
  File "/opt/conda/envs/learn-dev/lib/python3.7/site-packages/torch/multiprocessing/reductions.py", line 109, in rebuild_cuda_tensor
    event_sync_required)
RuntimeError: CUDA error: unknown error
```

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `train.py`
Content:
```
1 #!/usr/bin/env python
2 """Train models."""
3 import os
4 import signal
5 import torch
6 
7 import onmt.opts as opts
8 import onmt.utils.distributed
9 
10 from onmt.utils.misc import set_random_seed
11 from onmt.utils.logging import init_logger, logger
12 from onmt.train_single import main as single_main
13 from onmt.utils.parse import ArgumentParser
14 from onmt.inputters.inputter import build_dataset_iter, \
15     load_old_vocab, old_style_vocab, build_dataset_iter_multiple
16 
17 from itertools import cycle
18 
19 
20 def main(opt):
21     ArgumentParser.validate_train_opts(opt)
22     ArgumentParser.update_model_opts(opt)
23     ArgumentParser.validate_model_opts(opt)
24 
25     # Load checkpoint if we resume from a previous training.
26     if opt.train_from:
27         logger.info('Loading checkpoint from %s' % opt.train_from)
28         checkpoint = torch.load(opt.train_from,
29                                 map_location=lambda storage, loc: storage)
30         logger.info('Loading vocab from checkpoint at %s.' % opt.train_from)
31         vocab = checkpoint['vocab']
32     else:
33         vocab = torch.load(opt.data + '.vocab.pt')
34 
35     # check for code where vocab is saved instead of fields
36     # (in the future this will be done in a smarter way)
37     if old_style_vocab(vocab):
38         fields = load_old_vocab(
39             vocab, opt.model_type, dynamic_dict=opt.copy_attn)
40     else:
41         fields = vocab
42 
43     if len(opt.data_ids) > 1:
44         train_shards = []
45         for train_id in opt.data_ids:
46             shard_base = "train_" + train_id
47             train_shards.append(shard_base)
48         train_iter = build_dataset_iter_multiple(train_shards, fields, opt)
49     else:
50         train_iter = build_dataset_iter("train", fields, opt)
51 
52     nb_gpu = len(opt.gpu_ranks)
53 
54     if opt.world_size > 1:
55         queues = []
56         mp = torch.multiprocessing.get_context('spawn')
57         semaphore = mp.Semaphore(opt.world_size * opt.queue_size)
58         # Create a thread to listen for errors in the child processes.
59         error_queue = mp.SimpleQueue()
60         error_handler = ErrorHandler(error_queue)
61         # Train with multiprocessing.
62         procs = []
63         for device_id in range(nb_gpu):
64             q = mp.Queue(opt.queue_size)
65             queues += [q]
66             procs.append(mp.Process(target=run, args=(
67                 opt, device_id, error_queue, q, semaphore), daemon=True))
68             procs[device_id].start()
69             logger.info(" Starting process pid: %d  " % procs[device_id].pid)
70             error_handler.add_child(procs[device_id].pid)
71         producer = mp.Process(target=batch_producer,
72                               args=(train_iter, queues, semaphore, opt,),
73                               daemon=True)
74         producer.start()
75         error_handler.add_child(producer.pid)
76 
77         for p in procs:
78             p.join()
79         producer.terminate()
80 
81     elif nb_gpu == 1:  # case 1 GPU only
82         single_main(opt, 0)
83     else:   # case only CPU
84         single_main(opt, -1)
85 
86 
87 def batch_producer(generator_to_serve, queues, semaphore, opt):
88     init_logger(opt.log_file)
89     set_random_seed(opt.seed, False)
90     # generator_to_serve = iter(generator_to_serve)
91 
92     def pred(x):
93         """
94         Filters batches that belong only
95         to gpu_ranks of current node
96         """
97         for rank in opt.gpu_ranks:
98             if x[0] % opt.world_size == rank:
99                 return True
100 
101     generator_to_serve = filter(
102         pred, enumerate(generator_to_serve))
103 
104     def next_batch(device_id):
105         new_batch = next(generator_to_serve)
106         semaphore.acquire()
107         return new_batch[1]
108 
109     b = next_batch(0)
110 
111     for device_id, q in cycle(enumerate(queues)):
112         b.dataset = None
113         if isinstance(b.src, tuple):
114             b.src = tuple([_.to(torch.device(device_id))
115                            for _ in b.src])
116         else:
117             b.src = b.src.to(torch.device(device_id))
118         b.tgt = b.tgt.to(torch.device(device_id))
119         b.indices = b.indices.to(torch.device(device_id))
120         b.alignment = b.alignment.to(torch.device(device_id)) \
121             if hasattr(b, 'alignment') else None
122         b.src_map = b.src_map.to(torch.device(device_id)) \
123             if hasattr(b, 'src_map') else None
124 
125         # hack to dodge unpicklable `dict_keys`
126         b.fields = list(b.fields)
127         q.put(b, False)
128         b = next_batch(device_id)
129 
130 
131 def run(opt, device_id, error_queue, batch_queue, semaphore):
132     """ run process """
133     try:
134         gpu_rank = onmt.utils.distributed.multi_init(opt, device_id)
135         if gpu_rank != opt.gpu_ranks[device_id]:
136             raise AssertionError("An error occurred in \
137                   Distributed initialization")
138         single_main(opt, device_id, batch_queue, semaphore)
139     except KeyboardInterrupt:
140         pass  # killed by parent, do nothing
141     except Exception:
142         # propagate exception to parent process, keeping original traceback
143         import traceback
144         error_queue.put((opt.gpu_ranks[device_id], traceback.format_exc()))
145 
146 
147 class ErrorHandler(object):
148     """A class that listens for exceptions in children processes and propagates
149     the tracebacks to the parent process."""
150 
151     def __init__(self, error_queue):
152         """ init error handler """
153         import signal
154         import threading
155         self.error_queue = error_queue
156         self.children_pids = []
157         self.error_thread = threading.Thread(
158             target=self.error_listener, daemon=True)
159         self.error_thread.start()
160         signal.signal(signal.SIGUSR1, self.signal_handler)
161 
162     def add_child(self, pid):
163         """ error handler """
164         self.children_pids.append(pid)
165 
166     def error_listener(self):
167         """ error listener """
168         (rank, original_trace) = self.error_queue.get()
169         self.error_queue.put((rank, original_trace))
170         os.kill(os.getpid(), signal.SIGUSR1)
171 
172     def signal_handler(self, signalnum, stackframe):
173         """ signal handler """
174         for pid in self.children_pids:
175             os.kill(pid, signal.SIGINT)  # kill children processes
176         (rank, original_trace) = self.error_queue.get()
177         msg = """\n\n-- Tracebacks above this line can probably
178                  be ignored --\n\n"""
179         msg += original_trace
180         raise Exception(msg)
181 
182 
183 def _get_parser():
184     parser = ArgumentParser(description='train.py')
185 
186     opts.config_opts(parser)
187     opts.model_opts(parser)
188     opts.train_opts(parser)
189     return parser
190 
191 
192 if __name__ == "__main__":
193     parser = _get_parser()
194 
195     opt = parser.parse_args()
196     main(opt)
197
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/train.py b/train.py
--- a/train.py
+++ b/train.py
@@ -124,7 +124,7 @@
 
         # hack to dodge unpicklable `dict_keys`
         b.fields = list(b.fields)
-        q.put(b, False)
+        q.put(b)
         b = next_batch(device_id)

{"golden_diff": "diff --git a/train.py b/train.py\n--- a/train.py\n+++ b/train.py\n@@ -124,7 +124,7 @@\n \n         # hack to dodge unpicklable `dict_keys`\n         b.fields = list(b.fields)\n-        q.put(b, False)\n+        q.put(b)\n         b = next_batch(device_id)\n", "issue": "Queue full error - Multi-GPU 1M custom dataset\n```\r\nTraceback (most recent call last):\r\n  File \"/opt/conda/envs/learn-dev/lib/python3.7/multiprocessing/process.py\", line 297, in _bootstrap\r\n    self.run()\r\n  File \"/opt/conda/envs/learn-dev/lib/python3.7/multiprocessing/process.py\", line 99, in run\r\n    self._target(*self._args, **self._kwargs)\r\n  File \"/workspace/OpenNMT-py/train.py\", line 127, in batch_producer\r\n    q.put(b, False)\r\n  File \"/opt/conda/envs/learn-dev/lib/python3.7/multiprocessing/queues.py\", line 83, in put\r\n    raise Full\r\nqueue.Full\r\n[2019-06-04 07:32:18,291 INFO] Step 1200/100000; acc:  79.56; ppl:  1.82; xent: 0.60; lr: 1.00000; 33996/13919 tok/s;    401 sec\r\nTraceback (most recent call last):\r\n  File \"train.py\", line 196, in <module>\r\n    main(opt)\r\n  File \"train.py\", line 78, in main\r\n    p.join()\r\n  File \"/opt/conda/envs/learn-dev/lib/python3.7/multiprocessing/process.py\", line 140, in join\r\n    res = self._popen.wait(timeout)\r\n  File \"/opt/conda/envs/learn-dev/lib/python3.7/multiprocessing/popen_fork.py\", line 48, in wait\r\n    return self.poll(os.WNOHANG if timeout == 0.0 else 0)\r\n  File \"/opt/conda/envs/learn-dev/lib/python3.7/multiprocessing/popen_fork.py\", line 28, in poll\r\n    pid, sts = os.waitpid(self.pid, flag)\r\n  File \"train.py\", line 180, in signal_handler\r\n    raise Exception(msg)\r\nException: \r\n\r\n-- Tracebacks above this line can probably\r\n                 be ignored --\r\n\r\nTraceback (most recent call last):\r\n  File \"/workspace/OpenNMT-py/train.py\", line 138, in run\r\n    single_main(opt, device_id, batch_queue, semaphore)\r\n  File \"/workspace/OpenNMT-py/onmt/train_single.py\", line 139, in main\r\n    valid_steps=opt.valid_steps)\r\n  File \"/workspace/OpenNMT-py/onmt/trainer.py\", line 224, in train\r\n    self._accum_batches(train_iter)):\r\n  File \"/workspace/OpenNMT-py/onmt/trainer.py\", line 162, in _accum_batches\r\n    for batch in iterator:\r\n  File \"/workspace/OpenNMT-py/onmt/train_single.py\", line 116, in _train_iter\r\n    batch = batch_queue.get()\r\n  File \"/opt/conda/envs/learn-dev/lib/python3.7/multiprocessing/queues.py\", line 113, in get\r\n    return _ForkingPickler.loads(res)\r\n  File \"/opt/conda/envs/learn-dev/lib/python3.7/site-packages/torch/multiprocessing/reductions.py\", line 109, in rebuild_cuda_tensor\r\n    event_sync_required)\r\nRuntimeError: CUDA error: unknown error\r\n```\n", "before_files": [{"content": "#!/usr/bin/env python\n\"\"\"Train models.\"\"\"\nimport os\nimport signal\nimport torch\n\nimport onmt.opts as opts\nimport onmt.utils.distributed\n\nfrom onmt.utils.misc import set_random_seed\nfrom onmt.utils.logging import init_logger, logger\nfrom onmt.train_single import main as single_main\nfrom onmt.utils.parse import ArgumentParser\nfrom onmt.inputters.inputter import build_dataset_iter, \\\n    load_old_vocab, old_style_vocab, build_dataset_iter_multiple\n\nfrom itertools import cycle\n\n\ndef main(opt):\n    ArgumentParser.validate_train_opts(opt)\n    ArgumentParser.update_model_opts(opt)\n    ArgumentParser.validate_model_opts(opt)\n\n    # Load checkpoint if we resume from a previous training.\n    if opt.train_from:\n        logger.info('Loading checkpoint from %s' % opt.train_from)\n        checkpoint = torch.load(opt.train_from,\n                                map_location=lambda storage, loc: storage)\n        logger.info('Loading vocab from checkpoint at %s.' % opt.train_from)\n        vocab = checkpoint['vocab']\n    else:\n        vocab = torch.load(opt.data + '.vocab.pt')\n\n    # check for code where vocab is saved instead of fields\n    # (in the future this will be done in a smarter way)\n    if old_style_vocab(vocab):\n        fields = load_old_vocab(\n            vocab, opt.model_type, dynamic_dict=opt.copy_attn)\n    else:\n        fields = vocab\n\n    if len(opt.data_ids) > 1:\n        train_shards = []\n        for train_id in opt.data_ids:\n            shard_base = \"train_\" + train_id\n            train_shards.append(shard_base)\n        train_iter = build_dataset_iter_multiple(train_shards, fields, opt)\n    else:\n        train_iter = build_dataset_iter(\"train\", fields, opt)\n\n    nb_gpu = len(opt.gpu_ranks)\n\n    if opt.world_size > 1:\n        queues = []\n        mp = torch.multiprocessing.get_context('spawn')\n        semaphore = mp.Semaphore(opt.world_size * opt.queue_size)\n        # Create a thread to listen for errors in the child processes.\n        error_queue = mp.SimpleQueue()\n        error_handler = ErrorHandler(error_queue)\n        # Train with multiprocessing.\n        procs = []\n        for device_id in range(nb_gpu):\n            q = mp.Queue(opt.queue_size)\n            queues += [q]\n            procs.append(mp.Process(target=run, args=(\n                opt, device_id, error_queue, q, semaphore), daemon=True))\n            procs[device_id].start()\n            logger.info(\" Starting process pid: %d  \" % procs[device_id].pid)\n            error_handler.add_child(procs[device_id].pid)\n        producer = mp.Process(target=batch_producer,\n                              args=(train_iter, queues, semaphore, opt,),\n                              daemon=True)\n        producer.start()\n        error_handler.add_child(producer.pid)\n\n        for p in procs:\n            p.join()\n        producer.terminate()\n\n    elif nb_gpu == 1:  # case 1 GPU only\n        single_main(opt, 0)\n    else:   # case only CPU\n        single_main(opt, -1)\n\n\ndef batch_producer(generator_to_serve, queues, semaphore, opt):\n    init_logger(opt.log_file)\n    set_random_seed(opt.seed, False)\n    # generator_to_serve = iter(generator_to_serve)\n\n    def pred(x):\n        \"\"\"\n        Filters batches that belong only\n        to gpu_ranks of current node\n        \"\"\"\n        for rank in opt.gpu_ranks:\n            if x[0] % opt.world_size == rank:\n                return True\n\n    generator_to_serve = filter(\n        pred, enumerate(generator_to_serve))\n\n    def next_batch(device_id):\n        new_batch = next(generator_to_serve)\n        semaphore.acquire()\n        return new_batch[1]\n\n    b = next_batch(0)\n\n    for device_id, q in cycle(enumerate(queues)):\n        b.dataset = None\n        if isinstance(b.src, tuple):\n            b.src = tuple([_.to(torch.device(device_id))\n                           for _ in b.src])\n        else:\n            b.src = b.src.to(torch.device(device_id))\n        b.tgt = b.tgt.to(torch.device(device_id))\n        b.indices = b.indices.to(torch.device(device_id))\n        b.alignment = b.alignment.to(torch.device(device_id)) \\\n            if hasattr(b, 'alignment') else None\n        b.src_map = b.src_map.to(torch.device(device_id)) \\\n            if hasattr(b, 'src_map') else None\n\n        # hack to dodge unpicklable `dict_keys`\n        b.fields = list(b.fields)\n        q.put(b, False)\n        b = next_batch(device_id)\n\n\ndef run(opt, device_id, error_queue, batch_queue, semaphore):\n    \"\"\" run process \"\"\"\n    try:\n        gpu_rank = onmt.utils.distributed.multi_init(opt, device_id)\n        if gpu_rank != opt.gpu_ranks[device_id]:\n            raise AssertionError(\"An error occurred in \\\n                  Distributed initialization\")\n        single_main(opt, device_id, batch_queue, semaphore)\n    except KeyboardInterrupt:\n        pass  # killed by parent, do nothing\n    except Exception:\n        # propagate exception to parent process, keeping original traceback\n        import traceback\n        error_queue.put((opt.gpu_ranks[device_id], traceback.format_exc()))\n\n\nclass ErrorHandler(object):\n    \"\"\"A class that listens for exceptions in children processes and propagates\n    the tracebacks to the parent process.\"\"\"\n\n    def __init__(self, error_queue):\n        \"\"\" init error handler \"\"\"\n        import signal\n        import threading\n        self.error_queue = error_queue\n        self.children_pids = []\n        self.error_thread = threading.Thread(\n            target=self.error_listener, daemon=True)\n        self.error_thread.start()\n        signal.signal(signal.SIGUSR1, self.signal_handler)\n\n    def add_child(self, pid):\n        \"\"\" error handler \"\"\"\n        self.children_pids.append(pid)\n\n    def error_listener(self):\n        \"\"\" error listener \"\"\"\n        (rank, original_trace) = self.error_queue.get()\n        self.error_queue.put((rank, original_trace))\n        os.kill(os.getpid(), signal.SIGUSR1)\n\n    def signal_handler(self, signalnum, stackframe):\n        \"\"\" signal handler \"\"\"\n        for pid in self.children_pids:\n            os.kill(pid, signal.SIGINT)  # kill children processes\n        (rank, original_trace) = self.error_queue.get()\n        msg = \"\"\"\\n\\n-- Tracebacks above this line can probably\n                 be ignored --\\n\\n\"\"\"\n        msg += original_trace\n        raise Exception(msg)\n\n\ndef _get_parser():\n    parser = ArgumentParser(description='train.py')\n\n    opts.config_opts(parser)\n    opts.model_opts(parser)\n    opts.train_opts(parser)\n    return parser\n\n\nif __name__ == \"__main__\":\n    parser = _get_parser()\n\n    opt = parser.parse_args()\n    main(opt)\n", "path": "train.py"}], "after_files": [{"content": "#!/usr/bin/env python\n\"\"\"Train models.\"\"\"\nimport os\nimport signal\nimport torch\n\nimport onmt.opts as opts\nimport onmt.utils.distributed\n\nfrom onmt.utils.misc import set_random_seed\nfrom onmt.utils.logging import init_logger, logger\nfrom onmt.train_single import main as single_main\nfrom onmt.utils.parse import ArgumentParser\nfrom onmt.inputters.inputter import build_dataset_iter, \\\n    load_old_vocab, old_style_vocab, build_dataset_iter_multiple\n\nfrom itertools import cycle\n\n\ndef main(opt):\n    ArgumentParser.validate_train_opts(opt)\n    ArgumentParser.update_model_opts(opt)\n    ArgumentParser.validate_model_opts(opt)\n\n    # Load checkpoint if we resume from a previous training.\n    if opt.train_from:\n        logger.info('Loading checkpoint from %s' % opt.train_from)\n        checkpoint = torch.load(opt.train_from,\n                                map_location=lambda storage, loc: storage)\n        logger.info('Loading vocab from checkpoint at %s.' % opt.train_from)\n        vocab = checkpoint['vocab']\n    else:\n        vocab = torch.load(opt.data + '.vocab.pt')\n\n    # check for code where vocab is saved instead of fields\n    # (in the future this will be done in a smarter way)\n    if old_style_vocab(vocab):\n        fields = load_old_vocab(\n            vocab, opt.model_type, dynamic_dict=opt.copy_attn)\n    else:\n        fields = vocab\n\n    if len(opt.data_ids) > 1:\n        train_shards = []\n        for train_id in opt.data_ids:\n            shard_base = \"train_\" + train_id\n            train_shards.append(shard_base)\n        train_iter = build_dataset_iter_multiple(train_shards, fields, opt)\n    else:\n        train_iter = build_dataset_iter(\"train\", fields, opt)\n\n    nb_gpu = len(opt.gpu_ranks)\n\n    if opt.world_size > 1:\n        queues = []\n        mp = torch.multiprocessing.get_context('spawn')\n        semaphore = mp.Semaphore(opt.world_size * opt.queue_size)\n        # Create a thread to listen for errors in the child processes.\n        error_queue = mp.SimpleQueue()\n        error_handler = ErrorHandler(error_queue)\n        # Train with multiprocessing.\n        procs = []\n        for device_id in range(nb_gpu):\n            q = mp.Queue(opt.queue_size)\n            queues += [q]\n            procs.append(mp.Process(target=run, args=(\n                opt, device_id, error_queue, q, semaphore), daemon=True))\n            procs[device_id].start()\n            logger.info(\" Starting process pid: %d  \" % procs[device_id].pid)\n            error_handler.add_child(procs[device_id].pid)\n        producer = mp.Process(target=batch_producer,\n                              args=(train_iter, queues, semaphore, opt,),\n                              daemon=True)\n        producer.start()\n        error_handler.add_child(producer.pid)\n\n        for p in procs:\n            p.join()\n        producer.terminate()\n\n    elif nb_gpu == 1:  # case 1 GPU only\n        single_main(opt, 0)\n    else:   # case only CPU\n        single_main(opt, -1)\n\n\ndef batch_producer(generator_to_serve, queues, semaphore, opt):\n    init_logger(opt.log_file)\n    set_random_seed(opt.seed, False)\n    # generator_to_serve = iter(generator_to_serve)\n\n    def pred(x):\n        \"\"\"\n        Filters batches that belong only\n        to gpu_ranks of current node\n        \"\"\"\n        for rank in opt.gpu_ranks:\n            if x[0] % opt.world_size == rank:\n                return True\n\n    generator_to_serve = filter(\n        pred, enumerate(generator_to_serve))\n\n    def next_batch(device_id):\n        new_batch = next(generator_to_serve)\n        semaphore.acquire()\n        return new_batch[1]\n\n    b = next_batch(0)\n\n    for device_id, q in cycle(enumerate(queues)):\n        b.dataset = None\n        if isinstance(b.src, tuple):\n            b.src = tuple([_.to(torch.device(device_id))\n                           for _ in b.src])\n        else:\n            b.src = b.src.to(torch.device(device_id))\n        b.tgt = b.tgt.to(torch.device(device_id))\n        b.indices = b.indices.to(torch.device(device_id))\n        b.alignment = b.alignment.to(torch.device(device_id)) \\\n            if hasattr(b, 'alignment') else None\n        b.src_map = b.src_map.to(torch.device(device_id)) \\\n            if hasattr(b, 'src_map') else None\n\n        # hack to dodge unpicklable `dict_keys`\n        b.fields = list(b.fields)\n        q.put(b)\n        b = next_batch(device_id)\n\n\ndef run(opt, device_id, error_queue, batch_queue, semaphore):\n    \"\"\" run process \"\"\"\n    try:\n        gpu_rank = onmt.utils.distributed.multi_init(opt, device_id)\n        if gpu_rank != opt.gpu_ranks[device_id]:\n            raise AssertionError(\"An error occurred in \\\n                  Distributed initialization\")\n        single_main(opt, device_id, batch_queue, semaphore)\n    except KeyboardInterrupt:\n        pass  # killed by parent, do nothing\n    except Exception:\n        # propagate exception to parent process, keeping original traceback\n        import traceback\n        error_queue.put((opt.gpu_ranks[device_id], traceback.format_exc()))\n\n\nclass ErrorHandler(object):\n    \"\"\"A class that listens for exceptions in children processes and propagates\n    the tracebacks to the parent process.\"\"\"\n\n    def __init__(self, error_queue):\n        \"\"\" init error handler \"\"\"\n        import signal\n        import threading\n        self.error_queue = error_queue\n        self.children_pids = []\n        self.error_thread = threading.Thread(\n            target=self.error_listener, daemon=True)\n        self.error_thread.start()\n        signal.signal(signal.SIGUSR1, self.signal_handler)\n\n    def add_child(self, pid):\n        \"\"\" error handler \"\"\"\n        self.children_pids.append(pid)\n\n    def error_listener(self):\n        \"\"\" error listener \"\"\"\n        (rank, original_trace) = self.error_queue.get()\n        self.error_queue.put((rank, original_trace))\n        os.kill(os.getpid(), signal.SIGUSR1)\n\n    def signal_handler(self, signalnum, stackframe):\n        \"\"\" signal handler \"\"\"\n        for pid in self.children_pids:\n            os.kill(pid, signal.SIGINT)  # kill children processes\n        (rank, original_trace) = self.error_queue.get()\n        msg = \"\"\"\\n\\n-- Tracebacks above this line can probably\n                 be ignored --\\n\\n\"\"\"\n        msg += original_trace\n        raise Exception(msg)\n\n\ndef _get_parser():\n    parser = ArgumentParser(description='train.py')\n\n    opts.config_opts(parser)\n    opts.model_opts(parser)\n    opts.train_opts(parser)\n    return parser\n\n\nif __name__ == \"__main__\":\n    parser = _get_parser()\n\n    opt = parser.parse_args()\n    main(opt)\n", "path": "train.py"}]}