Compare text

Find the difference between two text files

Real-time editor

Hide whitespace changes

Hide unchanged lines

Disable line wrap

Layout

Diff precision

Text styles

Syntax highlighting

Diffchecker Desktop The most secure way to run Diffchecker. Get the Diffchecker Desktop app: your diffs never leave your computer!Get Desktop

EvolveDatacenterTCPDiff

Created 6 months agoDiff never expires

8 removals

Lines
Total
Removed

Characters
Total
Removed

To continue using this feature, upgrade to Diffchecker Pro View Pricing

383 lines

19 additions

Lines
Total
Added

Characters
Total
Added

To continue using this feature, upgrade to Diffchecker Pro View Pricing

389 lines

#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;

}

Saved diffs

Original text

Open file

#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;
}

}

Changed text

Open file

#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);