Fathom TCP level changes. TracedBuffer for keeping track of all buffers

to be traced. Adding tests for Fathom and TracedBuffer. A lot more.
Please read PR description.

4 files changed, 239 insertions, 18 deletions

diff --git a/test/core/iomgr/BUILD b/test/core/iomgr/BUILD
index fb0490a95f..a3d2ba2fc0 100644
--- a/test/core/iomgr/BUILD
+++ b/test/core/iomgr/BUILD
@@ -234,6 +234,19 @@ grpc_cc_test(
 )
 
 grpc_cc_test(
+    name = "buffer_list_test",
+    srcs = ["buffer_list_test.cc"],
+    language = "C++",
+    deps = [
+        "//:gpr",
+        "//:grpc",
+        "//test/core/util:gpr_test_util",
+        "//test/core/util:grpc_test_util",
+    ],
+)
+
+
+grpc_cc_test(
     name = "tcp_server_posix_test",
     srcs = ["tcp_server_posix_test.cc"],
     language = "C++",
diff --git a/test/core/iomgr/buffer_list_test.cc b/test/core/iomgr/buffer_list_test.cc
new file mode 100644
index 0000000000..9ffb71c85f
--- /dev/null
+++ b/test/core/iomgr/buffer_list_test.cc
@@ -0,0 +1,111 @@
+/*
+ *
+ * Copyright 2018 gRPC authors.
+ *
+ * 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 "src/core/lib/iomgr/port.h"
+
+#include "src/core/lib/iomgr/buffer_list.h"
+
+#include <grpc/grpc.h>
+
+#include "test/core/util/test_config.h"
+
+#ifdef GRPC_LINUX_ERRQUEUE
+
+static void TestShutdownFlushesListVerifier(void* arg,
+                                            grpc_core::Timestamps* ts,
+                                            grpc_error* error) {
+  GPR_ASSERT(error == GRPC_ERROR_NONE);
+  GPR_ASSERT(arg != nullptr);
+  gpr_atm* done = reinterpret_cast<gpr_atm*>(arg);
+  gpr_atm_rel_store(done, static_cast<gpr_atm>(1));
+}
+
+/** Tests that all TracedBuffer elements in the list are flushed out on
+ * shutdown.
+ * Also tests that arg is passed correctly.
+ */
+static void TestShutdownFlushesList() {
+  grpc_core::grpc_tcp_set_write_timestamps_callback(
+      TestShutdownFlushesListVerifier);
+  grpc_core::TracedBuffer* list = nullptr;
+#define NUM_ELEM 5
+  gpr_atm verifier_called[NUM_ELEM];
+  for (auto i = 0; i < NUM_ELEM; i++) {
+    gpr_atm_rel_store(&verifier_called[i], static_cast<gpr_atm>(0));
+    grpc_core::TracedBuffer::AddNewEntry(
+        &list, i, static_cast<void*>(&verifier_called[i]));
+  }
+  grpc_core::TracedBuffer::Shutdown(&list, GRPC_ERROR_NONE);
+  GPR_ASSERT(list == nullptr);
+  for (auto i = 0; i < NUM_ELEM; i++) {
+    GPR_ASSERT(gpr_atm_acq_load(&verifier_called[i]) ==
+               static_cast<gpr_atm>(1));
+  }
+}
+
+static void TestVerifierCalledOnAckVerifier(void* arg,
+                                            grpc_core::Timestamps* ts,
+                                            grpc_error* error) {
+  GPR_ASSERT(error == GRPC_ERROR_NONE);
+  GPR_ASSERT(arg != nullptr);
+  GPR_ASSERT(ts->acked_time.clock_type == GPR_CLOCK_REALTIME);
+  GPR_ASSERT(ts->acked_time.tv_sec == 123);
+  GPR_ASSERT(ts->acked_time.tv_nsec == 456);
+  gpr_atm* done = reinterpret_cast<gpr_atm*>(arg);
+  gpr_atm_rel_store(done, static_cast<gpr_atm>(1));
+}
+
+/** Tests that the timestamp verifier is called on an ACK timestamp.
+ */
+static void TestVerifierCalledOnAck() {
+  struct sock_extended_err serr;
+  serr.ee_data = 213;
+  serr.ee_info = grpc_core::SCM_TSTAMP_ACK;
+  struct grpc_core::scm_timestamping tss;
+  tss.ts[0].tv_sec = 123;
+  tss.ts[0].tv_nsec = 456;
+  grpc_core::grpc_tcp_set_write_timestamps_callback(
+      TestVerifierCalledOnAckVerifier);
+  grpc_core::TracedBuffer* list = nullptr;
+  gpr_atm verifier_called;
+  gpr_atm_rel_store(&verifier_called, static_cast<gpr_atm>(0));
+  grpc_core::TracedBuffer::AddNewEntry(&list, 213, &verifier_called);
+  grpc_core::TracedBuffer::ProcessTimestamp(&list, &serr, &tss);
+  GPR_ASSERT(gpr_atm_acq_load(&verifier_called) == static_cast<gpr_atm>(1));
+  GPR_ASSERT(list == nullptr);
+  grpc_core::TracedBuffer::Shutdown(&list, GRPC_ERROR_NONE);
+}
+
+static void TestTcpBufferList() {
+  TestVerifierCalledOnAck();
+  TestShutdownFlushesList();
+}
+
+int main(int argc, char** argv) {
+  grpc_test_init(argc, argv);
+  grpc_init();
+  TestTcpBufferList();
+  grpc_shutdown();
+  return 0;
+}
+
+#else /* GRPC_LINUX_ERRQUEUE */
+
+int main(int argc, char** argv) { return 1; }
+
+#endif /* GRPC_LINUX_ERRQUEUE */
diff --git a/test/core/iomgr/endpoint_tests.cc b/test/core/iomgr/endpoint_tests.cc
index 8db8ac5ed6..a9e8ba86c5 100644
--- a/test/core/iomgr/endpoint_tests.cc
+++ b/test/core/iomgr/endpoint_tests.cc
@@ -150,8 +150,8 @@ static void read_and_write_test_write_handler(void* data, grpc_error* error) {
                                &state->current_write_data);
       grpc_slice_buffer_reset_and_unref(&state->outgoing);
       grpc_slice_buffer_addn(&state->outgoing, slices, nslices);
-      grpc_endpoint_write(state->write_ep, &state->outgoing,
-                          &state->done_write);
+      grpc_endpoint_write(state->write_ep, &state->outgoing, &state->done_write,
+                          nullptr);
       gpr_free(slices);
       return;
     }
