How to use set_io_controls method in autotest

Best Python code snippet using autotest_python Github


Full Screen

...287 paths = glob.glob('/sys/block/sd*')288 if not paths:289 paths = glob.glob('/sys/block/hd*')290 return [os.path.basename(path) for path in paths]291def set_io_controls(container_name, disks=[], ioprio_classes=[PROPIO_NORMAL],292 io_shares=[95], io_limits=[0]):293 # set the propio controls for one container, for selected disks294 # writing directly to /dev/cgroup/container_name/io.io_service_level295 # without using containerd or container.py296 # See wiki ProportionalIOScheduler for definitions297 # ioprio_classes: list of service classes, one per disk298 # using numeric propio service classes as used by kernel API, namely299 # 1: RT, Real Time, aka PROPIO_PRIO300 # 2: BE, Best Effort, aka PROPIO_NORMAL301 # 3: PROPIO_IDLE302 # io_shares: list of disk-time-fractions, one per disk,303 # as percentage integer 0..100304 # io_limits: list of limit on/off, one per disk305 # 0: no limit, shares use of other containers' unused disk time306 # 1: limited, container's use of disk time is capped to given DTF307 # ioprio_classes defaults to best-effort308 # io_limit defaults to no limit, use slack time309 if not disks: # defaults to all drives310 disks = all_drive_names()311 io_shares = [io_shares [0]] * len(disks)312 ioprio_classes = [ioprio_classes[0]] * len(disks)313 io_limits = [io_limits [0]] * len(disks)314 if not (len(disks) == len(ioprio_classes) and len(disks) == len(io_shares)315 and len(disks) == len(io_limits)):316 raise error.AutotestError('Unequal number of values for io controls')317 service_level = io_attr(container_name, 'io_service_level')318 if not os.path.exists(service_level):319 return # kernel predates propio features320 # or io cgroup is mounted separately from cpusets321 disk_infos = []322 for disk,ioclass,limit,share in zip(disks, ioprio_classes,323 io_limits, io_shares):324 parts = (disk, str(ioclass), str(limit), str(share))325 disk_info = ' '.join(parts)326 utils.write_one_line(service_level, disk_info)327 disk_infos.append(disk_info)328 logging.debug('set_io_controls of %s to %s',329 container_name, ', '.join(disk_infos))330def abbrev_list(vals):331 """Condense unsigned (0,4,5,6,7,10) to '0,4-7,10'."""332 ranges = []333 lower = 0334 upper = -2335 for val in sorted(vals)+[-1]:336 if val != upper+1:337 if lower == upper:338 ranges.append(str(lower))339 elif lower <= upper:340 ranges.append('%d-%d' % (lower, upper))341 lower = val342 upper = val343 return ','.join(ranges)344def create_container_with_specific_mems_cpus(name, mems, cpus):345 need_fake_numa()346 os.mkdir(full_path(name))347 utils.write_one_line(cpuset_attr(name, 'mem_hardwall'), '1')348 utils.write_one_line(mems_path(name), ','.join(map(str, mems)))349 utils.write_one_line(cpus_path(name), ','.join(map(str, cpus)))350 logging.debug('container %s has %d cpus and %d nodes totalling %s bytes',351 name, len(cpus), len(get_mem_nodes(name)),352 utils.human_format(container_bytes(name)) )353def create_container_via_memcg(name, parent, bytes, cpus):354 # create container via direct memcg cgroup writes355 os.mkdir(full_path(name))356 nodes = utils.read_one_line(mems_path(parent))357 utils.write_one_line(mems_path(name), nodes) # inherit parent's nodes358 utils.write_one_line(memory_path(name)+'.limit_in_bytes', str(bytes))359 utils.write_one_line(cpus_path(name), ','.join(map(str, cpus)))360 logging.debug('Created container %s directly via memcg,'361 ' has %d cpus and %s bytes',362 name, len(cpus), utils.human_format(container_bytes(name)))363def _create_fake_numa_container_directly(name, parent, mbytes, cpus):364 need_fake_numa()365 lockfile = my_lock('inner') # serialize race between parallel tests366 try:367 # Pick specific mem nodes for new cpuset's exclusive use368 # For now, arbitrarily pick highest available node numbers369 needed_kbytes = mbytes * 1024370 nodes = sorted(list(available_exclusive_mem_nodes(parent)))371 kbytes = 0372 nodecnt = 0373 while kbytes < needed_kbytes and nodecnt < len(nodes):374 nodecnt += 1375 kbytes += node_avail_kbytes(nodes[-nodecnt])376 if kbytes < needed_kbytes:377 parent_mbytes = container_mbytes(parent)378 if mbytes > parent_mbytes:379 raise error.AutotestError(380 "New container's %d Mbytes exceeds "381 "parent container's %d Mbyte size"382 % (mbytes, parent_mbytes) )383 else:384 raise error.AutotestError(385 "Existing sibling containers hold "386 "%d Mbytes needed by new container"387 % ((needed_kbytes - kbytes)//1024) )388 mems = nodes[-nodecnt:]389 create_container_with_specific_mems_cpus(name, mems, cpus)390 finally:391 my_unlock(lockfile)392def create_container_directly(name, mbytes, cpus):393 parent = os.path.dirname(name)394 if fake_numa_containers:395 _create_fake_numa_container_directly(name, parent, mbytes, cpus)396 else:397 create_container_via_memcg(name, parent, mbytes<<20, cpus)398def create_container_with_mbytes_and_specific_cpus(name, mbytes,399 cpus=None, root=SUPER_ROOT, io={}, move_in=True, timeout=0):400 """\401 Create a cpuset container and move job's current pid into it402 Allocate the list "cpus" of cpus to that container403 name = arbitrary string tag404 mbytes = reqested memory for job in megabytes405 cpus = list of cpu indicies to associate with the cpuset406 defaults to all cpus avail with given root407 root = the parent cpuset to nest this new set within408 '': unnested top-level container409 io = arguments for proportional IO containers410 move_in = True: Move current process into the new container now.411 timeout = must be 0: persist until explicitly deleted.412 """413 need_mem_containers()414 if not container_exists(root):415 raise error.AutotestError('Parent container "%s" does not exist'416 % root)417 if cpus is None:418 # default to biggest container we can make under root419 cpus = get_cpus(root)420 else:421 cpus = set(cpus) # interface uses list422 if not cpus:423 raise error.AutotestError('Creating container with no cpus')424 name = os.path.join(root, name) # path relative to super_root425 if os.path.exists(full_path(name)):426 raise error.AutotestError('Container %s already exists' % name)427 create_container_directly(name, mbytes, cpus)428 set_io_controls(name, **io)429 if move_in:430 move_self_into_container(name)431 return name432def get_boot_numa():433 # get boot-time numa=fake=xyz option for current boot434 # eg numa=fake=nnn, numa=fake=nnnM, or nothing435 label = 'numa=fake='436 for arg in utils.read_one_line('/proc/cmdline').split():437 if arg.startswith(label):438 return arg[len(label):]...

Full Screen

Full Screen

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.

LambdaTest Learning Hubs:


You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run autotest automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?