dataset.py 27.8 KB
Newer Older
D
dongdaxiang 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13
#   Copyright (c) 2018 PaddlePaddle Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
14
"""This is defination of dataset class, which is high performance IO."""
D
dongdaxiang 已提交
15 16 17 18

from paddle.fluid.proto import data_feed_pb2
from google.protobuf import text_format
from . import core
D
dongdaxiang 已提交
19
__all__ = ['DatasetFactory', 'InMemoryDataset', 'QueueDataset']
D
dongdaxiang 已提交
20 21 22


class DatasetFactory(object):
23 24
    """
    DatasetFactory is a factory which create dataset by its name,
H
hutuxian 已提交
25
    you can create "QueueDataset" or "InMemoryDataset", or "FileInstantDataset",
26 27 28
    the default is "QueueDataset".

    Example:
29 30 31 32 33
        .. code-block:: python

          import paddle.fluid as fluid
          dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")

34
    """
D
dongdaxiang 已提交
35

D
dongdaxiang 已提交
36
    def __init__(self):
37
        """ Init. """
D
dongdaxiang 已提交
38 39
        pass

40
    def create_dataset(self, datafeed_class="QueueDataset"):
41
        """
H
hutuxian 已提交
42
        Create "QueueDataset" or "InMemoryDataset", or "FileInstantDataset",
43
        the default is "QueueDataset".
D
dongdaxiang 已提交
44

45 46 47 48
        Args:
            datafeed_class(str): datafeed class name, QueueDataset or InMemoryDataset.
                                 Default is QueueDataset.

D
dongdaxiang 已提交
49
        Examples:
50 51 52 53 54
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset()

55
        """
D
dongdaxiang 已提交
56 57
        try:
            dataset = globals()[datafeed_class]()
58
            return dataset
D
dongdaxiang 已提交
59 60 61 62 63 64
        except:
            raise ValueError("datafeed class %s does not exist" %
                             datafeed_class)


class DatasetBase(object):
65
    """ Base dataset class. """
D
dongdaxiang 已提交
66

D
dongdaxiang 已提交
67
    def __init__(self):
68
        """ Init. """
D
dongdaxiang 已提交
69 70 71 72
        # define class name here
        # to decide whether we need create in memory instance
        self.proto_desc = data_feed_pb2.DataFeedDesc()
        self.proto_desc.pipe_command = "cat"
X
xujiaqi01 已提交
73
        self.dataset = core.Dataset("MultiSlotDataset")
74
        self.thread_num = 1
J
jiaqi 已提交
75
        self.filelist = []
D
dongdaxiang 已提交
76 77 78 79 80 81

    def set_pipe_command(self, pipe_command):
        """
        Set pipe command of current dataset
        A pipe command is a UNIX pipeline command that can be used only

82 83 84 85 86 87
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset()
              dataset.set_pipe_command("python my_script.py")
88 89

        Args:
90
            pipe_command(str): pipe command
91

D
dongdaxiang 已提交
92 93 94
        """
        self.proto_desc.pipe_command = pipe_command

95 96 97 98 99 100 101 102
    def set_fea_eval(self, record_candidate_size, fea_eval=True):
        """
        set fea eval mode for slots shuffle to debug the importance level of
        slots(features), fea_eval need to be set True for slots shuffle.
        
        Args:
            record_candidate_size(int): size of instances candidate to shuffle 
                                        one slot
T
tianshuo78520a 已提交
103
            fea_eval(bool): whether enable fea eval mode to enable slots shuffle.
104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139
                            default is True.
            
        Examples:
            .. code-block:: python

            import paddle.fluid as fluid
            dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
            dataset.set_fea_eval(1000000, True)

        """
        if fea_eval:
            self.dataset.set_fea_eval(fea_eval, record_candidate_size)
        self.fea_eval = fea_eval

    def slots_shuffle(self, slots):
        """
        Slots Shuffle 
        Slots Shuffle is a shuffle method in slots level, which is usually used 
        in sparse feature with large scale of instances. To compare the metric, i.e.
        auc while doing slots shuffle on one or several slots with baseline to 
        evaluate the importance level of slots(features).
        
        Args:
            slots(list[string]): the set of slots(string) to do slots shuffle.

        Examples:
            import paddle.fluid as fluid
            dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
            dataset.set_merge_by_lineid()
            #suppose there is a slot 0
            dataset.slots_shuffle(['0'])
        """
        if self.fea_eval:
            slots_set = set(slots)
            self.dataset.slots_shuffle(slots_set)