@@ -294,7 +294,8 @@ static void multiple_shutdown_test(grpc_endpoint_test_config config) {
   grpc_slice_buffer_add(&slice_buffer, grpc_slice_from_copied_string("a"));
   grpc_endpoint_write(f.client_ep, &slice_buffer,
                       GRPC_CLOSURE_CREATE(inc_on_failure, &fail_count,
-                                          grpc_schedule_on_exec_ctx));
+                                          grpc_schedule_on_exec_ctx),
+                      nullptr);
   wait_for_fail_count(&fail_count, 3);
   grpc_endpoint_shutdown(f.client_ep,
                          GRPC_ERROR_CREATE_FROM_STATIC_STRING("Test Shutdown"));
diff --git a/test/core/iomgr/tcp_posix_test.cc b/test/core/iomgr/tcp_posix_test.cc
index 3e87831e44..648150a765 100644
--- a/test/core/iomgr/tcp_posix_test.cc
+++ b/test/core/iomgr/tcp_posix_test.cc
@@ -36,6 +36,9 @@
 #include <grpc/support/time.h>
 
 #include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/buffer_list.h"
+#include "src/core/lib/iomgr/ev_posix.h"
+#include "src/core/lib/iomgr/sockaddr_posix.h"
 #include "src/core/lib/slice/slice_internal.h"
 #include "test/core/iomgr/endpoint_tests.h"
 #include "test/core/util/test_config.h"
@@ -68,6 +71,48 @@ static void create_sockets(int sv[2]) {
   GPR_ASSERT(fcntl(sv[1], F_SETFL, flags | O_NONBLOCK) == 0);
 }
 
