Skip to content

Split workload when use multiprocessing for loading data. #151

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 4 commits into
base: dev
Choose a base branch
from
Open

Split workload when use multiprocessing for loading data. #151

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
9e158f0
fix dtype issue
matthew-z Dec 13, 2019
49548ad
Merge pull request #119 from NTMC-Community/padding_dtype
matthew-z Dec 14, 2019
0e5c04e
Update turorials/ranking/bert.ipynb (#149)
Albert-Ma Sep 26, 2020
9943969
Split workload when use multiprocessing to loading data.
Albert-Ma Sep 25, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions matchzoo/dataloader/callbacks/padding.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -130,8 +130,8 @@ def on_batch_unpacked(self, x: dict, y: np.ndarray):
"""Pad `x['text_left']` and `x['text_right]`."""

batch_size = len(x['id_left'])
pad_length_left = max(x['length_left'])
pad_length_right = max(x['length_right'])
pad_length_left = int(max(x['length_left']))
pad_length_right = int(max(x['length_right']))
if self._with_ngram:
ngram_length_left = max([len(w)
for k in x['ngram_left'] for w in k])
Expand Down
16 changes: 14 additions & 2 deletions matchzoo/dataloader/dataset.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -116,8 +116,20 @@ def __iter__(self):
"""Create a generator that iterate over the Batches."""
if self._resample or self._shuffle:
self.on_epoch_end()
for i in range(len(self)):
yield self[i]

worker_info = data.get_worker_info()
if worker_info is None:
# single-process
for i in range(len(self)):
yield self[i]
else:
# multi-process, split workload across all workers
per_work_load = int(math.ceil(len(self) / float(worker_info.num_workers)))
worker_id = worker_info.id
iter_start = worker_id * per_work_load
iter_end = min(iter_start + per_work_load, len(self))
for i in range(iter_start, iter_end):
yield self[i]

def on_epoch_end(self):
"""Reorganize the index array if needed."""
Expand Down
8 changes: 4 additions & 4 deletions tutorials/ranking/bert.ipynb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,9 +70,13 @@
" mode='pair',\n",
" num_dup=2,\n",
" num_neg=1\n",
" resample=True,\n",
" sort=False,\n",
" batch_size=20,\n",
")\n",
"testset = mz.dataloader.Dataset(\n",
" data_pack=test_pack_processed\n",
" batch_size=20,\n",
")"
]
},
Expand All @@ -85,15 +89,11 @@
"padding_callback = mz.models.Bert.get_default_padding_callback()\n",
"trainloader = mz.dataloader.DataLoader(\n",
" dataset=trainset,\n",
" batch_size=20,\n",
" stage='train',\n",
" resample=True,\n",
" sort=False,\n",
" callback=padding_callback\n",
")\n",
"testloader = mz.dataloader.DataLoader(\n",
" dataset=testset,\n",
" batch_size=20,\n",
" stage='dev',\n",
" callback=padding_callback\n",
")"
Expand Down