Merge branch 'master' into HEAD

2016-06-23 09:57:09 +08:00 · 2016-06-23 09:57:09 +08:00 · 06e749a53c
--- a/Multiverso.sln
+++ b/Multiverso.sln
@ -18,12 +18,12 @@ Global
 		release|x64 = release|x64
 	EndGlobalSection
 	GlobalSection(ProjectConfigurationPlatforms) = postSolution
-		{546681D6-495C-4AEE-BBC2-3CAEC86B5137}.debug_zmq|x64.ActiveCfg = debug|x64
-		{546681D6-495C-4AEE-BBC2-3CAEC86B5137}.debug_zmq|x64.Build.0 = debug|x64
+		{546681D6-495C-4AEE-BBC2-3CAEC86B5137}.debug_zmq|x64.ActiveCfg = debug_zmq|x64
+		{546681D6-495C-4AEE-BBC2-3CAEC86B5137}.debug_zmq|x64.Build.0 = debug_zmq|x64
 		{546681D6-495C-4AEE-BBC2-3CAEC86B5137}.debug|x64.ActiveCfg = debug|x64
 		{546681D6-495C-4AEE-BBC2-3CAEC86B5137}.debug|x64.Build.0 = debug|x64
-		{546681D6-495C-4AEE-BBC2-3CAEC86B5137}.release_zmq|x64.ActiveCfg = release|x64
-		{546681D6-495C-4AEE-BBC2-3CAEC86B5137}.release_zmq|x64.Build.0 = release|x64
+		{546681D6-495C-4AEE-BBC2-3CAEC86B5137}.release_zmq|x64.ActiveCfg = release_zmq|x64
+		{546681D6-495C-4AEE-BBC2-3CAEC86B5137}.release_zmq|x64.Build.0 = release_zmq|x64
 		{546681D6-495C-4AEE-BBC2-3CAEC86B5137}.release|x64.ActiveCfg = release|x64
 		{546681D6-495C-4AEE-BBC2-3CAEC86B5137}.release|x64.Build.0 = release|x64
 		{16F14058-B116-49D9-8BA0-209F3AFFE849}.debug_zmq|x64.ActiveCfg = debug_zmq|x64
--- a/Test/Test.vcxproj
+++ b/Test/Test.vcxproj
@ -1,10 +1,18 @@
 <?xml version="1.0" encoding="utf-8"?>
 <Project DefaultTargets="Build" ToolsVersion="12.0" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
  <ItemGroup Label="ProjectConfigurations">
+    <ProjectConfiguration Include="debug_zmq|x64">
+      <Configuration>debug_zmq</Configuration>
+      <Platform>x64</Platform>
+    </ProjectConfiguration>
    <ProjectConfiguration Include="debug|x64">
      <Configuration>debug</Configuration>
      <Platform>x64</Platform>
    </ProjectConfiguration>
+    <ProjectConfiguration Include="release_zmq|x64">
+      <Configuration>release_zmq</Configuration>
+      <Platform>x64</Platform>
+    </ProjectConfiguration>
    <ProjectConfiguration Include="release|x64">
      <Configuration>release</Configuration>
      <Platform>x64</Platform>
@ -26,6 +34,12 @@
    <PlatformToolset>v120</PlatformToolset>
    <CharacterSet>Unicode</CharacterSet>
  </PropertyGroup>
+  <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='debug_zmq|x64'" Label="Configuration">
+    <ConfigurationType>Application</ConfigurationType>
+    <UseDebugLibraries>true</UseDebugLibraries>
+    <PlatformToolset>v120</PlatformToolset>
+    <CharacterSet>Unicode</CharacterSet>
+  </PropertyGroup>
  <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='release|x64'" Label="Configuration">
    <ConfigurationType>Application</ConfigurationType>
    <UseDebugLibraries>false</UseDebugLibraries>
@ -33,26 +47,49 @@
    <WholeProgramOptimization>true</WholeProgramOptimization>
    <CharacterSet>Unicode</CharacterSet>
  </PropertyGroup>
