graph_node_test.cc

/* Copyright (c) 2020 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 <unistd.h>
#include <condition_variable>  // NOLINT
#include <fstream>
#include <iomanip>
#include <string>
#include <thread>  // NOLINT
#include <unordered_set>
#include <vector>
#include "google/protobuf/text_format.h"

#include "gtest/gtest.h"
#include "paddle/fluid/distributed/ps.pb.h"
#include "paddle/fluid/distributed/service/brpc_ps_client.h"
#include "paddle/fluid/distributed/service/brpc_ps_server.h"
#include "paddle/fluid/distributed/service/env.h"
#include "paddle/fluid/distributed/service/graph_brpc_client.h"
#include "paddle/fluid/distributed/service/graph_brpc_server.h"
#include "paddle/fluid/distributed/service/graph_py_service.h"
#include "paddle/fluid/distributed/service/ps_client.h"
#include "paddle/fluid/distributed/service/sendrecv.pb.h"
#include "paddle/fluid/distributed/service/service.h"
#include "paddle/fluid/distributed/table/graph/graph_node.h"
#include "paddle/fluid/framework/lod_tensor.h"
#include "paddle/fluid/framework/program_desc.h"
#include "paddle/fluid/framework/scope.h"
#include "paddle/fluid/framework/tensor_util.h"
#include "paddle/fluid/framework/variable.h"
#include "paddle/fluid/operators/math/math_function.h"
#include "paddle/fluid/platform/place.h"
#include "paddle/fluid/string/printf.h"

namespace framework = paddle::framework;
namespace platform = paddle::platform;
namespace operators = paddle::operators;
namespace math = paddle::operators::math;
namespace memory = paddle::memory;
namespace distributed = paddle::distributed;

void testSampleNodes(
    std::shared_ptr<paddle::distributed::GraphBrpcClient>& worker_ptr_) {
  std::vector<uint64_t> ids;
  auto pull_status = worker_ptr_->random_sample_nodes(0, 0, 6, ids);
  std::unordered_set<uint64_t> s;
  std::unordered_set<uint64_t> s1 = {37, 59};
  pull_status.wait();
  for (auto id : ids) s.insert(id);
  ASSERT_EQ(true, s.size() == s1.size());
  for (auto id : s) {
    ASSERT_EQ(true, s1.find(id) != s1.end());
  }
}

void testFeatureNodeSerializeInt() {
  std::string out =
      distributed::FeatureNode::parse_value_to_bytes<int32_t>({"123", "345"});
  std::vector<int32_t> out2 =
      distributed::FeatureNode::parse_bytes_to_array<int32_t>(out);
  ASSERT_EQ(out2[0], 123);
  ASSERT_EQ(out2[1], 345);
}

void testFeatureNodeSerializeInt64() {
  std::string out =
      distributed::FeatureNode::parse_value_to_bytes<int64_t>({"123", "345"});
  std::vector<int64_t> out2 =
      distributed::FeatureNode::parse_bytes_to_array<int64_t>(out);
  ASSERT_EQ(out2[0], 123);
  ASSERT_EQ(out2[1], 345);
}

void testFeatureNodeSerializeFloat32() {
  std::string out = distributed::FeatureNode::parse_value_to_bytes<float>(
      {"123.123", "345.123"});
  std::vector<float> out2 =
      distributed::FeatureNode::parse_bytes_to_array<float>(out);
  float eps;
  std::cout << "Float " << out2[0] << " " << 123.123 << std::endl;
  eps = out2[0] - 123.123;
  ASSERT_LE(eps * eps, 1e-5);
  eps = out2[1] - 345.123;
  ASSERT_LE(eps * eps, 1e-5);
}

void testFeatureNodeSerializeFloat64() {
  std::string out = distributed::FeatureNode::parse_value_to_bytes<double>(
      {"123.123", "345.123"});
  std::vector<double> out2 =
      distributed::FeatureNode::parse_bytes_to_array<double>(out);
  float eps;
  eps = out2[0] - 123.123;
  std::cout << "Float64 " << out2[0] << " " << 123.123 << std::endl;
  ASSERT_LE(eps * eps, 1e-5);
  eps = out2[1] - 345.123;
  ASSERT_LE(eps * eps, 1e-5);
}

void testSingleSampleNeighboor(
    std::shared_ptr<paddle::distributed::GraphBrpcClient>& worker_ptr_) {
  std::vector<std::vector<std::pair<uint64_t, float>>> vs;
  auto pull_status = worker_ptr_->batch_sample_neighbors(
      0, std::vector<uint64_t>(1, 37), 4, vs);
  pull_status.wait();

  std::unordered_set<uint64_t> s;
  std::unordered_set<uint64_t> s1 = {112, 45, 145};
  for (auto g : vs[0]) {
    s.insert(g.first);
  }
  ASSERT_EQ(s.size(), 3);
  for (auto g : s) {
    ASSERT_EQ(true, s1.find(g) != s1.end());
  }
  s.clear();
  s1.clear();
  vs.clear();
  pull_status = worker_ptr_->batch_sample_neighbors(
      0, std::vector<uint64_t>(1, 96), 4, vs);
  pull_status.wait();
  s1 = {111, 48, 247};
  for (auto g : vs[0]) {
    s.insert(g.first);
  }
  ASSERT_EQ(s.size(), 3);
  for (auto g : s) {
    ASSERT_EQ(true, s1.find(g) != s1.end());
  }
  vs.clear();
  pull_status = worker_ptr_->batch_sample_neighbors(0, {96, 37}, 4, vs, 0);
  pull_status.wait();
  ASSERT_EQ(vs.size(), 2);
}

void testAddNode(
    std::shared_ptr<paddle::distributed::GraphBrpcClient>& worker_ptr_) {
  worker_ptr_->clear_nodes(0);
  int total_num = 270000;
  uint64_t id;
  std::unordered_set<uint64_t> id_set;
  for (int i = 0; i < total_num; i++) {
    while (id_set.find(id = rand()) != id_set.end())
      ;
    id_set.insert(id);
  }
  std::vector<uint64_t> id_list(id_set.begin(), id_set.end());
  std::vector<bool> weight_list;
  auto status = worker_ptr_->add_graph_node(0, id_list, weight_list);
  status.wait();
  std::vector<uint64_t> ids[2];
  for (int i = 0; i < 2; i++) {
    auto sample_status =
        worker_ptr_->random_sample_nodes(0, i, total_num, ids[i]);
    sample_status.wait();
  }
  std::unordered_set<uint64_t> id_set_check(ids[0].begin(), ids[0].end());
  for (auto x : ids[1]) id_set_check.insert(x);
  ASSERT_EQ(id_set.size(), id_set_check.size());
  for (auto x : id_set) {
    ASSERT_EQ(id_set_check.find(x) != id_set_check.end(), true);
  }
  std::vector<uint64_t> remove_ids;
  for (auto p : id_set_check) {
    if (remove_ids.size() == 0)
      remove_ids.push_back(p);
    else if (remove_ids.size() < total_num / 2 && rand() % 2 == 1) {
      remove_ids.push_back(p);
    }
  }
  for (auto p : remove_ids) id_set_check.erase(p);
  status = worker_ptr_->remove_graph_node(0, remove_ids);
  status.wait();
  for (int i = 0; i < 2; i++) ids[i].clear();
  for (int i = 0; i < 2; i++) {
    auto sample_status =
        worker_ptr_->random_sample_nodes(0, i, total_num, ids[i]);
    sample_status.wait();
  }
  std::unordered_set<uint64_t> id_set_check1(ids[0].begin(), ids[0].end());
  for (auto x : ids[1]) id_set_check1.insert(x);
  ASSERT_EQ(id_set_check1.size(), id_set_check.size());
  for (auto x : id_set_check1) {
    ASSERT_EQ(id_set_check.find(x) != id_set_check.end(), true);
  }
}
void testBatchSampleNeighboor(
    std::shared_ptr<paddle::distributed::GraphBrpcClient>& worker_ptr_) {
  std::vector<std::vector<std::pair<uint64_t, float>>> vs;
  std::vector<std::uint64_t> v = {37, 96};
  auto pull_status = worker_ptr_->batch_sample_neighbors(0, v, 4, vs);
  pull_status.wait();
  std::unordered_set<uint64_t> s;
  std::unordered_set<uint64_t> s1 = {112, 45, 145};
  for (auto g : vs[0]) {
    s.insert(g.first);
  }
  ASSERT_EQ(s.size(), 3);
  for (auto g : s) {
    ASSERT_EQ(true, s1.find(g) != s1.end());
  }
  s.clear();
  s1.clear();
  s1 = {111, 48, 247};
  for (auto g : vs[1]) {
    s.insert(g.first);
  }
  ASSERT_EQ(s.size(), 3);
  for (auto g : s) {
    ASSERT_EQ(true, s1.find(g) != s1.end());
  }
}

void testCache();
void testGraphToBuffer();
// std::string nodes[] = {std::string("37\taa\t45;0.34\t145;0.31\t112;0.21"),
//                        std::string("96\tfeature\t48;1.4\t247;0.31\t111;1.21"),
//                        std::string("59\ttreat\t45;0.34\t145;0.31\t112;0.21"),
//                        std::string("97\tfood\t48;1.4\t247;0.31\t111;1.21")};

std::string edges[] = {
    std::string("37\t45\t0.34"),  std::string("37\t145\t0.31"),
    std::string("37\t112\t0.21"), std::string("96\t48\t1.4"),
    std::string("96\t247\t0.31"), std::string("96\t111\t1.21"),
    std::string("59\t45\t0.34"),  std::string("59\t145\t0.31"),
    std::string("59\t122\t0.21"), std::string("97\t48\t0.34"),
    std::string("97\t247\t0.31"), std::string("97\t111\t0.21")};
char edge_file_name[] = "edges.txt";

std::string nodes[] = {
    std::string("user\t37\ta 0.34\tb 13 14\tc hello\td abc"),
    std::string("user\t96\ta 0.31\tb 15 10\tc 96hello\td abcd"),
    std::string("user\t59\ta 0.11\tb 11 14"),
    std::string("user\t97\ta 0.11\tb 12 11"),
    std::string("item\t45\ta 0.21"),
    std::string("item\t145\ta 0.21"),
    std::string("item\t112\ta 0.21"),
    std::string("item\t48\ta 0.21"),
    std::string("item\t247\ta 0.21"),
    std::string("item\t111\ta 0.21"),
    std::string("item\t46\ta 0.21"),
    std::string("item\t146\ta 0.21"),
    std::string("item\t122\ta 0.21"),
    std::string("item\t49\ta 0.21"),
    std::string("item\t248\ta 0.21"),
    std::string("item\t113\ta 0.21")};
char node_file_name[] = "nodes.txt";

void prepare_file(char file_name[], bool load_edge) {
  std::ofstream ofile;
  ofile.open(file_name);
  if (load_edge) {
    for (auto x : edges) {
      ofile << x << std::endl;
    }
  } else {
    for (auto x : nodes) {
      ofile << x << std::endl;
    }
  }
  ofile.close();
}
void GetDownpourSparseTableProto(
    ::paddle::distributed::TableParameter* sparse_table_proto) {
  sparse_table_proto->set_table_id(0);
  sparse_table_proto->set_table_class("GraphTable");
  sparse_table_proto->set_shard_num(127);
  sparse_table_proto->set_type(::paddle::distributed::PS_SPARSE_TABLE);
  ::paddle::distributed::TableAccessorParameter* accessor_proto =
      sparse_table_proto->mutable_accessor();
  accessor_proto->set_accessor_class("CommMergeAccessor");
}

::paddle::distributed::PSParameter GetServerProto() {
  // Generate server proto desc
  ::paddle::distributed::PSParameter server_fleet_desc;
  ::paddle::distributed::ServerParameter* server_proto =
      server_fleet_desc.mutable_server_param();
  ::paddle::distributed::DownpourServerParameter* downpour_server_proto =
      server_proto->mutable_downpour_server_param();
  ::paddle::distributed::ServerServiceParameter* server_service_proto =
      downpour_server_proto->mutable_service_param();
  server_service_proto->set_service_class("GraphBrpcService");
  server_service_proto->set_server_class("GraphBrpcServer");
  server_service_proto->set_client_class("GraphBrpcClient");
  server_service_proto->set_start_server_port(0);
  server_service_proto->set_server_thread_num(12);

  ::paddle::distributed::TableParameter* sparse_table_proto =
      downpour_server_proto->add_downpour_table_param();
  GetDownpourSparseTableProto(sparse_table_proto);
  return server_fleet_desc;
}

::paddle::distributed::PSParameter GetWorkerProto() {
  ::paddle::distributed::PSParameter worker_fleet_desc;
  ::paddle::distributed::WorkerParameter* worker_proto =
      worker_fleet_desc.mutable_worker_param();

  ::paddle::distributed::DownpourWorkerParameter* downpour_worker_proto =
      worker_proto->mutable_downpour_worker_param();

  ::paddle::distributed::TableParameter* worker_sparse_table_proto =
      downpour_worker_proto->add_downpour_table_param();
  GetDownpourSparseTableProto(worker_sparse_table_proto);

  ::paddle::distributed::ServerParameter* server_proto =
      worker_fleet_desc.mutable_server_param();
  ::paddle::distributed::DownpourServerParameter* downpour_server_proto =
      server_proto->mutable_downpour_server_param();
  ::paddle::distributed::ServerServiceParameter* server_service_proto =
      downpour_server_proto->mutable_service_param();
  server_service_proto->set_service_class("GraphBrpcService");
  server_service_proto->set_server_class("GraphBrpcServer");
  server_service_proto->set_client_class("GraphBrpcClient");
  server_service_proto->set_start_server_port(0);
  server_service_proto->set_server_thread_num(12);

  ::paddle::distributed::TableParameter* server_sparse_table_proto =
      downpour_server_proto->add_downpour_table_param();
  GetDownpourSparseTableProto(server_sparse_table_proto);

  return worker_fleet_desc;
}

/*-------------------------------------------------------------------------*/

