Add Seq2seq example

chainermn/__init__.py

@@ -2,6 +2,7 @@
 
 from chainermn.communicators import create_communicator  # NOQA
 from chainermn.dataset import scatter_dataset  # NOQA
+from chainermn.dataset import DataSizeError  # NOQA
 from chainermn.link import MultiNodeChainList  # NOQA
 from chainermn.multi_node_evaluator import create_multi_node_evaluator  # NOQA
 from chainermn.multi_node_optimizer import create_multi_node_optimizer  # NOQA

chainermn/dataset.py

@@ -1,7 +1,53 @@
-import chainer.datasets
+import math
 import numpy
+import re
 import warnings
 
+import chainer.datasets
+
+
+class DataSizeError(RuntimeError):
+    def __init__(self, ds_size, pickled_size):
+        msg = """The dataset was too large to be scattered using MPI.
+
+        The length of the dataset is {} and it's size after being pickled
+        was {}. In the current MPI specification, the size cannot exceed
+        {}, which is so called 'INT_MAX'.
+
+        To solve this problem, please split the dataset into multiple
+        peaces and send/recv them separately.
+
+        Recommended sizes are indicated by ``slices()`` method.
+        """
+
+        INT_MAX = 2147483647
+        msg = msg.format(ds_size, pickled_size, INT_MAX)
+        super(DataSizeError, self).__init__(self, msg)
+
+        self.pickled_size = pickled_size
+        self.max_size = INT_MAX
+        self.dataset_len = ds_size
+
+    def num_split(self):
+        ps = self.pickled_size
+        mx = self.max_size
+        return (ps + mx - 1) // mx
+
+    def slices(self):
+        ds = self.dataset_len
+        nsplit = self.num_split()
+        size = math.ceil(ds / nsplit)
+
+        return [(b, min(e, ds)) for b, e in
+                ((i * size, (i + 1) * size) for i in range(0, nsplit))]
+
+
+def _parse_overflow_error(err):
+    msg = str(err)
+    m = re.search(r'integer (\d+) does not fit in', msg)
+    assert m is not None, "'{}' must include size of the message".format(msg)
+    return int(m.group(1))
+
 
 def scatter_dataset(dataset, comm, root=0, shuffle=False, seed=None):
     """Scatter the given dataset to the workers in the communicator.
@@ -53,10 +99,19 @@ def scatter_dataset(dataset, comm, root=0, shuffle=False, seed=None):
             if i == root:
                 mine = subds
             else:
-                comm.send(subds, dest=i)
+                try:
+                    comm.send(subds, dest=i)
+                except OverflowError as e:
+                    pickled_size = _parse_overflow_error(e)
+                    raise DataSizeError(len(dataset), pickled_size)
+
         return mine
     else:
-        return comm.recv(source=root)
+        try:
+            return comm.recv(source=0)
+        except OverflowError as e:
+            pickled_size = _parse_overflow_error(e)
+            raise DataSizeError(len(dataset), pickled_size)
 
 
 def get_n_iterations_for_one_epoch(dataset, local_batch_size, comm):

examples/seq2seq/README.md

@@ -0,0 +1,32 @@
+# ChainerMN seq2seq example
+
+An sample implementation of seq2seq model.
+
+## Data download and setup
+
+First, go to http://www.statmt.org/wmt15/translation-task.html#download and donwload necessary dataset.
+Let's assume you are in a working directory called `$WMT_DIR`.
+
+```
+$ cd $WMT_DIR
+$ wget http://www.statmt.org/wmt10/training-giga-fren.tar
+$ wget http://www.statmt.org/wmt15/dev-v2.tgz
+$ tar -xf training-giga-fren.tar
+$ tar -xf dev-v2.tgz
+$ ls 
+dev/  dev-v2.tgz  giga-fren.release2.fixed.en.gz  giga-fren.release2.fixed.fr.gz  training-giga-fren.tar
+
+```
+
+Next, you need to install required packages.
+
+```
+$ pip install nltk progressbar2
+
+## Run
+
+```bash
+$ cd $CHAINERMN
+```
+
+

examples/seq2seq/europal.orig.py

@@ -0,0 +1,82 @@
+from __future__ import unicode_literals
+
+import collections
+import gzip
+import io
+import os
+import re
+
+import numpy
+import progressbar
+
+
+split_pattern = re.compile(r'([.,!?"\':;)(])')
+digit_pattern = re.compile(r'\d')
+
+
+def split_sentence(s):
+    s = s.lower()
+    s = s.replace('\u2019', "'")
+    s = digit_pattern.sub('0', s)
+    words = []
+    for word in s.strip().split():
+        words.extend(split_pattern.split(word))
+    words = [w for w in words if w]
+    return words
+
+
+def open_file(path):
+    if path.endswith('.gz'):
+        return gzip.open(path, 'rt', 'utf-8')
+    else:
+        # Find gzipped version of the file
+        gz = path + '.gz'
+        if os.path.exists(gz):
+            return open_file(gz)
+        else:
+            return io.open(path, encoding='utf-8', errors='ignore')
+
+
+def count_lines(path):
+    with open_file(path) as f:
+        return sum([1 for _ in f])
+
+
+def read_file(path):
+    n_lines = count_lines(path)
+    bar = progressbar.ProgressBar()
+    with open_file(path) as f:
+        for line in bar(f, max_value=n_lines):
+            words = split_sentence(line)
+            yield words
+
+
+def count_words(path):
+    counts = collections.Counter()
+    for words in read_file(path):
+        for word in words:
+            counts[word] += 1
+
+    vocab = [word for (word, _) in counts.most_common(40000)]
+    return vocab
+
+
+def make_dataset(path, vocab):
+    word_id = {word: index for index, word in enumerate(vocab)}
+    dataset = []
+    token_count = 0
+    unknown_count = 0
+    for words in read_file(path):
+        array = make_array(word_id, words)
+        dataset.append(array)
+        token_count += array.size
+        unknown_count += (array == 1).sum()
+    print('# of tokens: %d' % token_count)
+    print('# of unknown: %d (%.2f %%)'
+          % (unknown_count, 100. * unknown_count / token_count))
+    return dataset
+
+
+def make_array(word_id, words):
+    ids = [word_id.get(word, 1) for word in words]
+    return numpy.array(ids, 'i')

examples/seq2seq/europal.py

@@ -0,0 +1,88 @@
+from __future__ import unicode_literals
+
+import collections
+import gzip
+import io
+import os
+import re
+
+import numpy
+import progressbar
+
+
+split_pattern = re.compile(r'([.,!?"\':;)(])')
+digit_pattern = re.compile(r'\d')
+
+
+def split_sentence(s):
+    s = s.lower()
+    s = s.replace('\u2019', "'")
+    s = digit_pattern.sub('0', s)
+    words = []
+    for word in s.strip().split():
+        words.extend(split_pattern.split(word))
+    words = [w for w in words if w]
+    return words
+
+
+def open_file(path):
+    if path.endswith('.gz'):
+        return gzip.open(path, 'rt', encoding='utf-8')
+    else:
+        # Find gzipped version of the file
+        gz = path + '.gz'
+        if os.path.exists(gz):
+            return open_file(gz)
+        else:
+            return io.open(path, encoding='utf-8', errors='ignore')
+
+
+def count_lines(path):
+    print(path)
+    with open_file(path) as f:
+        return sum([1 for _ in f])
+
+
+def read_file(path):
+    n_lines = count_lines(path)
+    bar = progressbar.ProgressBar()
+    with open_file(path) as f:
+        for line in bar(f, max_value=n_lines):
+            words = split_sentence(line)
+            yield words
+
+
+def count_words(path):
+    counts = collections.Counter()
+    for words in read_file(path):
+        for word in words:
+            counts[word] += 1
+
+    vocab = [word for (word, _) in counts.most_common(40000)]
+    return vocab
+
+
+def make_dataset(path, vocab):
+    word_id = {word: index for index, word in enumerate(vocab)}
+    dataset = []
+    token_count = 0
+    unknown_count = 0
+    for words in read_file(path):
+        array = make_array(word_id, words)
+        dataset.append(array)
+        token_count += array.size
+        unknown_count += (array == 1).sum()
+    print('# of tokens: %d' % token_count)
+    print('# of unknown: %d (%.2f %%)'
+          % (unknown_count, 100. * unknown_count / token_count))
+    return dataset
+
+
+def make_array(word_id, words):
+    ids = [word_id.get(word, 1) for word in words]
+    return numpy.array(ids, 'i')
+
+
+if __name__ == '__main__':
+    vocab = count_words('wmt/giga-fren.release2.fixed.en')
+    make_dataset('wmt/giga-fren.release2.fixed.en', vocab)