+  <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='release_zmq|x64'" Label="Configuration">
+    <ConfigurationType>Application</ConfigurationType>
+    <UseDebugLibraries>false</UseDebugLibraries>
+    <PlatformToolset>v120</PlatformToolset>
+    <WholeProgramOptimization>true</WholeProgramOptimization>
+    <CharacterSet>Unicode</CharacterSet>
+  </PropertyGroup>
  <Import Project="$(VCTargetsPath)\Microsoft.Cpp.props" />
  <ImportGroup Label="ExtensionSettings">
  </ImportGroup>
  <ImportGroup Condition="'$(Configuration)|$(Platform)'=='debug|x64'" Label="PropertySheets">
    <Import Project="$(UserRootDir)\Microsoft.Cpp.$(Platform).user.props" Condition="exists('$(UserRootDir)\Microsoft.Cpp.$(Platform).user.props')" Label="LocalAppDataPlatform" />
  </ImportGroup>
+  <ImportGroup Condition="'$(Configuration)|$(Platform)'=='debug_zmq|x64'" Label="PropertySheets">
+    <Import Project="$(UserRootDir)\Microsoft.Cpp.$(Platform).user.props" Condition="exists('$(UserRootDir)\Microsoft.Cpp.$(Platform).user.props')" Label="LocalAppDataPlatform" />
+  </ImportGroup>
  <ImportGroup Condition="'$(Configuration)|$(Platform)'=='release|x64'" Label="PropertySheets">
    <Import Project="$(UserRootDir)\Microsoft.Cpp.$(Platform).user.props" Condition="exists('$(UserRootDir)\Microsoft.Cpp.$(Platform).user.props')" Label="LocalAppDataPlatform" />
  </ImportGroup>
+  <ImportGroup Condition="'$(Configuration)|$(Platform)'=='release_zmq|x64'" Label="PropertySheets">
+    <Import Project="$(UserRootDir)\Microsoft.Cpp.$(Platform).user.props" Condition="exists('$(UserRootDir)\Microsoft.Cpp.$(Platform).user.props')" Label="LocalAppDataPlatform" />
+  </ImportGroup>
  <PropertyGroup Label="UserMacros" />
  <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='debug|x64'">
    <LinkIncremental>true</LinkIncremental>
    <IncludePath>$(ProjectDir);$(MSMPI_INC);$(ThirdPartyPath)\ZeroMQ 4.0.4\include;$(SolutionDir)/src;$(SolutionDir)/include;$(VC_IncludePath);$(WindowsSDK_IncludePath);D:\multiverso-next\lib</IncludePath>
    <LibraryPath>$(MSMPI_LIB64);$(SolutionDir)/x64/$(Configuration);$(VC_LibraryPath_x64);$(WindowsSDK_LibraryPath_x64);D:\multiverso-next\lib</LibraryPath>
  </PropertyGroup>
+  <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='debug_zmq|x64'">
+    <LinkIncremental>true</LinkIncremental>
+    <IncludePath>$(ProjectDir);$(MSMPI_INC);$(ThirdPartyPath)\ZeroMQ 4.0.4\include;$(SolutionDir)/src;$(SolutionDir)/include;$(VC_IncludePath);$(WindowsSDK_IncludePath);D:\multiverso-next\lib</IncludePath>
+    <LibraryPath>$(MSMPI_LIB64);$(SolutionDir)/x64/$(Configuration);$(VC_LibraryPath_x64);$(WindowsSDK_LibraryPath_x64);D:\multiverso-next\lib</LibraryPath>
+  </PropertyGroup>
  <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='release|x64'">
    <LinkIncremental>false</LinkIncremental>
    <IncludePath>$(ProjectDir);$(MSMPI_INC);$(ThirdPartyPath)\ZeroMQ 4.0.4\include;$(SolutionDir)/src;$(SolutionDir)/include;$(VC_IncludePath);$(WindowsSDK_IncludePath);</IncludePath>
    <LibraryPath>$(MSMPI_LIB64);$(SolutionDir)/x64/$(Configuration);$(VC_LibraryPath_x64);$(WindowsSDK_LibraryPath_x64);</LibraryPath>
  </PropertyGroup>
