pwrkap-gui 7.30-5 / usr / lib / pwrkap / ui_data.py

#!/usr/bin/python

"""Various data objects for the UI."""
# (C) Copyright IBM Corp. 2008-2009
# Licensed under the GPLv2.

import cPickle as pickle
import socket
import datetime
import traceback
import time
import pwrkap_data
import threading

# Event types
HOST_CONNECTED = 1
HOST_DISCONNECTED = 2
HOST_DOMAIN_DATA_RECEIVED = 3

DEFAULT_MAX_AGE = 1800
class historical_data:
	"""Storage for power cap/use data."""

	def __init__(self, max_age):
		"""Create storage."""
		self.clear()
		self.max_age = max_age

	def clear(self):
		"""Clear all data."""
		self.time = []
		self.seconds = []
		self.cap = []
		self.power = []
		self.energy = []
		self.util_details = {}

	def get_max_sample_age(self):
		"""Retrieve the maximum record age."""
		return self.max_age

	def set_max_sample_age(self, new_max_age):
		"""Set the maximum age that records are kept."""
		self.max_age = new_max_age
		self.scrub_samples()

	def scrub_samples(self):
		"""Remove old records."""
		now = datetime.datetime.utcnow()
		for i in range(0, len(self.time)):
			if (now - self.time[i]).seconds <= self.max_age:
				del self.time[0:i]
				del self.seconds[0:i]
				del self.cap[0:i]
				del self.power[0:i]
				del self.energy[0:i]
				for dev in self.util_details.keys():
					del self.util_details[dev][0:i]
				break

	def retrieve_as_lists(self, since = None):
		"""Retrieve all data as a tuple of lists."""
		if since == None:
			return (self.seconds, self.cap, self.power, \
				self.energy, self.util_details)
	
		for i in range(0, len(self.time)):
			if self.time[i] > since:
				x = {}
				for dev in self.util_details.keys():
					x[dev] = self.util_details[dev][i:]
				return (self.seconds[i:], self.cap[i:], \
					self.power[i:], self.energy[i:], x)
		return None

	def store(self, time_stamp, cap, power, energy, util_details):
		"""Store a snapshot of time, cap, power and utilization."""
		if len(self.time) > 0:
			last_time = self.time[-1]
			if time_stamp <= last_time:
				return
		self.time.append(time_stamp)
		self.seconds.append(time.mktime(time_stamp.timetuple()))
		self.cap.append(cap)
		self.power.append(power)
		self.energy.append(energy)
		for dev in util_details.keys():
			if not self.util_details.has_key(dev):
				self.util_details[dev] = [0] * (len(self.time) - 1)
			self.util_details[dev].append(util_details[dev])
		self.scrub_samples()

	def get_most_recent(self):
		"""Retrieve most recent samples."""
		if len(self.time) == 0:
			return None
		idx = len(self.time) - 1
		ud = {}
		for dev in self.util_details.keys():
			ud[dev] = self.util_details[dev][-1]
		return (self.time[-1], self.cap[-1], self.power[-1], \
			self.energy[-1], ud)

