Rietveld Code Review Tool
Help | Bug tracker | Discussion group | Source code | Sign in
(1471)

Unified Diff: src/mpi/distributed-simulator-impl.cc

Issue 109068: Ns-3 MPI Distributed Simulation (Closed)
Patch Set: Fix valgrind and merged Created 14 years ago
Use n/p to move between diff chunks; N/P to move between comments. Please Sign in to add in-line comments.
Jump to:
View side-by-side diff with in-line comments
Download patch
« no previous file with comments | « src/mpi/distributed-simulator-impl.h ('k') | src/mpi/mpi-interface.h » ('j') | no next file with comments »
Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
Index: src/mpi/distributed-simulator-impl.cc
===================================================================
new file mode 100644
--- /dev/null
+++ b/src/mpi/distributed-simulator-impl.cc
@@ -0,0 +1,521 @@
+/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
+/*
+ * Copyright (c) 2005,2006 INRIA
+ *
+ * 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
+ *
+ * Author: George Riley <riley@ece.gatech.edu>
+ */
+
+#include "distributed-simulator-impl.h"
+#include "mpi-interface.h"
+
+#include "ns3/simulator.h"
+#include "ns3/scheduler.h"
+#include "ns3/event-impl.h"
+#include "ns3/channel.h"
+#include "ns3/node-container.h"
+#include "ns3/ptr.h"
+#include "ns3/pointer.h"
+#include "ns3/assert.h"
+#include "ns3/log.h"
+
+#include <math.h>
+
+#ifdef NS3_MPI
+#include <mpi.h>
+#endif
+
+NS_LOG_COMPONENT_DEFINE ("DistributedSimulatorImpl");
+
+namespace ns3 {
+
+NS_OBJECT_ENSURE_REGISTERED (DistributedSimulatorImpl);
+
+LbtsMessage::~LbtsMessage ()
+{
+}
+
+Time
+LbtsMessage::GetSmallestTime ()
+{
+ return m_smallestTime;
+}
+
+uint32_t
+LbtsMessage::GetTxCount ()
+{
+ return m_txCount;
+}
+
+uint32_t
+LbtsMessage::GetRxCount ()
+{
+ return m_rxCount;
+}
+uint32_t
+LbtsMessage::GetMyId ()
+{
+ return m_myId;
+}
+
+Time DistributedSimulatorImpl::m_lookAhead = Seconds (0);
+
+TypeId
+DistributedSimulatorImpl::GetTypeId (void)
+{
+ static TypeId tid = TypeId ("ns3::DistributedSimulatorImpl")
+ .SetParent<Object> ()
+ .AddConstructor<DistributedSimulatorImpl> ()
+ ;
+ return tid;
+}
+
+DistributedSimulatorImpl::DistributedSimulatorImpl ()
+{
+#ifdef NS3_MPI
+ m_myId = MpiInterface::GetSystemId ();
+ m_systemCount = MpiInterface::GetSize ();
+
+ // Allocate the LBTS message buffer
+ m_pLBTS = new LbtsMessage[m_systemCount];
+ m_grantedTime = Seconds (0);
+#else
+ NS_FATAL_ERROR ("Can't use distributed simulator without MPI compiled in");
+#endif
+
+ m_stop = false;
+ // uids are allocated from 4.
+ // uid 0 is "invalid" events
+ // uid 1 is "now" events
+ // uid 2 is "destroy" events
+ m_uid = 4;
+ // before ::Run is entered, the m_currentUid will be zero
+ m_currentUid = 0;
+ m_currentTs = 0;
+ m_currentContext = 0xffffffff;
+ m_unscheduledEvents = 0;
+ m_events = 0;
+}
+
+DistributedSimulatorImpl::~DistributedSimulatorImpl ()
+{
+}
+
+void
+DistributedSimulatorImpl::DoDispose (void)
+{
+ while (!m_events->IsEmpty ())
+ {
+ Scheduler::Event next = m_events->RemoveNext ();
+ next.impl->Unref ();
+ }
+ m_events = 0;
+ delete [] m_pLBTS;
+ SimulatorImpl::DoDispose ();
+}
+
+void
+DistributedSimulatorImpl::Destroy ()
+{
+ while (!m_destroyEvents.empty ())
+ {
+ Ptr<EventImpl> ev = m_destroyEvents.front ().PeekEventImpl ();
+ m_destroyEvents.pop_front ();
+ NS_LOG_LOGIC ("handle destroy " << ev);
+ if (!ev->IsCancelled ())
+ {
+ ev->Invoke ();
+ }
+ }
+
+ MpiInterface::Destroy ();
+}
+
+
+void
+DistributedSimulatorImpl::CalculateLookAhead (void)
+{
+#ifdef NS3_MPI
+ if (MpiInterface::GetSize () <= 1)
+ {
+ DistributedSimulatorImpl::m_lookAhead = Seconds (0);
+ m_grantedTime = Seconds (0);
+ }
+ else
+ {
+ NodeContainer c = NodeContainer::GetGlobal ();
+ for (NodeContainer::Iterator iter = c.Begin (); iter != c.End (); ++iter)
+ {
+ if ((*iter)->GetSystemId () != MpiInterface::GetSystemId ())
+ {
+ continue;
+ }
+
+ for (uint32_t i = 0; i < (*iter)->GetNDevices (); ++i)
+ {
+ Ptr<NetDevice> localNetDevice = (*iter)->GetDevice (i);
+ // only works for p2p links currently
+ if (!localNetDevice->IsPointToPoint ())
+ {
+ continue;
+ }
+ Ptr<Channel> channel = localNetDevice->GetChannel ();
+ if (channel == 0)
+ {
+ continue;
+ }
+
+ // grab the adjacent node
+ Ptr<Node> remoteNode;
+ if (channel->GetDevice (0) == localNetDevice)
+ {
+ remoteNode = (channel->GetDevice (1))->GetNode ();
+ }
+ else
+ {
+ remoteNode = (channel->GetDevice (0))->GetNode ();
+ }
+
+ // if it's not remote, don't consider it
+ if (remoteNode->GetSystemId () == MpiInterface::GetSystemId ())
+ {
+ continue;
+ }
+
+ // compare delay on the channel with current value of
+ // m_lookAhead. if delay on channel is smaller, make
+ // it the new lookAhead.
+ TimeValue delay;
+ channel->GetAttribute ("Delay", delay);
+ if (DistributedSimulatorImpl::m_lookAhead.IsZero ())
+ {
+ DistributedSimulatorImpl::m_lookAhead = delay.Get ();
+ m_grantedTime = delay.Get ();
+ }
+ if (delay.Get ().GetSeconds () < DistributedSimulatorImpl::m_lookAhead.GetSeconds ())
+ {
+ DistributedSimulatorImpl::m_lookAhead = delay.Get ();
+ m_grantedTime = delay.Get ();
+ }
+ }
+ }
+ }
+#else
+ NS_FATAL_ERROR ("Can't use distributed simulator without MPI compiled in");
+#endif
+}
+
+void
+DistributedSimulatorImpl::SetScheduler (ObjectFactory schedulerFactory)
+{
+ Ptr<Scheduler> scheduler = schedulerFactory.Create<Scheduler> ();
+
+ if (m_events != 0)
+ {
+ while (!m_events->IsEmpty ())
+ {
+ Scheduler::Event next = m_events->RemoveNext ();
+ scheduler->Insert (next);
+ }
+ }
+ m_events = scheduler;
+}
+
+void
+DistributedSimulatorImpl::ProcessOneEvent (void)
+{
+ Scheduler::Event next = m_events->RemoveNext ();
+
+ NS_ASSERT (next.key.m_ts >= m_currentTs);
+ m_unscheduledEvents--;
+
+ NS_LOG_LOGIC ("handle " << next.key.m_ts);
+ m_currentTs = next.key.m_ts;
+ m_currentContext = next.key.m_context;
+ m_currentUid = next.key.m_uid;
+ next.impl->Invoke ();
+ next.impl->Unref ();
+}
+
+bool
+DistributedSimulatorImpl::IsFinished (void) const
+{
+ return m_events->IsEmpty () || m_stop;
+}
+
+uint64_t
+DistributedSimulatorImpl::NextTs (void) const
+{
+ NS_ASSERT (!m_events->IsEmpty ());
+ Scheduler::Event ev = m_events->PeekNext ();
+ return ev.key.m_ts;
+}
+
+Time
+DistributedSimulatorImpl::Next (void) const
+{
+ return TimeStep (NextTs ());
+}
+
+void
+DistributedSimulatorImpl::Run (void)
+{
+#ifdef NS3_MPI
+ CalculateLookAhead ();
+ m_stop = false;
+ while (!m_events->IsEmpty () && !m_stop)
+ {
+ Time nextTime = Next ();
+ if (nextTime > m_grantedTime)
+ { // Can't process, calculate a new LBTS
+ // First receive any pending messages
+ MpiInterface::ReceiveMessages ();
+ // reset next time
+ nextTime = Next ();
+ // And check for send completes
+ MpiInterface::TestSendComplete ();
+ // Finally calculate the lbts
+ LbtsMessage lMsg (MpiInterface::GetRxCount (), MpiInterface::GetTxCount (), m_myId, nextTime);
+ m_pLBTS[m_myId] = lMsg;
+ MPI_Allgather (&lMsg, sizeof (LbtsMessage), MPI_BYTE, m_pLBTS,
+ sizeof (LbtsMessage), MPI_BYTE, MPI_COMM_WORLD);
+ Time smallestTime = m_pLBTS[0].GetSmallestTime ();
+ // The totRx and totTx counts insure there are no transient
+ // messages; If totRx != totTx, there are transients,
+ // so we don't update the granted time.
+ uint32_t totRx = m_pLBTS[0].GetRxCount ();
+ uint32_t totTx = m_pLBTS[0].GetTxCount ();
+
+ for (uint32_t i = 1; i < m_systemCount; ++i)
+ {
+ if (m_pLBTS[i].GetSmallestTime () < smallestTime)
+ {
+ smallestTime = m_pLBTS[i].GetSmallestTime ();
+ }
+ totRx += m_pLBTS[i].GetRxCount ();
+ totTx += m_pLBTS[i].GetTxCount ();
+
+ }
+ if (totRx == totTx)
+ {
+ m_grantedTime = smallestTime + DistributedSimulatorImpl::m_lookAhead;
+ }
+ }
+ if (nextTime <= m_grantedTime)
+ { // Save to process
+ ProcessOneEvent ();
+ }
+ }
+
+ // If the simulator stopped naturally by lack of events, make a
+ // consistency test to check that we didn't lose any events along the way.
+ NS_ASSERT (!m_events->IsEmpty () || m_unscheduledEvents == 0);
+#else
+ NS_FATAL_ERROR ("Can't use distributed simulator without MPI compiled in");
+#endif
+}
+
+uint32_t DistributedSimulatorImpl::GetSystemId () const
+{
+ return m_myId;
+}
+
+void
+DistributedSimulatorImpl::RunOneEvent (void)
+{
+ ProcessOneEvent ();
+}
+
+void
+DistributedSimulatorImpl::Stop (void)
+{
+ m_stop = true;
+}
+
+void
+DistributedSimulatorImpl::Stop (Time const &time)
+{
+ Simulator::Schedule (time, &Simulator::Stop);
+}
+
+//
+// Schedule an event for a _relative_ time in the future.
+//
+EventId
+DistributedSimulatorImpl::Schedule (Time const &time, EventImpl *event)
+{
+ Time tAbsolute = time + TimeStep (m_currentTs);
+
+ NS_ASSERT (tAbsolute.IsPositive ());
+ NS_ASSERT (tAbsolute >= TimeStep (m_currentTs));
+ Scheduler::Event ev;
+ ev.impl = event;
+ ev.key.m_ts = (uint64_t) tAbsolute.GetTimeStep ();
+ ev.key.m_context = GetContext ();
+ ev.key.m_uid = m_uid;
+ m_uid++;
+ m_unscheduledEvents++;
+ m_events->Insert (ev);
+ return EventId (event, ev.key.m_ts, ev.key.m_context, ev.key.m_uid);
+}
+
+void
+DistributedSimulatorImpl::ScheduleWithContext (uint32_t context, Time const &time, EventImpl *event)
+{
+ NS_LOG_FUNCTION (this << context << time.GetTimeStep () << m_currentTs << event);
+
+ Scheduler::Event ev;
+ ev.impl = event;
+ ev.key.m_ts = m_currentTs + time.GetTimeStep ();
+ ev.key.m_context = context;
+ ev.key.m_uid = m_uid;
+ m_uid++;
+ m_unscheduledEvents++;
+ m_events->Insert (ev);
+}
+
+EventId
+DistributedSimulatorImpl::ScheduleNow (EventImpl *event)
+{
+ Scheduler::Event ev;
+ ev.impl = event;
+ ev.key.m_ts = m_currentTs;
+ ev.key.m_context = GetContext ();
+ ev.key.m_uid = m_uid;
+ m_uid++;
+ m_unscheduledEvents++;
+ m_events->Insert (ev);
+ return EventId (event, ev.key.m_ts, ev.key.m_context, ev.key.m_uid);
+}
+
+EventId
+DistributedSimulatorImpl::ScheduleDestroy (EventImpl *event)
+{
+ EventId id (Ptr<EventImpl> (event, false), m_currentTs, 0xffffffff, 2);
+ m_destroyEvents.push_back (id);
+ m_uid++;
+ return id;
+}
+
+Time
+DistributedSimulatorImpl::Now (void) const
+{
+ return TimeStep (m_currentTs);
+}
+
+Time
+DistributedSimulatorImpl::GetDelayLeft (const EventId &id) const
+{
+ if (IsExpired (id))
+ {
+ return TimeStep (0);
+ }
+ else
+ {
+ return TimeStep (id.GetTs () - m_currentTs);
+ }
+}
+
+void
+DistributedSimulatorImpl::Remove (const EventId &id)
+{
+ if (id.GetUid () == 2)
+ {
+ // destroy events.
+ for (DestroyEvents::iterator i = m_destroyEvents.begin (); i != m_destroyEvents.end (); i++)
+ {
+ if (*i == id)
+ {
+ m_destroyEvents.erase (i);
+ break;
+ }
+ }
+ return;
+ }
+ if (IsExpired (id))
+ {
+ return;
+ }
+ Scheduler::Event event;
+ event.impl = id.PeekEventImpl ();
+ event.key.m_ts = id.GetTs ();
+ event.key.m_context = id.GetContext ();
+ event.key.m_uid = id.GetUid ();
+ m_events->Remove (event);
+ event.impl->Cancel ();
+ // whenever we remove an event from the event list, we have to unref it.
+ event.impl->Unref ();
+
+ m_unscheduledEvents--;
+}
+
+void
+DistributedSimulatorImpl::Cancel (const EventId &id)
+{
+ if (!IsExpired (id))
+ {
+ id.PeekEventImpl ()->Cancel ();
+ }
+}
+
+bool
+DistributedSimulatorImpl::IsExpired (const EventId &ev) const
+{
+ if (ev.GetUid () == 2)
+ {
+ if (ev.PeekEventImpl () == 0
+ || ev.PeekEventImpl ()->IsCancelled ())
+ {
+ return true;
+ }
+ // destroy events.
+ for (DestroyEvents::const_iterator i = m_destroyEvents.begin (); i != m_destroyEvents.end (); i++)
+ {
+ if (*i == ev)
+ {
+ return false;
+ }
+ }
+ return true;
+ }
+ if (ev.PeekEventImpl () == 0
+ || ev.GetTs () < m_currentTs
+ || (ev.GetTs () == m_currentTs
+ && ev.GetUid () <= m_currentUid)
+ || ev.PeekEventImpl ()->IsCancelled ())
+ {
+ return true;
+ }
+ else
+ {
+ return false;
+ }
+}
+
+Time
+DistributedSimulatorImpl::GetMaximumSimulationTime (void) const
+{
+ // XXX: I am fairly certain other compilers use other non-standard
+ // post-fixes to indicate 64 bit constants.
+ return TimeStep (0x7fffffffffffffffLL);
+}
+
+uint32_t
+DistributedSimulatorImpl::GetContext (void) const
+{
+ return m_currentContext;
+}
+
+} // namespace ns3
« no previous file with comments | « src/mpi/distributed-simulator-impl.h ('k') | src/mpi/mpi-interface.h » ('j') | no next file with comments »

Powered by Google App Engine
RSS Feeds Recent Issues | This issue
This is Rietveld f62528b