kvstore.h

Go to the documentation of this file.

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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.
 */

#ifndef MXNET_KVSTORE_H_
#define MXNET_KVSTORE_H_
#include <dmlc/io.h>
#include <vector>
#include <utility>
#include <unordered_map>
#include <string>
#include <functional>
#include <atomic>
#include "../../src/kvstore/gradient_compression.h"
#include "./ndarray.h"
#if MXNET_USE_DIST_KVSTORE
#include "ps/ps.h"
#endif  // MXNET_USE_DIST_KVSTORE

namespace mxnet {

enum class KVStoreServerProfilerCommand {
  kSetConfig, kState, kPause, kDump
};

class KVStore {
 public:
  virtual ~KVStore() {}

  static KVStore *Create(const char *type = "local");

  inline const std::string& type() { return type_; }

  virtual void SetGradientCompression(const std::vector<std::pair<std::string, std::string> >
                                      & kwargs) = 0;

  virtual void Init(const std::vector<int>& keys,
                    const std::vector<NDArray>& values) = 0;
  virtual void Init(const std::vector<std::string>& str_keys,
                    const std::vector<NDArray>& values) = 0;
  virtual void Push(const std::vector<int>& keys,
                    const std::vector<NDArray>& values,
                    int priority = 0)  = 0;

  virtual void Push(const std::vector<std::string>& str_keys,
                    const std::vector<NDArray>& values,
                    int priority = 0)  = 0;
  virtual void Pull(const std::vector<int>& keys,
                    const std::vector<NDArray*>& values,
                    int priority = 0, bool ignore_sparse = true) = 0;
  virtual void Pull(const std::vector<std::string>& str_keys,
                    const std::vector<NDArray*>& values,
                    int priority = 0, bool ignore_sparse = true) = 0;

  virtual void Broadcast(const std::vector<int>& vkeys,
                         const std::vector<int>& okeys,
                         const std::vector<NDArray>& values,
                         const std::vector<NDArray*>& outs,
                         int priority = 0) = 0;

  virtual void Broadcast(const std::vector<std::string>& str_vkeys,
                         const std::vector<std::string>& str_okeys,
                         const std::vector<NDArray>& values,
                         const std::vector<NDArray*>& outs,
                         int priority = 0) = 0;

  virtual void PushPull(const std::vector<int>& vkeys,
                        const std::vector<int>& okeys,
                        const std::vector<NDArray>& values,
                        const std::vector<NDArray*>& outs,
                        int priority = 0) = 0;

  virtual void PushPull(const std::vector<std::string>& str_vkeys,
                        const std::vector<std::string>& str_okeys,
                        const std::vector<NDArray>& values,
                        const std::vector<NDArray*>& outs,
                        int priority = 0) = 0;
  virtual void PullRowSparse(const std::vector<int>& str_keys,
                             const std::vector<std::pair<NDArray*, NDArray>>& val_rowids,
                             int priority = 0) = 0;

  virtual void PullRowSparse(const std::vector<std::string>& str_keys,
                             const std::vector<std::pair<NDArray*, NDArray>>& val_rowids,
                             int priority = 0) = 0;

  typedef std::function<void(int, const NDArray&, NDArray*)> Updater;
  typedef std::function<void(const std::string&, const NDArray&, NDArray*)> StrUpdater;
  virtual void set_updater(const Updater& updater) {
    CHECK(updater) << "invalid updater";
    updater_ = updater;
  }

  virtual void set_updater(const StrUpdater& updater) {
    CHECK(updater) << "invalid updater";
    str_updater_ = updater;
  }

  /******************************************************
   * the following are used for multi-machines.
   ******************************************************/

  static void InitPSEnv(const std::unordered_map<std::string, std::string>& envs) {
#if MXNET_USE_DIST_KVSTORE
    ps::Environment::Init(envs);
#else
    LOG(FATAL) << "compile with USE_DIST_KVSTORE=1 to init parameter server's environment";
#endif  // MXNET_USE_DIST_KVSTORE
  }

  static bool IsWorkerNode() {
#if MXNET_USE_DIST_KVSTORE
    const char* role_str = ps::Environment::Get()->find("DMLC_ROLE");
    return (role_str == nullptr) || (!strcmp(role_str, "worker"));
#else
    return true;
#endif  // MXNET_USE_DIST_KVSTORE
  }

  static bool IsServerNode() {
#if MXNET_USE_DIST_KVSTORE
    const char* role_str = ps::Environment::Get()->find("DMLC_ROLE");
    return (role_str != nullptr) && (!strcmp(role_str, "server"));
#else
    return false;
#endif  // MXNET_USE_DIST_KVSTORE
  }

  void set_barrier_before_exit(const bool barrier_before_exit) {
#if MXNET_USE_DIST_KVSTORE
    if (!IsWorkerNode()) LOG(FATAL) << "barrier_before_exit takes effect only on worker nodes";
    barrier_before_exit_ = barrier_before_exit;
#else
    LOG(FATAL) << "compile with USE_DIST_KVSTORE=1 to enable barrier";
#endif
  }

  static bool IsSchedulerNode() {
#if MXNET_USE_DIST_KVSTORE
    const char* role_str = ps::Environment::Get()->find("DMLC_ROLE");
    return (role_str != nullptr) && (!strcmp(role_str, "scheduler"));
#else
    return false;
#endif  // MXNET_USE_DIST_KVSTORE
  }

  virtual int get_rank() const {
    return 0;
  }

  virtual int get_group_size() const {
    return 1;
  }

  virtual int get_num_dead_node(int node_id, int timeout = 60) const {
    return 0;
  }

  virtual void Barrier() { }

  virtual void SendCommandToServers(int cmd_id, const std::string& cmd_body) { }

  virtual void SetServerProfilerCommand(const KVStoreServerProfilerCommand type,
                                        const std::string& params) {
    LOG(INFO) << "Unable to pass server the profiler command. If you are using "
              << "distributed kvstore, you need to compile with USE_DIST_KVSTORE=1."
              << "If you are training on single machine, then there is no server process"
              << "to profile. Please profile the worker process instead.";
  }

  typedef std::function<void(int, const std::string&)> Controller;

  virtual void RunServer(const Controller& controller) { }

 protected:
  Updater updater_;

  StrUpdater str_updater_;

  std::string type_;

  std::shared_ptr<kvstore::GradientCompression> gradient_compression_;

  std::atomic<bool> barrier_before_exit_{true};
};

}  // namespace mxnet
#endif  // MXNET_KVSTORE_H_