legacy-libs/grpc-cloned/deps/grpc/src/core/lib/iomgr/buffer_list.cc

   1 /*
   2  *
   3  * Copyright 2018 gRPC authors.
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  */
  18
  19 #include <grpc/support/port_platform.h>
  20
  21 #include "src/core/lib/iomgr/buffer_list.h"
  22 #include "src/core/lib/iomgr/port.h"
  23
  24 #include <grpc/support/log.h>
  25
  26 #ifdef GRPC_LINUX_ERRQUEUE
  27 #include <netinet/in.h>
  28 #include <string.h>
  29 #include <time.h>
  30
  31 #include "src/core/lib/gprpp/memory.h"
  32
  33 namespace grpc_core {
  34 namespace {
  35 /** Fills gpr_timespec gts based on values from timespec ts */
  36 void fill_gpr_from_timestamp(gpr_timespec* gts, const struct timespec* ts) {
  37   gts->tv_sec = ts->tv_sec;
  38   gts->tv_nsec = static_cast<int32_t>(ts->tv_nsec);
  39   gts->clock_type = GPR_CLOCK_REALTIME;
  40 }
  41
  42 void default_timestamps_callback(void* arg, grpc_core::Timestamps* ts,
  43                                  grpc_error* shudown_err) {
  44   gpr_log(GPR_DEBUG, "Timestamps callback has not been registered");
  45 }
  46
  47 /** The saved callback function that will be invoked when we get all the
  48  * timestamps that we are going to get for a TracedBuffer. */
  49 void (*timestamps_callback)(void*, grpc_core::Timestamps*,
  50                             grpc_error* shutdown_err) =
  51     default_timestamps_callback;
  52
  53 /* Used to extract individual opt stats from cmsg, so as to avoid troubles with
  54  * unaligned reads */
  55 template <typename T>
  56 T read_unaligned(const void* ptr) {
  57   T val;
  58   memcpy(&val, ptr, sizeof(val));
  59   return val;
  60 }
  61
  62 /* Extracts opt stats from the tcp_info struct \a info to \a metrics */
  63 void extract_opt_stats_from_tcp_info(ConnectionMetrics* metrics,
  64                                      const grpc_core::tcp_info* info) {
  65   if (info == nullptr) {
  66     return;
  67   }
  68   if (info->length > offsetof(grpc_core::tcp_info, tcpi_sndbuf_limited)) {
  69     metrics->recurring_retrans.set(info->tcpi_retransmits);
  70     metrics->is_delivery_rate_app_limited.set(
  71         info->tcpi_delivery_rate_app_limited);
  72     metrics->congestion_window.set(info->tcpi_snd_cwnd);
  73     metrics->reordering.set(info->tcpi_reordering);
  74     metrics->packet_retx.set(info->tcpi_total_retrans);
  75     metrics->pacing_rate.set(info->tcpi_pacing_rate);
  76     metrics->data_notsent.set(info->tcpi_notsent_bytes);
  77     if (info->tcpi_min_rtt != UINT32_MAX) {
  78       metrics->min_rtt.set(info->tcpi_min_rtt);
  79     }
  80     metrics->packet_sent.set(info->tcpi_data_segs_out);
  81     metrics->delivery_rate.set(info->tcpi_delivery_rate);
  82     metrics->busy_usec.set(info->tcpi_busy_time);
  83     metrics->rwnd_limited_usec.set(info->tcpi_rwnd_limited);
  84     metrics->sndbuf_limited_usec.set(info->tcpi_sndbuf_limited);
  85   }
  86   if (info->length > offsetof(grpc_core::tcp_info, tcpi_dsack_dups)) {
  87     metrics->data_sent.set(info->tcpi_bytes_sent);
  88     metrics->data_retx.set(info->tcpi_bytes_retrans);
  89     metrics->packet_spurious_retx.set(info->tcpi_dsack_dups);
  90   }
  91 }
  92
  93 /** Extracts opt stats from the given control message \a opt_stats to the
  94  * connection metrics \a metrics */
  95 void extract_opt_stats_from_cmsg(ConnectionMetrics* metrics,
  96                                  const cmsghdr* opt_stats) {
  97   if (opt_stats == nullptr) {
  98     return;
  99   }
 100   const auto* data = CMSG_DATA(opt_stats);
 101   constexpr int64_t cmsg_hdr_len = CMSG_ALIGN(sizeof(struct cmsghdr));
 102   const int64_t len = opt_stats->cmsg_len - cmsg_hdr_len;
 103   int64_t offset = 0;
 104
 105   while (offset < len) {
 106     const auto* attr = reinterpret_cast<const nlattr*>(data + offset);
 107     const void* val = data + offset + NLA_HDRLEN;
 108     switch (attr->nla_type) {
 109       case TCP_NLA_BUSY: {
 110         metrics->busy_usec.set(read_unaligned<uint64_t>(val));
 111         break;
 112       }
 113       case TCP_NLA_RWND_LIMITED: {
 114         metrics->rwnd_limited_usec.set(read_unaligned<uint64_t>(val));
 115         break;
 116       }
 117       case TCP_NLA_SNDBUF_LIMITED: {
 118         metrics->sndbuf_limited_usec.set(read_unaligned<uint64_t>(val));
 119         break;
 120       }
 121       case TCP_NLA_PACING_RATE: {
 122         metrics->pacing_rate.set(read_unaligned<uint64_t>(val));
 123         break;
 124       }
 125       case TCP_NLA_DELIVERY_RATE: {
 126         metrics->delivery_rate.set(read_unaligned<uint64_t>(val));
 127         break;
 128       }
 129       case TCP_NLA_DELIVERY_RATE_APP_LMT: {
 130         metrics->is_delivery_rate_app_limited.set(read_unaligned<uint8_t>(val));
 131         break;
 132       }
 133       case TCP_NLA_SND_CWND: {
 134         metrics->congestion_window.set(read_unaligned<uint32_t>(val));
 135         break;
 136       }
 137       case TCP_NLA_MIN_RTT: {
 138         metrics->min_rtt.set(read_unaligned<uint32_t>(val));
 139         break;
 140       }
 141       case TCP_NLA_SRTT: {
 142         metrics->srtt.set(read_unaligned<uint32_t>(val));
 143         break;
 144       }
 145       case TCP_NLA_RECUR_RETRANS: {
 146         metrics->recurring_retrans.set(read_unaligned<uint8_t>(val));
 147         break;
 148       }
 149       case TCP_NLA_BYTES_SENT: {
 150         metrics->data_sent.set(read_unaligned<uint64_t>(val));
 151         break;
 152       }
 153       case TCP_NLA_DATA_SEGS_OUT: {
 154         metrics->packet_sent.set(read_unaligned<uint64_t>(val));
 155         break;
 156       }
 157       case TCP_NLA_TOTAL_RETRANS: {
 158         metrics->packet_retx.set(read_unaligned<uint64_t>(val));
 159         break;
 160       }
 161       case TCP_NLA_DELIVERED: {
 162         metrics->packet_delivered.set(read_unaligned<uint32_t>(val));
 163         break;
 164       }
 165       case TCP_NLA_DELIVERED_CE: {
 166         metrics->packet_delivered_ce.set(read_unaligned<uint32_t>(val));
 167         break;
 168       }
 169       case TCP_NLA_BYTES_RETRANS: {
 170         metrics->data_retx.set(read_unaligned<uint64_t>(val));
 171         break;
 172       }
 173       case TCP_NLA_DSACK_DUPS: {
 174         metrics->packet_spurious_retx.set(read_unaligned<uint32_t>(val));
 175         break;
 176       }
 177       case TCP_NLA_REORDERING: {
 178         metrics->reordering.set(read_unaligned<uint32_t>(val));
 179         break;
 180       }
 181       case TCP_NLA_SND_SSTHRESH: {
 182         metrics->snd_ssthresh.set(read_unaligned<uint32_t>(val));
 183         break;
 184       }
 185     }
 186     offset += NLA_ALIGN(attr->nla_len);
 187   }
 188 }
 189
 190 static int get_socket_tcp_info(grpc_core::tcp_info* info, int fd) {
 191   memset(info, 0, sizeof(*info));
 192   info->length = sizeof(*info) - sizeof(socklen_t);
 193   return getsockopt(fd, IPPROTO_TCP, TCP_INFO, info, &(info->length));
 194 }
 195 } /* namespace */
 196
 197 void TracedBuffer::AddNewEntry(TracedBuffer** head, uint32_t seq_no, int fd,
 198                                void* arg) {
 199   GPR_DEBUG_ASSERT(head != nullptr);
 200   TracedBuffer* new_elem = New<TracedBuffer>(seq_no, arg);
 201   /* Store the current time as the sendmsg time. */
 202   new_elem->ts_.sendmsg_time.time = gpr_now(GPR_CLOCK_REALTIME);
 203   new_elem->ts_.scheduled_time.time = gpr_inf_past(GPR_CLOCK_REALTIME);
 204   new_elem->ts_.sent_time.time = gpr_inf_past(GPR_CLOCK_REALTIME);
 205   new_elem->ts_.acked_time.time = gpr_inf_past(GPR_CLOCK_REALTIME);
 206
 207   if (get_socket_tcp_info(&new_elem->ts_.info, fd) == 0) {
 208     extract_opt_stats_from_tcp_info(&new_elem->ts_.sendmsg_time.metrics,
 209                                     &new_elem->ts_.info);
 210   }
 211   if (*head == nullptr) {
 212     *head = new_elem;
 213     return;
 214   }
 215   /* Append at the end. */
 216   TracedBuffer* ptr = *head;
 217   while (ptr->next_ != nullptr) {
 218     ptr = ptr->next_;
 219   }
 220   ptr->next_ = new_elem;
 221 }
 222
 223 void TracedBuffer::ProcessTimestamp(TracedBuffer** head,
 224                                     struct sock_extended_err* serr,
 225                                     struct cmsghdr* opt_stats,
 226                                     struct scm_timestamping* tss) {
 227   GPR_DEBUG_ASSERT(head != nullptr);
 228   TracedBuffer* elem = *head;
 229   TracedBuffer* next = nullptr;
 230   while (elem != nullptr) {
 231     /* The byte number refers to the sequence number of the last byte which this
 232      * timestamp relates to. */
 233     if (serr->ee_data >= elem->seq_no_) {
 234       switch (serr->ee_info) {
 235         case SCM_TSTAMP_SCHED:
 236           fill_gpr_from_timestamp(&(elem->ts_.scheduled_time.time),
 237                                   &(tss->ts[0]));
 238           extract_opt_stats_from_cmsg(&(elem->ts_.scheduled_time.metrics),
 239                                       opt_stats);
 240           elem = elem->next_;
 241           break;
 242         case SCM_TSTAMP_SND:
 243           fill_gpr_from_timestamp(&(elem->ts_.sent_time.time), &(tss->ts[0]));
 244           extract_opt_stats_from_cmsg(&(elem->ts_.sent_time.metrics),
 245                                       opt_stats);
 246           elem = elem->next_;
 247           break;
 248         case SCM_TSTAMP_ACK:
 249           fill_gpr_from_timestamp(&(elem->ts_.acked_time.time), &(tss->ts[0]));
 250           extract_opt_stats_from_cmsg(&(elem->ts_.acked_time.metrics),
 251                                       opt_stats);
 252           /* Got all timestamps. Do the callback and free this TracedBuffer.
 253            * The thing below can be passed by value if we don't want the
 254            * restriction on the lifetime. */
 255           timestamps_callback(elem->arg_, &(elem->ts_), GRPC_ERROR_NONE);
 256           next = elem->next_;
 257           Delete<TracedBuffer>(elem);
 258           *head = elem = next;
 259           break;
 260         default:
 261           abort();
 262       }
 263     } else {
 264       break;
 265     }
 266   }
 267 }
 268
 269 void TracedBuffer::Shutdown(TracedBuffer** head, void* remaining,
 270                             grpc_error* shutdown_err) {
 271   GPR_DEBUG_ASSERT(head != nullptr);
 272   TracedBuffer* elem = *head;
 273   while (elem != nullptr) {
 274     timestamps_callback(elem->arg_, &(elem->ts_), shutdown_err);
 275     auto* next = elem->next_;
 276     Delete<TracedBuffer>(elem);
 277     elem = next;
 278   }
 279   *head = nullptr;
 280   if (remaining != nullptr) {
 281     timestamps_callback(remaining, nullptr, shutdown_err);
 282   }
 283   GRPC_ERROR_UNREF(shutdown_err);
 284 }
 285
 286 void grpc_tcp_set_write_timestamps_callback(void (*fn)(void*,
 287                                                        grpc_core::Timestamps*,
 288                                                        grpc_error* error)) {
 289   timestamps_callback = fn;
 290 }
 291 } /* namespace grpc_core */
 292
 293 #else /* GRPC_LINUX_ERRQUEUE */
 294
 295 namespace grpc_core {
 296 void grpc_tcp_set_write_timestamps_callback(void (*fn)(void*,
 297                                                        grpc_core::Timestamps*,
 298                                                        grpc_error* error)) {
 299   gpr_log(GPR_DEBUG, "Timestamps callback is not enabled for this platform");
 300 }
 301 } /* namespace grpc_core */
 302
 303 #endif /* GRPC_LINUX_ERRQUEUE */