tl-system / ns.py Goto Github PK

Describe the bug

1. According to the formula, we should use the active set of the last update time to calculate the incremental rounds.

def put(self, packet, upstream_update=None, upstream_store=None)

the code updates the active set before calculating the increments. As a result, we are using the current active set to calculate the increments.

self.active_set.add(self.flow_classes(flow_id)) 
weight_sum = 0.0
for i in self.active_set:
       weight_sum += self.weights[i]
self.vtime += (now - self.last_update) / weight_sum

2. The active set might change when departure occurs, so we should use the same method to update the virtual time upon departure as upon arrival. However, in

def update_stats(self, packet)

the code only replaces the last update time but does not update the virtual time. Consequently, we will miss a large part of the calculation of virtual time.

Describe the bug
In the line 113 of ns.utils.generators.MAP_MSP_generator.BMAP_generator, "nextpr = np.hstack((nextpr, np.diag(sojourn) * Dk))" should be "nextpr = np.hstack((nextpr, np.diag(sojourn) @ Dk))".

Describe the bug
Hello, I analyze the sp.py, and find a mismatch for different formats of priority, which may result in some bugs.

The input priorities is a list or a dict, with the following comment:

This can be either a list or a dictionary. If it is a list, it uses the flow_id
as its index to look for the flow's corresponding priority. If it is a dictionary,
it contains (flow_id -> priority) pairs for each possible flow_id.

But self.prio_queue_count has a different map:

if isinstance(priorities, list):
    self.prio_queue_count = [0 for __ in range(len(priorities))] # flow_id -> count
elif isinstance(priorities, dict):
    self.prio_queue_count = {prio: 0 for prio in priorities.values()} # priority -> count
else:
    raise ValueError(
        'Priorities must be either a list or a dictionary.')

Specifically, if priorities is a list, self.prio_queue_count will map flow_id to count. Otherwise, self.prio_queue_count will map the priority of the flow_id to count.

To Reproduce
You can run the following code and see a bug:

import numpy as np
import simpy
from ns.packet.dist_generator import DistPacketGenerator
from ns.packet.sink import PacketSink
from ns.scheduler.sp import SPServer

np.random.seed(42)

def arrival_1():
    """ Packets arrive with a constant interval of 1.5 seconds. """
    return 1.5


def arrival_2():
    """ Packets arrive with a constant interval of 2.0 seconds. """
    return 2.0

def packet_size():
    return int(np.random.exponential(100))

DEBUG = True

env = simpy.Environment()
sp = SPServer(env, 100, [1,100], debug=DEBUG) # line[1]
# sp = SPServer(env, 100, {0:10, 1:1}, debug=DEBUG) # line[2]
# sp = SPServer(env, 100, {0:1, 1:10}, debug=DEBUG) # line[3]
ps = PacketSink(env, rec_flow_ids=False, debug=DEBUG)

pg1 = DistPacketGenerator(env, "flow_1", arrival_1, packet_size, flow_id=0)
pg2 = DistPacketGenerator(env, "flow_2", arrival_2, packet_size, flow_id=1)

pg1.out = sp
pg2.out = sp
sp.out = ps

env.run(until=20)

The error:

Time 1.5, flow_id 0, packet_id 1
Sent out packet 1 of priority 1
Traceback (most recent call last):
  File "D:\anaconda3\lib\site-packages\ns\packet\dist_generator.py", line 80, in run
    self.out.put(packet)
  File "D:\anaconda3\lib\site-packages\ns\scheduler\sp.py", line 179, in put
    self.prio_queue_count[prio] += 1
IndexError: list index out of range

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "e:\Study\2021-summer\ns.py\examples\sp.py", line 36, in <module>
    env.run(until=20)
  File "D:\anaconda3\lib\site-packages\simpy\core.py", line 254, in run
    self.step()
  File "D:\anaconda3\lib\site-packages\simpy\core.py", line 206, in step
    raise exc
IndexError: list index out of range

Confusion
Moreover, I also want to know whether a larger number will result in a higher priority or a lower priority. Because, for the above test code, line[2] and line[3] act as the same: flow_0 has a high priority in both codes, regardless of the priority value.

Thank you for this great tool. It's very simple.
Can I use your simulator to do very simple routing i.e. static routing?

Context

I am trying to implement a MLFQ [1] scheduler using the SPServer component. In MLFQ scheduler, flow priority changes during flow lifetime depending on the number of transmitted bytes. In this case (1) the mapping flow -> class does not depend solely on the flow_id but also (2) on a runtime state that can be inferred looking at other packet fields e.g., sequence number. Notice that the defintion of flow is arbitrary and could depend on dynamic parameters (num. transmitted packets, time-to-live, etc.) other than a static flow_id.