+static void create_inet_sockets(int sv[2]) {
+  gpr_log(GPR_INFO, "create sockets");
+  /* Prepare listening socket */
+  struct sockaddr_in addr;
+  memset(&addr, 0, sizeof(struct sockaddr_in));
+  addr.sin_family = AF_INET;
+  int sock = socket(AF_INET, SOCK_STREAM, 0);
+  GPR_ASSERT(sock);
+  GPR_ASSERT(bind(sock, (sockaddr*)&addr, sizeof(sockaddr_in)) == 0);
+  listen(sock, 1);
+
+  /* Prepare client socket and connect to server */
+  socklen_t len = sizeof(sockaddr_in);
+  GPR_ASSERT(getsockname(sock, (sockaddr*)&addr, &len) == 0);
+
+  gpr_log(GPR_INFO, "%d\n", addr.sin_port);
+  char* addra = inet_ntoa(addr.sin_addr);
+  gpr_log(GPR_INFO, "%s\n", addra);
+
+  int client = socket(AF_INET, SOCK_STREAM, 0);
+  GPR_ASSERT(client);
+  int ret;
+  do {
+    ret = connect(client, (sockaddr*)&addr, sizeof(sockaddr_in));
+  } while (ret == -1 && errno == EINTR);
+
+  /* Accept client connection */
+  len = sizeof(socklen_t);
+  int server;
+  do {
+    server = accept(sock, (sockaddr*)&addr, (socklen_t*)&len);
+  } while (server == -1 && errno == EINTR);
+  GPR_ASSERT(server != -1);
+
+  sv[0] = server;
+  sv[1] = client;
+  int flags = fcntl(sv[0], F_GETFL, 0);
+  GPR_ASSERT(fcntl(sv[0], F_SETFL, flags | O_NONBLOCK) == 0);
+  flags = fcntl(sv[1], F_GETFL, 0);
+  GPR_ASSERT(fcntl(sv[1], F_SETFL, flags | O_NONBLOCK) == 0);
+}
+
 static ssize_t fill_socket(int fd) {
   ssize_t write_bytes;
   ssize_t total_bytes = 0;
@@ -289,6 +334,7 @@ static grpc_slice* allocate_blocks(size_t num_bytes, size_t slice_size,
 
 static void write_done(void* user_data /* write_socket_state */,
                        grpc_error* error) {
+  GPR_ASSERT(error == GRPC_ERROR_NONE);
   struct write_socket_state* state =
       static_cast<struct write_socket_state*>(user_data);
   gpr_log(GPR_INFO, "Write done callback called");
@@ -314,17 +360,22 @@ void drain_socket_blocking(int fd, size_t num_bytes, size_t read_size) {
 
   for (;;) {
     grpc_pollset_worker* worker = nullptr;
+    gpr_log(GPR_INFO, "in loop");
     gpr_mu_lock(g_mu);
+    gpr_log(GPR_INFO, "in locked polling");
     GPR_ASSERT(GRPC_LOG_IF_ERROR(
         "pollset_work",
         grpc_pollset_work(g_pollset, &worker,
                           grpc_timespec_to_millis_round_up(
                               grpc_timeout_milliseconds_to_deadline(10)))));
+    gpr_log(GPR_INFO, "done locked polling");
     gpr_mu_unlock(g_mu);
 
     do {
+      gpr_log(GPR_INFO, "doing a read");
       bytes_read =
           read(fd, buf, bytes_left > read_size ? read_size : bytes_left);
+      gpr_log(GPR_INFO, "done with read");
     } while (bytes_read < 0 && errno == EINTR);
     GPR_ASSERT(bytes_read >= 0);
     for (i = 0; i < bytes_read; ++i) {
@@ -340,10 +391,24 @@ void drain_socket_blocking(int fd, size_t num_bytes, size_t read_size) {
   gpr_free(buf);
 }
 
+/* Verifier for timestamps callback for write_test */
+void timestamps_verifier(void* arg, grpc_core::Timestamps* ts,
+                         grpc_error* error) {
+  GPR_ASSERT(error == GRPC_ERROR_NONE);
+  GPR_ASSERT(arg != nullptr);
+  GPR_ASSERT(ts->sendmsg_time.clock_type == GPR_CLOCK_REALTIME);
+  GPR_ASSERT(ts->scheduled_time.clock_type == GPR_CLOCK_REALTIME);
+  GPR_ASSERT(ts->acked_time.clock_type == GPR_CLOCK_REALTIME);
+  gpr_atm* done_timestamps = (gpr_atm*)arg;
+  gpr_atm_rel_store(done_timestamps, static_cast<gpr_atm>(1));
+}
+
 /* Write to a socket using the grpc_tcp API, then drain it directly.
    Note that if the write does not complete immediately we need to drain the
-   socket in parallel with the read. */
-static void write_test(size_t num_bytes, size_t slice_size) {
+   socket in parallel with the read. If collect_timestamps is true, it will
+   try to get timestamps for the write. */
+static void write_test(size_t num_bytes, size_t slice_size,
+                       bool collect_timestamps) {
   int sv[2];
   grpc_endpoint* ep;
   struct write_socket_state state;
@@ -356,19 +421,27 @@ static void write_test(size_t num_bytes, size_t slice_size) {
       grpc_timespec_to_millis_round_up(grpc_timeout_seconds_to_deadline(20));
   grpc_core::ExecCtx exec_ctx;
 
+  if (collect_timestamps && !grpc_event_engine_can_track_errors()) {
+    return;
+  }
+
   gpr_log(GPR_INFO,
           "Start write test with %" PRIuPTR " bytes, slice size %" PRIuPTR,
           num_bytes, slice_size);
 
-  create_sockets(sv);
+  if (collect_timestamps) {
+    create_inet_sockets(sv);
+  } else {
+    create_sockets(sv);
+  }
 
   grpc_arg a[1];
   a[0].key = const_cast<char*>(GRPC_ARG_TCP_READ_CHUNK_SIZE);
   a[0].type = GRPC_ARG_INTEGER,
   a[0].value.integer = static_cast<int>(slice_size);
   grpc_channel_args args = {GPR_ARRAY_SIZE(a), a};
-  ep = grpc_tcp_create(grpc_fd_create(sv[1], "write_test", false), &args,
-                       "test");
+  ep = grpc_tcp_create(grpc_fd_create(sv[1], "write_test", collect_timestamps),
+                       &args, "test");
   grpc_endpoint_add_to_pollset(ep, g_pollset);
 
   state.ep = ep;
@@ -381,18 +454,28 @@ static void write_test(size_t num_bytes, size_t slice_size) {
   GRPC_CLOSURE_INIT(&write_done_closure, write_done, &state,
                     grpc_schedule_on_exec_ctx);
 
-  grpc_endpoint_write(ep, &outgoing, &write_done_closure);
+  gpr_atm done_timestamps;
+  gpr_atm_rel_store(&done_timestamps, static_cast<gpr_atm>(0));
+  grpc_endpoint_write(ep, &outgoing, &write_done_closure,
+                      grpc_event_engine_can_track_errors() && collect_timestamps
+                          ? (void*)&done_timestamps
+                          : nullptr);
+  gpr_log(GPR_INFO, "about to drain");
   drain_socket_blocking(sv[0], num_bytes, num_bytes);
+  gpr_log(GPR_INFO, "done drain");
+  exec_ctx.Flush();
   gpr_mu_lock(g_mu);
   for (;;) {
     grpc_pollset_worker* worker = nullptr;
-    if (state.write_done) {
+    if (state.write_done &&
+        (!(grpc_event_engine_can_track_errors() && collect_timestamps) ||
+         gpr_atm_acq_load(&done_timestamps) == static_cast<gpr_atm>(1))) {
       break;
     }
     GPR_ASSERT(GRPC_LOG_IF_ERROR(
         "pollset_work", grpc_pollset_work(g_pollset, &worker, deadline)));
     gpr_mu_unlock(g_mu);
-
+    exec_ctx.Flush();
     gpr_mu_lock(g_mu);
   }
   gpr_mu_unlock(g_mu);
@@ -488,6 +571,7 @@ static void release_fd_test(size_t num_bytes, size_t slice_size) {
 }
 
 void run_tests(void) {
+  gpr_log(GPR_INFO, "run tests");
   size_t i = 0;
 
   read_test(100, 8192);
@@ -496,15 +580,25 @@ void run_tests(void) {
   read_test(10000, 1);
   large_read_test(8192);
   large_read_test(1);
-
-  write_test(100, 8192);
-  write_test(100, 1);
-  write_test(100000, 8192);
-  write_test(100000, 1);
-  write_test(100000, 137);
+  gpr_log(GPR_INFO, "done read tests");
+
+  write_test(100, 8192, false);
+  write_test(100, 1, false);
+  write_test(100000, 8192, false);
+  write_test(100000, 1, false);
+  write_test(100000, 137, false);
+  gpr_log(GPR_INFO, "done normal write tests");
+
+  write_test(100, 8192, true);
+  write_test(100, 1, true);
+  write_test(100000, 8192, true);
+  write_test(100000, 1, true);
+  write_test(100, 137, true);
+  gpr_log(GPR_INFO, "done super write tests");
 
   for (i = 1; i < 1000; i = GPR_MAX(i + 1, i * 5 / 4)) {
-    write_test(40320, i);
+    write_test(40320, i, false);
+    write_test(40320, i, true);
   }
 
   release_fd_test(100, 8192);
@@ -549,6 +643,8 @@ int main(int argc, char** argv) {
   grpc_closure destroyed;
   grpc_test_init(argc, argv);
   grpc_init();
+  gpr_log(GPR_INFO, "here");
+  grpc_core::grpc_tcp_set_write_timestamps_callback(timestamps_verifier);
   {
     grpc_core::ExecCtx exec_ctx;
     g_pollset = static_cast<grpc_pollset*>(gpr_zalloc(grpc_pollset_size()));