D
dongdaxiang 已提交
140 141 142 143
    def set_batch_size(self, batch_size):
        """
        Set batch size. Will be effective during training

144 145 146 147 148 149
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset()
              dataset.set_batch_size(128)
D
dongdaxiang 已提交
150 151

        Args:
152
            batch_size(int): batch size
D
dongdaxiang 已提交
153 154 155 156

        """
        self.proto_desc.batch_size = batch_size

157
    def set_thread(self, thread_num):
158 159 160
        """
        Set thread num, it is the num of readers.

161 162 163 164 165 166
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset()
               dataset.set_thread(12)
167 168

        Args:
169
            thread_num(int): thread num
170
        """
171
        self.dataset.set_thread_num(thread_num)
172
        self.thread_num = thread_num
173 174

    def set_filelist(self, filelist):
175 176 177
        """
        Set file list in current worker.

178 179 180 181 182 183
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset()
              dataset.set_filelist(['a.txt', 'b.txt'])
184 185

        Args:
186
            filelist(list): file list
187
        """
188
        self.dataset.set_filelist(filelist)
J
jiaqi 已提交
189
        self.filelist = filelist
190

D
dongdaxiang 已提交
191
    def set_use_var(self, var_list):
192 193 194
        """
        Set Variables which you will use.

195 196 197 198 199 200
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset()
              dataset.set_use_var([data, label])
201 202

        Args:
203
            var_list(list): variable list
204
        """
205
        multi_slot = self.proto_desc.multi_slot_desc
D
dongdaxiang 已提交
206
        for var in var_list:
207
            slot_var = multi_slot.slots.add()
D
dongdaxiang 已提交
208 209 210 211
            slot_var.is_used = True
            slot_var.name = var.name
            if var.lod_level == 0:
                slot_var.is_dense = True
212
                slot_var.shape.extend(var.shape)
213
            if var.dtype == core.VarDesc.VarType.FP32:
D
dongdaxiang 已提交
214
                slot_var.type = "float"
215
            elif var.dtype == core.VarDesc.VarType.INT64:
D
dongdaxiang 已提交
216 217 218 219 220 221
                slot_var.type = "uint64"
            else:
                raise ValueError(
                    "Currently, fluid.dataset only supports dtype=float32 and dtype=int64"
                )

222
    def set_hdfs_config(self, fs_name, fs_ugi):
223 224 225
        """
        Set hdfs config: fs name ad ugi

226 227 228 229 230 231
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset()
              dataset.set_hdfs_config("my_fs_name", "my_fs_ugi")
232 233

        Args:
234 235
            fs_name(str): fs name
            fs_ugi(str): fs ugi
236
        """
237 238
        self.dataset.set_hdfs_config(fs_name, fs_ugi)

239
    def _prepare_to_run(self):
240 241 242 243
        """
        Set data_feed_desc before load or shuffle,
        user no need to call this function.
        """
J
jiaqi 已提交
244 245 246
        if self.thread_num > len(self.filelist):
            self.thread_num = len(self.filelist)
        self.dataset.set_thread_num(self.thread_num)
247
        self.dataset.set_data_feed_desc(self.desc())
J
jiaqi 已提交
248 249 250 251
        self.dataset.create_readers()

    def _finish_to_run(self):
        self.dataset.destroy_readers()
252

D
dongdaxiang 已提交
253 254 255 256
    def desc(self):
        """
        Returns a protobuf message for this DataFeedDesc

257 258 259 260 261 262
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset()
              print(dataset.desc())
D
dongdaxiang 已提交
263 264 265 266 267 268

        Returns:
            A string message
        """
        return text_format.MessageToString(self.proto_desc)

