ndn-subscription-consumer.cpp

/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
/**
 * Copyright (c) 2011-2015  Regents of the University of California.
 *
 * This file is part of ndnSIM. See AUTHORS for complete list of ndnSIM authors and
 * contributors.
 *
 * ndnSIM is free software: you can redistribute it and/or modify it under the terms
 * of the GNU General Public License as published by the Free Software Foundation,
 * either version 3 of the License, or (at your option) any later version.
 *
 * ndnSIM 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
 * ndnSIM, e.g., in COPYING.md file.  If not, see <http://www.gnu.org/licenses/>.
 **/

#include "ns3/ptr.h"
#include "ns3/log.h"
#include "ns3/simulator.h"
#include "ns3/packet.h"
#include "ns3/callback.h"
#include "ns3/string.h"
#include "ns3/boolean.h"
#include "ns3/uinteger.h"
#include "ns3/integer.h"
#include "ns3/double.h"

#include <string>
#include <algorithm>
#include <set>
#include "ndn-subscription-consumer.hpp"
#include <ndn-cxx/lp/tags.hpp>
#include "ndn-cxx/util/time.hpp"

namespace ns3 {
namespace ndn {

NS_LOG_COMPONENT_DEFINE("ndn.SubscriptionConsumer");

NS_OBJECT_ENSURE_REGISTERED(SubscriptionConsumer);

TypeId
SubscriptionConsumer::GetTypeId(void)
{
  static TypeId tid =
    TypeId("ns3::ndn::SubscriptionConsumer")
      .SetGroupName("Ndn")
      .SetParent<Consumer>()
      .AddConstructor<SubscriptionConsumer>()
      .AddAttribute("SubscriptionInterval", "Interval of subscription interest", TimeValue(Seconds(1.0)),
										MakeTimeAccessor(&SubscriptionConsumer::m_subscriptionInterval), MakeTimeChecker())
			.AddTraceSource("ConsumerStreamsReused",
										  "Number of consumer streams detected as already existing or equivalent to existing ones",
										  MakeTraceSourceAccessor (&SubscriptionConsumer::interestSatisfactionRatio),
											"ns3::TracedValueCallback::Double")
			.AddTraceSource("ValidSubCount",
											"Number of consumer streams answered by the producers",
											MakeTraceSourceAccessor (&SubscriptionConsumer::validSubCountPerc),
											"ns3::TracedValueCallback::Double")
			.AddTraceSource("FirstReceivedDataDelay",
											"Time between the application start and the time the first data was received",
											MakeTraceSourceAccessor (&SubscriptionConsumer::firstDataDelay),
											"ns3::TracedValueCallback::Time");

  return tid;
}

SubscriptionConsumer::SubscriptionConsumer()
: Consumer(), subscriptionCount(1), hasReceivedFirstdata(false) {}

void
SubscriptionConsumer::StartApplication()
{
	startTime = Simulator::Now();
	hasReceivedFirstdata = false;
	firstDataDelay = Seconds(0);
	Consumer::StartApplication();
}

size_t SubscriptionConsumer::globalSubscriptionCount = 0;

void
SubscriptionConsumer::ScheduleNextPacket()
{
  NS_LOG_FUNCTION_NOARGS();

  SendPacket();
}

void
SubscriptionConsumer::SendPacket()
{
  NS_LOG_FUNCTION_NOARGS();

  if (!m_active)
    return;

  Ptr<UniformRandomVariable> rand = CreateObject<UniformRandomVariable>();

  if (randomInterestPrefixes.empty())
	  {
		  subscriptionCount = 1;
		  randomInterestPrefixes.resize(1);
		  randomInterestPrefixes[0] = m_interestName;
	  }
  auto rng = CreateObject<UniformRandomVariable>();
  BOOST_ASSERT_MSG(subscriptionCount > 0 , "Subscription count must be set");
  NS_LOG_DEBUG("randomBrokerPrefixes: " << randomInterestPrefixes.size());
  NS_LOG_DEBUG("subscriptionCount: " << subscriptionCount);
//  globalSubscriptionCount += subscriptionCount;
//  NS_LOG_UNCOND("globalSubscriptionCount: " << globalSubscriptionCount);

  for (int i=0; i<subscriptionCount; i++)
  {
  	uint prefixIdx;
  	do { prefixIdx = rng->GetInteger(0, randomInterestPrefixes.size()-1); }
  	while (pickedPrefixes.find(prefixIdx) != pickedPrefixes.end());
  	pickedPrefixes.insert(prefixIdx);

  	auto it = randomInterestPrefixes.begin();
  	std::advance(it, prefixIdx);
	  shared_ptr<Name> nameWithTransformation = make_shared<Name>(*it);
	  if ( useParams || !transformationsAndParameters.empty())
		{
		  if(!transformationsAndParameters.empty())
			{
			  params.clearTranformations();
			  params.clearTransformationParams();
			  for (const auto& transformationAndParams : this->transformationsAndParameters)
				{
				  params.addTransformation(transformationAndParams.first);
				  for (const auto& paramAndChoices : transformationAndParams.second)
					{
					  uint i = rand->GetInteger(0, paramAndChoices.second.size()-1);
					  params.addTransformationParam(transformationAndParams.first,
																					paramAndChoices.first, paramAndChoices.second.at(i));
					}
				}
			}
		  params.appendTransformationsToName(nameWithTransformation);
		}

//	  toBeReceived.push_back(*nameWithTransformation);
	  shared_ptr<Interest> interest = make_shared<Interest>(*nameWithTransformation);
	  interest->setNonce(m_rand->GetValue(0, std::numeric_limits<uint32_t>::max()));
	  interest->setCanBePrefix(true);
	  time::milliseconds subInterval(m_subscriptionInterval.GetMilliSeconds());
	  interest->setSubscription(subInterval);
	  time::milliseconds interestLifeTime(m_interestLifeTime.GetMilliSeconds());
	  interest->setInterestLifetime(interestLifeTime);

	  if ( useParams ) interest->setParameters(params.wireEncode());
//	  NS_LOG_INFO ("Requesting Interest: \n" << *interest);

//	  WillSendOutInterest(i);

	  NS_LOG_DEBUG("Sending " << interest->getName() << " " << m_subscriptionInterval.GetSeconds() << "sec");
	  m_transmittedInterests(interest, this, m_face);
	  m_appLink->onReceiveInterest(*interest);
	  avgDataDelay[*nameWithTransformation] = 0;
	  localPrefixNames[*nameWithTransformation] = 0;
  }
  interestTransmission = time::steady_clock::now();
}

SubscriptionConsumer::randomTransformations
	SubscriptionConsumer::transformationsAndParameters;

void
SubscriptionConsumer::setRandomTransformations
	(SubscriptionConsumer::randomTransformations randomStuff)
{
  transformationsAndParameters = randomStuff;
}

SubscriptionConsumer::InterestPrefixes
	SubscriptionConsumer::randomInterestPrefixes;

void
SubscriptionConsumer::setRandomInterestPrefixes(InterestPrefixes prefAndCounts)
{
  randomInterestPrefixes = prefAndCounts;
}

SubscriptionConsumer::InterestPrefixes
SubscriptionConsumer::getRandomInterestPrefixes()
{
	return randomInterestPrefixes;
}

void
SubscriptionConsumer::addPrefix(Name prefix)
{
	prefixList.push_back(prefix);
}

void
SubscriptionConsumer::resetPrefixList()
{
	prefixList.clear();
}

void
SubscriptionConsumer::OnData(shared_ptr<const Data> data)
{
	NS_LOG_FUNCTION(data->getName() << m_subscriptionInterval.GetMinutes());
	auto it = std::find_if(localPrefixNames.begin(), localPrefixNames.end(), [&](const std::pair<Name, uint> item){
		return item.first.isPrefixOf(data->getName());
	});
	BOOST_ASSERT(it != localPrefixNames.end());

	auto data_timestamp = data->getName().at(-1).toSteadyClockTimeStamp();
	long time_since_transmission = time::duration_cast<time::milliseconds>(data_timestamp - interestTransmission).count();
	long time_until_next = time_since_transmission / m_subscriptionInterval.GetMilliSeconds();
	long drift = std::min(time_until_next, (long)m_subscriptionInterval.GetMilliSeconds() - time_until_next);
//	NS_LOG_UNCOND("Drift: " << drift << "ms");
	BOOST_ASSERT(drift < 10000);


	localPrefixNames[it->first] += 1;
	avgDataDelay[it->first] += time::duration_cast<time::milliseconds>(time::steady_clock::now() - data_timestamp).count();

	receivedDatapkts++;
	if (!hasReceivedFirstdata)
		{
			hasReceivedFirstdata = true;
			firstDataDelay = Simulator::Now() - startTime;
			firstDataTime = Simulator::Now();
		}

//	if (!toBeReceived.empty())
//		{
//			Time currOperationTime = Simulator::Now() - startTime;
////			if (currOperationTime > 4 * m_subscriptionInterval)
////				{
////					BOOST_THROW_EXCEPTION(std::logic_error("Too much data has been lost on node " + ::ndn::to_string(m_node->GetId()) + " missing " + ::ndn::to_string(toBeReceived.size()) + "sources"));
////				}
//			auto it = std::find_if(toBeReceived.begin(), toBeReceived.end(),
//				  		   [&](const Name& name) { return name.isPrefixOf(data->getName()); });
//			if (it != toBeReceived.end())
//				{
//					toBeReceived.erase(it);
//					NS_LOG_DEBUG("First data from " << *it << " received, " << ::ndn::to_string(toBeReceived.size()) << " remaining");
//				}
//		}

	App::OnData(data);
}

void
SubscriptionConsumer::OnTimeout(uint32_t sequenceNumber)
{
	BOOST_THROW_EXCEPTION(std::logic_error("There should not be interest timeouts"));
	Consumer::OnTimeout(sequenceNumber);
}

void
SubscriptionConsumer::StopApplication()
{
	NS_LOG_FUNCTION_NOARGS();

	if (hasReceivedFirstdata)
		{
			const Time streamDuration = Simulator::Now() - firstDataTime;

			const uint validSubCount = std::count_if(localPrefixNames.begin(), localPrefixNames.end(),
																							 [](const auto& item) { return item.second > 0; });
			this->validSubCountPerc = (100.0 * validSubCount)/subscriptionCount;

			const long expectedreceivedDatas = std::ceil(streamDuration / m_subscriptionInterval) + 1;// - acceptableLoss;

			for (auto it=localPrefixNames.begin(); it!=localPrefixNames.end(); it++)
				{
					interestSatisfactionRatio += 100.0 * it->second / expectedreceivedDatas;
					avgDataDelay[it->first] /= it->second;
				}

			interestSatisfactionRatio /= validSubCount;
		}
	App::StopApplication();
}

double
SubscriptionConsumer::getInterestSatisfactionRatio() const
{
	return interestSatisfactionRatio.Get();
}

double
SubscriptionConsumer::getValidSubCountPerc() const
{
	return validSubCountPerc.Get();
}

Time
SubscriptionConsumer::getFirstDataDelay() const
{
	if (!hasReceivedFirstdata)
		NS_LOG_LOGIC("Has not received any data");
	return firstDataDelay;
}

bool
SubscriptionConsumer::receivedFirstData() const
{
	return hasReceivedFirstdata;
}

void
SubscriptionConsumer::setSubscriptionCount(size_t count)
{
	subscriptionCount = count;
}

size_t
SubscriptionConsumer::getSubscriptionCount() const
{
	return subscriptionCount;
}

std::set<uint>
SubscriptionConsumer::getPickedPrefixes () const
{
	return pickedPrefixes;
}

Time
SubscriptionConsumer::GetSubInterval() const
{
	return m_subscriptionInterval;
}

std::map<Name, double>
SubscriptionConsumer::getAvgDataDelay() const
{
	return avgDataDelay;
}

} // namespace ndn
} // namespace ns3