<commit_before>/*
 *
 * 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 <grpc/support/port_platform.h>

#include "src/core/lib/iomgr/buffer_list.h"
#include "src/core/lib/iomgr/port.h"

#include <grpc/support/log.h>

#ifdef GRPC_LINUX_ERRQUEUE
#include <netinet/in.h>
#include <string.h>
#include <time.h>

#include "src/core/lib/gprpp/memory.h"

namespace grpc_core {
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;
}

void default_timestamps_callback(void* /*arg*/, grpc_core::Timestamps* /*ts*/,
                                 grpc_error_handle /*shudown_err*/) {
  gpr_log(GPR_DEBUG, "Timestamps callback has not been registered");
}

/** 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_handle shutdown_err) =
    default_timestamps_callback;

/* Used to extract individual opt stats from cmsg, so as to avoid troubles with
 * unaligned reads */
template <typename T>
T read_unaligned(const void* ptr) {
  T val;
  memcpy(&val, ptr, sizeof(val));
  return val;
}

/* Extracts opt stats from the tcp_info struct \a info to \a metrics */
void extract_opt_stats_from_tcp_info(ConnectionMetrics* metrics,
                                     const grpc_core::tcp_info* info) {
  if (info == nullptr) {
    return;
  }
  if (info->length > offsetof(grpc_core::tcp_info, tcpi_sndbuf_limited)) {
    metrics->recurring_retrans.emplace(info->tcpi_retransmits);
    metrics->is_delivery_rate_app_limited.emplace(
        info->tcpi_delivery_rate_app_limited);
    metrics->congestion_window.emplace(info->tcpi_snd_cwnd);
    metrics->reordering.emplace(info->tcpi_reordering);
    metrics->packet_retx.emplace(info->tcpi_total_retrans);
    metrics->pacing_rate.emplace(info->tcpi_pacing_rate);
    metrics->data_notsent.emplace(info->tcpi_notsent_bytes);
    if (info->tcpi_min_rtt != UINT32_MAX) {
      metrics->min_rtt.emplace(info->tcpi_min_rtt);
    }
    metrics->packet_sent.emplace(info->tcpi_data_segs_out);
    metrics->delivery_rate.emplace(info->tcpi_delivery_rate);
    metrics->busy_usec.emplace(info->tcpi_busy_time);
    metrics->rwnd_limited_usec.emplace(info->tcpi_rwnd_limited);
    metrics->sndbuf_limited_usec.emplace(info->tcpi_sndbuf_limited);
  }
  if (info->length > offsetof(grpc_core::tcp_info, tcpi_dsack_dups)) {
    metrics->data_sent.emplace(info->tcpi_bytes_sent);
    metrics->data_retx.emplace(info->tcpi_bytes_retrans);
    metrics->packet_spurious_retx.emplace(info->tcpi_dsack_dups);
  }
}

/** Extracts opt stats from the given control message \a opt_stats to the
 * connection metrics \a metrics */
