Texte vergleichen

Finde den Unterschied zwischen zwei Textdateien

Live-Editor

Leerzeichen ausblenden

Gleiches ausblenden

Zeilenumbruch aus

Ansicht

Vergleichsgenauigkeit

Textstile

Syntaxhervorhebung

Diffchecker Desktop Der sicherste Weg, Diffchecker zu nutzen. Hol dir die Desktop-App: Deine Diffs verlassen nie deinen Computer!Desktop holen

EvolveDatacenterTCPDiff

Erstellt vor 6 MonatenDiff läuft nie ab

8 Entfernungen

Zeilen
Gesamt
Entfernt

Zeichen
Gesamt
Entfernt

Um diese Funktion weiterhin zu nutzen, aktualisiere auf Diffchecker Pro Preise anzeigen

383 Zeilen

19 Hinzufügungen

Zeilen
Gesamt
Hinzugefügt

Zeichen
Gesamt
Hinzugefügt

Um diese Funktion weiterhin zu nutzen, aktualisiere auf Diffchecker Pro Preise anzeigen

389 Zeilen

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

}

Gespeicherte Diffs

Originaltext

Datei öffnen

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

}

Bearbeitung

Datei öffnen

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