pwrkap 7.30-5 / usr / lib / pwrkap / pwrkap_data.py

#!/usr/bin/python

"""Common data types for pwrkap drivers."""
# (C) Copyright IBM Corp. 2008-2009
# Licensed under the GPLv2.
import math
import transitions
import datetime
import threading
import traceback
import lazy_log

ONE_SECOND = datetime.timedelta(0, 0, 1)

def average_utilization(util_map):
	"""Calculate the average utilization from a map of device -> utilization."""
	assert len(util_map) > 0

	sum = 0.0
	for key in util_map.keys():
		sum = sum + util_map[key]
	return float(sum) / len(util_map)

class meter:
	"""Abstract base class for meter implementations."""

	def read(self):
		"""Read the meter."""
		pass

	def inventory(self):
		"""Return an inventory of the meter capabilities."""
		pass

	def get_latency(self):
		"""Return the average latency of the meter in seconds."""
		pass

class power_meter(meter):
	"""Abstract base class for power meters.  read() returns Watts."""
	pass

class energy_meter(meter):
	"""Abstract base class for energy meters.  read() returns Joules."""
	pass

import power_energy_meter

class device:
	"""Abstract base class for power-managed devices."""

	def get_prefix(self):
		"""Return the type of this device."""
		pass

	def get_power_states(self):
		"""Return a mapping of all possible power states to the device's ability to perform while in that state (in percent)."""
		pass

	def get_max_power_state(self):
		"""Return the maximum power state."""
		pass

	def set_max_power_state(self, max_pstate):
		"""Set the maximum power state."""
		pass

	def get_current_power_state(self):
		"""Return the current power state."""
		pass

	def snapshot(self):
		"""Return a snapshot of the current state of the device."""
		pass

	def inventory(self):
		"""Return an inventory of the device capabilities."""
		pass

	def get_utilization_details(self):
		"""Return a dictionary containing {subdevice: utilization} pairs.  If there are no subdevices, return a single {device: utilization} pair."""
		pass

	def get_name(self):
		"""Return the name of this device."""
		pass

	def snapshot(self):
		"""Take a snapshot of this device."""
		key = self.get_name()
		obj = {	"state": self.get_current_power_state(), \
			"max_state": self.get_max_power_state(), \
			"util_details": self.get_utilization_details()}

		return (key, obj)

	def start_load(self):
		"""Start a load for training purposes."""
		pass

	def stop_load(self):
		"""Stop the training load."""
		pass

class device_domain(device):
	"""A collection of devices that must be power-managed together."""

	def __init__(self, devices):
		"""Create a domain of power-managed devices."""
		assert len(devices) > 0
		self.devices = devices
		self.must_set_all = False

	def get_prefix(self):
		"""Return the prefix of the domain."""
		return "domain"

	def get_current_power_state(self):
		"""Return the current power state."""
		highest_seen = None
		for dev in self.devices:
			cps = dev.get_current_power_state()
			if highest_seen == None or highest_seen < cps:
				highest_seen = cps
		return highest_seen

	def get_power_states(self):
		"""Return a list of all possible power states."""
		return self.devices[0].get_power_states()

	def get_max_power_state(self):
		"""Return the maximum power state."""
		return self.devices[0].get_max_power_state()

	def set_max_power_state(self, max_pstate):
		"""Set the maximum power state."""
		if not self.must_set_all:
			return self.devices[0].set_max_power_state(max_pstate)
		res = True
		for dev in self.devices:
			res = res and dev.set_max_power_state(max_pstate)
		return res

	def get_utilization_details(self):
		"""Merge and return maps of device utilization."""
		dom_util_map = {}

		for dev in self.devices:
			dev_util_map = dev.get_utilization_details()
			assert dev_util_map != None
			dom_util_map.update(dev_util_map)

		return dom_util_map

	def snapshot(self):
		"""Take a snapshot of this device domain."""
		snap_list = {}
		for dev in self.devices:
			(k, v) = dev.snapshot()
			snap_list[k] = v
		return snap_list

	def inventory(self):
		"""Take an inventory of this device domain."""
		inv_list = {}
		for dev in self.devices:
			(k, v) = dev.inventory()
			inv_list[k] = v
		return inv_list

	def get_device(self):
		"""Return a device that represents this domain."""
		return self.devices[0]

	def start_load(self):
		loaded = []

		for dev in self.devices:
			if dev.start_load():
				loaded.append(dev)
			else:
				self.stop_load_for(loaded)	
				return False
		return True

	def stop_load_for(self, devices):
		for dev in devices:
			dev.stop_load()

	def stop_load(self):
		self.stop_load_for(self.devices)

