<commit_before>/**
 * @file
 * @author <NAME>
 * @date 2012.02.13
 * @brief Implementation of the Asynchronous Acknowledgement Gerät and friends
 */

#include <map>
#include <set>
#include <deque>
#include <iostream>
#include <cstring>

#include "async_ack_geraet.hxx"
#include "io.hxx"

using namespace std;

/* The AAG handles its destruction by setting the aag field of all associated
 * senders and receivers to NULL, so both of those classes must gracefully
 * handle this condition.
 */

AAGReceiver::AAGReceiver(AsyncAckGeraet* aag_, DeletionStrategy ds)
: InputNetworkGeraet(ds),
  aag(aag_)
{
  aag->assocReceiver(this);
}

AAGReceiver::~AAGReceiver() {
  if (aag)
    aag->disassocReceiver(this);
}

void AAGReceiver::receive(NetworkConnection::seq_t seq,
                          const byte* data, unsigned len)
throw() {
  if (aag && aag->incomming(seq)) {
    receiveAccepted(seq, data, len);
  }
}


AAGSender::AAGSender(AsyncAckGeraet* aag_,
                     DeletionStrategy ds,
                     NetworkConnection* cxn_)
: OutputNetworkGeraet(cxn_? cxn_ : aag_->cxn, ds),
  aag(aag_)
{
  aag->assocSender(this);
}

AAGSender::~AAGSender() {
  if (aag)
    aag->disassocSender(this);
}

NetworkConnection::seq_t AAGSender::send(byte* data, unsigned len)
throw() {
  byte* tmp = data;
  NetworkConnection::seq_t seq = cxn->writeHeader(tmp, channel);

  cxn->send(data, len);

  // If no aag exists, we will presumably be deleted soon, so just
  // continue and send the packet without entering it into the
  // acknowledgement system.
  if (aag)
    aag->add(seq, this);

  return seq;
}

void AAGSender::forget(NetworkConnection::seq_t seq) throw() {
  if (aag)
    aag->remove(seq);
}


ReliableSender::ReliableSender(AsyncAckGeraet* aag,
                               DeletionStrategy ds)
: AAGSender(aag, ds),
  nextId(0)
{ }

unsigned ReliableSender::send(byte* data, unsigned len) throw() {
  NetworkConnection::seq_t seq = AAGSender::send(data, len);
  unsigned id = nextId++;
  pending.insert(make_pair(seq, make_pair(id, vector<byte>(data, data+len))));
  return id;
}

void ReliableSender::ack(NetworkConnection::seq_t seq) throw() {
  delivered(pending[seq].first);
  pending.erase(seq);
}

void ReliableSender::nak(NetworkConnection::seq_t seq) throw() {
  pair<unsigned, vector<byte> >& item = pending[seq];
  NetworkConnection::seq_t ns = AAGSender::send(&item.second[0],
                                                item.second.size());
  pending[ns] = item;
  pending.erase(seq);
}


const NetworkConnection::geraet_num AsyncAckGeraet::num =
  NetworkConnection::registerGeraetCreator(&creator, 1);

InputNetworkGeraet* AsyncAckGeraet::creator(NetworkConnection* cxn) {
  return cxn->aag;
}

AsyncAckGeraet::AsyncAckGeraet(NetworkConnection* cxn)
: AAGReceiver(this, InputNetworkGeraet::DSIntrinsic),
  AAGSender(this, OutputNetworkGeraet::DSIntrinsic, cxn),
  lastGreatestSeq(0), timeSinceTxn(0)
{
}

AsyncAckGeraet::~AsyncAckGeraet() {
  //Notify all receivers and senders that we no longer exist
  for (set<AAGReceiver*>::const_iterator it = receivers.begin();
       it != receivers.end(); ++it)
    (*it)->aag = NULL;
  for (set<AAGSender*>::const_iterator it = senders.begin();
       it != senders.end(); ++it)
    (*it)->aag = NULL;
}

