add checkpoint util class and implement

paddle/fluid/operators/listen_and_serv_op.cc

@@ -187,6 +187,7 @@ void ListenAndServOp::RunSyncLoop(framework::Executor *executor,
     for (auto &var : sparse_vars) {
       var->GetMutable<framework::SelectedRows>()->mutable_rows()->clear();
     }
+
     rpc_service_->SetCond(1);
     // FIXME(typhoonzero): use another condition to sync wait clients get.
     rpc_service_->WaitClientGet(fan_in);

python/paddle/fluid/io.py

@@ -13,21 +13,17 @@
 # limitations under the License.
 
 import os
+import time
+import shutil
 
 from paddle.fluid.evaluator import Evaluator
 from paddle.fluid.framework import Program, Parameter, default_main_program, Variable
 from . import core
 
 __all__ = [
-    'save_vars',
-    'save_params',
-    'save_persistables',
-    'load_vars',
-    'load_params',
-    'load_persistables',
-    'save_inference_model',
-    'load_inference_model',
-    'get_inference_program',
+    'save_vars', 'save_params', 'save_persistables', 'load_vars', 'load_params',
+    'load_persistables', 'save_inference_model', 'load_inference_model',
+    'get_inference_program', 'save_checkpoint', 'load_checkpoint'
 ]
 
 
@@ -195,6 +191,8 @@ def load_vars(executor,
         load_var_map = {}
         for each_var in vars:
             assert isinstance(each_var, Variable)
+            if each_var.type == core.VarDesc.VarType.RAW:
+                continue
             new_var = _clone_var_in_block_(load_block, each_var)
             if filename is None:
                 load_block.append_op(
@@ -454,3 +452,167 @@ def get_parameter_value_by_name(name, executor, program=None):
         program = default_main_program()
     var = program.global_block().var(name)
     return get_parameter_value(var, executor)
+
+
+SUCCESS_MARK_FILENAME = "_SUCCESS"
+
+
+def save_checkpoint(executor,
+                    dirname=None,
+                    max_num_checkpoints=3,
+                    save_interval_secs=600,
+                    main_program=None):
+    """
+    Save Checkpoint will save persistable LodTensor variables from main_program in checkpoint directory,
+    directory named by serial number from 0 to (n -1), save_checkpoint use LRU strategy
+    to keep numbers of checkpoint directory,  the numbers of checkpoint directory are max_num_checkpoints at most,
+    The interval time between two save_checkpoint must great than or equal to save_interval_secs.
+
+    :param dirname
+    :param max_num_checkpoints
+    :param save_secs
+    :param main_program
+    """
+    if dirname is None:
+        dirname = os.getcwd()
+
+    if not os.path.isdir(dirname):
+        os.makedirs(dirname)
+
+    serial = _get_lastest_checkpoint_dir(dirname)
+    if serial >= 0 and not _interval_secs_exceed(
+            os.path.join(dirname, str(serial)), save_interval_secs):
+        return
+
+    serial = serial + 1
+    cur_dir = os.path.join(dirname, str(serial))
+
+    save_vars(
+        executor,
+        dirname=cur_dir,
+        main_program=main_program,
+        vars=None,
+        predicate=_is_checkpoint_var,
+        filename=None)
+    _write_success(cur_dir)
+    _lru_delete(dirname, max_num_checkpoints)
+
+
+def load_checkpoint(executor, dirname=None, main_program=None):
+    """
+    Load checkpoint from directory by executor,
+    it will find lastest checkpoint file and load it auto.
+
+    :param executor
+    :param dirname
+    :param main_program
+    """
+
+    if dirname is None:
+        dirname = os.getcwd()
+
+    serial = _get_lastest_checkpoint_dir(dirname)
+
+    if serial < 0:
+        return
+    cur_dir = os.path.join(dirname, str(serial))
+
+    load_vars(
+        executor,
+        dirname=cur_dir,
+        main_program=main_program,
+        predicate=_is_checkpoint_var,
+        filename=None)
+
+
+def _is_checkpoint_var(var):
+    """
+    checkpoint will not save or load all the variables.
+    var type is FEED_MINIBATCH/FETCH_LIST/RAW and var name is end with @GRAD are discarded.
+
+    :param var
+    """
+    if var.desc.type() == core.VarDesc.VarType.FEED_MINIBATCH or \
+            var.desc.type() == core.VarDesc.VarType.FETCH_LIST or \
+            var.desc.type() == core.VarDesc.VarType.RAW:
+        return False
+
+    if var.name.endswith("@GRAD"):
+        return False
+
+    return var.persistable
+
+
+def _interval_secs_exceed(dirname, save_interval_secs):
+    dir_time = os.path.getmtime(dirname)
+    if save_interval_secs > (time.time() - dir_time):
+        return False
+    return True
+
+
+def _lru_delete(dirname, max_num_checkpoints=3):
+    dirs = os.listdir(dirname)
+    serials = []
+    for serial in dirs:
+        try:
+            serials.append(int(serial))
+        except ValueError:
+            continue
+
+    if len(serials) <= max_num_checkpoints:
+        return
+
+    serials.sort(reverse=True)
+    serials = serials[max_num_checkpoints:]
+    for serial in serials:
+        cur_dir = os.path.join(dirname, str(serial))
+        shutil.rmtree(cur_dir)
+
+
+def _write_success(dirname):
+    """
+    write an empty _SUCCESS file to checkpoint dir, indicate this checkpoint is correct.
+
+    :param dirname
+    """
+    success_file = os.path.join(dirname, SUCCESS_MARK_FILENAME)
+    with open(success_file, 'a'):
+        pass
+
+
+def _get_lastest_checkpoint_dir(checkpoint_dir):
+    """
+    get the latest file in checkpoint directory, the _SUCCESS file must exist in the directory
+
+    :param checkpoint_dir
+    """
+    if not checkpoint_dir.strip():
+        return -1
+
+    def has_success(checkpoint_dir, cur_dir):
+        """
+        is _SUCCESS in this dir
+        """
+        if not os.path.isdir(os.path.join(checkpoint_dir, cur_dir)):
+            return -1
+
+        try:
+            int(cur_dir)
+        except ValueError:
+            return -1
+
+        success_path = os.path.join(checkpoint_dir, cur_dir,
+                                    SUCCESS_MARK_FILENAME)
+        if os.path.isfile(success_path):
+            return int(cur_dir)
+
+    if not os.path.isdir(checkpoint_dir):
+        return -1
+
+    current_dir = -1
+    dirs = os.listdir(checkpoint_dir)
+    for cur_dir in dirs:
+        success_num = has_success(checkpoint_dir, cur_dir)
+        if success_num > current_dir:
+            current_dir = success_num
+    return current_dir