Merge branch 'develop' of https://github.com/PaddlePaddle/Paddle into…

… auto_code_gen_pr_2

cmake/external/xxhash.cmake

@@ -31,8 +31,12 @@ ENDIF()
 
 if (WIN32)
   set(XXHASH_LIBRARIES "${XXHASH_INSTALL_DIR}/lib/xxhash.lib")
+  set(XXHASH_CMAKE_C_FLAGS "${CMAKE_C_FLAGS} /wd4710 /wd4711")
+  set(XXHASH_CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} /wd4710 /wd4711")
 else()
   set(XXHASH_LIBRARIES "${XXHASH_INSTALL_DIR}/lib/libxxhash.a")
+  set(XXHASH_CMAKE_C_FLAGS ${CMAKE_C_FLAGS})
+  set(XXHASH_CMAKE_CXX_FLAGS ${CMAKE_CXX_FLAGS})
 endif ()
 
 if(WIN32)
@@ -55,6 +59,12 @@ if(WIN32)
                       -DCMAKE_GENERATOR=${CMAKE_GENERATOR}
                       -DCMAKE_GENERATOR_PLATFORM=${CMAKE_GENERATOR_PLATFORM}
                       -DBUILD_SHARED_LIBS=OFF
+                      -DCMAKE_CXX_FLAGS=${XXHASH_CMAKE_CXX_FLAGS}
+                      -DCMAKE_CXX_FLAGS_RELEASE=${CMAKE_CXX_FLAGS_RELEASE}
+                      -DCMAKE_CXX_FLAGS_DEBUG=${CMAKE_CXX_FLAGS_DEBUG}
+                      -DCMAKE_C_FLAGS=${XXHASH_CMAKE_C_FLAGS}
+                      -DCMAKE_C_FLAGS_DEBUG=${CMAKE_C_FLAGS_DEBUG}
+                      -DCMAKE_C_FLAGS_RELEASE=${CMAKE_C_FLAGS_RELEASE}
                       ${OPTIONAL_CACHE_ARGS}
       TEST_COMMAND      ""
       BUILD_BYPRODUCTS ${XXHASH_LIBRARIES}

paddle/fluid/distributed/fleet.cc

@@ -570,8 +570,6 @@ void FleetWrapper::LoadModel(const std::string& path, const int mode) {
   ret.wait();
   if (ret.get() != 0) {
     LOG(ERROR) << "load model from path:" << path << " failed";
-    sleep(sleep_seconds_before_fail_exit_);
-    exit(-1);
   }
 }
 
@@ -596,8 +594,6 @@ void FleetWrapper::SaveModel(const std::string& path, const int mode) {
   int32_t feasign_cnt = ret.get();
   if (feasign_cnt == -1) {
     LOG(ERROR) << "save model failed";
-    sleep(sleep_seconds_before_fail_exit_);
-    exit(-1);
   }
 }
 

paddle/fluid/distributed/fleet_executor/carrier.cc