std::string ip_ = "127.0.0.1", ip2 = "127.0.0.1";
uint32_t port_ = 5209, port2 = 5210;

std::vector<std::string> host_sign_list_;

std::shared_ptr<paddle::distributed::GraphBrpcServer> pserver_ptr_,
    pserver_ptr2;

std::shared_ptr<paddle::distributed::GraphBrpcClient> worker_ptr_;

void RunServer() {
  LOG(INFO) << "init first server";
  ::paddle::distributed::PSParameter server_proto = GetServerProto();

  auto _ps_env = paddle::distributed::PaddlePSEnvironment();
  _ps_env.set_ps_servers(&host_sign_list_, 2);  // test
  pserver_ptr_ = std::shared_ptr<paddle::distributed::GraphBrpcServer>(
      (paddle::distributed::GraphBrpcServer*)
          paddle::distributed::PSServerFactory::create(server_proto));
  std::vector<framework::ProgramDesc> empty_vec;
  framework::ProgramDesc empty_prog;
  empty_vec.push_back(empty_prog);
  pserver_ptr_->configure(server_proto, _ps_env, 0, empty_vec);
  LOG(INFO) << "first server, run start(ip,port)";
  pserver_ptr_->start(ip_, port_);
  pserver_ptr_->build_peer2peer_connection(0);
  LOG(INFO) << "init first server Done";
}

