simple-distributed-mpi-comm.cc

Go to the documentation of this file.

/*
 *  Copyright 2018. Lawrence Livermore National Security, LLC.
 *
 * 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: Steven Smith <smith84@llnl.gov>
 */
 
#include "mpi-test-fixtures.h"
 
#include "ns3/core-module.h"
#include "ns3/internet-stack-helper.h"
#include "ns3/ipv4-address-helper.h"
#include "ns3/ipv4-global-routing-helper.h"
#include "ns3/ipv4-list-routing-helper.h"
#include "ns3/ipv4-static-routing-helper.h"
#include "ns3/mpi-interface.h"
#include "ns3/network-module.h"
#include "ns3/nix-vector-helper.h"
#include "ns3/on-off-helper.h"
#include "ns3/packet-sink-helper.h"
#include "ns3/packet-sink.h"
#include "ns3/point-to-point-helper.h"
 
#include <mpi.h>
 
using namespace ns3;
 
NS_LOG_COMPONENT_DEFINE("SimpleDistributedMpiComm");
 
const int NS_COLOR = 1;
const int NOT_NS_COLOR = NS_COLOR + 1;
 
void
ReportRank(int color, MPI_Comm splitComm)
{
    int otherId = 0;
    int otherSize = 1;
 
    MPI_Comm_rank(splitComm, &otherId);
    MPI_Comm_size(splitComm, &otherSize);
 
    if (color == NS_COLOR)
    {
        RANK0COUT("ns-3 rank:  ");
    }
    else
    {
        RANK0COUT("Other rank: ");
    }
 
    RANK0COUTAPPEND("in MPI_COMM_WORLD: " << SinkTracer::GetWorldRank() << ":"
                                          << SinkTracer::GetWorldSize() << ", in splitComm: "
                                          << otherId << ":" << otherSize << std::endl);
 
} // ReportRank()
 