class IllegalDomain(Exception): pass

class power_domain_volatile_data:
	"""Dummy class to isolate power domain data that can't be preserved."""
	def __init__(self):
		self.signal = threading.Condition()

	def __getstate__(self):
		pass

	def __setstate__(self, state):
		self.signal = threading.Condition()

NUM_UTIL_BUCKETS = 4
ENFORCEMENT_INTERVAL = 30
MEASUREMENT_PERIOD = 15
ENFORCEMENT_SNAPSHOTS = 2
DEFAULT_SNAPSHOTS_TO_KEEP = 1000
class power_domain:
	"""A collection of power-managed device domains, a power
	meter, and various routines to manage them."""

	def __init__(self, domains, idomains, power_meter, energy_meter, cap):
		"""Create a power domain."""
		global NUM_UTIL_BUCKETS, DEFAULT_SNAPSHOTS_TO_KEEP

		assert len(domains) > 0
		self.power_meter = power_meter
		self.energy_meter = energy_meter
		if self.energy_meter == None:
			self.energy_meter = power_energy_meter.power_energy_meter(self.power_meter)
		self.domains = domains
		self.id = next_power_domain_id()
		self.cap = cap
		self.inter_domains = idomains
		self.snap_store = transitions.snapshot_store(DEFAULT_SNAPSHOTS_TO_KEEP)
		self.trans_store = transitions.transition_store(self.snap_store, self.inter_domains, NUM_UTIL_BUCKETS)
		self.check_idomain()
		self.control_loop_active = False
		self.need_enforcement = False
		self.volatile = power_domain_volatile_data()
		self.control_loop_should_exit = False
		self.last_enforcement = datetime.datetime.utcnow()

	def choose_domains_for_training(self):
		"""Return the smallest set of domains that are needed to
collect training data."""
		training = []

		for dom in self.domains:
			dev = dom.get_device()
			already_covered = False
			for idom in self.inter_domains:
				if idom[0] == dev:
					training.append(dom)
					already_covered = True
					break
				elif dev in idom:
					already_covered = True
			if not already_covered:
				training.append(dom)

		return training

	def check_idomain(self):
		"""Check interchangeable domains for problems."""
		# No empty idoms
		for idom in self.inter_domains:
			assert len(idom) > 0
		
		# All devices must be part of an idom.
		devs = set()
		for dom in self.domains:
			for dev in dom.devices:
				devs.add(dev)
		for idom in self.inter_domains:
			for dev in idom:
				assert dev in devs
				devs.remove(dev)
		assert len(devs) == 0

	def get_utilization_details(self):
		"""Merge and return maps of domain utilization."""
		pdom_util_map = {}

		for dom in self.domains:
			dom_util_map = dom.get_utilization_details()
			assert dom_util_map != None
			pdom_util_map.update(dom_util_map)

		return pdom_util_map

	def get_energy_use(self):
		"""Return the power domain's energy use, in Joules."""
		return self.energy_meter.read()

	def get_power_use(self):
		"""Return the power domain's power use, in Watts."""
		return self.power_meter.read()

	def get_cap(self):
		"""Return the power domain's maximum usage."""
		return self.cap

	def set_cap(self, cap):
		"""Set a new cap for this domain."""
		self.cap = cap
		if self.control_loop_active:
			self.need_enforcement = True

			# Signal the control loop
			self.volatile.signal.acquire()
			self.volatile.signal.notify()
			self.volatile.signal.release()
		else:
			self.enforce_cap()
		return True

	def exit_control_loop(self):
		"""Terminate the control loop."""
		if not self.control_loop_active:
			return
		self.control_loop_should_exit = True
		self.volatile.signal.acquire()
		self.volatile.signal.notify()
		self.volatile.signal.release()

	def control_loop(self):
		try:
			self.do_control_loop()
		except Exception, ex:
			traceback.print_exc()

	def do_control_loop(self):
		"""Control loop for power use regulation."""
		global ENFORCEMENT_INTERVAL, MEASUREMENT_PERIOD, ENFORCEMENT_SNAPSHOTS
		snapshots_since_enforcement = 0

		self.control_loop_should_exit = False
		self.control_loop_active = True
		while True:
			# Are we being told to exit?
			if self.control_loop_should_exit:
				self.control_loop_active = False
				return

			before = datetime.datetime.utcnow()

			# Take snapshot
			(name, props) = self.process_snapshot()
			usage = props["power"]
			lazy_log.logger.log((name, props))
			snapshots_since_enforcement = snapshots_since_enforcement + 1

			# Figure out if we need to run the enforcement loop
			nowtime = datetime.datetime.utcnow()
			if self.need_enforcement or \
			   (nowtime - self.last_enforcement).seconds > ENFORCEMENT_INTERVAL or \
			   snapshots_since_enforcement >= ENFORCEMENT_SNAPSHOTS:
				snapshots_since_enforcement = 0
				self.need_enforcement = False
				self.do_enforce_cap(usage)
				self.last_enforcement = datetime.datetime.utcnow()

			# Now sleep for a bit?
			after = datetime.datetime.utcnow()

			if (after - before).seconds < MEASUREMENT_PERIOD:
				self.volatile.signal.acquire()
				self.volatile.signal.wait(MEASUREMENT_PERIOD - (after - before).seconds)
				self.volatile.signal.release()

	def snapshot(self):
		"""Return a (key, obj) representation of the power domain."""
		def snapshot_domains(list):
			"""Snapshot a list of domains."""
			snap_list = []
			for domain in list:
				snap_list.append(domain.snapshot())
			return snap_list

		ud = self.get_utilization_details()
		aud = average_utilization(ud)

		key = self.name()
		obj = { "utilization":	aud, \
			"power":	self.get_power_use(), \
			"energy":	self.get_energy_use(), \
			"domains":	snapshot_domains(self.domains), \
			"cap":		self.get_cap(), \
			"util_details": ud}

		return (key, obj)

	def process_snapshot(self):
		"""Capture and record a snapshot."""
		snap = self.snapshot()
		self.trans_store.consider_snapshot(snap[1])
		return snap

	def inventory(self):
		"""Return a (key, obj) representation of the power domain's \
capabilities."""
		def inventory_domains(list):
			"""Inventory a list of domains."""
			inv_list = []
			for domain in list:
				inv_list.append(domain.inventory())
			return inv_list

		(pmeter_name, pmeter_data) = self.power_meter.inventory()
		(emeter_name, emeter_data) = self.energy_meter.inventory()
		key = self.name()
		obj = { "domains":	inventory_domains(self.domains), \
			"pmeter":	{pmeter_name: pmeter_data},
			"emeter":	{emeter_name: emeter_data}}

		return (key, obj)

	def name(self):
		"""Return the domain's name."""
		return "pwrdom" + str(self.id)

	def enforce_cap(self):
		"""Try to enforce the power cap on a one-off basis."""
		return self.do_enforce_cap(self.get_power_use())

	def do_enforce_cap(self, power_use):
		"""Try to enforce the power cap given a power usage reading."""
		recent_transitions = set()
		recent_devs = set()

		# Make it so that we don't go back to where we started from
		for domain in self.domains:
			state = domain.get_current_power_state()
			recent_transitions.add((domain, state))

		# XXX: Fudge things a bit here--give ourselves 10W of headroom
		remaining_delta = self.get_cap() - power_use - 10
		original_delta = remaining_delta

		print "***********"
		print "delta=%(delta)d cap=%(cap)d usage=%(use)d" % {"delta": remaining_delta, "cap": self.get_cap(), "use": power_use}
		delta = self.change_power_target(remaining_delta, recent_transitions, recent_devs)
		while delta != None:
			remaining_delta = remaining_delta - delta
			if remaining_delta * original_delta < 0:
				break
			delta = self.change_power_target(remaining_delta, recent_transitions, recent_devs)

		print "Remaining delta=%(rem)dW" % {"rem": remaining_delta}
		if remaining_delta < 0:
			return remaining_delta

		return 0

	def change_power_target(self, delta, recent, recent_devs):
		"""Take one step towards changing the power use target."""
		proposals = []

		if delta == 0:
			return None

		# For each device domain, construct a set of transitions
		# from the (current state, utilization) to another state
		# for which we know the power cost.  XXX: If there is no
		# data for (c0, u) -> (c1), can we use (c0, u') -> (c1)
		# instead?
		for domain in self.domains:
			dev_proposals = self.trans_store.propose_transitions(domain)
			for prop in dev_proposals:
				# Don't monkey around with CPUs we've already modified
				if prop.device in recent_devs:
					continue

				# Ignore proposals that don't change power use.
				if prop.power_impact == 0:
					continue

				# If we have to decrease power use, eliminate
				# options that consume more energy.
				if delta < 0 and prop.power_impact > 0:
					continue

				# If delta positive, do not pick any option that results
				# in a performance decrease.  This greedy algorithm only
				# cares about now; it does not look ahead.
				if delta > 0 and prop.performance_impact < 0:
					continue

				# Eliminate transitions that exceed the desired delta
				# if the delta is positive.
				if delta > 0 and prop.power_impact > delta:
					continue

				# Else, add to proposal list.  Note that it is
				# quite valid to have options that increase
				# peformance and decrease energy use!
				proposals.append(prop)

		# If there are no transitions left, we're stuck; exit
		if len(proposals) == 0:
			return None

		# Sort transitions in order of goodness.
		proposals.sort(cmp = transitions.compare_proposals)

		if True:
			print "------------------"
			for x in proposals:
				print (x.device.get_device().inventory()[0], x.new_state, 100*x.performance_impact, \
					x.power_impact, x.performance_impact / x.power_impact)

		# Pick the best one
		proposal = None
		for prop in proposals:
			if (prop.device, prop.new_state) in recent:
				continue
			proposal = prop
			break
		if proposal == None:
			return None
		print ("I CHOOSE", proposal)

		# Implement it.
		proposal.device.set_max_power_state(proposal.new_state)

		# Remember that fact.
		recent.add((proposal.device, proposal.new_state))
		recent_devs.add(proposal.device)

		return proposal.power_impact

	def start_load(self):
		loaded = []

		for dom in self.domains:
			if dom.start_load():
				loaded.append(dom)
			else:
				self.stop_load_for(loaded)	
				return False
		return True

	def stop_load_for(self, domains):
		for dom in domains:
			dom.stop_load()

	def stop_load(self):
		self.stop_load_for(self.domains)

def detect_idomains_for_devices(devices):
	"""Compute identical-domains for a list of devices."""
	idomains = []

	for device in devices:
		(name, data) = device.inventory()
		prefix = device.get_prefix()

		dev_idomain = []
		for idomain in idomains:
			(idom_name, idom_data) = idomain[0].inventory()
			idom_prefix = idomain[0].get_prefix()
			if idom_prefix == prefix and idom_data == data:
				dev_idomain = idomain
				break
		if len(dev_idomain) == 0:
			idomains.append(dev_idomain)
		dev_idomain.append(device)

	return idomains

next_dom_id = 0
def next_power_domain_id():
	"""Return a unique power domain identifier."""
	global next_dom_id

	next_dom_id = next_dom_id + 1
	return next_dom_id - 1