Problem
In this case, different calls to the function for same flow_id will lead to a different class_id mapping, depending on the state at invocation time. This can cause inconsistencies when update_stats is called on old buffered packets that are served when the state has already changed. For example, a packet enqueued when a flow was mapped to class 0, but served when the same flow is mapped to class 1 (because its state, number of transmitted packet, is changed), will try to update_stats related to class 1 instead of class 0.

Describe the solution you'd like
A better implementation of flow_classes in my opinion would take the packet instead of the flow_id as its input parameter. Since flow_id can be found as a packet field, this solution would not sacrify current implementation but would allow to transparently keep the same class mapping one would have when packet is generated, even if the invocation to flow_classes happens later.

Additional context
[1] Bai, W., Chen, L., Chen, K., Han, D., Tian, C. and Wang, H., 2015. Information-Agnostic Flow Scheduling for Commodity Data Centers. In 12th USENIX Symposium on Networked Systems Design and Implementation (NSDI 15).

Problem: Implementation of supporting class-based fair scheduling.

Class-based fair scheduling is very common in real networks. I think it will be awesome if ns.py can support this. In ns.py==0.2.4, it can somehow provide the classed-based serving, but I am afraid it is different from the way that the real mechanism works.

Take Class Based Weighted Fair Queuing (CBWFQ) as an example, it provides congestion-management support for user-defined traffic classes. The mapping between flows to traffic classes are user-defined. A FIFO queue is reserved for each class, and a flow belonging to a class is directed to the queue for that class. The weight of each class is also user-defined.

The mechanism of CB-DRR or CB-SP is similar to the above.

Describe the solution you'd like
To implement a dictionary "flows_to_classes" that acts as a mapping between flow IDs and class IDs at a CBWFQ.
For example, in a fattree network, the dictionary will match a tri-tuple (flow ID, node ID, port ID) to a class ID at a WFQ.
The number of the classes, and the weights of each class are user-defined, independent of the actual flows passing through the port.

Additional context
This is a bug in how ns.py==0.2.4 works in terms of CBWFQ.
For example, in examples/fattree.py, a weight is assigned to each flow (Line 45). I think we should assign the weights to classes instead of queues, and use a flows_to_classes mapping.

I have a couple of concerns regarding the implementation of TCP.

1. Cumulative ACKs. Since TCP ACKs are cumulative, I expect an incoming ACK to stop all timeouts of the currently in-flight packets it acknowledges. However, it seems that your implementation only deletes the timer associated to the packet that triggered the ACK at the receiver. At least when tested with TCPReno congestion control, I experience lot of retransmissions in the simulation.
2. Slow start. When a timeout expires in TCPReno, the new ssthresh should be set to cwnd/2. However, I don't find this behavior implemented in the code. Moreover, the slow start phase should end when self.cwnd + self.mss is strictly smaller than self.ssthesh and not when smaller or equal.
3. Fast Recovery. In the fast recovery phase, after the third duplicate ACK is received, TCPReno would inflate the window of 1 MSS for each received ACK and transmit new segments if the congestion window allows. In the implementation, the window is correctly inflated, but in the if self.dupack > 3 clause, the tcp_generator retranmits old segments.

If I plot the evolution of the congestion window of a TCP stream going through a single transmission queue, I get a graph very different from the typical TCPReno window evolution. Am I interpreting the code in the wrong way?

Thank you for the clarification and for the very good code base.

Dear Developers,
I just ran the code of "fattree.py" and tried to print the "perhop_times" attribute of each packet sink in order to check the detailed information for each packet. I found that the "perhop_times" attribute of packet sink is always empty after running the simulation. Could I know whether this is as expected after running the simulation?

Brief Description
The the dict items of flow.pkt_sink.perhop_times are all empty after running the simulation.

To Reproduce
Steps to reproduce the behavior:

1. Add the "print(all_flows[flow_id].pkt_sink.perhop_times)" following the line "print(all_flows[flow_id].pkt_sink.arrivals)" at the end of the file "fattree.py".
2. Run the "fattree.py"
3. Check the output. Found all the items of perhop_times are empty.

Thank you for reading my request!

The time cost of the program increases rapidly as the number of packet increases, which makes it really hard to do experiments of large scale traffics. Is there a way to speed up the program?

Is your feature request related to a problem? Please describe.
The Transmission Control Protocol (TCP) is one of the main protocols of the Internet protocol suite, and its implementation is missing from ns.py.

Describe the solution you'd like
Implement a TCPPacketGenerator based on RFC793. Flow size, duration, and congestion control mechanisms should be configurable.

Describe alternatives you've considered
Recording TCP packet traces and replaying them using trace generator. However, this is unrealistic and fails to capture inter-flow dynamics.

I'd like to cite this repository for my bachelor's thesis. It would be great if you could add a CITATION.cff file to the repo to ease citation by researchers.