void RunServer2() {
  LOG(INFO) << "init second server";
  ::paddle::distributed::PSParameter server_proto2 = GetServerProto();

  auto _ps_env2 = paddle::distributed::PaddlePSEnvironment();
  _ps_env2.set_ps_servers(&host_sign_list_, 2);  // test
  pserver_ptr2 = std::shared_ptr<paddle::distributed::GraphBrpcServer>(
      (paddle::distributed::GraphBrpcServer*)
          paddle::distributed::PSServerFactory::create(server_proto2));
  std::vector<framework::ProgramDesc> empty_vec2;
  framework::ProgramDesc empty_prog2;
  empty_vec2.push_back(empty_prog2);
  pserver_ptr2->configure(server_proto2, _ps_env2, 1, empty_vec2);
  pserver_ptr2->start(ip2, port2);
  pserver_ptr2->build_peer2peer_connection(1);
}

void RunClient(
    std::map<uint64_t, std::vector<paddle::distributed::Region>>& dense_regions,
    int index, paddle::distributed::PsBaseService* service) {
  ::paddle::distributed::PSParameter worker_proto = GetWorkerProto();
  paddle::distributed::PaddlePSEnvironment _ps_env;
  auto servers_ = host_sign_list_.size();
  _ps_env = paddle::distributed::PaddlePSEnvironment();
  _ps_env.set_ps_servers(&host_sign_list_, servers_);
  worker_ptr_ = std::shared_ptr<paddle::distributed::GraphBrpcClient>(
      (paddle::distributed::GraphBrpcClient*)
          paddle::distributed::PSClientFactory::create(worker_proto));
  worker_ptr_->configure(worker_proto, dense_regions, _ps_env, 0);
  worker_ptr_->set_shard_num(127);
  worker_ptr_->set_local_channel(index);
  worker_ptr_->set_local_graph_service(
      (paddle::distributed::GraphBrpcService*)service);
}