269 270 271 272 273 274
    def _dynamic_adjust_before_train(self, thread_num):
        pass

    def _dynamic_adjust_after_train(self):
        pass

D
dongdaxiang 已提交
275 276

class InMemoryDataset(DatasetBase):
277 278
    """
    InMemoryDataset, it will load data into memory
D
dongdaxiang 已提交
279 280
    and shuffle data before training.
    This class should be created by DatasetFactory
281 282

    Example:
283
        dataset = paddle.fluid.DatasetFactory().create_dataset("InMemoryDataset")
284
    """
D
dongdaxiang 已提交
285

D
dongdaxiang 已提交
286
    def __init__(self):
287
        """ Init. """
288 289
        super(InMemoryDataset, self).__init__()
        self.proto_desc.name = "MultiSlotInMemoryDataFeed"
290
        self.fleet_send_batch_size = None
291
        self.is_user_set_queue_num = False
J
jiaqi 已提交
292
        self.queue_num = None
293 294
        self.parse_ins_id = False
        self.parse_content = False
295
        self.merge_by_lineid = False
296
        self.fleet_send_sleep_seconds = None
J
jiaqi 已提交
297 298 299 300 301 302

    def _prepare_to_run(self):
        """
        Set data_feed_desc before load or shuffle,
        user no need to call this function.
        """
303
        if self.thread_num <= 0:
304
            self.thread_num = 1
J
jiaqi 已提交
305 306 307 308
        self.dataset.set_thread_num(self.thread_num)
        if self.queue_num is None:
            self.queue_num = self.thread_num
        self.dataset.set_queue_num(self.queue_num)
309 310
        self.dataset.set_parse_ins_id(self.parse_ins_id)
        self.dataset.set_parse_content(self.parse_content)
J
jiaqi 已提交
311 312 313 314
        self.dataset.set_data_feed_desc(self.desc())
        self.dataset.create_channel()
        self.dataset.create_readers()

315 316
    def _dynamic_adjust_before_train(self, thread_num):
        if not self.is_user_set_queue_num:
H
hutuxian 已提交
317
            self.dataset.dynamic_adjust_channel_num(thread_num, False)
318 319 320 321
        self.dataset.dynamic_adjust_readers_num(thread_num)

    def _dynamic_adjust_after_train(self):
        if not self.is_user_set_queue_num:
H
hutuxian 已提交
322
            self.dataset.dynamic_adjust_channel_num(self.thread_num, False)
323 324
        self.dataset.dynamic_adjust_readers_num(self.thread_num)

J
jiaqi 已提交
325 326 327 328 329
    def set_queue_num(self, queue_num):
        """
        Set Dataset output queue num, training threads get data from queues

        Args:
330
            queue_num(int): dataset output queue num
J
jiaqi 已提交
331 332 333 334 335 336 337 338 339

        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
              dataset.set_queue_num(12)

        """
340
        self.is_user_set_queue_num = True
J
jiaqi 已提交
341 342
        self.queue_num = queue_num

343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376
    def set_parse_ins_id(self, parse_ins_id):
        """
        Set id Dataset need to parse insid

        Args:
            parse_ins_id(bool): if parse ins_id or not

        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
              dataset.set_parse_ins_id(True)

        """
        self.parse_ins_id = parse_ins_id

    def set_parse_content(self, parse_content):
        """
        Set if Dataset need to parse content

        Args:
            parse_content(bool): if parse content or not

        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
              dataset.set_parse_content(True)

        """
        self.parse_content = parse_content

377
    def set_fleet_send_batch_size(self, fleet_send_batch_size=1024):
J
jiaqi 已提交
378
        """
379
        Set fleet send batch size, default is 1024
J
jiaqi 已提交
380 381 382 383 384 385 386 387 388 389 390 391 392

        Args:
            fleet_send_batch_size(int): fleet send batch size

        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
              dataset.set_fleet_send_batch_size(800)

        """
        self.fleet_send_batch_size = fleet_send_batch_size
393

