private Boolean run(String args[]) throws IOException {
        try {
            if (conf.getBoolean(
                    HboxConfiguration.HBOX_TF_INPUT_PS_ENABLE, HboxConfiguration.DEFAULT_HBOX_TF_INPUT_PS_ENABLE)) {
                prepareInputFiles();
            } else if (conf.get(HboxConfiguration.HBOX_INPUT_STRATEGY, HboxConfiguration.DEFAULT_HBOX_INPUT_STRATEGY)
                            .equals("PLACEHOLDER")
                    && conf.getBoolean(
                            HboxConfiguration.HBOX_PLACEHOLDER_WHOLE_ENABLE,
                            HboxConfiguration.DEFAULT_HBOX_PLACEHOLDER_WHOLE_ENABLE)) {
                prepareInputFiles();
            } else {
                if (this.role.equals(HboxConstants.WORKER)) {
                    prepareInputFiles();
                } else if (conf.getBoolean(
                        HboxConfiguration.HBOX_TF_EVALUATOR, HboxConfiguration.DEFAULT_HBOX_TF_EVALUATOR)) {
                    createLocalInputDir();
                }
            }
            if (this.conf.getBoolean(
                    HboxConfiguration.HBOX_CONTAINER_AUTO_CREATE_OUTPUT_DIR,
                    HboxConfiguration.DEFAULT_HBOX_CONTAINER_AUTO_CREATE_OUTPUT_DIR)) {
                createLocalOutputDir();
            }
        } catch (InterruptedException e) {
            LOG.error("Container prepare inputs failed!", e);
            this.reportFailedAndExit();
        } catch (ExecutionException e) {
            LOG.error("Container prepare inputs failed!", e);
            this.reportFailedAndExit();
        }

        if (hboxAppType.equals("MPI") || hboxAppType.equals("TENSORNET") || hboxAppType.equals("HOROVOD")) {
            reLinksFiles();
        }

        if (("TENSORFLOW".equals(hboxAppType) || "TENSOR2TENSOR".equals(hboxAppType)) && !single) {
            LOG.info("Reserved available port: " + reservedSocket.getLocalPort());
            if (!this.role.equals(HboxConstants.EVALUATOR)
                    || conf.getBoolean(
                            HboxConfiguration.HBOX_TF_DISTRIBUTION_STRATEGY,
                            HboxConfiguration.DEFAULT_HBOX_TF_DISTRIBUTION_STRATEGY)) {
                amClient.reportReservedPort(
                        envs.get(ApplicationConstants.Environment.NM_HOST.toString()),
                        reservedSocket.getLocalPort(),
                        this.role,
                        this.index);
            }
            this.localAddress =
                    envs.get(ApplicationConstants.Environment.NM_HOST.toString()) + ":" + reservedSocket.getLocalPort();
            LOG.info("Current localAddress: " + this.localAddress);

            while (!heartbeatThread.isHboxTrainCompleted()) {
                // TODO may be need encode use Base64 while used in Env
                this.clusterDef = amClient.getClusterDef();
                if (this.clusterDef != null) {
                    LOG.info("Cluster def is: " + this.clusterDef);
                    break;
                }
                Utilities.sleep(this.conf.getInt(
                        HboxConfiguration.HBOX_CONTAINER_UPDATE_APP_STATUS_INTERVAL,
                        HboxConfiguration.DEFAULT_HBOX_CONTAINER_UPDATE_APP_STATUS_INTERVAL));
            }
            if (heartbeatThread.isHboxTrainCompleted()) {
                return false;
            }
            Gson gson = new Gson();
            Map<String, Object> clusterInfo = new HashMap<>();
            Map<String, Object> taskInfo = new HashMap<>();
            Map<String, Object> tfConfigInfo = new HashMap<>();
            switch (hboxAppType) {
                case "TENSORFLOW":
                    {
                        // cluster info
                        clusterInfo = (Map) gson.fromJson(this.clusterDef, clusterInfo.getClass());
                        // task info
                        taskInfo.put("type", this.role);
                        taskInfo.put("index", this.index);
                        tfConfigInfo.put("cluster", clusterInfo);
                        tfConfigInfo.put("task", taskInfo);
                        LOG.info("Tensorflow distribute mode needs the TF_CONFIG: " + new Gson().toJson(tfConfigInfo));
                    }
                    break;
                case "TENSOR2TENSOR":
                    {
                        if (this.role.equals("ps")) {
                            taskInfo.put("type", this.role);
                        } else {
                            taskInfo.put("type", "master");
                        }
                        taskInfo.put("index", this.index);
                        clusterInfo = (Map)
                                gson.fromJson(this.clusterDef.replaceAll("worker", "master"), clusterInfo.getClass());
                        tfConfigInfo.put("cluster", clusterInfo);
                        tfConfigInfo.put("task", taskInfo);
                        tfConfigInfo.put("environment", "cloud");
                        LOG.info("Tensor2Tensor distribute mode needs the TF_CONFIG: "
                                + new Gson().toJson(tfConfigInfo));
                    }
                    break;
            }
            this.tfConfig = new Gson().toJson(tfConfigInfo);
        }

        if (hboxAppType.equals("DISTLIGHTGBM")) {
            LOG.info("Reserved available port: " + reservedSocket.getLocalPort());
            this.lightGBMLocalPort = reservedSocket.getLocalPort();
            InetAddress address = null;
            try {
                address = InetAddress.getByName(envs.get(ApplicationConstants.Environment.NM_HOST.toString()));
            } catch (UnknownHostException e) {
                LOG.info("acquire host ip failed " + e);
                reportFailedAndExit();
            }
            String ipPortStr = address.getHostAddress() + " " + reservedSocket.getLocalPort();
            LOG.info("lightGBM ip port string is: " + ipPortStr);
            amClient.reportLightGbmIpPort(containerId, ipPortStr);
            String lightGBMIpPortStr = null;
            while (!heartbeatThread.isHboxTrainCompleted()) {
                // TODO may be need encode use Base64 while used in Env
                lightGBMIpPortStr = amClient.getLightGbmIpPortStr();
                if (lightGBMIpPortStr != null) {
                    LOG.info("lightGBM IP PORT list is: " + lightGBMIpPortStr);
                    break;
                }
                Utilities.sleep(this.conf.getInt(
                        HboxConfiguration.HBOX_CONTAINER_UPDATE_APP_STATUS_INTERVAL,
                        HboxConfiguration.DEFAULT_HBOX_CONTAINER_UPDATE_APP_STATUS_INTERVAL));
            }
            if (heartbeatThread.isHboxTrainCompleted()) {
                return false;
            }
            Type type = new TypeToken<ConcurrentHashMap<String, String>>() {}.getType();
            ConcurrentHashMap<String, String> map = new Gson().fromJson(lightGBMIpPortStr, type);
            PrintWriter writer = new PrintWriter("lightGBMlist.txt", "UTF-8");
            for (String str : map.keySet()) {
                writer.println(map.get(str));
            }
            writer.close();
        }

        if (hboxAppType.equals("DISTLIGHTLDA")) {
            if (this.role.equals(HboxConstants.PS)) {
                LOG.info("Reserved available port: " + reservedSocket.getLocalPort());
                this.lightLDALocalPort = reservedSocket.getLocalPort();
                InetAddress address = null;
                try {
                    address = InetAddress.getByName(envs.get(ApplicationConstants.Environment.NM_HOST.toString()));
                } catch (UnknownHostException e) {
                    LOG.info("acquire host ip failed " + e);
                    reportFailedAndExit();
                }
                String ipPortStr = this.index + " " + address.getHostAddress() + ":" + reservedSocket.getLocalPort();
                this.lightLDAEndpoint = address.getHostAddress() + ":" + reservedSocket.getLocalPort();
                LOG.info("lightLDA ip port string is: " + ipPortStr);
                amClient.reportLightLdaIpPort(containerId, ipPortStr);
            }
            if (this.role.equals(HboxConstants.WORKER)) {
                String lightLDAIpPortStr = null;
                while (!heartbeatThread.isHboxTrainCompleted()) {
                    // TODO may be need encode use Base64 while used in Env
                    lightLDAIpPortStr = amClient.getLightLdaIpPortStr();
                    if (lightLDAIpPortStr != null) {
                        LOG.info("lightLDA IP PORT list is: " + lightLDAIpPortStr);
                        break;
                    }
                    Utilities.sleep(this.conf.getInt(
                            HboxConfiguration.HBOX_CONTAINER_UPDATE_APP_STATUS_INTERVAL,
                            HboxConfiguration.DEFAULT_HBOX_CONTAINER_UPDATE_APP_STATUS_INTERVAL));
                }
                if (heartbeatThread.isHboxTrainCompleted()) {
                    return false;
                }
                Type type = new TypeToken<ConcurrentHashMap<String, String>>() {}.getType();
                ConcurrentHashMap<String, String> map = new Gson().fromJson(lightLDAIpPortStr, type);
                PrintWriter writer = new PrintWriter("lightLdaEndPoints.txt", "UTF-8");
                for (String str : map.keySet()) {
                    writer.println(map.get(str));
                }
                writer.close();
            }
        }

        if (hboxAppType.equals("DISTTORCH")) {
            if (this.index == 0) {
                this.torchRank0Port = reservedSocket.getLocalPort();
                LOG.info("Reserved available port: " + torchRank0Port);
                try {
                    InetAddress address =
                            InetAddress.getByName(envs.get(ApplicationConstants.Environment.NM_HOST.toString()));
                    this.torchRank0IP = address.getHostAddress() + ":" + torchRank0Port;
                } catch (UnknownHostException e) {
                    LOG.info("acquire host ip failed " + e);
                    reportFailedAndExit();
                }
                LOG.info("torch rank 0 ip port string is: " + torchRank0Port);
                amClient.reportTorchRank0IP(torchRank0IP);
            } else {
                while (!heartbeatThread.isHboxTrainCompleted()) {
                    // TODO may be need encode use Base64 while used in Env
                    this.torchRank0IP = amClient.getTorchRank0IP();
                    if (torchRank0IP != null) {
                        LOG.info("Torch Rank 0 IP:Port is: " + torchRank0IP);
                        break;
                    }
                    Utilities.sleep(this.conf.getInt(
                            HboxConfiguration.HBOX_CONTAINER_UPDATE_APP_STATUS_INTERVAL,
                            HboxConfiguration.DEFAULT_HBOX_CONTAINER_UPDATE_APP_STATUS_INTERVAL));
                }
                if (heartbeatThread.isHboxTrainCompleted()) {
                    return false;
                }
            }
        }

        /**
         * set TF_CLUSTER_DEF in env
         * python script can load cluster def use "json.loads(os.environ["TF_CLUSTER_DEF"])"
         */
        List<String> envList = new ArrayList<>(20);

        LOG.debug("hadoop user name:" + System.getenv(HboxConstants.Environment.HADOOP_USER_NAME.toString()));
        String containerExecType = conf.get(
                        HboxConfiguration.CONTAINER_EXECUTOR_TYPE, HboxConfiguration.DEFAULT_CONTAINER_EXECUTOR_TYPE)
                .toUpperCase();

        if (conf.get(HboxConfiguration.HBOX_CONTAINER_ENV) != null) {
            String[] env = StringUtils.split(conf.get(HboxConfiguration.HBOX_CONTAINER_ENV), "|");
            for (String envPair : env) {
                if (StringUtils.split(envPair, "=").length != 2) {
                    LOG.error(envPair + " is not correct");
                } else {
                    envList.add(envPair);
                }
            }
        }

        inheritEnv(envList, "PATH");
        inheritEnv(envList, "JAVA_HOME");
        envList.add("INDEX=" + this.index);
        envList.add("HADOOP_VERSION=" + VersionInfo.getVersion());
        pathListEnv(envList, "HADOOP_CONF_DIR", "./", System.getenv("HADOOP_CONF_DIR"));
        inheritEnv(envList, "HADOOP_HOME");
        inheritEnv(envList, "HADOOP_HDFS_HOME");
        inheritEnv(envList, "HADOOP_USER_NAME");
        inheritEnv(envList, "HADOOP_COMMON_HOME");
        inheritEnv(envList, "HADOOP_YARN_HOME");
        inheritEnv(envList, "CONTAINER_ID");
        if (null != System.getenv("JAVA_HOME")) {
            pathListEnv(
                    envList,
                    "LD_LIBRARY_PATH",
                    System.getenv("LD_LIBRARY_PATH"),
                    System.getenv("JAVA_HOME") + "/jre/lib/amd64/server",
                    System.getenv("HADOOP_HOME"));
        } else {
            pathListEnv(envList, "LD_LIBRARY_PATH", System.getenv("LD_LIBRARY_PATH"), System.getenv("HADOOP_HOME"));
        }
        pathListEnv(envList, "CLASSPATH", "./", System.getenv("CLASSPATH"), System.getProperty("java.class.path"));
        envList.add("PYTHONUNBUFFERED=1");
        envList.add(
                "HBOX_CONTAINER_LOG_DIR=" + System.getenv(HboxConstants.Environment.HBOX_CONTAINER_LOG_DIR.toString()));

        if ("TENSORFLOW".equals(hboxAppType) || "TENSOR2TENSOR".equals(hboxAppType)) {
            envList.add(HboxConstants.Environment.HBOX_TF_INDEX.toString() + "=" + this.index);
            envList.add(HboxConstants.Environment.HBOX_TF_ROLE.toString() + "=" + this.role);
            envList.add(HboxConstants.Environment.HBOX_TN_INDEX.toString() + "=" + this.index);
            envList.add(HboxConstants.Environment.HBOX_TN_ROLE.toString() + "=" + this.role);
            envList.add(HboxConstants.Environment.HBOX_TN_CLUSTER_DEF.toString() + "=" + this.clusterDef);
            if (!single) {
                envList.add(HboxConstants.Environment.HBOX_TF_CONFIG.toString() + "=" + this.tfConfig);
                envList.add(HboxConstants.Environment.HBOX_TF_CLUSTER_DEF.toString() + "=" + this.clusterDef);
                envList.add(HboxConstants.Environment.HBOX_LOCAL_ADDRESS.toString() + "=" + this.localAddress);
            }
        } else if (hboxAppType.equals("MXNET")) {
            if (!singleMx) {
                String dmlcID;
                if (this.role.equals("worker")) {
                    dmlcID = "DMLC_WORKER_ID";
                } else {
                    dmlcID = "DMLC_SERVER_ID";
                }
                envList.add("DMLC_PS_ROOT_URI=" + System.getenv("DMLC_PS_ROOT_URI"));
                envList.add("DMLC_PS_ROOT_PORT=" + System.getenv("DMLC_PS_ROOT_PORT"));
                envList.add("DMLC_NUM_WORKER=" + System.getenv("DMLC_NUM_WORKER"));
                envList.add("DMLC_NUM_SERVER=" + System.getenv("DMLC_NUM_SERVER"));
                envList.add(dmlcID + "=" + this.index);
                envList.add("DMLC_ROLE=" + this.role);
            }
        } else if (hboxAppType.equals("DISTXGBOOST")) {
            envList.add("DMLC_TRACKER_URI=" + System.getenv("DMLC_TRACKER_URI"));
            envList.add("DMLC_TRACKER_PORT=" + System.getenv("DMLC_TRACKER_PORT"));
            envList.add("DMLC_NUM_WORKER=" + System.getenv("DMLC_NUM_WORKER"));
            envList.add("DMLC_TASK_ID=" + this.index);
            envList.add("DMLC_ROLE=" + this.role);
        } else if (hboxAppType.equals("DISTLIGHTGBM")) {
            envList.add("LIGHTGBM_NUM_MACHINE="
                    + System.getenv(HboxConstants.Environment.HBOX_LIGHTGBM_WORKER_NUM.toString()));
            envList.add("LIGHTGBM_LOCAL_LISTEN_PORT=" + this.lightGBMLocalPort);
        } else if (hboxAppType.equals("DISTLIGHTLDA")) {
            envList.add("LIGHTLDA_WORKER_NUM="
                    + System.getenv(HboxConstants.Environment.HBOX_LIGHTLDA_WORKER_NUM.toString()));
            envList.add(
                    "LIGHTLDA_SERVER_NUM=" + System.getenv(HboxConstants.Environment.HBOX_LIGHTLDA_PS_NUM.toString()));
            envList.add("LIGHTLDA_RANK=" + this.index);
            envList.add("LIGHTLDA_SERVER_ENDPOINT=" + this.lightLDAEndpoint);
            envList.add("LIGHTLDA_ROLE=" + this.role);
        } else if (hboxAppType.equals("VPC")) {
            envList.add("PYTHONPATH=" + System.getenv("PYTHONPATH"));
        } else if (hboxAppType.equals("DIGITS")) {
            envList.add("PYTHONPATH=" + System.getenv("PYTHONPATH"));
        } else if (hboxAppType.equals("MPI") || hboxAppType.equals("TENSORNET") || hboxAppType.equals("HOROVOD")) {
            final String mpiExtraLdLibraryPath = conf.get(HboxConfiguration.HBOX_MPI_EXTRA_LD_LIBRARY_PATH);
            if (mpiExtraLdLibraryPath != null) {
                LOG.info("prepend " + mpiExtraLdLibraryPath + " to LD_LIBRARY_PATH");
            }

            String mpiInstallDir = Paths.get(conf.get(
                            HboxConfiguration.HBOX_MPI_INSTALL_DIR, HboxConfiguration.DEFAULT_HBOX_MPI_INSTALL_DIR))
                    .toAbsolutePath()
                    .toString();

            envList.add("OPAL_PREFIX=" + mpiInstallDir);
            // for rsh agent, will use $HOME as working dir
            envList.add("HOME=" + this.mpiAppDir);
            envList.add("PWD=" + this.mpiAppDir);
            envList.add(HboxConstants.Environment.HBOX_TF_INDEX + "=" + this.index);
            envList.add("HBOX_CUSTOM_EXIT="
                    + conf.get(HboxConfiguration.HBOX_CUSTOM_EXIT, HboxConfiguration.DEFAULT_HBOX_CUSTOM_EXIT));
            pathListEnv(
                    envList,
                    "LD_LIBRARY_PATH",
                    mpiExtraLdLibraryPath,
                    mpiInstallDir + File.separator + "lib",
                    mpiInstallDir + File.separator + "lib" + File.separator + "openmpi",
                    mpiInstallDir + File.separator + "lib" + File.separator + "pmix",
                    System.getenv("LD_LIBRARY_PATH"));
        } else if (hboxAppType.equals("XFLOW")) {
            String dmlcID;
            String heapprofile;
            if (this.role.equals("worker")) {
                dmlcID = "DMLC_WORKER_ID";
                heapprofile = "./W";
            } else {
                dmlcID = "DMLC_SERVER_ID";
                heapprofile = "./S";
            }
            envList.add("DMLC_PS_ROOT_URI=" + System.getenv("DMLC_PS_ROOT_URI"));
            envList.add("DMLC_PS_ROOT_PORT=" + System.getenv("DMLC_PS_ROOT_PORT"));
            envList.add("DMLC_NUM_WORKER=" + System.getenv("DMLC_NUM_WORKER"));
            envList.add("DMLC_NUM_SERVER=" + System.getenv("DMLC_NUM_SERVER"));
            envList.add(dmlcID + "=" + this.index);
            envList.add("DMLC_ROLE=" + this.role);
            envList.add("HEAPPROFILE=" + heapprofile + this.index);
        } else if (hboxAppType.equals("DISTTORCH")) {
            envList.add("INIT_METHOD=tcp://" + this.torchRank0IP);
            envList.add("RANK=" + this.index);
            envList.add("WORLD_SIZE=" + System.getenv("WORLD_SIZE"));
            String[] torchRank0IPs = this.torchRank0IP.split(":");
            envList.add("master_addr=" + torchRank0IPs[0]);
            envList.add("master_port=" + torchRank0IPs[1]);
        } else if (hboxAppType.equals("XDL")) {
            if (!conf.getBoolean(
                    HboxConfiguration.HBOX_TF_MODE_SINGLE, HboxConfiguration.DEFAULT_HBOX_TF_MODE_SINGLE)) {
                envList.add("TASK_NAME=" + this.role);
                envList.add("TASK_INDEX=" + this.index);
                envList.add("ZK_ADDR=" + envs.get("ZK_ADDR"));
                if (role.equalsIgnoreCase(HboxConstants.SCHEDULER)) {
                    envList.add("PS_NUM=" + envs.get("PS_NUM"));
                    envList.add("PS_CPU_CORES=" + envs.get("PS_CPU_CORES"));
                    envList.add("PS_MEMORY_M=" + envs.get("PS_MEMORY_M"));
                } else if (role.equalsIgnoreCase(HboxConstants.WORKER)) {
                    envList.add("TASK_NUM=" + envs.get("TASK_NUM"));
                }
            }
        }
        // if user's environments too long, write environments to file inputFileList.txt, s3 input to file
        // s3InputFileList.txt
        if (conf.get(HboxConfiguration.HBOX_INPUT_STRATEGY, HboxConfiguration.DEFAULT_HBOX_INPUT_STRATEGY)
                .equals("PLACEHOLDER")) {
            if (this.inputFileList != null && this.inputFileList.trim() != "") {
                envList.add(HboxConstants.Environment.HBOX_INPUT_FILE_LIST.toString() + "=" + this.inputFileList);
                if (envList.toString().length()
                        > conf.getInt(
                                HboxConfiguration.HBOX_ENV_MAXLENGTH, HboxConfiguration.DEFAULT_HBOX_ENV_MAXLENGTH)) {
                    LOG.warn("Current container environments length "
                            + envList.toString().length() + " exceed the configuration "
                            + HboxConfiguration.HBOX_ENV_MAXLENGTH + " "
                            + conf.getInt(
                                    HboxConfiguration.HBOX_ENV_MAXLENGTH,
                                    HboxConfiguration.DEFAULT_HBOX_ENV_MAXLENGTH));
                    envList.remove(envList.size() - 1);
                    LOG.warn("InputFile list had written to local file: inputFileList.txt !!");
                    PrintWriter writer = new PrintWriter("inputFileList.txt", "UTF-8");
                    writer.println(this.inputFileList);
                    writer.close();
                }
            }
            if (this.s3InputUrlList != null && this.s3InputUrlList.trim() != "") {
                envList.add(HboxConstants.Environment.HBOX_S3_INPUT_FILE_LIST.toString() + "=" + this.s3InputUrlList);
                if (envList.toString().length()
                        > conf.getInt(
                                HboxConfiguration.HBOX_ENV_MAXLENGTH, HboxConfiguration.DEFAULT_HBOX_ENV_MAXLENGTH)) {
                    LOG.warn("Current container environments length "
                            + envList.toString().length() + " exceed the configuration "
                            + HboxConfiguration.HBOX_ENV_MAXLENGTH + " "
                            + conf.getInt(
                                    HboxConfiguration.HBOX_ENV_MAXLENGTH,
                                    HboxConfiguration.DEFAULT_HBOX_ENV_MAXLENGTH));
                    envList.remove(envList.size() - 1);
                    LOG.warn("InputFile list had written to local file: s3InputFileList.txt !!");
                    PrintWriter writer = new PrintWriter("s3InputFileList.txt", "UTF-8");
                    writer.println(this.s3InputUrlList);
                    writer.close();
                }
            }
        }

        // String containerType = conf.get(HboxConfiguration.CONTAINER_EXECUTOR_TYPE,
        // HboxConfiguration.DEFAULT_CONTAINER_EXECUTOR_TYPE).toUpperCase();
        if (hboxAppType.equals("VPC") || hboxAppType.equals("DIGITS")) {
            String dockerPort = envs.get("DOCKER_PORT");
            String password = envs.get("DOCKER_PASSWORD");
            String vpcCommandAndPasswd = "root@" + envs.get(ApplicationConstants.Environment.NM_HOST.toString())
                    + " -p " + dockerPort + ":" + password;
            amClient.reportVPCCommandAndPasswd(containerId, vpcCommandAndPasswd);
            String duration =
                    this.conf.get(HboxConfiguration.HBOX_VPC_DURATION, HboxConfiguration.DEFAULT_HBOX_VPC_DURATION);
            if (hboxAppType.equals("VPC")) {
                if (duration.equals("0")) {
                    args = new String[] {"sleep", "32767d"};
                } else {
                    args = new String[] {"sleep", duration};
                }
            } else {
                int digitsPort = reservedSocket.getLocalPort();
                LOG.info("Reserved digits available port: " + digitsPort);
                String digitsUrl =
                        "http://" + envs.get(ApplicationConstants.Environment.NM_HOST.toString()) + ":" + digitsPort;
                amClient.reportDigitsUrl(containerId, digitsUrl);
                String digitsShellname = "digits_" + containerId.toString() + ".sh";
                String digitsServerCmd = "python -m digits -p " + digitsPort + " &";
                String digitsSleepCmd;
                if (duration.equals("0")) {
                    digitsSleepCmd = "sleep 32767d";
                } else {
                    digitsSleepCmd = "sleep " + duration;
                }
                PrintWriter writer = new PrintWriter(digitsShellname, "UTF-8");
                writer.println(digitsServerCmd);
                writer.println(digitsSleepCmd);
                writer.close();
                args = new String[] {"/bin/sh", digitsShellname};
            }
        } else if (hboxAppType.equals("MPI") || hboxAppType.equals("TENSORNET") || hboxAppType.equals("HOROVOD")) {

            String mpiInstallDir = Paths.get(conf.get(
                            HboxConfiguration.HBOX_MPI_INSTALL_DIR, HboxConfiguration.DEFAULT_HBOX_MPI_INSTALL_DIR))
                    .toAbsolutePath()
                    .toString();

            // If command not starts with absolute path, should add mpiInstallDir prefix to command
            if (args.length > 0 && !args[0].startsWith("/")) {
                args[0] = mpiInstallDir + File.separator + "bin" + File.separator + args[0];
            }

        } else {
            if (containerExecType.equals("DOCKER")) {
                String dockerPort = envs.get("DOCKER_PORT");
                String password = envs.get("DOCKER_PASSWORD");
                String vpcCommandAndPasswd = "root@" + envs.get(ApplicationConstants.Environment.NM_HOST.toString())
                        + " -p " + dockerPort + ":" + password;
                amClient.reportVPCCommandAndPasswd(containerId, vpcCommandAndPasswd);
            }
        }
        LOG.info("Executing command: " + String.join(" ", args));

        // propagate opentelemetry context to the subprocess
        HboxOpenTelemetry.injectIntoEnvList(envList);

        Runtime rt = Runtime.getRuntime();

        // close reserved socket as tf will bind this port later
        this.reservedSocket.close();
        String[] env = envList.toArray(new String[envList.size()]);
        File dir = null;
        if (hboxAppType.equals("MPI") || hboxAppType.equals("TENSORNET") || hboxAppType.equals("HOROVOD")) {
            dir = new File(this.mpiAppDir);
        }
        final Process hboxProcess = containerLaunch.exec(args, env, envs, dir);

        final Span span = Span.current();
        span.addEvent("process_launched");

        Date now = new Date();
        heartbeatThread.setContainersStartTime(now.toString());

        if (this.conf.getBoolean(HboxConfiguration.HBOX_INPUT_STREAM, HboxConfiguration.DEFAULT_HBOX_INPUT_STREAM)
                || conf.get(HboxConfiguration.HBOX_INPUT_STRATEGY, HboxConfiguration.DEFAULT_HBOX_INPUT_STRATEGY)
                        .equals("STREAM")) {
            LOG.info("Starting thread to redirect stdin of hbox process");
            Thread stdinRedirectThread = new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        OutputStreamWriter osw = new OutputStreamWriter(hboxProcess.getOutputStream());
                        File gzFile = new File("inputformatCache.gz");
                        GZIPOutputStream gos = new GZIPOutputStream(new FileOutputStream(gzFile));
                        boolean isCache = conf.getBoolean(
                                HboxConfiguration.HBOX_INPUTFORMAT_CACHE,
                                HboxConfiguration.DEFAULT_HBOX_INPUTFORMAT_CACHE);
                        List<InputSplit> inputs = Arrays.asList(amClient.getStreamInputSplit(containerId));
                        JobConf jobConf = new JobConf(conf);
                        RecordReader reader;
                        InputFormat inputFormat = ReflectionUtils.newInstance(
                                conf.getClass(
                                        HboxConfiguration2.HBOX_INPUTF0RMAT_CLASS,
                                        HboxConfiguration2.DEFAULT_HBOX_INPUTF0RMAT_CLASS,
                                        InputFormat.class),
                                jobConf);
                        for (int j = 0;
                                j
                                        < conf.getInt(
                                                HboxConfiguration.HBOX_STREAM_EPOCH,
                                                HboxConfiguration.DEFAULT_HBOX_STREAM_EPOCH);
                                j++) {
                            LOG.info("Epoch " + (j + 1) + " starting...");
                            for (int i = 0, len = inputs.size(); i < len; i++) {
                                LOG.info("split " + (i + 1) + " is handling...");
                                reader = inputFormat.getRecordReader(inputs.get(i), jobConf, Reporter.NULL);
                                Object key = reader.createKey();
                                Object value = reader.createValue();
                                Boolean finished = false;
                                while (!finished) {
                                    try {
                                        finished = !reader.next(key, value);
                                        if (finished) {
                                            break;
                                        }
                                        osw.write(value.toString());
                                        osw.write("\n");
                                        if (j == 0 && isCache) {
                                            if (conf.getInt(
                                                            HboxConfiguration.HBOX_STREAM_EPOCH,
                                                            HboxConfiguration.DEFAULT_HBOX_STREAM_EPOCH)
                                                    > 1) {
                                                gos.write(value.toString().getBytes());
                                                gos.write("\n".getBytes());
                                                if ((gzFile.length() / 1024 / 1024)
                                                        > conf.getInt(
                                                                HboxConfiguration.HBOX_INPUTFORMAT_CACHESIZE_LIMIT,
                                                                HboxConfiguration
                                                                        .DEFAULT_HBOX_INPUTFORMAT_CACHESIZE_LIMIT)) {
                                                    LOG.info("Inputformat cache file size is:"
                                                            + gzFile.length() / 1024 / 1024 + "M "
                                                            + "beyond the limit size:"
                                                            + conf.getInt(
                                                                    HboxConfiguration.HBOX_INPUTFORMAT_CACHESIZE_LIMIT,
                                                                    HboxConfiguration
                                                                            .DEFAULT_HBOX_INPUTFORMAT_CACHESIZE_LIMIT)
                                                            + "M.");
                                                    gzFile.delete();
                                                    LOG.info("Local cache file deleted and will not use cache.");
                                                    isCache = false;
                                                }
                                            }
                                        }
                                    } catch (EOFException e) {
                                        finished = true;
                                        e.printStackTrace();
                                    }
                                }
                                reader.close();
                                LOG.info("split " + (i + 1) + " is finished.");
                            }
                            LOG.info("Epoch " + (j + 1) + " finished.");
                            if (isCache) {
                                break;
                            }
                        }
                        osw.close();
                        gos.close();
                    } catch (Exception e) {
                        LOG.warn("Exception in thread stdinRedirectThread");
                        e.printStackTrace();
                    }
                }
            });
            stdinRedirectThread.start();
        }

        List<OutputInfo> outputs = Arrays.asList(amClient.getOutputLocation());
        if ((this.conf.getBoolean(HboxConfiguration.HBOX_OUTPUT_STREAM, HboxConfiguration.DEFAULT_HBOX_OUTPUT_STREAM)
                        || this.conf
                                .get(
                                        HboxConfiguration.HBOX_OUTPUT_STRATEGY,
                                        HboxConfiguration.DEFAULT_HBOX_OUTPUT_STRATEGY)
                                .equals("STREAM"))
                && outputs.size() > 0) {
            LOG.info("Starting thread to redirect stream stdout of hbox process");
            final Thread stdoutRedirectThread = new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        BufferedReader reader;
                        reader = new BufferedReader(new InputStreamReader(hboxProcess.getInputStream()));
                        List<OutputInfo> outputs = Arrays.asList(amClient.getOutputLocation());
                        JobConf jobConf = new JobConf(conf);
                        jobConf.setOutputKeyClass(Text.class);
                        jobConf.setOutputValueClass(Text.class);
                        jobConf.setBoolean("mapred.output.compress", true);
                        jobConf.set("mapred.output.compression.codec", "org.apache.hadoop.io.compress.GzipCodec");
                        jobConf.setOutputFormat(TextOutputFormat.class);
                        Path remotePath =
                                new Path(outputs.get(0).getDfsLocation() + "/_temporary/" + containerId.toString());
                        FileSystem dfs = remotePath.getFileSystem(jobConf);
                        // FileOutputFormat.setOutputPath(jobConf, remotePath.makeQualified(dfs));
                        jobConf.set(
                                HboxConstants.STREAM_OUTPUT_DIR,
                                remotePath.makeQualified(dfs).toString());

                        OutputFormat outputFormat = ReflectionUtils.newInstance(
                                conf.getClass(
                                        HboxConfiguration2.HBOX_OUTPUTFORMAT_CLASS,
                                        HboxConfiguration2.DEFAULT_HBOX_OUTPUTF0RMAT_CLASS,
                                        OutputFormat.class),
                                jobConf);
                        outputFormat.checkOutputSpecs(dfs, jobConf);
                        JobID jobID = new JobID(new SimpleDateFormat("yyyyMMddHHmm").format(new Date()), 0);
                        TaskAttemptID taId = new TaskAttemptID(new TaskID(jobID, true, 0), 0);
                        jobConf.set("mapred.tip.id", taId.getTaskID().toString());
                        jobConf.set("mapred.task.id", taId.toString());
                        jobConf.set("mapred.job.id", jobID.toString());
                        amClient.reportMapedTaskID(containerId, taId.toString());
                        RecordWriter writer = outputFormat.getRecordWriter(dfs, jobConf, "part-r", Reporter.NULL);
                        String hboxStreamResultLine;
                        while ((hboxStreamResultLine = reader.readLine()) != null) {
                            // LOG.info(hboxStreamResultLine);
                            writer.write(null, hboxStreamResultLine);
                            // reader.skip(1);
                        }
                        writer.close(Reporter.NULL);
                        reader.close();
                    } catch (Exception e) {
                        LOG.warn("Exception in thread stdoutRedirectThread");
                        e.printStackTrace();
                    }
                }
            });
            stdoutRedirectThread.start();
        } else {
            LOG.info("Starting thread to redirect stdout of hbox process");
            Thread stdoutRedirectThread = new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        BufferedReader reader;
                        reader = new BufferedReader(new InputStreamReader(hboxProcess.getInputStream()));
                        String hboxStdoutLog;
                        while ((hboxStdoutLog = reader.readLine()) != null) {
                            LOG.info(hboxStdoutLog);
                            if (conf.getBoolean(
                                    HboxConfiguration.HBOX_CONTAINER_RUNNING_LOG_ENABLE,
                                    HboxConfiguration.DEFAULT_HBOX_CONTAINER_RUNNING_LOG_ENABLE)) {
                                heartbeatThread.appendContainerStdOut(hboxStdoutLog);
                            }
                        }
                    } catch (Exception e) {
                        LOG.warn("Exception in thread stdoutRedirectThread");
                        e.printStackTrace();
                    }
                }
            });
            stdoutRedirectThread.start();
        }

        LOG.info("Starting thread to redirect stderr of hbox process");
        Thread stderrRedirectThread = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    BufferedReader reader;
                    reader = new BufferedReader(new InputStreamReader(hboxProcess.getErrorStream()));
                    String hboxStderrLog;
                    while ((hboxStderrLog = reader.readLine()) != null) {
                        if (conf.getBoolean(
                                        HboxConfiguration.HBOX_CONTAINER_RUNNING_LOG_ENABLE,
                                        HboxConfiguration.DEFAULT_HBOX_CONTAINER_RUNNING_LOG_ENABLE)
                                && !(hboxAppType.equals("HOROVOD")
                                        || hboxAppType.equals("MPI")
                                        || hboxAppType.equals("TENSORNET")
                                        || hboxAppType.equals("VPC")
                                        || hboxAppType.equals("DIGITS"))) {
                            heartbeatThread.appendContainerStdErr(hboxStderrLog);
                        }
                        if (hboxStderrLog.contains("reporter progress")) {
                            heartbeatThread.setProgressLog(hboxStderrLog);
                        } else if (hboxStderrLog.contains("FSDataInputStream#read error")
                                || hboxStderrLog.contains("Byte-buffer read unsupported by input stream")
                                || hboxStderrLog.contains(
                                        "org.apache.hadoop.fs.FSDataInputStream.read(FSDataInputStream.java")) {
                            continue;
                        } else {
                            LOG.info(hboxStderrLog);
                            if (hboxAppType.equals("MPI")
                                    || hboxAppType.equals("TENSORNET")
                                    || hboxAppType.equals("HOROVOD")) {
                                if (hboxStderrLog.contains("Permission denied")) {
                                    LOG.info("bind failed, now am retry.");
                                    reportFailedAndExit();
                                }
                            }
                        }
                    }
                } catch (Exception e) {
                    LOG.warn("Error in thread stderrRedirectThread");
                    e.printStackTrace();
                }
            }
        });
        stderrRedirectThread.start();

        // amClient.reportStatus(containerId, HboxContainerStatus.RUNNING);
        heartbeatThread.setContainerStatus(HboxContainerStatus.RUNNING);

        if (!(hboxAppType.equals("VPC") || hboxAppType.equals("DIGITS"))) {
            // Start Board process
            int boardIndex = this.conf.getInt(
                    HboxConfiguration.HBOX_TF_BOARD_WORKER_INDEX, HboxConfiguration.DEFAULT_HBOX_TF_BOARD_WORKER_INDEX);
            Boolean boardEnable = this.conf.getBoolean(
                    HboxConfiguration.HBOX_TF_BOARD_ENABLE, HboxConfiguration.DEFAULT_HBOX_TF_BOARD_ENABLE);
            Boolean boardPsEnable = this.conf.getBoolean(
                    HboxConfiguration.HBOX_TF_BOARD_PS_ENABLE, HboxConfiguration.DEFAULT_HBOX_TF_BOARD_PS_ENABLE);
            if ("TENSORNET".equals(hboxAppType)) {
                boardIndex += 1;
            }
            if (boardEnable) {
                if (boardPsEnable) {
                    if (this.role.equals(HboxConstants.PS) && boardIndex == this.index) {
                        Socket boardReservedSocket = new Socket();
                        try {
                            Utilities.getReservePort(
                                    boardReservedSocket,
                                    InetAddress.getByName(localHost).getHostAddress(),
                                    reservePortBegin,
                                    reservePortEnd);
                        } catch (IOException e) {
                            LOG.error("Can not get available port");
                            reportFailedAndExit();
                        }
                        String boardHost = envs.get(ApplicationConstants.Environment.NM_HOST.toString());
                        String boardLogDir = this.conf.get(
                                HboxConfiguration.HBOX_TF_BOARD_LOG_DIR,
                                HboxConfiguration.DEFAULT_HBOX_TF_BOARD_LOG_DIR);
                        int boardPort = boardReservedSocket.getLocalPort();
                        String boardCommand;
                        if ("TENSORFLOW".equals(hboxAppType) || "TENSOR2TENSOR".equals(hboxAppType)) {
                            int boardReloadInterval = this.conf.getInt(
                                    HboxConfiguration.HBOX_TF_BOARD_RELOAD_INTERVAL,
                                    HboxConfiguration.DEFAULT_HBOX_TF_BOARD_RELOAD_INTERVAL);
                            boardCommand = "tensorboard --host=" + boardHost + " --port=" + boardPort
                                    + " --reload_interval=" + boardReloadInterval + " --logdir=" + boardLogDir;
                        } else if ("TENSORNET".equals(hboxAppType)) {
                            String pythonHome = conf.get(
                                    HboxConfiguration.HBOX_PYTHON_HOME, HboxConfiguration.DEFAULT_HBOX_PYTHON_HOME);
                            int boardReloadInterval = this.conf.getInt(
                                    HboxConfiguration.HBOX_TF_BOARD_RELOAD_INTERVAL,
                                    HboxConfiguration.DEFAULT_HBOX_TF_BOARD_RELOAD_INTERVAL);
                            boardCommand = pythonHome + "python " + pythonHome + "tensorboard --host=" + boardHost
                                    + " --port=" + boardPort + " --reload_interval=" + boardReloadInterval
                                    + " --logdir=" + boardLogDir;
                        } else {
                            int boardCacheTimeout = this.conf.getInt(
                                    HboxConfiguration.HBOX_BOARD_CACHE_TIMEOUT,
                                    HboxConfiguration.DEFAULT_HBOX_BOARD_CACHE_TIMEOUT);
                            boardCommand = "visualDL --host=" + boardHost + " --port=" + boardPort + " --logdir="
                                    + boardLogDir + " --cache_timeout=" + boardCacheTimeout;
                            String modelpb = this.conf.get(
                                    HboxConfiguration.HBOX_BOARD_MODELPB, HboxConfiguration.DEFAULT_HBOX_BOARD_MODELPB);
                            if (!(modelpb.equals("") || modelpb == null)) {
                                boardCommand = boardCommand + " --model_pb=" + modelpb;
                            }
                        }
                        String boardUrl = "http://" + boardHost + ":" + boardPort;
                        LOG.info("Executing board command:" + boardCommand);
                        boardReservedSocket.close();
                        try {
                            final Process boardProcess = rt.exec(boardCommand, env);
                            LOG.info("Starting thread to redirect stdout of board process");
                            Thread boardStdoutRedirectThread = new Thread(new Runnable() {
                                @Override
                                public void run() {
                                    try {
                                        BufferedReader reader;
                                        reader = new BufferedReader(
                                                new InputStreamReader(boardProcess.getInputStream()));
                                        String boardStdoutLog;
                                        while ((boardStdoutLog = reader.readLine()) != null) {
                                            LOG.debug(boardStdoutLog);
                                        }
                                    } catch (Exception e) {
                                        LOG.warn("Exception in thread boardStdoutRedirectThread");
                                        e.printStackTrace();
                                    }
                                }
                            });
                            boardStdoutRedirectThread.start();

                            LOG.info("Starting thread to redirect stderr of board process");
                            Thread boardStderrRedirectThread = new Thread(new Runnable() {
                                @Override
                                public void run() {
                                    try {
                                        BufferedReader reader;
                                        reader = new BufferedReader(
                                                new InputStreamReader(boardProcess.getErrorStream()));
                                        String boardStderrLog;
                                        while ((boardStderrLog = reader.readLine()) != null) {
                                            LOG.debug(boardStderrLog);
                                        }
                                    } catch (Exception e) {
                                        LOG.warn("Error in thread boardStderrRedirectThread");
                                        e.printStackTrace();
                                    }
                                }
                            });
                            boardStderrRedirectThread.start();
                            amClient.reportTensorBoardURL(boardUrl);
                            LOG.info("Container index is " + index + ", report board url:" + boardUrl);
                        } catch (Exception e) {
                            LOG.error("Board Process failed. For more detail: " + e);
                        }
                    }
                } else {
                    if (this.role.equals(HboxConstants.WORKER) && boardIndex == this.index) {
                        Socket boardReservedSocket = new Socket();
                        try {
                            Utilities.getReservePort(
                                    boardReservedSocket,
                                    InetAddress.getByName(localHost).getHostAddress(),
                                    reservePortBegin,
                                    reservePortEnd);
                        } catch (IOException e) {
                            LOG.error("Can not get available port");
                            reportFailedAndExit();
                        }
                        String boardHost = envs.get(ApplicationConstants.Environment.NM_HOST.toString());
                        String boardLogDir = this.conf.get(
                                HboxConfiguration.HBOX_TF_BOARD_LOG_DIR,
                                HboxConfiguration.DEFAULT_HBOX_TF_BOARD_LOG_DIR);
                        int boardPort = boardReservedSocket.getLocalPort();
                        String boardCommand;
                        if ("TENSORFLOW".equals(hboxAppType) || "TENSOR2TENSOR".equals(hboxAppType)) {
                            int boardReloadInterval = this.conf.getInt(
                                    HboxConfiguration.HBOX_TF_BOARD_RELOAD_INTERVAL,
                                    HboxConfiguration.DEFAULT_HBOX_TF_BOARD_RELOAD_INTERVAL);
                            boardCommand = "tensorboard --host=" + boardHost + " --port=" + boardPort
                                    + " --reload_interval=" + boardReloadInterval + " --logdir=" + boardLogDir;
                        } else if ("TENSORNET".equals(hboxAppType)) {
                            String pythonHome = conf.get(
                                    HboxConfiguration.HBOX_PYTHON_HOME, HboxConfiguration.DEFAULT_HBOX_PYTHON_HOME);
                            int boardReloadInterval = this.conf.getInt(
                                    HboxConfiguration.HBOX_TF_BOARD_RELOAD_INTERVAL,
                                    HboxConfiguration.DEFAULT_HBOX_TF_BOARD_RELOAD_INTERVAL);
                            boardCommand = pythonHome + "/python " + pythonHome + "/tensorboard --host=" + boardHost
                                    + " --port=" + boardPort + " --reload_interval=" + boardReloadInterval
                                    + " --logdir=" + boardLogDir;
                        } else {
                            int boardCacheTimeout = this.conf.getInt(
                                    HboxConfiguration.HBOX_BOARD_CACHE_TIMEOUT,
                                    HboxConfiguration.DEFAULT_HBOX_BOARD_CACHE_TIMEOUT);
                            boardCommand = "visualDL --host=" + boardHost + " --port=" + boardPort + " --logdir="
                                    + boardLogDir + " --cache_timeout=" + boardCacheTimeout;
                            String modelpb = this.conf.get(
                                    HboxConfiguration.HBOX_BOARD_MODELPB, HboxConfiguration.DEFAULT_HBOX_BOARD_MODELPB);
                            if (!(modelpb.equals("") || modelpb == null)) {
                                boardCommand = boardCommand + " --model_pb=" + modelpb;
                            }
                        }
                        String boardUrl = "http://" + boardHost + ":" + boardPort;
                        LOG.info("Executing board command:" + boardCommand);
                        boardReservedSocket.close();
                        try {
                            final Process boardProcess = rt.exec(boardCommand, env);
                            LOG.info("Starting thread to redirect stdout of board process");
                            Thread boardStdoutRedirectThread = new Thread(new Runnable() {
                                @Override
                                public void run() {
                                    try {
                                        BufferedReader reader;
                                        reader = new BufferedReader(
                                                new InputStreamReader(boardProcess.getInputStream()));
                                        String boardStdoutLog;
                                        while ((boardStdoutLog = reader.readLine()) != null) {
                                            LOG.debug(boardStdoutLog);
                                        }
                                    } catch (Exception e) {
                                        LOG.warn("Exception in thread boardStdoutRedirectThread");
                                        e.printStackTrace();
                                    }
                                }
                            });
                            boardStdoutRedirectThread.start();

                            LOG.info("Starting thread to redirect stderr of board process");
                            Thread boardStderrRedirectThread = new Thread(new Runnable() {
                                @Override
                                public void run() {
                                    try {
                                        BufferedReader reader;
                                        reader = new BufferedReader(
                                                new InputStreamReader(boardProcess.getErrorStream()));
                                        String boardStderrLog;
                                        while ((boardStderrLog = reader.readLine()) != null) {
                                            LOG.debug(boardStderrLog);
                                        }
                                    } catch (Exception e) {
                                        LOG.warn("Error in thread boardStderrRedirectThread");
                                        e.printStackTrace();
                                    }
                                }
                            });
                            boardStderrRedirectThread.start();
                            amClient.reportTensorBoardURL(boardUrl);
                            LOG.info("Container index is " + index + ", report board url:" + boardUrl);
                        } catch (Exception e) {
                            LOG.error("Board Process failed. For more detail: " + e);
                        }
                    }
                }
            }
        }
        String cudaEnv = "";
        if (Long.parseLong(envs.get(HboxConstants.Environment.HBOX_CONTAIENR_GPU_NUM.toString())) > 0) {
            // get the container gpu assigned info
            String gpuUsedCommand = "curl --compressed -H Accept:application/json -XGET http://"
                    + envs.get(ApplicationConstants.Environment.NM_HOST.toString()) + ":"
                    + String.valueOf(conf.getInt(
                            HboxConfiguration.HOBX_NM_WEBAPP_PORT, HboxConfiguration.DEFAULT_HOBX_NM_WEBAPP_PORT))
                    + "/ws/v1/node/resources/yarn.io%2Fgpu";
            LOG.info("get gpu info cmd: " + gpuUsedCommand);
            final Process getGPUUsedProcess = Runtime.getRuntime().exec(gpuUsedCommand, env);
            LOG.info("Starting thread to get the gpu used info");
            try {
                StringBuilder gpuDevice = new StringBuilder();
                BufferedReader reader;
                reader = new BufferedReader(new InputStreamReader(getGPUUsedProcess.getInputStream()));
                String line = reader.readLine();
                LOG.debug("nodemanager info: " + line);
                if (line == null) {
                    LOG.error("Can't get the gpu info from nodemanager.");
                } else {
                    Gson gson = new GsonBuilder()
                            .registerTypeAdapter(
                                    new TypeToken<Map<String, Object>>() {}.getType(),
                                    new JsonDeserializer<Map<String, Object>>() {
                                        @Override
                                        public Map<String, Object> deserialize(
                                                JsonElement json, Type typeOfT, JsonDeserializationContext context)
                                                throws JsonParseException {
                                            Map<String, Object> treeMap = new HashMap<String, Object>();
                                            JsonObject jsonObject = json.getAsJsonObject();
                                            Set<Map.Entry<String, JsonElement>> entrySet = jsonObject.entrySet();
                                            for (Map.Entry<String, JsonElement> entry : entrySet) {
                                                treeMap.put(entry.getKey(), entry.getValue());
                                            }
                                            return treeMap;
                                        }
                                    })
                            .create();
                    Map<String, Object> gpuInfo =
                            gson.fromJson(line, new TypeToken<Map<String, Object>>() {}.getType());
                    if (gpuInfo.containsKey(HboxConstants.ASSIGNED_GPU_DEVICES.toString())) {
                        List<Map> assignedInfo = gson.fromJson(
                                gpuInfo.get(HboxConstants.ASSIGNED_GPU_DEVICES.toString())
                                        .toString(),
                                List.class);
                        for (Map info : assignedInfo) {
                            if (info.get("containerId").equals(containerId.toString())) {
                                gpuDevice
                                        .append(
                                                StringUtils.split(
                                                        info.get("index").toString(), ".")[0])
                                        .append(",");
                                LOG.info("container: " + containerId.toString() + " assigned gpu: "
                                        + info.get("index").toString());
                            }
                        }
                    } else {
                        LOG.info("Current nodemanager haven't assigned gpu.");
                    }
                }
                reader.close();
                getGPUUsedProcess.waitFor();
                if (gpuDevice.length() > 0) {
                    cudaEnv = gpuDevice.toString().substring(0, gpuDevice.length() - 1);
                }
            } catch (Exception e) {
                LOG.warn("Exception in thread get the gpu used info process stdoutRedirectThread");
                e.printStackTrace();
            }
            LOG.info("the gpu used info for container " + containerId + " : " + cudaEnv);
        }
        amClient.reportGPUDevice(containerId, cudaEnv);

        int updateAppStatusInterval = this.conf.getInt(
                HboxConfiguration.HBOX_CONTAINER_UPDATE_APP_STATUS_INTERVAL,
                HboxConfiguration.DEFAULT_HBOX_CONTAINER_UPDATE_APP_STATUS_INTERVAL);
        // add cpu,gpu,memory metrics
        this.hboxCmdProcessId = getPidOfProcess(hboxProcess);
        LOG.info("hboxCmdProcessId is:" + this.hboxCmdProcessId);
        containerReporter = new ContainerReporter(
                amClient, conf, containerId, cudaEnv, this.hboxCmdProcessId, containerExecType.equals("DOCKER"));
        containerReporter.setDaemon(true);
        containerReporter.start();

        if (hboxAppType.equals("MPI") || hboxAppType.equals("TENSORNET") || hboxAppType.equals("HOROVOD")) {
            int updateAppStatusRetry = this.conf.getInt(
                    HboxConfiguration.HBOX_MPI_CONTAINER_UPDATE_APP_STATUS_RETRY,
                    HboxConfiguration.DEFAULT_HBOX_MPI_CONTAINER_UPDATE_APP_STATUS_RETRY);
            while (true) {
                if (this.amClient.isApplicationCompleted()) {
                    int code = 0;
                    this.uploadOutputFiles();
                    File customExit = new File(this.mpiAppDir + "/"
                            + conf.get(HboxConfiguration.HBOX_CUSTOM_EXIT, HboxConfiguration.DEFAULT_HBOX_CUSTOM_EXIT));
                    if (customExit.exists()) {
                        try {
                            String exitCode = FileUtils.readLines(customExit).get(0);
                            code = Integer.parseInt(exitCode);
                            span.addEvent("mpi_custom_exit", Attributes.of(AttributeKey.longKey("hbox.mpi.exit"), (long)
                                    code));
                        } catch (Exception e) {
                            LOG.error("Hbox get wrong custom exit code!", e);
                            span.recordException(e);
                            code = 1; // mark as error
                        }
                    } else {
                        try {
                            code = hboxProcess.exitValue();
                            LOG.info("MPI process exit code is: " + code);
                            span.addEvent(
                                    "mpi_exit", Attributes.of(AttributeKey.longKey("hbox.mpi.exit"), (long) code));
                        } catch (final IllegalThreadStateException e) {
                            LOG.warn("MPI Process is still running while app is finished");
                            code = 0; // maybe other work fails
                            span.addEvent("mpi_still_running");
                        }
                    }
                    return code == 0;
                } else {
                    final int signalID = amClient.getSignal();
                    if (signalID >= 0) {
                        rt.exec("kill -" + signalID + " -" + this.hboxCmdProcessId);
                        LOG.info("Send the signal " + signalID + " to process " + this.hboxCmdProcessId);
                        amClient.sendSignal(-1);
                    }
                }
                Utilities.sleep(updateAppStatusInterval);
            }
        } else {
            int code = -1;
            while (code == -1 && !heartbeatThread.isHboxTrainCompleted()) {
                Utilities.sleep(updateAppStatusInterval);
                try {
                    code = hboxProcess.exitValue();
                    LOG.info("Container process exit code is: " + code);
                } catch (IllegalThreadStateException e) {
                    LOG.debug("Hbox Process is running");
                    final int signalID = amClient.getSignal();
                    if (signalID >= 0) {
                        rt.exec("kill -" + signalID + " -" + this.hboxCmdProcessId);
                        LOG.info("Send the signal " + signalID + " to process " + this.hboxCmdProcessId);
                        amClient.sendSignal(-1);
                    }
                }
            }
            // code status:
            /**
             * code=0: container is successful
             * code>0: container is failed, for example: python 137 means being kill SIGKILL
             * code=-1: container is killed by website
             */
            this.exitCode = code;
            boolean uploadWhenFailed = this.conf.getBoolean(
                    HboxConfiguration.HBOX_FAILED_UPLOAD, HboxConfiguration.DEFAULT_HBOX_FAILED_UPLOAD);
            if (uploadWhenFailed && code > 0) {
                this.uploadOutputFiles();
                return false;
            }

            if ((this.role.equals(HboxConstants.PS) || this.role.equals(HboxConstants.SCHEDULER))
                    && !this.hboxAppType.equals("DISTLIGHTLDA")) {
                if (code == -1) {
                    this.uploadOutputFiles();
                    hboxProcess.destroy();
                    return true;
                } else if (code == 0) {
                    this.uploadOutputFiles();
                    return true;
                }
                return false;
            }

            if (this.role.equals("server")) {
                if (code == -1) {
                    hboxProcess.destroy();
                    return true;
                } else if (code == 0) {
                    return true;
                }
                return false;
            }
            // As role is worker
            if (code == 0) {
                this.uploadOutputFiles();
            } else {
                return false;
            }
            return true;
        }
    }
build reLinksFiles for cacheFiles and cacheArchives for mpi app
core/src/main/java/net/qihoo/hbox/container/HboxContainer.java