refactor: librabit

include/rabit/internal/engine.h

@@ -0,0 +1,263 @@
+/*!
+ *  Copyright (c) 2014 by Contributors
+ * \file engine.h
+ * \brief This file defines the core interface of rabit library
+ * \author Tianqi Chen, Nacho, Tianyi
+ */
+#ifndef RABIT_INTERNAL_ENGINE_H_
+#define RABIT_INTERNAL_ENGINE_H_
+#include <string>
+#include "../serializable.h"
+
+namespace MPI {
+/*! \brief MPI data type just to be compatible with MPI reduce function*/
+class Datatype;
+}
+
+/*! \brief namespace of rabit */
+namespace rabit {
+/*! \brief core interface of the engine */
+namespace engine {
+/*! \brief interface of core Allreduce engine */
+class IEngine {
+ public:
+  /*!
+   * \brief Preprocessing function, that is called before AllReduce,
+   *        used to prepare the data used by AllReduce
+   * \param arg additional possible argument used to invoke the preprocessor
+   */
+  typedef void (PreprocFunction) (void *arg);
+  /*!
+   * \brief reduce function, the same form of MPI reduce function is used,
+   *        to be compatible with MPI interface
+   *        In all the functions, the memory is ensured to aligned to 64-bit
+   *        which means it is OK to cast src,dst to double* int* etc
+   * \param src pointer to source space
+   * \param dst pointer to destination reduction
+   * \param count total number of elements to be reduced (note this is total number of elements instead of bytes)
+   *              the definition of the reduce function should be type aware
+   * \param dtype the data type object, to be compatible with MPI reduce
+   */
+  typedef void (ReduceFunction) (const void *src,
+                                 void *dst, int count,
+                                 const MPI::Datatype &dtype);
+  /*! \brief virtual destructor */
+  virtual ~IEngine() {}
+  /*!
+   * \brief performs in-place Allreduce, on sendrecvbuf
+   *        this function is NOT thread-safe
+   * \param sendrecvbuf_ buffer for both sending and receiving data
+   * \param type_nbytes the number of bytes the type has
+   * \param count number of elements to be reduced
+   * \param reducer reduce function
+   * \param prepare_func Lazy preprocessing function, if it is not NULL, prepare_fun(prepare_arg)
+   *                     will be called by the function before performing Allreduce in order to initialize the data in sendrecvbuf.
+   *                     If the result of Allreduce can be recovered directly, then prepare_func will NOT be called
+   * \param prepare_arg argument used to pass into the lazy preprocessing function
+   */
+  virtual void Allreduce(void *sendrecvbuf_,
+                         size_t type_nbytes,
+                         size_t count,
+                         ReduceFunction reducer,
+                         PreprocFunction prepare_fun = NULL,
+                         void *prepare_arg = NULL) = 0;
+  /*!
+   * \brief broadcasts data from root to every other node
+   * \param sendrecvbuf_ buffer for both sending and receiving data
+   * \param size the size of the data to be broadcasted
+   * \param root the root worker id to broadcast the data
+   */
+  virtual void Broadcast(void *sendrecvbuf_, size_t size, int root) = 0;
+  /*!
+   * \brief explicitly re-initialize everything before calling LoadCheckPoint
+   *    call this function when IEngine throws an exception,
+   *    this function should only be used for test purposes
+   */
+  virtual void InitAfterException(void) = 0;
+  /*!
+   * \brief loads the latest check point
+   * \param global_model pointer to the globally shared model/state
+   *   when calling this function, the caller needs to guarantee that the global_model
+   *   is the same in all nodes
+   * \param local_model pointer to the local model that is specific to current node/rank
+   *   this can be NULL when no local model is needed
+   *
+   * \return the version number of the model loaded
+   *     if returned version == 0, this means no model has been CheckPointed
+   *     the p_model is not touched, users should do necessary initialization by themselves
+   *
+   *   Common usage example:
+   *      int iter = rabit::LoadCheckPoint(&model);
+   *      if (iter == 0) model.InitParameters();
+   *      for (i = iter; i < max_iter; ++i) {
+   *        do many things, include allreduce
+   *        rabit::CheckPoint(model);
+   *      }
+   *
+   * \sa CheckPoint, VersionNumber
+   */
+  virtual int LoadCheckPoint(Serializable *global_model,
+                             Serializable *local_model = NULL) = 0;
+  /*!
+   * \brief checkpoints the model, meaning a stage of execution was finished
+   *  every time we call check point, a version number increases by ones
+   *
+   * \param global_model pointer to the globally shared model/state
+   *   when calling this function, the caller needs to guarantee that the global_model
+   *   is the same in every node
+   * \param local_model pointer to the local model that is specific to current node/rank
+   *   this can be NULL when no local state is needed
+   *
+   * NOTE: local_model requires explicit replication of the model for fault-tolerance, which will
+   *       bring replication cost in CheckPoint function. global_model does not need explicit replication.
+   *       So, only CheckPoint with global_model if possible
+   *
+   * \sa LoadCheckPoint, VersionNumber
+   */
+  virtual void CheckPoint(const Serializable *global_model,
+                          const Serializable *local_model = NULL) = 0;
+  /*!
+   * \brief This function can be used to replace CheckPoint for global_model only,
+   *   when certain condition is met (see detailed explanation).
+   *
+   *   This is a "lazy" checkpoint such that only the pointer to global_model is
+   *   remembered and no memory copy is taken. To use this function, the user MUST ensure that:
+   *   The global_model must remain unchanged until the last call of Allreduce/Broadcast in the current version finishes.
+   *   In other words, global_model can be changed only between the last call of
+   *   Allreduce/Broadcast and LazyCheckPoint in the current version
+   *
+   *   For example, suppose the calling sequence is:
+   *   LazyCheckPoint, code1, Allreduce, code2, Broadcast, code3, LazyCheckPoint
+   *
+   *   If the user can only change global_model in code3, then LazyCheckPoint can be used to
+   *   improve the efficiency of the program.
+   * \param global_model pointer to the globally shared model/state
+   *   when calling this function, the caller needs to guarantee that global_model
+   *   is the same in every node
+   * \sa LoadCheckPoint, CheckPoint, VersionNumber
+   */
+  virtual void LazyCheckPoint(const Serializable *global_model) = 0;
+  /*!
+   * \return version number of the current stored model,
+   *         which means how many calls to CheckPoint we made so far
+   * \sa LoadCheckPoint, CheckPoint
+   */
+  virtual int VersionNumber(void) const = 0;
+  /*! \brief gets rank of current node */
+  virtual int GetRank(void) const = 0;
+  /*! \brief gets total number of nodes */
+  virtual int GetWorldSize(void) const = 0;
+  /*! \brief whether we run in distribted mode */
+  virtual bool IsDistributed(void) const = 0;
+  /*! \brief gets the host name of the current node */
+  virtual std::string GetHost(void) const = 0;
+  /*!
+   * \brief prints the msg in the tracker,
+   *    this function can be used to communicate progress information to
+   *    the user who monitors the tracker
+   * \param msg message to be printed in the tracker
+   */
+  virtual void TrackerPrint(const std::string &msg) = 0;
+};
+
+/*! \brief initializes the engine module */
+void Init(int argc, char *argv[]);
+/*! \brief finalizes the engine module */
+void Finalize(void);
+/*! \brief singleton method to get engine */
+IEngine *GetEngine(void);
+
+/*! \brief namespace that contains stubs to be compatible with MPI */
+namespace mpi {
+/*!\brief enum of all operators */
+enum OpType {
+  kMax = 0,
+  kMin = 1,
+  kSum = 2,
+  kBitwiseOR = 3
+};
+/*!\brief enum of supported data types */
+enum DataType {
+  kChar = 0,
+  kUChar = 1,
+  kInt = 2,
+  kUInt = 3,
+  kLong = 4,
+  kULong = 5,
+  kFloat = 6,
+  kDouble = 7,
+  kLongLong = 8,
+  kULongLong = 9
+};
+}  // namespace mpi
+/*!
+ * \brief perform in-place Allreduce, on sendrecvbuf
+ *   this is an internal function used by rabit to be able to compile with MPI
+ *   do not use this function directly
+ * \param sendrecvbuf buffer for both sending and receiving data
+ * \param type_nbytes the number of bytes the type has
+ * \param count number of elements to be reduced
+ * \param reducer reduce function
+ * \param dtype the data type
+ * \param op the reduce operator type
+ * \param prepare_func Lazy preprocessing function, lazy prepare_fun(prepare_arg)
+ *                     will be called by the function before performing Allreduce, to initialize the data in sendrecvbuf_.
+ *                     If the result of Allreduce can be recovered directly, then prepare_func will NOT be called
+ * \param prepare_arg argument used to pass into the lazy preprocessing function.
+ */
+void Allreduce_(void *sendrecvbuf,
+                size_t type_nbytes,
+                size_t count,
+                IEngine::ReduceFunction red,
+                mpi::DataType dtype,
+                mpi::OpType op,
+                IEngine::PreprocFunction prepare_fun = NULL,
+                void *prepare_arg = NULL);
+
+/*!
+ * \brief handle for customized reducer, used to handle customized reduce
+ *  this class is mainly created for compatiblity issues with MPI's customized reduce
+ */
+class ReduceHandle {
+ public:
+  // constructor
+  ReduceHandle(void);
+  // destructor
+  ~ReduceHandle(void);
+  /*!
+   * \brief initialize the reduce function,
+   *   with the type the reduce function needs to deal with
+   *   the reduce function MUST be communicative
+   */
+  void Init(IEngine::ReduceFunction redfunc, size_t type_nbytes);
+  /*!
+   * \brief customized in-place all reduce operation
+   * \param sendrecvbuf the in place send-recv buffer
+   * \param type_n4bytes size of the type, in terms of 4bytes
+   * \param count number of elements to send
+   * \param prepare_func Lazy preprocessing function, lazy prepare_fun(prepare_arg)
+   *                     will be called by the function before performing Allreduce in order to initialize the data in sendrecvbuf_.
+   *                     If the result of Allreduce can be recovered directly, then prepare_func will NOT be called
+   * \param prepare_arg argument used to pass into the lazy preprocessing function
+   */
+  void Allreduce(void *sendrecvbuf,
+                 size_t type_nbytes,
+                 size_t count,
+                 IEngine::PreprocFunction prepare_fun = NULL,
+                 void *prepare_arg = NULL);
+  /*! \return the number of bytes occupied by the type */
+  static int TypeSize(const MPI::Datatype &dtype);
+
+ protected:
+  // handle function field
+  void *handle_;
+  // reduce function of the reducer
+  IEngine::ReduceFunction *redfunc_;
+  // handle to the type field
+  void *htype_;
+  // the created type in 4 bytes
+  size_t created_type_nbytes_;
+};
+}  // namespace engine
+}  // namespace rabit
+#endif  // RABIT_INTERNAL_ENGINE_H_