394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410
    def set_fleet_send_sleep_seconds(self, fleet_send_sleep_seconds=0):
        """
        Set fleet send sleep time, default is 0

        Args:
            fleet_send_sleep_seconds(int): fleet send sleep time

        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
              dataset.set_fleet_send_sleep_seconds(2)

        """
        self.fleet_send_sleep_seconds = fleet_send_sleep_seconds

411
    def set_merge_by_lineid(self, merge_size=2):
412 413 414 415 416
        """
        Set merge by line id, instances of same line id will be merged after
        shuffle, you should parse line id in data generator.

        Args:
417
            merge_size(int): ins size to merge. default is 2.
418 419 420 421 422 423 424 425 426

        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
              dataset.set_merge_by_lineid()

        """
427
        self.dataset.set_merge_by_lineid(merge_size)
428
        self.merge_by_lineid = True
429
        self.parse_ins_id = True
430

431 432 433 434 435 436 437 438 439 440
    def set_generate_unique_feasigns(self, generate_uni_feasigns, shard_num):
        self.dataset.set_generate_unique_feasigns(generate_uni_feasigns)
        self.gen_uni_feasigns = generate_uni_feasigns
        self.local_shard_num = shard_num

    def generate_local_tables_unlock(self, table_id, fea_dim, read_thread_num,
                                     consume_thread_num, shard_num):
        self.dataset.generate_local_tables_unlock(
            table_id, fea_dim, read_thread_num, consume_thread_num, shard_num)

441
    def load_into_memory(self):
442 443 444
        """
        Load data into memory

445 446 447 448 449 450 451 452
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
              filelist = ["a.txt", "b.txt"]
              dataset.set_filelist(filelist)
              dataset.load_into_memory()
453
        """
454
        self._prepare_to_run()
455
        self.dataset.load_into_memory()
D
dongdaxiang 已提交
456

457
    def preload_into_memory(self, thread_num=None):
J
jiaqi 已提交
458 459 460
        """
        Load data into memory in async mode

461 462 463
        Args:
            thread_num(int): preload thread num

J
jiaqi 已提交
464 465 466 467 468 469 470 471 472 473 474
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
              filelist = ["a.txt", "b.txt"]
              dataset.set_filelist(filelist)
              dataset.preload_into_memory()
              dataset.wait_preload_done()
        """
        self._prepare_to_run()
475 476 477 478
        if thread_num is None:
            thread_num = self.thread_num
        self.dataset.set_preload_thread_num(thread_num)
        self.dataset.create_preload_readers()
J
jiaqi 已提交
479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495
        self.dataset.preload_into_memory()

    def wait_preload_done(self):
        """
        Wait preload_into_memory done

        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
              filelist = ["a.txt", "b.txt"]
              dataset.set_filelist(filelist)
              dataset.preload_into_memory()
              dataset.wait_preload_done()
        """
        self.dataset.wait_preload_done()
496
        self.dataset.destroy_preload_readers()
J
jiaqi 已提交
497

D
dongdaxiang 已提交
498
    def local_shuffle(self):
499 500 501
        """
        Local shuffle

502 503 504 505 506 507 508 509 510
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
              filelist = ["a.txt", "b.txt"]
              dataset.set_filelist(filelist)
              dataset.load_into_memory()
              dataset.local_shuffle()
511
        """
512
        self.dataset.local_shuffle()
D
dongdaxiang 已提交
513

514
    def global_shuffle(self, fleet=None, thread_num=12):
515 516
        """
        Global shuffle.
517 518 519
        Global shuffle can be used only in distributed mode. i.e. multiple
        processes on single machine or multiple machines training together.
        If you run in distributed mode, you should pass fleet instead of None.
520

521
        Examples:
522 523 524 525 526 527 528 529 530
            .. code-block:: python

              import paddle.fluid as fluid
              from paddle.fluid.incubate.fleet.parameter_server.pslib import fleet
              dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
              filelist = ["a.txt", "b.txt"]
              dataset.set_filelist(filelist)
              dataset.load_into_memory()
              dataset.global_shuffle(fleet)
531 532

        Args:
533
            fleet(Fleet): fleet singleton. Default None.
534
            thread_num(int): shuffle thread num. Default is 12.
535

536
        """
