<commit_before>/*
 * Farsight2 - Farsight TFRC implementation
 *
 * Copyright 2010 Collabora Ltd.
 *  @author: <NAME> <<EMAIL>>
 * Copyright 2010 Nokia Corp.
 *
 * tfrc.c - An implemention of TCP Friendly rate control, RFCs 5348 and 4828
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
 */

#include "tfrc.h"

#include <math.h>
#include <string.h>

/* for gst_util_uint64_scale_round */
#include <gst/gst.h>

/*
 * ALL TIMES ARE IN MILLISECONDS
 * bitrates are in bytes/sec
 */

#if 0
#define DEBUG_RECEIVER(receiver, format, ...) \
  g_debug ("TFRC-R (%p): " format , receiver,  __VA_ARGS__)
#else
#define DEBUG_RECEIVER(...)
#endif

#if 0
#define DEBUG_SENDER(sender, format, ...) \
  g_debug ("TFRC-S (%p): " format,  sender,  __VA_ARGS__)
#else
#define DEBUG_SENDER(...)
#endif

#define DEFAULT_MSS 1460

#define SECOND (1000)

/*
 * @s: segment size in bytes
 * @R: RTT in milli seconds (instead of seconds)
 * @p: loss event per packet transmitted
 *
 *                              s
 * X_Bps = -----------------------------------------------
 *         R * (sqrt(2*p/3) + 12*sqrt(3*p/8)*p*(1+32*p^2))
 *
 * Returns: The bitrate in bytes/s
 */
static gdouble
calculate_bitrate (gdouble s, gdouble R, gdouble p)
{
  gdouble f = sqrt (2 * p / 3) + 12 * sqrt (3 * p / 8) * p * (1 + 32 * p * p);

  return (SECOND * s) / (R * f);
}

#define RECEIVE_RATE_HISTORY_SIZE      (4)

struct ReceiveRateItem {
  guint64 timestamp;
  guint rate;
};

struct _TfrcSender {
  guint computed_rate; /* The rate computer from the TCP throughput equation */

  gboolean sp;
  guint header_size;
  guint average_packet_size; /* 16 times larger */
  gboolean use_inst_rate; /* use inst_rate instead of rate */

  guint mss; /* max segment size */
  guint rate; /* maximum allowed sending rate in bytes/sec */
  guint inst_rate; /* corrected maximum allowed sending rate */
  guint averaged_rtt;
  guint sqmean_rtt;
  guint64 tld; /* Time Last Doubled during slow-start */

  guint64 nofeedback_timer_expiry;

  guint retransmission_timeout; /* RTO */

  struct ReceiveRateItem receive_rate_history[RECEIVE_RATE_HISTORY_SIZE];

  gdouble last_loss_event_rate;

  gboolean sent_packet;
};

TfrcSender *
tfrc_sender_new (guint segment_size, guint64 now)
{
  TfrcSender *sender = g_slice_new0 (TfrcSender);

  /* initialized as described in RFC 5348 4.2 */
  sender->mss = DEFAULT_MSS;
  sender->average_packet_size = segment_size << 4;
  sender->rate = segment_size;

  sender->retransmission_timeout = 2 * SECOND;
  sender->nofeedback_timer_expiry = now + (2 * SECOND); /* 2 seconds */
  return sender;
}

TfrcSender *
tfrc_sender_new_sp (guint64 now, guint initial_average_packet_size)
{
  TfrcSender *sender = tfrc_sender_new (1460, now);

  sender->sp = TRUE;
  /* Specified in RFC 4828 Section 3 second bullet */
  sender->header_size = 40;
  sender->average_packet_size = initial_average_packet_size << 4;

  return sender;
}

void
tfrc_sender_use_inst_rate (TfrcSender *sender, gboolean use_inst_rate)
{
  sender->use_inst_rate = use_inst_rate;
}


void
tfrc_sender_free (TfrcSender *sender)
{
  g_slice_free (TfrcSender, sender);
}

static guint
sender_get_segment_size (TfrcSender *sender)
{
  if (sender->sp)
    return sender->mss;
  else
    return sender->average_packet_size >> 4;
}

void
tfrc_sender_on_first_rtt (TfrcSender *sender, guint64 now)
{
  sender->receive_rate_history[0].rate = G_MAXUINT;
  sender->receive_rate_history[0].timestamp = now;
}

static guint get_max_receive_rate (TfrcSender *sender, gboolean ignore_max_uint)
{
  guint max_rate = 0;
  guint i;

  for (i = 0; i < RECEIVE_RATE_HISTORY_SIZE; i++)
  {
    if (G_UNLIKELY (sender->receive_rate_history[i].rate == G_MAXUINT))
    {
      if (ignore_max_uint)
        return max_rate;
      else
        return G_MAXUINT;
    }
    max_rate = MAX (max_rate, sender->receive_rate_history[i].rate);
  }

  return max_rate;
}

static void
add_to_receive_rate_history (TfrcSender *sender, guint receive_rate,
    guint64 now)
{
  int i;

  for (i = RECEIVE_RATE_HISTORY_SIZE - 1; i > 0; i --)
    sender->receive_rate_history[i] = sender->receive_rate_history[i-1];

  sender->receive_rate_history[0].rate = receive_rate;
  sender->receive_rate_history[0].timestamp = now;
}

static guint
maximize_receive_rate_history (TfrcSender *sender, guint receive_rate,
    guint64 now)
{
  guint max_rate;

  DEBUG_SENDER (sender, "MAXIMIZE recv: %u", receive_rate);

  add_to_receive_rate_history (sender, receive_rate, now);

  max_rate = get_max_receive_rate (sender, TRUE);

  memset (sender->receive_rate_history, 0,
      sizeof(struct ReceiveRateItem) * RECEIVE_RATE_HISTORY_SIZE);

  sender->receive_rate_history[0].rate = max_rate;
  sender->receive_rate_history[0].timestamp = now;

  return max_rate;
}

