fleet_util.py

#   Copyright (c) 2019 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.
"""Fleet Utils."""

import collections
import copy
import json
import logging
import math
import numpy as np
import os
import sys
import time
import paddle.fluid as fluid
from paddle.fluid import core
from paddle.fluid.log_helper import get_logger
from paddle.distributed.fleet.utils.fs import LocalFS, HDFSClient, AFSClient
from . import utils

OpRole = core.op_proto_and_checker_maker.OpRole

__all__ = ["FleetUtil", "GPUPSUtil"]

_logger = get_logger(
    __name__, logging.INFO, fmt='%(asctime)s %(levelname)s: %(message)s'
)

fleet = None


class FleetUtil(object):
    """
    FleetUtil provides some common functions for users' convenience.

    Examples:
        .. code-block:: python

          from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
          fleet_util = FleetUtil()
          fleet_util.rank0_print("my log")

    """

    def __init__(self, mode="pslib"):
        global fleet
        op_maker = core.op_proto_and_checker_maker
        self.op_role_key = op_maker.kOpRoleAttrName()
        if mode == "pslib":
            from paddle.fluid.incubate.fleet.parameter_server.pslib import (
                fleet as fleet_pslib,
            )

            fleet = fleet_pslib
        elif mode == "transpiler":
            from paddle.fluid.incubate.fleet.parameter_server.distribute_transpiler import (
                fleet as fleet_transpiler,
            )

            fleet = fleet_transpiler
        else:
            raise ValueError(
                "Please choose one mode from [\"pslib\", \"transpiler\"]"
            )

    def rank0_print(self, s):
        """
        Worker of rank 0 print some log.

        Args:
            s(str): string to print

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()
              fleet_util.rank0_print("my log")

        """
        if fleet.worker_index() != 0:
            return
        print(s)
        sys.stdout.flush()

    def rank0_info(self, s):
        """
        Worker of rank 0 print some log info.

        Args:
            s(str): string to log

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()
              fleet_util.rank0_info("my log info")

        """
        if fleet.worker_index() != 0:
            return
        _logger.info(s)

    def rank0_error(self, s):
        """
        Worker of rank 0 print some log error.

        Args:
            s(str): string to log

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()
              fleet_util.rank0_error("my log error")

        """
        if fleet.worker_index() != 0:
            return
        _logger.error(s)

    def set_zero(
        self,
        var_name,
        scope=fluid.global_scope(),
        place=fluid.CPUPlace(),
        param_type="int64",
    ):
        """
        Set tensor of a Variable to zero.

        Args:
            var_name(str): name of Variable
            scope(Scope): Scope object, default is fluid.global_scope()
            place(Place): Place object, default is fluid.CPUPlace()
            param_type(str): param data type, default is int64

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()
              fleet_util.set_zero(myvar.name, myscope)

        """
        param = scope.var(var_name).get_tensor()
        param_array = np.zeros(param._get_dims()).astype(param_type)
        param.set(param_array, place)

    def print_global_auc(
        self,
        scope=fluid.global_scope(),
        stat_pos="_generated_var_2",
        stat_neg="_generated_var_3",
        print_prefix="",
    ):
        r"""
        Print global auc of all distributed workers.

        Args:
            scope(Scope): Scope object, default is fluid.global_scope()
            stat_pos(str): name of auc pos bucket Variable
            stat_neg(str): name of auc neg bucket Variable
            print_prefix(str): prefix of print auc

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()
              fleet_util.print_global_auc(myscope, stat_pos=stat_pos.name,
                                          stat_neg=stat_neg.name)

              # below is part of model
              emb = my_slot_net(slots, label) # emb can be fc layer of size 1
              similarity_norm = fluid.layers.sigmoid(fluid.layers.clip(\
                  emb, min=-15.0, max=15.0), name="similarity_norm")\
              binary_predict = fluid.layers.concat(input=[\
                  fluid.layers.elementwise_sub(\
                      fluid.layers.ceil(similarity_norm), similarity_norm),\
                  similarity_norm], axis=1)
              auc, batch_auc, [batch_stat_pos, batch_stat_neg, stat_pos, \
                  stat_neg] = fluid.layers.auc(input=binary_predict,\
                                               label=label, curve='ROC',\
                                               num_thresholds=4096)

        """
        auc_value = self.get_global_auc(scope, stat_pos, stat_neg)
        self.rank0_print(print_prefix + " global auc = %s" % auc_value)

    def get_global_auc(
        self,
        scope=fluid.global_scope(),
        stat_pos="_generated_var_2",
        stat_neg="_generated_var_3",
    ):
        """
        Get global auc of all distributed workers.

        Args:
            scope(Scope): Scope object, default is fluid.global_scope()
            stat_pos(str): name of auc pos bucket Variable
            stat_neg(str): name of auc neg bucket Variable

        Returns:
            auc_value(float), total_ins_num(int)

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()
              auc_value, _ = fleet_util.get_global_auc(myscope,
                                                       stat_pos=stat_pos,
                                                       stat_neg=stat_neg)

        """
        if scope.find_var(stat_pos) is None or scope.find_var(stat_neg) is None:
            self.rank0_print("not found auc bucket")
            return None
        fleet._role_maker._barrier_worker()
        # auc pos bucket
        pos = np.array(scope.find_var(stat_pos).get_tensor())
        # auc pos bucket shape
        old_pos_shape = np.array(pos.shape)
        # reshape to one dim
        pos = pos.reshape(-1)
        global_pos = np.copy(pos) * 0
        # mpi allreduce
        fleet._role_maker._all_reduce(pos, global_pos)
        # reshape to its original shape
        global_pos = global_pos.reshape(old_pos_shape)

        # auc neg bucket
        neg = np.array(scope.find_var(stat_neg).get_tensor())
        old_neg_shape = np.array(neg.shape)
        neg = neg.reshape(-1)
        global_neg = np.copy(neg) * 0
        fleet._role_maker._all_reduce(neg, global_neg)
        global_neg = global_neg.reshape(old_neg_shape)

        # calculate auc
        num_bucket = len(global_pos[0])
        area = 0.0
        pos = 0.0
        neg = 0.0
        new_pos = 0.0
        new_neg = 0.0
        total_ins_num = 0
        for i in range(num_bucket):
            index = num_bucket - 1 - i
            new_pos = pos + global_pos[0][index]
            total_ins_num += global_pos[0][index]
            new_neg = neg + global_neg[0][index]
            total_ins_num += global_neg[0][index]
            area += (new_neg - neg) * (pos + new_pos) / 2
            pos = new_pos
            neg = new_neg

        auc_value = None
        if pos * neg == 0 or total_ins_num == 0:
            auc_value = 0.5
        else:
            auc_value = area / (pos * neg)

        fleet._role_maker._barrier_worker()
        return auc_value

    def load_fleet_model_one_table(self, table_id, path):
        """
        load pslib model to one table

        Args:
            table_id(int): load model to one table, default is None, which mean
                           load all table.
            path(str): model path

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()
              fleet_util.load_fleet_model("hdfs:/my/model/path", table_id=1)
        """
        fleet.load_one_table(table_id, path)

    def load_fleet_model(self, path, mode=0):
        """
        load pslib model

        Args:
            path(str): model path
            mode(str): 0 or 1, which means load checkpoint or delta model,
                       default is 0

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()

              fleet_util.load_fleet_model("hdfs:/my/model/path")

              fleet_util.load_fleet_model("hdfs:/my/model/path", mode=0)

        """
        fleet.init_server(path, mode=mode)

    def save_fleet_model(self, path, mode=0):
        """
        save pslib model

        Args:
            path(str): model path
            mode(str): 0 or 1, which means save checkpoint or delta model,
                       default is 0

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()
              fleet_util.save_fleet_model("hdfs:/my/model/path")

        """
        fleet.save_persistables(None, path, mode=mode)

    def _get_xbox_str(
        self,
        output_path,
        day,
        model_path,
        xbox_base_key,
        data_path,
        hadoop_fs_name,
        monitor_data={},
        mode="patch",
    ):
        xbox_dict = collections.OrderedDict()
        if mode == "base":
            xbox_dict["id"] = str(xbox_base_key)
        elif mode == "patch":
            xbox_dict["id"] = str(int(time.time()))
        else:
            print("warning: unknown mode %s, set it to patch" % mode)
            mode = "patch"
            xbox_dict["id"] = str(int(time.time()))
        xbox_dict["key"] = str(xbox_base_key)
        if model_path.startswith("hdfs:") or model_path.startswith("afs:"):
            model_path = model_path[model_path.find(":") + 1 :]
        xbox_dict["input"] = hadoop_fs_name + model_path.rstrip("/") + "/000"
        xbox_dict["record_count"] = "111111"
        xbox_dict["partition_type"] = "2"
        xbox_dict["job_name"] = "default_job_name"
        xbox_dict["ins_tag"] = "feasign"
        xbox_dict["ins_path"] = data_path
        job_id_with_host = os.popen("echo -n ${JOB_ID}").read().strip()
        instance_id = os.popen("echo -n ${INSTANCE_ID}").read().strip()
        start_pos = instance_id.find(job_id_with_host)
        end_pos = instance_id.find("--")
        if start_pos != -1 and end_pos != -1:
            job_id_with_host = instance_id[start_pos:end_pos]
        xbox_dict["job_id"] = job_id_with_host
        # currently hard code here, set monitor_data empty string
        xbox_dict["monitor_data"] = ""
        xbox_dict["monitor_path"] = (
            output_path.rstrip("/") + "/monitor/" + day + ".txt"
        )
        xbox_dict["mpi_size"] = str(fleet.worker_num())
        return json.dumps(xbox_dict)

    def write_model_donefile(
        self,
        output_path,
        day,
        pass_id,
        xbox_base_key,
        hadoop_fs_name,
        hadoop_fs_ugi,
        hadoop_home="$HADOOP_HOME",
        donefile_name="donefile.txt",
    ):
        """
        write donefile when save model

        Args:
            output_path(str): output path
            day(str|int): training day
            pass_id(str|int): training pass id
            xbox_base_key(str|int): xbox base key
            hadoop_fs_name(str): hdfs/afs fs name
            hadoop_fs_ugi(str): hdfs/afs fs ugi
            hadoop_home(str): hadoop home, default is "$HADOOP_HOME"
            donefile_name(str): donefile name, default is "donefile.txt"

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()
              fleet_util.write_model_donefile(output_path="hdfs:/my/output",
                                              model_path="hdfs:/my/model",
                                              day=20190723,
                                              pass_id=66,
                                              xbox_base_key=int(time.time()),
                                              hadoop_fs_name="hdfs://xxx",
                                              hadoop_fs_ugi="user,passwd")

        """
        day = str(day)
        pass_id = str(pass_id)
        xbox_base_key = int(xbox_base_key)

        if pass_id != "-1":
            suffix_name = "/%s/%s/" % (day, pass_id)
            model_path = output_path.rstrip("/") + suffix_name
        else:
            suffix_name = "/%s/0/" % day
            model_path = output_path.rstrip("/") + suffix_name

        if fleet.worker_index() == 0:
            donefile_path = output_path + "/" + donefile_name
            content = "%s\t%lu\t%s\t%s\t%d" % (
                day,
                xbox_base_key,
                model_path,
                pass_id,
                0,
            )
            configs = {
                "fs.default.name": hadoop_fs_name,
                "hadoop.job.ugi": hadoop_fs_ugi,
            }
            client = HDFSClient(hadoop_home, configs)
            if client.is_file(donefile_path):
                pre_content = client.cat(donefile_path)
                pre_content_list = pre_content.split("\n")
                day_list = [i.split("\t")[0] for i in pre_content_list]
                pass_list = [i.split("\t")[3] for i in pre_content_list]
                exist = False
                for i in range(len(day_list)):
                    if int(day) == int(day_list[i]) and int(pass_id) == int(
                        pass_list[i]
                    ):
                        exist = True
                        break
                if not exist:
                    with open(donefile_name, "w") as f:
                        f.write(pre_content + "\n")
                        f.write(content + "\n")
                    client.delete(donefile_path)
                    client.upload(donefile_name, output_path)
                    self.rank0_error(
                        "write %s/%s %s succeed" % (day, pass_id, donefile_name)
                    )
                else:
                    self.rank0_error(
                        "not write %s because %s/%s already "
                        "exists" % (donefile_name, day, pass_id)
                    )
            else:
                with open(donefile_name, "w") as f:
                    f.write(content + "\n")
                client.upload(donefile_name, output_path)
                self.rank0_error(
                    "write %s/%s %s succeed" % (day, pass_id, donefile_name)
                )
        fleet._role_maker._barrier_worker()

    def write_xbox_donefile(
        self,
        output_path,
        day,
        pass_id,
        xbox_base_key,
        data_path,
        hadoop_fs_name,
        hadoop_fs_ugi,
        monitor_data={},
        hadoop_home="$HADOOP_HOME",
        donefile_name=None,
    ):
        """
        write delta donefile or xbox base donefile

        Args:
            output_path(str): output path
            day(str|int): training day of model
            pass_id(str|int): training pass id of model
            xbox_base_key(str|int): xbox base key
            data_path(str|list): training data path
            hadoop_fs_name(str): hdfs/afs fs name
            hadoop_fs_ugi(str): hdfs/afs fs ugi
            monitor_data(dict): metrics
            hadoop_home(str): hadoop home, default is "$HADOOP_HOME"
            donefile_name(str): donefile name, default is None"

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()
              fleet_util.write_xbox_donefile(
                  output_path="hdfs:/my/output/",
                  model_path="hdfs:/my/output/20190722/01",
                  day=20190722,
                  pass_id=1,
                  xbox_base_key=int(time.time()),
                  data_path="hdfs:/my/data/",
                  hadoop_fs_name="hdfs://xxx",
                  hadoop_fs_ugi="user,passwd",
                  monitor_data={}
                  )

        """
        day = str(day)
        pass_id = str(pass_id)
        xbox_base_key = int(xbox_base_key)
        mode = None

        if pass_id != "-1":
            mode = "patch"
            suffix_name = "/%s/delta-%s/" % (day, pass_id)
            model_path = output_path.rstrip("/") + suffix_name
            if donefile_name is None:
                donefile_name = "xbox_patch_done.txt"
        else:
            mode = "base"
            suffix_name = "/%s/base/" % day
            model_path = output_path.rstrip("/") + suffix_name
            if donefile_name is None:
                donefile_name = "xbox_base_done.txt"

        if isinstance(data_path, list):
            data_path = ",".join(data_path)

        if fleet.worker_index() == 0:
            donefile_path = output_path + "/" + donefile_name
            xbox_str = self._get_xbox_str(
                output_path,
                day,
                model_path,
                xbox_base_key,
                data_path,
                hadoop_fs_name,
                monitor_data={},
                mode=mode,
            )
            configs = {
                "fs.default.name": hadoop_fs_name,
                "hadoop.job.ugi": hadoop_fs_ugi,
            }
            client = HDFSClient(hadoop_home, configs)
            if client.is_file(donefile_path):
                pre_content = client.cat(donefile_path)
                last_dict = json.loads(pre_content.split("\n")[-1])
                last_day = last_dict["input"].split("/")[-3]
                last_pass = last_dict["input"].split("/")[-2].split("-")[-1]
                exist = False
                if (
                    int(day) < int(last_day)
                    or int(day) == int(last_day)
                    and int(pass_id) <= int(last_pass)
                ):
                    exist = True
                if not exist:
                    with open(donefile_name, "w") as f:
                        f.write(pre_content + "\n")
                        f.write(xbox_str + "\n")
                    client.delete(donefile_path)
                    client.upload(donefile_name, output_path)
                    self.rank0_error(
                        "write %s/%s %s succeed" % (day, pass_id, donefile_name)
                    )
                else:
                    self.rank0_error(
                        "not write %s because %s/%s already "
                        "exists" % (donefile_name, day, pass_id)
                    )
            else:
                with open(donefile_name, "w") as f:
                    f.write(xbox_str + "\n")
                client.upload(donefile_name, output_path)
                self.rank0_error(
                    "write %s/%s %s succeed" % (day, pass_id, donefile_name)
                )
        fleet._role_maker._barrier_worker()

    def write_cache_donefile(
        self,
        output_path,
        day,
        pass_id,
        key_num,
        hadoop_fs_name,
        hadoop_fs_ugi,
        hadoop_home="$HADOOP_HOME",
        donefile_name="sparse_cache.meta",
        **kwargs
    ):
        """
        write cache donefile

        Args:
            output_path(str): output path
            day(str|int): training day of model
            pass_id(str|int): training pass id of model
            key_num(str|int): save cache return value
            hadoop_fs_name(str): hdfs/afs fs name
            hadoop_fs_ugi(str): hdfs/afs fs ugi
            hadoop_home(str): hadoop home, default is "$HADOOP_HOME"
            donefile_name(str): donefile name, default is "sparse_cache.meta"
            kwargs(dict): user defined properties
                          file_num(int): cache file num
                          table_id(int): cache table id

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()
              fleet_util.write_cache_donefile(
                  output_path="hdfs:/my/output/",
                  day=20190722,
                  pass_id=1,
                  key_num=123456,
                  hadoop_fs_name="hdfs://xxx",
                  hadoop_fs_ugi="user,passwd",
                  )

        """
        day = str(day)
        pass_id = str(pass_id)
        key_num = int(key_num)
        file_num = kwargs.get("file_num", 16)
        table_id = kwargs.get("table_id", 0)

        if pass_id != "-1":
            suffix_name = "/%s/delta-%s/%03d_cache" % (day, pass_id, table_id)
            model_path = output_path.rstrip("/") + suffix_name
        else:
            suffix_name = "/%s/base/%03d_cache" % (day, table_id)
            model_path = output_path.rstrip("/") + suffix_name

        if fleet.worker_index() == 0:
            donefile_path = model_path + "/" + donefile_name
            configs = {
                "fs.default.name": hadoop_fs_name,
                "hadoop.job.ugi": hadoop_fs_ugi,
            }
            client = HDFSClient(hadoop_home, configs)
            if client.is_file(donefile_path):
                self.rank0_error(
                    "not write because %s already exists" % donefile_path
                )
            else:
                meta_str = "file_prefix:part\npart_num:%s\nkey_num:%d\n" % (
                    file_num,
                    key_num,
                )
                with open(donefile_name, "w") as f:
                    f.write(meta_str)
                client.upload(donefile_name, model_path)
                self.rank0_error("write %s succeed" % donefile_path)
        fleet._role_maker._barrier_worker()

    def load_model(self, output_path, day, pass_id):
        """
        load pslib model

        Args:
            output_path(str): output path
            day(str|int): training day
            pass_id(str|int): training pass id

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()
              fleet_util.load_model("hdfs:/my/path", 20190722, 88)

        """
        day = str(day)
        pass_id = str(pass_id)
        suffix_name = "/%s/%s/" % (day, pass_id)
        load_path = output_path + suffix_name
        self.rank0_error("going to load_model %s" % load_path)
        self.load_fleet_model(load_path)
        self.rank0_error("load_model done")

    def save_model(self, output_path, day, pass_id):
        """
        save pslib model

        Args:
            output_path(str): output path
            day(str|int): training day
            pass_id(str|int): training pass id

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()
              fleet_util.save_model("hdfs:/my/path", 20190722, 88)

        """
        day = str(day)
        pass_id = str(pass_id)
        suffix_name = "/%s/%s/" % (day, pass_id)
        model_path = output_path + suffix_name
        self.rank0_print("going to save_model %s" % model_path)
        self.save_fleet_model(model_path)
        self.rank0_print("save_model done")

    def save_batch_model(self, output_path, day):
        """
        save batch model

        Args:
            output_path(str): output path
            day(str|int): training day

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()
              fleet_util.save_batch_model("hdfs:/my/path", 20190722)

        """
        day = str(day)
        suffix_name = "/%s/0/" % day
        model_path = output_path + suffix_name
        self.rank0_print("going to save_model %s" % model_path)
        fleet.save_persistables(None, model_path, mode=3)
        self.rank0_print("save_batch_model done")

    def save_delta_model(self, output_path, day, pass_id):
        """
        save delta model

        Args:
            output_path(str): output path
            day(str|int): training day
            pass_id(str|int): training pass id

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()
              fleet_util.save_batch_model("hdfs:/my/path", 20190722, 88)

        """
        day = str(day)
        pass_id = str(pass_id)
        suffix_name = "/%s/delta-%s/" % (day, pass_id)
        model_path = output_path + suffix_name
        self.rank0_print("going to save_delta_model %s" % model_path)
        fleet.save_persistables(None, model_path, mode=1)
        self.rank0_print("save_delta_model done")

    def save_xbox_base_model(self, output_path, day):
        """
        save xbox base model

        Args:
            output_path(str): output path
            day(str|int): training day

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()
              fleet_util.save_xbox_base_model("hdfs:/my/path", 20190722, 88)

        """
        day = str(day)
        suffix_name = "/%s/base/" % day
        model_path = output_path + suffix_name
        self.rank0_print("going to save_xbox_base_model " + model_path)
        fleet.save_persistables(None, model_path, mode=2)
        self.rank0_print("save_xbox_base_model done")

    def save_cache_model(self, output_path, day, pass_id, mode=1, **kwargs):
        """
        save cache model

        Args:
            output_path(str): output path
            day(str|int): training day
            pass_id(str|int): training pass id
            mode(str|int): save mode
            kwargs(dict): user defined properties
                          table_id(int): table id to save cache

        Returns:
            key_num(int): cache key num

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()
              fleet_util.save_cache_model("hdfs:/my/path", 20190722, 88)

        """
        day = str(day)
        pass_id = str(pass_id)
        mode = int(mode)
        table_id = kwargs.get("table_id", 0)
        suffix_name = "/%s/delta-%s" % (day, pass_id)
        model_path = output_path.rstrip("/") + suffix_name
        self.rank0_print("going to save_cache_model %s" % model_path)
        key_num = fleet.save_cache_model(
            None, model_path, mode=mode, table_id=table_id
        )
        self.rank0_print("save_cache_model done")
        return key_num

    def save_cache_base_model(self, output_path, day, **kwargs):
        """
        save cache model

        Args:
            output_path(str): output path
            day(str|int): training day
            pass_id(str|int): training pass id
            kwargs(dict): user defined properties
                          table_id(int): table id to save cache

        Returns:
            key_num(int): cache key num

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()
              fleet_util.save_cache_base_model("hdfs:/my/path", 20190722)

        """
        day = str(day)
        table_id = kwargs.get("table_id", 0)
        suffix_name = "/%s/base" % day
        model_path = output_path.rstrip("/") + suffix_name
        self.rank0_print("going to save_cache_base_model %s" % model_path)
        key_num = fleet.save_cache_model(
            None, model_path, mode=2, table_id=table_id
        )
        self.rank0_print("save_cache_base_model done")
        return key_num

    def pull_all_dense_params(self, scope, program):
        """
        pull all dense params in trainer of rank 0

        Args:
            scope(Scope): fluid Scope
            program(Program): fluid Program

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()
              fleet_util.pull_all_dense_params(my_scope, my_program)

        """
        fleet._role_maker._barrier_worker()
        if fleet._role_maker.is_first_worker():
            prog_id = str(id(program))
            tables = (
                fleet._opt_info["program_id_to_worker"][prog_id]
                .get_desc()
                .dense_table
            )
            prog_conf = fleet._opt_info['program_configs'][prog_id]
            prog_tables = {}
            for key in prog_conf:
                if "dense" not in key:
                    continue
                for table_id in prog_conf[key]:
                    prog_tables[int(table_id)] = 0
            for table in tables:
                if int(table.table_id) not in prog_tables:
                    continue
                var_name_list = []
                for i in range(0, len(table.dense_variable_name)):
                    var_name = table.dense_variable_name[i]
                    if scope.find_var(var_name) is None:
                        raise ValueError(
                            "var "
                            + var_name
                            + " not found in scope "
                            + "when pull dense"
                        )
                    var_name_list.append(var_name)
                fleet._fleet_ptr.pull_dense(
                    scope, int(table.table_id), var_name_list
                )
        fleet._role_maker._barrier_worker()

    def save_paddle_inference_model(
        self,
        executor,
        scope,
        program,
        feeded_vars,
        target_vars,
        output_path,
        day,
        pass_id,
        hadoop_fs_name,
        hadoop_fs_ugi,
        hadoop_home="$HADOOP_HOME",
        save_combine=True,
    ):
        """
        save paddle inference model, and upload to hdfs dnn_plugin path

        Args:
            executor(Executor): fluid Executor
            scope(Scope): fluid Scope
            program(Program): fluid Program
            feeded_vars(list[Variable]): feed vars
            target_vars(list[variable]): fetch vars
            output_path(str): hdfs/afs output path
            day(str|int): training day
            pass_id(str|int): training pass
            hadoop_fs_name(str): hadoop fs name
            hadoop_fs_ugi(str): hadoop fs ugi
            hadoop_home(str): hadoop home, default is "$HADOOP_HOME"
            save_combine(bool): whether to save in a file or separate files,
                                default is True

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()
              fleet_util.save_paddle_inference_model(exe,
                                                     join_scope,
                                                     join_program,
                                                     feeded_vars,
                                                     target_vars,
                                                     "hdfs:/my/output/path/",
                                                     day=20190727,
                                                     pass_id=6,
                                                     hadoop_fs_name="xxx",
                                                     hadoop_fs_ugi="xxx,xxx")
        """
        day = str(day)
        pass_id = str(pass_id)
        feeded_var_names = [i.name for i in feeded_vars]
        model_name = "inference_model"
        # pull dense before save
        self.pull_all_dense_params(scope, program)
        if fleet.worker_index() == 0:
            with fluid.scope_guard(scope):
                if save_combine:
                    fluid.io.save_inference_model(
                        dirname=model_name,
                        feeded_var_names=feeded_var_names,
                        target_vars=target_vars,
                        executor=executor,
                        main_program=program.clone(),
                        params_filename="params",
                    )
                else:
                    fluid.io.save_inference_model(
                        dirname=model_name,
                        feeded_var_names=feeded_var_names,
                        target_vars=target_vars,
                        executor=executor,
                        main_program=program.clone(),
                    )

            configs = {
                "fs.default.name": hadoop_fs_name,
                "hadoop.job.ugi": hadoop_fs_ugi,
            }
            client = HDFSClient(hadoop_home, configs)

            if pass_id == "-1":
                dest = "%s/%s/base/dnn_plugin/" % (output_path, day)
            else:
                dest = "%s/%s/delta-%s/dnn_plugin/" % (
                    output_path,
                    day,
                    pass_id,
                )
            if not client.is_exist(dest):
                client.makedirs(dest)

            client.upload(model_name, dest, multi_processes=5, overwrite=True)

        fleet._role_maker._barrier_worker()

    def save_paddle_params(
        self,
        executor,
        scope,
        program,
        model_name,
        output_path,
        day,
        pass_id,
        hadoop_fs_name,
        hadoop_fs_ugi,
        hadoop_home="$HADOOP_HOME",
        var_names=None,
        save_combine=True,
    ):
        """
        save paddle model, and upload to hdfs dnn_plugin path

        Args:
            executor(Executor): fluid Executor
            scope(Scope): fluid Scope
            program(Program): fluid Program
            model_name(str): save model local dir or filename
            output_path(str): hdfs/afs output path
            day(str|int): training day
            pass_id(str|int): training pass
            hadoop_fs_name(str): hadoop fs name
            hadoop_fs_ugi(str): hadoop fs ugi
            hadoop_home(str): hadoop home, default is "$HADOOP_HOME"
            var_names(list): save persistable var names, default is None
            save_combine(bool): whether to save in a file or separate files,
                                default is True

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()
              fleet_util.save_paddle_params(exe,
                                            join_scope,
                                            join_program,
                                            "paddle_dense.model.0",
                                            "hdfs:/my/output/path/",
                                            day=20190727,
                                            pass_id=6,
                                            hadoop_fs_name="xxx",
                                            hadoop_fs_ugi="xxx,xxx",
                                            var_names=join_all_var_names)
              fleet_util.save_paddle_params(exe,
                                            join_scope,
                                            join_program,
                                            "paddle_dense.model.usr.0",
                                            "hdfs:/my/output/path/",
                                            day=20190727,
                                            pass_id=6,
                                            hadoop_fs_name="xxx",
                                            hadoop_fs_ugi="xxx,xxx",
                                            var_names=join_user_var_names)
              fleet_util.save_paddle_params(exe,
                                            join_scope,
                                            join_program,
                                            "paddle_dense.model.item.0",
                                            "hdfs:/my/output/path/",
                                            day=20190727,
                                            pass_id=6,
                                            hadoop_fs_name="xxx",
                                            hadoop_fs_ugi="xxx,xxx",
                                            var_names=join_user_item_names)

        """
        day = str(day)
        pass_id = str(pass_id)
        # pull dense before save
        self.pull_all_dense_params(scope, program)
        if fleet.worker_index() == 0:
            vars = [program.global_block().var(i) for i in var_names]
            with fluid.scope_guard(scope):
                if save_combine:
                    fluid.io.save_vars(
                        executor, "./", program, vars=vars, filename=model_name
                    )
                else:
                    fluid.io.save_vars(executor, model_name, program, vars=vars)

            configs = {
                "fs.default.name": hadoop_fs_name,
                "hadoop.job.ugi": hadoop_fs_ugi,
            }
            client = HDFSClient(hadoop_home, configs)

            if pass_id == "-1":
                dest = "%s/%s/base/dnn_plugin/" % (output_path, day)
            else:
                dest = "%s/%s/delta-%s/dnn_plugin/" % (
                    output_path,
                    day,
                    pass_id,
                )
            if not client.is_exist(dest):
                client.mkdirs(dest)
            client.upload(model_name, dest, multi_processes=5, overwrite=True)

        fleet._role_maker._barrier_worker()

    def get_last_save_xbox_base(
        self,
        output_path,
        hadoop_fs_name,
        hadoop_fs_ugi,
        hadoop_home="$HADOOP_HOME",
    ):
        r"""
        get last saved base xbox info from xbox_base_done.txt

        Args:
            output_path(str): output path
            hadoop_fs_name(str): hdfs/afs fs_name
            hadoop_fs_ugi(str): hdfs/afs fs_ugi
            hadoop_home(str): hadoop home, default is "$HADOOP_HOME"

        Returns:
            [last_save_day, last_path, xbox_base_key]
            last_save_day(int): day of saved model
            last_path(str): model path
            xbox_base_key(int): xbox key

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()
              last_save_day, last_path, xbox_base_key = \
                  fleet_util.get_last_save_xbox_base("hdfs:/my/path", 20190722,
                                                     88)

        """
        donefile_path = output_path + "/xbox_base_done.txt"
        configs = {
            "fs.default.name": hadoop_fs_name,
            "hadoop.job.ugi": hadoop_fs_ugi,
        }
        client = HDFSClient(hadoop_home, configs)
        if not client.is_file(donefile_path):
            return [-1, -1, int(time.time())]
        pre_content = client.cat(donefile_path)
        last_dict = json.loads(pre_content.split("\n")[-1])
        last_day = int(last_dict["input"].split("/")[-3])
        last_path = "/".join(last_dict["input"].split("/")[:-1])
        xbox_base_key = int(last_dict["key"])
        return [last_day, last_path, xbox_base_key]

    def get_last_save_xbox(
        self,
        output_path,
        hadoop_fs_name,
        hadoop_fs_ugi,
        hadoop_home="$HADOOP_HOME",
    ):
        r"""
        get last saved xbox info from xbox_patch_done.txt

        Args:
            output_path(str): output path
            hadoop_fs_name(str): hdfs/afs fs_name
            hadoop_fs_ugi(str): hdfs/afs fs_ugi
            hadoop_home(str): hadoop home, default is "$HADOOP_HOME"

        Returns:
            [last_save_day, last_save_pass, last_path, xbox_base_key]
            last_save_day(int): day of saved model
            last_save_pass(int): pass id of saved
            last_path(str): model path
            xbox_base_key(int): xbox key

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()
              last_save_day, last_save_pass, last_path, xbox_base_key = \
                  fleet_util.get_last_save_xbox("hdfs:/my/path", 20190722, 88)

        """
        donefile_path = output_path + "/xbox_patch_done.txt"
        configs = {
            "fs.default.name": hadoop_fs_name,
            "hadoop.job.ugi": hadoop_fs_ugi,
        }
        client = HDFSClient(hadoop_home, configs)
        if not client.is_file(donefile_path):
            return [-1, -1, "", int(time.time())]
        pre_content = client.cat(donefile_path)
        last_dict = json.loads(pre_content.split("\n")[-1])
        last_day = int(last_dict["input"].split("/")[-3])
        last_pass = int(last_dict["input"].split("/")[-2].split("-")[-1])
        last_path = "/".join(last_dict["input"].split("/")[:-1])
        xbox_base_key = int(last_dict["key"])
        return [last_day, last_pass, last_path, xbox_base_key]

    def get_last_save_model(
        self,
        output_path,
        hadoop_fs_name,
        hadoop_fs_ugi,
        hadoop_home="$HADOOP_HOME",
    ):
        r"""
        get last saved model info from donefile.txt

        Args:
            output_path(str): output path
            hadoop_fs_name(str): hdfs/afs fs_name
            hadoop_fs_ugi(str): hdfs/afs fs_ugi
            hadoop_home(str): hadoop home, default is "$HADOOP_HOME"

        Returns:
            [last_save_day, last_save_pass, last_path, xbox_base_key]
            last_save_day(int): day of saved model
            last_save_pass(int): pass id of saved
            last_path(str): model path
            xbox_base_key(int): xbox key

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()
              last_save_day, last_save_pass, last_path, xbox_base_key = \
                  fleet_util.get_last_save_model("hdfs:/my/path", 20190722, 88)

        """
        last_save_day = -1
        last_save_pass = -1
        last_path = ""
        donefile_path = output_path + "/donefile.txt"
        configs = {
            "fs.default.name": hadoop_fs_name,
            "hadoop.job.ugi": hadoop_fs_ugi,
        }
        client = HDFSClient(hadoop_home, configs)
        if not client.is_file(donefile_path):
            return [-1, -1, "", int(time.time())]
        content = client.cat(donefile_path)
        content = content.split("\n")[-1].split("\t")
        last_save_day = int(content[0])
        last_save_pass = int(content[3])
        last_path = content[2]
        xbox_base_key = int(content[1])
        return [last_save_day, last_save_pass, last_path, xbox_base_key]

    def get_online_pass_interval(
        self, days, hours, split_interval, split_per_pass, is_data_hourly_placed
    ):
        """
        get online pass interval

        Args:
            days(str): days to train
            hours(str): hours to train
            split_interval(int|str): split interval
            split_per_pass(int}str): split per pass
            is_data_hourly_placed(bool): is data hourly placed

        Returns:
            online_pass_interval(list)

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()
              online_pass_interval = fleet_util.get_online_pass_interval(
                  days="{20190720..20190729}",
                  hours="{0..23}",
                  split_interval=5,
                  split_per_pass=2,
                  is_data_hourly_placed=False)

        """
        days = os.popen("echo -n " + days).read().split(" ")
        hours = os.popen("echo -n " + hours).read().split(" ")
        split_interval = int(split_interval)
        split_per_pass = int(split_per_pass)
        splits_per_day = 24 * 60 // split_interval
        pass_per_day = splits_per_day // split_per_pass
        left_train_hour = int(hours[0])
        right_train_hour = int(hours[-1])

        start = 0
        split_path = []
        for i in range(splits_per_day):
            h = start // 60
            m = start % 60
            if h < left_train_hour or h > right_train_hour:
                start += split_interval
                continue
            if is_data_hourly_placed:
                split_path.append("%02d" % h)
            else:
                split_path.append("%02d%02d" % (h, m))
            start += split_interval

        start = 0
        online_pass_interval = []
        for i in range(pass_per_day):
            online_pass_interval.append([])
            for j in range(start, start + split_per_pass):
                online_pass_interval[i].append(split_path[j])
            start += split_per_pass

        return online_pass_interval

    def get_global_metrics(
        self,
        scope=fluid.global_scope(),
        stat_pos_name="_generated_var_2",
        stat_neg_name="_generated_var_3",
        sqrerr_name="sqrerr",
        abserr_name="abserr",
        prob_name="prob",
        q_name="q",
        pos_ins_num_name="pos",
        total_ins_num_name="total",
    ):
        r"""
        get global metrics, including auc, bucket_error, mae, rmse,
        actual_ctr, predicted_ctr, copc, mean_predict_qvalue, total_ins_num.

        Args:
            scope(Scope): Scope object, default is fluid.global_scope()
            stat_pos_name(str): name of auc pos bucket Variable
            stat_neg_name(str): name of auc neg bucket Variable
            sqrerr_name(str): name of sqrerr Variable
            abserr_name(str): name of abserr Variable
            prob_name(str): name of prob Variable
            q_name(str): name of q Variable
            pos_ins_num_name(str): name of pos ins num Variable
            total_ins_num_name(str): name of total ins num Variable

        Returns:
            [auc, bucket_error, mae, rmse, actual_ctr, predicted_ctr, copc,
             mean_predict_qvalue, total_ins_num]

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()
              metric_list = fleet_util.get_global_metrics(myscope,
                                                          stat_pos.name,
                                                          stat_neg.name,
                                                          local_sqrerr.name,
                                                          local_abserr.name,
                                                          local_prob.name,
                                                          local_q.name,
                                                          local_pos_ins.name,
                                                          local_total_ins.name)

              # below is part of example model
              label = fluid.layers.data(name="click", shape=[-1, 1],\
                  dtype="int64", lod_level=0, append_batch_size=False)
              emb = my_slot_net(slots, label) # emb can be fc layer of size 1
              similarity_norm = fluid.layers.sigmoid(fluid.layers.clip(\
                  emb, min=-15.0, max=15.0), name="similarity_norm")\
              binary_predict = fluid.layers.concat(input=[\
                  fluid.layers.elementwise_sub(\
                      fluid.layers.ceil(similarity_norm), similarity_norm),\
                  similarity_norm], axis=1)
              auc, batch_auc, [batch_stat_pos, batch_stat_neg, stat_pos, \
                  stat_neg] = fluid.layers.auc(input=binary_predict,\
                                               label=label, curve='ROC',\
                                               num_thresholds=4096)
              local_sqrerr, local_abserr, local_prob, local_q, local_pos_ins,\
                  local_total_ins = fluid.contrib.layers.ctr_metric_bundle(\
                      similarity_norm, label)

        """
        if (
            scope.find_var(stat_pos_name) is None
            or scope.find_var(stat_neg_name) is None
        ):
            self.rank0_print("not found auc bucket")
            return [None] * 9
        elif scope.find_var(sqrerr_name) is None:
            self.rank0_print("not found sqrerr_name=%s" % sqrerr_name)
            return [None] * 9
        elif scope.find_var(abserr_name) is None:
            self.rank0_print("not found abserr_name=%s" % abserr_name)
            return [None] * 9
        elif scope.find_var(prob_name) is None:
            self.rank0_print("not found prob_name=%s" % prob_name)
            return [None] * 9
        elif scope.find_var(q_name) is None:
            self.rank0_print("not found q_name=%s" % q_name)
            return [None] * 9
        elif scope.find_var(pos_ins_num_name) is None:
            self.rank0_print("not found pos_ins_num_name=%s" % pos_ins_num_name)
            return [None] * 9
        elif scope.find_var(total_ins_num_name) is None:
            self.rank0_print(
                "not found total_ins_num_name=%s" % total_ins_num_name
            )
            return [None] * 9

        # barrier worker to ensure all workers finished training
        fleet._role_maker._barrier_worker()

        # get auc
        auc = self.get_global_auc(scope, stat_pos_name, stat_neg_name)
        pos = np.array(scope.find_var(stat_pos_name).get_tensor())
        # auc pos bucket shape
        old_pos_shape = np.array(pos.shape)
        # reshape to one dim
        pos = pos.reshape(-1)
        global_pos = np.copy(pos) * 0
        # mpi allreduce
        fleet._role_maker._all_reduce(pos, global_pos)
        # reshape to its original shape
        global_pos = global_pos.reshape(old_pos_shape)
        # auc neg bucket
        neg = np.array(scope.find_var(stat_neg_name).get_tensor())
        old_neg_shape = np.array(neg.shape)
        neg = neg.reshape(-1)
        global_neg = np.copy(neg) * 0
        fleet._role_maker._all_reduce(neg, global_neg)
        global_neg = global_neg.reshape(old_neg_shape)

        num_bucket = len(global_pos[0])

        def get_metric(name):
            metric = np.array(scope.find_var(name).get_tensor())
            old_metric_shape = np.array(metric.shape)
            metric = metric.reshape(-1)
            global_metric = np.copy(metric) * 0
            fleet._role_maker._all_reduce(metric, global_metric)
            global_metric = global_metric.reshape(old_metric_shape)
            return global_metric[0]

        global_sqrerr = get_metric(sqrerr_name)
        global_abserr = get_metric(abserr_name)
        global_prob = get_metric(prob_name)
        global_q_value = get_metric(q_name)
        # note: get ins_num from auc bucket is not actual value,
        # so get it from metric op
        pos_ins_num = get_metric(pos_ins_num_name)
        total_ins_num = get_metric(total_ins_num_name)
        neg_ins_num = total_ins_num - pos_ins_num

        mae = global_abserr / total_ins_num
        rmse = math.sqrt(global_sqrerr / total_ins_num)
        return_actual_ctr = pos_ins_num / total_ins_num
        predicted_ctr = global_prob / total_ins_num
        mean_predict_qvalue = global_q_value / total_ins_num
        copc = 0.0
        if abs(predicted_ctr > 1e-6):
            copc = return_actual_ctr / predicted_ctr

        # calculate bucket error
        last_ctr = -1.0
        impression_sum = 0.0
        ctr_sum = 0.0
        click_sum = 0.0
        error_sum = 0.0
        error_count = 0.0
        click = 0.0
        show = 0.0
        ctr = 0.0
        adjust_ctr = 0.0
        relative_error = 0.0
        actual_ctr = 0.0
        relative_ctr_error = 0.0
        k_max_span = 0.01
        k_relative_error_bound = 0.05
        for i in range(num_bucket):
            click = global_pos[0][i]
            show = global_pos[0][i] + global_neg[0][i]
            ctr = float(i) / num_bucket
            if abs(ctr - last_ctr) > k_max_span:
                last_ctr = ctr
                impression_sum = 0.0
                ctr_sum = 0.0
                click_sum = 0.0
            impression_sum += show
            ctr_sum += ctr * show
            click_sum += click
            if impression_sum == 0:
                continue
            adjust_ctr = ctr_sum / impression_sum
            if adjust_ctr == 0:
                continue
            relative_error = math.sqrt(
                (1 - adjust_ctr) / (adjust_ctr * impression_sum)
            )
            if relative_error < k_relative_error_bound:
                actual_ctr = click_sum / impression_sum
                relative_ctr_error = abs(actual_ctr / adjust_ctr - 1)
                error_sum += relative_ctr_error * impression_sum
                error_count += impression_sum
                last_ctr = -1

        bucket_error = error_sum / error_count if error_count > 0 else 0.0

        return [
            auc,
            bucket_error,
            mae,
            rmse,
            return_actual_ctr,
            predicted_ctr,
            copc,
            mean_predict_qvalue,
            int(total_ins_num),
        ]

    def print_global_metrics(
        self,
        scope=fluid.global_scope(),
        stat_pos_name="_generated_var_2",
        stat_neg_name="_generated_var_3",
        sqrerr_name="sqrerr",
        abserr_name="abserr",
        prob_name="prob",
        q_name="q",
        pos_ins_num_name="pos",
        total_ins_num_name="total",
        print_prefix="",
    ):
        r"""
        print global metrics, including auc, bucket_error, mae, rmse,
        actual_ctr, predicted_ctr, copc, mean_predict_qvalue, total_ins_num.

        Args:
            scope(Scope): Scope object, default is fluid.global_scope()
            stat_pos_name(str): name of auc pos bucket Variable
            stat_neg_name(str): name of auc neg bucket Variable
            sqrerr_name(str): name of sqrerr Variable
            abserr_name(str): name of abserr Variable
            prob_name(str): name of prob Variable
            q_name(str): name of q Variable
            pos_ins_num_name(str): name of pos ins num Variable
            total_ins_num_name(str): name of total ins num Variable
            print_prefix(str): print prefix

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()
              fleet_util.print_global_metrics(myscope,
                                              stat_pos.name,
                                              stat_neg.name,
                                              local_sqrerr.name,
                                              local_abserr.name,
                                              local_prob.name,
                                              local_q.name,
                                              local_pos_ins.name,
                                              local_total_ins.name)

              # below is part of model
              label = fluid.layers.data(name="click", shape=[-1, 1],\
                  dtype="int64", lod_level=0, append_batch_size=False)
              emb = my_slot_net(slots, label) # emb can be fc layer of size 1
              similarity_norm = fluid.layers.sigmoid(fluid.layers.clip(\
                  emb, min=-15.0, max=15.0), name="similarity_norm")\
              binary_predict = fluid.layers.concat(input=[\
                  fluid.layers.elementwise_sub(\
                      fluid.layers.ceil(similarity_norm), similarity_norm),\
                  similarity_norm], axis=1)
              auc, batch_auc, [batch_stat_pos, batch_stat_neg, stat_pos, \
                  stat_neg] = fluid.layers.auc(input=binary_predict,\
                                               label=label, curve='ROC',\
                                               num_thresholds=4096)
              local_sqrerr, local_abserr, local_prob, local_q, local_pos_ins, \
                  local_total_ins = fluid.contrib.layers.ctr_metric_bundle(\
                      similarity_norm, label)

        """
        if (
            scope.find_var(stat_pos_name) is None
            or scope.find_var(stat_neg_name) is None
        ):
            self.rank0_print("not found auc bucket")
            return
        elif scope.find_var(sqrerr_name) is None:
            self.rank0_print("not found sqrerr_name=%s" % sqrerr_name)
            return
        elif scope.find_var(abserr_name) is None:
            self.rank0_print("not found abserr_name=%s" % abserr_name)
            return
        elif scope.find_var(prob_name) is None:
            self.rank0_print("not found prob_name=%s" % prob_name)
            return
        elif scope.find_var(q_name) is None:
            self.rank0_print("not found q_name=%s" % q_name)
            return
        elif scope.find_var(pos_ins_num_name) is None:
            self.rank0_print("not found pos_ins_num_name=%s" % pos_ins_num_name)
            return
        elif scope.find_var(total_ins_num_name) is None:
            self.rank0_print(
                "not found total_ins_num_name=%s" % total_ins_num_name
            )
            return

        (
            auc,
            bucket_error,
            mae,
            rmse,
            actual_ctr,
            predicted_ctr,
            copc,
            mean_predict_qvalue,
            total_ins_num,
        ) = self.get_global_metrics(
            scope,
            stat_pos_name,
            stat_neg_name,
            sqrerr_name,
            abserr_name,
            prob_name,
            q_name,
            pos_ins_num_name,
            total_ins_num_name,
        )
        self.rank0_print(
            "%s global AUC=%.6f BUCKET_ERROR=%.6f MAE=%.6f "
            "RMSE=%.6f Actural_CTR=%.6f Predicted_CTR=%.6f "
            "COPC=%.6f MEAN Q_VALUE=%.6f Ins number=%s"
            % (
                print_prefix,
                auc,
                bucket_error,
                mae,
                rmse,
                actual_ctr,
                predicted_ctr,
                copc,
                mean_predict_qvalue,
                total_ins_num,
            )
        )

    def program_type_trans(self, prog_dir, prog_fn, is_text):
        return utils.program_type_trans(prog_dir, prog_fn, is_text)

    def draw_from_program_file(
        self, model_filename, is_text, output_dir, output_filename
    ):
        """draw program from file"""
        program = utils.load_program(model_filename, is_text)
        utils.graphviz(program.global_block(), output_dir, output_filename)

    def draw_from_program(self, program, output_dir, output_name):
        """draw Program"""
        utils.graphviz(program.global_block(), output_dir, output_name)

    def check_two_programs(self, config):
        train_prog = utils.load_program(
            config.train_prog_path, config.is_text_train_program
        )
        pruned_prog = utils.load_program(
            config.pruned_prog_path, config.is_text_pruned_program
        )
        if config.draw:
            pruned_dir = os.path.dirname(config.pruned_prog_path)
            self.draw_from_program(
                pruned_prog, pruned_dir, config.draw_out_name
            )
        res = utils.check_pruned_program_vars(train_prog, pruned_prog)
        if res:
            _logger.info("check_programs succeed.")
        else:
            _logger.info(
                "check_programs failed. pruned program and train program not match!"
            )
        return res

    def check_vars_and_dump(self, config):
        _logger.info("start check_vars_and_dump.")
        results = utils.check_saved_vars_try_dump(
            config.dump_model_dir,
            config.dump_program_filename,
            config.is_text_dump_program,
            config.feed_config,
            config.fetch_config,
            config.batch_size,
            config.save_params_filename,
        )
        _logger.info("check_vars_and_dump succeed.")
        return results

    def parse_program_proto(self, prog_path, is_text, output_dir):
        """
        Parse program.proto into a more readable format.
        This function will generate three files:
        output_dir/vars_all.log,
        output_dir/vars_persistable.log,
        output_dir/ops.log.

        Args:
            prog_path(str): proto file path to be parsed.
            is_text(bool): proto file is human-readale format or not(binary).
            output_dir(str): output dir.

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import FleetUtil
              fleet_util = FleetUtil()
              program_path = "./program.pbtxt"
              is_text = True
              output_dir = "/tmp/"
              fleet_util.parse_program_proto(program_path, is_text, output_dir)
        """
        program = utils.load_program(prog_path, is_text)
        utils.parse_program(program, output_dir)

    def _is_optimizer_op(self, op):
        return self.op_role_key in op.attr_names and int(
            op.all_attrs()[self.op_role_key]
        ) & int(OpRole.Optimize)

    def split_program_by_device(self, program):
        ops_list = []
        type_list = []
        pre = None
        type_cpu = "cpu"
        for op in program.global_block().ops:
            if self._is_optimizer_op(op):
                break
            if op.has_attr("op_device"):
                cur_attr = (
                    op.attr("op_device")
                    if op.attr("op_device") != ""
                    else type_cpu
                )
                if pre is None or pre != cur_attr:
                    ops_list.append([])
                    type_list.append(cur_attr)
                ops_list[-1].append(op)
                pre = cur_attr
        l = len(type_list)
        i = 0
        type_heter = None
        while i < l:
            while i < l and type_list[i] == type_cpu:
                i += 1
            if i == l:
                break

            type_heter = type_list[i]
            i += 1
            start = i
            valid = True
            while i < l and type_list[i] != type_heter:
                if type_list[i] != type_cpu:
                    valid = False
                    break
                i += 1

            if i == l:
                break
            elif not valid:
                continue

            for j in range(start, i):
                for op in ops_list[j]:
                    op._set_attr("op_device", type_heter)
                type_list[j] = type_heter
                j += 1

        pre = None
        merged_ops_list = []
        merged_type_list = []
        for i in range(l):
            if pre is None or pre != type_list[i]:
                merged_ops_list.append([])
                merged_type_list.append(type_list[i])
            merged_ops_list[-1].extend(ops_list[i])
            pre = type_list[i]

        data_vars = set()
        for k in program.global_block().vars:
            var = program.global_block().var(k)
            if not var.persistable:
                data_vars.add(var.name)

        l = len(merged_ops_list)
        inputs_pre = set()
        outputs_pre = set()
        in_from_pre = [[] for i in range(l)]
        for i in range(l):
            inputs = set()
            outputs = set()
            for op in merged_ops_list[i]:
                for input in op.input_names:
                    for tmp in op.input(input):
                        if tmp not in outputs:
                            inputs.add(tmp)
                for output in op.output_names:
                    for tmp in op.output(output):
                        outputs.add(tmp)
            if i == 0:
                in_from_pre[i] = []
            elif i == 1:
                in_from_pre[i] = (outputs_pre | data_vars) & inputs
            else:
                in_from_pre[i] = outputs_pre & inputs
            inputs_pre = copy.deepcopy(inputs)
            outputs_pre = copy.deepcopy(outputs)

        l = len(in_from_pre)
        start_list = []
        end_list = []
        send_list = [[] for i in range(l)]
        sum = 0
        program_list = []
        for i in range(l):
            start_list.append(sum)
            end_list.append(sum + len(merged_ops_list[i]) - 1)
            sum += len(merged_ops_list[i])
            if i < l - 1:
                send_list[i].extend(list(in_from_pre[i + 1]))
            prog = program.clone()
            if merged_type_list[i] != type_cpu:
                prog = prog._prune_with_input(
                    list(in_from_pre[i]), list(send_list[i])
                )
                program_list.append(prog)
            else:
                program_list.append(prog)
        recv_list = [list(i) for i in in_from_pre]
        found = False
        heter_index = None
        for i in range(len(merged_type_list)):
            t = merged_type_list[i]
            if t != type_cpu:
                if found:
                    print("only one region of program can be heter")
                found = True
                heter_index = i
        if heter_index is None:
            print("warning: non heter program")
            return None
        else:
            return [
                start_list[heter_index],
                end_list[heter_index],
                send_list[heter_index],
                recv_list[heter_index],
                program_list[heter_index],
            ]