537 538
        trainer_num = 1
        if fleet is not None:
X
xujiaqi01 已提交
539
            fleet._role_maker.barrier_worker()
540
            trainer_num = fleet.worker_num()
541
        if self.fleet_send_batch_size is None:
542 543 544
            self.fleet_send_batch_size = 1024
        if self.fleet_send_sleep_seconds is None:
            self.fleet_send_sleep_seconds = 0
545
        self.dataset.register_client2client_msg_handler()
546
        self.dataset.set_trainer_num(trainer_num)
J
jiaqi 已提交
547
        self.dataset.set_fleet_send_batch_size(self.fleet_send_batch_size)
548
        self.dataset.set_fleet_send_sleep_seconds(self.fleet_send_sleep_seconds)
549
        if fleet is not None:
X
xujiaqi01 已提交
550
            fleet._role_maker.barrier_worker()
551
        self.dataset.global_shuffle(thread_num)
552
        if fleet is not None:
X
xujiaqi01 已提交
553
            fleet._role_maker.barrier_worker()
554 555 556
        if self.merge_by_lineid:
            self.dataset.merge_by_lineid()
        if fleet is not None:
X
xujiaqi01 已提交
557
            fleet._role_maker.barrier_worker()
D
dongdaxiang 已提交
558

559 560 561 562
    def release_memory(self):
        """
        Release InMemoryDataset memory data, when data will not be used again.

563 564 565 566 567 568 569 570 571 572 573 574 575 576 577
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              from paddle.fluid.incubate.fleet.parameter_server.pslib import fleet
              dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
              filelist = ["a.txt", "b.txt"]
              dataset.set_filelist(filelist)
              dataset.load_into_memory()
              dataset.global_shuffle(fleet)
              exe = fluid.Executor(fluid.CPUPlace())
              exe.run(fluid.default_startup_program())
              exe.train_from_dataset(fluid.default_main_program(), dataset)
              dataset.release_memory()

578 579
        """
        self.dataset.release_memory()
D
dongdaxiang 已提交
580

581 582 583 584 585 586 587 588 589 590 591 592 593 594
    def get_memory_data_size(self, fleet=None):
        """
        Get memory data size, user can call this function to know the num
        of ins in all workers after load into memory.

        Note:
            This function may cause bad performance, because it has barrier

        Args:
            fleet(Fleet): Fleet Object.

        Returns:
            The size of memory data.

595 596 597 598 599 600 601 602 603 604
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              from paddle.fluid.incubate.fleet.parameter_server.pslib import fleet
              dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
              filelist = ["a.txt", "b.txt"]
              dataset.set_filelist(filelist)
              dataset.load_into_memory()
              print dataset.get_memory_data_size(fleet)
605 606 607 608 609 610 611

        """
        import numpy as np
        local_data_size = self.dataset.get_memory_data_size()
        local_data_size = np.array([local_data_size])
        if fleet is not None:
            global_data_size = local_data_size * 0
X
xujiaqi01 已提交
612 613
            fleet._role_maker.all_reduce_worker(local_data_size,
                                                global_data_size)
614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631
            return global_data_size[0]
        return local_data_size[0]

    def get_shuffle_data_size(self, fleet=None):
        """
        Get shuffle data size, user can call this function to know the num
        of ins in all workers after local/global shuffle.

        Note:
            This function may cause bad performance to local shuffle,
            because it has barrier. It does not affect global shuffle.

        Args:
            fleet(Fleet): Fleet Object.

        Returns:
            The size of shuffle data.

632 633 634 635 636 637 638 639 640 641 642
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              from paddle.fluid.incubate.fleet.parameter_server.pslib import fleet
              dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
              filelist = ["a.txt", "b.txt"]
              dataset.set_filelist(filelist)
              dataset.load_into_memory()
              dataset.global_shuffle(fleet)
              print dataset.get_shuffle_data_size(fleet)
643 644 645 646 647 648 649

        """
        import numpy as np
        local_data_size = self.dataset.get_shuffle_data_size()
        local_data_size = np.array([local_data_size])
        if fleet is not None:
            global_data_size = local_data_size * 0