static void
update_receive_rate_history (TfrcSender *sender, guint receive_rate,
    guint64 now)
{
  guint i;

  add_to_receive_rate_history (sender, receive_rate, now);

  for (i = 1; i < RECEIVE_RATE_HISTORY_SIZE; i++)
    if (sender->receive_rate_history[i].rate &&
        sender->receive_rate_history[i].timestamp <
        now - (2 * sender->averaged_rtt))
      sender->receive_rate_history[i].rate = 0;
}

const guint t_mbi = 64; /* the maximum backoff interval of 64 seconds */

static guint
compute_initial_rate (guint mss, guint rtt)
{
  if (G_UNLIKELY (rtt == 0))
    return 0;

  return (SECOND * MIN (4 * mss, MAX (2 * mss, 4380))) / rtt;
}

/* RFC 5348 section 4.3 step 4 second part */
static void
recompute_sending_rate (TfrcSender *sender, guint recv_limit,
    gdouble loss_event_rate, guint64 now)
{
  if (loss_event_rate > 0) {
    /* congestion avoidance phase */
    sender->computed_rate = calculate_bitrate (sender_get_segment_size (sender),
        sender->averaged_rtt, loss_event_rate);
    sender->rate = MAX (MIN (sender->computed_rate, recv_limit),
            sender_get_segment_size (sender)/t_mbi);
    DEBUG_SENDER (sender, "congestion avoidance: %u (computed: %u ss: %u)",
        sender->rate, sender->computed_rate, sender_get_segment_size (sender));
  } else if (now - sender->tld >= sender->averaged_rtt) {
    /* initial slow-start */
    sender->rate = MAX (MIN (2 * sender->rate, recv_limit),
        compute_initial_rate (sender->mss, sender->averaged_rtt));
    sender->tld = now;
    DEBUG_SENDER (sender, "initial slow start: %u", sender->rate);
  }
}

void
tfrc_sender_on_feedback_packet (TfrcSender *sender, guint64 now,
    guint rtt, guint receive_rate, gdouble loss_event_rate,
    gboolean is_data_limited)
{
  guint recv_limit; /* the limit on the sending rate computed from X_recv_set */

  g_return_if_fail (rtt > 0 && rtt <= 10 * SECOND);

  /* On first feedback packet, set he rate based on the mss and rtt */
  if (sender->tld == 0) {
    sender->rate = compute_initial_rate (sender->mss, rtt);
    sender->tld = now;
    DEBUG_SENDER (sender, "fb: initial rate: %u", sender->rate);
  }

  /* Apply the steps from RFC 5348 section 4.3 */

  /* Step 1 (calculating the rtt) is done before calling this function */

  /* Step 2: Update the RTT */
  if (sender->averaged_rtt == 0)
    sender->averaged_rtt = rtt;
  else
    sender->averaged_rtt = (sender->averaged_rtt * 9 + rtt) / 10;

  if (sender->averaged_rtt == 0)
    sender->averaged_rtt = 1;

  /* Step 3: Update the timeout interval */
  sender->retransmission_timeout = MAX (4 * sender->averaged_rtt,
      SECOND * 2 * sender_get_segment_size (sender) / sender->rate );

  /* Step 4: Update the allowed sending rate */


  if (G_UNLIKELY (is_data_limited)) {
    /* the entire interval covered by the feedback packet
       was a data-limited interval */
    if (loss_event_rate > sender->last_loss_event_rate) {
      guint i;
      /* the feedback packet reports a new loss event or an
         increase in the loss event rate p */

      /* Halve entries in the receive rate history */
      for (i = 0; i < RECEIVE_RATE_HISTORY_SIZE; i++)
        sender->receive_rate_history[i].rate /= 2;

      receive_rate *= 0.85;

      recv_limit = maximize_receive_rate_history (sender, receive_rate,
          now);
      DEBUG_SENDER (sender, "fb: data limited, new loss event %f > %f,"
          " recv_limit: %u", loss_event_rate, sender->last_loss_event_rate,
          recv_limit);
    } else {
      recv_limit = 2 * maximize_receive_rate_history (sender, receive_rate,
          now);
      DEBUG_SENDER (sender, "fb: data limited, no new loss event %f <= %f,"
          " recv_limit: %u", loss_event_rate, sender->last_loss_event_rate,
          recv_limit);
    }
  } else {
    /* typical behavior */
    update_receive_rate_history (sender, receive_rate, now);
    recv_limit = get_max_receive_rate (sender, FALSE);
    if (recv_limit < G_MAXUINT / 2)
      recv_limit *= 2;
    else
      recv_limit = G_MAXUINT;
    DEBUG_SENDER (sender, "fb: not data limited, recv_limit: %u",
        recv_limit);
  }

  recompute_sending_rate (sender, recv_limit, loss_event_rate, now);

  /* Step 5: calculate the instantaneous
     transmit rate, X_inst, following Section 4.5.
  */

  if (sender->sqmean_rtt)
    sender->sqmean_rtt = 0.9 * sender->sqmean_rtt + sqrt(rtt) / 10;
  else
    sender->sqmean_rtt = sqrt(rtt);

  sender->inst_rate = sender->rate * sender->sqmean_rtt / sqrt(rtt);
  if (sender->inst_rate < sender_get_segment_size (sender) / t_mbi)
    sender->inst_rate = sender_get_segment_size (sender) / t_mbi;

  /* Step 6: Reset the nofeedback timer to expire after RTO seconds. */

  sender->nofeedback_timer_expiry = now + sender->retransmission_timeout;
  sender->sent_packet = FALSE;

  sender->last_loss_event_rate = loss_event_rate;
}

