<commit_before>// This file is a part of the IncludeOS unikernel - www.includeos.org
//
// Copyright 2015-2016 Oslo and Akershus University College of Applied Sciences
// and <NAME>
//
// 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.

// #define DEBUG
// #define DEBUG2

#include <common> // Ensures/Expects
#include <net/tcp/connection.hpp>
#include <net/tcp/connection_states.hpp>
#include <net/tcp/tcp.hpp>
#include <net/tcp/tcp_errors.hpp>

using namespace net::tcp;
using namespace std;

Connection::Connection(TCP& host, Socket local, Socket remote, ConnectCallback callback)
  : host_(host),
    local_{std::move(local)}, remote_{std::move(remote)},
    is_ipv6_(local_.address().is_v6()),
    state_(&Connection::Closed::instance()),
    prev_state_(state_),
    cb{host_.window_size()},
    read_request(nullptr),
    writeq(),
    recv_wnd_getter{TCP::global_recv_wnd},
    on_connect_{std::move(callback)},
    on_disconnect_({this, &Connection::default_on_disconnect}),
    rtx_timer({this, &Connection::rtx_timeout}),
    timewait_dack_timer({this, &Connection::dack_timeout}),
    queued_(false),
    dack_{0},
    last_ack_sent_{cb.RCV.NXT},
    smss_{host_.MSS()}
{
  setup_congestion_control();
  //printf("<Connection> Created %p %s  ACTIVE: %u\n", this,
  //        to_string().c_str(), host_.active_connections());
}

Connection::~Connection()
{
  //printf("<Connection> Deleted %p %s  ACTIVE: %u\n", this,
  //        to_string().c_str(), host_.active_connections());
  rtx_clear();
}

void Connection::_on_read(size_t recv_bufsz, ReadCallback cb)
{
  (void) recv_bufsz;
  if(read_request == nullptr)
  {
    Expects(bufalloc != nullptr);
    read_request.reset(
      new Read_request(this->cb.RCV.NXT, host_.min_bufsize(), host_.max_bufsize(), cb, bufalloc.get()));
  }
  // read request is already set, only reset if new size.
  else
  {
    //printf("on_read already set\n");
    read_request->callback = cb;
    // this will flush the current data to the user (if any)
    read_request->reset(this->cb.RCV.NXT);

    // due to throwing away buffers (and all data) we also
    // need to clear the sack list if anything is stored here.
    if(sack_list)
      sack_list->clear();
  }
}

Connection_ptr Connection::retrieve_shared() {
  return host_.retrieve_shared(this);
}

/*
  This is most likely used in a PASSIVE open
*/
/*Connection::Connection(TCP& host, port_t local_port, ConnectCallback cb)
  : Connection(host, local_port, Socket(), std::move(cb))
{
}*/

Connection::TCB::TCB(const uint32_t recvwin)
  : SND{ 0, 0, default_window_size, 0, 0, 0, default_mss, 0, false },
    ISS{(seq_t)4815162342},
    RCV{ 0, recvwin, 0, 0, 0 },
    IRS{0},
    ssthresh{recvwin},
    cwnd{0},
    recover{0},
    TS_recent{0}
{
}

Connection::TCB::TCB()
  : Connection::TCB(default_window_size)
{
}

void Connection::reset_callbacks()
{
  on_disconnect_ = {this, &Connection::default_on_disconnect};
  on_connect_.reset();
  writeq.on_write(nullptr);
  on_close_.reset();
  recv_wnd_getter.reset();
  if(read_request)
    read_request->callback.reset();
}

uint16_t Connection::MSDS() const noexcept {
  return std::min(host_.MSS(), cb.SND.MSS) + sizeof(Header);
}

size_t Connection::receive(seq_t seq, const uint8_t* data, size_t n, bool PUSH) {
  // should not be called without an read request
  Expects(read_request);
  Expects(n > 0);
  size_t received{0};

  while(n)
  {
    auto read = read_request->insert(seq, data + received, n, PUSH);

    n -= read; // subtract amount of data left to insert
    received += read; // add to the total recieved
    seq += read; // advance the sequence number
  }

  return received;
}


void Connection::write(buffer_t buffer)
{
  if (UNLIKELY(buffer->size() == 0)) {
    throw TCP_error("Can't write zero bytes to TCP stream");
  }

  // Only write if allowed
  if(state_->is_writable())
  {
    // add to queue
    writeq.push_back(std::move(buffer));

    // request packets if connected, else let ACK clock do the writing
    if(state_->is_connected())
      host_.request_offer(*this);
  }
}