class pwrkap_host:
	"""Represents a connection to a pwrkap server."""

	def __init__(self, host, port):
		"""Connect to a given host:port for pwrkap data."""
		self.host = host
		self.port = port
		self.domains = {}
		self.socket = None
		self.sockfile = None
		self.exit_loop = False
		self.event_listeners = []

	def get_name(self):
		"""Create name."""
		return "%s:%d" % (self.host, self.port)

	def add_listener(self, obj):
		"""Add an event listener."""
		self.event_listeners.append(obj)

	def remove_listener(self, obj):
		"""Remove an event listener."""
		self.event_listeners.remove(obj)

	def connect(self):
		"""Connect to the server."""
		assert self.socket == None

		self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
		print "Connecting to %s:%d" % (self.host, self.port)
		self.socket.connect((self.host, self.port))
		self.sockfile = self.socket.makefile()

	def set_up_socket(self):
		"""Prepare socket for use."""
		# Read list of (domain, domain_descr) tuples
		domains = pickle.load(self.sockfile)
		for (name, descr) in domains:
			phd = pwrkap_host_domain(self, self.port, name)
			self.domains[name] = phd

		# Read list of old snapshots (time, (domain, snapshot)) tuples
		# being careful to look for (time, 'live')
		live_time = None
		old_data = []
		print (self, 1, datetime.datetime.utcnow())
		while True:
			(time, obj) = pickle.load(self.sockfile)
			if obj == 'live':
				live_time = time
				break;
			old_data.append((time, obj))
		print (self, 2, datetime.datetime.utcnow())

		# Now recalibrate the timestamps.
		assert live_time != None
		now = datetime.datetime.utcnow()
		time_delta = now - live_time

		#1-2 and 3-4 are slow
		# Recalibrate the historical data and store
		for (time, obj) in old_data:
			(domain, snap) = obj
			x = (time + time_delta, snap)
			if "energy" in snap.keys():
				energy = snap["energy"]
			else:
				energy = None
			src_ud = {}
			dst_ud = {}
			if snap.has_key("util_details"):
				src_ud = snap["util_details"]
			else:
				src_ud = {"dev": snap["utilization"]}
			for dev in src_ud:
				dst_ud["%s:%s:%s:%s" % (self.host, self.port, domain, dev)] = 100.0 * src_ud[dev]
			self.domains[domain].historical_data.store(time + time_delta, \
				snap["cap"], snap["power"], energy, dst_ud)

		# Set initial values of cap/power/util to the last snapshot
		# regardless of how old it is
		for dom_name in self.domains.keys():
			self.domains[dom_name].fudge_data()
		print (self, 3, datetime.datetime.utcnow())

	def disconnect(self):
		"""Disconnect from server and erase all data."""
		assert self.socket != None
		self.sockfile.close()
		self.sockfile = None
		self.socket.shutdown(socket.SHUT_RDWR)
		self.socket = None
		self.domains = {}

	def dispatch(self):
		"""Dispatch incoming data."""
		while True:
			(time, obj) = pickle.load(self.sockfile)
			(dom, snap) = obj
			time = datetime.datetime.utcnow()
			self.domains[dom].receive_snapshot(time, snap)

	def exit_control_loop(self):
		"""Exit control loop."""
		assert self.socket != None
		self.exit_loop = True
		self.sockfile.close()
		self.socket.shutdown(socket.SHUT_RDWR)

	def control_loop(self):
		"""Main control loop."""
		self.exit_loop = False

		while not self.exit_loop:
			try:
				self.connect()
			except Exception, e:
				print e
				traceback.print_exc()
				self.socket = None
				time.sleep(1)
				continue
			try:
				self.set_up_socket()
			except Exception, e:
				print e
				traceback.print_exc()
				self.sockfile.close()
				self.socket.shutdown(socket.SHUT_RDWR)
				self.socket = None
				continue
			self.send_event(HOST_CONNECTED)
			try:
				self.dispatch()
			except Exception, e:
				print e
				traceback.print_exc()
				self.disconnect()
				self.send_event(HOST_DISCONNECTED)

	def send_command(self, command):
		"""Send a command."""
		assert self.sockfile != None
		pickled = pickle.dump(command, self.sockfile, pickle.HIGHEST_PROTOCOL)
		self.sockfile.flush()

	def send_event(self, event_type, affected_object = None):
		"""Notify listeners of an event."""
		if affected_object == None:
			affected_object = self

		for listener in self.event_listeners:
			try:
				listener.pwrkap_event(event_type, affected_object)
			except Exception, e:
				traceback.print_exc()
				print e

class pwrkap_client_domain:
	"""Abstract base class for power cap client domains."""

	def receive_snapshot(self, time, snap):
		"""Receive a snapshot."""
		pass

	def send_command(self, command):
		"""Send a command."""
		pass

	def set_cap(self, new_cap):
		"""Set a power cap."""
		pass

	def set_max_sample_age(self, new_max_age):
		"""Set the maximum sample age."""
		pass

	def get_cap(self):
		"""Retrieve this domain's power cap."""
		pass

	def get_power(self):
		"""Retrieve this domain's power usage."""
		pass

	def get_energy(self):
		"""Retrieve this domain's energy usage."""
		pass

	def get_utilization_details(self):
		"""Retrieve this domain's utilization data."""
		pass

	def get_max_sample_age(self):
		"""Retrieve this domain's maximum record age."""
		pass

	def is_affected_by(self, other):
		"""Return true if a change to the other domain affects this one."""
		pass

	def get_historical_data_as_lists(self, since = None):
		"""Retrieve sample data as a series of lists."""
		pass

	def get_name(self):
		"""Create name."""
		pass