X
xujiaqi01 已提交
650 651
            fleet._role_maker.all_reduce_worker(local_data_size,
                                                global_data_size)
652 653 654
            return global_data_size[0]
        return local_data_size[0]

X
xjqbest 已提交
655

D
dongdaxiang 已提交
656
class QueueDataset(DatasetBase):
657 658 659
    """
    QueueDataset, it will process data streamly.

660 661 662 663 664 665
    Examples:
        .. code-block:: python

          import paddle.fluid as fluid
          dataset = fluid.DatasetFactory().create_dataset("QueueDataset")

666
    """
D
dongdaxiang 已提交
667

D
dongdaxiang 已提交
668
    def __init__(self):
669
        """
D
dongdaxiang 已提交
670 671
        Initialize QueueDataset
        This class should be created by DatasetFactory
672
        """
673
        super(QueueDataset, self).__init__()
D
dongdaxiang 已提交
674
        self.proto_desc.name = "MultiSlotDataFeed"
X
xujiaqi01 已提交
675

676 677 678 679 680 681 682 683 684 685 686 687 688 689
    def _prepare_to_run(self):
        """
        Set data_feed_desc/thread num/filelist before run,
        user no need to call this function.
        """
        if self.thread_num > len(self.filelist):
            self.thread_num = len(self.filelist)
        if self.thread_num == 0:
            self.thread_num = 1
        self.dataset.set_thread_num(self.thread_num)
        self.dataset.set_filelist(self.filelist)
        self.dataset.set_data_feed_desc(self.desc())
        self.dataset.create_readers()

X
xujiaqi01 已提交
690
    def local_shuffle(self):
691
        """
692
        Local shuffle data.
D
dongdaxiang 已提交
693

D
dongdaxiang 已提交
694 695
        Local shuffle is not supported in QueueDataset
        NotImplementedError will be raised
696 697 698 699 700 701 702 703

        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("QueueDataset")
              dataset.local_shuffle()

704 705 706
        Raises:
            NotImplementedError: QueueDataset does not support local shuffle

707
        """
D
dongdaxiang 已提交
708 709 710
        raise NotImplementedError(
            "QueueDataset does not support local shuffle, "
            "please use InMemoryDataset for local_shuffle")
X
xujiaqi01 已提交
711

712
    def global_shuffle(self, fleet=None):
713
        """
714 715
        Global shuffle data.

D
dongdaxiang 已提交
716 717
        Global shuffle is not supported in QueueDataset
        NotImplementedError will be raised
718

719 720 721
        Args:
            fleet(Fleet): fleet singleton. Default None.

722 723 724 725 726 727 728 729
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              from paddle.fluid.incubate.fleet.parameter_server.pslib import fleet
              dataset = fluid.DatasetFactory().create_dataset("QueueDataset")
              dataset.global_shuffle(fleet)

730 731 732
        Raises:
            NotImplementedError: QueueDataset does not support global shuffle

733
        """
D
dongdaxiang 已提交
734 735 736
        raise NotImplementedError(
            "QueueDataset does not support global shuffle, "
            "please use InMemoryDataset for global_shuffle")
H
hutuxian 已提交
737 738 739 740 741


class FileInstantDataset(DatasetBase):
    """
    FileInstantDataset, it will process data streamly.
742 743 744 745 746 747

    Examples:
        .. code-block:: python

          import paddle.fluid as fluid
          dataset = fluid.DatasetFactory.create_dataset("FileInstantDataset")
H
hutuxian 已提交
748 749 750 751
    """

    def __init__(self):
        """
752 753
        Initialize FileInstantDataset
        This class should be created by DatasetFactory
H
hutuxian 已提交
754 755 756 757 758 759
        """
        super(FileInstantDataset, self).__init__()
        self.proto_desc.name = "MultiSlotFileInstantDataFeed"

    def local_shuffle(self):
        """
760 761
        Local shuffle
        FileInstantDataset does not support local shuffle
H
hutuxian 已提交
762 763 764 765 766 767 768 769
        """
        raise NotImplementedError(
            "FileInstantDataset does not support local shuffle, "
            "please use InMemoryDataset for local_shuffle")

    def global_shuffle(self, fleet=None):
        """
        Global shuffle
770
        FileInstantDataset does not support global shuffle
H
hutuxian 已提交
771 772 773 774
        """
        raise NotImplementedError(
            "FileInstantDataset does not support global shuffle, "
            "please use InMemoryDataset for global_shuffle")