class GPUPSUtil(FleetUtil):
    """
    GPUPSUtil provides some common functions for users' convenience.

    Examples:
        .. code-block:: python

          from paddle.fluid.incubate.fleet.utils.fleet_util import GPUPSUtil
          fleet_util = GPUPSUtil()
          fleet_util.rank0_print("my log")
    """

    def __init__(self, fs_client=None):
        super(GPUPSUtil, self).__init__("pslib")
        self._afs = fs_client
        # self._afs = fs_client._fs

    def init(self, fs_name, fs_user, fs_passwd, fs_conf):
        r"""
        init for fs config

        Args:
            fs_name(str): fs name
            fs_user(str): fs user
            fs_passwd(str): fs password
            fs_conf(str): fs and afs conf path

        Returns:
            None

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import GPUPSUtil
              fleet_util = GPUPSUtil()
              fleet_util.init(20190722, 88, 88, "./afs.conf")
        """
        self._afs.init(fs_name, fs_user, fs_passwd, fs_conf)

    def set_fsclient(self, fs_client):
        r"""
        set fs_client for fs config

        Args:
            fs_client(AFSClient): fs_client object

        Returns:
            None

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import GPUPSUtil
              from paddle.distributed.fleet.utils.fs import AFSClient
              hdfs_client = AFSClient()
              fleet_util = GPUPSUtil()
              fleet_util.set_fsclient(hdfs_client)
        """
        self._afs = fs_client

    def get_last_save_xbox_base(self, output_path):
        r"""
        get last saved base xbox info from xbox_base_done.txt

        Args:
            output_path(str): output path

        Returns:
            [last_save_day, last_path, xbox_base_key]
            last_save_day(int): day of saved model
            last_path(str): model path
            xbox_base_key(int): xbox key

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import GPUPSUtil
              from paddle.distributed.fleet.utils.fs import AFSClient
              hdfs_client = AFSClient()
              fleet_util = GPUPSUtil()
              fleet_util.set_fsclient(hdfs_client)
              last_save_day, last_path, xbox_base_key = \
                  fleet_util.get_last_save_xbox_base("hdfs:/my/path")

        """
        donefile_path = output_path + "/xbox_base_done.txt"

        if not self._afs.is_file(donefile_path):
            return [-1, -1, int(time.time())]
        self._afs.download(donefile_path, "./xbox_base_done.txt")
        # pre_content = self._afs.cat(donefile_path)
        pre_content = ""
        with open("xbox_base_done.txt", "r") as f:
            pre_content = f.read()
        pre_content = pre_content.strip()
        last_dict = json.loads(pre_content.split("\n")[-1])
        last_day = int(last_dict["input"].split("/")[-3])
        last_path = "/".join(last_dict["input"].split("/")[:-1])
        xbox_base_key = int(last_dict["key"])
        return [last_day, last_path, xbox_base_key]

    def get_last_save_xbox(self, output_path):
        r"""
        get last saved xbox info from xbox_patch_done.txt

        Args:
            output_path(str): output path

        Returns:
            [last_save_day, last_save_pass, last_path, xbox_base_key]
            last_save_day(int): day of saved model
            last_save_pass(int): pass id of saved
            last_path(str): model path
            xbox_base_key(int): xbox key

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import GPUPSUtil
              from paddle.distributed.fleet.utils.fs import AFSClient
              hdfs_client = AFSClient()
              fleet_util = GPUPSUtil()
              fleet_util.set_fsclient(hdfs_client)
              last_save_day, last_save_pass, last_path, xbox_base_key = \
                  fleet_util.get_last_save_xbox("hdfs:/my/path")

        """
        donefile_path = output_path + "/xbox_patch_done.txt"

        if not self._afs.is_file(donefile_path):
            return [-1, -1, "", int(time.time())]
        self._afs.download(donefile_path, "xbox_patch_done.txt")
        pre_content = ""
        with open("xbox_patch_done.txt", "r") as f:
            pre_content = f.read()
        pre_content = pre_content.strip()
        last_dict = json.loads(pre_content.split("\n")[-1])
        last_day = int(last_dict["input"].split("/")[-3])
        last_pass = int(last_dict["input"].split("/")[-2].split("-")[-1])
        last_path = "/".join(last_dict["input"].split("/")[:-1])
        xbox_base_key = int(last_dict["key"])
        os.remove("xbox_patch_done.txt")
        return [last_day, last_pass, last_path, xbox_base_key]

    def get_last_save_model(self, output_path):
        r"""
        get last saved model info from donefile.txt

        Args:
            output_path(str): output path

        Returns:
            [last_save_day, last_save_pass, last_path, xbox_base_key]
            last_save_day(int): day of saved model
            last_save_pass(int): pass id of saved
            last_path(str): model path
            xbox_base_key(int): xbox key

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import GPUPSUtil
              from paddle.distributed.fleet.utils.fs import AFSClient
              hdfs_client = AFSClient()
              fleet_util = GPUPSUtil()
              fleet_util.set_fsclient(hdfs_client)
              last_save_day, last_save_pass, last_path, xbox_base_key = \
                  fleet_util.get_last_save_model("hdfs:/my/path")

        """
        last_save_day = -1
        last_save_pass = -1
        last_path = ""
        donefile_path = output_path + "/donefile.txt"
        if not self._afs.is_file(donefile_path):
            return [-1, -1, "", int(time.time())]
        self._afs.download(donefile_path, "./donefile.txt")
        content = ""
        with open("donefile.txt", "r") as f:
            content = f.read()
        content = content.strip().split("\n")[-1].split("\t")
        last_save_day = int(content[0])
        last_save_pass = int(content[3])
        last_path = content[2]
        xbox_base_key = int(content[1])
        os.remove("donefile.txt")
        return [last_save_day, last_save_pass, last_path, xbox_base_key]

    def write_model_donefile(
        self,
        output_path,
        day,
        pass_id,
        xbox_base_key,
        donefile_name="donefile.txt",
    ):
        """
        write donefile when save model

        Args:
            output_path(str): output path
            day(str|int): training day
            pass_id(str|int): training pass id
            xbox_base_key(str|int): xbox base key
            donefile_name(str): donefile name, default is "donefile.txt"

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import GPUPSUtil
              from paddle.distributed.fleet.utils.fs import AFSClient
              hdfs_client = AFSClient()
              fleet_util = GPUPSUtil()
              fleet_util.set_fsclient(hdfs_client)
              fleet_util.write_model_donefile(output_path="hdfs:/my/output",
                                              model_path="hdfs:/my/model",
                                              day=20190723,
                                              pass_id=66,
                                              xbox_base_key=int(time.time()))

        """
        day = str(day)
        pass_id = str(pass_id)
        xbox_base_key = int(xbox_base_key)

        if pass_id != "-1":
            suffix_name = "/%s/%s/" % (day, pass_id)
            model_path = output_path.rstrip("/") + suffix_name
        else:
            suffix_name = "/%s/0/" % day
            model_path = output_path.rstrip("/") + suffix_name

        if fleet.worker_index() == 0:
            donefile_path = output_path + "/" + donefile_name
            content = "%s\t%lu\t%s\t%s\t%d" % (
                day,
                xbox_base_key,
                model_path,
                pass_id,
                0,
            )
            if self._afs.is_file(donefile_path):
                self._afs.download(donefile_path, donefile_name)
                pre_content = ""
                with open(donefile_name, "r") as f:
                    pre_content = f.read()
                pre_content_list = pre_content.strip().split("\n")
                day_list = [i.split("\t")[0] for i in pre_content_list]
                pass_list = [i.split("\t")[3] for i in pre_content_list]
                os.remove(donefile_name)
                exist = False
                for i in range(len(day_list)):
                    if int(day) == int(day_list[i]) and int(pass_id) == int(
                        pass_list[i]
                    ):
                        exist = True
                        break
                if not exist:
                    with open(donefile_name, "w") as f:
                        f.write(pre_content.strip() + "\n")
                        f.write(content + "\n")
                    self._afs.delete(donefile_path)
                    self._afs.upload(donefile_name, donefile_path)
                    self.rank0_error(
                        "write %s/%s %s succeed" % (day, pass_id, donefile_name)
                    )
                else:
                    self.rank0_error(
                        "not write %s because %s/%s already "
                        "exists" % (donefile_name, day, pass_id)
                    )
            else:
                with open(donefile_name, "w") as f:
                    f.write(content + "\n")
                self._afs.upload(donefile_name, donefile_path)
                self.rank0_error(
                    "write %s/%s %s succeed" % (day, pass_id, donefile_name)
                )

    def write_xbox_donefile(
        self,
        output_path,
        day,
        pass_id,
        xbox_base_key,
        data_path,
        hadoop_fs_name,
        hadoop_fs_ugi,
        monitor_data={},
        hadoop_home="$HADOOP_HOME",
        donefile_name=None,
    ):
        """
        write delta donefile or xbox base donefile

        Args:
            output_path(str): output path
            day(str|int): training day of model
            pass_id(str|int): training pass id of model
            xbox_base_key(str|int): xbox base key
            data_path(str|list): training data path
            monitor_data(dict): metrics
            hadoop_home(str): hadoop home, default is "$HADOOP_HOME"
            donefile_name(str): donefile name, default is None"

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import GPUPSUtil
              from paddle.distributed.fleet.utils.fs import AFSClient
              hdfs_client = AFSClient()
              fleet_util = GPUPSUtil()
              fleet_util.set_fsclient(hdfs_client)
              fleet_util.write_xbox_donefile(
                  output_path="hdfs:/my/output/",
                  model_path="hdfs:/my/output/20190722/01",
                  day=20190722,
                  pass_id=1,
                  xbox_base_key=int(time.time()),
                  data_path="hdfs:/my/data/",
                  monitor_data={})

        """
        day = str(day)
        pass_id = str(pass_id)
        xbox_base_key = int(xbox_base_key)
        mode = None
        if pass_id != "-1":
            mode = "patch"
            suffix_name = "/%s/delta-%s/" % (day, pass_id)
            model_path = output_path.rstrip("/") + suffix_name
            if donefile_name is None:
                donefile_name = "xbox_patch_done.txt"
        else:
            mode = "base"
            suffix_name = "/%s/base/" % day
            model_path = output_path.rstrip("/") + suffix_name
            if donefile_name is None:
                donefile_name = "xbox_base_done.txt"

        if isinstance(data_path, list):
            data_path = ",".join(data_path)
        if fleet.worker_index() == 0:
            donefile_path = output_path + "/" + donefile_name
            xbox_str = self._get_xbox_str(
                output_path,
                day,
                model_path,
                xbox_base_key,
                data_path,
                hadoop_fs_name,
                monitor_data={},
                mode=mode,
            )

            if self._afs.is_exist(donefile_path):
                self.rank0_info("exist %s succeed" % (donefile_path))
                self._afs.download(donefile_path, donefile_name)
                pre_content = ""
                with open(donefile_name, "r") as f:
                    pre_content = f.read()
                last_dict = json.loads(pre_content.strip().split("\n")[-1])
                last_day = last_dict["input"].split("/")[-3]
                last_pass = last_dict["input"].split("/")[-2].split("-")[-1]

                os.remove(donefile_name)
                self.rank0_info("remove %s succeed" % (donefile_name))
                exist = False
                if (
                    int(day) < int(last_day)
                    or int(day) == int(last_day)
                    and int(pass_id) <= int(last_pass)
                ):
                    exist = True
                if not exist:
                    with open(donefile_name, "w") as f:
                        f.write(pre_content.strip() + "\n")
                        f.write(xbox_str + "\n")
                    self._afs.delete(donefile_path)
                    self._afs.upload(donefile_name, donefile_path)
                    self.rank0_info(
                        "write %s/%s %s succeed" % (day, pass_id, donefile_name)
                    )
                else:
                    self.rank0_info(
                        "not write %s because %s/%s already "
                        "exists" % (donefile_name, day, pass_id)
                    )
            else:
                with open(donefile_name, "w") as f:
                    f.write(xbox_str + "\n")
                self._afs.upload(donefile_name, donefile_path)
                self.rank0_error(
                    "write %s/%s %s succeed" % (day, pass_id, donefile_name)
                )

    def write_cache_donefile(
        self,
        output_path,
        day,
        pass_id,
        key_num,
        donefile_name="sparse_cache.meta",
        **kwargs
    ):
        """
        write cache donefile

        Args:
            output_path(str): output path
            day(str|int): training day of model
            pass_id(str|int): training pass id of model
            key_num(str|int): save cache return value
            donefile_name(str): donefile name, default is "sparse_cache.meta"
            kwargs(dict): user defined properties
                          file_num(int): cache file num
                          table_id(int): cache table id

        Examples:
            .. code-block:: python

              from paddle.fluid.incubate.fleet.utils.fleet_util import GPUPSUtil
              from paddle.distributed.fleet.utils.fs import AFSClient
              hdfs_client = AFSClient()
              fleet_util = GPUPSUtil()
              fleet_util.set_fsclient(hdfs_client)
              fleet_util.write_cache_donefile(
                  output_path="hdfs:/my/output/",
                  day=20190722,
                  pass_id=1,
                  key_num=123456)

        """
        day = str(day)
        pass_id = str(pass_id)
        key_num = int(key_num)
        file_num = kwargs.get("file_num", 16)
        table_id = kwargs.get("table_id", 0)

        if pass_id != "-1":
            suffix_name = "/%s/delta-%s/%03d_cache" % (day, pass_id, table_id)
            model_path = output_path.rstrip("/") + suffix_name
        else:
            suffix_name = "/%s/base/%03d_cache" % (day, table_id)
            model_path = output_path.rstrip("/") + suffix_name

        if fleet.worker_index() == 0:
            donefile_path = model_path + "/" + donefile_name

            if self._afs.is_file(donefile_path):
                self.rank0_error(
                    "not write because %s already exists" % donefile_path
                )
            else:
                meta_str = "file_prefix:part\npart_num:%s\nkey_num:%d\n" % (
                    file_num,
                    key_num,
                )
                with open(donefile_name, "w") as f:
                    f.write(meta_str)
                self._afs.upload(donefile_name, donefile_path)
                self.rank0_error("write %s succeed" % donefile_path)

    def _get_xbox_str(
        self,
        output_path,
        day,
        model_path,
        xbox_base_key,
        data_path,
        hadoop_fs_name,
        monitor_data={},
        mode="patch",
    ):
        xbox_dict = collections.OrderedDict()
        if mode == "base":
            xbox_dict["id"] = str(xbox_base_key)
        elif mode == "patch":
            xbox_dict["id"] = str(int(time.time()))
        else:
            print("warning: unknown mode %s, set it to patch" % mode)
            mode = "patch"
            xbox_dict["id"] = str(int(time.time()))
        xbox_dict["key"] = str(xbox_base_key)
        if model_path.startswith("hdfs:") or model_path.startswith("afs:"):
            model_path = model_path[model_path.find(":") + 1 :]
        xbox_dict["input"] = hadoop_fs_name + model_path.rstrip("/") + "/000"
        xbox_dict["record_count"] = "111111"
        xbox_dict["partition_type"] = "2"
        xbox_dict["job_name"] = "default_job_name"
        xbox_dict["ins_tag"] = "feasign"
        xbox_dict["ins_path"] = data_path
        xbox_dict["job_id"] = os.environ.get("PADDLE_JOB_ID", "")
        # currently hard code here, set monitor_data empty string
        xbox_dict["monitor_data"] = ""
        xbox_dict["monitor_path"] = (
            output_path.rstrip("/") + "/monitor/" + day + ".txt"
        )
        xbox_dict["mpi_size"] = str(fleet.worker_num())
        return json.dumps(xbox_dict)