static void
update_limits(TfrcSender *sender, guint timer_limit, guint64 now)
{
  if (timer_limit < sender_get_segment_size (sender) / t_mbi)
    timer_limit = sender_get_segment_size (sender) / t_mbi;

  memset (sender->receive_rate_history, 0,
      sizeof(struct ReceiveRateItem) * RECEIVE_RATE_HISTORY_SIZE);

  sender->receive_rate_history[0].rate = timer_limit / 2;
  sender->receive_rate_history[0].timestamp = now;

  recompute_sending_rate (sender, timer_limit,
      sender->last_loss_event_rate, now);
}


void
tfrc_sender_no_feedback_timer_expired (TfrcSender *sender, guint64 now)
{
  guint receive_rate = get_max_receive_rate (sender, FALSE);
  guint recover_rate = compute_initial_rate (sender->mss, sender->averaged_rtt);

  if (sender->averaged_rtt == 0 && sender->sent_packet) {
    /* We do not have X_Bps or recover_rate yet.
     * Halve the allowed sending rate.
     */

    sender->rate = MAX ( sender->rate / 2,
        sender_get_segment_size (sender) / t_mbi);
    DEBUG_SENDER (sender, "no_fb: no p, initial, halve rate: %u", sender->rate);
  } else if (((sender->last_loss_event_rate > 0 &&
              receive_rate < recover_rate) ||
          (sender->last_loss_event_rate == 0 &&
              sender->rate < 2 * recover_rate)) &&
      !sender->sent_packet) {
    /* Don't halve the allowed sending rate. */
    /* do nothing */
    DEBUG_SENDER (sender, "no_fb: idle, do nothing recv: %u recover: %u",
        receive_rate, recover_rate);
  } else if (sender->last_loss_event_rate == 0) {
    /* We do not have X_Bps yet.
     * Halve the allowed sending rate.
     */
    sender->rate = MAX ( sender->rate / 2,
        sender_get_segment_size (sender) / t_mbi);
    DEBUG_SENDER (sender, "no_fb: no p, halve rate: %u recover: %u, sent: %u", sender->rate,
        recover_rate, sender->sent_packet);
  } else if (sender->computed_rate / 2 > receive_rate) {
    /* 2 * X_recv was already limiting the sending rate.
     * Halve the allowed sending rate.
   */
    DEBUG_SENDER (sender, "no_fb: computed rate %u > 2 * recv_rate %u",
        sender->computed_rate, receive_rate);
    update_limits(sender, receive_rate, now);
  } else {
    DEBUG_SENDER (sender, "no_fb: ELSE computed: %u", sender->computed_rate);
    update_limits(sender, sender->computed_rate / 2, now);
  }

  g_assert (sender->rate != 0);

  sender->nofeedback_timer_expiry = now + MAX ( 4 * sender->averaged_rtt,
      SECOND * 2 * sender_get_segment_size (sender) / sender->rate);
  sender->sent_packet = FALSE;
}

void
tfrc_sender_sending_packet (TfrcSender *sender, guint size)
{
  /* this should be:
   * avg = size + (avg * 15/16)
   */
  sender->average_packet_size =
      size + ((15 * sender->average_packet_size) >> 4);

  sender->sent_packet = TRUE;
}

guint
tfrc_sender_get_send_rate (TfrcSender *sender)
{
  guint rate;

  if (!sender)
    return DEFAULT_MSS;

  if (sender->use_inst_rate)
    rate = sender->inst_rate;
  else
    rate = sender->rate;

  if (sender->sp)
    return rate * (sender->average_packet_size >> 4) /
        ((sender->average_packet_size >> 4) + sender->header_size);
  else
    return rate;
}

guint64
tfrc_sender_get_no_feedback_timer_expiry (TfrcSender *sender)
{
  return sender->nofeedback_timer_expiry;
}

guint
tfrc_sender_get_averaged_rtt (TfrcSender *sender)
{
  return sender->averaged_rtt;
}


#define NDUPACK 3 /* Number of packets to receive after a loss before declaring the loss event */
#define LOSS_EVENTS_MAX (9)
#define LOSS_INTERVALS_MAX (8)
#define MAX_HISTORY_SIZE (LOSS_EVENTS_MAX * 2) /* 2 is a random number */
#define MIN_HISTORY_DURATION (10)

typedef struct  {
  guint64 first_timestamp;
  guint first_seqnum;
  guint64 first_recvtime;

  guint64 last_timestamp;
  guint last_seqnum;
  guint64 last_recvtime;
} ReceivedInterval;

struct _TfrcReceiver {
  GQueue received_intervals;

  gboolean sp;

  guint sender_rtt;
  guint receive_rate;
  guint max_receive_rate;
  guint max_receive_rate_ss;
  guint64 feedback_timer_expiry;

  guint first_loss_interval;

  gdouble loss_event_rate;

  gboolean feedback_sent_on_last_timer;

  guint received_bytes;
  guint prev_received_bytes;
  guint64 received_bytes_reset_time;
  guint64 prev_received_bytes_reset_time;
  guint received_packets;
  guint prev_received_packets;
  guint sender_rtt_on_last_feedback;
};

TfrcReceiver *
tfrc_receiver_new (guint64 now)
{
  TfrcReceiver *receiver = g_slice_new0 (TfrcReceiver);

  g_queue_init (&receiver->received_intervals);
  receiver->received_bytes_reset_time = now;
  receiver->prev_received_bytes_reset_time = now;

  return receiver;
}