void RunBrpcPushSparse() {
  testCache();
  setenv("http_proxy", "", 1);
  setenv("https_proxy", "", 1);
  prepare_file(edge_file_name, 1);
  prepare_file(node_file_name, 0);
  auto ph_host = paddle::distributed::PSHost(ip_, port_, 0);
  host_sign_list_.push_back(ph_host.serialize_to_string());

  // test-start
  auto ph_host2 = paddle::distributed::PSHost(ip2, port2, 1);
  host_sign_list_.push_back(ph_host2.serialize_to_string());
  // test-end
  // Srart Server
  std::thread* server_thread = new std::thread(RunServer);
  std::thread* server_thread2 = new std::thread(RunServer2);
  sleep(1);

  std::map<uint64_t, std::vector<paddle::distributed::Region>> dense_regions;
  dense_regions.insert(
      std::pair<uint64_t, std::vector<paddle::distributed::Region>>(0, {}));
  auto regions = dense_regions[0];

  RunClient(dense_regions, 0, pserver_ptr_->get_service());

  /*-----------------------Test Server Init----------------------------------*/
  auto pull_status =
      worker_ptr_->load(0, std::string(edge_file_name), std::string("e>"));
  srand(time(0));
  pull_status.wait();
  std::vector<std::vector<std::pair<uint64_t, float>>> vs;
  testSampleNodes(worker_ptr_);
  sleep(5);
  testSingleSampleNeighboor(worker_ptr_);
  testBatchSampleNeighboor(worker_ptr_);
  pull_status = worker_ptr_->batch_sample_neighbors(
      0, std::vector<uint64_t>(1, 10240001024), 4, vs);
  pull_status.wait();
  ASSERT_EQ(0, vs[0].size());
  paddle::distributed::GraphTable* g =
      (paddle::distributed::GraphTable*)pserver_ptr_->table(0);
  size_t ttl = 6;
  g->make_neighbor_sample_cache(4, ttl);
  int round = 5;
  while (round--) {
    vs.clear();
    pull_status = worker_ptr_->batch_sample_neighbors(
        0, std::vector<uint64_t>(1, 37), 1, vs);
    pull_status.wait();

    for (int i = 0; i < ttl; i++) {
      std::vector<std::vector<std::pair<uint64_t, float>>> vs1;
      pull_status = worker_ptr_->batch_sample_neighbors(
          0, std::vector<uint64_t>(1, 37), 1, vs1);
      pull_status.wait();
      ASSERT_EQ(vs[0].size(), vs1[0].size());

      for (int j = 0; j < vs[0].size(); j++) {
        ASSERT_EQ(vs[0][j].first, vs1[0][j].first);
      }
    }
  }

  std::vector<distributed::FeatureNode> nodes;
  pull_status = worker_ptr_->pull_graph_list(0, 0, 0, 1, 1, nodes);
  pull_status.wait();
  ASSERT_EQ(nodes.size(), 1);
  ASSERT_EQ(nodes[0].get_id(), 37);
  nodes.clear();
  pull_status = worker_ptr_->pull_graph_list(0, 0, 1, 4, 1, nodes);
  pull_status.wait();
  ASSERT_EQ(nodes.size(), 1);
  ASSERT_EQ(nodes[0].get_id(), 59);
  for (auto g : nodes) {
    std::cout << g.get_id() << std::endl;
  }
  distributed::GraphPyServer server1, server2;
  distributed::GraphPyClient client1, client2;
  std::string ips_str = "127.0.0.1:5211;127.0.0.1:5212";
  std::vector<std::string> edge_types = {std::string("user2item")};
  std::vector<std::string> node_types = {std::string("user"),
                                         std::string("item")};
  VLOG(0) << "make 2 servers";
  server1.set_up(ips_str, 127, node_types, edge_types, 0);
  server2.set_up(ips_str, 127, node_types, edge_types, 1);

  server1.add_table_feat_conf("user", "a", "float32", 1);
  server1.add_table_feat_conf("user", "b", "int32", 2);
  server1.add_table_feat_conf("user", "c", "string", 1);
  server1.add_table_feat_conf("user", "d", "string", 1);
  server1.add_table_feat_conf("item", "a", "float32", 1);

  server2.add_table_feat_conf("user", "a", "float32", 1);
  server2.add_table_feat_conf("user", "b", "int32", 2);
  server2.add_table_feat_conf("user", "c", "string", 1);
  server2.add_table_feat_conf("user", "d", "string", 1);
  server2.add_table_feat_conf("item", "a", "float32", 1);

  client1.set_up(ips_str, 127, node_types, edge_types, 0);

  client1.add_table_feat_conf("user", "a", "float32", 1);
  client1.add_table_feat_conf("user", "b", "int32", 2);
  client1.add_table_feat_conf("user", "c", "string", 1);
  client1.add_table_feat_conf("user", "d", "string", 1);
  client1.add_table_feat_conf("item", "a", "float32", 1);

  client2.set_up(ips_str, 127, node_types, edge_types, 1);

  client2.add_table_feat_conf("user", "a", "float32", 1);
  client2.add_table_feat_conf("user", "b", "int32", 2);
  client2.add_table_feat_conf("user", "c", "string", 1);
  client2.add_table_feat_conf("user", "d", "string", 1);
  client2.add_table_feat_conf("item", "a", "float32", 1);

  server1.start_server(false);
  std::cout << "first server done" << std::endl;
  server2.start_server(false);
  std::cout << "second server done" << std::endl;
  client1.start_client();
  std::cout << "first client done" << std::endl;
  client2.start_client();
  std::cout << "first client done" << std::endl;
  std::cout << "started" << std::endl;
  VLOG(0) << "come to set local server";
  client1.bind_local_server(0, server1);
  VLOG(0) << "first bound";
  client2.bind_local_server(1, server2);
  VLOG(0) << "second bound";
  client1.load_node_file(std::string("user"), std::string(node_file_name));
  client1.load_node_file(std::string("item"), std::string(node_file_name));
  client1.load_edge_file(std::string("user2item"), std::string(edge_file_name),
                         0);
  nodes.clear();

  nodes = client1.pull_graph_list(std::string("user"), 0, 1, 4, 1);

  ASSERT_EQ(nodes[0].get_id(), 59);
  nodes.clear();

  // Test Pull by step

  std::unordered_set<uint64_t> count_item_nodes;
  // pull by step 2
  for (int test_step = 1; test_step < 4; test_step++) {
    count_item_nodes.clear();
    std::cout << "check pull graph list by step " << test_step << std::endl;
    for (int server_id = 0; server_id < 2; server_id++) {
      for (int start_step = 0; start_step < test_step; start_step++) {
        nodes = client1.pull_graph_list(std::string("item"), server_id,
                                        start_step, 12, test_step);
        for (auto g : nodes) {
          count_item_nodes.insert(g.get_id());
        }
        nodes.clear();
      }
    }
    ASSERT_EQ(count_item_nodes.size(), 12);
  }

  vs = client1.batch_sample_neighbors(std::string("user2item"),
                                      std::vector<uint64_t>(1, 96), 4);
  ASSERT_EQ(vs[0].size(), 3);
  std::vector<uint64_t> node_ids;
  node_ids.push_back(96);
  node_ids.push_back(37);
  vs = client1.batch_sample_neighbors(std::string("user2item"), node_ids, 4);

  ASSERT_EQ(vs.size(), 2);
  std::vector<uint64_t> nodes_ids = client2.random_sample_nodes("user", 0, 6);
  ASSERT_EQ(nodes_ids.size(), 2);
  ASSERT_EQ(true, (nodes_ids[0] == 59 && nodes_ids[1] == 37) ||
                      (nodes_ids[0] == 37 && nodes_ids[1] == 59));

  // Test get node feat
  node_ids.clear();
  node_ids.push_back(37);
  node_ids.push_back(96);
  std::vector<std::string> feature_names;
  feature_names.push_back(std::string("c"));
  feature_names.push_back(std::string("d"));
  auto node_feat =
      client1.get_node_feat(std::string("user"), node_ids, feature_names);
  ASSERT_EQ(node_feat.size(), 2);
  ASSERT_EQ(node_feat[0].size(), 2);
  VLOG(0) << "get_node_feat: " << node_feat[0][0];
  VLOG(0) << "get_node_feat: " << node_feat[0][1];
  VLOG(0) << "get_node_feat: " << node_feat[1][0];
  VLOG(0) << "get_node_feat: " << node_feat[1][1];

  node_feat[1][0] = "helloworld";

  client1.set_node_feat(std::string("user"), node_ids, feature_names,
                        node_feat);

  // sleep(5);
  node_feat =
      client1.get_node_feat(std::string("user"), node_ids, feature_names);
  VLOG(0) << "get_node_feat: " << node_feat[1][0];
  ASSERT_TRUE(node_feat[1][0] == "helloworld");

  // Test string
  node_ids.clear();
  node_ids.push_back(37);
  node_ids.push_back(96);
  // std::vector<std::string> feature_names;
  feature_names.clear();
  feature_names.push_back(std::string("a"));
  feature_names.push_back(std::string("b"));
  node_feat =
      client1.get_node_feat(std::string("user"), node_ids, feature_names);
  ASSERT_EQ(node_feat.size(), 2);
  ASSERT_EQ(node_feat[0].size(), 2);
  VLOG(0) << "get_node_feat: " << node_feat[0][0].size();
  VLOG(0) << "get_node_feat: " << node_feat[0][1].size();
  VLOG(0) << "get_node_feat: " << node_feat[1][0].size();
  VLOG(0) << "get_node_feat: " << node_feat[1][1].size();

  std::remove(edge_file_name);
  std::remove(node_file_name);
  testAddNode(worker_ptr_);
  LOG(INFO) << "Run stop_server";
  worker_ptr_->stop_server();
  LOG(INFO) << "Run finalize_worker";
  worker_ptr_->finalize_worker();
  testFeatureNodeSerializeInt();
  testFeatureNodeSerializeInt64();
  testFeatureNodeSerializeFloat32();
  testFeatureNodeSerializeFloat64();
  testGraphToBuffer();
  client1.stop_server();
}