+  <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='release_zmq|x64'">
+    <LinkIncremental>false</LinkIncremental>
+    <IncludePath>$(ProjectDir);$(MSMPI_INC);$(ThirdPartyPath)\ZeroMQ 4.0.4\include;$(SolutionDir)/src;$(SolutionDir)/include;$(VC_IncludePath);$(WindowsSDK_IncludePath);</IncludePath>
+    <LibraryPath>$(MSMPI_LIB64);$(SolutionDir)/x64/$(Configuration);$(VC_LibraryPath_x64);$(WindowsSDK_LibraryPath_x64);</LibraryPath>
+  </PropertyGroup>
  <ItemDefinitionGroup Condition="'$(Configuration)|$(Platform)'=='debug|x64'">
    <ClCompile>
      <PrecompiledHeader>
@ -67,6 +104,20 @@
      <AdditionalDependencies>Multiverso.lib;kernel32.lib;user32.lib;gdi32.lib;winspool.lib;comdlg32.lib;advapi32.lib;shell32.lib;ole32.lib;oleaut32.lib;uuid.lib;odbc32.lib;odbccp32.lib;%(AdditionalDependencies)</AdditionalDependencies>
    </Link>
  </ItemDefinitionGroup>
+  <ItemDefinitionGroup Condition="'$(Configuration)|$(Platform)'=='debug_zmq|x64'">
+    <ClCompile>
+      <PrecompiledHeader>
+      </PrecompiledHeader>
+      <WarningLevel>Level3</WarningLevel>
+      <Optimization>Disabled</Optimization>
+      <PreprocessorDefinitions>WIN32;_DEBUG;_CONSOLE;_LIB;%(PreprocessorDefinitions)</PreprocessorDefinitions>
+    </ClCompile>
+    <Link>
+      <SubSystem>Console</SubSystem>
+      <GenerateDebugInformation>true</GenerateDebugInformation>
+      <AdditionalDependencies>Multiverso.lib;kernel32.lib;user32.lib;gdi32.lib;winspool.lib;comdlg32.lib;advapi32.lib;shell32.lib;ole32.lib;oleaut32.lib;uuid.lib;odbc32.lib;odbccp32.lib;%(AdditionalDependencies)</AdditionalDependencies>
+    </Link>
+  </ItemDefinitionGroup>
  <ItemDefinitionGroup Condition="'$(Configuration)|$(Platform)'=='release|x64'">
    <ClCompile>
      <WarningLevel>Level3</WarningLevel>
@ -90,6 +141,29 @@
      <UseLibraryDependencyInputs>false</UseLibraryDependencyInputs>
    </ProjectReference>
  </ItemDefinitionGroup>
+  <ItemDefinitionGroup Condition="'$(Configuration)|$(Platform)'=='release_zmq|x64'">
+    <ClCompile>
+      <WarningLevel>Level3</WarningLevel>
+      <PrecompiledHeader>
+      </PrecompiledHeader>
+      <Optimization>MaxSpeed</Optimization>
+      <FunctionLevelLinking>true</FunctionLevelLinking>
+      <IntrinsicFunctions>true</IntrinsicFunctions>
+      <PreprocessorDefinitions>WIN32;NDEBUG;_CONSOLE;_LIB;%(PreprocessorDefinitions)</PreprocessorDefinitions>
+      <OpenMPSupport>true</OpenMPSupport>
+    </ClCompile>
+    <Link>
+      <SubSystem>Console</SubSystem>
+      <GenerateDebugInformation>true</GenerateDebugInformation>
+      <EnableCOMDATFolding>true</EnableCOMDATFolding>
+      <OptimizeReferences>true</OptimizeReferences>
+      <AdditionalDependencies>Multiverso.lib;kernel32.lib;user32.lib;gdi32.lib;winspool.lib;comdlg32.lib;advapi32.lib;shell32.lib;ole32.lib;oleaut32.lib;uuid.lib;odbc32.lib;odbccp32.lib;%(AdditionalDependencies);</AdditionalDependencies>
+      <AdditionalLibraryDirectories>D:\multiverso-next\x64\release</AdditionalLibraryDirectories>
+    </Link>
+    <ProjectReference>
+      <UseLibraryDependencyInputs>false</UseLibraryDependencyInputs>
+    </ProjectReference>
+  </ItemDefinitionGroup>
  <ItemGroup>
    <ClCompile Include="gtest\gtest-all.cpp" />
    <ClCompile Include="main.cpp" />