TfrcReceiver *
tfrc_receiver_new_sp (guint64 now)
{
  TfrcReceiver *receiver = tfrc_receiver_new (now);

  receiver->sp = TRUE;

  return receiver;
}

void
tfrc_receiver_free (TfrcReceiver *receiver)
{
  ReceivedInterval *ri;

  while ((ri = g_queue_pop_tail (&receiver->received_intervals)))
    g_slice_free (ReceivedInterval, ri);

  g_slice_free (TfrcReceiver, receiver);
}

/*
 * @s:  segment size in bytes
 * @R: RTT in milli seconds (instead of seconds)
 * @rate: the sending rate
 *
 * Returns the 1/p that would produce this sending rate
 * This is used to compute the first loss interval
 */

static gdouble
compute_first_loss_interval (gdouble s, gdouble R, gdouble rate)
{
  gdouble p_min = 0;
  gdouble p_max = 1;
  gdouble p;
  gdouble computed_rate;

  /* Use an iterative process to find p
   * it would be faster to use a table, but I'm lazy
   */

  do {
    p = (p_min + p_max) / 2;
    computed_rate = calculate_bitrate (s, R, p);

    if (computed_rate < rate)
      p_max = p;
    else
      p_min = p;

  } while (computed_rate < 0.95 * rate || computed_rate > 1.05 * rate);

  return 1 / p;
}