include/rabit/internal/io.h

@@ -0,0 +1,106 @@
+/*!
+ *  Copyright (c) 2014 by Contributors
+ * \file io.h
+ * \brief utilities with different serializable implementations
+ * \author Tianqi Chen
+ */
+#ifndef RABIT_INTERNAL_IO_H_
+#define RABIT_INTERNAL_IO_H_
+#include <cstdio>
+#include <vector>
+#include <cstring>
+#include <string>
+#include <algorithm>
+#include "./utils.h"
+#include "../serializable.h"
+
+namespace rabit {
+namespace utils {
+/*! \brief re-use definition of dmlc::SeekStream */
+typedef dmlc::SeekStream SeekStream;
+/*! \brief fixed size memory buffer */
+struct MemoryFixSizeBuffer : public SeekStream {
+ public:
+  MemoryFixSizeBuffer(void *p_buffer, size_t buffer_size)
+      : p_buffer_(reinterpret_cast<char*>(p_buffer)),
+        buffer_size_(buffer_size) {
+    curr_ptr_ = 0;
+  }
+  virtual ~MemoryFixSizeBuffer(void) {}
+  virtual size_t Read(void *ptr, size_t size) {
+    utils::Assert(curr_ptr_ + size <= buffer_size_,
+                  "read can not have position excceed buffer length");
+    size_t nread = std::min(buffer_size_ - curr_ptr_, size);
+    if (nread != 0) std::memcpy(ptr, p_buffer_ + curr_ptr_, nread);
+    curr_ptr_ += nread;
+    return nread;
+  }
+  virtual void Write(const void *ptr, size_t size) {
+    if (size == 0) return;
+    utils::Assert(curr_ptr_ + size <=  buffer_size_,
+                  "write position exceed fixed buffer size");
+    std::memcpy(p_buffer_ + curr_ptr_, ptr, size);
+    curr_ptr_ += size;
+  }
+  virtual void Seek(size_t pos) {
+    curr_ptr_ = static_cast<size_t>(pos);
+  }
+  virtual size_t Tell(void) {
+    return curr_ptr_;
+  }
+  virtual bool AtEnd(void) const {
+    return curr_ptr_ == buffer_size_;
+  }
+
+ private:
+  /*! \brief in memory buffer */
+  char *p_buffer_;
+  /*! \brief current pointer */
+  size_t buffer_size_;
+  /*! \brief current pointer */
+  size_t curr_ptr_;
+};  // class MemoryFixSizeBuffer
+
+/*! \brief a in memory buffer that can be read and write as stream interface */
+struct MemoryBufferStream : public SeekStream {
+ public:
+  explicit MemoryBufferStream(std::string *p_buffer)
+      : p_buffer_(p_buffer) {
+    curr_ptr_ = 0;
+  }
+  virtual ~MemoryBufferStream(void) {}
+  virtual size_t Read(void *ptr, size_t size) {
+    utils::Assert(curr_ptr_ <= p_buffer_->length(),
+                  "read can not have position excceed buffer length");
+    size_t nread = std::min(p_buffer_->length() - curr_ptr_, size);
+    if (nread != 0) std::memcpy(ptr, &(*p_buffer_)[0] + curr_ptr_, nread);
+    curr_ptr_ += nread;
+    return nread;
+  }
+  virtual void Write(const void *ptr, size_t size) {
+    if (size == 0) return;
+    if (curr_ptr_ + size > p_buffer_->length()) {
+      p_buffer_->resize(curr_ptr_+size);
+    }
+    std::memcpy(&(*p_buffer_)[0] + curr_ptr_, ptr, size);
+    curr_ptr_ += size;
+  }
+  virtual void Seek(size_t pos) {
+    curr_ptr_ = static_cast<size_t>(pos);
+  }
+  virtual size_t Tell(void) {
+    return curr_ptr_;
+  }
+  virtual bool AtEnd(void) const {
+    return curr_ptr_ == p_buffer_->length();
+  }
+
+ private:
+  /*! \brief in memory buffer */
+  std::string *p_buffer_;
+  /*! \brief current pointer */
+  size_t curr_ptr_;
+};  // class MemoryBufferStream
+}  // namespace utils
+}  // namespace rabit
+#endif  // RABIT_INTERNAL_IO_H_

