diff options
| -rw-r--r-- | rumba/model.py | 6 | ||||
| -rw-r--r-- | rumba/prototypes/irati.py | 3 | ||||
| -rw-r--r-- | rumba/prototypes/ouroboros.py | 79 | ||||
| -rw-r--r-- | rumba/prototypes/rlite.py | 3 | ||||
| -rw-r--r-- | rumba/testbeds/faketestbed.py | 5 |
5 files changed, 75 insertions, 21 deletions
diff --git a/rumba/model.py b/rumba/model.py index 2749ae7..42112be 100644 --- a/rumba/model.py +++ b/rumba/model.py @@ -874,6 +874,10 @@ class Experiment(object): def prototype_name(self): raise Exception('prototype_name() method not implemented') + @abc.abstractmethod + def _terminate_prototype(self): + raise Exception('terminate_prototype() method not implemented') + def swap_in(self): # Realize the experiment testbed (testbed-specific) start = time.time() @@ -884,6 +888,8 @@ class Experiment(object): def swap_out(self): start = time.time() + # Terminate prototype gracefully + self._terminate_prototype() for node in self.nodes: if node.ssh_config.client is not None: node.ssh_config.client.close() diff --git a/rumba/prototypes/irati.py b/rumba/prototypes/irati.py index bf3ce8c..0807e10 100644 --- a/rumba/prototypes/irati.py +++ b/rumba/prototypes/irati.py @@ -426,3 +426,6 @@ class Experiment(mod.Experiment): conf_files.setdefault(node, []).append( 'normal.%s.%s.dif' % (node.name, dif.name)) return conf_files + + def _terminate_prototype(self): + return diff --git a/rumba/prototypes/ouroboros.py b/rumba/prototypes/ouroboros.py index 6dec9e4..c588405 100644 --- a/rumba/prototypes/ouroboros.py +++ b/rumba/prototypes/ouroboros.py @@ -25,11 +25,13 @@ # import time +import subprocess import rumba.ssh_support as ssh import rumba.model as mod import rumba.multiprocess as m_processing import rumba.log as log +import rumba.testbeds.faketestbed as fake logger = log.get_logger(__name__) @@ -53,13 +55,33 @@ class Experiment(mod.Experiment): def prototype_name(self): return 'ouroboros' + def exec_local_cmd(self, cmd): + try: + logger.info(cmd) + subprocess.check_call(cmd.split(' ')) + except subprocess.CalledProcessError as e: + logger.error("Return code was " + str(e.returncode)) + raise + + def exec_local_cmds(self, cmds): + for cmd in cmds: + self.exec_local_cmd(cmd) + def setup_ouroboros(self): - for node in self.nodes: - ssh.execute_command(self.testbed, node.ssh_config, - "sudo nohup irmd > /dev/null &", - time_out=None) + if isinstance(self.testbed, fake.Testbed): + self.irmd = subprocess.Popen(["irmd"]) + logger.info("Started IRMd, sleeping 2 seconds...") + time.sleep(2) + else: + for node in self.nodes: + ssh.execute_command(self.testbed, node.ssh_config, + "sudo nohup irmd > /dev/null &", + time_out=None) def install_ouroboros(self): + if isinstance(self.testbed, fake.Testbed): + return + packages = ["cmake", "protobuf-c-compiler", "git", "libfuse-dev", "libgcrypt20-dev", "libssl-dev"] @@ -97,11 +119,11 @@ class Experiment(mod.Experiment): cmd = "irm i c n " + ipcp.name if isinstance(ipcp.dif, mod.ShimEthDIF): - # NOTE: Here to test with fake testbed - if ipcp.ifname is None: - ipcp.ifname = "eth0" - cmd += " type eth-llc if_name " + ipcp.ifname - cmd += " layer " + ipcp.dif.name + if isinstance(self.testbed, fake.Testbed): + cmd += " type local layer " + ipcp.dif.name + else: + cmd += " type eth-llc if_name " + ipcp.ifname + cmd += " layer " + ipcp.dif.name elif isinstance(ipcp.dif, mod.NormalDIF): cmd += " type normal" if ipcp.dif_bootstrapper: @@ -131,8 +153,11 @@ class Experiment(mod.Experiment): # Postpone registrations self.r_ipcps[ipcp] = cmds2 - ssh.execute_commands(self.testbed, node.ssh_config, cmds, - time_out=None) + if isinstance(self.testbed, fake.Testbed): + self.exec_local_cmds(cmds) + else: + ssh.execute_commands(self.testbed, node.ssh_config, cmds, + time_out=None) def enroll_dif(self, el): for e in el: @@ -141,9 +166,13 @@ class Experiment(mod.Experiment): # Execute postponed registration if e['enroller'] in self.r_ipcps: - ssh.execute_commands(self.testbed, - e['enroller'].node.ssh_config, - self.r_ipcps[e['enroller']], time_out=None) + if isinstance(self.testbed, fake.Testbed): + self.exec_local_cmds(self.r_ipcps[e['enroller']]) + else: + ssh.execute_commands(self.testbed, + e['enroller'].node.ssh_config, + self.r_ipcps[e['enroller']], + time_out=None) self.r_ipcps.pop(e['enroller'], None) cmd = "irm r n " + ipcp.name @@ -158,9 +187,12 @@ class Experiment(mod.Experiment): cmd += " layer " + dif_b.name cmds.append(cmd) - ssh.execute_commands(self.testbed, - e['enrollee'].node.ssh_config, - cmds, time_out=None) + if isinstance(self.testbed, fake.Testbed): + self.exec_local_cmds(cmds) + else: + ssh.execute_commands(self.testbed, + e['enrollee'].node.ssh_config, + cmds, time_out=None) def setup_flows(self, el, comp): for e in el: @@ -168,9 +200,12 @@ class Experiment(mod.Experiment): cmd = "irm i conn n " + ipcp.name + " comp " + \ comp + " dst " + e['dst'].name - ssh.execute_command(self.testbed, - ipcp.node.ssh_config, - cmd, time_out=None) + if isinstance(self.testbed, fake.Testbed): + self.exec_local_cmd(cmd) + else: + ssh.execute_command(self.testbed, + ipcp.node.ssh_config, + cmd, time_out=None) def _install_prototype(self): logger.info("Installing Ouroboros...") @@ -192,3 +227,7 @@ class Experiment(mod.Experiment): self.setup_flows(dt, comp="dt") logger.info("All done, have fun!") + + def _terminate_prototype(self): + if isinstance(self.testbed, fake.Testbed): + self.irmd.terminate() diff --git a/rumba/prototypes/rlite.py b/rumba/prototypes/rlite.py index dca693b..e1b3745 100644 --- a/rumba/prototypes/rlite.py +++ b/rumba/prototypes/rlite.py @@ -176,3 +176,6 @@ class Experiment(mod.Experiment): logger.info("IPCPs registered to their lower DIFs on all nodes") self.enroll_ipcps() logger.info("enrollment completed in all DIFs") + + def _terminate_prototype(self): + return diff --git a/rumba/testbeds/faketestbed.py b/rumba/testbeds/faketestbed.py index 44994e3..b021aca 100644 --- a/rumba/testbeds/faketestbed.py +++ b/rumba/testbeds/faketestbed.py @@ -38,4 +38,7 @@ class Testbed(mod.Testbed): mod.Testbed.__init__(self, exp_name, username, password, proj_name) def swap_in(self, experiment): - logger.info("[Fake testbed] experiment swapped in") + logger.info("Experiment swapped in") + + def swap_out(self, experiment): + logger.info("Experiment swapped out") |