/* Implements RFC 5348 section 5 */
static gdouble
calculate_loss_event_rate (TfrcReceiver *receiver, guint64 now)
{
  guint64 loss_event_times[LOSS_EVENTS_MAX];
  guint loss_event_seqnums[LOSS_EVENTS_MAX];
  guint loss_event_pktcount[LOSS_EVENTS_MAX];
  guint loss_intervals[LOSS_EVENTS_MAX];
  const gdouble weights[8] = { 1.0, 1.0, 1.0, 1.0, 0.8, 0.6, 0.4, 0.2 };
  gint max_index = -1;
  GList *item;
  guint max_seqnum = 0;
  gint i;
  guint max_interval;
  gdouble I_tot0 = 0;
  gdouble I_tot1 = 0;
  gdouble W_tot = 0;
  gdouble I_tot;

  if (receiver->sender_rtt == 0)
    return 0;

  if (receiver->received_intervals.length < 2)
    return 0;

  DEBUG_RECEIVER (receiver, "start loss event rate computation (rtt: %u)",
      receiver->sender_rtt);

  for (item = g_queue_peek_head_link (&receiver->received_intervals)->next;
       item;
       item = item->next) {
    ReceivedInterval *current = item->data;
    ReceivedInterval *prev = item->prev->data;
    guint64 start_ts;
    guint start_seqnum;

    max_seqnum = current->last_seqnum;

    DEBUG_RECEIVER (receiver, "Loss: ts %"G_GUINT64_FORMAT
        "->%"G_GUINT64_FORMAT" seq %"G_GUINT64_FORMAT"->%"G_GUINT64_FORMAT,
        prev->last_timestamp, current->first_timestamp, prev->last_seqnum,
        current->first_seqnum);

    /* If the current loss is entirely within one RTT of the beginning of the
     * last loss, lets merge it into there
     */
    if (max_index >= 0 && current->first_timestamp <
        loss_event_times[max_index % LOSS_EVENTS_MAX] + receiver->sender_rtt) {
      loss_event_pktcount[max_index % LOSS_EVENTS_MAX] +=
          current->first_seqnum - prev->last_seqnum;
      DEBUG_RECEIVER (receiver, "Merged: pktcount[%u] = %u", max_index,
          loss_event_pktcount[max_index % LOSS_EVENTS_MAX]);
      continue;
    }

    if (max_index >= 0 && prev->last_timestamp <
        loss_event_times[max_index % LOSS_EVENTS_MAX] + receiver->sender_rtt) {
      /* This is the case where a loss event ends in a middle of an interval
       * without packets, then we close this loss event and start a new one
       */
      start_ts = loss_event_times[max_index % LOSS_EVENTS_MAX] +
          receiver->sender_rtt;
      start_seqnum = prev->last_seqnum +
          gst_util_uint64_scale_round (
            current->first_seqnum - prev->last_seqnum,
            start_ts - prev->last_timestamp,
            1 + current->first_timestamp - prev->last_timestamp);
      loss_event_pktcount[max_index % LOSS_EVENTS_MAX] +=
          start_seqnum - prev->last_seqnum - 1;
      DEBUG_RECEIVER (receiver,
          "Loss ends inside loss interval pktcount[%u] = %u",
          max_index, loss_event_pktcount[max_index % LOSS_EVENTS_MAX]);
    } else {
      /* this is the case where the packet loss starts an entirely new loss
       * event
       */
      start_ts = prev->last_timestamp +
          gst_util_uint64_scale_round (1,
              current->first_timestamp - prev->last_timestamp,
              current->first_seqnum - prev->last_seqnum);
      start_seqnum = prev->last_seqnum + 1;
    }

    DEBUG_RECEIVER (receiver, "start_ts: %u seqnum: %u", start_ts,
        start_seqnum);

    /* Now we have one or more loss events that start
     * during this interval of lost packets, if there is more than one
     * all but the last one are of RTT length
     */
    while (start_ts <= current->first_timestamp) {
      max_index ++;

      loss_event_times[max_index % LOSS_EVENTS_MAX] = start_ts;
      loss_event_seqnums[max_index % LOSS_EVENTS_MAX] = start_seqnum;
      if (current->first_timestamp == prev->last_timestamp) {
        /* if current->first_ts == prev->last_ts,<commit_msg>int64 in tfrc<commit_after>/*
 * Farsight2 - Farsight TFRC implementation
 *
 * Copyright 2010 Collabora Ltd.
 *  @author: <NAME> <<EMAIL>>
 * Copyright 2010 Nokia Corp.
 *
 * tfrc.c - An implemention of TCP Friendly rate control, RFCs 5348 and 4828
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
 */

#include "tfrc.h"

#include <math.h>
#include <string.h>

/* for gst_util_uint64_scale_round */
#include <gst/gst.h>

/*
 * ALL TIMES ARE IN MILLISECONDS
 * bitrates are in bytes/sec
 */

#if 0
#define DEBUG_RECEIVER(receiver, format, ...) \
  g_debug ("TFRC-R (%p): " format , receiver,  __VA_ARGS__)
#else
#define DEBUG_RECEIVER(...)
#endif

#if 0
#define DEBUG_SENDER(sender, format, ...) \
  g_debug ("TFRC-S (%p): " format,  sender,  __VA_ARGS__)
#else
#define DEBUG_SENDER(...)
#endif

#define DEFAULT_MSS 1460

#define SECOND (1000)

/*
 * @s: segment size in bytes
 * @R: RTT in milli seconds (instead of seconds)
 * @p: loss event per packet transmitted
 *
 *                              s
 * X_Bps = -----------------------------------------------
 *         R * (sqrt(2*p/3) + 12*sqrt(3*p/8)*p*(1+32*p^2))
 *
 * Returns: The bitrate in bytes/s
 */
static gdouble
calculate_bitrate (gdouble s, gdouble R, gdouble p)
{
  gdouble f = sqrt (2 * p / 3) + 12 * sqrt (3 * p / 8) * p * (1 + 32 * p * p);

  return (SECOND * s) / (R * f);
}

#define RECEIVE_RATE_HISTORY_SIZE      (4)

struct ReceiveRateItem {
  guint64 timestamp;
  guint rate;
};

struct _TfrcSender {
  guint computed_rate; /* The rate computer from the TCP throughput equation */

  gboolean sp;
  guint header_size;
  guint average_packet_size; /* 16 times larger */
  gboolean use_inst_rate; /* use inst_rate instead of rate */

  guint mss; /* max segment size */
  guint rate; /* maximum allowed sending rate in bytes/sec */
  guint inst_rate; /* corrected maximum allowed sending rate */
  guint averaged_rtt;
  guint sqmean_rtt;
  guint64 tld; /* Time Last Doubled during slow-start */

  guint64 nofeedback_timer_expiry;

  guint retransmission_timeout; /* RTO */

  struct ReceiveRateItem receive_rate_history[RECEIVE_RATE_HISTORY_SIZE];

  gdouble last_loss_event_rate;

  gboolean sent_packet;
};

TfrcSender *
tfrc_sender_new (guint segment_size, guint64 now)
{
  TfrcSender *sender = g_slice_new0 (TfrcSender);

  /* initialized as described in RFC 5348 4.2 */
  sender->mss = DEFAULT_MSS;
  sender->average_packet_size = segment_size << 4;
  sender->rate = segment_size;

  sender->retransmission_timeout = 2 * SECOND;
  sender->nofeedback_timer_expiry = now + (2 * SECOND); /* 2 seconds */
  return sender;
}

TfrcSender *
tfrc_sender_new_sp (guint64 now, guint initial_average_packet_size)
{
  TfrcSender *sender = tfrc_sender_new (1460, now);

  sender->sp = TRUE;
  /* Specified in RFC 4828 Section 3 second bullet */
  sender->header_size = 40;
  sender->average_packet_size = initial_average_packet_size << 4;

  return sender;
}

void
tfrc_sender_use_inst_rate (TfrcSender *sender, gboolean use_inst_rate)
{
  sender->use_inst_rate = use_inst_rate;
}


void
tfrc_sender_free (TfrcSender *sender)
{
  g_slice_free (TfrcSender, sender);
}

static guint
sender_get_segment_size (TfrcSender *sender)
{
  if (sender->sp)
    return sender->mss;
  else
    return sender->average_packet_size >> 4;
}

void
tfrc_sender_on_first_rtt (TfrcSender *sender, guint64 now)
{
  sender->receive_rate_history[0].rate = G_MAXUINT;
  sender->receive_rate_history[0].timestamp = now;
}

static guint get_max_receive_rate (TfrcSender *sender, gboolean ignore_max_uint)
{
  guint max_rate = 0;
  guint i;

  for (i = 0; i < RECEIVE_RATE_HISTORY_SIZE; i++)
  {
    if (G_UNLIKELY (sender->receive_rate_history[i].rate == G_MAXUINT))
    {
      if (ignore_max_uint)
        return max_rate;
      else
        return G_MAXUINT;
    }
    max_rate = MAX (max_rate, sender->receive_rate_history[i].rate);
  }

  return max_rate;
}

static void
add_to_receive_rate_history (TfrcSender *sender, guint receive_rate,
    guint64 now)
{
  int i;

  for (i = RECEIVE_RATE_HISTORY_SIZE - 1; i > 0; i --)
    sender->receive_rate_history[i] = sender->receive_rate_history[i-1];

  sender->receive_rate_history[0].rate = receive_rate;
  sender->receive_rate_history[0].timestamp = now;
}

static guint
maximize_receive_rate_history (TfrcSender *sender, guint receive_rate,
    guint64 now)
{
  guint max_rate;

  DEBUG_SENDER (sender, "MAXIMIZE recv: %u", receive_rate);

  add_to_receive_rate_history (sender, receive_rate, now);

  max_rate = get_max_receive_rate (sender, TRUE);

  memset (sender->receive_rate_history, 0,
      sizeof(struct ReceiveRateItem) * RECEIVE_RATE_HISTORY_SIZE);

  sender->receive_rate_history[0].rate = max_rate;
  sender->receive_rate_history[0].timestamp = now;

  return max_rate;
}

static void
update_receive_rate_history (TfrcSender *sender, guint receive_rate,
    guint64 now)
{
  guint i;

  add_to_receive_rate_history (sender, receive_rate, now);

  for (i = 1; i < RECEIVE_RATE_HISTORY_SIZE; i++)
    if (sender->receive_rate_history[i].rate &&
        sender->receive_rate_history[i].timestamp <
        now - (2 * sender->averaged_rtt))
      sender->receive_rate_history[i].rate = 0;
}

const guint t_mbi = 64; /* the maximum backoff interval of 64 seconds */

static guint
compute_initial_rate (guint mss, guint rtt)
{
  if (G_UNLIKELY (rtt == 0))
    return 0;

  return (SECOND * MIN (4 * mss, MAX (2 * mss, 4380))) / rtt;
}

/* RFC 5348 section 4.3 step 4 second part */
static void
recompute_sending_rate (TfrcSender *sender, guint recv_limit,
    gdouble loss_event_rate, guint64 now)
{
  if (loss_event_rate > 0) {
    /* congestion avoidance phase */
    sender->computed_rate = calculate_bitrate (sender_get_segment_size (sender),
        sender->averaged_rtt, loss_event_rate);
    sender->rate = MAX (MIN (sender->computed_rate, recv_limit),
            sender_get_segment_size (sender)/t_mbi);
    DEBUG_SENDER (sender, "congestion avoidance: %u (computed: %u ss: %u)",
        sender->rate, sender->computed_rate, sender_get_segment_size (sender));
  } else if (now - sender->tld >= sender->averaged_rtt) {
    /* initial slow-start */
    sender->rate = MAX (MIN (2 * sender->rate, recv_limit),
        compute_initial_rate (sender->mss, sender->averaged_rtt));
    sender->tld = now;
    DEBUG_SENDER (sender, "initial slow start: %u", sender->rate);
  }
}

void
tfrc_sender_on_feedback_packet (TfrcSender *sender, guint64 now,
    guint rtt, guint receive_rate, gdouble loss_event_rate,
    gboolean is_data_limited)
{
  guint recv_limit; /* the limit on the sending rate computed from X_recv_set */

  g_return_if_fail (rtt > 0 && rtt <= 10 * SECOND);

  /* On first feedback packet, set he rate based on the mss and rtt */
  if (sender->tld == 0) {
    sender->rate = compute_initial_rate (sender->mss, rtt);
    sender->tld = now;
    DEBUG_SENDER (sender, "fb: initial rate: %u", sender->rate);
  }

  /* Apply the steps from RFC 5348 section 4.3 */

  /* Step 1 (calculating the rtt) is done before calling this function */

  /* Step 2: Update the RTT */
  if (sender->averaged_rtt == 0)
    sender->averaged_rtt = rtt;
  else
    sender->averaged_rtt = (sender->averaged_rtt * 9 + rtt) / 10;

  if (sender->averaged_rtt == 0)
    sender->averaged_rtt = 1;

  /* Step 3: Update the timeout interval */
  sender->retransmission_timeout = MAX (4 * sender->averaged_rtt,
      SECOND * 2 * sender_get_segment_size (sender) / sender->rate );

  /* Step 4: Update the allowed sending rate */


  if (G_UNLIKELY (is_data_limited)) {
    /* the entire interval covered by the feedback packet
       was a data-limited interval */
    if (loss_event_rate > sender->last_loss_event_rate) {
      guint i;
      /* the feedback packet reports a new loss event or an
         increase in the loss event rate p */

      /* Halve entries in the receive rate history */
      for (i = 0; i < RECEIVE_RATE_HISTORY_SIZE; i++)
        sender->receive_rate_history[i].rate /= 2;

      receive_rate *= 0.85;

      recv_limit = maximize_receive_rate_history (sender, receive_rate,
          now);
      DEBUG_SENDER (sender, "fb: data limited, new loss event %f > %f,"
          " recv_limit: %u", loss_event_rate, sender->last_loss_event_rate,
          recv_limit);
    } else {
      recv_limit = 2 * maximize_receive_rate_history (sender, receive_rate,
          now);
      DEBUG_SENDER (sender, "fb: data limited, no new loss event %f <= %f,"
          " recv_limit: %u", loss_event_rate, sender->last_loss_event_rate,
          recv_limit);
    }
  } else {
    /* typical behavior */
    update_receive_rate_history (sender, receive_rate, now);
    recv_limit = get_max_receive_rate (sender, FALSE);
    if (recv_limit < G_MAXUINT / 2)
      recv_limit *= 2;
    else
      recv_limit = G_MAXUINT;
    DEBUG_SENDER (sender, "fb: not data limited, recv_limit: %u",
        recv_limit);
  }

  recompute_sending_rate (sender, recv_limit, loss_event_rate, now);

  /* Step 5: calculate the instantaneous
     transmit rate, X_inst, following Section 4.5.
  */

  if (sender->sqmean_rtt)
    sender->sqmean_rtt = 0.9 * sender->sqmean_rtt + sqrt(rtt) / 10;
  else
    sender->sqmean_rtt = sqrt(rtt);

  sender->inst_rate = sender->rate * sender->sqmean_rtt / sqrt(rtt);
  if (sender->inst_rate < sender_get_segment_size (sender) / t_mbi)
    sender->inst_rate = sender_get_segment_size (sender) / t_mbi;

  /* Step 6: Reset the nofeedback timer to expire after RTO seconds. */

  sender->nofeedback_timer_expiry = now + sender->retransmission_timeout;
  sender->sent_packet = FALSE;

  sender->last_loss_event_rate = loss_event_rate;
}

static void
update_limits(TfrcSender *sender, guint timer_limit, guint64 now)
{
  if (timer_limit < sender_get_segment_size (sender) / t_mbi)
    timer_limit = sender_get_segment_size (sender) / t_mbi;

  memset (sender->receive_rate_history, 0,
      sizeof(struct ReceiveRateItem) * RECEIVE_RATE_HISTORY_SIZE);

  sender->receive_rate_history[0].rate = timer_limit / 2;
  sender->receive_rate_history[0].timestamp = now;

  recompute_sending_rate (sender, timer_limit,
      sender->last_loss_event_rate, now);
}


void
tfrc_sender_no_feedback_timer_expired (TfrcSender *sender, guint64 now)
{
  guint receive_rate = get_max_receive_rate (sender, FALSE);
  guint recover_rate = compute_initial_rate (sender->mss, sender->averaged_rtt);

  if (sender->averaged_rtt == 0 && sender->sent_packet) {
    /* We do not have X_Bps or recover_rate yet.
     * Halve the allowed sending rate.
     */

    sender->rate = MAX ( sender->rate / 2,
        sender_get_segment_size (sender) / t_mbi);
    DEBUG_SENDER (sender, "no_fb: no p, initial, halve rate: %u", sender->rate);
  } else if (((sender->last_loss_event_rate > 0 &&
              receive_rate < recover_rate) ||
          (sender->last_loss_event_rate == 0 &&
              sender->rate < 2 * recover_rate)) &&
      !sender->sent_packet) {
    /* Don't halve the allowed sending rate. */
    /* do nothing */
    DEBUG_SENDER (sender, "no_fb: idle, do nothing recv: %u recover: %u",
        receive_rate, recover_rate);
  } else if (sender->last_loss_event_rate == 0) {
    /* We do not have X_Bps yet.
     * Halve the allowed sending rate.
     */
    sender->rate = MAX ( sender->rate / 2,
        sender_get_segment_size (sender) / t_mbi);
    DEBUG_SENDER (sender, "no_fb: no p, halve rate: %u recover: %u, sent: %u", sender->rate,
        recover_rate, sender->sent_packet);
  } else if (sender->computed_rate / 2 > receive_rate) {
    /* 2 * X_recv was already limiting the sending rate.
     * Halve the allowed sending rate.
   */
    DEBUG_SENDER (sender, "no_fb: computed rate %u > 2 * recv_rate %u",
        sender->computed_rate, receive_rate);
    update_limits(sender, receive_rate, now);
  } else {
    DEBUG_SENDER (sender, "no_fb: ELSE computed: %u", sender->computed_rate);
    update_limits(sender, sender->computed_rate / 2, now);
  }

  g_assert (sender->rate != 0);

  sender->nofeedback_timer_expiry = now + MAX ( 4 * sender->averaged_rtt,
      SECOND * 2 * sender_get_segment_size (sender) / sender->rate);
  sender->sent_packet = FALSE;
}

void
tfrc_sender_sending_packet (TfrcSender *sender, guint size)
{
  /* this should be:
   * avg = size + (avg * 15/16)
   */
  sender->average_packet_size =
      size + ((15 * sender->average_packet_size) >> 4);

  sender->sent_packet = TRUE;
}

guint
tfrc_sender_get_send_rate (TfrcSender *sender)
{
  guint rate;

  if (!sender)
    return DEFAULT_MSS;

  if (sender->use_inst_rate)
    rate = sender->inst_rate;
  else
    rate = sender->rate;

  if (sender->sp)
    return rate * (sender->average_packet_size >> 4) /
        ((sender->average_packet_size >> 4) + sender->header_size);
  else
    return rate;
}

guint64
tfrc_sender_get_no_feedback_timer_expiry (TfrcSender *sender)
{
  return sender->nofeedback_timer_expiry;
}

guint
tfrc_sender_get_averaged_rtt (TfrcSender *sender)
{
  return sender->averaged_rtt;
}


#define NDUPACK 3 /* Number of packets to receive after a loss before declaring the loss event */
#define LOSS_EVENTS_MAX (9)
#define LOSS_INTERVALS_MAX (8)
#define MAX_HISTORY_SIZE (LOSS_EVENTS_MAX * 2) /* 2 is a random number */
#define MIN_HISTORY_DURATION (10)

typedef struct  {
  guint64 first_timestamp;
  guint first_seqnum;
  guint64 first_recvtime;

  guint64 last_timestamp;
  guint last_seqnum;
  guint64 last_recvtime;
} ReceivedInterval;

struct _TfrcReceiver {
  GQueue received_intervals;

  gboolean sp;

  guint sender_rtt;
  guint receive_rate;
  guint max_receive_rate;
  guint max_receive_rate_ss;
  guint64 feedback_timer_expiry;

  guint first_loss_interval;

  gdouble loss_event_rate;

  gboolean feedback_sent_on_last_timer;

  guint received_bytes;
  guint prev_received_bytes;
  guint64 received_bytes_reset_time;
  guint64 prev_received_bytes_reset_time;
  guint received_packets;
  guint prev_received_packets;
  guint sender_rtt_on_last_feedback;
};

TfrcReceiver *
tfrc_receiver_new (guint64 now)
{
  TfrcReceiver *receiver = g_slice_new0 (TfrcReceiver);

  g_queue_init (&receiver->received_intervals);
  receiver->received_bytes_reset_time = now;
  receiver->prev_received_bytes_reset_time = now;

  return receiver;
}


TfrcReceiver *
tfrc_receiver_new_sp (guint64 now)
{
  TfrcReceiver *receiver = tfrc_receiver_new (now);

  receiver->sp = TRUE;

  return receiver;
}

void
tfrc_receiver_free (TfrcReceiver *receiver)
{
  ReceivedInterval *ri;

  while ((ri = g_queue_pop_tail (&receiver->received_intervals)))
    g_slice_free (ReceivedInterval, ri);

  g_slice_free (TfrcReceiver, receiver);
}

/*
 * @s:  segment size in bytes
 * @R: RTT in milli seconds (instead of seconds)
 * @rate: the sending rate
 *
 * Returns the 1/p that would produce this sending rate
 * This is used to compute the first loss interval
 */

static gdouble
compute_first_loss_interval (gdouble s, gdouble R, gdouble rate)
{
  gdouble p_min = 0;
  gdouble p_max = 1;
  gdouble p;
  gdouble computed_rate;

  /* Use an iterative process to find p
   * it would be faster to use a table, but I'm lazy
   */

  do {
    p = (p_min + p_max) / 2;
    computed_rate = calculate_bitrate (s, R, p);

    if (computed_rate < rate)
      p_max = p;
    else
      p_min = p;

  } while (computed_rate < 0.95 * rate || computed_rate > 1.05 * rate);

  return 1 / p;
}


/* Implements RFC 5348 section 5 */
static gdouble
calculate_loss_event_rate (TfrcReceiver *receiver, guint64 now)
{
  guint64 loss_event_times[LOSS_EVENTS_MAX];
  guint loss_event_seqnums[LOSS_EVENTS_MAX];
  guint loss_event_pktcount[LOSS_EVENTS_MAX];
  guint loss_intervals[LOSS_EVENTS_MAX];
  const gdouble weights[8] = { 1.0, 1.0, 1.0, 1.0, 0.8, 0.6, 0.4, 0.2 };
  gint max_index = -1;
  GList *item;
  guint max_seqnum = 0;
  gint i;
  guint max_interval;
  gdouble I_tot0 = 0;
  gdouble I_tot1 = 0;
  gdouble W_tot = 0;
  gdouble I_tot;

  if (receiver->sender_rtt == 0)
    return 0;

  if (receiver->received_intervals.length < 2)
    return 0;

  DEBUG_RECEIVER (receiver, "start loss event rate computation (rtt: %u)",
      receiver->sender_rtt);

  for (item = g_queue_peek_head_link (&receiver->received_intervals)->next;
       item;
       item = item->next) {
    ReceivedInterval *current = item->data;
    ReceivedInterval *prev = item->prev->data;
    guint64 start_ts;
    guint start_seqnum;

    max_seqnum = current->last_seqnum;

    DEBUG_RECEIVER (receiver, "Loss: ts %"G_GUINT64_FORMAT
        "->%"G_GUINT64_FORMAT" seq %u->%u",
        prev->last_timestamp, current->first_timestamp, prev->last_seqnum,
        current->first_seqnum);

    /* If the current loss is entirely within one RTT of the beginning of the
     * last loss, lets merge it into there
     */
    if (max_index >= 0 && current->first_timestamp <
        loss_event_times[max_index % LOSS_EVENTS_MAX] + receiver->sender_rtt) {
      loss_event_pktcount[max_index % LOSS_EVENTS_MAX] +=
          current->first_seqnum - prev->last_seqnum;
      DEBUG_RECEIVER (receiver, "Merged: pktcount[%u] = %u", max_index,
          loss_event_pktcount[max_index % LOSS_EVENTS_MAX]);
      continue;
    }

    if (max_index >= 0 && prev->last_timestamp <
        loss_event_times[max_index % LOSS_EVENTS_MAX] + receiver->sender_rtt) {
      /* This is the case where a loss event ends in a middle of an interval
       * without packets, then we close this loss event and start a new one
       */
      start_ts = loss_event_times[max_index % LOSS_EVENTS_MAX] +
          receiver->sender_rtt;
      start_seqnum = prev->last_seqnum +
          gst_util_uint64_scale_round (
            current->first_seqnum - prev->last_seqnum,
            start_ts - prev->last_timestamp,
            1 + current->first_timestamp - prev->last_timestamp);
      loss_event_pktcount[max_index % LOSS_EVENTS_MAX] +=
          start_seqnum - prev->last_seqnum - 1;
      DEBUG_RECEIVER (receiver,
          "Loss ends inside loss interval pktcount[%u] = %u",
          max_index, loss_event_pktcount[max_index % LOSS_EVENTS_MAX]);
    } else {
      /* this is the case where the packet loss starts an entirely new loss
       * event
       */
      start_ts = prev->last_timestamp +
          gst_util_uint64_scale_round (1,
              current->first_timestamp - prev->last_timestamp,
              current->first_seqnum - prev->last_seqnum);
      start_seqnum = prev->last_seqnum + 1;
    }

    DEBUG_RECEIVER (receiver, "start_ts: %" G_GUINT64_FORMAT " seqnum: %u",
        start_ts, start_seqnum);

    /* Now we have one or more loss events that start
     * during this interval of lost packets, if there is more than one
     * all but the last one are of RTT length
     */
    while (start_ts <= current->first_timestamp) {
      max_index ++;

      loss_event_times[max_index % LOSS_EVENTS_MAX] = start_ts;
      loss_event_seqnums[max_index % LOSS_EVENTS_MAX] = start_seqnum;
      if (current->first_timestamp == prev->last_timestamp) {
        /* if current->first_ts == prev->last_ts,
{'fraction_non_alphanumeric': 0.10887555994229747, 'fraction_numerical': 0.015741654645306608, 'mean_word_length': 3.4512785851075813, 'pattern_counts': {'":': 0, '<': 57, '<?xml version=': 0, '>': 373, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5293861', 'n_tokens_mistral': 14493, 'n_tokens_neox': 13665, 'n_words': 3547}