include/rabit/internal/rabit-inl.h

@@ -0,0 +1,328 @@
+/*!
+ * Copyright by Contributors
+ * \file rabit-inl.h
+ * \brief implementation of inline template function for rabit interface
+ *
+ * \author Tianqi Chen
+ */
+#ifndef RABIT_INTERNAL_RABIT_INL_H_
+#define RABIT_INTERNAL_RABIT_INL_H_
+// use engine for implementation
+#include <vector>
+#include <string>
+#include "./io.h"
+#include "./utils.h"
+#include "../rabit.h"
+
+namespace rabit {
+namespace engine {
+namespace mpi {
+// template function to translate type to enum indicator
+template<typename DType>
+inline DataType GetType(void);
+template<>
+inline DataType GetType<char>(void) {
+  return kChar;
+}
+template<>
+inline DataType GetType<unsigned char>(void) {
+  return kUChar;
+}
+template<>
+inline DataType GetType<int>(void) {
+  return kInt;
+}
+template<>
+inline DataType GetType<unsigned int>(void) { // NOLINT(*)
+  return kUInt;
+}
+template<>
+inline DataType GetType<long>(void) {  // NOLINT(*)
+  return kLong;
+}
+template<>
+inline DataType GetType<unsigned long>(void) { // NOLINT(*)
+  return kULong;
+}
+template<>
+inline DataType GetType<float>(void) {
+  return kFloat;
+}
+template<>
+inline DataType GetType<double>(void) {
+  return kDouble;
+}
+template<>
+inline DataType GetType<long long>(void) { // NOLINT(*)
+  return kLongLong;
+}
+template<>
+inline DataType GetType<unsigned long long>(void) { // NOLINT(*)
+  return kULongLong;
+}
+}  // namespace mpi
+}  // namespace engine
+
+namespace op {
+struct Max {
+  static const engine::mpi::OpType kType = engine::mpi::kMax;
+  template<typename DType>
+  inline static void Reduce(DType &dst, const DType &src) { // NOLINT(*)
+    if (dst < src) dst = src;
+  }
+};
+struct Min {
+  static const engine::mpi::OpType kType = engine::mpi::kMin;
+  template<typename DType>
+  inline static void Reduce(DType &dst, const DType &src) { // NOLINT(*)
+    if (dst > src) dst = src;
+  }
+};
+struct Sum {
+  static const engine::mpi::OpType kType = engine::mpi::kSum;
+  template<typename DType>
+  inline static void Reduce(DType &dst, const DType &src) { // NOLINT(*)
+    dst += src;
+  }
+};
+struct BitOR {
+  static const engine::mpi::OpType kType = engine::mpi::kBitwiseOR;
+  template<typename DType>
+  inline static void Reduce(DType &dst, const DType &src) { // NOLINT(*)
+    dst |= src;
+  }
+};
+template<typename OP, typename DType>
+inline void Reducer(const void *src_, void *dst_, int len, const MPI::Datatype &dtype) {
+  const DType *src = (const DType*)src_;
+  DType *dst = (DType*)dst_;  // NOLINT(*)
+  for (int i = 0; i < len; ++i) {
+    OP::Reduce(dst[i], src[i]);
+  }
+}
+}  // namespace op
+
+// intialize the rabit engine
+inline void Init(int argc, char *argv[]) {
+  engine::Init(argc, argv);
+}
+// finalize the rabit engine
+inline void Finalize(void) {
+  engine::Finalize();
+}
+// get the rank of current process
+inline int GetRank(void) {
+  return engine::GetEngine()->GetRank();
+}
+// the the size of the world
+inline int GetWorldSize(void) {
+  return engine::GetEngine()->GetWorldSize();
+}
+// whether rabit is distributed
+inline bool IsDistributed(void) {
+  return engine::GetEngine()->IsDistributed();
+}
+// get the name of current processor
+inline std::string GetProcessorName(void) {
+  return engine::GetEngine()->GetHost();
+}
+// broadcast data to all other nodes from root
+inline void Broadcast(void *sendrecv_data, size_t size, int root) {
+  engine::GetEngine()->Broadcast(sendrecv_data, size, root);
+}
+template<typename DType>
+inline void Broadcast(std::vector<DType> *sendrecv_data, int root) {
+  size_t size = sendrecv_data->size();
+  Broadcast(&size, sizeof(size), root);
+  if (sendrecv_data->size() != size) {
+    sendrecv_data->resize(size);
+  }
+  if (size != 0) {
+    Broadcast(&(*sendrecv_data)[0], size * sizeof(DType), root);
+  }
+}
+inline void Broadcast(std::string *sendrecv_data, int root) {
+  size_t size = sendrecv_data->length();
+  Broadcast(&size, sizeof(size), root);
+  if (sendrecv_data->length() != size) {
+    sendrecv_data->resize(size);
+  }
+  if (size != 0) {
+    Broadcast(&(*sendrecv_data)[0], size * sizeof(char), root);
+  }
+}
+
+// perform inplace Allreduce
+template<typename OP, typename DType>
+inline void Allreduce(DType *sendrecvbuf, size_t count,
+                      void (*prepare_fun)(void *arg),
+                      void *prepare_arg) {
+  engine::Allreduce_(sendrecvbuf, sizeof(DType), count, op::Reducer<OP, DType>,
+                     engine::mpi::GetType<DType>(), OP::kType, prepare_fun, prepare_arg);
+}
+
+// C++11 support for lambda prepare function
+#if DMLC_USE_CXX11
+inline void InvokeLambda_(void *fun) {
+  (*static_cast<std::function<void()>*>(fun))();
+}
+template<typename OP, typename DType>
+inline void Allreduce(DType *sendrecvbuf, size_t count, std::function<void()> prepare_fun) {
+  engine::Allreduce_(sendrecvbuf, sizeof(DType), count, op::Reducer<OP, DType>,
+                     engine::mpi::GetType<DType>(), OP::kType, InvokeLambda_, &prepare_fun);
+}
+#endif  // C++11
+
+// print message to the tracker
+inline void TrackerPrint(const std::string &msg) {
+  engine::GetEngine()->TrackerPrint(msg);
+}
+#ifndef RABIT_STRICT_CXX98_
+inline void TrackerPrintf(const char *fmt, ...) {
+  const int kPrintBuffer = 1 << 10;
+  std::string msg(kPrintBuffer, '\0');
+  va_list args;
+  va_start(args, fmt);
+  vsnprintf(&msg[0], kPrintBuffer, fmt, args);
+  va_end(args);
+  msg.resize(strlen(msg.c_str()));
+  TrackerPrint(msg);
+}
+#endif
+// load latest check point
+inline int LoadCheckPoint(Serializable *global_model,
+                          Serializable *local_model) {
+  return engine::GetEngine()->LoadCheckPoint(global_model, local_model);
+}
+// checkpoint the model, meaning we finished a stage of execution
+inline void CheckPoint(const Serializable *global_model,
+                       const Serializable *local_model) {
+  engine::GetEngine()->CheckPoint(global_model, local_model);
+}
+// lazy checkpoint the model, only remember the pointer to global_model
+inline void LazyCheckPoint(const Serializable *global_model) {
+  engine::GetEngine()->LazyCheckPoint(global_model);
+}
+// return the version number of currently stored model
+inline int VersionNumber(void) {
+  return engine::GetEngine()->VersionNumber();
+}
+// ---------------------------------
+// Code to handle customized Reduce
+// ---------------------------------
+// function to perform reduction for Reducer
+template<typename DType, void (*freduce)(DType &dst, const DType &src)>
+inline void ReducerSafe_(const void *src_, void *dst_, int len_, const MPI::Datatype &dtype) {
+  const size_t kUnit = sizeof(DType);
+  const char *psrc = reinterpret_cast<const char*>(src_);
+  char *pdst = reinterpret_cast<char*>(dst_);
+  DType tdst, tsrc;
+  for (int i = 0; i < len_; ++i) {
+    // use memcpy to avoid alignment issue
+    std::memcpy(&tdst, pdst + i * kUnit, sizeof(tdst));
+    std::memcpy(&tsrc, psrc + i * kUnit, sizeof(tsrc));
+    freduce(tdst, tsrc);
+    std::memcpy(pdst + i * kUnit, &tdst, sizeof(tdst));
+  }
+}
+// function to perform reduction for Reducer
+template<typename DType, void (*freduce)(DType &dst, const DType &src)> // NOLINT(*)
+inline void ReducerAlign_(const void *src_, void *dst_,
+                          int len_, const MPI::Datatype &dtype) {
+  const DType *psrc = reinterpret_cast<const DType*>(src_);
+  DType *pdst = reinterpret_cast<DType*>(dst_);
+  for (int i = 0; i < len_; ++i) {
+    freduce(pdst[i], psrc[i]);
+  }
+}
+template<typename DType, void (*freduce)(DType &dst, const DType &src)>  // NOLINT(*)
+inline Reducer<DType, freduce>::Reducer(void) {
+  // it is safe to directly use handle for aligned data types
+  if (sizeof(DType) == 8 || sizeof(DType) == 4 || sizeof(DType) == 1) {
+    this->handle_.Init(ReducerAlign_<DType, freduce>, sizeof(DType));
+  } else {
+    this->handle_.Init(ReducerSafe_<DType, freduce>, sizeof(DType));
+  }
+}
+template<typename DType, void (*freduce)(DType &dst, const DType &src)> // NOLINT(*)
+inline void Reducer<DType, freduce>::Allreduce(DType *sendrecvbuf, size_t count,
+                                               void (*prepare_fun)(void *arg),
+                                               void *prepare_arg) {
+  handle_.Allreduce(sendrecvbuf, sizeof(DType), count, prepare_fun, prepare_arg);
+}
+// function to perform reduction for SerializeReducer
+template<typename DType>
+inline void SerializeReducerFunc_(const void *src_, void *dst_,
+                                  int len_, const MPI::Datatype &dtype) {
+  int nbytes = engine::ReduceHandle::TypeSize(dtype);
+  // temp space
+  DType tsrc, tdst;
+  for (int i = 0; i < len_; ++i) {
+    utils::MemoryFixSizeBuffer fsrc((char*)(src_) + i * nbytes, nbytes); // NOLINT(*)
+    utils::MemoryFixSizeBuffer fdst((char*)(dst_) + i * nbytes, nbytes); // NOLINT(*)
+    tsrc.Load(fsrc);
+    tdst.Load(fdst);
+    // govern const check
+    tdst.Reduce(static_cast<const DType &>(tsrc), nbytes);
+    fdst.Seek(0);
+    tdst.Save(fdst);
+  }
+}
+template<typename DType>
+inline SerializeReducer<DType>::SerializeReducer(void) {
+  handle_.Init(SerializeReducerFunc_<DType>, sizeof(DType));
+}
+// closure to call Allreduce
+template<typename DType>
+struct SerializeReduceClosure {
+  DType *sendrecvobj;
+  size_t max_nbyte, count;
+  void (*prepare_fun)(void *arg);
+  void *prepare_arg;
+  std::string *p_buffer;
+  // invoke the closure
+  inline void Run(void) {
+    if (prepare_fun != NULL) prepare_fun(prepare_arg);
+    for (size_t i = 0; i < count; ++i) {
+      utils::MemoryFixSizeBuffer fs(BeginPtr(*p_buffer) + i * max_nbyte, max_nbyte);
+      sendrecvobj[i].Save(fs);
+    }
+  }
+  inline static void Invoke(void *c) {
+    static_cast<SerializeReduceClosure<DType>*>(c)->Run();
+  }
+};
+template<typename DType>
+inline void SerializeReducer<DType>::Allreduce(DType *sendrecvobj,
+                                               size_t max_nbyte, size_t count,
+                                               void (*prepare_fun)(void *arg),
+                                               void *prepare_arg) {
+  buffer_.resize(max_nbyte * count);
+  // setup closure
+  SerializeReduceClosure<DType> c;
+  c.sendrecvobj = sendrecvobj; c.max_nbyte = max_nbyte; c.count = count;
+  c.prepare_fun = prepare_fun; c.prepare_arg = prepare_arg; c.p_buffer = &buffer_;
+  // invoke here
+  handle_.Allreduce(BeginPtr(buffer_), max_nbyte, count,
+                    SerializeReduceClosure<DType>::Invoke, &c);
+  for (size_t i = 0; i < count; ++i) {
+    utils::MemoryFixSizeBuffer fs(BeginPtr(buffer_) + i * max_nbyte, max_nbyte);
+    sendrecvobj[i].Load(fs);
+  }
+}
+
+#if DMLC_USE_CXX11
+template<typename DType, void (*freduce)(DType &dst, const DType &src)>  // NOLINT(*)g
+inline void Reducer<DType, freduce>::Allreduce(DType *sendrecvbuf, size_t count,
+                                               std::function<void()> prepare_fun) {
+  this->Allreduce(sendrecvbuf, count, InvokeLambda_, &prepare_fun);
+}
+template<typename DType>
+inline void SerializeReducer<DType>::Allreduce(DType *sendrecvobj,
+                                               size_t max_nbytes, size_t count,
+                                               std::function<void()> prepare_fun) {
+  this->Allreduce(sendrecvobj, max_nbytes, count, InvokeLambda_, &prepare_fun);
+}
+#endif
+}  // namespace rabit
+#endif  // RABIT_INTERNAL_RABIT_INL_H_