void extract_opt_stats_from_cmsg(ConnectionMetrics* metrics,
                                 const cmsghdr* opt_stats) {
  if (opt_stats == nullptr) {
    return;
  }
  const auto* data = CMSG_DATA(opt_stats);
  constexpr int64_t cmsg_hdr_len = CMSG_ALIGN(sizeof(struct cmsghdr));
  const int64_t len = opt_stats->cmsg_len - cmsg_hdr_len;
  int64_t offset = 0;

  while (offset < len) {
    const auto* attr = reinterpret_cast<const nlattr*>(data + offset);
    const void* val = data + offset + NLA_HDRLEN;
    switch (attr->nla_type) {
      case TCP_NLA_BUSY: {
        metrics->busy_usec.emplace(read_unaligned<uint64_t>(val));
        break;
      }
      case TCP_NLA_RWND_LIMITED: {
        metrics->rwnd_limited_usec.emplace(read_unaligned<uint64_t>(val));
        break;
      }
      case TCP_NLA_SNDBUF_LIMITED: {
        metrics->sndbuf_limited_usec.emplace(read_unaligned<uint64_t>(val));
        break;
      }
      case TCP_NLA_PACING_RATE: {
        metrics->pacing_rate.emplace(read_unaligned<uint64_t>(val));
        break;
      }
      case TCP_NLA_DELIVERY_RATE: {
        metrics->delivery_rate.emplace(read_unaligned<uint64_t>(val));
        break;
      }
      case TCP_NLA_DELIVERY_RATE_APP_LMT: {
        metrics->is_delivery_rate_app_limited.emplace(
            read_unaligned<uint8_t>(val));
        break;
      }
      case TCP_NLA_SND_CWND: {
        metrics->congestion_window.emplace(read_unaligned<uint32_t>(val));
        break;
      }
      case TCP_NLA_MIN_RTT: {
        metrics->min_rtt.emplace(read_unaligned<uint32_t>(val));
        break;
      }
      case TCP_NLA_SRTT: {
        metrics->srtt.emplace(read_unaligned<uint32_t>(val));
        break;
      }
      case TCP_NLA_RECUR_RETRANS: {
        metrics->recurring_retrans.emplace(read_unaligned<uint8_t>(val));
        break;
      }
      case TCP_NLA_BYTES_SENT: {
        metrics->data_sent.emplace(read_unaligned<uint64_t>(val));
        break;
      }
      case TCP_NLA_DATA_SEGS_OUT: {
        metrics->packet_sent.emplace(read_unaligned<uint64_t>(val));
        break;
      }
      case TCP_NLA_TOTAL_RETRANS: {
        metrics->packet_retx.emplace(read_unaligned<uint64_t>(val));
        break;
      }
      case TCP_NLA_DELIVERED: {
        metrics->packet_delivered.emplace(read_unaligned<uint32_t>(val));
        break;
      }
      case TCP_NLA_DELIVERED_CE: {
        metrics->packet_delivered_ce.emplace(read_unaligned<uint32_t>(val));
        break;
      }
      case TCP_NLA_BYTES_RETRANS: {
        metrics->data_retx.emplace(read_unaligned<uint64_t>(val));
        break;
      }
      case TCP_NLA_DSACK_DUPS: {
        metrics->packet_spurious_retx.emplace(read_unaligned<uint32_t>(val));
        break;
      }
      case TCP_NLA_REORDERING: {
        metrics->reordering.emplace(read_unaligned<uint32_t>(val));
        break;
      }
      case TCP_NLA_SND_SSTHRESH: {
        metrics->snd_ssthresh.emplace(read_unaligned<uint32_t>(val));
        break;
      }
    }
    offset += NLA_ALIGN(attr->nla_len);
  }
}

static int get_socket_tcp_info(grpc_core::tcp_info* info, int fd) {
  memset(info, 0, sizeof(*info));
  info->length = sizeof(*info) - sizeof(socklen_t);
  return getsockopt(fd, IPPROTO_TCP, TCP_INFO, info, &(info->length));
}
} /* namespace */

void TracedBuffer::AddNewEntry(TracedBuffer** head, uint32_t seq_no, int fd,
                               void* arg) {
  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.time = gpr_now(GPR_CLOCK_REALTIME);
  new_elem->ts_.scheduled_time.time = gpr_inf_past(GPR_CLOCK_REALTIME);
  new_elem->ts_.sent_time.time = gpr_inf_past(GPR_CLOCK_REALTIME);
  new_elem->ts_.acked_time.time = gpr_inf_past(GPR_CLOCK_REALTIME);

  if (get_socket_tcp_info(&new_elem->ts_.info, fd) == 0) {
    extract_opt_stats_from_tcp_info(&new_elem->ts_.sendmsg_time.metrics,
                                    &new_elem->ts_.info);
  }
  if (*head == nullptr) {
    *head = new_elem;
    return;
  }
  /* Append at the end. */
  TracedBuffer* ptr = *head;
  while (ptr->next_ != nullptr) {
    ptr = ptr->next_;
  }
  ptr->next_ = new_elem;
}