class pwrkap_aggregate_domain(pwrkap_client_domain):
	"""Aggregate of multiple power domains."""

	def __init__(self, domains, name):
		"""Create an aggregate domain of domains."""
		self.domains = domains

		self.historical_data = historical_data(DEFAULT_MAX_AGE)
		self.domain_profiles = {}
		for dom in self.domains:
			self.domain_profiles[dom] = None
		self.last_aggregation = None
		self.aggregation_lock = threading.Lock()
		self.name = name

	def get_name(self):
		"""Create name."""
		return self.name

	def unfilled_domain_profiles(self):
		"""Determine if we can use caching mechanism."""
		for dom in self.domains:
			if self.domain_profiles[dom] == None:
				return True
		return False

	def aggregate_snapshots(self, since):
		"""Aggregate snapshots together."""
		self.aggregation_lock.acquire()
		self.__aggregate_snapshots_a(since)
		self.aggregation_lock.release()

	def __aggregate_snapshots_a(self, since):
		"""Aggregate snapshots together (no locking)."""

		def find_next_sample(data):
			"""Find the next sample to process."""
			lowest_timestamp = None
			domain = None
			for dom in domain_data_keys:
				x = data[dom]
				if x[0] >= len(x[1]):
					continue
				if lowest_timestamp == None or x[1][x[0]] < lowest_timestamp:
					lowest_timestamp = x[1][x[0]]
					domain = dom
			if domain == None:
				return None
			dom_info = data[domain]
			cursor = dom_info[0]
			dom_info[0] = cursor + 1
			# Pull out the utilization data
			ud = {}
			for dev in dom_info[5].keys():
				ud[dev] = dom_info[5][dev][cursor]
			# return domain, timestamp, cap, power, energy, utilization
			return (domain, lowest_timestamp, dom_info[2][cursor], \
				dom_info[3][cursor], dom_info[4][cursor], ud)

		# Figure out if we need a full scan or if we can re-use
		# previously calculated data
		last_time = self.last_aggregation
		if last_time == None or self.unfilled_domain_profiles():
			self.historical_data.clear()
			last_time = since

		# Prepare our lists of items
		domain_data = {}
		to_read = 0
		ztime = datetime.datetime.utcnow()
		for dom in self.domains:
			x = dom.get_historical_data_as_lists(last_time)
			if x == None:
				continue
			(t, c, p, e, ud) = x
			to_read = to_read + len(t)
			domain_data[dom] = [0, t, c, p, e, ud]
			
		print "Ugh, I have %d samples to read." % to_read

		# Process all samples in order
		atime = datetime.datetime.utcnow()
		print ("listmake time", (atime-ztime))

		doms = self.domain_profiles.keys()
		domain_data_keys = domain_data.keys()

		sample = find_next_sample(domain_data)
		while sample != None:
			# Update the domain's profile
			d = sample[0]
			xtime = sample[1]
			self.domain_profiles[d] = sample[2:]

			sample = find_next_sample(domain_data)

			if sample != None:
				assert xtime <= sample[1]

			# Coalesce all profiles with the same timestamp.
			xtime = xtime - (xtime % 15)
			if sample != None:
				new_xtime = sample[1] - (sample[1] % 15)

				if new_xtime == xtime:
					continue

			# Otherwise, snapshot the world.
			cap = 0
			power = 0
			energy = 0
			util_details = {}
			num_doms = len(doms)
			if num_doms == 0:
				continue

			for dom in doms:
				res = self.domain_profiles[dom]
				if res == None:
					num_doms = num_doms - 1
					continue
				(c, p, e, ud) = res
				cap = cap + c
				power = power + p
				if e != None:
					energy = energy + e
				util_details.update(ud)

			utilization = pwrkap_data.average_utilization(util_details)
			utctime = datetime.datetime.fromtimestamp(xtime)
			self.historical_data.store(utctime, cap, power, \
				energy, util_details)
		btime = datetime.datetime.utcnow()
		print ("aggloop", (btime-atime))
		self.last_aggregation = datetime.datetime.utcnow()
		return

	def receive_snapshot(self, time, snap):
		"""Receive a snapshot."""
		assert False

	def send_command(self, command):
		"""Send a command."""
		assert False

	def set_cap(self, new_cap):
		"""Set a power cap."""
		# Figure out old caps so we can redistribute proportionally
		old_cap = 0
		missing_domains = False
		for dom in self.domains:
			c = dom.get_cap()
			if c == None:
				missing_domains = True
			old_cap = old_cap + c
		if missing_domains:
			print "BOO!  Some domain is missing a cap!"
			return

		# Now reallocate cap
		for dom in self.domains:
			dom.set_cap(new_cap * dom.get_cap() / old_cap)

	def set_max_sample_age(self, new_max_age):
		"""Set the maximum sample age."""
		for dom in self.domains:
			dom.set_max_sample_age(new_max_age)

	def get_cap(self):
		"""Retrieve this domain's power cap."""
		cap = 0
		for dom in self.domains:
			c = dom.get_cap()
			if c != None:
				cap = cap + c
		return cap

	def get_power(self):
		"""Retrieve this domain's power usage."""
		power = 0
		for dom in self.domains:
			p = dom.get_power()
			if p != None:
				power = power + p
		return power

	def get_energy(self):
		"""Retrieve this domain's energy usage."""
		energy = 0
		for dom in self.domains:
			e = dom.get_energy()
			if e != None:
				energy = energy + e
		return energy

	def get_utilization_details(self):
		"""Retrieve this domain's utilization."""
		ud = {}
		for dom in self.domains:
			ud.update(dom.get_utilization_details())
		return ud

	def get_max_sample_age(self):
		"""Retrieve this domain's maximum record age."""
		if len(self.domains) == 0:
			return None

		return self.domains[0].get_max_sample_age()

	def is_affected_by(self, other):
		"""Return true if a change to the other domain affects this one."""
		if other in self.domains:
			return True
		return False

	def get_historical_data_as_lists(self, since = None):
		"""Retrieve sample data as a series of lists."""
		if len(self.domains) == 1:
			return self.domains[0].get_historical_data_as_lists(since)

		if since == None:
			max_age = self.get_max_sample_age()
			td = datetime.timedelta(0, max_age)
			since = (datetime.datetime.utcnow() - td)
		self.aggregate_snapshots(since)
		return self.historical_data.retrieve_as_lists(since)

