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.0It 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()