void TracedBuffer::ProcessTimestamp(TracedBuffer** head,<commit_msg>Provide correct len to getsockopt(TCP_INFO). (#26837)<commit_after>/*
 *
 * 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 <grpc/support/port_platform.h>

#include "src/core/lib/iomgr/buffer_list.h"
#include "src/core/lib/iomgr/port.h"

#include <grpc/support/log.h>

#ifdef GRPC_LINUX_ERRQUEUE
#include <netinet/in.h>
#include <string.h>
#include <time.h>

#include "src/core/lib/gprpp/memory.h"

namespace grpc_core {
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;
}

void default_timestamps_callback(void* /*arg*/, grpc_core::Timestamps* /*ts*/,
                                 grpc_error_handle /*shudown_err*/) {
  gpr_log(GPR_DEBUG, "Timestamps callback has not been registered");
}

/** 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_handle shutdown_err) =
    default_timestamps_callback;

/* Used to extract individual opt stats from cmsg, so as to avoid troubles with
 * unaligned reads */
template <typename T>
T read_unaligned(const void* ptr) {
  T val;
  memcpy(&val, ptr, sizeof(val));
  return val;
}

/* Extracts opt stats from the tcp_info struct \a info to \a metrics */
void extract_opt_stats_from_tcp_info(ConnectionMetrics* metrics,
                                     const grpc_core::tcp_info* info) {
  if (info == nullptr) {
    return;
  }
  if (info->length > offsetof(grpc_core::tcp_info, tcpi_sndbuf_limited)) {
    metrics->recurring_retrans.emplace(info->tcpi_retransmits);
    metrics->is_delivery_rate_app_limited.emplace(
        info->tcpi_delivery_rate_app_limited);
    metrics->congestion_window.emplace(info->tcpi_snd_cwnd);
    metrics->reordering.emplace(info->tcpi_reordering);
    metrics->packet_retx.emplace(info->tcpi_total_retrans);
    metrics->pacing_rate.emplace(info->tcpi_pacing_rate);
    metrics->data_notsent.emplace(info->tcpi_notsent_bytes);
    if (info->tcpi_min_rtt != UINT32_MAX) {
      metrics->min_rtt.emplace(info->tcpi_min_rtt);
    }
    metrics->packet_sent.emplace(info->tcpi_data_segs_out);
    metrics->delivery_rate.emplace(info->tcpi_delivery_rate);
    metrics->busy_usec.emplace(info->tcpi_busy_time);
    metrics->rwnd_limited_usec.emplace(info->tcpi_rwnd_limited);
    metrics->sndbuf_limited_usec.emplace(info->tcpi_sndbuf_limited);
  }
  if (info->length > offsetof(grpc_core::tcp_info, tcpi_dsack_dups)) {
    metrics->data_sent.emplace(info->tcpi_bytes_sent);
    metrics->data_retx.emplace(info->tcpi_bytes_retrans);
    metrics->packet_spurious_retx.emplace(info->tcpi_dsack_dups);
  }
}

/** Extracts opt stats from the given control message \a opt_stats to the
 * connection metrics \a metrics */
void extract_opt_stats_from_cmsg(ConnectionMetrics* metrics,
                                 const cmsghdr* opt_stats) {
  if (opt_stats == nullptr) {
    return;
  }
  const auto* data = CMSG_DATA(opt_stats);
  constexpr int64_t cmsg_hdr_len = CMSG_ALIGN(sizeof(struct cmsghdr));
  const int64_t len = opt_stats->cmsg_len - cmsg_hdr_len;
  int64_t offset = 0;

  while (offset < len) {
    const auto* attr = reinterpret_cast<const nlattr*>(data + offset);
    const void* val = data + offset + NLA_HDRLEN;
    switch (attr->nla_type) {
      case TCP_NLA_BUSY: {
        metrics->busy_usec.emplace(read_unaligned<uint64_t>(val));
        break;
      }
      case TCP_NLA_RWND_LIMITED: {
        metrics->rwnd_limited_usec.emplace(read_unaligned<uint64_t>(val));
        break;
      }
      case TCP_NLA_SNDBUF_LIMITED: {
        metrics->sndbuf_limited_usec.emplace(read_unaligned<uint64_t>(val));
        break;
      }
      case TCP_NLA_PACING_RATE: {
        metrics->pacing_rate.emplace(read_unaligned<uint64_t>(val));
        break;
      }
      case TCP_NLA_DELIVERY_RATE: {
        metrics->delivery_rate.emplace(read_unaligned<uint64_t>(val));
        break;
      }
      case TCP_NLA_DELIVERY_RATE_APP_LMT: {
        metrics->is_delivery_rate_app_limited.emplace(
            read_unaligned<uint8_t>(val));
        break;
      }
      case TCP_NLA_SND_CWND: {
        metrics->congestion_window.emplace(read_unaligned<uint32_t>(val));
        break;
      }
      case TCP_NLA_MIN_RTT: {
        metrics->min_rtt.emplace(read_unaligned<uint32_t>(val));
        break;
      }
      case TCP_NLA_SRTT: {
        metrics->srtt.emplace(read_unaligned<uint32_t>(val));
        break;
      }
      case TCP_NLA_RECUR_RETRANS: {
        metrics->recurring_retrans.emplace(read_unaligned<uint8_t>(val));
        break;
      }
      case TCP_NLA_BYTES_SENT: {
        metrics->data_sent.emplace(read_unaligned<uint64_t>(val));
        break;
      }
      case TCP_NLA_DATA_SEGS_OUT: {
        metrics->packet_sent.emplace(read_unaligned<uint64_t>(val));
        break;
      }
      case TCP_NLA_TOTAL_RETRANS: {
        metrics->packet_retx.emplace(read_unaligned<uint64_t>(val));
        break;
      }
      case TCP_NLA_DELIVERED: {
        metrics->packet_delivered.emplace(read_unaligned<uint32_t>(val));
        break;
      }
      case TCP_NLA_DELIVERED_CE: {
        metrics->packet_delivered_ce.emplace(read_unaligned<uint32_t>(val));
        break;
      }
      case TCP_NLA_BYTES_RETRANS: {
        metrics->data_retx.emplace(read_unaligned<uint64_t>(val));
        break;
      }
      case TCP_NLA_DSACK_DUPS: {
        metrics->packet_spurious_retx.emplace(read_unaligned<uint32_t>(val));
        break;
      }
      case TCP_NLA_REORDERING: {
        metrics->reordering.emplace(read_unaligned<uint32_t>(val));
        break;
      }
      case TCP_NLA_SND_SSTHRESH: {
        metrics->snd_ssthresh.emplace(read_unaligned<uint32_t>(val));
        break;
      }
    }
    offset += NLA_ALIGN(attr->nla_len);
  }
}

static int get_socket_tcp_info(grpc_core::tcp_info* info, int fd) {
  memset(info, 0, sizeof(*info));
  info->length = offsetof(grpc_core::tcp_info, length);
  return getsockopt(fd, IPPROTO_TCP, TCP_INFO, info, &(info->length));
}
} /* namespace */

void TracedBuffer::AddNewEntry(TracedBuffer** head, uint32_t seq_no, int fd,
                               void* arg) {
  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.time = gpr_now(GPR_CLOCK_REALTIME);
  new_elem->ts_.scheduled_time.time = gpr_inf_past(GPR_CLOCK_REALTIME);
  new_elem->ts_.sent_time.time = gpr_inf_past(GPR_CLOCK_REALTIME);
  new_elem->ts_.acked_time.time = gpr_inf_past(GPR_CLOCK_REALTIME);

  if (get_socket_tcp_info(&new_elem->ts_.info, fd) == 0) {
    extract_opt_stats_from_tcp_info(&new_elem->ts_.sendmsg_time.metrics,
                                    &new_elem->ts_.info);
  }
  if (*head == nullptr) {
    *head = new_elem;
    return;
  }
  /* Append at the end. */
  TracedBuffer* ptr = *head;
  while (ptr->next_ != nullptr) {
    ptr = ptr->next_;
  }
  ptr->next_ = new_elem;
}

void TracedBuffer::ProcessTimestamp(TracedBuffer** head,
{'fraction_non_alphanumeric': 0.1144701521155567, 'fraction_numerical': 0.007861434232391665, 'mean_word_length': 3.350013900472616, 'pattern_counts': {'":': 0, '<': 59, '<?xml version=': 0, '>': 209, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4312365', 'n_tokens_mistral': 5608, 'n_tokens_neox': 5519, 'n_words': 940}