include/rabit/internal/timer.h

@@ -0,0 +1,41 @@
+/*!
+ * Copyright by Contributors
+ * \file timer.h
+ * \brief This file defines the utils for timing
+ * \author Tianqi Chen, Nacho, Tianyi
+ */
+#ifndef RABIT_INTERNAL_TIMER_H_
+#define RABIT_INTERNAL_TIMER_H_
+#include <time.h>
+#ifdef __MACH__
+#include <mach/clock.h>
+#include <mach/mach.h>
+#endif
+#include "./utils.h"
+
+namespace rabit {
+namespace utils {
+/*!
+ * \brief return time in seconds, not cross platform, avoid to use this in most places
+ */
+inline double GetTime(void) {
+  #ifdef __MACH__
+  clock_serv_t cclock;
+  mach_timespec_t mts;
+  host_get_clock_service(mach_host_self(), CALENDAR_CLOCK, &cclock);
+  utils::Check(clock_get_time(cclock, &mts) == 0, "failed to get time");
+  mach_port_deallocate(mach_task_self(), cclock);
+  return static_cast<double>(mts.tv_sec) + static_cast<double>(mts.tv_nsec) * 1e-9;
+  #else
+  #if defined(__unix__) || defined(__linux__)
+  timespec ts;
+  utils::Check(clock_gettime(CLOCK_REALTIME, &ts) == 0, "failed to get time");
+  return static_cast<double>(ts.tv_sec) + static_cast<double>(ts.tv_nsec) * 1e-9;
+  #else
+  return static_cast<double>(time(NULL));
+  #endif
+  #endif
+}
+}  // namespace utils
+}  // namespace rabit
+#endif  // RABIT_INTERNAL_TIMER_H_