void Connection::offer(size_t& packets)
{
  debug2("<Connection::offer> %s got offered [%u] packets. Usable window is %u.\n",
        to_string().c_str(), packets, usable_window());

  // write until we either cant send more (window closes or no more in queue),
  // or we're out of packets.

  static bool congested=false;
  if (!can_send())
  {
    if (sendq_remaining() > 0)
    {
      printf("Window_blocked %d data %zu empty packets sw=%ld fs=%ld WND=%d\n",sendq_remaining(),packets,(int64_t)send_window(), (int64_t)flight_size(),cb.SND.WND);
      congested=true;
    }
  }
  if (can_send() && congested)
  {
    printf("Window released congestion removed %zu packets\n",packets);
    congested=false;
  }
<commit_msg>tcp: removed uneccesary old print<commit_after>// This file is a part of the IncludeOS unikernel - www.includeos.org
//
// Copyright 2015-2016 Oslo and Akershus University College of Applied Sciences
// and <NAME>
//
// 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.

// #define DEBUG
// #define DEBUG2

#include <common> // Ensures/Expects
#include <net/tcp/connection.hpp>
#include <net/tcp/connection_states.hpp>
#include <net/tcp/tcp.hpp>
#include <net/tcp/tcp_errors.hpp>

using namespace net::tcp;
using namespace std;

Connection::Connection(TCP& host, Socket local, Socket remote, ConnectCallback callback)
  : host_(host),
    local_{std::move(local)}, remote_{std::move(remote)},
    is_ipv6_(local_.address().is_v6()),
    state_(&Connection::Closed::instance()),
    prev_state_(state_),
    cb{host_.window_size()},
    read_request(nullptr),
    writeq(),
    recv_wnd_getter{TCP::global_recv_wnd},
    on_connect_{std::move(callback)},
    on_disconnect_({this, &Connection::default_on_disconnect}),
    rtx_timer({this, &Connection::rtx_timeout}),
    timewait_dack_timer({this, &Connection::dack_timeout}),
    queued_(false),
    dack_{0},
    last_ack_sent_{cb.RCV.NXT},
    smss_{host_.MSS()}
{
  setup_congestion_control();
  //printf("<Connection> Created %p %s  ACTIVE: %u\n", this,
  //        to_string().c_str(), host_.active_connections());
}

Connection::~Connection()
{
  //printf("<Connection> Deleted %p %s  ACTIVE: %u\n", this,
  //        to_string().c_str(), host_.active_connections());
  rtx_clear();
}

void Connection::_on_read(size_t recv_bufsz, ReadCallback cb)
{
  (void) recv_bufsz;
  if(read_request == nullptr)
  {
    Expects(bufalloc != nullptr);
    read_request.reset(
      new Read_request(this->cb.RCV.NXT, host_.min_bufsize(), host_.max_bufsize(), cb, bufalloc.get()));
  }
  // read request is already set, only reset if new size.
  else
  {
    //printf("on_read already set\n");
    read_request->callback = cb;
    // this will flush the current data to the user (if any)
    read_request->reset(this->cb.RCV.NXT);

    // due to throwing away buffers (and all data) we also
    // need to clear the sack list if anything is stored here.
    if(sack_list)
      sack_list->clear();
  }
}

Connection_ptr Connection::retrieve_shared() {
  return host_.retrieve_shared(this);
}

/*
  This is most likely used in a PASSIVE open
*/
/*Connection::Connection(TCP& host, port_t local_port, ConnectCallback cb)
  : Connection(host, local_port, Socket(), std::move(cb))
{
}*/

Connection::TCB::TCB(const uint32_t recvwin)
  : SND{ 0, 0, default_window_size, 0, 0, 0, default_mss, 0, false },
    ISS{(seq_t)4815162342},
    RCV{ 0, recvwin, 0, 0, 0 },
    IRS{0},
    ssthresh{recvwin},
    cwnd{0},
    recover{0},
    TS_recent{0}
{
}

Connection::TCB::TCB()
  : Connection::TCB(default_window_size)
{
}

void Connection::reset_callbacks()
{
  on_disconnect_ = {this, &Connection::default_on_disconnect};
  on_connect_.reset();
  writeq.on_write(nullptr);
  on_close_.reset();
  recv_wnd_getter.reset();
  if(read_request)
    read_request->callback.reset();
}

uint16_t Connection::MSDS() const noexcept {
  return std::min(host_.MSS(), cb.SND.MSS) + sizeof(Header);
}

size_t Connection::receive(seq_t seq, const uint8_t* data, size_t n, bool PUSH) {
  // should not be called without an read request
  Expects(read_request);
  Expects(n > 0);
  size_t received{0};

  while(n)
  {
    auto read = read_request->insert(seq, data + received, n, PUSH);

    n -= read; // subtract amount of data left to insert
    received += read; // add to the total recieved
    seq += read; // advance the sequence number
  }

  return received;
}


void Connection::write(buffer_t buffer)
{
  if (UNLIKELY(buffer->size() == 0)) {
    throw TCP_error("Can't write zero bytes to TCP stream");
  }

  // Only write if allowed
  if(state_->is_writable())
  {
    // add to queue
    writeq.push_back(std::move(buffer));

    // request packets if connected, else let ACK clock do the writing
    if(state_->is_connected())
      host_.request_offer(*this);
  }
}

void Connection::offer(size_t& packets)
{
  debug2("<Connection::offer> %s got offered [%u] packets. Usable window is %u.\n",
        to_string().c_str(), packets, usable_window());

  // write until we either cant send more (window closes or no more in queue),
  // or we're out of packets.

{'fraction_non_alphanumeric': 0.1413011480718281, 'fraction_numerical': 0.010106957119026592, 'mean_word_length': 3.84640989063243, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 44, 'https://': 0, 'lorem ipsum': 0, 'www.': 4, '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': '4121094', 'n_tokens_mistral': 3507, 'n_tokens_neox': 3391, 'n_words': 905}