examples/python/network-nonlinear/network-nonlinear.py

   1 # Copyright (c) 2006-2023. The SimGrid Team. All rights reserved.
   2 #
   3 # This program is free software; you can redistribute it and/or modify it
   4 # under the terms of the license (GNU LGPL) which comes with this package.
   5
   6 """
   7 This example shows how to simulate a non-linear resource sharing for network links.
   8 """
   9
  10 import functools
  11 import sys
  12 from simgrid import Actor, Engine, Comm, Mailbox, NetZone, Link, LinkInRoute, this_actor
  13
  14 class Sender:
  15     """
  16     Send a series of messages to mailbox "receiver"
  17     """
  18     def __init__(self, msg_count: int, msg_size=int(1e6)):
  19         self.msg_count = msg_count
  20         self.msg_size = msg_size
  21
  22     # Actors that are created as object will execute their __call__ method.
  23     # So, the following constitutes the main function of the Sender actor.
  24     def __call__(self):
  25         pending_comms = []
  26         mbox = Mailbox.by_name("receiver")
  27
  28         for i in range(self.msg_count):
  29             msg = "Message " + str(i)
  30             size = self.msg_size * (i + 1)
  31             this_actor.info("Send '%s' to '%s, msg size: %d'" % (msg, mbox.name, size))
  32             comm = mbox.put_async(msg, size)
  33             pending_comms.append(comm)
  34
  35         this_actor.info("Done dispatching all messages")
  36
  37         # Now that all message exchanges were initiated, wait for their completion in one single call
  38         Comm.wait_all(pending_comms)
  39
  40         this_actor.info("Goodbye now!")
  41
  42 class Receiver:
  43     """
  44     Receiver actor: wait for N messages on the mailbox "receiver"
  45     """
  46
  47     def __init__(self, msg_count=10):
  48         self.msg_count = msg_count
  49
  50     def __call__(self):
  51         mbox = Mailbox.by_name("receiver")
  52
  53         pending_msgs = []
  54         pending_comms = []
  55
  56         this_actor.info("Wait for %d messages asynchronously" % self.msg_count)
  57         for _ in range(self.msg_count):
  58             comm, data = mbox.get_async()
  59             pending_comms.append(comm)
  60             pending_msgs.append(data)
  61
  62         while pending_comms:
  63             index = Comm.wait_any(pending_comms)
  64             msg = pending_msgs[index].get()
  65             this_actor.info("I got '%s'." % msg)
  66             del pending_comms[index]
  67             del pending_msgs[index]
  68
  69 ####################################################################################################
  70 def link_nonlinear(link: Link, capacity: float, n: int) -> float:
  71     """
  72     Non-linear resource sharing for links
  73
  74     Note that the callback is called twice in this example:
  75     1) link UP: with the number of active flows (from 9 to 1)
  76     2) link DOWN: with 0 active flows. A crosstraffic communication is happing
  77     in the down link, but it's not considered as an active flow.
  78     """
  79     # emulates a degradation in link according to the number of flows
  80     # you probably want something more complex than that and based on real
  81     # experiments
  82     capacity = min(capacity, capacity * (1.0 - (n - 1) / 10.0))
  83     this_actor.info("Link %s, %d active communications, new capacity %f" % (link.name, n, capacity))
  84     return capacity
  85
  86 def load_platform():
  87     """
  88     Create a simple 2-hosts platform
  89      ________                 __________
  90     | Sender |===============| Receiver |
  91     |________|    Link1      |__________|
  92
  93     """
  94     zone = NetZone.create_full_zone("Zone1")
  95     sender = zone.create_host("sender", 1).seal()
  96     receiver = zone.create_host("receiver", 1).seal()
  97
  98     link = zone.create_split_duplex_link("link1", 1e6)
  99     # setting same callbacks (could be different) for link UP/DOWN in split-duplex link
 100     link.link_up.set_sharing_policy(Link.SharingPolicy.NONLINEAR,
 101                                     functools.partial(link_nonlinear, link.link_up))
 102     link.link_down.set_sharing_policy(Link.SharingPolicy.NONLINEAR,
 103                                       functools.partial(link_nonlinear, link.link_down))
 104     link.set_latency(10e-6).seal()
 105
 106     # create routes between nodes
 107     zone.add_route(sender, receiver, [link])
 108     zone.seal()
 109
 110     # create actors Sender/Receiver
 111     Actor.create("receiver", receiver, Receiver(9))
 112     Actor.create("sender", sender, Sender(9))
 113
 114 ###################################################################################################
 115
 116 if __name__ == '__main__':
 117     e = Engine(sys.argv)
 118
 119     # create platform
 120     load_platform()
 121
 122     # runs the simulation
 123     e.run()
 124
 125     # explicitly deleting Engine object to avoid segfault during cleanup phase.
 126     # During Engine destruction, the cleanup of std::function linked to link_non_linear callback is called.
 127     # If we let the cleanup by itself, it fails trying on its destruction because the python main program
 128     # has already freed its variables
 129     del e