@@ -48,12 +48,16 @@ bool Carrier::EnqueueInterceptorMessage(
     // handle control message
     return true;
   } else {
-    if (creating_interceptors_) {
-      // Cannot handle the message to interceptor since interceptors
-      // are still under creating. Will enqueue into a tmp stack.
-      VLOG(3) << "Receiving message while creating interceptors.";
-      message_tmp_.emplace_back(interceptor_message);
-      return true;
+    {
+      std::unique_lock<std::mutex> lock_creating(creating_flag_mutex_);
+      if (creating_interceptors_) {
+        std::unique_lock<std::mutex> lock_message(tmp_message_mutex_);
+        // Cannot handle the message to interceptor since interceptors
+        // are still under creating. Will enqueue into a tmp stack.
+        VLOG(3) << "Receiving message while creating interceptors.";
+        message_tmp_.emplace_back(interceptor_message);
+        return true;
+      }
     }
     int64_t dst_id = interceptor_message.dst_id();
     Interceptor* dst_interceptor = GetInterceptor(dst_id);
@@ -112,16 +116,24 @@ Interceptor* Carrier::SetInterceptor(int64_t interceptor_id,
 
 void Carrier::SetCreatingFlag(bool flag) {
   // set the creating flag
+  creating_flag_mutex_.lock();
   VLOG(3) << "Carrier is set the creating flag from " << creating_interceptors_
           << " to " << flag << ".";
   creating_interceptors_ = flag;
+  creating_flag_mutex_.unlock();
   if (!flag) {
     // finish create interceptors outside, handle tmp messsages
     HandleTmpMessages();
   }
 }
 
 void Carrier::HandleTmpMessages() {
+  // NOTE: It's ok lock on the tmp_message_mutex_ here, when enter this
+  // `HandleTmpMessages` method, the creating_interceptors_ flag
+  // must be false, therefore, there won't have conflict with the
+  // lock on the tmp_message_mutex_ inside `EnqueueInterceptorMessage`
+  // on the same thread.
+  std::unique_lock<std::mutex> lock(tmp_message_mutex_);
   VLOG(3) << "Carrier has received " << message_tmp_.size()
           << " messages during creating interceptors.";
   for (const auto& msg : message_tmp_) {
@@ -147,7 +159,9 @@ void Carrier::CreateInterceptors() {
     }
     // The carrier will be always waiting for outside initializer
     // since there is no interceptor has been created during auto init
+    creating_flag_mutex_.lock();
     creating_interceptors_ = false;
+    creating_flag_mutex_.unlock();
     HandleTmpMessages();
   }
 }

paddle/fluid/distributed/fleet_executor/carrier.h

@@ -15,6 +15,7 @@
 #pragma once
 
 #include <memory>
+#include <mutex>
 #include <string>
 #include <unordered_map>
 #include <vector>
@@ -78,7 +79,9 @@ class Carrier final {
       interceptor_idx_to_interceptor_;
 
   std::vector<InterceptorMessage> message_tmp_{};
+  std::mutex tmp_message_mutex_;
   bool creating_interceptors_{true};
+  std::mutex creating_flag_mutex_;
   bool is_init_{false};
 };
 

paddle/fluid/distributed/fleet_executor/message_bus.cc

@@ -51,15 +51,11 @@ void MessageBus::Init(
 #endif
 
   ListenPort();
-
-  std::call_once(once_flag_, []() {
-    std::atexit([]() { MessageBus::Instance().Release(); });
-  });
 }
 
 bool MessageBus::IsInit() const { return is_init_; }
 
-void MessageBus::Release() {
+MessageBus::~MessageBus() {
   VLOG(3) << "Message bus releases resource.";
 #if defined(PADDLE_WITH_DISTRIBUTE) && defined(PADDLE_WITH_PSCORE) && \
     !defined(PADDLE_WITH_ASCEND_CL)

paddle/fluid/distributed/fleet_executor/message_bus.h

@@ -50,11 +50,11 @@ class MessageBus final {
 
   bool IsInit() const;
 
-  void Release();
-
   // called by Interceptor, send InterceptorMessage to dst
   bool Send(const InterceptorMessage& interceptor_message);
 
+  ~MessageBus();
+
   DISABLE_COPY_AND_ASSIGN(MessageBus);
 
  private:

paddle/fluid/distributed/fleet_executor/test/CMakeLists.txt

@@ -2,3 +2,7 @@ set_source_files_properties(interceptor_ping_pong_test.cc PROPERTIES COMPILE_FLA
 set_source_files_properties(compute_interceptor_test.cc PROPERTIES COMPILE_FLAGS ${DISTRIBUTE_COMPILE_FLAGS})
 cc_test(interceptor_ping_pong_test SRCS interceptor_ping_pong_test.cc DEPS fleet_executor ${BRPC_DEPS})
 cc_test(compute_interceptor_test SRCS compute_interceptor_test.cc DEPS fleet_executor ${BRPC_DEPS})
+if(WITH_DISTRIBUTE AND WITH_PSCORE AND NOT (WITH_ASCEND OR WITH_ASCEND_CL))
+set_source_files_properties(interceptor_ping_pong_with_brpc_test.cc PROPERTIES COMPILE_FLAGS ${DISTRIBUTE_COMPILE_FLAGS})
+cc_test(interceptor_ping_pong_with_brpc_test SRCS interceptor_ping_pong_with_brpc_test.cc DEPS fleet_executor ${BRPC_DEPS})
+endif()

paddle/fluid/distributed/fleet_executor/test/interceptor_ping_pong_with_brpc_test.cc

@@ -0,0 +1,128 @@
+/* Copyright (c) 2021 PaddlePaddle Authors. All Rights Reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License. */
+
+#include <sys/socket.h>
+#include <time.h>
+#include <iostream>
+#include <unordered_map>
+
+#include "gtest/gtest.h"
+
+#include "paddle/fluid/distributed/fleet_executor/carrier.h"
+#include "paddle/fluid/distributed/fleet_executor/interceptor.h"
+#include "paddle/fluid/distributed/fleet_executor/message_bus.h"
+
+namespace paddle {
+namespace distributed {
+
+class PingPongInterceptor : public Interceptor {
+ public:
+  PingPongInterceptor(int64_t interceptor_id, TaskNode* node)
+      : Interceptor(interceptor_id, node) {
+    RegisterMsgHandle([this](const InterceptorMessage& msg) { PingPong(msg); });
+  }
+
+  void PingPong(const InterceptorMessage& msg) {
+    std::cout << GetInterceptorId() << " recv msg, count=" << count_
+              << std::endl;
+    ++count_;
+    if (count_ == 20 && GetInterceptorId() == 0) {
+      InterceptorMessage stop;
+      stop.set_message_type(STOP);
+      Send(0, stop);
+      Send(1, stop);
+      return;
+    }
+
+    InterceptorMessage resp;
+    int64_t dst = GetInterceptorId() == 0 ? 1 : 0;
+    Send(dst, resp);
+  }
+
+ private:
+  int count_{0};
+};
+
+REGISTER_INTERCEPTOR(PingPong, PingPongInterceptor);
+
+TEST(InterceptorTest, PingPong) {
+  std::cout << "Ping pong test through brpc" << std::endl;
+  unsigned int seed = time(0);
+  // random generated two ports in from 6000 to 9000
+  int port0 = 6000 + rand_r(&seed) % 3000;
+  int port1 = port0 + 1;
+
+  // using socket to check the availability of the port
+  int server_fd = -1;
+  server_fd = socket(AF_INET, SOCK_STREAM, 0);
+  int opt = 1;
+  linger ling;
+  ling.l_onoff = 1;
+  ling.l_linger = 0;
+  setsockopt(server_fd, SOL_SOCKET, SO_LINGER, &ling, sizeof(ling));
+  setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
+  struct sockaddr_in address;
+  address.sin_family = AF_INET;
+  address.sin_addr.s_addr = INADDR_ANY;
+  address.sin_port = htons(port0);
+  while (bind(server_fd, (struct sockaddr*)&address, sizeof(address)) == -1) {
+    port0++;
+    address.sin_port = htons(port0);
+  }
+  close(server_fd);
+
+  // use another socket to check another port
+  server_fd = socket(AF_INET, SOCK_STREAM, 0);
+  setsockopt(server_fd, SOL_SOCKET, SO_LINGER, &ling, sizeof(ling));
+  setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
+  port1 = port0 + 1;
+  address.sin_port = htons(port1);
+  while (bind(server_fd, (struct sockaddr*)&address, sizeof(address)) == -1) {
+    port1++;
+    address.sin_port = htons(port1);
+  }
+  close(server_fd);
+
+  std::string ip0 = "127.0.0.1:" + std::to_string(port0);
+  std::string ip1 = "127.0.0.1:" + std::to_string(port1);
+  std::cout << "ip0: " << ip0 << std::endl;
+  std::cout << "ip1: " << ip1 << std::endl;
+
+  int pid = fork();
+  if (pid == 0) {
+    MessageBus& msg_bus = MessageBus::Instance();
+    msg_bus.Init({{0, 0}, {1, 1}}, {{0, ip0}, {1, ip1}}, ip0);
+
+    Carrier& carrier = Carrier::Instance();
+
+    Interceptor* a = carrier.SetInterceptor(
+        0, InterceptorFactory::Create("PingPong", 0, nullptr));
+    carrier.SetCreatingFlag(false);
+
+    InterceptorMessage msg;
+    a->Send(1, msg);
+  } else {
+    MessageBus& msg_bus = MessageBus::Instance();
+    msg_bus.Init({{0, 0}, {1, 1}}, {{0, ip0}, {1, ip1}}, ip1);
+
+    Carrier& carrier = Carrier::Instance();
+
+    carrier.SetInterceptor(1,
+                           InterceptorFactory::Create("PingPong", 1, nullptr));
+    carrier.SetCreatingFlag(false);
+  }
+}
+
+}  // namespace distributed
+}  // namespace paddle

paddle/fluid/distributed/index_dataset/CMakeLists.txt

@@ -1,7 +1,10 @@
 proto_library(index_dataset_proto SRCS index_dataset.proto)
 cc_library(index_wrapper SRCS index_wrapper.cc DEPS index_dataset_proto fs)
-cc_library(index_sampler SRCS index_sampler.cc DEPS index_wrapper)
-
+if(WITH_MKLDNN)
+  cc_library(index_sampler SRCS index_sampler.cc DEPS xxhash index_wrapper mkldnn)
+else()
+  cc_library(index_sampler SRCS index_sampler.cc DEPS xxhash index_wrapper)
+endif()
 if(WITH_PYTHON)
   py_proto_compile(index_dataset_py_proto SRCS index_dataset.proto)
 endif()

paddle/fluid/distributed/index_dataset/index_dataset.proto

@@ -19,6 +19,7 @@ message IndexNode {
   required uint64 id = 1;
   required bool is_leaf = 2;
   required float probability = 3;
+  optional string item_name = 4;
 }
 
 message TreeMeta {
@@ -29,4 +30,4 @@ message TreeMeta {
 message KVItem {
   required bytes key = 1;
   required bytes value = 2;
-}
+}

paddle/fluid/distributed/index_dataset/index_sampler.cc

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 #include "paddle/fluid/distributed/index_dataset/index_sampler.h"
+#include "paddle/fluid/framework/data_feed.h"
 
 namespace paddle {
 namespace distributed {
@@ -69,6 +70,67 @@ std::vector<std::vector<uint64_t>> LayerWiseSampler::sample(
   }
   return outputs;
 }
+void LayerWiseSampler::sample_from_dataset(
+    const uint16_t sample_slot,
+    std::vector<paddle::framework::Record>* src_datas,
+    std::vector<paddle::framework::Record>* sample_results) {
+  sample_results->clear();
+  for (auto& data : *src_datas) {
+    VLOG(1) << "src data size = " << src_datas->size();
+    VLOG(1) << "float data size = " << data.float_feasigns_.size();
+    // data.Print();
+    uint64_t start_idx = sample_results->size();
+    VLOG(1) << "before sample, sample_results.size = " << start_idx;
+    uint64_t sample_feasign_idx = -1;
+    bool sample_sign = false;
+    for (unsigned int i = 0; i < data.uint64_feasigns_.size(); i++) {
+      VLOG(1) << "slot" << i << " = " << data.uint64_feasigns_[i].slot();
+      if (data.uint64_feasigns_[i].slot() == sample_slot) {
+        sample_sign = true;
+        sample_feasign_idx = i;
+      }
+      if (sample_sign) break;
+    }
+
+    VLOG(1) << "sample_feasign_idx: " << sample_feasign_idx;
+    if (sample_sign) {
+      auto target_id =
+          data.uint64_feasigns_[sample_feasign_idx].sign().uint64_feasign_;
+      auto travel_codes = tree_->GetTravelCodes(target_id, start_sample_layer_);
+      auto travel_path = tree_->GetNodes(travel_codes);
+      for (unsigned int j = 0; j < travel_path.size(); j++) {
+        paddle::framework::Record instance(data);
+        instance.uint64_feasigns_[sample_feasign_idx].sign().uint64_feasign_ =
+            travel_path[j].id();
+        sample_results->push_back(instance);
+        for (int idx_offset = 0; idx_offset < layer_counts_[j]; idx_offset++) {
+          int sample_res = 0;
+          do {
+            sample_res = sampler_vec_[j]->Sample();
+          } while (layer_ids_[j][sample_res].id() == travel_path[j].id());
+          paddle::framework::Record instance(data);
+          instance.uint64_feasigns_[sample_feasign_idx].sign().uint64_feasign_ =
+              layer_ids_[j][sample_res].id();
+          VLOG(1) << "layer id :" << layer_ids_[j][sample_res].id();
+          // sample_feasign_idx + 1 == label's id
+          instance.uint64_feasigns_[sample_feasign_idx + 1]
+              .sign()
+              .uint64_feasign_ = 0;
+          sample_results->push_back(instance);
+        }
+        VLOG(1) << "layer end!!!!!!!!!!!!!!!!!!";
+      }
+    }
+  }
+  VLOG(1) << "after sample, sample_results.size = " << sample_results->size();
+  return;
+}
+
+std::vector<uint64_t> float2int(std::vector<double> tmp) {
+  std::vector<uint64_t> tmp_int;
+  for (auto i : tmp) tmp_int.push_back(uint64_t(i));
+  return tmp_int;
+}
 
 }  // end namespace distributed
 }  // end namespace paddle