include/rabit/internal/utils.h

@@ -0,0 +1,191 @@
+/*!
+ *  Copyright (c) 2014 by Contributors
+ * \file utils.h
+ * \brief simple utils to support the code
+ * \author Tianqi Chen
+ */
+#ifndef RABIT_INTERNAL_UTILS_H_
+#define RABIT_INTERNAL_UTILS_H_
+#define _CRT_SECURE_NO_WARNINGS
+#include <cstdio>
+#include <string>
+#include <cstdlib>
+#include <vector>
+
+#ifndef RABIT_STRICT_CXX98_
+#include <cstdarg>
+#endif
+
+#if !defined(__GNUC__)
+#define fopen64 std::fopen
+#endif
+#ifdef _MSC_VER
+// NOTE: sprintf_s is not equivalent to snprintf,
+// they are equivalent when success, which is sufficient for our case
+#define snprintf sprintf_s
+#define vsnprintf vsprintf_s
+#else
+#ifdef _FILE_OFFSET_BITS
+#if _FILE_OFFSET_BITS == 32
+#pragma message("Warning: FILE OFFSET BITS defined to be 32 bit")
+#endif
+#endif
+
+#ifdef __APPLE__
+#define off64_t off_t
+#define fopen64 std::fopen
+#endif
+
+extern "C" {
+#include <sys/types.h>
+}
+#endif
+
+#ifdef _MSC_VER
+typedef unsigned char uint8_t;
+typedef unsigned __int16 uint16_t;
+typedef unsigned __int32 uint32_t;
+typedef unsigned __int64 uint64_t;
+typedef __int64 int64_t;
+#else
+#include <inttypes.h>
+#endif
+
+namespace rabit {
+/*! \brief namespace for helper utils of the project */
+namespace utils {
+
+/*! \brief error message buffer length */
+const int kPrintBuffer = 1 << 12;
+
+#ifndef RABIT_CUSTOMIZE_MSG_
+/*!
+ * \brief handling of Assert error, caused by inappropriate input
+ * \param msg error message
+ */
+inline void HandleAssertError(const char *msg) {
+  fprintf(stderr, "AssertError:%s\n", msg);
+  exit(-1);
+}
+/*!
+ * \brief handling of Check error, caused by inappropriate input
+ * \param msg error message
+ */
+inline void HandleCheckError(const char *msg) {
+  fprintf(stderr, "%s\n", msg);
+  exit(-1);
+}
+inline void HandlePrint(const char *msg) {
+  printf("%s", msg);
+}
+inline void HandleLogPrint(const char *msg) {
+  fprintf(stderr, "%s", msg);
+  fflush(stderr);
+}
+#else
+#ifndef RABIT_STRICT_CXX98_
+// include declarations, some one must implement this
+void HandleAssertError(const char *msg);
+void HandleCheckError(const char *msg);
+void HandlePrint(const char *msg);
+#endif
+#endif
+#ifdef RABIT_STRICT_CXX98_
+// these function pointers are to be assigned
+extern "C" void (*Printf)(const char *fmt, ...);
+extern "C" int (*SPrintf)(char *buf, size_t size, const char *fmt, ...);
+extern "C" void (*Assert)(int exp, const char *fmt, ...);
+extern "C" void (*Check)(int exp, const char *fmt, ...);
+extern "C" void (*Error)(const char *fmt, ...);
+#else
+/*! \brief printf, prints messages to the console */
+inline void Printf(const char *fmt, ...) {
+  std::string msg(kPrintBuffer, '\0');
+  va_list args;
+  va_start(args, fmt);
+  vsnprintf(&msg[0], kPrintBuffer, fmt, args);
+  va_end(args);
+  HandlePrint(msg.c_str());
+}
+/*! \brief portable version of snprintf */
+inline int SPrintf(char *buf, size_t size, const char *fmt, ...) {
+  va_list args;
+  va_start(args, fmt);
+  int ret = vsnprintf(buf, size, fmt, args);
+  va_end(args);
+  return ret;
+}
+
+/*! \brief assert a condition is true, use this to handle debug information */
+inline void Assert(bool exp, const char *fmt, ...) {
+  if (!exp) {
+    std::string msg(kPrintBuffer, '\0');
+    va_list args;
+    va_start(args, fmt);
+    vsnprintf(&msg[0], kPrintBuffer, fmt, args);
+    va_end(args);
+    HandleAssertError(msg.c_str());
+  }
+}
+
+/*!\brief same as assert, but this is intended to be used as a message for users */
+inline void Check(bool exp, const char *fmt, ...) {
+  if (!exp) {
+    std::string msg(kPrintBuffer, '\0');
+    va_list args;
+    va_start(args, fmt);
+    vsnprintf(&msg[0], kPrintBuffer, fmt, args);
+    va_end(args);
+    HandleCheckError(msg.c_str());
+  }
+}
+
+/*! \brief report error message, same as check */
+inline void Error(const char *fmt, ...) {
+  {
+    std::string msg(kPrintBuffer, '\0');
+    va_list args;
+    va_start(args, fmt);
+    vsnprintf(&msg[0], kPrintBuffer, fmt, args);
+    va_end(args);
+    HandleCheckError(msg.c_str());
+  }
+}
+#endif
+
+/*! \brief replace fopen, report error when the file open fails */
+inline std::FILE *FopenCheck(const char *fname, const char *flag) {
+  std::FILE *fp = fopen64(fname, flag);
+  Check(fp != NULL, "can not open file \"%s\"\n", fname);
+  return fp;
+}
+}  // namespace utils
+// easy utils that can be directly accessed in xgboost
+/*! \brief get the beginning address of a vector */
+template<typename T>
+inline T *BeginPtr(std::vector<T> &vec) {  // NOLINT(*)
+  if (vec.size() == 0) {
+    return NULL;
+  } else {
+    return &vec[0];
+  }
+}
+/*! \brief get the beginning address of a vector */
+template<typename T>
+inline const T *BeginPtr(const std::vector<T> &vec) {  // NOLINT(*)
+  if (vec.size() == 0) {
+    return NULL;
+  } else {
+    return &vec[0];
+  }
+}
+inline char* BeginPtr(std::string &str) {  // NOLINT(*)
+  if (str.length() == 0) return NULL;
+  return &str[0];
+}
+inline const char* BeginPtr(const std::string &str) {
+  if (str.length() == 0) return NULL;
+  return &str[0];
+}
+}  // namespace rabit
+#endif  // RABIT_INTERNAL_UTILS_H_