Comparar texto

Encuentra la diferencia entre dos archivos de texto

Editor live

Ocultar espacios en blanco

Ocultar sin cambios

Sin ajuste de línea

Vista

Nivel de detalle

Estilos de texto

Resaltado de sintaxis

Diffchecker Desktop La forma más segura de usar Diffchecker. ¡Obtén la app de Diffchecker Desktop: tus diffs nunca salen de tu computadora!Obtener Desktop

EvolveDatacenterTCPDiff

Creado hace 6 mesesEl diff nunca expira

8 eliminaciones

Líneas
Total
Eliminado

Caracteres
Total
Eliminado

Para continuar usando esta función, actualice a Diffchecker Pro Ver precios

383 líneas

19 adiciones

Líneas
Total
Añadido

Caracteres
Total
Añadido

Para continuar usando esta función, actualice a Diffchecker Pro Ver precios

389 líneas

#include "tcp-datacenter.h"

namespace ns3 {

Ptr<RdmaQueuePair> qp = CreateObject<RdmaQueuePair>(pg, sip, dip, sport, dport);

qp->SetSize(size);

qp->SetWin(win);

qp->SetBaseRtt(baseRtt);

qp->SetVarWin(m_var_win);

qp->SetAppNotifyCallback(notifyAppFinish);

qp->stopTime = stopTime;

uint32_t nic_idx = GetNicIdxOfQp(qp);

m_nic[nic_idx].qpGrp->AddQp(qp);

uint64_t key = GetQpKey(dip.Get(), sport, pg);

m_qpMap[key] = qp;

qp->powerEnabled = true;

if (m_nic[nic_idx].dev == NULL) {

std::cout << "sip " << sip << " dip " << dip << " sport " << sport << " dport " << dport << std::endl;

}

DataRate m_bps = m_nic[nic_idx].dev->GetDataRate();

if(win)

qp->SetWin(m_bps.GetBitRate() * 1 * baseRtt * 1e-9 / 8);

qp->SetWin(m_bps.GetBitRate() * 1.0 * baseRtt * 1e-9 / 8);

qp->m_rate = m_bps;

qp->m_rate = DataRate(m_bps.GetBitRate() / 10);

qp->m_max_rate = m_bps;

qp->hp.m_curRate = m_bps;

qp->hp.m_curRate = DataRate(m_bps.GetBitRate() / 10);

m_nic[nic_idx].dev->NewQp(qp);

}

int TcpDataCenter::ReceiveCnp(Ptr<Packet> p, CustomHeader &ch) {

uint32_t qIndex = ch.cnp.qIndex;

if (qIndex == 1) {

std::cout << "TCP--ignore\n";

return 0;

}

uint16_t udpport = ch.cnp.fid;

Ptr<RdmaQueuePair> qp = GetQp(ch.sip, udpport, qIndex);

if (qp == NULL)

std::cout << "ERROR: QCN NIC cannot find the flow\n";

uint32_t nic_idx = GetNicIdxOfQp(qp);

Ptr<QbbNetDevice> dev = m_nic[nic_idx].dev;

if (qp->m_rate == 0)

{

qp->m_rate = dev->GetDataRate();

qp->hp.m_curRate = dev->GetDataRate();

}

return 0;

}

