diff options
| -rw-r--r-- | rumba/testbeds/qemu.py | 266 |
1 files changed, 189 insertions, 77 deletions
diff --git a/rumba/testbeds/qemu.py b/rumba/testbeds/qemu.py index 2f5491a..45e5cc0 100644 --- a/rumba/testbeds/qemu.py +++ b/rumba/testbeds/qemu.py @@ -17,69 +17,135 @@ # License along with this library; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, # MA 02110-1301 USA -import multiprocessing +from Queue import Empty +from multiprocessing import Process, Queue, cpu_count +from os import geteuid import rumba.model as mod -from subprocess import Popen +from subprocess import Popen, check_call, CalledProcessError class Testbed(mod.Testbed): - def __init__(self, exp_name, username, vm_img_folder, proj_name="ARCFIRE", password="", use_vhost=True): + def __init__(self, exp_name, username, vm_img_folder, proj_name="ARCFIRE", password="", + use_vhost=True, qemu_out_folder=""): mod.Testbed.__init__(self, exp_name, username, password, proj_name) self.vms = {} self.shims = [] self.vm_img_path = vm_img_folder self.vhost = use_vhost + self.qemu_folder = qemu_out_folder + self.boot_processes = [] + + @staticmethod + def _run_command_chain(commands, results_queue, error_queue): + """ + Runs (sequentially) the command list. + + On error, breaks and dumps it in error_queue, and interrupts as soon as it is non-empty. + + :type commands: list + :type results_queue: Queue + :type error_queue: Queue + :param commands: list of commands to execute + :param results_queue: Queue of results of parallel processes + :param error_queue: Queue of error(s) encountered + :return: None + """ + try: + for command in commands: + if not error_queue.empty(): + break + print('DEBUG: executing >> {}'.format(command)) + check_call(command.split()) + + results_queue.put("Command chain ran correctly.") + except CalledProcessError as e: + error_queue.put(str(e)) def create_experiment(self, experiment): """ :type experiment mod.Experiment - :rtype str :param experiment: The experiment running - :return the script to be run to create the experiment's vms """ + if geteuid() != 0: + print("ERROR: QEMU testbed requires root access: please use sudo.") + return # TODO I'd rather raise an appropriate error... + print("[QEMU testbed] swapping in") - command_str = "" # Building bridges and taps - + shim_processes = [] + r_queue = Queue() + e_queue = Queue() + print(experiment.dif_ordering) for shim in experiment.dif_ordering: + command_list = [] if not isinstance(shim, mod.ShimEthDIF): # Nothing to do here continue self.shims.append(shim) - command_str += 'sudo brctl addbr %(br)s\n' \ - 'sudo ip link set %(br)s up\n' \ - '\n' % {'br': shim.name} + command_list += ('sudo brctl addbr %(br)s\n' + 'sudo ip link set %(br)s up' + % {'br': shim.name} + ).split('\n') for node in shim.members: # type:mod.Node name = node.name vm = self.vms.setdefault(name, {'vm': node, 'ports': []}) port_id = len(vm['ports']) + 1 tap_id = '%s.%02x' % (name, port_id) - command_str += 'sudo ip tuntap add mode tap name %(tap)s\n' \ - 'sudo ip link set %(tap)s up\n' \ - 'sudo brctl addif %(br)s %(tap)s\n\n' \ - % {'tap': tap_id, 'br': shim.name} + command_list += ('sudo ip tuntap add mode tap name %(tap)s\n' + 'sudo ip link set %(tap)s up\n' + 'sudo brctl addif %(br)s %(tap)s' + % {'tap': tap_id, 'br': shim.name} + ).split('\n') if shim.link_speed > 0: speed = '%dmbit' % shim.link_speed # Rate limit the traffic transmitted on the TAP interface - command_str += 'sudo tc qdisc add dev %(tap)s handle 1: root ' \ - 'htb default 11\n' \ - 'sudo tc class add dev %(tap)s parent 1: classid ' \ - '1:1 htb rate 10gbit\n' \ - 'sudo tc class add dev %(tap)s parent 1:1 classid ' \ - '1:11 htb rate %(speed)s\n' \ - % {'tap': tap_id, 'speed': speed} + command_list += ('sudo tc qdisc add dev %(tap)s handle 1: root ' + 'htb default 11\n' + 'sudo tc class add dev %(tap)s parent 1: classid ' + '1:1 htb rate 10gbit\n' + 'sudo tc class add dev %(tap)s parent 1:1 classid ' + '1:11 htb rate %(speed)s' + % {'tap': tap_id, 'speed': speed} + ).split('\n') vm['ports'].append({'tap_id': tap_id, 'shim': shim, 'port_id': port_id}) # TODO deal with Ip address (shim UDP DIF). + # Avoid stacking processes if one failed before. + if not e_queue.empty(): + break + # Launch commands asynchronously + process = Process(target=self._run_command_chain, args=(command_list, r_queue, e_queue)) + shim_processes.append(process) + process.start() + + # Wait for all processes to be over. + total_processes = len(shim_processes) + max_waiting_time = 2 * total_processes + over_processes = 0 + + while max_waiting_time > 0 and over_processes < total_processes: + # Check for errors + if not e_queue.empty(): + print('Testbed instantiation failed: {}'.format(str(e_queue.get()))) + return # TODO: again, I would prefer a specific exception + try: + # Check for results + result = r_queue.get(timeout=1) + if result == "Command chain ran correctly.": + over_processes += 1 + print('DEBUG: %s of %s processes completed.' % (over_processes, total_processes)) + except Empty: + max_waiting_time -= 1 + # Building vms - boot_batch_size = max(1, multiprocessing.cpu_count() // 2) + boot_batch_size = max(1, cpu_count() // 2) booting_budget = boot_batch_size boot_backoff = 12 base_port = 2222 @@ -105,22 +171,24 @@ class Testbed(mod.Testbed): host_fwd_str = 'hostfwd=tcp::%(fwdp)s-:22' % vars_dict vars_dict['hostfwdstr'] = host_fwd_str - command_str += 'qemu-system-x86_64 ' + command = 'qemu-system-x86_64 ' # TODO manage non default images - command_str += '-kernel %(vmimgpath)s/bzImage ' \ - '-append "console=ttyS0" ' \ - '-initrd %(vmimgpath)s/rootfs.cpio ' % vars_dict - command_str += '-nographic ' \ - '-display none ' \ - '--enable-kvm ' \ - '-smp 1 ' \ - '-m %(memory)sM ' \ - '-device %(frontend)s,mac=%(mac)s,netdev=mgmt ' \ - '-netdev user,id=mgmt,%(hostfwdstr)s ' \ - '-vga std ' \ - '-pidfile rina-%(id)s.pid ' \ - '-serial file:%(vmname)s.log ' \ - % vars_dict + command += ('-kernel %(vmimgpath)s/bzImage ' + '-append "console=ttyS0" ' + '-initrd %(vmimgpath)s/rootfs.cpio ' + % vars_dict) + command += ('-nographic ' + '-display none ' + '--enable-kvm ' + '-smp 1 ' + '-m %(memory)sM ' + '-device %(frontend)s,mac=%(mac)s,netdev=mgmt ' + '-netdev user,id=mgmt,%(hostfwdstr)s ' + '-vga std ' + '-pidfile rina-%(id)s.pid ' + '-serial file:%(vmname)s.log ' + % vars_dict + ) del vars_dict @@ -129,63 +197,107 @@ class Testbed(mod.Testbed): mac = '00:0a:0a:0a:%02x:%02x' % (vmid, port['port_id']) port['mac'] = mac - command_str += '-device %(frontend)s,mac=%(mac)s,netdev=data%(idx)s ' \ - '-netdev tap,ifname=%(tap)s,id=data%(idx)s,script=no,' \ - 'downscript=no%(vhost)s ' \ - % {'mac': mac, 'tap': tap_id, 'idx': port['port_id'], - 'frontend': vm_frontend, - 'vhost': ',vhost=on' if self.vhost else ''} - - command_str += '&\n' + command += ('-device %(frontend)s,mac=%(mac)s,netdev=data%(idx)s ' + '-netdev tap,ifname=%(tap)s,id=data%(idx)s,script=no,' + 'downscript=no%(vhost)s ' + % {'mac': mac, 'tap': tap_id, 'idx': port['port_id'], + 'frontend': vm_frontend, 'vhost': ',vhost=on' if self.vhost else ''} + ) booting_budget -= 1 if booting_budget <= 0: - command_str += 'sleep %s\n' % boot_backoff - booting_budget = boot_batch_size + pass # TODO manage the backoff time + # command_str += 'sleep %s\n' % boot_backoff + # booting_budget = boot_batch_size + + with open(self.qemu_folder + '/qemu_out{}'.format(vmid), 'w') as out_file: + print('DEBUG: executing >> {}'.format(command)) + self.boot_processes.append(Popen(command.split(), stdout=out_file)) + pass vmid += 1 - Popen([command_str], shell=True, executable="/bin/bash").wait() # TODO something more elegant maybe? - return command_str + return - def _make_down_script(self): + def __del__(self): """ :rtype str :return: The script to tear down the experiment """ - command_str = 'kill_qemu() {\n' \ - ' PIDFILE=$1\n' \ - ' PID=$(cat $PIDFILE)\n' \ - ' if [ -n $PID ]; then\n' \ - ' kill $PID\n' \ - ' while [ -n "$(ps -p $PID -o comm=)" ]; do\n' \ - ' sleep 1\n' \ - ' done\n' \ - ' fi\n' \ - '\n' \ - ' rm $PIDFILE\n' \ - '}\n\n' + # TERM qemu processes + for process in self.boot_processes: + process.terminate() - for vm_name, vm in self.vms.items(): - command_str += 'kill_qemu rina-%(id)s.pid\n' % {'id': vm['id']} - - command_str += '\n' + # Wait for them to shut down + for process in self.boot_processes: + process.wait() + port_processes = [] + error_queue = Queue() + results_queue = Queue() for vm_name, vm in self.vms.items(): for port in vm['ports']: tap = port['tap_id'] shim = port['shim'] - command_str += 'sudo brctl delif %(br)s %(tap)s\n' \ - 'sudo ip link set %(tap)s down\n' \ - 'sudo ip tuntap del mode tap name %(tap)s\n\n' \ - % {'tap': tap, 'br': shim.name} + commands = [] + + commands += ('sudo brctl delif %(br)s %(tap)s\n' + 'sudo ip link set %(tap)s down\n' + 'sudo ip tuntap del mode tap name %(tap)s' + % {'tap': tap, 'br': shim.name} + ).split('\n') + process = Process(target=self._run_command_chain, args=(commands, results_queue, error_queue)) + port_processes.append(process) + process.start() + + total_processes = len(port_processes) + max_waiting_time = 2 * total_processes + over_processes = 0 + + while max_waiting_time > 0 and over_processes < total_processes: + # Check for errors + if not error_queue.empty(): + print('Failure while shutting down: {}'.format(str(error_queue.get()))) + over_processes += 1 + try: + # Check for results + result = results_queue.get(timeout=1) + if result == "Command chain ran correctly.": + over_processes += 1 + print('DEBUG: %s of %s tear-down port processes completed.' % (over_processes, total_processes)) + except Empty: + max_waiting_time -= 1 + + error_queue = Queue() + results_queue = Queue() + shim_processes = [] for shim in self.shims: - command_str += 'sudo ip link set %(br)s down\n' \ - 'sudo brctl delbr %(br)s\n' \ - '\n' % {'br': shim.name} - return command_str + commands = [] + commands += ('sudo ip link set %(br)s down\n' + 'sudo brctl delbr %(br)s' + % {'br': shim.name} + ).split('\n') + process = Process(target=self._run_command_chain, args=(commands, results_queue, error_queue)) + shim_processes.append(process) + process.start() - def __del__(self): - script = self._make_down_script() - Popen([script], shell=True, executable="/bin/bash") + total_processes = len(shim_processes) + max_waiting_time = 2 * total_processes + over_processes = 0 + + while max_waiting_time > 0 and over_processes < total_processes: + # Check for errors + if not error_queue.empty(): + print('Failure while shutting down: {}'.format(str(error_queue.get()))) + over_processes += 1 + try: + # Check for results + result = results_queue.get(timeout=1) + if result == "Command chain ran correctly.": + over_processes += 1 + print('DEBUG: %s of %s tear-down shim processes completed.' % (over_processes, total_processes)) + except Empty: + max_waiting_time -= 1 + + return |