Describe the bug
In function update_stats of sp.py, the debug message should use packet.flow_id instead of packet.packet_id to map a flow to its class.

if self.debug:
    print(
        f"Sent out packet {packet.packet_id} from flow {packet.flow_id} "
        f"belonging to class {self.flow_classes(packet.packet_id)} "
        f"of priority {packet.prio[self.element_id]}")

To Reproduce
Run the following code, where the flow class is assigned at runtime depending on the number of transmitted packets, by looking up the flows dictionary:

"""
A basic example that connects two packet generators to a network wire with
a propagation delay distribution, and then to a packet sink.
"""

from functools import partial
import simpy
from bisect import bisect_left

from ns.packet.dist_generator import DistPacketGenerator
from ns.packet.sink import PacketSink
from ns.scheduler.sp import SPServer
from ns.switch.switch import SimplePacketSwitch


# priortiy queues per server
k = 2
# number of servers
n = 2
# rate of each server
mu = 100
# demotion thresholds
demotion_thresholds = [5]

def arrival_1():
    """ Packets arrive with a constant interval of 1.5 seconds. """
    return 1.5


def arrival_2():
    """ Packets arrive with a constant interval of 2.0 seconds. """
    return 2.0


def packet_size():
    return 100


def flow_to_classes(f_id, flows, thresholds):
    attained_service = flows[f_id].packets_sent
    return bisect_left(thresholds, attained_service)
    #return (f_id + n_id + fib[f_id]) % n_classes_per_port

env = simpy.Environment()

# simple switch with infinite rate, just used as entrypoint node
# for flows to handle routing to different servers
switch = SimplePacketSwitch(
    env,
    nports=n,
    port_rate=0,  # in bits/second
    buffer_size=None,  # in packets
    debug=True)

pg1 = DistPacketGenerator(env, "flow_1", arrival_1, packet_size, flow_id=0)
pg2 = DistPacketGenerator(env, "flow_2", arrival_2, packet_size, flow_id=1)

# connect all packet generators to switch
pg1.out = switch
pg2.out = switch

# route flows
fib = {0: 0, 1: 1}
switch.demux.fib = fib

flows = {
    0: pg1,
    1: pg2
}
flow_classes = partial(flow_to_classes, flows=flows, thresholds=demotion_thresholds)
servers = []
for i in range(n):
    sp = SPServer(
        env,
        mu,
        list(range(k)),
        flow_classes=flow_classes,
        debug=True
        )
    
    ps = PacketSink(env, rec_flow_ids=False, debug=True)

    sp.out = ps
    switch.ports[i].out = sp

    servers.append(sp)

env.run(until=20)

Expected behavior
You get following exception:

Traceback (most recent call last):
  File "C:\Users\d067567\AppData\Local\anaconda3\envs\ns.py\lib\site-packages\ns\scheduler\sp.py", line 190, in run
    self.update_stats(packet)
  File "C:\Users\d067567\AppData\Local\anaconda3\envs\ns.py\lib\site-packages\ns\scheduler\sp.py", line 107, in update_stats
    f"belonging to class {self.flow_classes(packet.packet_id)} "
  File "g:\My Drive\Politecnico\Thesis\python-ns\examples\spatial_diversity.py", line 40, in flow_to_classes
    attained_service = flows[f_id].packets_sent
KeyError: 6

Is your feature request related to a problem? Please describe.
The current design of flow IDs is based on the assumption that unique int-type numbers can be assigned to the flows throughout the entire network, across all the sources. This may not be ideal in many situations.

Describe the solution you'd like
Rather than using int as its type, redesign flow IDs and all its dependent data structures so that it can also be a string. All lists in the current implementation that directly used flow IDs as indexes should use a dict instead.

Describe alternatives you've considered
One can imagine that the source ID, which is currently a string, can be used in conjunction with the flow ID to meet the original usage requirements of flow IDs. But this may be cumbersome as it requires source IDs to be set, in addition to setting flow IDs. In many cases, it may be easier to just set the flow IDs to strings that are unique system-wide.

Additional context

I believe the lines 37&38 should be inside the while loop. After exiting the while loop, the function "run" will call the timeout_callback, but will always end after that.

def run(self):
    """ The generator function used in simulations. """
    while self.env.now < self.timer_expiry:
        yield self.env.timeout(self.timer_expiry - self.env.now)

    if not self.stopped:
        self.timeout_callback(self.timer_id)

Specifically, when timeout_callback includes a call to "restart", this will have no effect. I assume the intended use of restart is to continue the timer with a new timeout, but that will not happen unless lines 37-38 are inside the while loop.

def restart(self, timeout):
    """ Restarting the timer with a new timeout value. """
    self.timer_started = self.env.now
    self.timer_expiry = self.timer_started + timeout