int TcpDataCenter::ReceiveAck(Ptr<Packet> p, CustomHeader &ch) {

uint16_t qIndex = ch.ack.pg;

uint16_t port = ch.ack.dport;

uint32_t seq = ch.ack.seq;

uint8_t cnp = (ch.ack.flags >> qbbHeader::FLAG_CNP) & 1;

Ptr<RdmaQueuePair> qp = GetQp(ch.sip, port, qIndex);

if (qp == NULL) {

std::cout << "ERROR: " << "node:" << m_node->GetId() << ' ' << (ch.l3Prot == 0xFC ? "ACK" : "NACK") << " NIC cannot find the flow\n";

return 0;

}

uint32_t nic_idx = GetNicIdxOfQp(qp);

Ptr<QbbNetDevice> dev = m_nic[nic_idx].dev;

if (m_ack_interval == 0)

std::cout << "ERROR: shouldn't receive ack\n";

else {

if (!m_backto0) {

qp->Acknowledge(seq);

} else {

uint32_t goback_seq = seq / m_chunk * m_chunk;

qp->Acknowledge(goback_seq);

}

if (qp->IsFinished()) {

QpComplete(qp);

}

if (ch.l3Prot == 0xFD)

RecoverQueue(qp);

HandleAckHp(qp, p, ch);

dev->TriggerTransmit();

return 0;

}

#define PRINT_LOG 0

void TcpDataCenter::HandleAckHp(Ptr<RdmaQueuePair> qp, Ptr<Packet> p, CustomHeader &ch) {

uint32_t ack_seq = ch.ack.seq;

if (ack_seq > qp->hp.m_lastUpdateSeq) {

UpdateRatePower(qp, p, ch, false);

} else {

FastReactPower(qp, p, ch);

}

void TcpDataCenter::UpdateRatePower(Ptr<RdmaQueuePair> qp, Ptr<Packet> p, CustomHeader &ch, bool fast_react) {

uint32_t next_seq = qp->snd_nxt;

double prevRtt = qp->m_baseRtt;

double prevCompletion = Simulator::Now().GetNanoSeconds();

std::map<uint32_t, double>::iterator it = qp->rates.find(ch.ack.seq);

if (it != qp->rates.end()) {

qp->rates.erase(it);

prevRtt = Simulator::Now().GetNanoSeconds() - it->second;

qp->m_baseRtt = std::min(uint64_t(Simulator::Now().GetNanoSeconds() - it->second), qp->m_baseRtt);

prevCompletion = Simulator::Now().GetNanoSeconds();

}

if (qp->hp.m_lastUpdateSeq == 0) {

qp->prevRtt = prevRtt;

qp->prevCompletion = Simulator::Now().GetNanoSeconds();

qp->hp.m_lastUpdateSeq = next_seq;

}else {

double max_c = 0;

double U = 0;

uint64_t dt = 0;

double A = ( double(prevRtt - qp->prevRtt) / (prevCompletion - qp->prevCompletion) + 1 );

double dt_sample = prevCompletion - qp->prevCompletion;

if (A < 0.5)

if (dt_sample < 1.0) dt_sample = 1.0;

A = 0.5;

double A = ( double(prevRtt - qp->prevRtt) / dt_sample + 1 );

if (A < 0.5) A = 0.5;

double power = ( A ) * (prevRtt);

double powerx = (power) / (1.05 * qp->m_baseRtt);

double powerx = (power) / (1.03 * qp->m_baseRtt);

double u = powerx;

uint8_t cnp = (ch.ack.flags >> qbbHeader::FLAG_CNP) & 1;

if (cnp) {

if (u < 2.0) u = 2.0;

}

if (u > U) {

U = u;

dt = prevCompletion - qp->prevCompletion;

dt = (uint64_t)dt_sample;

}

DataRate new_rate;

if (dt > 1.0 * qp->m_baseRtt)

dt = 1.0 * qp->m_baseRtt;

if (U < 0) {

U = qp->hp.u;

}

qp->hp.u = (qp->hp.u * (1.0 * qp->m_baseRtt - dt) + U * dt) / double(1.0 * qp->m_baseRtt);

max_c = qp->hp.u;

new_rate = (0.7 * ( qp->hp.m_curRate / max_c + DataRate("150Mbps") ) + 0.3 * qp->hp.m_curRate);

new_rate = (0.8 * ( qp->hp.m_curRate / max_c + DataRate("250Mbps") ) + 0.2 * qp->hp.m_curRate);

if (new_rate < m_minRate)

new_rate = m_minRate;

if (new_rate > qp->m_max_rate)

new_rate = qp->m_max_rate;

qp->prevRtt = prevRtt;

qp->prevCompletion = Simulator::Now().GetNanoSeconds();

ChangeRate(qp, new_rate);

if (!fast_react) {

qp->hp.m_curRate = new_rate;

}

if (!fast_react) {

if (next_seq > qp->hp.m_lastUpdateSeq)

qp->hp.m_lastUpdateSeq = next_seq;

}

void TcpDataCenter::FastReactPower(Ptr<RdmaQueuePair> qp, Ptr<Packet> p, CustomHeader &ch) {

if (m_fast_react)

UpdateRatePower(qp, p, ch, true);

}

void TcpDataCenter::Setup(QpCompleteCallback cb) {

for (uint32_t i = 0; i < m_nic.size(); i++) {

Ptr<QbbNetDevice> dev = m_nic[i].dev;

if (dev == NULL)

continue;

dev->m_rdmaEQ->m_qpGrp = m_nic[i].qpGrp;

dev->m_rdmaReceiveCb = MakeCallback(&TcpDataCenter::Receive, this);

dev->m_rdmaLinkDownCb = MakeCallback(&TcpDataCenter::SetLinkDown, this);

dev->m_rdmaPktSent = MakeCallback(&TcpDataCenter::PktSent, this);

dev->m_rdmaEQ->m_rdmaGetNxtPkt = MakeCallback(&TcpDataCenter::GetNxtPacket, this);

}

m_qpCompleteCallback = cb;

}

void TcpDataCenter::DeleteQueuePair(Ptr<RdmaQueuePair> qp) {

uint64_t key = GetQpKey(qp->dip.Get(), qp->sport, qp->m_pg);

m_qpMap.erase(key);

}

Ptr<RdmaRxQueuePair> TcpDataCenter::GetRxQp(uint32_t sip, uint32_t dip, uint16_t sport, uint16_t dport, uint16_t pg, bool create) {

uint64_t key = ((uint64_t)dip << 32) | ((uint64_t)pg << 16) | (uint64_t)dport;

auto it = m_rxQpMap.find(key);

if (it != m_rxQpMap.end())

return it->second;

if (create) {

Ptr<RdmaRxQueuePair> q = CreateObject<RdmaRxQueuePair>();

q->sip = sip;

q->dip = dip;

q->sport = sport;

q->dport = dport;

q->m_ecn_source.qIndex = pg;

m_rxQpMap[key] = q;

return q;

}

return NULL;

}

void TcpDataCenter::DeleteRxQp(uint32_t dip, uint16_t pg, uint16_t dport) {

uint64_t key = ((uint64_t)dip << 32) | ((uint64_t)pg << 16) | (uint64_t)dport;

m_rxQpMap.erase(key);

}

int TcpDataCenter::ReceiveUdp(Ptr<Packet> p, CustomHeader &ch) {

uint8_t ecnbits = ch.GetIpv4EcnBits();

uint32_t payload_size = p->GetSize() - ch.GetSerializedSize();

Ptr<RdmaRxQueuePair> rxQp = GetRxQp(ch.dip, ch.sip, ch.udp.dport, ch.udp.sport, ch.udp.pg, true);

if (ecnbits != 0) {

rxQp->m_ecn_source.ecnbits |= ecnbits;

rxQp->m_ecn_source.qfb++;

}

rxQp->m_ecn_source.total++;

rxQp->m_milestone_rx = m_ack_interval;

int x = ReceiverCheckSeq(ch.udp.seq, rxQp, payload_size);

if (x == 1 || x == 2) {

qbbHeader seqh;

seqh.SetSeq(rxQp->ReceiverNextExpectedSeq);

seqh.SetPG(ch.udp.pg);

seqh.SetSport(ch.udp.dport);

seqh.SetDport(ch.udp.sport);

if (ecnbits)

seqh.SetCnp();

Ptr<Packet> newp = Create<Packet>(std::max(60 - 14 - 20 - (int)seqh.GetSerializedSize(), 0));

newp->AddHeader(seqh);

Ipv4Header head;

head.SetDestination(Ipv4Address(ch.sip));

head.SetSource(Ipv4Address(ch.dip));

head.SetProtocol(x == 1 ? 0xFC : 0xFD);

head.SetTtl(64);

head.SetPayloadSize(newp->GetSize());

head.SetIdentification(rxQp->m_ipid++);

newp->AddHeader(head);

AddHeader(newp, 0x800);

uint32_t nic_idx = GetNicIdxOfRxQp(rxQp);

m_nic[nic_idx].dev->RdmaEnqueueHighPrioQ(newp);

m_nic[nic_idx].dev->TriggerTransmit();

}

return 0;

}

int TcpDataCenter::Receive(Ptr<Packet> p, CustomHeader &ch) {

if (ch.l3Prot == 0x11) {

ReceiveUdp(p, ch);

} else if (ch.l3Prot == 0xFF) {

ReceiveCnp(p, ch);

} else if (ch.l3Prot == 0xFD) {

ReceiveAck(p, ch);

} else if (ch.l3Prot == 0xFC) {

ReceiveAck(p, ch);

}

return 0;

}

void TcpDataCenter::RecoverQueue(Ptr<RdmaQueuePair> qp) {

qp->snd_nxt = qp->snd_una;

}

void TcpDataCenter::QpComplete(Ptr<RdmaQueuePair> qp) {

NS_ASSERT(!m_qpCompleteCallback.IsNull());

if (m_cc_mode == 1) {

Simulator::Cancel(qp->mlx.m_eventUpdateAlpha);

Simulator::Cancel(qp->mlx.m_eventDecreaseRate);

Simulator::Cancel(qp->mlx.m_rpTimer);

}

m_qpCompleteCallback(qp);

qp->m_notifyAppFinish();

DeleteQueuePair(qp);

}

void TcpDataCenter::SetLinkDown(Ptr<QbbNetDevice> dev) {

printf("RdmaHw: node:%u a link down\n", m_node->GetId());

}

void TcpDataCenter::AddTableEntry(Ipv4Address &dstAddr, uint32_t intf_idx) {

uint32_t dip = dstAddr.Get();

m_rtTable[dip].push_back(intf_idx);

}

void TcpDataCenter::ClearTable() {

m_rtTable.clear();

}

void TcpDataCenter::RedistributeQp() {

for (uint32_t i = 0; i < m_nic.size(); i++) {

if (m_nic[i].dev == NULL)

continue;

m_nic[i].qpGrp->Clear();

}

for (auto &it : m_qpMap) {

Ptr<RdmaQueuePair> qp = it.second;

uint32_t nic_idx = GetNicIdxOfQp(qp);

m_nic[nic_idx].qpGrp->AddQp(qp);

m_nic[nic_idx].dev->ReassignedQp(qp);

}

Ptr<Packet> TcpDataCenter::GetNxtPacket(Ptr<RdmaQueuePair> qp) {

uint32_t payload_size = qp->GetBytesLeft();

if (m_mtu < payload_size)

payload_size = m_mtu;

Ptr<Packet> p = Create<Packet> (payload_size);

SeqTsHeader seqTs;

seqTs.SetSeq (qp->snd_nxt);

seqTs.SetPG (qp->m_pg);

p->AddHeader (seqTs);

UdpHeader udpHeader;

udpHeader.SetDestinationPort (qp->dport);

udpHeader.SetSourcePort (qp->sport);

p->AddHeader (udpHeader);

Ipv4Header ipHeader;

ipHeader.SetSource (qp->sip);

ipHeader.SetDestination (qp->dip);

ipHeader.SetProtocol (0x11);

ipHeader.SetPayloadSize (p->GetSize());

ipHeader.SetTtl (64);

ipHeader.SetTos (0);

ipHeader.SetIdentification (qp->m_ipid);

p->AddHeader(ipHeader);

PppHeader ppp;

ppp.SetProtocol (0x0021);

p->AddHeader (ppp);

qp->snd_nxt += payload_size;

qp->m_ipid++;

return p;

}

void TcpDataCenter::PktSent(Ptr<RdmaQueuePair> qp, Ptr<Packet> pkt, Time interframeGap) {

qp->lastPktSize = pkt->GetSize();

uint32_t seq = qp->snd_nxt;

qp->rates[qp->snd_nxt] = Simulator::Now().GetNanoSeconds();

UpdateNextAvail(qp, interframeGap, pkt->GetSize());

}

void TcpDataCenter::UpdateNextAvail(Ptr<RdmaQueuePair> qp, Time interframeGap, uint32_t pkt_size) {

Time sendingTime;

if (m_rateBound)

sendingTime = interframeGap + qp->m_rate.CalculateBytesTxTime(pkt_size);

else

sendingTime = interframeGap + qp->m_max_rate.CalculateBytesTxTime(pkt_size);

qp->m_nextAvail = Simulator::Now() + sendingTime;

}

void TcpDataCenter::ChangeRate(Ptr<RdmaQueuePair> qp, DataRate new_rate) {

#if 1

Time sendingTime = qp->m_rate.CalculateBytesTxTime(qp->lastPktSize);

Time new_sendintTime = new_rate.CalculateBytesTxTime(qp->lastPktSize);

qp->m_nextAvail = qp->m_nextAvail + new_sendintTime - sendingTime;

uint32_t nic_idx = GetNicIdxOfQp(qp);

m_nic[nic_idx].dev->UpdateNextAvail(qp->m_nextAvail);

#endif

qp->m_rate = new_rate;

}

Diferencias guardadas

Texto original

Abrir archivo

#include "tcp-datacenter.h"

namespace ns3 {

void TcpDataCenter::AddQueuePair(uint64_t size, uint16_t pg, Ipv4Address sip, Ipv4Address dip, uint16_t sport, uint16_t dport, uint32_t win, uint64_t baseRtt, Callback<void> notifyAppFinish, Time stopTime) {
	Ptr<RdmaQueuePair> qp = CreateObject<RdmaQueuePair>(pg, sip, dip, sport, dport);
	qp->SetSize(size);
	qp->SetWin(win);
	qp->SetBaseRtt(baseRtt);
	qp->SetVarWin(m_var_win);
	qp->SetAppNotifyCallback(notifyAppFinish);
	qp->stopTime = stopTime;

uint32_t nic_idx = GetNicIdxOfQp(qp);
	m_nic[nic_idx].qpGrp->AddQp(qp);
	uint64_t key = GetQpKey(dip.Get(), sport, pg);
	m_qpMap[key] = qp;

qp->powerEnabled = true;

if (m_nic[nic_idx].dev == NULL) {
		std::cout << "sip " << sip << " dip " << dip << " sport " << sport  << " dport " << dport << std::endl;
	}
	DataRate m_bps = m_nic[nic_idx].dev->GetDataRate();
	if(win)
		qp->SetWin(m_bps.GetBitRate() * 1 * baseRtt * 1e-9 / 8);
	qp->m_rate = m_bps;
	qp->m_max_rate = m_bps;
    
    qp->hp.m_curRate = m_bps;

m_nic[nic_idx].dev->NewQp(qp);
}

int TcpDataCenter::ReceiveCnp(Ptr<Packet> p, CustomHeader &ch) {
	uint32_t qIndex = ch.cnp.qIndex;
	if (qIndex == 1) {
		std::cout << "TCP--ignore\n";
		return 0;
	}
	uint16_t udpport = ch.cnp.fid;

Ptr<RdmaQueuePair> qp = GetQp(ch.sip, udpport, qIndex);
	if (qp == NULL)
		std::cout << "ERROR: QCN NIC cannot find the flow\n";
	uint32_t nic_idx = GetNicIdxOfQp(qp);
	Ptr<QbbNetDevice> dev = m_nic[nic_idx].dev;

if (qp->m_rate == 0)
	{
		qp->m_rate = dev->GetDataRate();
        qp->hp.m_curRate = dev->GetDataRate();
	}
	return 0;
}

int TcpDataCenter::ReceiveAck(Ptr<Packet> p, CustomHeader &ch) {
	uint16_t qIndex = ch.ack.pg;
	uint16_t port = ch.ack.dport;
	uint32_t seq = ch.ack.seq;
	uint8_t cnp = (ch.ack.flags >> qbbHeader::FLAG_CNP) & 1;
	
	Ptr<RdmaQueuePair> qp = GetQp(ch.sip, port, qIndex);
	if (qp == NULL) {
		std::cout << "ERROR: " << "node:" << m_node->GetId() << ' ' << (ch.l3Prot == 0xFC ? "ACK" : "NACK") << " NIC cannot find the flow\n";
		return 0;
	}

uint32_t nic_idx = GetNicIdxOfQp(qp);
	Ptr<QbbNetDevice> dev = m_nic[nic_idx].dev;
	if (m_ack_interval == 0)
		std::cout << "ERROR: shouldn't receive ack\n";
	else {
		if (!m_backto0) {
			qp->Acknowledge(seq);
		} else {
			uint32_t goback_seq = seq / m_chunk * m_chunk;
			qp->Acknowledge(goback_seq);
		}
		if (qp->IsFinished()) {
			QpComplete(qp);
		}
	}
	if (ch.l3Prot == 0xFD)
		RecoverQueue(qp);

HandleAckHp(qp, p, ch);

dev->TriggerTransmit();
	return 0;
}

#define PRINT_LOG 0

void TcpDataCenter::HandleAckHp(Ptr<RdmaQueuePair> qp, Ptr<Packet> p, CustomHeader &ch) {
	uint32_t ack_seq = ch.ack.seq;

if (ack_seq > qp->hp.m_lastUpdateSeq) {
		UpdateRatePower(qp, p, ch, false);
	} else {
		FastReactPower(qp, p, ch);
	}
}

void TcpDataCenter::UpdateRatePower(Ptr<RdmaQueuePair> qp, Ptr<Packet> p, CustomHeader &ch, bool fast_react) {
	uint32_t next_seq = qp->snd_nxt;
	double prevRtt = qp->m_baseRtt;
	double prevCompletion = Simulator::Now().GetNanoSeconds();
	std::map<uint32_t, double>::iterator it = qp->rates.find(ch.ack.seq);

if (it != qp->rates.end()) {
		qp->rates.erase(it);
		prevRtt = Simulator::Now().GetNanoSeconds() - it->second;
		qp->m_baseRtt = std::min(uint64_t(Simulator::Now().GetNanoSeconds() - it->second), qp->m_baseRtt);
		prevCompletion = Simulator::Now().GetNanoSeconds();
	}
	if (qp->hp.m_lastUpdateSeq == 0) {
		qp->prevRtt = prevRtt;
		qp->prevCompletion = Simulator::Now().GetNanoSeconds();
		qp->hp.m_lastUpdateSeq = next_seq;
	}else {
        double max_c = 0;
        double U = 0;
        uint64_t dt = 0;

double A = ( double(prevRtt - qp->prevRtt) / (prevCompletion - qp->prevCompletion) + 1  );
        if (A < 0.5)
            A = 0.5;
        double power = ( A ) * (prevRtt);
        double powerx = (power) / (1.05 * qp->m_baseRtt);
        double u = powerx;

if (u > U) {
            U = u;
            dt = prevCompletion - qp->prevCompletion;
        }

DataRate new_rate;

if (dt > 1.0 * qp->m_baseRtt)
            dt = 1.0 * qp->m_baseRtt;

if (U < 0) {
            U = qp->hp.u;
        }
        qp->hp.u = (qp->hp.u * (1.0 * qp->m_baseRtt - dt) + U * dt) / double(1.0 * qp->m_baseRtt);

max_c = qp->hp.u;
        new_rate = (0.7 * ( qp->hp.m_curRate / max_c + DataRate("150Mbps") ) + 0.3 * qp->hp.m_curRate);

if (new_rate < m_minRate)
            new_rate = m_minRate;
        if (new_rate > qp->m_max_rate)
            new_rate = qp->m_max_rate;

qp->prevRtt = prevRtt;
        qp->prevCompletion = Simulator::Now().GetNanoSeconds();
        
        ChangeRate(qp, new_rate);
        
        if (!fast_react) {
            qp->hp.m_curRate = new_rate;
        }

if (!fast_react) {
			if (next_seq > qp->hp.m_lastUpdateSeq)
				qp->hp.m_lastUpdateSeq = next_seq;
		}
	}
}

void TcpDataCenter::FastReactPower(Ptr<RdmaQueuePair> qp, Ptr<Packet> p, CustomHeader &ch) {
	if (m_fast_react)
		UpdateRatePower(qp, p, ch, true);
}

void TcpDataCenter::Setup(QpCompleteCallback cb) {
	for (uint32_t i = 0; i < m_nic.size(); i++) {
		Ptr<QbbNetDevice> dev = m_nic[i].dev;
		if (dev == NULL)
			continue;
		dev->m_rdmaEQ->m_qpGrp = m_nic[i].qpGrp;
		dev->m_rdmaReceiveCb = MakeCallback(&TcpDataCenter::Receive, this);
		dev->m_rdmaLinkDownCb = MakeCallback(&TcpDataCenter::SetLinkDown, this);
		dev->m_rdmaPktSent = MakeCallback(&TcpDataCenter::PktSent, this);
		dev->m_rdmaEQ->m_rdmaGetNxtPkt = MakeCallback(&TcpDataCenter::GetNxtPacket, this);
	}
	m_qpCompleteCallback = cb;
}

void TcpDataCenter::DeleteQueuePair(Ptr<RdmaQueuePair> qp) {
	uint64_t key = GetQpKey(qp->dip.Get(), qp->sport, qp->m_pg);
	m_qpMap.erase(key);
}

Ptr<RdmaRxQueuePair> TcpDataCenter::GetRxQp(uint32_t sip, uint32_t dip, uint16_t sport, uint16_t dport, uint16_t pg, bool create) {
	uint64_t key = ((uint64_t)dip << 32) | ((uint64_t)pg << 16) | (uint64_t)dport;
	auto it = m_rxQpMap.find(key);
	if (it != m_rxQpMap.end())
		return it->second;
	if (create) {
		Ptr<RdmaRxQueuePair> q = CreateObject<RdmaRxQueuePair>();
		q->sip = sip;
		q->dip = dip;
		q->sport = sport;
		q->dport = dport;
		q->m_ecn_source.qIndex = pg;
		m_rxQpMap[key] = q;
		return q;
	}
	return NULL;
}

void TcpDataCenter::DeleteRxQp(uint32_t dip, uint16_t pg, uint16_t dport) {
	uint64_t key = ((uint64_t)dip << 32) | ((uint64_t)pg << 16) | (uint64_t)dport;
	m_rxQpMap.erase(key);
}

int TcpDataCenter::ReceiveUdp(Ptr<Packet> p, CustomHeader &ch) {
	uint8_t ecnbits = ch.GetIpv4EcnBits();

uint32_t payload_size = p->GetSize() - ch.GetSerializedSize();

Ptr<RdmaRxQueuePair> rxQp = GetRxQp(ch.dip, ch.sip, ch.udp.dport, ch.udp.sport, ch.udp.pg, true);
	if (ecnbits != 0) {
		rxQp->m_ecn_source.ecnbits |= ecnbits;
		rxQp->m_ecn_source.qfb++;
	}
	rxQp->m_ecn_source.total++;
	rxQp->m_milestone_rx = m_ack_interval;

int x = ReceiverCheckSeq(ch.udp.seq, rxQp, payload_size);
	if (x == 1 || x == 2) {
		qbbHeader seqh;
		seqh.SetSeq(rxQp->ReceiverNextExpectedSeq);
		seqh.SetPG(ch.udp.pg);
		seqh.SetSport(ch.udp.dport);
		seqh.SetDport(ch.udp.sport);
		if (ecnbits)
			seqh.SetCnp();

Ptr<Packet> newp = Create<Packet>(std::max(60 - 14 - 20 - (int)seqh.GetSerializedSize(), 0));
		newp->AddHeader(seqh);

Ipv4Header head;
		head.SetDestination(Ipv4Address(ch.sip));
		head.SetSource(Ipv4Address(ch.dip));
		head.SetProtocol(x == 1 ? 0xFC : 0xFD);
		head.SetTtl(64);
		head.SetPayloadSize(newp->GetSize());
		head.SetIdentification(rxQp->m_ipid++);

newp->AddHeader(head);
		AddHeader(newp, 0x800);
		uint32_t nic_idx = GetNicIdxOfRxQp(rxQp);
		m_nic[nic_idx].dev->RdmaEnqueueHighPrioQ(newp);
		m_nic[nic_idx].dev->TriggerTransmit();
	}
	return 0;
}

int TcpDataCenter::Receive(Ptr<Packet> p, CustomHeader &ch) {
	if (ch.l3Prot == 0x11) {
		ReceiveUdp(p, ch);
	} else if (ch.l3Prot == 0xFF) {
		ReceiveCnp(p, ch);
	} else if (ch.l3Prot == 0xFD) {
		ReceiveAck(p, ch);
	} else if (ch.l3Prot == 0xFC) {
		ReceiveAck(p, ch);
	}
	return 0;
}

void TcpDataCenter::RecoverQueue(Ptr<RdmaQueuePair> qp) {
	qp->snd_nxt = qp->snd_una;
}

void TcpDataCenter::QpComplete(Ptr<RdmaQueuePair> qp) {
	NS_ASSERT(!m_qpCompleteCallback.IsNull());
	if (m_cc_mode == 1) {
		Simulator::Cancel(qp->mlx.m_eventUpdateAlpha);
		Simulator::Cancel(qp->mlx.m_eventDecreaseRate);
		Simulator::Cancel(qp->mlx.m_rpTimer);
	}

m_qpCompleteCallback(qp);

qp->m_notifyAppFinish();

DeleteQueuePair(qp);
}

void TcpDataCenter::SetLinkDown(Ptr<QbbNetDevice> dev) {
	printf("RdmaHw: node:%u a link down\n", m_node->GetId());
}

void TcpDataCenter::AddTableEntry(Ipv4Address &dstAddr, uint32_t intf_idx) {
	uint32_t dip = dstAddr.Get();
	m_rtTable[dip].push_back(intf_idx);
}

void TcpDataCenter::ClearTable() {
	m_rtTable.clear();
}

void TcpDataCenter::RedistributeQp() {
	for (uint32_t i = 0; i < m_nic.size(); i++) {
		if (m_nic[i].dev == NULL)
			continue;
		m_nic[i].qpGrp->Clear();
	}

for (auto &it : m_qpMap) {
		Ptr<RdmaQueuePair> qp = it.second;
		uint32_t nic_idx = GetNicIdxOfQp(qp);
		m_nic[nic_idx].qpGrp->AddQp(qp);
		m_nic[nic_idx].dev->ReassignedQp(qp);
	}
}

Ptr<Packet> TcpDataCenter::GetNxtPacket(Ptr<RdmaQueuePair> qp) {
	uint32_t payload_size = qp->GetBytesLeft();
	if (m_mtu < payload_size)
		payload_size = m_mtu;
	Ptr<Packet> p = Create<Packet> (payload_size);
	SeqTsHeader seqTs;
	seqTs.SetSeq (qp->snd_nxt);
	seqTs.SetPG (qp->m_pg);
	p->AddHeader (seqTs);
	UdpHeader udpHeader;
	udpHeader.SetDestinationPort (qp->dport);
	udpHeader.SetSourcePort (qp->sport);
	p->AddHeader (udpHeader);
	Ipv4Header ipHeader;
	ipHeader.SetSource (qp->sip);
	ipHeader.SetDestination (qp->dip);
	ipHeader.SetProtocol (0x11);
	ipHeader.SetPayloadSize (p->GetSize());
	ipHeader.SetTtl (64);
	ipHeader.SetTos (0);
	ipHeader.SetIdentification (qp->m_ipid);
	p->AddHeader(ipHeader);
	PppHeader ppp;
	ppp.SetProtocol (0x0021);
	p->AddHeader (ppp);

qp->snd_nxt += payload_size;
	qp->m_ipid++;

return p;
}

void TcpDataCenter::PktSent(Ptr<RdmaQueuePair> qp, Ptr<Packet> pkt, Time interframeGap) {
	qp->lastPktSize = pkt->GetSize();
	uint32_t seq = qp->snd_nxt;
	qp->rates[qp->snd_nxt] = Simulator::Now().GetNanoSeconds();
	UpdateNextAvail(qp, interframeGap, pkt->GetSize());

}

void TcpDataCenter::UpdateNextAvail(Ptr<RdmaQueuePair> qp, Time interframeGap, uint32_t pkt_size) {
	Time sendingTime;
	if (m_rateBound)
		sendingTime = interframeGap + qp->m_rate.CalculateBytesTxTime(pkt_size);
	else
		sendingTime = interframeGap + qp->m_max_rate.CalculateBytesTxTime(pkt_size);
	qp->m_nextAvail = Simulator::Now() + sendingTime;
}

void TcpDataCenter::ChangeRate(Ptr<RdmaQueuePair> qp, DataRate new_rate) {
#if 1
	Time sendingTime = qp->m_rate.CalculateBytesTxTime(qp->lastPktSize);
	Time new_sendintTime = new_rate.CalculateBytesTxTime(qp->lastPktSize);
	qp->m_nextAvail = qp->m_nextAvail + new_sendintTime - sendingTime;
	uint32_t nic_idx = GetNicIdxOfQp(qp);
	m_nic[nic_idx].dev->UpdateNextAvail(qp->m_nextAvail);
#endif
	qp->m_rate = new_rate;
}

}

Texto modificado

Abrir archivo

#include "tcp-datacenter.h"

namespace ns3 {

uint32_t nic_idx = GetNicIdxOfQp(qp);
	m_nic[nic_idx].qpGrp->AddQp(qp);
	uint64_t key = GetQpKey(dip.Get(), sport, pg);
	m_qpMap[key] = qp;

qp->powerEnabled = true;

if (m_nic[nic_idx].dev == NULL) {
		std::cout << "sip " << sip << " dip " << dip << " sport " << sport  << " dport " << dport << std::endl;
	}
	DataRate m_bps = m_nic[nic_idx].dev->GetDataRate();
	if(win)
		qp->SetWin(m_bps.GetBitRate() * 1.0 * baseRtt * 1e-9 / 8);
	qp->m_rate = DataRate(m_bps.GetBitRate() / 10);
	qp->m_max_rate = m_bps;
    
    qp->hp.m_curRate = DataRate(m_bps.GetBitRate() / 10);

m_nic[nic_idx].dev->NewQp(qp);
}

int TcpDataCenter::ReceiveCnp(Ptr<Packet> p, CustomHeader &ch) {
	uint32_t qIndex = ch.cnp.qIndex;
	if (qIndex == 1) {
		std::cout << "TCP--ignore\n";
		return 0;
	}
	uint16_t udpport = ch.cnp.fid;

if (qp->m_rate == 0)
	{
		qp->m_rate = dev->GetDataRate();
        qp->hp.m_curRate = dev->GetDataRate();
	}
	return 0;
}

HandleAckHp(qp, p, ch);

dev->TriggerTransmit();
	return 0;
}

#define PRINT_LOG 0

void TcpDataCenter::HandleAckHp(Ptr<RdmaQueuePair> qp, Ptr<Packet> p, CustomHeader &ch) {
	uint32_t ack_seq = ch.ack.seq;

if (ack_seq > qp->hp.m_lastUpdateSeq) {
		UpdateRatePower(qp, p, ch, false);
	} else {
		FastReactPower(qp, p, ch);
	}
}

double dt_sample = prevCompletion - qp->prevCompletion;
        if (dt_sample < 1.0) dt_sample = 1.0;
        double A = ( double(prevRtt - qp->prevRtt) / dt_sample + 1  );
        if (A < 0.5) A = 0.5;
        double power = ( A ) * (prevRtt);
        double powerx = (power) / (1.03 * qp->m_baseRtt);
        double u = powerx;

uint8_t cnp = (ch.ack.flags >> qbbHeader::FLAG_CNP) & 1;
        if (cnp) {
             if (u < 2.0) u = 2.0;
        }

if (u > U) {
            U = u;
            dt = (uint64_t)dt_sample;
        }

DataRate new_rate;

if (dt > 1.0 * qp->m_baseRtt)
            dt = 1.0 * qp->m_baseRtt;

if (U < 0) {
            U = qp->hp.u;
        }
        qp->hp.u = (qp->hp.u * (1.0 * qp->m_baseRtt - dt) + U * dt) / double(1.0 * qp->m_baseRtt);

max_c = qp->hp.u;
        new_rate = (0.8 * ( qp->hp.m_curRate / max_c + DataRate("250Mbps") ) + 0.2 * qp->hp.m_curRate);