add network cpp test (#565)

tests/cpp/graph_index_test.cc

+/*!
+ *  Copyright (c) 2019 by Contributors
+ * \file msg_queue.cc
+ * \brief Message queue for DGL distributed training.
+ */
 #include <gtest/gtest.h>
 #include <dgl/graph.h>
 

tests/cpp/message_queue_test.cc

+/*!
+ *  Copyright (c) 2019 by Contributors
+ * \file msg_queue.cc
+ * \brief Message queue for DGL distributed training.
+ */
+#include <gtest/gtest.h>
+#include <string>
+
+#include "../src/graph/network/msg_queue.h"
+
+using std::string;
+using dgl::network::MessageQueue;
+
+TEST(MessageQueueTest, AddRemove) {
+  MessageQueue queue(5, 1);  // size:5, num_of_producer:1
+  char buff[10];
+  queue.Add("111", 3);
+  queue.Add("22", 2);
+  EXPECT_EQ(0, queue.Add("xxxx", 4, false));  // non-blocking add
+  queue.Remove(buff, 3);
+  EXPECT_EQ(string(buff, 3), string("111"));
+  queue.Remove(buff, 2);
+  EXPECT_EQ(string(buff, 2), string("22"));
+  queue.Add("33333", 5);
+  queue.Remove(buff, 5);
+  EXPECT_EQ(string(buff, 5), string("33333"));
+  EXPECT_EQ(0, queue.Remove(buff, 10, false));  // non-blocking remove
+  EXPECT_EQ(queue.Add("666666", 6), -1);  // exceed buffer size
+  queue.Add("55555", 5);
+  EXPECT_EQ(queue.Remove(buff, 3), -1);  // message too long
+}
+
+TEST(MessageQueueTest, EmptyAndNoMoreAdd) {
+  MessageQueue queue(5, 2);  // size:5, num_of_producer:2
+  char buff[10];
+  EXPECT_EQ(queue.EmptyAndNoMoreAdd(), false);
+  queue.Signal(1);
+  queue.Signal(1);
+  EXPECT_EQ(queue.EmptyAndNoMoreAdd(), false);
+  queue.Signal(2);
+  EXPECT_EQ(queue.EmptyAndNoMoreAdd(), true);
+  EXPECT_EQ(queue.Remove(buff, 5), 0);
+}
\ No newline at end of file

tests/cpp/tcp_socket_test.cc

+/*!
+ *  Copyright (c) 2019 by Contributors
+ * \file msg_queue.cc
+ * \brief Message queue for DGL distributed training.
+ */
+#include <gtest/gtest.h>
+#include <string.h>
+#include <unistd.h>
+#include <string>
+
+#include "../src/graph/network/tcp_socket.h"
+
+using std::string;
+using dgl::network::TCPSocket;
+
+const size_t kDataLength = 9999999; // 9.5 MB
+
+TEST(TCPSocket, SendRecieve) {
+  int pid = fork();
+  const char * msg = "0123456789";
+  ASSERT_GE(pid, 0);
+  if (pid > 0) {  // parent: server
+    TCPSocket server;
+    TCPSocket client;
+    string cl_ip;
+    int cl_port;
+    char serbuff[10];
+    memset(serbuff, '\0', 10);
+    server.SetTimeout(5 * 60 * 1000);
+
+    ASSERT_TRUE(server.Bind("127.0.0.1", 2049));
+    ASSERT_TRUE(server.Listen(3));
+    ASSERT_TRUE(server.Accept(&client, &cl_ip, &cl_port));
+
+    // Small block
+    for (int i = 0; i < 9999; ++i) {
+      int tmp;
+      int recieved_bytes = 0;
+      while (recieved_bytes < 10) {
+        int max_len = 10 - recieved_bytes;
+        tmp = client.Receive(&serbuff[recieved_bytes], max_len);
+        ASSERT_GE(tmp, 0);
+        recieved_bytes += tmp;
+      }
+      ASSERT_EQ(string("0123456789"), string(serbuff, 10));
+      int sent_bytes = 0;
+      while (sent_bytes < 10) {
+        int max_len = 10 - sent_bytes;
+        tmp = client.Send(&msg[sent_bytes], max_len);
+        ASSERT_GE(tmp, 0);
+        sent_bytes += tmp;
+      }
+    }
+
+    // Big block
+    char* bigdata = new char[kDataLength];
+    char* bigbuff = new char[kDataLength];
+    memset(bigdata, 'x', kDataLength);
+    memset(bigbuff, '\0', kDataLength);
+    int recieved_bytes = 0;
+    while (recieved_bytes < kDataLength) {
+      int max_len = kDataLength - recieved_bytes;
+      int tmp = client.Receive(&bigbuff[recieved_bytes], max_len);
+      ASSERT_GE(tmp, 0);
+      recieved_bytes += tmp;
+    }
+    for (size_t i = 0; i < kDataLength; ++i) {
+      ASSERT_EQ(bigbuff[i], 'x');
+    }
+    int sent_bytes = 0;
+    while (sent_bytes < kDataLength) {
+      int max_len = kDataLength - sent_bytes;
+      int tmp = client.Send(&bigdata[sent_bytes], max_len);
+      ASSERT_GE(tmp, 0);
+      sent_bytes += tmp;
+    }
+  } else {  // child: client
+    sleep(3);   // wait for server
+    TCPSocket client;
+    ASSERT_TRUE(client.Connect("127.0.0.1", 2049));
+    char clibuff[10];
+    memset(clibuff, '\0', 10);
+
+    // Small block
+    for (int i = 0; i < 9999; ++i) {
+      int tmp;
+      int sent_bytes = 0;
+      while (sent_bytes < 10) {
+        int max_len = 10 - sent_bytes;
+        tmp = client.Send(&msg[sent_bytes], max_len);
+        ASSERT_GE(tmp, 0);
+        sent_bytes += tmp;
+      }
+      int recieved_bytes = 0;
+      while (recieved_bytes < 10) {
+      	int max_len = 10 - recieved_bytes;
+        tmp = client.Receive(&clibuff[recieved_bytes], max_len);
+        ASSERT_GE(tmp, 0);
+        recieved_bytes += tmp;
+      }
+      ASSERT_EQ(string("0123456789"), string(clibuff, 10));
+    }
+
+    // Big block
+    char* bigdata = new char[kDataLength];
+    char* bigbuff = new char[kDataLength];
+    memset(bigdata, 'x', kDataLength);
+    memset(bigbuff, '\0', kDataLength);
+    int sent_bytes = 0;
+    while (sent_bytes < kDataLength) {
+      int max_len = kDataLength - sent_bytes;
+      int tmp = client.Send(&bigdata[sent_bytes], max_len);
+      ASSERT_GE(tmp, 0);
+      sent_bytes += tmp;
+    }
+    int recieved_bytes = 0;
+    while (recieved_bytes < kDataLength) {
+      int max_len = kDataLength - recieved_bytes;
+      int tmp = client.Receive(&bigbuff[recieved_bytes], max_len);
+      ASSERT_GE(tmp, 0);
+      recieved_bytes += tmp;
+    }
+    for (size_t i = 0; i < kDataLength; ++i) {
+      ASSERT_EQ(bigbuff[i], 'x');
+    }
+  }
+  wait(0);
+}
\ No newline at end of file