Adding docs and cleaning up
diff --git a/src/core/lib/iomgr/buffer_list.cc b/src/core/lib/iomgr/buffer_list.cc
index 4f0b522..8d1645d 100644
--- a/src/core/lib/iomgr/buffer_list.cc
+++ b/src/core/lib/iomgr/buffer_list.cc
@@ -31,14 +31,12 @@
namespace grpc_core {
void TracedBuffer::AddNewEntry(TracedBuffer** head, uint32_t seq_no,
void* arg) {
- gpr_log(GPR_INFO, "Adding new entry %u", seq_no);
GPR_DEBUG_ASSERT(head != nullptr);
TracedBuffer* new_elem = New<TracedBuffer>(seq_no, arg);
/* Store the current time as the sendmsg time. */
new_elem->ts_.sendmsg_time = gpr_now(GPR_CLOCK_REALTIME);
if (*head == nullptr) {
*head = new_elem;
- gpr_log(GPR_INFO, "returning");
return;
}
/* Append at the end. */
@@ -47,16 +45,18 @@
ptr = ptr->next_;
}
ptr->next_ = new_elem;
- gpr_log(GPR_INFO, "returning");
}
namespace {
+/** Fills gpr_timespec gts based on values from timespec ts */
void fill_gpr_from_timestamp(gpr_timespec* gts, const struct timespec* ts) {
gts->tv_sec = ts->tv_sec;
gts->tv_nsec = static_cast<int32_t>(ts->tv_nsec);
gts->clock_type = GPR_CLOCK_REALTIME;
}
+/** The saved callback function that will be invoked when we get all the
+ * timestamps that we are going to get for a TracedBuffer. */
void (*timestamps_callback)(void*, grpc_core::Timestamps*,
grpc_error* shutdown_err);
} /* namespace */
@@ -64,12 +64,10 @@
void TracedBuffer::ProcessTimestamp(TracedBuffer** head,
struct sock_extended_err* serr,
struct scm_timestamping* tss) {
- gpr_log(GPR_INFO, "Got timestamp %d", serr->ee_data);
GPR_DEBUG_ASSERT(head != nullptr);
TracedBuffer* elem = *head;
TracedBuffer* next = nullptr;
while (elem != nullptr) {
- gpr_log(GPR_INFO, "looping");
/* The byte number refers to the sequence number of the last byte which this
* timestamp relates to. For scheduled and send, we are interested in the
* timestamp for the first byte, whereas for ack, we are interested in the
@@ -77,17 +75,14 @@
if (serr->ee_data >= elem->seq_no_) {
switch (serr->ee_info) {
case SCM_TSTAMP_SCHED:
- gpr_log(GPR_INFO, "type sched\n");
fill_gpr_from_timestamp(&(elem->ts_.scheduled_time), &(tss->ts[0]));
elem = elem->next_;
break;
case SCM_TSTAMP_SND:
- gpr_log(GPR_INFO, "type send\n");
fill_gpr_from_timestamp(&(elem->ts_.sent_time), &(tss->ts[0]));
elem = elem->next_;
break;
case SCM_TSTAMP_ACK:
- gpr_log(GPR_INFO, "type ack\n");
if (serr->ee_data >= elem->seq_no_) {
fill_gpr_from_timestamp(&(elem->ts_.acked_time), &(tss->ts[0]));
/* Got all timestamps. Do the callback and free this TracedBuffer.
diff --git a/src/core/lib/iomgr/buffer_list.h b/src/core/lib/iomgr/buffer_list.h
index d42f97f..0f66dcc 100644
--- a/src/core/lib/iomgr/buffer_list.h
+++ b/src/core/lib/iomgr/buffer_list.h
@@ -37,20 +37,34 @@
gpr_timespec acked_time;
};
+/** TracedBuffer is a class to keep track of timestamps for a specific buffer in
+ * the TCP layer. We are only tracking timestamps for Linux kernels and hence
+ * this class would only be used by Linux platforms. For all other platforms,
+ * TracedBuffer would be an empty class.
+ *
+ * The timestamps collected are according to grpc_core::Timestamps declared
+ * above.
+ *
+ * A TracedBuffer list is kept track of using the head element of the list. If
+ * the head element of the list is nullptr, then the list is empty.
+ */
#ifdef GRPC_LINUX_ERRQUEUE
class TracedBuffer {
public:
- /** Add a new entry in the TracedBuffer list pointed to by head */
+ /** Add a new entry in the TracedBuffer list pointed to by head. Also saves
+ * sendmsg_time with the current timestamp. */
static void AddNewEntry(grpc_core::TracedBuffer** head, uint32_t seq_no,
void* arg);
- /** Processes a timestamp received */
+ /** Processes a received timestamp based on sock_extended_err and
+ * scm_timestamping structures. It will invoke the timestamps callback if the
+ * timestamp type is SCM_TSTAMP_ACK. */
static void ProcessTimestamp(grpc_core::TracedBuffer** head,
struct sock_extended_err* serr,
struct scm_timestamping* tss);
- /** Calls the callback for each traced buffer in the list with timestamps that
- * it has. */
+ /** Cleans the list by calling the callback for each traced buffer in the list
+ * with timestamps that it has. */
static void Shutdown(grpc_core::TracedBuffer** head,
grpc_error* shutdown_err);
@@ -58,24 +72,23 @@
GPRC_ALLOW_CLASS_TO_USE_NON_PUBLIC_NEW
TracedBuffer(int seq_no, void* arg)
- : seq_no_(seq_no), arg_(arg), next_(nullptr) {
- gpr_log(GPR_INFO, "seq_no %d", seq_no_);
- }
+ : seq_no_(seq_no), arg_(arg), next_(nullptr) {}
uint32_t seq_no_; /* The sequence number for the last byte in the buffer */
void* arg_; /* The arg to pass to timestamps_callback */
- grpc_core::Timestamps ts_;
- grpc_core::TracedBuffer* next_;
+ grpc_core::Timestamps ts_; /* The timestamps corresponding to this buffer */
+ grpc_core::TracedBuffer* next_; /* The next TracedBuffer in the list */
};
#else /* GRPC_LINUX_ERRQUEUE */
class TracedBuffer {};
#endif /* GRPC_LINUX_ERRQUEUE */
-/** Sets the timestamp callback */
+/** Sets the callback function to call when timestamps for a write are
+ * collected. The callback does not own a reference to error. */
void grpc_tcp_set_write_timestamps_callback(void (*fn)(void*,
grpc_core::Timestamps*,
grpc_error* error));
-}; // namespace grpc_core
+}; /* namespace grpc_core */
#endif /* GRPC_CORE_LIB_IOMGR_BUFFER_LIST_H */
diff --git a/src/core/lib/iomgr/endpoint.h b/src/core/lib/iomgr/endpoint.h
index ea39ea6..1f590a8 100644
--- a/src/core/lib/iomgr/endpoint.h
+++ b/src/core/lib/iomgr/endpoint.h
@@ -72,6 +72,8 @@
\a slices may be mutated at will by the endpoint until cb is called.
No guarantee is made to the content of slices after a write EXCEPT that
it is a valid slice buffer.
+ \a arg is platform specific. It is currently only used by TCP on linux
+ platforms as an argument that would be forwarded to the timestamps callback.
*/
void grpc_endpoint_write(grpc_endpoint* ep, grpc_slice_buffer* slices,
grpc_closure* cb, void* arg);
diff --git a/src/core/lib/iomgr/endpoint_cfstream.cc b/src/core/lib/iomgr/endpoint_cfstream.cc
index 70d674f..df2cf50 100644
--- a/src/core/lib/iomgr/endpoint_cfstream.cc
+++ b/src/core/lib/iomgr/endpoint_cfstream.cc
@@ -268,7 +268,7 @@
}
static void CFStreamWrite(grpc_endpoint* ep, grpc_slice_buffer* slices,
- grpc_closure* cb, void *arg) {
+ grpc_closure* cb, void* arg) {
CFStreamEndpoint* ep_impl = reinterpret_cast<CFStreamEndpoint*>(ep);
if (grpc_tcp_trace.enabled()) {
gpr_log(GPR_DEBUG, "CFStream endpoint:%p write (%p, %p) length:%zu",
diff --git a/src/core/lib/iomgr/internal_errqueue.h b/src/core/lib/iomgr/internal_errqueue.h
index 92292e9..bbe3377 100644
--- a/src/core/lib/iomgr/internal_errqueue.h
+++ b/src/core/lib/iomgr/internal_errqueue.h
@@ -16,6 +16,12 @@
*
*/
+/* This file contains constants defined in <linux/errqueue.h> and
+ * <linux/net_tstamp.h> so as to allow collecting network timestamps in the
+ * kernel. This file allows tcp_posix.cc to compile on platforms that do not
+ * have <linux/errqueue.h> and <linux/net_tstamp.h>.
+ */
+
#ifndef GRPC_CORE_LIB_IOMGR_INTERNAL_ERRQUEUE_H
#define GRPC_CORE_LIB_IOMGR_INTERNAL_ERRQUEUE_H
diff --git a/src/core/lib/iomgr/tcp_posix.cc b/src/core/lib/iomgr/tcp_posix.cc
index 97251e7..4300a9f 100644
--- a/src/core/lib/iomgr/tcp_posix.cc
+++ b/src/core/lib/iomgr/tcp_posix.cc
@@ -106,12 +106,16 @@
grpc_resource_user* resource_user;
grpc_resource_user_slice_allocator slice_allocator;
- grpc_core::TracedBuffer* head;
- gpr_mu traced_buffer_lock;
- void* outgoing_buffer_arg;
- int bytes_counter;
- bool socket_ts_enabled;
- gpr_atm stop_error_notification;
+ grpc_core::TracedBuffer* head; /* List of traced buffers */
+ gpr_mu traced_buffer_lock; /* Lock for access to list of traced buffers */
+ void* outgoing_buffer_arg; /* buffer arg provided on grpc_endpoint_write */
+ int bytes_counter; /* Current TCP relative sequence number. Used for
+ timestamping traced buffers. */
+ bool socket_ts_enabled; /* True if timestamping options are set on the socket
+ */
+ gpr_atm
+ stop_error_notification; /* Set to 1 if we do not want to be notified on
+ errors anymore */
};
struct backup_poller {
@@ -360,7 +364,6 @@
grpc_tcp* tcp = reinterpret_cast<grpc_tcp*>(ep);
grpc_slice_buffer_reset_and_unref_internal(&tcp->last_read_buffer);
if (grpc_event_engine_can_track_errors()) {
- // gpr_log(GPR_INFO, "stop errors");
gpr_atm_no_barrier_store(&tcp->stop_error_notification, true);
grpc_fd_notify_on_error(tcp->em_fd, nullptr);
}
@@ -539,6 +542,8 @@
static bool tcp_write_with_timestamps(grpc_tcp* tcp, struct msghdr* msg,
size_t sending_length,
ssize_t* sent_length, grpc_error** error);
+
+/** The callback function to be invoked when we get an error on the socket. */
static void tcp_handle_error(void* arg /* grpc_tcp */, grpc_error* error);
#ifdef GRPC_LINUX_ERRQUEUE
@@ -547,13 +552,14 @@
ssize_t* sent_length,
grpc_error** error) {
if (!tcp->socket_ts_enabled) {
- // gpr_log(GPR_INFO, "setting options yo");
uint32_t opt = grpc_core::kTimestampingSocketOptions;
if (setsockopt(tcp->fd, SOL_SOCKET, SO_TIMESTAMPING,
static_cast<void*>(&opt), sizeof(opt)) != 0) {
*error = tcp_annotate_error(GRPC_OS_ERROR(errno, "setsockopt"), tcp);
grpc_slice_buffer_reset_and_unref_internal(tcp->outgoing_buffer);
- gpr_log(GPR_INFO, "failed to set");
+ if (grpc_tcp_trace.enabled()) {
+ gpr_log(GPR_ERROR, "Failed to set timestamping options on the socket.");
+ }
return false;
}
tcp->socket_ts_enabled = true;
@@ -589,18 +595,29 @@
return true;
}
+/** Reads \a cmsg to derive timestamps from the control messages. If a valid
+ * timestamp is found, the traced buffer list is updated with this timestamp.
+ * The caller of this function should be looping on the control messages found
+ * in \a msg. \a cmsg should point to the control message that the caller wants
+ * processed.
+ * On return, a pointer to a control message is returned. On the next iteration,
+ * CMSG_NXTHDR(msg, ret_val) should be passed as \a cmsg. */
struct cmsghdr* process_timestamp(grpc_tcp* tcp, msghdr* msg,
struct cmsghdr* cmsg) {
auto next_cmsg = CMSG_NXTHDR(msg, cmsg);
if (next_cmsg == nullptr) {
- gpr_log(GPR_ERROR, "Received timestamp without extended error");
+ if (grpc_tcp_trace.enabled()) {
+ gpr_log(GPR_ERROR, "Received timestamp without extended error");
+ }
return cmsg;
}
if (!(next_cmsg->cmsg_level == SOL_IP || next_cmsg->cmsg_level == SOL_IPV6) ||
!(next_cmsg->cmsg_type == IP_RECVERR ||
next_cmsg->cmsg_type == IPV6_RECVERR)) {
- gpr_log(GPR_ERROR, "Unexpected cmsg");
+ if (grpc_tcp_trace.enabled()) {
+ gpr_log(GPR_ERROR, "Unexpected control message");
+ }
return cmsg;
}
@@ -609,14 +626,13 @@
auto serr = reinterpret_cast<struct sock_extended_err*>(CMSG_DATA(next_cmsg));
if (serr->ee_errno != ENOMSG ||
serr->ee_origin != SO_EE_ORIGIN_TIMESTAMPING) {
- gpr_log(GPR_ERROR, "Unexpected cmsg");
+ gpr_log(GPR_ERROR, "Unexpected control message");
return cmsg;
}
/* The error handling can potentially be done on another thread so we need
* to protect the traced buffer list. A lock free list might be better. Using
* a simple mutex for now. */
gpr_mu_lock(&tcp->traced_buffer_lock);
- // gpr_log(GPR_INFO, "processing timestamp");
grpc_core::TracedBuffer::ProcessTimestamp(&tcp->head, serr, tss);
gpr_mu_unlock(&tcp->traced_buffer_lock);
return next_cmsg;
@@ -624,14 +640,12 @@
/** For linux platforms, reads the socket's error queue and processes error
* messages from the queue. Returns true if all the errors processed were
- * timestamps. Returns false if the any of the errors were not timestamps. For
+ * timestamps. Returns false if any of the errors were not timestamps. For
* non-linux platforms, error processing is not enabled currently, and hence
* crashes out.
*/
static bool process_errors(grpc_tcp* tcp) {
- // gpr_log(GPR_INFO, "process errors");
while (true) {
- // gpr_log(GPR_INFO, "looping");
struct iovec iov;
iov.iov_base = nullptr;
iov.iov_len = 0;
@@ -654,24 +668,22 @@
int r, saved_errno;
do {
- // gpr_log(GPR_INFO, "error recvmsg");
r = recvmsg(tcp->fd, &msg, MSG_ERRQUEUE);
saved_errno = errno;
} while (r < 0 && saved_errno == EINTR);
if (r == -1 && saved_errno == EAGAIN) {
- // gpr_log(GPR_INFO, "here");
return true; /* No more errors to process */
}
if (r == -1) {
- // gpr_log(GPR_INFO, "%d", saved_errno);
return false;
}
- if ((msg.msg_flags & MSG_CTRUNC) == 1) {
- gpr_log(GPR_INFO, "Error message was truncated.");
+ if (grpc_tcp_trace.enabled()) {
+ if ((msg.msg_flags & MSG_CTRUNC) == 1) {
+ gpr_log(GPR_INFO, "Error message was truncated.");
+ }
}
- // gpr_log(GPR_INFO, "%d %lu", r, msg.msg_controllen);
if (msg.msg_controllen == 0) {
/* There was no control message read. Return now */
return true;
@@ -680,10 +692,12 @@
cmsg = CMSG_NXTHDR(&msg, cmsg)) {
if (cmsg->cmsg_level != SOL_SOCKET ||
cmsg->cmsg_type != SCM_TIMESTAMPING) {
- /* Got a weird one, not a timestamp */
- gpr_log(GPR_INFO, "weird %d %d %d", r, cmsg->cmsg_level,
- cmsg->cmsg_type);
- continue;
+ /* Got a weird control message, not a timestamp */
+ if (grpc_tcp_trace.enabled()) {
+ gpr_log(GPR_INFO, "weird control message cmsg_level:%d cmsg_type:%d",
+ cmsg->cmsg_level, cmsg->cmsg_type);
+ }
+ return false;
}
process_timestamp(tcp, &msg, cmsg);
}
@@ -691,7 +705,6 @@
}
static void tcp_handle_error(void* arg /* grpc_tcp */, grpc_error* error) {
- // gpr_log(GPR_INFO, "grpc_tcp_handle_error");
grpc_tcp* tcp = static_cast<grpc_tcp*>(arg);
if (grpc_tcp_trace.enabled()) {
gpr_log(GPR_INFO, "TCP:%p got_error: %s", tcp, grpc_error_string(error));
@@ -701,15 +714,10 @@
static_cast<bool>(gpr_atm_acq_load(&tcp->stop_error_notification))) {
/* We aren't going to register to hear on error anymore, so it is safe to
* unref. */
- // gpr_log(GPR_INFO, "%p %d", error,
- // static_cast<bool>(gpr_atm_acq_load(&tcp->stop_error_notification)));
- // gpr_log(GPR_INFO, "unref");
grpc_core::TracedBuffer::Shutdown(&tcp->head, GRPC_ERROR_REF(error));
TCP_UNREF(tcp, "error");
- // gpr_log(GPR_INFO, "here");
} else {
if (!process_errors(tcp)) {
- // gpr_log(GPR_INFO, "no timestamps");
/* This was not a timestamps error. This was an actual error. Set the
* read and write closures to be ready.
*/
@@ -719,7 +727,6 @@
GRPC_CLOSURE_INIT(&tcp->error_closure, tcp_handle_error, tcp,
grpc_schedule_on_exec_ctx);
grpc_fd_notify_on_error(tcp->em_fd, &tcp->error_closure);
- // gpr_log(GPR_INFO, "udhar se");
}
}
@@ -798,7 +805,6 @@
sent_length = sendmsg(tcp->fd, &msg, SENDMSG_FLAGS);
} while (sent_length < 0 && errno == EINTR);
}
- // gpr_log(GPR_INFO, "sent length %ld", sent_length);
if (sent_length < 0) {
if (errno == EAGAIN) {
@@ -869,7 +875,6 @@
const char* str = grpc_error_string(error);
gpr_log(GPR_INFO, "write: %s", str);
}
- // gpr_log(GPR_INFO, "scheduling callback");
GRPC_CLOSURE_SCHED(cb, error);
TCP_UNREF(tcp, "write");
}
@@ -913,14 +918,12 @@
if (grpc_tcp_trace.enabled()) {
gpr_log(GPR_INFO, "write: delayed");
}
- // gpr_log(GPR_INFO, "notify");
notify_on_write(tcp);
} else {
if (grpc_tcp_trace.enabled()) {
const char* str = grpc_error_string(error);
gpr_log(GPR_INFO, "write: %s", str);
}
- // gpr_log(GPR_INFO, "sched");
GRPC_CLOSURE_SCHED(cb, error);
}
}
@@ -1069,7 +1072,7 @@
tcp->release_fd_cb = done;
grpc_slice_buffer_reset_and_unref_internal(&tcp->last_read_buffer);
if (grpc_event_engine_can_track_errors()) {
- // gpr_log(GPR_INFO, "stop errors");
+ /* Stop errors notification. */
gpr_atm_no_barrier_store(&tcp->stop_error_notification, true);
grpc_fd_notify_on_error(tcp->em_fd, nullptr);
}
diff --git a/src/core/lib/iomgr/tcp_posix.h b/src/core/lib/iomgr/tcp_posix.h
index 322af62..eff825c 100644
--- a/src/core/lib/iomgr/tcp_posix.h
+++ b/src/core/lib/iomgr/tcp_posix.h
@@ -57,9 +57,4 @@
void grpc_tcp_destroy_and_release_fd(grpc_endpoint* ep, int* fd,
grpc_closure* done);
-/** Sets the callback function to call when timestamps for a write are
- * collected. */
-void grpc_tcp_set_write_timestamps_callback(void (*fn)(void*,
- grpc_core::Timestamps*));
-
#endif /* GRPC_CORE_LIB_IOMGR_TCP_POSIX_H */
diff --git a/test/core/iomgr/tcp_posix_test.cc b/test/core/iomgr/tcp_posix_test.cc
index 648150a..6447cc2 100644
--- a/test/core/iomgr/tcp_posix_test.cc
+++ b/test/core/iomgr/tcp_posix_test.cc
@@ -72,7 +72,6 @@
}
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));
@@ -86,10 +85,6 @@
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;
@@ -337,9 +332,7 @@
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");
gpr_mu_lock(g_mu);
- gpr_log(GPR_INFO, "Signalling write done");
state->write_done = 1;
GPR_ASSERT(
GRPC_LOG_IF_ERROR("pollset_kick", grpc_pollset_kick(g_pollset, nullptr)));
@@ -360,22 +353,17 @@
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) {
@@ -460,9 +448,7 @@
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 (;;) {
@@ -571,7 +557,6 @@
}
void run_tests(void) {
- gpr_log(GPR_INFO, "run tests");
size_t i = 0;
read_test(100, 8192);
@@ -580,21 +565,18 @@
read_test(10000, 1);
large_read_test(8192);
large_read_test(1);
- 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, false);
@@ -643,7 +625,6 @@
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;