class pwrkap_host_domain(pwrkap_client_domain):
	"""Represents a power domain on a pwrkap server."""

	def __init__(self, pwrkap_host, port, name):
		"""Create a power domain."""
		global DEFAULT_MAX_AGE

		self.historical_data = historical_data(DEFAULT_MAX_AGE)
		self.name = name
		self.host = pwrkap_host
		self.port = port
		self.cap = None
		self.power = None
		self.energy = None
		self.utilization_details = None
		self.last_event_trigger = datetime.datetime.utcnow()

	def get_name(self):
		"""Create name."""
		return self.name

	def receive_snapshot(self, time, snap):
		"""Receive a snapshot."""
		if snap.has_key("energy"):
			energy = snap["energy"]
		else:
			energy = None
		if snap.has_key("util_details"):
			ud = snap["util_details"]
		else:
			ud = {"dev": 100.0 * snap["utilization"]}
		dst_ud = {}
		for dev in ud:
			dst_ud["%s:%s:%s:%s" % (self.host.host, self.port, self.name, dev)] = 100.0 * ud[dev]
		self.historical_data.store(time, snap["cap"], snap["power"], \
					   energy, dst_ud)
		self.cap = snap["cap"]
		self.power = snap["power"]
		self.energy = energy
		self.utilization_details = dst_ud
		self.host.send_event(HOST_DOMAIN_DATA_RECEIVED, self)

	def get_historical_data_as_lists(self, since = None):
		"""Retrieve sample data as a series of lists."""
		return self.historical_data.retrieve_as_lists(since)

	def send_command(self, command):
		"""Send a command."""
		cmd = [self.name]
		for x in command:
			cmd.append(x)
		self.host.send_command(cmd)

	def set_cap(self, new_cap):
		"""Set a power cap."""
		self.cap = new_cap
		self.send_command(["cap", "%d" % new_cap])

	def set_max_sample_age(self, new_max_age):
		"""Set the maximum age that snapshots are kept."""
		self.historical_data.set_max_sample_age(new_max_age)

	def fudge_data(self):
		"""Set initial values of cached data to the last snapshot."""
		res = self.historical_data.get_most_recent()
		if res == None:
			return
		(time, self.cap, self.power, self.energy, self.utilization_details) = res

	def get_cap(self):
		"""Retrieve this domain's power cap."""
		return self.cap

	def get_power(self):
		"""Retrieve this domain's power usage."""
		return self.power

	def get_energy(self):
		"""Retrieve this domain's energy usage."""
		return self.energy

	def get_utilization_details(self):
		"""Return utilization details."""
		return self.utilization_details

	def get_max_sample_age(self):
		"""Retrieve this domain's maximum record age."""
		return self.historical_data.get_max_sample_age()

	def is_affected_by(self, other):
		"""Return true if a change to the other domain affects this one."""
		if other == self:
			return True
		return False