lengyijun
1/3/2019 - 3:23 PM
topology-tcp-piece.cc
/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
/*
* This program is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License version 2 as
* published by the Free Software Foundation;
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program; if not, write to the Free Software
* Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
*
*/
#include <ctime>
#include <sstream>
#include "ns3/core-module.h"
#include "ns3/network-module.h"
#include "ns3/internet-module.h"
#include "ns3/point-to-point-module.h"
#include "ns3/applications-module.h"
#include "ns3/ipv4-nix-vector-helper.h"
#include "ns3/topology-read-module.h"
#include <list>
#include "wifi-example-apps.h"
#include "ns3/netanim-module.h"
using namespace ns3;
NS_LOG_COMPONENT_DEFINE ("TopologyCreationExperiment");
const long writeSize=1072;
const long totaltotalTxBytes=4*1024*1024;
const long totalTxBytes=128*1024;
int count;
const int listenPort=12345;
void SendStuffWithTag (Ptr<Socket> sock, Ipv4Address dstaddr, uint16_t port,TimestampTag tag);
void BindSock (Ptr<Socket> sock, Ptr<NetDevice> netdev);
void dstSocketRecv (Ptr<Socket> socket);
void HandleAccept (Ptr<Socket> s, const Address& from);
void WriteUntilBufferFull (Ptr<Socket>, uint32_t);
void Init(Ptr<Node> n, Ipv4Address dstaddr, uint16_t port);
void SetFinTag();
void SetReqTag();
std::map<uint32_t,std::vector<Ipv4Address>> ipv4NeighMap;
std::map<uint32_t,std::vector<TimestampTag>> tagMap;
std::map<Ptr<Socket>,uint32_t> byteCountMap;
std::map<Ptr<Socket>,TimestampTag> socketTagMap;
TimestampTag FinTag; //tag the last piece of the 4M bulk
TimestampTag ReqTag; //request the latest bulk
int main (int argc, char *argv[])
{
count=totaltotalTxBytes/totalTxBytes;
std::string format ("Inet");
// std::string input ("src/topology-read/examples/double.txt"); //2
// std::string input ("src/topology-read/examples/triple.txt"); //3
// std::string input ("src/topology-read/examples/tree.txt"); //10
std::string input ("src/topology-read/examples/output.txt"); //1000
// std::string input ("src/topology-read/examples/Inet_toposample.txt"); //4000
// Set up command line parameters used to control the experiment.
CommandLine cmd;
cmd.AddValue ("format", "Format to use for data input [Orbis|Inet|Rocketfuel].",
format);
cmd.AddValue ("input", "Name of the input file.",
input);
cmd.Parse (argc, argv);
// ------------------------------------------------------------
// -- Read topology data.
// --------------------------------------------
// Pick a topology reader based in the requested format.
TopologyReaderHelper topoHelp;
topoHelp.SetFileName (input);
topoHelp.SetFileType (format);
Ptr<TopologyReader> inFile = topoHelp.GetTopologyReader ();
NodeContainer nodes;
if (inFile != 0)
{
nodes = inFile->Read ();
}
if (inFile->LinksSize () == 0)
{
NS_LOG_ERROR ("Problems reading the topology file. Failing.");
return -1;
}
// ------------------------------------------------------------
// -- Create nodes and network stacks
// --------------------------------------------
NS_LOG_INFO ("creating internet stack");
InternetStackHelper stack;
// Setup NixVector Routing
// Ipv4NixVectorHelper nixRouting;
// stack.SetRoutingHelper (nixRouting); // has effect on the next Install ()
stack.Install (nodes);
NS_LOG_INFO ("creating ip4 addresses");
Ipv4AddressHelper address;
address.SetBase ("10.0.0.0", "255.255.255.252");
int totlinks = inFile->LinksSize ();
NS_LOG_INFO ("creating node containers");
NodeContainer* nc = new NodeContainer[totlinks];
TopologyReader::ConstLinksIterator iter;
int i = 0;
for ( iter = inFile->LinksBegin (); iter != inFile->LinksEnd (); iter++, i++ )
{
nc[i] = NodeContainer (iter->GetFromNode (), iter->GetToNode ());
}
NS_LOG_INFO ("creating net device containers");
NetDeviceContainer* ndc = new NetDeviceContainer[totlinks];
PointToPointHelper p2p;
for (int i = 0; i < totlinks; i++)
{
// p2p.SetChannelAttribute ("Delay", TimeValue(MilliSeconds(weight[i])));
p2p.SetChannelAttribute ("Delay", StringValue ("2ms"));
p2p.SetDeviceAttribute ("DataRate", StringValue ("5Mbps"));
// p2p.SetDeviceAttribute ("Mtu", UintegerValue (65536));
ndc[i] = p2p.Install (nc[i]);
}
// it crates little subnets, one for each couple of nodes.
NS_LOG_INFO ("creating ipv4 interfaces");
Ipv4InterfaceContainer* ipic = new Ipv4InterfaceContainer[totlinks];
for (int i = 0; i < totlinks; i++)
{
ipic[i] = address.Assign (ndc[i]);
address.NewNetwork ();
}
NS_LOG_INFO ("creating ipv4NeighMap");
i=0;
for ( iter = inFile->LinksBegin (); iter != inFile->LinksEnd (); iter++, i++ )
{
uint32_t from=iter->GetFromNode()->GetId();
uint32_t to=iter->GetToNode()->GetId();
std::map<uint32_t,std::vector<Ipv4Address>>::iterator it= ipv4NeighMap.find(from);
if (it==ipv4NeighMap.end()){
std::vector<Ipv4Address> v;
v.push_back(ipic[i].GetAddress(1));
ipv4NeighMap.insert(std::make_pair(from,v));
}else{
ipv4NeighMap[from].push_back(ipic[i].GetAddress(1));
}
std::map<uint32_t,std::vector<Ipv4Address>>::iterator it1= ipv4NeighMap.find(to);
if (it1==ipv4NeighMap.end()){
std::vector<Ipv4Address> v;
v.push_back(ipic[i].GetAddress(0));
ipv4NeighMap.insert(std::make_pair(to,v));
}else{
ipv4NeighMap[to].push_back(ipic[i].GetAddress(0));
}
}
InetSocketAddress dst = InetSocketAddress (Ipv4Address::GetAny(),listenPort);
for ( unsigned int i = 0; i < nodes.GetN (); i++ )
{
Ptr<Socket> dstSocket = Socket::CreateSocket (nodes.Get(i), TypeId::LookupByName ("ns3::TcpSocketFactory"));
dstSocket->Bind (dst);
dstSocket->Listen();
dstSocket->SetAcceptCallback (
MakeNullCallback<bool, Ptr<Socket>, const Address &> (),
MakeCallback (&HandleAccept)
);
}
// AnimationInterface anim ("animation.xml");
// p2p.EnablePcapAll ("socket-bound-static-routing");
Simulator::Schedule(Seconds(0.01),&SetReqTag);
Simulator::Schedule(Seconds(0.02),&SetFinTag);
for(int i=0;i<count;i++){
Simulator::Schedule(Seconds (0.1+i*0.0001),&Init,nodes.Get(0),ipic[0].GetAddress(1),listenPort);
}
NS_LOG_INFO ("Run Simulation.");
Simulator::Run ();
Simulator::Destroy ();
delete[] ipic;
delete[] ndc;
delete[] nc;
NS_LOG_INFO ("Done.");
return 0;
}
void SetReqTag(){
ReqTag.SetTimestamp (Simulator::Now ());
}
void SetFinTag(){
FinTag.SetTimestamp (Simulator::Now ());
//FinTag != ReqTag
NS_ASSERT (FinTag.Compare(ReqTag)==1);
}
void Init( Ptr<Node> n,Ipv4Address dstaddr, uint16_t port)
{
Ptr<Socket> sendSocket = Socket::CreateSocket (n, TypeId::LookupByName ("ns3::TcpSocketFactory"));
sendSocket->SetAttribute("SndBufSize", UintegerValue(1024*1024));
sendSocket->Bind ();
TimestampTag initTag;
initTag.SetTimestamp (Simulator::Now ());
NS_ASSERT(initTag.Compare(ReqTag)==1);
NS_ASSERT(initTag.Compare(FinTag)==1);
Ptr<Packet> p = Create<Packet> ();
p->AddPaddingAtEnd (26);
p->AddByteTag(initTag);
sendSocket->Connect (InetSocketAddress (dstaddr,port));
sendSocket->Send(p);
return;
}
void HandleAccept (Ptr<Socket> s, const Address& from)
{
// NS_LOG_INFO("HandleAccept");
s->SetRecvCallback (MakeCallback (&dstSocketRecv));
}
void SendStuffWithTag(Ptr<Socket> sock, Ipv4Address dstaddr, uint16_t port,TimestampTag tag)
{
socketTagMap.insert(std::make_pair(sock,tag));
sock->Connect (InetSocketAddress (dstaddr,port));
sock->SetSendCallback (MakeCallback (&WriteUntilBufferFull));
WriteUntilBufferFull (sock,sock->GetTxAvailable ());
return;
}
void BindSock (Ptr<Socket> sock, Ptr<NetDevice> netdev)
{
sock->BindToNetDevice (netdev);
return;
}
void dstSocketRecv (Ptr<Socket> socket)
{
Address from;
Ptr<Packet> packet = socket->RecvFrom (from);
ByteTagIterator it=packet-> GetByteTagIterator() ;
TimestampTag tag;
//a piece of 4M data, has no tag
//we don't need to bother it
if(!it.HasNext()){
// NS_LOG_INFO("No tag");
return;
}
it.Next().GetTag(tag);
Ipv4Address ipv4from=InetSocketAddress::ConvertFrom (from).GetIpv4 ();
Ptr<Node> n=socket->GetNode();
//the ReqTag
//we should send the whole data back
if(tag.Compare(ReqTag)==0){
NS_LOG_INFO("ReqTag");
it.Next().GetTag(tag); //update tag to the request data tag
//this tag must be in the tagMap
NS_ASSERT(tag.Compare(ReqTag)==1);
NS_ASSERT(tag.Compare(FinTag)==1);
Ptr<Socket> sendSocket = Socket::CreateSocket (n, TypeId::LookupByName ("ns3::TcpSocketFactory"));
sendSocket->Connect (InetSocketAddress (ipv4from,listenPort));
sendSocket->SetAttribute("SndBufSize", UintegerValue(2*1024*1024)); //no need , too small
SendStuffWithTag (sendSocket,ipv4from ,listenPort, tag);
return;
}
uint32_t id=socket->GetNode()->GetId();
std::vector<Ipv4Address> v=ipv4NeighMap.find(id)->second;
//the FinTag
//the whole block is received
if(tag.Compare(FinTag)==0){
NS_LOG_INFO("FinTag");
it.Next().GetTag(tag);
NS_ASSERT(tag.Compare(ReqTag)==1);
NS_ASSERT(tag.Compare(FinTag)==1);
//broadcast new block
for(unsigned int i=0;i<v.size();i++){
if(v[i].IsEqual(ipv4from)){
continue; //don't send back
}
Ptr<Socket> sendSocket= Socket::CreateSocket (n, TypeId::LookupByName ("ns3::TcpSocketFactory"));
// sendSocket->SetAttribute("SndBufSize", UintegerValue(2*1024*1024)); //no need , too small
Ptr<Packet> p = Create<Packet> ();
p->AddPaddingAtEnd (26);
p->AddByteTag(tag);
sendSocket->Connect (InetSocketAddress (v[i],listenPort));
sendSocket->Send(p);
}
return;
}
NS_ASSERT(!it.HasNext());
//judge header is new or old
std::map<uint32_t,std::vector<TimestampTag>>::iterator tagit= tagMap.find(id);
if(tagit!=tagMap.end()){
std::vector<TimestampTag> tagvector= tagit->second;
for(unsigned int i=0;i<tagvector.size();i++){
if(tag.Compare(tagvector[i])==0){
//old tag
//this packet is a header message, and it tell me that the node has a packet which I have already
//so I'll not bother it
NS_LOG_INFO("The "<<id<<" node receive a packet with an old HEADER");
return;
}
}
}
NS_LOG_INFO(Simulator::Now()<<": The "<<id<<" node receive a packet with a NEW HEADER");
//deal with new header
tagMap[id].push_back(tag);
Ptr<Socket> sendSocket = Socket::CreateSocket (n, TypeId::LookupByName ("ns3::TcpSocketFactory"));
sendSocket->Connect (InetSocketAddress (ipv4from,listenPort));
//send a packet with tag and ReqTag
Ptr<Packet> p = Create<Packet> ();
p->AddPaddingAtEnd (26);
p->AddByteTag(ReqTag);
p->AddByteTag(tag);
sendSocket->Send(p);
return;
}
void WriteUntilBufferFull (Ptr<Socket> localSocket, uint32_t txSpace)
{
std::map<Ptr<Socket>,uint32_t>::iterator it=byteCountMap.find(localSocket);
if(it==byteCountMap.end()){
byteCountMap.insert(std::make_pair(localSocket,0));
}
while (byteCountMap[localSocket]< totalTxBytes && localSocket->GetTxAvailable () > 0)
{
std::cout<<"currentTxBytes: "<<byteCountMap[localSocket]<<";totalTxBytes: "<<totalTxBytes<<std::endl;
uint32_t left = totalTxBytes -byteCountMap[localSocket];
uint32_t dataOffset = byteCountMap[localSocket]% writeSize;
uint32_t toWrite = writeSize - dataOffset;
toWrite = std::min (toWrite, left);
toWrite = std::min (toWrite, localSocket->GetTxAvailable ());
Ptr<Packet> p = Create<Packet> ();
p->AddPaddingAtEnd (toWrite);
//the last packet
if(totalTxBytes==toWrite+byteCountMap[localSocket]){
NS_LOG_INFO("the last packet");
std::map<Ptr<Socket>,TimestampTag>::iterator it = socketTagMap.find(localSocket);
if(it!=socketTagMap.end()){
p->AddByteTag (FinTag);
p->AddByteTag (it->second);
}else{
NS_LOG_ERROR("CANNOT FOUND SOCKET KEY IN socketTagMap");
}
}
int amountSent = localSocket->Send (p);
if(amountSent < 0)
{
// we will be called again when new tx space becomes available.
NS_LOG_WARN("amountSent < 0");
return;
}
byteCountMap[localSocket]+= amountSent;
}
localSocket->Close ();
}
//----------------------------------------------------------------------
//-- TimestampTag
//------------------------------------------------------
TypeId
TimestampTag::GetTypeId (void)
{
static TypeId tid = TypeId ("TimestampTag")
.SetParent<Tag> ()
.AddConstructor<TimestampTag> ()
.AddAttribute ("Timestamp",
"Some momentous point in time!",
EmptyAttributeValue (),
MakeTimeAccessor (&TimestampTag::GetTimestamp),
MakeTimeChecker ())
;
return tid;
}
TypeId
TimestampTag::GetInstanceTypeId (void) const
{
return GetTypeId ();
}
uint32_t
TimestampTag::GetSerializedSize (void) const
{
return 8;
}
void
TimestampTag::Serialize (TagBuffer i) const
{
int64_t t = m_timestamp.GetNanoSeconds ();
i.Write ((const uint8_t *)&t, 8);
}
void
TimestampTag::Deserialize (TagBuffer i)
{
int64_t t;
i.Read ((uint8_t *)&t, 8);
m_timestamp = NanoSeconds (t);
}
void
TimestampTag::SetTimestamp (Time time)
{
m_timestamp = time;
}
Time
TimestampTag::GetTimestamp (void) const
{
return m_timestamp;
}
void
TimestampTag::Print (std::ostream &os) const
{
os << "t=" << m_timestamp;
}
//=0: equal
//=1 not equal
int TimestampTag::Compare(TimestampTag& t)const
{
int x=m_timestamp.Compare(t.GetTimestamp());
if(x==0){
return 0;
}else{
return 1;
}
}