void testCache() {
  ::paddle::distributed::ScaledLRU<::paddle::distributed::SampleKey,
                                   ::paddle::distributed::SampleResult>
      st(1, 2, 4);
  char* str = new char[7];
  strcpy(str, "54321");
  ::paddle::distributed::SampleResult* result =
      new ::paddle::distributed::SampleResult(5, str);
  ::paddle::distributed::SampleKey skey = {6, 1};
  std::vector<std::pair<::paddle::distributed::SampleKey,
                        paddle::distributed::SampleResult>>
      r;
  st.query(0, &skey, 1, r);
  ASSERT_EQ((int)r.size(), 0);

  st.insert(0, &skey, result, 1);
  for (int i = 0; i < st.get_ttl(); i++) {
    st.query(0, &skey, 1, r);
    ASSERT_EQ((int)r.size(), 1);
    char* p = (char*)r[0].second.buffer.get();
    for (int j = 0; j < r[0].second.actual_size; j++) ASSERT_EQ(p[j], str[j]);
    r.clear();
  }
  st.query(0, &skey, 1, r);
  ASSERT_EQ((int)r.size(), 0);
  str = new char[10];
  strcpy(str, "54321678");
  result = new ::paddle::distributed::SampleResult(strlen(str), str);
  st.insert(0, &skey, result, 1);
  for (int i = 0; i < st.get_ttl() / 2; i++) {
    st.query(0, &skey, 1, r);
    ASSERT_EQ((int)r.size(), 1);
    char* p = (char*)r[0].second.buffer.get();
    for (int j = 0; j < r[0].second.actual_size; j++) ASSERT_EQ(p[j], str[j]);
    r.clear();
  }
  str = new char[18];
  strcpy(str, "343332d4321");
  result = new ::paddle::distributed::SampleResult(strlen(str), str);
  st.insert(0, &skey, result, 1);
  for (int i = 0; i < st.get_ttl(); i++) {
    st.query(0, &skey, 1, r);
    ASSERT_EQ((int)r.size(), 1);
    char* p = (char*)r[0].second.buffer.get();
    for (int j = 0; j < r[0].second.actual_size; j++) ASSERT_EQ(p[j], str[j]);
    r.clear();
  }
  st.query(0, &skey, 1, r);
  ASSERT_EQ((int)r.size(), 0);
}
void testGraphToBuffer() {
  ::paddle::distributed::GraphNode s, s1;
  s.set_feature_size(1);
  s.set_feature(0, std::string("hhhh"));
  s.set_id(65);
  int size = s.get_size(true);
  char str[size];
  s.to_buffer(str, true);
  s1.recover_from_buffer(str);
  ASSERT_EQ(s.get_id(), s1.get_id());
  VLOG(0) << s.get_feature(0);
  VLOG(0) << s1.get_feature(0);
}

TEST(RunBrpcPushSparse, Run) { RunBrpcPushSparse(); }