int
main(int argc, char* argv[])
{
    bool nix = true;
    bool nullmsg = false;
    bool tracing = false;
    bool init = false;
    bool verbose = false;
    bool testing = false;
 
    // Parse command line
    CommandLine cmd(__FILE__);
    cmd.AddValue("nix", "Enable the use of nix-vector or global routing", nix);
    cmd.AddValue("nullmsg",
                 "Enable the use of null-message synchronization (instead of granted time window)",
                 nullmsg);
    cmd.AddValue("tracing", "Enable pcap tracing", tracing);
    cmd.AddValue("init", "ns-3 should initialize MPI by calling MPI_Init", init);
    cmd.AddValue("verbose", "verbose output", verbose);
    cmd.AddValue("test", "Enable regression test output", testing);
    cmd.Parse(argc, argv);
 
    // Defer reporting the configuration until we know the communicator
 
    // Distributed simulation setup; by default use granted time window algorithm.
    if (nullmsg)
    {
        GlobalValue::Bind("SimulatorImplementationType",
                          StringValue("ns3::NullMessageSimulatorImpl"));
    }
    else
    {
        GlobalValue::Bind("SimulatorImplementationType",
                          StringValue("ns3::DistributedSimulatorImpl"));
    }
 
    // MPI_Init
 
    if (init)
    {
        // Initialize MPI directly
        MPI_Init(&argc, &argv);
    }
    else
    {
        // Let ns-3 call MPI_Init and MPI_Finalize
        MpiInterface::Enable(&argc, &argv);
    }
 
    SinkTracer::Init();
 
    auto worldSize = SinkTracer::GetWorldSize();
    auto worldRank = SinkTracer::GetWorldRank();
 
    if ((!init) && (worldSize != 2))
    {
        RANK0COUT("This simulation requires exactly 2 logical processors if --init is not set."
                  << std::endl);
        return 1;
    }
 
    if (worldSize < 2)
    {
        RANK0COUT("This simulation requires 2  or more logical processors." << std::endl);
        return 1;
    }
 
    // Set up the MPI communicator for ns-3
    //  Condition                         ns-3 Communicator
    //  a.  worldSize = 2    copy of MPI_COMM_WORLD
    //  b.  worldSize > 2    communicator of ranks 1-2
 
    // Flag to record that we created a communicator so we can free it at the end.
    bool freeComm = false;
    // The new communicator, if we create one
    MPI_Comm splitComm = MPI_COMM_WORLD;
    // The list of ranks assigned to ns-3
    std::string ns3Ranks;
    // Tag for whether this rank should go into a new communicator
    int color = MPI_UNDEFINED;
 
    if (worldSize == 2)
    {
        std::stringstream ss;
        color = NS_COLOR;
        ss << "MPI_COMM_WORLD (" << worldSize << " ranks)";
        ns3Ranks = ss.str();
        splitComm = MPI_COMM_WORLD;
        freeComm = false;
    }
    else
    {
        //  worldSize > 2    communicator of ranks 1-2
 
        // Put ranks 1-2 in the new communicator
        if (worldRank == 1 || worldRank == 2)
        {
            color = NS_COLOR;
        }
        else
        {
            color = NOT_NS_COLOR;
        }
        std::stringstream ss;
        ss << "Split [1-2] (out of " << worldSize << " ranks) from MPI_COMM_WORLD";
        ns3Ranks = ss.str();
 
        // Now create the new communicator
        MPI_Comm_split(MPI_COMM_WORLD, color, worldRank, &splitComm);
        freeComm = true;
    }
 
    if (init)
    {
        MpiInterface::Enable(splitComm);
    }
 
    // Report the configuration from rank 0 only
    RANK0COUT(cmd.GetName() << "\n");
    RANK0COUT("\n");
    RANK0COUT("Configuration:\n");
    RANK0COUT("Routing:           " << (nix ? "nix-vector" : "global") << "\n");
    RANK0COUT("Synchronization:   " << (nullmsg ? "null-message" : "granted time window (YAWNS)")
                                    << "\n");
    RANK0COUT("MPI_Init called:   "
              << (init ? "explicitly by this program" : "implicitly by ns3::MpiInterface::Enable()")
              << "\n");
    RANK0COUT("ns-3 Communicator: " << ns3Ranks << "\n");
    RANK0COUT("PCAP tracing:      " << (tracing ? "" : "not") << " enabled\n");
    RANK0COUT("\n");
    RANK0COUT("Rank assignments:" << std::endl);
 
    if (worldRank == 0)
    {
        ReportRank(color, splitComm);
    }
 
    if (verbose)
    {
        // Circulate a token to have each rank report in turn
        int token;
 
        if (worldRank == 0)
        {
            token = 1;
        }
        else
        {
            MPI_Recv(&token, 1, MPI_INT, worldRank - 1, 0, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
            ReportRank(color, splitComm);
        }
 
        MPI_Send(&token, 1, MPI_INT, (worldRank + 1) % worldSize, 0, MPI_COMM_WORLD);
 
        if (worldRank == 0)
        {
            MPI_Recv(&token, 1, MPI_INT, worldSize - 1, 0, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
        }
    } // circulate token to report rank
 
    RANK0COUT(std::endl);
 
    if (color != NS_COLOR)
    {
        // Do other work outside the ns-3 communicator
 
        // In real use of a separate communicator from ns-3
        // the other tasks would be running another simulator
        // or other desired work here..
 
        // Our work is done, just wait for everyone else to finish.
 
        MpiInterface::Disable();
 
        if (init)
        {
            MPI_Finalize();
        }
 
        return 0;
    }
 
    // The code below here is essentially the same as simple-distributed.cc
    // --------------------------------------------------------------------
 
    // We use a trace instead of relying on NS_LOG
 
    if (verbose)
    {
        LogComponentEnable("PacketSink", LOG_LEVEL_INFO);
    }
 
    uint32_t systemId = MpiInterface::GetSystemId();
    uint32_t systemCount = MpiInterface::GetSize();
 
    // Check for valid distributed parameters.
    // Both this script and simple-distributed.cc will work
    // with arbitrary numbers of ranks, as long as there are at least 2.
    if (systemCount < 2)
    {
        RANK0COUT("This simulation requires at least 2 logical processors." << std::endl);
        return 1;
    }
 
    // Some default values
    Config::SetDefault("ns3::OnOffApplication::PacketSize", UintegerValue(512));
    Config::SetDefault("ns3::OnOffApplication::DataRate", StringValue("1Mbps"));
    Config::SetDefault("ns3::OnOffApplication::MaxBytes", UintegerValue(512));
 
    // Create leaf nodes on left with system id 0
    NodeContainer leftLeafNodes;
    leftLeafNodes.Create(4, 0);
 
    // Create router nodes.  Left router
    // with system id 0, right router with
    // system id 1
    NodeContainer routerNodes;
    Ptr<Node> routerNode1 = CreateObject<Node>(0);
    Ptr<Node> routerNode2 = CreateObject<Node>(1);
    routerNodes.Add(routerNode1);
    routerNodes.Add(routerNode2);
 
    // Create leaf nodes on left with system id 1
    NodeContainer rightLeafNodes;
    rightLeafNodes.Create(4, 1);
 
    PointToPointHelper routerLink;
    routerLink.SetDeviceAttribute("DataRate", StringValue("5Mbps"));
    routerLink.SetChannelAttribute("Delay", StringValue("5ms"));
 
    PointToPointHelper leafLink;
    leafLink.SetDeviceAttribute("DataRate", StringValue("1Mbps"));
    leafLink.SetChannelAttribute("Delay", StringValue("2ms"));
 
    // Add link connecting routers
    NetDeviceContainer routerDevices;
    routerDevices = routerLink.Install(routerNodes);
 
    // Add links for left side leaf nodes to left router
    NetDeviceContainer leftRouterDevices;
    NetDeviceContainer leftLeafDevices;
    for (uint32_t i = 0; i < 4; ++i)
    {
        NetDeviceContainer temp = leafLink.Install(leftLeafNodes.Get(i), routerNodes.Get(0));
        leftLeafDevices.Add(temp.Get(0));
        leftRouterDevices.Add(temp.Get(1));
    }
 
    // Add links for right side leaf nodes to right router
    NetDeviceContainer rightRouterDevices;
    NetDeviceContainer rightLeafDevices;
    for (uint32_t i = 0; i < 4; ++i)
    {
        NetDeviceContainer temp = leafLink.Install(rightLeafNodes.Get(i), routerNodes.Get(1));
        rightLeafDevices.Add(temp.Get(0));
        rightRouterDevices.Add(temp.Get(1));
    }
 
    InternetStackHelper stack;
    Ipv4NixVectorHelper nixRouting;
    Ipv4StaticRoutingHelper staticRouting;
 
    Ipv4ListRoutingHelper list;
    list.Add(staticRouting, 0);
    list.Add(nixRouting, 10);
 
    if (nix)
    {
        stack.SetRoutingHelper(list); // has effect on the next Install ()
    }
 
    stack.InstallAll();
 
    Ipv4InterfaceContainer routerInterfaces;
    Ipv4InterfaceContainer leftLeafInterfaces;
    Ipv4InterfaceContainer leftRouterInterfaces;
    Ipv4InterfaceContainer rightLeafInterfaces;
    Ipv4InterfaceContainer rightRouterInterfaces;
 
    Ipv4AddressHelper leftAddress;
    leftAddress.SetBase("10.1.1.0", "255.255.255.0");
 
    Ipv4AddressHelper routerAddress;
    routerAddress.SetBase("10.2.1.0", "255.255.255.0");
 
    Ipv4AddressHelper rightAddress;
    rightAddress.SetBase("10.3.1.0", "255.255.255.0");
 
    // Router-to-Router interfaces
    routerInterfaces = routerAddress.Assign(routerDevices);
 
    // Left interfaces
    for (uint32_t i = 0; i < 4; ++i)
    {
        NetDeviceContainer ndc;
        ndc.Add(leftLeafDevices.Get(i));
        ndc.Add(leftRouterDevices.Get(i));
        Ipv4InterfaceContainer ifc = leftAddress.Assign(ndc);
        leftLeafInterfaces.Add(ifc.Get(0));
        leftRouterInterfaces.Add(ifc.Get(1));
        leftAddress.NewNetwork();
    }
 
    // Right interfaces
    for (uint32_t i = 0; i < 4; ++i)
    {
        NetDeviceContainer ndc;
        ndc.Add(rightLeafDevices.Get(i));
        ndc.Add(rightRouterDevices.Get(i));
        Ipv4InterfaceContainer ifc = rightAddress.Assign(ndc);
        rightLeafInterfaces.Add(ifc.Get(0));
        rightRouterInterfaces.Add(ifc.Get(1));
        rightAddress.NewNetwork();
    }
 
    if (!nix)
    {
        Ipv4GlobalRoutingHelper::PopulateRoutingTables();
    }
 
    if (tracing)
    {
        if (systemId == 0)
        {
            routerLink.EnablePcap("router-left", routerDevices, true);
            leafLink.EnablePcap("leaf-left", leftLeafDevices, true);
        }
 
        if (systemId == 1)
        {
            routerLink.EnablePcap("router-right", routerDevices, true);
            leafLink.EnablePcap("leaf-right", rightLeafDevices, true);
        }
    }
 
    // Create a packet sink on the right leafs to receive packets from left leafs
    uint16_t port = 50000;
    if (systemId == 1)
    {
        Address sinkLocalAddress(InetSocketAddress(Ipv4Address::GetAny(), port));
        PacketSinkHelper sinkHelper("ns3::UdpSocketFactory", sinkLocalAddress);
        ApplicationContainer sinkApp;
        for (uint32_t i = 0; i < 4; ++i)
        {
            auto apps = sinkHelper.Install(rightLeafNodes.Get(i));
            auto sink = DynamicCast<PacketSink>(apps.Get(0));
            NS_ASSERT_MSG(sink, "Couldn't get PacketSink application.");
            if (testing)
            {
                sink->TraceConnectWithoutContext("RxWithAddresses",
                                                 MakeCallback(&SinkTracer::SinkTrace));
            }
            sinkApp.Add(apps);
        }
        sinkApp.Start(Seconds(1.0));
        sinkApp.Stop(Seconds(5));
    }
 
    // Create the OnOff applications to send
    if (systemId == 0)
    {
        OnOffHelper clientHelper("ns3::UdpSocketFactory", Address());
        clientHelper.SetAttribute("OnTime", StringValue("ns3::ConstantRandomVariable[Constant=1]"));
        clientHelper.SetAttribute("OffTime",
                                  StringValue("ns3::ConstantRandomVariable[Constant=0]"));
 
        ApplicationContainer clientApps;
        for (uint32_t i = 0; i < 4; ++i)
        {
            AddressValue remoteAddress(InetSocketAddress(rightLeafInterfaces.GetAddress(i), port));
            clientHelper.SetAttribute("Remote", remoteAddress);
            clientApps.Add(clientHelper.Install(leftLeafNodes.Get(i)));
        }
        clientApps.Start(Seconds(1.0));
        clientApps.Stop(Seconds(5));
    }
 
    RANK0COUT(std::endl);
 
    Simulator::Stop(Seconds(5));
    Simulator::Run();
    Simulator::Destroy();
 
    // --------------------------------------------------------------------
    // Conditional cleanup based on whether we built a communicator
    // and called MPI_Init
 
    if (freeComm)
    {
        MPI_Comm_free(&splitComm);
    }
 
    if (testing)
    {
        SinkTracer::Verify(4);
    }
 
    // Clean up the ns-3 MPI execution environment
    // This will call MPI_Finalize if MpiInterface::Initialize was called
    MpiInterface::Disable();
 
    if (init)
    {
        // We called MPI_Init, so we have to call MPI_Finalize
        MPI_Finalize();
    }
 
    return 0;
}