H
hutuxian 已提交
775 776 777 778 779 780 781 782 783 784


class BoxPSDataset(InMemoryDataset):
    """
    BoxPSDataset: derived from InMemoryDataset.

    Examples:
        .. code-block:: python

          import paddle.fluid as fluid
H
hutuxian 已提交
785
          dataset = fluid.DatasetFactory().create_dataset("BoxPSDataset")
H
hutuxian 已提交
786 787 788 789
    """

    def __init__(self):
        """
790 791
        Initialize BoxPSDataset
        This class should be created by DatasetFactory
H
hutuxian 已提交
792 793 794 795
        """
        super(BoxPSDataset, self).__init__()
        self.boxps = core.BoxPS(self.dataset)

H
hutuxian 已提交
796 797 798 799 800 801 802 803 804
    def set_date(self, date):
        """
        Workaround for date
        """
        year = int(date[:4])
        month = int(date[4:6])
        day = int(date[6:])
        self.boxps.set_date(year, month, day)

H
hutuxian 已提交
805 806
    def begin_pass(self):
        """
807
        Begin Pass
H
hutuxian 已提交
808 809 810 811 812 813 814 815 816
        Notify BoxPS to load sparse parameters of next pass to GPU Memory 

        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("BoxPSDataset")
              dataset.begin_pass()
        """
H
hutuxian 已提交
817 818 819 820
        self.boxps.begin_pass()

    def end_pass(self):
        """
821
        End Pass
H
hutuxian 已提交
822 823 824 825 826 827 828 829
        Notify BoxPS that current pass ended 
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("BoxPSDataset")
              dataset.end_pass()
        """
H
hutuxian 已提交
830 831 832 833
        self.boxps.end_pass()

    def wait_preload_done(self):
        """
T
tianshuo78520a 已提交
834
        Wait async preload done
835
        Wait Until Feed Pass Done
H
hutuxian 已提交
836 837 838 839 840 841 842 843 844 845
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("BoxPSDataset")
              filelist = ["a.txt", "b.txt"]
              dataset.set_filelist(filelist)
              dataset.preload_into_memory()
              dataset.wait_preload_done()
        """
H
hutuxian 已提交
846 847 848 849
        self.boxps.wait_feed_pass_done()

    def load_into_memory(self):
        """
H
hutuxian 已提交
850 851 852 853 854 855 856 857 858 859
        Load next pass into memory and notify boxps to fetch its emb from SSD
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("BoxPSDataset")
              filelist = ["a.txt", "b.txt"]
              dataset.set_filelist(filelist)
              dataset.load_into_memory()
	    """
H
hutuxian 已提交
860 861 862 863 864
        self._prepare_to_run()
        self.boxps.load_into_memory()

    def preload_into_memory(self):
        """
H
hutuxian 已提交
865 866 867 868 869 870 871 872 873 874
        Begin async preload next pass while current pass may be training
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("BoxPSDataset")
              filelist = ["a.txt", "b.txt"]
              dataset.set_filelist(filelist)
              dataset.preload_into_memory()
        """
H
hutuxian 已提交
875 876
        self._prepare_to_run()
        self.boxps.preload_into_memory()
H
hutuxian 已提交
877 878 879 880 881

    def _dynamic_adjust_before_train(self, thread_num):
        if not self.is_user_set_queue_num:
            self.dataset.dynamic_adjust_channel_num(thread_num, True)
        self.dataset.dynamic_adjust_readers_num(thread_num)