void AsyncAckGeraet::update(unsigned et) throw() {
  //Send acknowledgement packet if there is anything to acknowledge
  if (!toAcknowledge.empty()) {
    //Update recent queue
    for (seq_t i = 0; i != (*toAcknowledge.rbegin())+1; ++i)
      recentQueue.push_back(i+lastGreatestSeq);
    //Remove stale entries
    while (recentQueue.size() >= 1024) {
      recentNak.erase(recentQueue.front());
      recentQueue.pop_front();
    }

    sendAck(toAcknowledge, lastGreatestSeq);
    //Record the packets we NAKed
    for (seq_t i = 0; i != *toAcknowledge.rbegin(); ++i)
      recentNak.insert(i+lastGreatestSeq);

    //Clear current acks
    lastGreatestSeq += 1+*toAcknowledge.rbegin();
    toAcknowledge.clear();
  }

  //Check for long-term unacknowledged acks<commit_msg>Fix AsyncAckGeraet not properly maintaining the recent NAK list.<commit_after>/**
 * @file
 * @author <NAME>
 * @date 2012.02.13
 * @brief Implementation of the Asynchronous Acknowledgement Gerät and friends
 */

#include <map>
#include <set>
#include <deque>
#include <iostream>
#include <cstring>

#include "async_ack_geraet.hxx"
#include "io.hxx"

using namespace std;

/* The AAG handles its destruction by setting the aag field of all associated
 * senders and receivers to NULL, so both of those classes must gracefully
 * handle this condition.
 */

AAGReceiver::AAGReceiver(AsyncAckGeraet* aag_, DeletionStrategy ds)
: InputNetworkGeraet(ds),
  aag(aag_)
{
  aag->assocReceiver(this);
}

AAGReceiver::~AAGReceiver() {
  if (aag)
    aag->disassocReceiver(this);
}

void AAGReceiver::receive(NetworkConnection::seq_t seq,
                          const byte* data, unsigned len)
throw() {
  if (aag && aag->incomming(seq)) {
    receiveAccepted(seq, data, len);
  }
}


AAGSender::AAGSender(AsyncAckGeraet* aag_,
                     DeletionStrategy ds,
                     NetworkConnection* cxn_)
: OutputNetworkGeraet(cxn_? cxn_ : aag_->cxn, ds),
  aag(aag_)
{
  aag->assocSender(this);
}

AAGSender::~AAGSender() {
  if (aag)
    aag->disassocSender(this);
}

NetworkConnection::seq_t AAGSender::send(byte* data, unsigned len)
throw() {
  byte* tmp = data;
  NetworkConnection::seq_t seq = cxn->writeHeader(tmp, channel);

  cxn->send(data, len);

  // If no aag exists, we will presumably be deleted soon, so just
  // continue and send the packet without entering it into the
  // acknowledgement system.
  if (aag)
    aag->add(seq, this);

  return seq;
}

void AAGSender::forget(NetworkConnection::seq_t seq) throw() {
  if (aag)
    aag->remove(seq);
}


ReliableSender::ReliableSender(AsyncAckGeraet* aag,
                               DeletionStrategy ds)
: AAGSender(aag, ds),
  nextId(0)
{ }

unsigned ReliableSender::send(byte* data, unsigned len) throw() {
  NetworkConnection::seq_t seq = AAGSender::send(data, len);
  unsigned id = nextId++;
  pending.insert(make_pair(seq, make_pair(id, vector<byte>(data, data+len))));
  return id;
}

void ReliableSender::ack(NetworkConnection::seq_t seq) throw() {
  delivered(pending[seq].first);
  pending.erase(seq);
}

void ReliableSender::nak(NetworkConnection::seq_t seq) throw() {
  pair<unsigned, vector<byte> >& item = pending[seq];
  NetworkConnection::seq_t ns = AAGSender::send(&item.second[0],
                                                item.second.size());
  pending[ns] = item;
  pending.erase(seq);
}


const NetworkConnection::geraet_num AsyncAckGeraet::num =
  NetworkConnection::registerGeraetCreator(&creator, 1);

InputNetworkGeraet* AsyncAckGeraet::creator(NetworkConnection* cxn) {
  return cxn->aag;
}

AsyncAckGeraet::AsyncAckGeraet(NetworkConnection* cxn)
: AAGReceiver(this, InputNetworkGeraet::DSIntrinsic),
  AAGSender(this, OutputNetworkGeraet::DSIntrinsic, cxn),
  lastGreatestSeq(0), timeSinceTxn(0)
{
}

AsyncAckGeraet::~AsyncAckGeraet() {
  //Notify all receivers and senders that we no longer exist
  for (set<AAGReceiver*>::const_iterator it = receivers.begin();
       it != receivers.end(); ++it)
    (*it)->aag = NULL;
  for (set<AAGSender*>::const_iterator it = senders.begin();
       it != senders.end(); ++it)
    (*it)->aag = NULL;
}

void AsyncAckGeraet::update(unsigned et) throw() {
  //Send acknowledgement packet if there is anything to acknowledge
  if (!toAcknowledge.empty()) {
    //Update recent queue
    for (seq_t i = 0; i != ((seq_t)((*toAcknowledge.rbegin())+1)); ++i) {
      recentNak.insert(i+lastGreatestSeq);
      recentQueue.push_back(i+lastGreatestSeq);
    }
    //Remove stale entries
    while (recentQueue.size() >= 1024) {
      recentNak.erase(recentQueue.front());
      recentQueue.pop_front();
    }

    sendAck(toAcknowledge, lastGreatestSeq);

    //Clear current acks
    lastGreatestSeq += 1+*toAcknowledge.rbegin();
    toAcknowledge.clear();
  }

  //Check for long-term unacknowledged acks
{'fraction_non_alphanumeric': 0.13565174497471943, 'fraction_numerical': 0.0050561104945122705, 'mean_word_length': 3.9724095646842428, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 53, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6174019', 'n_tokens_mistral': 2826, 'n_tokens_neox': 2750, 'n_words': 613}