--- a/Test/main.cpp
+++ b/Test/main.cpp
@ -89,7 +89,7 @@ void TestArray(int argc, char* argv[]) {
  multiverso::SetCMDFlag("sync", true);
  MV_Init(&argc, argv);

-  size_t array_size = 5;
+  size_t array_size = 500;

  ArrayWorker<int>* shared_array = MV_CreateTable(ArrayTableOption<int>(array_size));

@ -99,23 +99,23 @@ void TestArray(int argc, char* argv[]) {
  std::vector<int> delta(array_size);
  for (int i = 0; i < array_size; ++i)
    delta[i] = static_cast<int>(i);
+
  int* data = new int[array_size];

-  int iter = 1000000000;
-
+  int iter = 10 * (MV_Rank() + 10);
  for (int i = 0; i < iter; ++i) {
    shared_array->Add(delta.data(), array_size);
    shared_array->Get(data, array_size);
    for (int k = 0; k < array_size; ++k) {
      if (data[k] != delta[k] * (i + 1) * MV_NumWorkers()) {
-        std::cout << "i + 1 = " << i + 1 << " k = " << k << std::endl;
-        for (int j = 0; j < array_size; ++j) {
-          std::cout << data[j] << " ";
-        }
-        exit(1);
+        // std::cout << "i + 1 = " << i + 1 << " k = " << k << std::endl;
+        // for (int j = 0; j < array_size; ++j) {
+          // std::cout << data[j] << " ";
+        // }
+        // exit(1);
      }
    }
-    if (i % 1000 == 0) { printf("iter = %d\n", i); fflush(stdout); }
+    { printf("iter = %d\n", i); fflush(stdout); }
  }
  MV_ShutDown();
 }
--- a/include/multiverso/net/zmq_net.h
+++ b/include/multiverso/net/zmq_net.h
@ -11,10 +11,15 @@
 #include "multiverso/message.h"
 #include "multiverso/util/log.h"
 #include "multiverso/util/net_util.h"
+#include "multiverso/util/configure.h"

 #include <zmq.h>

 namespace multiverso {
+
+MV_DEFINE_string(machine_file, "", "path of machine file");
+MV_DEFINE_int(port, 55555 , "port used to communication");
+
 class ZMQNetWrapper : public NetInterface {
 public:
  // argc >= 2
@ -23,10 +28,10 @@ public:
  void Init(int* argc, char** argv) override {
    // get machine file 
    if (active_) return;
-    CHECK(*argc > 2);
+    // CHECK(*argc > 2);
    std::vector<std::string> machine_lists;
-    ParseMachineFile(argv[1], &machine_lists);
-    int port = atoi(argv[2]);
+    ParseMachineFile(MV_CONFIG_machine_file, &machine_lists);
+    int port = MV_CONFIG_port; // atoi(argv[2]);

    size_ = static_cast<int>(machine_lists.size());
    CHECK(size_ > 0);
--- a/src/server.cpp
+++ b/src/server.cpp
@ -74,6 +74,7 @@ public:
    int num_worker = Zoo::Get()->num_workers();
    worker_get_clocks_.reset(new VectorClock(num_worker));
    worker_add_clocks_.reset(new VectorClock(num_worker));
+    num_waited_add_.resize(num_worker, 0);
  }

  // make some modification to suit to the sync server
@ -83,18 +84,13 @@ public:
    explicit VectorClock(int n) : 
      local_clock_(n, 0), global_clock_(0), size_(0) {}

-    static bool except_max_int_compare(int a, int b) {
-      return (b == std::numeric_limits<int>::max() ? false : a < b);
-    }
-
    // Return true when all clock reach a same number
    virtual bool Update(int i) {
      ++local_clock_[i];
      if (global_clock_ < *(std::min_element(std::begin(local_clock_),
        std::end(local_clock_)))) {
        ++global_clock_;
-        if (global_clock_ == *(std::max_element(std::begin(local_clock_),
-          std::end(local_clock_), except_max_int_compare))) {
+        if (global_clock_ == max_element()) {
          return true;
        }
      }
@ -106,8 +102,7 @@ public:
      if (global_clock_ < *(std::min_element(std::begin(local_clock_),
        std::end(local_clock_)))) {
        ++global_clock_;
-        if (global_clock_ == *(std::max_element(std::begin(local_clock_),
-          std::end(local_clock_), except_max_int_compare))) {
+        if (global_clock_ == max_element()) {
          return true;
        }
      }
@ -124,6 +119,14 @@ public:
    int local_clock(int i) const { return local_clock_[i]; }
    int global_clock() const { return global_clock_; }

+  private:
+    int max_element() const {
+      int max = -1;
+      for (auto val : local_clock_) {
+        max = (val != std::numeric_limits<int>::max() && val > max) ? val : max;
+      }
+      return max;
+    }
  protected:
    std::vector<int> local_clock_;
    int global_clock_;
@ -136,6 +139,7 @@ protected:
    if (worker_get_clocks_->local_clock(worker) >
        worker_get_clocks_->global_clock()) {
      msg_add_cache_.Push(msg);
+      ++num_waited_add_[worker];
      return;
    }
    // 2. Process Add
@ -158,8 +162,7 @@ protected:
    int worker = Zoo::Get()->rank_to_worker_id(msg->src());
    if (worker_add_clocks_->local_clock(worker) >
        worker_add_clocks_->global_clock() || 
-        worker_get_clocks_->local_clock(worker) > 
-        worker_get_clocks_->global_clock()) {
+        num_waited_add_[worker] > 0) {
      // Will wait for other worker finished Add
      msg_get_cache_.Push(msg);
      return;
@ -174,13 +177,14 @@ protected:
        int add_worker = Zoo::Get()->rank_to_worker_id(add_msg->src());
        Server::ProcessAdd(add_msg);
        CHECK(!worker_add_clocks_->Update(add_worker));
+        --num_waited_add_[add_worker];
      }
    }
  }

  void ProcessFinishTrain(MessagePtr& msg) {
    int worker = Zoo::Get()->rank_to_worker_id(msg->src());
-    Log::Debug("[ProcessFinishTrain] Server %d, worker %d has finished training.\n", 
+    Log::Info("[ProcessFinishTrain] Server %d, worker %d has finished training.\n", 
               Zoo::Get()->server_rank(), worker);
    if (worker_get_clocks_->FinishTrain(worker)) {
      CHECK(msg_get_cache_.Empty());
@ -207,6 +211,7 @@ protected:
 private:
  std::unique_ptr<VectorClock> worker_get_clocks_;
  std::unique_ptr<VectorClock> worker_add_clocks_;
+  std::vector<int> num_waited_add_;

  MtQueue<MessagePtr> msg_add_cache_;
  MtQueue<MessagePtr> msg_get_cache_;
--- a/src/zoo.cpp
+++ b/src/zoo.cpp
@ -150,7 +150,7 @@ void Zoo::FinishTrain() {
    msg->set_type(MsgType::Server_Finish_Train);
    SendTo(actor::kCommunicator, msg);
  }
-  Log::Debug("rank %d finish train\n", Zoo::Get()->rank());
+  Log::Info("rank %d finish train\n", Zoo::Get()->rank());
 }