Using Python with ns-3

Though ns-3 is written in C++, it has an Python binding. The code sample below shows how to use Python to create an network experiment.

• It creates a network of three nodes. Two of them are connected via point to point link and two of them are connected via Ethernet.

      # Topology
      #         10.1.1.0
      #    n0 -------------- n1   n2
      #       point-to-point  |    |
      #                       ======
      #                     LAN 10.1.2.0

• It setup a Tcp socket client on node n2 which keep sending packets to a Tcp socket server on node n0.

• To simulate packet drops, a error model was used to generate 1 packet drop per 100 packets.
• The data transfer is captured via Pcap.

Tip
The ns.core module is created in a subtle way in ns-3 Python binding that most of the Python IDEs cannot get the intellisense from the module. That is because all the classes/functions in ns.core are imported dynamically via code (See [ns_root]/src/core/bindings/core.py).

The only way so far I found to work around it is remove the core.py file and rebuild. After that a core.so module will be generated instead of a _core.so. IDEs are happy with core.so.

Here is the code:

import sys
from ns3 import *


def create_nodes(n):
    return (Node() for _ in range(n))


def node_container(*nodes):
    c = NodeContainer()
    for n in nodes:
        c.Add(n)
    return c


def connect_nodes_with_p2p_channel(nodes, rate, delay, pcap_name):
    helper = PointToPointHelper()
    helper.SetDeviceAttribute("DataRate", StringValue(rate))
    helper.SetChannelAttribute("Delay", StringValue(delay))
    devices = helper.Install(nodes)
    if pcap_name is not None:
        helper.EnablePcapAll(pcap_name)
    return devices


def connect_nodes_with_csma_channel(nodes, rate, delay, pcap_name):
    helper = CsmaHelper()
    helper.SetChannelAttribute("DataRate", StringValue(rate))
    helper.SetChannelAttribute("Delay", StringValue(delay))
    devices = helper.Install(nodes)
    if pcap_name is not None:
        helper.EnablePcapAll(pcap_name)
    return devices


def install_internet_stack(nodes):
    stack = InternetStackHelper()
    stack.Install(nodes)


def set_receive_error(device, error_packet_seq_ids):
    of = ObjectFactory()
    of.SetTypeId('ns3::ReceiveListErrorModel')
    em = of.Create()
    em.SetList(error_packet_seq_ids)
    device.SetAttribute("ReceiveErrorModel", PointerValue(em))
    return em


def create_interfaces(devices, ip_base, mask):
    address = Ipv4AddressHelper()
    address.SetBase(Ipv4Address(ip_base), Ipv4Mask(mask))
    interfaces = address.Assign(devices)
    return interfaces


def get_protocol_type_id(protocol):
    if protocol == 'tcp':
        return 'ns3::TcpSocketFactory'
    elif protocol == 'udp':
        return 'ns3::UdpSocketFactory'
    else:
        return None


def install_packet_sink_app(node, protocol, port):
    protocol_type_id = get_protocol_type_id(protocol)
    helper = PacketSinkHelper(protocol_type_id, Address(InetSocketAddress(Ipv4Address.GetAny(), port)))
    app = helper.Install(node)
    return app


def install_on_off_app(node, protocol, server_address, server_port, rate=None, packet_size=None, on_time=None, off_time=None):
    protocol_type_id = get_protocol_type_id(protocol)
    helper = OnOffHelper(protocol_type_id, Address(InetSocketAddress(server_address, server_port)))
    if rate is not None:
        helper.SetAttribute("DataRate", DataRateValue(DataRate(rate)))
    if packet_size is not None:
        helper.SetAttribute("PacketSize", UintegerValue(packet_size))
    if on_time is not None:
        helper.SetAttribute("OnTime", get_constant_random_variable(on_time))
    if off_time is not None:
        helper.SetAttribute("OffTime", get_constant_random_variable(off_time))

    app = helper.Install(node)
    return app


def get_constant_random_variable(constant):
    # return StringValue('ns3::ConstantRandomVariable[Constant=%d]' % constant)
    v = ConstantRandomVariable()
    v.SetAttribute('Constant', StringValue(str(constant)))
    return PointerValue(v)


def main():
    cmd = CommandLine()
    cmd.Parse(sys.argv)

    # Topology
    #         10.1.1.0
    #    n0 -------------- n1   n2
    #       point-to-point  |    |
    #                       ======
    #                     LAN 10.1.2.0

    n0, n1, n2 = create_nodes(3)

    p2p_nodes = node_container(n0, n1)
    p2p_devices = connect_nodes_with_p2p_channel(p2p_nodes, '2Mbps', '2ms', 'p2p')

    csma_nodes = node_container(n1, n2)
    csma_devices = connect_nodes_with_csma_channel(csma_nodes, '10Mbps', '1ms', 'csma')

    install_internet_stack(n0)
    install_internet_stack(n1)
    install_internet_stack(n2)

    set_receive_error(p2p_devices.Get(0), [100 * i for i in range(10)])

    p2p_interfaces = create_interfaces(p2p_devices, '10.1.1.0', '255.255.255.0')
    csma_interfaces = create_interfaces(csma_devices, '10.1.2.0', '255.255.255.0')

    server_address = p2p_interfaces.GetAddress(0)
    server_port = 8888
    server_app = install_packet_sink_app(n0, 'tcp', server_port)
    server_app.Start(Seconds(1.0))
    server_app.Stop(Seconds(10.0))

    client_address = csma_interfaces.GetAddress(1)
    on_off_app = install_on_off_app(n3, 'tcp', server_address, server_port, rate='2Mbps', on_time=1, off_time=0)
    on_off_app.Start(Seconds(2.0))
    on_off_app.Stop(Seconds(10.0))

    Ipv4GlobalRoutingHelper.PopulateRoutingTables()

    Simulator.Stop(Seconds(20.0))
    Simulator.Run()
    Simulator.Destroy()


def log_cwnd(oldval, newval):
    print('cwnd old: %d, new: %d' % (oldval, newval))


def start_flow(socket, address, port):
    def write_until_buffer_full(socket, available):
        print(available)
        n = 10000
        while socket.GetTxAvailable() > 0 and n > 0:
            to_write = min(4 * 1024 * 1024, socket.GetTxAvailable())
            sent = socket.Send(Packet(to_write), 0)
            if sent < 0:
                return
            n -= 1

    socket.Connect(InetSocketAddress(address, port))
    socket.SetSendCallback(write_until_buffer_full)
    write_until_buffer_full(socket, socket.GetTxAvailable())


if __name__ == '__main__':
    main()