diff --git a/TODO b/TODO
new file mode 100644
index 0000000..08150be
--- /dev/null
+++ b/TODO
@@ -0,0 +1 @@
+remove thrift cruft
diff --git a/etc/NodeManager.cfg b/etc/NodeManager.cfg
index a47bccf..48f4044 100644
--- a/etc/NodeManager.cfg
+++ b/etc/NodeManager.cfg
@@ -80,7 +80,6 @@
 statsInterval = 0.0
 ;accountingHost = clustermanager
 ;accountingPort = 2228
-;bind = 0.0.0.0 ; not supported (Thrift is missing support to specify what to bind to!)
 
 [Security]
 authAndEncrypt = False
diff --git a/etc/TashiDefaults.cfg b/etc/TashiDefaults.cfg
index fd034eb..101da69 100644
--- a/etc/TashiDefaults.cfg
+++ b/etc/TashiDefaults.cfg
@@ -57,7 +57,6 @@
 allowDuplicateNames = False
 ;accountingHost = clustermanager
 ;accountingPort = 2228
-;bind = 0.0.0.0 ; not supported (Thrift is missing support to specify what to bind to!)
 
 [GetentOverride]
 baseData = tashi.clustermanager.data.Pickled
@@ -110,7 +109,6 @@
 clusterManagerHost = localhost 
 clusterManagerPort = 9882
 statsInterval = 0.0
-;bind = 0.0.0.0 ; not supported (Thrift is missing support to specify what to bind to!)
 
 [Qemu]
 qemuBin = /usr/bin/kvm
diff --git a/src/tashi/agents/locality-server.py b/src/tashi/agents/locality-server.py
deleted file mode 100755
index a2b64a5..0000000
--- a/src/tashi/agents/locality-server.py
+++ /dev/null
@@ -1,228 +0,0 @@
-#!/usr/bin/python
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-#
-#   http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied.  See the License for the
-# specific language governing permissions and limitations
-# under the License.
-
-# this module provides a service to locate servers that are close
-# to a VM. Uses all-pairs shortest path algorithm. Need to provide
-# a topology for the underlying physical infrastructure.
-
-from socket import gethostname
-import os
-import threading
-import time
-import socket
-
-from tashi.services.ttypes import *
-
-from thrift import Thrift
-from thrift.transport import TSocket
-from thrift.transport import TTransport
-from thrift.protocol import TBinaryProtocol
-from thrift.server import TServer
-
-from tashi.services import clustermanagerservice
-from tashi.util import getConfig, createClient
-from tashi.parallel import *
-
-import tashi.services.layoutlocality.localityservice as localityservice
-
-from numpy import *
-from scipy import *
-
-cnames = {}
-def cannonicalName(hn):
-	try:
-		if cnames.has_key(hn):
-			return cnames[hn]
-		r = socket.gethostbyname_ex(hn)[0]
-		cnames[hn] = r
-		return r
-	except:
-		return hn
-
-# define matrix multiplication that can be used to calculate a min-plus
-# distance product
-def genMul(A, B, add, mult):
-	'''generalized matrix multiplication'''
-	C = zeros((shape(A)[0], shape(B)[1]))
-	for i in range(shape(C)[0]):
-		for j in range(shape(C)[1]):
-			C[i,j] = add(mult(A[i,:], B[:,j]))
-	return C
-
-def addHost(graph, hostVals, host):
-	if not graph.has_key(host):
-		graph[host] = []
-	if not hostVals.has_key(host):
-		hostVals[host] = len(hostVals)
-
-def graphConnect(graph, h1, h2):
-	if not h1 in graph[h2]:
-		graph[h2].append(h1)
-	if not h2 in graph[h1]:
-		graph[h1].append(h2)
-
-def graphFromFile(fn = 'serverLayout', graph = {}, hostVals = {}):
-	f = open(fn)
-	for line in f.readlines():
-		line = line.split()
-		if len(line) < 1:
-			continue
-		server = cannonicalName(line[0].strip())
-
-		addHost(graph, hostVals, server)
-		for peer in line[1:]:
-			peer = cannonicalName(peer.strip())
-			addHost(graph, hostVals, peer)
-			graphConnect(graph, server, peer)
-	return graph, hostVals
-
-def graphFromTashi(client, transport, graph={}, hostVals={}):
-	print 'getting graph'
-	if not transport.isOpen():
-		transport.open()
-	hosts = client.getHosts()
-	instances = client.getInstances()
-	for instance in instances:
-		host = [cannonicalName(h.name) for h in hosts if h.id == instance.hostId]
-		if len(host) <1 :
-			print 'cant find vm host'
-			continue
-		host = host[0]
-		print 'host is ', host
-		addHost(graph, hostVals, host)
-		print 'added host'
-		vmhost = cannonicalName(instance.name)
-		addHost(graph, hostVals, vmhost)
-		print 'added vm'
-		graphConnect(graph, host, vmhost)
-		print 'connected'
-	print 'returning from graphFromTashi'
-	return graph, hostVals
-
-
-
-def graphToArray(graph, hostVals):
-	a = zeros((len(hostVals), len(hostVals)))
-	for host in graph.keys():
-		if not hostVals.has_key(host):
-			continue
-		a[hostVals[host], hostVals[host]] = 1
-		for peer in graph[host]:
-			if not hostVals.has_key(peer):
-				continue
-			a[hostVals[host], hostVals[peer]] = 1
-	a[a==0] = inf
-	for i in range(shape(a)[0]):
-		a[i,i]=0
-	return a
-
-def shortestPaths(graphArray):
-	a = graphArray
-	for i in range(math.ceil(math.log(shape(a)[0],2))):
-		a = genMul(a,a,min,plus)
-	return a
-
-def plus(A, B):
-	return A + B
-
-
-def getHopCountMatrix(sourceHosts, destHosts, array, hostVals):
-	a = zeros((len(sourceHosts), len(destHosts)))
-	a[a==0] = inf
-	for i in range(len(sourceHosts)):
-		sh = cannonicalName(sourceHosts[i])
-		shv = None
-		if hostVals.has_key(sh):
-			shv = hostVals[sh]
-		else:
-			print 'host not found', sh
-			continue
-		for j in range(len(destHosts)):
-			dh = cannonicalName(destHosts[j])
-			dhv = None
-			if hostVals.has_key(dh):
-				dhv = hostVals[dh]
-			else:
-				print 'dest not found', dh
-				continue
-			print sh, dh, i,j, shv, dhv, array[shv, dhv]
-			a[i,j] = array[shv, dhv]
-	return a
-
-
-class LocalityService:
-	def __init__(self):
-		(config, configFiles) = getConfig(["Agent"])
-		self.port = int(config.get('LocalityService', 'port'))
-		print 'Locality service on port %i' % self.port
-		self.processor = localityservice.Processor(self)
-		self.transport = TSocket.TServerSocket(self.port)
-		self.tfactory = TTransport.TBufferedTransportFactory()
-		self.pfactory = TBinaryProtocol.TBinaryProtocolFactory()
-		self.server = TServer.TThreadedServer(self.processor,
-							self.transport,
-							self.tfactory,
-							self.pfactory)
-                
-		self.hostVals =[]
-		self.array = array([[]])
-		self.rtime = 0
-
-
-		self.fileName = os.path.expanduser(config.get("LocalityService", "staticLayout"))
-		(self.client, self.transport) = createClient(config)
-
-		self.server.serve()
-
-	@synchronizedmethod
-	def refresh(self):
-		if time.time() - self.rtime < 10:
-			return
-		g, self.hostVals = graphFromFile(self.fileName)
-		try:
-			g, self.hostVals = graphFromTashi(self.client, self.transport, g, self.hostVals)
-		except e:
-			print e
-			print 'could not get instance list from cluster manager'
-		print 'graph to array'
-		a = graphToArray(g, self.hostVals)
-		print 'calling shortest paths ', a.shape
-		self.array = shortestPaths(a)
-		print 'computed shortest paths'
-		print self.array
-		print self.hostVals
-	@synchronizedmethod
-	def getHopCountMatrix(self, sourceHosts, destHosts):
-		self.refresh()
-		print 'getting hop count matrix for', sourceHosts, destHosts
-		hcm =  getHopCountMatrix(sourceHosts, destHosts, self.array, self.hostVals)
-		print hcm
-		return hcm
-
-
-def main():
-
-	#XXXstroucki This code has not been updated for several years.
-	# It may still be useful as an example.
-	import sys
-	sys.exit(0);
-
-	ls = LocalityService()
-
-if __name__ == "__main__":
-	main()
diff --git a/src/tashi/client/client.py b/src/tashi/client/client.py
deleted file mode 100755
index 71b5b20..0000000
--- a/src/tashi/client/client.py
+++ /dev/null
@@ -1,213 +0,0 @@
-#! /usr/bin/env python
-
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-# 
-#   http://www.apache.org/licenses/LICENSE-2.0
-# 
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied.  See the License for the
-# specific language governing permissions and limitations
-# under the License.    
-
-import inspect
-import os
-import sys
-import types
-from tashi.services.ttypes import *
-from thrift.protocol.TBinaryProtocol import TBinaryProtocol
-from thrift.transport.TTransport import TBufferedTransport
-from thrift.transport.TSocket import TSocket
-
-from tashi.services import clustermanagerservice
-from tashi import vmStates
-
-from tashi.util import getConfig
-
-def makeHTMLTable(list):
-	(stdin_r, stdin_w) = os.pipe()
-
-# XXXpipe: find number of columns in current window
-	pipe = os.popen("tput cols")
-	columns = pipe.read().strip()
-	keys = {}
-	for k in list:
-		for k2 in k.__dict__.keys():
-			if (not k2.endswith("Obj")):
-				keys[k2] = k2
-	if ('id' in keys):
-		del keys['id']
-		keylist = ['id'] + keys.keys()
-	else:
-		keylist = keys.keys()
-	output = "<html>"
-	output = output + "<table>"
-	output = output + "<tr>"
-	for k in keylist:
-		output = output + "<td>%s</td>" % (k)
-	output = output + "</tr>"
-	for k in list:
-		output = output + "<tr>"
-		for k2 in keylist:
-			if (k2 == "state"):
-				output = output + "<td>%s</td>" % (str(vmStates[k.__dict__.get(k2, None)]))
-			else:
-				output = output + "<td>%s</td>" % (str(k.__dict__.get(k2, None)))
-		output = output + "</tr>"
-	output = output + "</table>"
-	output = output + "</html>"
-	pid = os.fork()
-	if (pid == 0):
-		os.close(stdin_w)
-		os.dup2(stdin_r, 0)
-		os.close(stdin_r)
-		os.execl("/usr/bin/lynx", "/usr/bin/lynx", "-width=%s" % (columns), "-dump", "-stdin")
-		sys.exit(-1)
-	os.close(stdin_r)
-	os.write(stdin_w, output)
-	os.close(stdin_w)
-	os.waitpid(pid, 0)
-
-def getFunction(argv):
-	"""Tries to determine the name of the function requested by the user -- may be called multiple times if the binary name is 'client'"""
-	function = "None"
-	if (len(argv) > 0):
-		function = argv[0].strip()
-		if (function.rfind("/") != -1):
-			function = function[function.rfind("/")+1:]
-		if (function.rfind(".") != -1):
-			function = function[:function.rfind(".")]
-	return function
-
-def getFunctionInfo(m):
-	"""Gets a string that describes a function from the interface"""
-	f = getattr(clustermanagerservice.Iface, m)
-	argspec = inspect.getargspec(f)[0][1:]
-	return m + inspect.formatargspec(argspec)
-
-def usage():
-	"""Print program usage"""
-	print "Available methods:"
-	for m in methods:
-		print "\t" + getFunctionInfo(m)
-	print
-	print "Examples:"
-	print "\tgetInstances"
-	print "\taddUser 'User(d={\"username\":\"foobar\"})'"
-	print "\tremoveUser 2"
-	print "\tcreateVM 1 1"
-
-def simpleType(obj):
-	"""Determines whether an object is a simple type -- used as a helper function to pprint"""
-	if (type(obj) is not type([])):
-		if (not getattr(obj, "__dict__", None)):
-			return True
-	return False
-
-def pprint(obj, depth = 0, key = None):
-	"""My own version of pprint that prints out a dict in a readable, but slightly more compact format"""
-	valueManip = lambda x: x
-	if (key):
-		keyString = key + ": "
-		if (key == "state"):
-			valueManip = lambda x: vmStates[x]
-	else:
-		keyString = ""
-	if (type(obj) is type([])):
-		if (reduce(lambda x, y: x and simpleType(y), obj, True)):
-			print (" " * (depth * INDENT)) + keyString + str(obj)
-		else:
-			print (" " * (depth * INDENT)) + keyString + "["
-			for o in obj:
-				pprint(o, depth + 1)
-			print (" " * (depth * INDENT)) + "]"
-	elif (getattr(obj, "__dict__", None)):
-		if (reduce(lambda x, y: x and simpleType(y), obj.__dict__.itervalues(), True)):
-			print (" " * (depth * INDENT)) + keyString + str(obj)
-		else:
-			print (" " * (depth * INDENT)) + keyString + "{"
-			for (k, v) in obj.__dict__.iteritems():
-				pprint(v, depth + 1, k)
-			print (" " * (depth * INDENT)) + "}"
-	else:
-		print (" " * (depth * INDENT)) + keyString + str(valueManip(obj))
-
-def main():
-	"""Main function for the client program"""
-	global INDENT, methods, exitCode
-	exitCode = 0
-	INDENT = (os.getenv("INDENT", 4))
-	methods = filter(lambda x: not x.startswith("__"), clustermanagerservice.Iface.__dict__.keys())
-	function = getFunction(sys.argv)
-	if (function == "client"):
-		function = getFunction(sys.argv[1:])
-	if (function == "--makesyms"):
-		for m in methods:
-			os.symlink(sys.argv[0], m)
-		sys.exit(0)
-	if (function == "--rmsyms"):
-		for m in methods:
-			os.unlink(m)
-		sys.exit(0)
-
-	(config,configFiles) = getConfig(["Client"])
-	cfgHost = config.get('Client', 'clusterManagerHost')
-	cfgPort = config.get('Client', 'clusterManagerPort')
-	cfgTimeout = float(config.get('Client', 'clusterManagerTimeout'))
-	host = os.getenv('TASHI_CM_HOST', cfgHost)
-	port = os.getenv('TASHI_CM_PORT', cfgPort)
-	timeout = float(os.getenv('TASHI_CM_TIMEOUT', cfgTimeout)) * 1000.0
-
-	socket = TSocket(host, int(port))
-	socket.setTimeout(timeout)
-	transport = TBufferedTransport(socket)
-	protocol = TBinaryProtocol(transport)
-	client = clustermanagerservice.Client(protocol)
-	client._transport = transport
-	client._transport.open()
-	f = getattr(client, function, None)
-	if not f:
-		usage()
-		sys.exit(-1)
-	args = map(lambda x: eval(x), sys.argv[1:])
-	try:
-		res = f(*args)
-		def cmp(x, y):
-			try:
-				if (x.id < y.id):
-					return -1
-				elif (y.id < x.id):
-					return 1
-				else:
-					return 0
-			except Exception, e:
-				return 0
-		if (type(res) == types.ListType):
-			res.sort(cmp)
-		if (os.getenv("USE_HTML_TABLES")):
-			try:
-				makeHTMLTable(res)
-			except:
-				pprint(res)
-		else:
-			pprint(res)
-	except TashiException, e:
-		print e.msg
-		exitCode = e.errno
-	except TypeError, e:
-		print e
-		print "\t" + getFunctionInfo(function)
-		exitCode = -1
-	finally:
-		client._transport.close()
-	sys.exit(exitCode)
-
-if __name__ == "__main__":
-	main()
diff --git a/src/tashi/client/test.py b/src/tashi/client/test.py
deleted file mode 100644
index a53eefa..0000000
--- a/src/tashi/client/test.py
+++ /dev/null
@@ -1,314 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-# 
-#   http://www.apache.org/licenses/LICENSE-2.0
-# 
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied.  See the License for the
-# specific language governing permissions and limitations
-# under the License.    
-
-import unittest
-import logging
-import sys
-import signal
-import os.path
-import copy
-import time
-import random
-from ConfigParser import ConfigParser
-
-from tashi.services.ttypes import *
-from thrift.transport.TSocket import TSocket
-from thrift.protocol.TBinaryProtocol import TBinaryProtocol
-from thrift.transport.TTransport import TBufferedTransport
-
-from tashi.services import clustermanagerservice
-from tashi.messaging.threadpool import synchronized
-from tashi.messaging.tashimessaging import TestTashiSubscriber
-
-from tashi.util import getConfig
-
-import tashi.client.client
-
-class ClientConnection(object):
-	'''Creates an rpc proxy'''
-	def __init__(self, host, port):
-		self.host = host
-		self.port = port
-		self.transport = TBufferedTransport(TSocket(host, int(port)))
-		self.protocol = TBinaryProtocol(self.transport)
-		self.client = clustermanagerservice.Client(self.protocol)
-		self.client._transport = self.transport
-		self.client._transport.open()
-	def __del__(self):
-		self.client._transport.close()
-
-def incrementor(init=0):
-	while 1:
-		yield init
-		init = init + 1
-
-# FIXME: don't duplicate code from clustermanager
-# def getConfig(args):
-#	 config = ConfigParser()
-#	 configFiles = [
-#		'/usr/share/tashi/ClusterManagerDefaults.cfg',
-#		'/etc/tashi/ClusterManager.cfg',
-#		os.path.expanduser('~/.tashi/ClusterManager.cfg')
-#		] + ([args[0]] if len(args) > 0 else [])
-
-#	 configFiles = config.read(configFiles)
-#	 if len(configFiles) == 0:
-#		 print >>sys.stderr, 'Unable to find the configuration file\n'
-#		 sys.exit(3)
-
-#	 return config
-
-
-class TestClient(unittest.TestCase):
-	@synchronized()
-	def getPortNum(self):
-		return self.portnum.next()
-
-	"""macro test cases for single-host tests
-
-	Assumes cwd is 'src/tashi/client/'
-	"""
-	def setUp(self):
-		"""Create a CM and single NM on local host"""
-		logging.info('setting up test')
-		
-		(self.config, self.configfiles) = getConfig([])
-
-		self.port = 1717		# FIXME: take this (and other things) from config file
-		self.portnum = incrementor(self.port)
-
-		self.cwd = os.getcwd()
-		self.srcd = os.path.dirname(os.path.dirname(self.cwd))
-		
-		self.environ = copy.copy(os.environ)
-		self.environ['PYTHONPATH'] = self.srcd
-		logging.info('base path = %s' % self.srcd)
-
-		self.nm = os.spawnlpe(os.P_NOWAIT, 'python', 'python', 
-							  os.path.join(self.srcd, 'tashi', 'nodemanager', 'nodemanager.py'),
-							  self.environ)
-		self.cm = os.spawnlpe(os.P_WAIT, 'python', 'python',
-							 os.path.join(self.srcd,  'tashi', 'clustermanager', 'clustermanager.py'),
-							 '--drop', '--create',
-							 os.path.expanduser('~/.tashi/ClusterManager.cfg'),
-							 self.environ)
-		self.cm = os.spawnlpe(os.P_NOWAIT, 'python', 'python',
-							 os.path.join(self.srcd,  'tashi', 'clustermanager', 'clustermanager.py'),
-							 os.path.expanduser('~/.tashi/ClusterManager.cfg'),
-							 self.environ)
-		# since we are spawning with P_NOWAIT, we need to sleep to ensure that the CM is listening
-		time.sleep(1)
-		try:
-			self.connection = ClientConnection('localhost', self.config.get('ClusterManagerService', 'port'))
-		except Exception, e:
-			logging.warning('client connection failed')
-			ex = None
-			try:
-				logging.warning("setUp killing node manager " + str(self.nm))
-				os.kill(self.nm, signal.SIGKILL)
-			except Exception, e:
-				ex = e
-				logging.warning('could not kill node manager: '+ str(e))
-			try:
-				logging.warning('setUp killing cluster manager ' + str(self.cm))
-				os.kill(self.cm, signal.SIGKILL)
-			except Exception, e:
-				ex = e
-				logging.warning('could not kill cluster manager: ' + str(e))
-			if e != None:
-				raise e
-
-		logging.info('node manager PID: %i' % self.nm)
-	def tearDown(self):
-		'''Kill the CM and NM that were created by setUP'''
-		logging.info('tearing down test')
-		ex = None
-		try:
-			logging.debug("killing cluster manager " + str(self.cm))
-			os.kill(self.cm, signal.SIGKILL)
-		except Exception, e:
-			ex = e
-			logging.error('Could not kill cluster manager: ' + str(e))
-			
-		try:
-			logging.debug("killing node manager " + str(self.nm))
-			os.kill(self.nm, signal.SIGKILL)
-		except Exception, e:
-			ex = e
-			logging.error('Could not kill node manager: ' + str(e))
-		if ex != None:
-			raise ex
-	def testSetup(self):
-		'''empty test to ensure that setUp code works'''
-		logging.info('setting up')
-	def testHostManagement(self):
-		'''test adding/removing/listing hosts
-
-		Right now this just adds a single host: localhost.  Eventually
-		it should 1) take a list of hosts from a test configuration
-		file, 2) ensure that all were added, 3) remove a random
-		subset, 4) ensure that they were correctly removed, 5) remove
-		all, 6) ensure that they were correctly removed.'''
-
-		# get empty host list
-		hosts = self.connection.client.getHosts()
-		self.assertEqual(hosts, [], 'starting host list not empty: ' + str(hosts) )
-
-		# add a host
-		host = Host()
-		host.hostname = 'localhost'
-		host.enabled=True
-		self.connection.client.addHost(host)
-		hosts = self.connection.client.getHosts()
-		self.assertEqual(len(hosts), 1, 'wrong number of hosts %i, should be %i' % (len(hosts), 1) )
-		self.assertEqual(hosts[0].hostname, 'localhost', 'wrong hostname: ' + str(hosts[0].hostname) )
-
-		# remove first host
-		hid = hosts[0].id
-		self.connection.client.removeHost(hid)
-		hosts = self.connection.client.getHosts()
-		self.assertEqual(hosts, [], 'host list not empty after remove: ' + str(hosts) )
-
-	def testMessaging(self):
-		'''test messaging system started by CM
-
-		tests messages published directly, through events in the CM,
-		and the log system'''
-		# FIXME: add tests for generating events as a side-effect of
-		# rpc commands, as well as logging in the CM
-		portnum = self.getPortNum()
-		self.sub = TestTashiSubscriber(self.config, portnum)
-		self.assertEqual(self.sub.messageQueue.qsize(), 0)
-		self.pub = tashi.messaging.thriftmessaging.PublisherThrift(self.config.get('MessageBroker', 'host'),
-																   int(self.config.get('MessageBroker', 'port')))
-		self.pub.publish({'message-type':'text', 'message':'Hello World!'})
-		time.sleep(0.5)
-		print '*** QSIZE', self.sub.messageQueue.qsize()
-		self.assertEqual(self.sub.messageQueue.qsize(), 1)
-
-		self.log = logging.getLogger(__name__)
-		messageHandler = tashi.messaging.tashimessaging.TashiLogHandler(self.config)
-		self.log.addHandler(messageHandler)
-		# FIXME: why can't we log messages with severity below 'warning'?
-		self.log.warning('test log message')
-		time.sleep(0.5)
-		self.assertEqual(self.sub.messageQueue.qsize(), 2)
-
-		# This should generate at least one log message
-#		 hosts = self.connection.client.getHosts()
-#		 time.sleep(0.5)
-#		 if (self.sub.messageQueue.qsize() <= 2):
-#			 self.fail()
-
-	def testUserManagement(self):
-		'''test adding/removing/listing users
-
-		same as testHostManagement, but with users'''
-		usernames = ['sleepy', 'sneezy', 'dopey', 'doc',
-					 'grumpy', 'bashful', 'happy']
-		# add all users
-		for un in usernames:
-			user = User()
-			user.username = un
-			self.connection.client.addUser(user)
-		# ensure that all were added
-		users = self.connection.client.getUsers()
-		self.assertEqual(len(usernames), len(users))
-		for user in users:
-			usernames.remove(user.username)
-		self.assertEqual(0, len(usernames))
-		# remove a random subset
-		rm = random.sample(users, 4)
-		for user in rm:
-			self.connection.client.removeUser(user.id)
-			users.remove(user)
-		newUsers = self.connection.client.getUsers()
-		# This ensures that the remaining ones are what we expect:
-		for user in newUsers:
-			# if there is a user remaining that we asked to be removed,
-			# this will throw an exception
-			users.remove(user)
-		# if a user was removed that we did not intend, this will
-		# throw an exception
-		self.assertEqual(0, len(users))
-
-#	 def testInstanceConfigurationManagement(self):
-#		 '''test adding/removing/listing instance configurations
-
-#		 same as testHostManagement, but with instance configurations'''
-#		 self.fail('test not implemented')
-	def testHardDiskConfigurationManagement(self):
-		'''test adding/removing/listing hard disk configurations
-
-		same as testHostManagement, but with hard disk configurations'''
-
-		user = User(d={'username':'sleepy'})
-		self.connection.client.addUser(user)
-		users = self.connection.client.getUsers()
-
-		per = PersistentImage()
-		per.userId = users[0].id
-		per.name = 'sleepy-PersistentImage'
-		self.connection.client.addPersistentImage(per)
-		pers = self.connection.client.getPersistentImages()
-
-		inst = InstanceConfiguration()
-		inst.name = 'sleepy-inst'
-		inst.memory = 512
-		inst.cores = 1
-		self.connection.client.addInstanceConfiguration(inst)
-		insts = self.connection.client.getInstanceConfigurations()
-
-		hdc = HardDiskConfiguration()
-		hdc.index = 0
-		hdc.persistentImageId = pers[0].id
-		hdc.persistent = False
-		hdc.instanceConfigurationId = insts[0].id
-
-#	 def testCreateDestroyShutdown(self):
-#		 '''test creating/destroying/shutting down VMs
-
-#		 not implemented'''
-#		 self.fail('test not implemented')
-#	 def testSuspendResume(self):
-#		 '''test suspending/resuming VMs
-
-#		 not implemented'''
-#		 self.fail('test not implemented')
-#	 def testMigrate(self):
-#		 '''test migration
-
-#		 not implemented'''
-#		 self.fail('test not implemented')
-#	 def testPauseUnpause(self):
-#		 '''test pausing/unpausing VMs
-
-#		 not implemented'''
-#		 self.fail('test not implemented')
-
-
-##############################
-# Test Code
-##############################
-if __name__ == '__main__':
-	logging.basicConfig(level=logging.NOTSET,
-						format="%(asctime)s %(levelname)s:\t %(message)s",
-						stream=sys.stdout)
-
-	suite = unittest.TestLoader().loadTestsFromTestCase(TestClient)
-	unittest.TextTestRunner(verbosity=2).run(suite)
-
diff --git a/src/tashi/messaging/messageBroker.py b/src/tashi/messaging/messageBroker.py
deleted file mode 100644
index c21b57a..0000000
--- a/src/tashi/messaging/messageBroker.py
+++ /dev/null
@@ -1,49 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-# 
-#   http://www.apache.org/licenses/LICENSE-2.0
-# 
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied.  See the License for the
-# specific language governing permissions and limitations
-# under the License.    
-
-import ConfigParser
-import getopt
-
-import os
-import sys
-import time
-
-import thriftmessaging
-
-options = []
-long_options = ['port=']
-
-# FIXME: should initialize from config file
-params = {"port":1717}
-
-try:
-	optlist, args = getopt.getopt(sys.argv[1:], options, long_options)
-except getopt.GetoptError, err:
-	print str(err)
-	sys.exit(2)
-
-for opt in optlist:
-	if opt[0] == "--port":
-		try:
-			params["port"] = int(opt[1])
-		except:
-			print "--port expects an integer, got %s" % opt[1]
-			sys.exit(0)
-
-print "Starting message broker on port %i" % params["port"]
-broker = thriftmessaging.MessageBrokerThrift(params["port"], daemon=False)
-
diff --git a/src/tashi/messaging/messaging.py b/src/tashi/messaging/messaging.py
deleted file mode 100644
index c421d5c..0000000
--- a/src/tashi/messaging/messaging.py
+++ /dev/null
@@ -1,337 +0,0 @@
-#!/usr/bin/python
-
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-# 
-#   http://www.apache.org/licenses/LICENSE-2.0
-# 
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied.  See the License for the
-# specific language governing permissions and limitations
-# under the License.    
-
-import threading
-import thread
-import sys
-import os
-import socket
-import Queue
-import copy
-import random
-import traceback
-
-from threadpool import ThreadPoolClass, threadpool, ThreadPool
-from threadpool import threadpoolmethod, threaded, synchronized, synchronizedmethod
-
-class RWLock(object):
-	"""RWLock: Simple reader/writer lock implementation
-	FIXME: this implementation will starve writers!
-	Methods:
-		acquire() : take lock for read access
-		release() : release lock from read access
-		acquireWrite() : take lock for write access
-		releaseWrite() : release lock from write access"""
-	def __init__(self):
-		self.lock = threading.Condition()
-		self.readers = 0
-	def acquire(self):
-		self.lock.acquire()
-		self.readers = self.readers + 1
-		self.lock.release()
-	def release(self):
-		self.lock.acquire()
-		self.readers = self.readers - 1
-		self.lock.notify()
-		self.lock.release()
-	def acquireWrite(self):
-		self.lock.acquire()
-		while self.readers > 0:
-			self.lock.wait()
-	def releaseWrite(self):
-		self.lock.notify()
-		self.lock.release()
-
-
-
-class MessageBroker(object):
-	def __init__(self):
-		self.sublock = RWLock()
-		self.subscribers = []
-		self.random = random.Random()
-	def log(self, msg):
-		print "MessageBroker: Got log: '%s'" % str(msg)
-		return msg
-	def addSubscriber(self, subscriber):
-		self.sublock.acquireWrite()
-		self.subscribers.append(subscriber)
-		l = len(self.subscribers)
-		self.sublock.releaseWrite()
-		return l
-	def publish(self, message):
-		removesubs = []
-		i = self.random.randint(0,100)
-
-#		 subscribers = self.getSubscribers()
-#		 random.shuffle(subscribers)
-
-		self.sublock.acquire()
-
-		sys.stdout.flush()
-
-		for subscriber in self.subscribers:
-			try:
-				sys.stdout.flush()
-				assert(subscriber != self)
-				subscriber.publish(message)
-				sys.stdout.flush()
-			except Exception, e:
-				print e
-				removesubs.append(subscriber)
-
-		self.sublock.release()
-
-		if len(removesubs) > 0:
-			print "detected %i failed subscribers" % len(removesubs)
-			sys.stdout.flush()
-			self.sublock.acquireWrite()
-			for subscriber in removesubs:
-				try:
-					self.subscribers.remove(subscriber)
-				except:
-					pass
-			self.sublock.releaseWrite()
-	def getSubscribers(self):
-		self.sublock.acquire()
-		subs = copy.copy(self.subscribers)
-		self.sublock.release()
-		return subs
-	def removeSubscriber(self, subscriber):
-		self.sublock.acquireWrite()
-		try:
-			self.subscribers.remove(subscriber)
-		except:
-			pass
-		self.sublock.releaseWrite()
-	def publishList(self, messages):
-		for message in messages:
-			self.publish(message)
-
-class Subscriber(object):
-	def __init__(self, broker, pmatch={}, nmatch={}, synchronized=False):
-		self.broker = broker
-		self.lock = threading.Lock()
-		self.synchronized = synchronized
-		self.pmatch={}
-		self.nmatch={}
-		broker.addSubscriber(self)
-	def publish(self, message):
-		sys.stdout.flush()
-		msg = message
-		try:
-			if self.synchronized:
-				self.lock.acquire()
-			msg = self.filter(msg)
-			if (msg != None):
-				self.handle(msg)
-			if self.synchronized:
-				self.lock.release()
-		except Exception, x:
-			if self.synchronized:
-				self.lock.release()
-			print '%s, %s, %s' % (type(x), x, traceback.format_exc())
-	def publishList(self, messages):
-		for message in messages:
-			self.publish(message)
-	def handle(self, message):
-		print "Subscriber Default Handler: '%s'" % message
-	def setMatch(self, pmatch={}, nmatch={}):
-		self.lock.acquire()
-		self.pmatch=pmatch
-		self.nmatch=nmatch
-		self.lock.release()
-	def filter(self, message):
-		"""filter(self, message) : the filter function returns
-		the message, modified to be passed to the handler.
-		Returning (None) indicates that this is not a message
-		we are interested in, and it will not be passed to the
-		handler."""
-		send = True
-		for key in self.pmatch.keys():
-			if (not message.has_key(key)):
-				send = False
-				break
-			if self.pmatch[key] != None:
-				if message[key] != self.pmatch[key]:
-					send = False
-					break
-		if send == False:
-			return None
-		for key in message.keys():
-			if self.nmatch.has_key(key):
-				if self.nmatch[key] == None:
-					send = False
-					break
-				if self.nmatch[key] == message[key]:
-					send = False
-					break
-		if send == False:
-			return None
-		return message
-
-
-	
-class Publisher(object):
-	'''Superclass for pub/sub publishers
-
-	FIXME: use finer-grained locking'''
-	def __init__(self, broker, aggregate=100):
-		self.pending = []
-		self.pendingLock = threading.Lock()
-		self.aggregateSize = aggregate
-		self.broker = broker
-	@synchronizedmethod
-	def publish(self, message):
-		if message.has_key('aggregate') and message['aggregate'] == 'True':
-			self.aggregate(message)
-			return
-		else:
-			self.broker.publish(message)
-	@synchronizedmethod
-	def publishList(self, messages):
-		self.broker.publishList(messages)
-	@synchronizedmethod
-	def aggregate(self, message):
-		# we can make this lock-less by using a queue for pending
-		# messages
-		self.pendingLock.acquire()
-		self.pending.append(message)
-		if len(self.pending) >= self.aggregateSize:
-			self.broker.publishList(self.pending)
-			self.pending = []
-		self.pendingLock.release()
-	@synchronizedmethod
-	def setBroker(self, broker):
-		self.broker = broker
-
-##############################
-# Testing Code
-##############################
-import time
-import unittest
-import sys
-import logging
-
-		
-class TestSubscriber(Subscriber):
-	def __init__(self, *args, **kwargs):
-		self.queue = Queue.Queue()
-		Subscriber.__init__(self, *args, **kwargs)
-	def handle(self, message):
-		self.queue.put(message)
-
-class TestMessaging(unittest.TestCase):
-	def setUp(self):
-		self.broker = MessageBroker()
-		self.publisher = Publisher(self.broker)
-		self.subscriber = TestSubscriber(self.broker)
-	def testPublish(self):
-		self.publisher.publish( {'message':'hello world'} )
-		self.assertEqual(self.subscriber.queue.qsize(), 1)
-	def testPublishList(self):
-		nrmsgs = 10
-		msgs = []
-		for i in range(nrmsgs):
-			msgs.append( {'msgnum':str(i)} )
-		self.publisher.publishList( msgs )
-		self.assertEqual(self.subscriber.queue.qsize(), nrmsgs)
-	def testAggregate(self):
-		nrmsgs = self.publisher.aggregateSize
-		for i in range(nrmsgs):
-			self.assertEqual(self.subscriber.queue.qsize(), 0)
-			self.publisher.aggregate( {'msgnum':str(i)} )
-		self.assertEqual(self.subscriber.queue.qsize(), nrmsgs)
-	def testAggregateKeyword(self):
-		nrmsgs = self.publisher.aggregateSize
-		for i in range(nrmsgs):
-			self.assertEqual(self.subscriber.queue.qsize(), 0)
-			self.publisher.publish( {'msgnum':str(i), 'aggregate':'True'} )
-		self.assertEqual(self.subscriber.queue.qsize(), nrmsgs)
-
-if __name__ == '__main__':
-
-	logging.basicConfig(level=logging.INFO,
-						format="%(asctime)s %(levelname)s:\t %(message)s",
-						stream=sys.stdout)
-
-	suite = unittest.TestLoader().loadTestsFromTestCase(TestMessaging)
-	unittest.TextTestRunner(verbosity=2).run(suite) 
-
-	sys.exit(0)
-
-
-##############################
-# Old/Unused testing code
-##############################
-
-
-
-	print 'testing removeSubscriber'
-	broker.removeSubscriber(subscriber)
-	publisher.publish( {'message':"you shouldn't see this"} )
-
-	nsub = NullSubscriber(broker)
-	print 'timing publish'
-	nrmsg = 100000
-	tt = time.time()
-	for i in range(nrmsg):
-#		publisher.publish( {"message":"hello world!"} )
-		publisher.publish( {} )
-	tt = time.time() - tt
-	print "Published %i messages in %f seconds, %f msg/s"%(nrmsg,
-														   tt,
-														   nrmsg/tt)
-	broker.removeSubscriber(nsub)
-
-	class SlowSubscriber(Subscriber):
-		def handle(self, message):
-			print 'called slow subscriber with message', message
-			time.sleep(1)
-			print 'returning from slow subscriber with message', message
-	class ThreadedSubscriber(Subscriber):
-		@threaded
-		def handle(self, message):
-			print 'called threaded subscriber with message', message
-			time.sleep(1)
-			print 'returning from threaded subscriber with message', message
-	class ThreadPoolSubscriber(Subscriber, ThreadPoolClass):
-		@threadpoolmethod
-		def handle(self, message):
-			print 'called threadpool subscriber with message', message
-			time.sleep(1)
-			print 'returning from threadpool subscriber with message', message
-
-
-
-	tsub = ThreadedSubscriber(broker)
-	for i in range(8):
-		publisher.publish( {"msg":str(i)} )
-	broker.removeSubscriber(tsub)
-	time.sleep(3)
-
-	tpsub = ThreadPoolSubscriber(broker)
-	for i in range(8):
-		publisher.publish( {"msg":str(i)} )
-	broker.removeSubscriber(tpsub)
-	time.sleep(3)
-
-	ssub = SlowSubscriber(broker)
-	for i in range(4):
-		publisher.publish( {"msg":str(i)} )
-	broker.removeSubscriber(ssub)
diff --git a/src/tashi/messaging/soapmessaging.py b/src/tashi/messaging/soapmessaging.py
deleted file mode 100755
index be35fc9..0000000
--- a/src/tashi/messaging/soapmessaging.py
+++ /dev/null
@@ -1,229 +0,0 @@
-#! /usr/bin/env python
-
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-# 
-#   http://www.apache.org/licenses/LICENSE-2.0
-# 
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied.  See the License for the
-# specific language governing permissions and limitations
-# under the License.    
-
-from messaging import *
-
-import cPickle
-import soaplib.wsgi_soap
-import cherrypy.wsgiserver
-from soaplib.service import soapmethod
-from soaplib.serializers.primitive import *
-import SOAPpy.WSDL
-import time
-
-class MessageBrokerSoap(soaplib.wsgi_soap.SimpleWSGISoapApp, MessageBroker):
-	def __init__(self, port):
-		soaplib.wsgi_soap.SimpleWSGISoapApp.__init__(self)
-		MessageBroker.__init__(self)
-		self.port = port
-		def trdfn():
-			service = self
-			server = cherrypy.wsgiserver.CherryPyWSGIServer(("0.0.0.0",port), service)
-			server.start()
-		threading.Thread(target=trdfn).start()
-
-
-	@soapmethod(Array(String), Array(String), _returns=Null)
-	def log(self, keys, values):
-		message = {}
-		if len(keys) != len(values):
-			raise Exception, "Different lengths for keys and values"
-		for i in range(len(keys)):
-			message[keys[i]] = values[i]
-		MessageBroker.log(self, message)
-
-	@soapmethod(String, Integer, _returns=Null)
-	def addSubscriber(self, host, port):
-		subscriber = SubscriberSoapProxy(host, port)
-		MessageBroker.addSubscriber(self, subscriber)
-	
-	@soapmethod(String, Integer, _returns=Null)
-	def removeSubscriber(self, host, port):
-		# should this method really be able to peek into subscriber.host/port 
-		subscriber = None
-		subscribers = self.getSubscribers()
-		for subscriber in subscribers:
-			if subscriber.host == host and subscriber.port == port:
-				subscriber = subscriber
-		if subscriber != None:
-			MessageBroker.removeSubscriber(self, subscriber)
-		
-
-	@soapmethod(Array(String), Array(String), _returns=Null)
-	def publish(self, keys, values):
-		message = {}
-		if len(keys) != len(values):
-			raise Exception, "Different lengths for keys and values"
-		for i in range(len(keys)):
-			message[keys[i]] = values[i]
-		MessageBroker.publish(self, message)
-
-
-
-class MessageBrokerSoapProxy(object):
-	def __init__(self, host, port):
-		self.host = host
-		self.port = port
-		self.connection = SOAPpy.WSDL.Proxy("http://%s:%i/.wsdl"%(host,port))
-	def log(self, message):
-		keys = []
-		values = []
-		for k,v in message.items():
-			keys.append(k)
-			values.append(v)
-		self.connection.log(keys=keys, values=values)
-	def addSubscriber(self, subscriber):
-		self.connection.addSubscriber(host=subscriber.host, port=subscriber.port)
-	def publish(self, message):
-		keys = []
-		values = []
-		for k,v in message.items():
-			keys.append(k)
-			values.append(v)
-		self.connection.publish(keys=keys, values=values)
-	def removeSubscriber(self, subscriber):
-		self.connection.removeSubscriber(host=subscriber.host, port=subscriber.port)
-
-
-
-
-class SubscriberSoap(soaplib.wsgi_soap.SimpleWSGISoapApp, Subscriber):
-	def __init__(self, broker, port, synchronized=False):
-		soaplib.wsgi_soap.SimpleWSGISoapApp.__init__(self)
-		Subscriber.__init__(self, synchronized=synchronized)
-		self.host = socket.gethostname()
-		self.port = port
-		self.broker = broker
-		self.server = None
-		def trdfn():
-			service = self
-			self.server = cherrypy.wsgiserver.CherryPyWSGIServer(("0.0.0.0",port), service)
-			self.server.start()
-		threading.Thread(target=trdfn).start()
-#		broker.log("Subscriber started")
-		broker.addSubscriber(self)
-	@soapmethod(Array(String), Array(String), _returns=Integer)
-	def publish(self, keys, values):
-		message = {}
-		if len(keys) != len(values):
-			raise Exception, "Different lengths for keys and values"
-		for i in range(len(keys)):
-			message[keys[i]] = values[i]
-		Subscriber.publish(self, message)
-		return 0
-	def stop(self):
-		self.server.stop()
-
-class SubscriberSoapProxy(object):
-	def __init__(self, host, port):
-		self.host = host
-		self.port = port
-		self.connection = SOAPpy.WSDL.Proxy("http://%s:%i/.wsdl"%(host,port))
-	def publish(self, message):
-		keys = []
-		values = []
-		for k,v in message.items():
-			keys.append(k)
-			values.append(v)
-		self.connection.publish(keys=keys, values=values)
-
-
-####################
-# Testing Code 
-####################
-
-class CustomSubscriber(SubscriberSoap):
-	def handle(self, message):
-		print "Custom Subscriber: '%s'" % str(message)
-
-class NullSubscriber(SubscriberSoap):
-	def handle(self, message):
-		pass
-
-
-if __name__ == '__main__':
-	try:
-		portnum = 1717
-
-		print "\ntesting message broker"
-		broker = MessageBrokerSoap(portnum)
-		proxy = MessageBrokerSoapProxy("localhost", portnum)
-		portnum = portnum + 1 
-
-		print "\ntesting log function"
-		proxy.log( {"message":"Hello World!"} )
-#		proxy.log("It looks like log works")
-
-		print "\ntesting subscriber proxy"
-		subscriber = SubscriberSoap(proxy, portnum)
-		portnum = portnum + 1
-
-		print "\ntesting custom subscriber"
-		csub = CustomSubscriber(proxy, portnum)
-		portnum = portnum + 1
-
-		print "\ntesting publish"
-		proxy.publish( {"message":"Hello World!"} )
-
-		print "\ntesting stop"
-		subscriber.stop()
-		proxy.publish( {"message":"Everybody here?"} )
-
-		print "\ntesting removeSubscriber"
-		proxy.removeSubscriber(csub)
-		proxy.publish( {"message":"Nobody home"} )
-		proxy.addSubscriber(csub)
-		proxy.publish( {"message":"You're back!"} )
-
-		print "\ntesting filter"
-		csub.setMatch( {"print":"yes"} )
-		proxy.publish( {"print":"yes", "message":"this should be printed"} )
-		proxy.publish( {"print":"no", "message":"this should NOT be printed"} )
-		csub.setMatch()
-
-		print "\ntesting publish performance"
-		proxy.removeSubscriber(csub)
-		nrmsg = 10000
-		tt = time.time()
-		for i in range(nrmsg):
-			proxy.publish( {"message":"msg %i"%i} )
-		tt = time.time() - tt
-		print "Published %i messages in %f seconds, %f msg/s"%(nrmsg,
-															   tt,
-															   nrmsg/tt)
-
-		print "\ntesting publish/subscribe performance"
-		nsub = NullSubscriber(proxy, portnum)
-		portnum = portnum + 1
-		nrmsg = 10000
-		tt = time.time()
-		for i in range(nrmsg):
-			proxy.publish( {"message":"msg %i"%i} )
-		tt = time.time() - tt
-		print "Published %i messages in %f seconds, %f msg/s"%(nrmsg,
-															   tt,
-															   nrmsg/tt)
-
-																   
-
-	except Exception, e:
-#		raise e
-		print  e
-		sys.exit(0)
-	sys.exit(0)
diff --git a/src/tashi/messaging/tashimessaging.py b/src/tashi/messaging/tashimessaging.py
deleted file mode 100644
index 006400f..0000000
--- a/src/tashi/messaging/tashimessaging.py
+++ /dev/null
@@ -1,148 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-# 
-#   http://www.apache.org/licenses/LICENSE-2.0
-# 
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied.  See the License for the
-# specific language governing permissions and limitations
-# under the License.    
-
-from thriftmessaging import *
-import logging
-import Queue
-from ConfigParser import ConfigParser
-import time
-import socket
-import signal
-
-class TashiLogHandler(logging.Handler, PublisherThrift):
-	def __init__(self, config, *args, **kwargs):
-		self.messages = Queue.Queue()
-		self.config = config
-		logging.Handler.__init__(self, *args, **kwargs)
-		PublisherThrift.__init__(self, 
-								 config.get('MessageBroker', 'host'),
-								 int(config.get('MessageBroker', 'port')))
-	def emit(self, record):
-		# 'args', 'created', 'exc_info', 'exc_text', 'filename',
-		# 'funcName', 'getMessage', 'levelname', 'levelno', 'lineno',
-		# 'module', 'msecs', 'msg', 'name', 'pathname', 'process',
-		# 'relativeCreated', 'thread', 'threadName']
-		msg = {}
-		# args
-		# created
-		# exc_info
-		# exc_text
-		msg['log-filename'] = str(record.filename)
-		msg['log-funcname'] = str(record.funcName)
-		msg['log-levelname'] = str(record.levelname)
-		msg['log-level'] = str(record.levelno)
-		msg['log-lineno'] = str(record.lineno)
-		msg['log-module'] = str(record.module)
-		msg['log-msecs'] = str(record.msecs)
-		msg['log-message'] = str(record.msg)
-		msg['log-name'] = str(record.name)
-		msg['log-pathname'] = str(record.pathname)
-		msg['log-process'] = str(record.process)
-		# relativeCreated
-		msg['log-thread'] = str(record.thread)
-		msg['log-threadname'] = str(record.threadName)
-
-		# standard message fields
-		msg['timestamp'] = str(time.time())
-		msg['hostname'] = socket.gethostname()
-		msg['message-type'] = 'log'
-
-		self.messages.put(msg)
-		self.publish(msg)
-
-class TashiSubscriber(SubscriberThrift):
-	def __init__(self, config, port, **kwargs):
-		sys.stdout.flush()
-		brokerPort = int(config.get('MessageBroker', 'port'))
-		self.broker = MessageBrokerThriftProxy(config.get('MessageBroker', 'host'), brokerPort)
-		SubscriberThrift.__init__(self, self.broker, port, **kwargs)
-
-		
-
-##############################
-# Test Code
-##############################
-import unittest
-import sys
-
-class TestTashiSubscriber(TashiSubscriber):
-	def __init__(self, *args, **kwargs):
-		self.messageQueue = Queue.Queue()
-		TashiSubscriber.__init__(self, *args, **kwargs)
-	def handle(self, message):
-		self.messageQueue.put(message)
-
-
-def incrementor(start = 0):
-	while True:
-		a = start
-		start = start + 1
-		yield a
-increment = incrementor()
-
-class TestTashiMessaging(unittest.TestCase):
-	def setUp(self):
-		self.configFiles = [ '../../../etc/TestConfig.cfg']
-		self.config = ConfigParser()
-		self.configFiles = self.config.read(self.configFiles)
-		self.port = int(self.config.get('MessageBroker', 'port'))
-
-		try:
-			self.brokerPid = os.spawnlpe(os.P_NOWAIT, 'python', 'python', 
-										 './messageBroker.py', 
-										 '--port', str(self.port),
-										 os.environ)
-			self.port = self.port + 1
-			# FIXME: what's the best way to wait for the broker to be ready?
-			time.sleep(1)
-		except Exception, e:
-			sys.exit(0)
-		self.initialized = True
-		self.log = logging.getLogger('TestTashiMessaging')
-		self.handler = TashiLogHandler(self.config)
-		self.log.addHandler(self.handler)
-		self.sub = TestTashiSubscriber(self.config, int(self.port) + increment.next())
-	def tearDown(self):
-		os.kill(self.brokerPid, signal.SIGKILL)
-		# FIXME: wait for the port to be ready again
-		time.sleep(2)
-		self.log.removeHandler(self.handler)
-#		 self.sub.broker.removeSubscriber(self.sub)
-		pass
-	def testLog(self):
-		self.log.log(50, "Hello World!")
-		self.handler.messages.get(timeout=5)
-		self.sub.messageQueue.get(timeout=5)
-		self.assertEqual(self.handler.messages.qsize(), 0)
-		self.assertEqual(self.sub.messageQueue.qsize(), 0)
-	def testPublish(self):
-		sys.stdout.flush()
-		self.port = self.port + 1
-		self.handler.publish({'message':'hello world'})
-		self.sub.messageQueue.get(timeout=5)
-		self.assertEqual(self.sub.messageQueue.qsize(), 0)
-		
-
-if __name__=='__main__':
-
-
-#	 logging.basicConfig(level=logging.INFO,
-#						 format="%(asctime)s %(levelname)s:\t %(message)s",
-#						 stream=sys.stdout)
-
-	suite = unittest.TestLoader().loadTestsFromTestCase(TestTashiMessaging)
-	unittest.TextTestRunner(verbosity=2).run(suite)
diff --git a/src/tashi/messaging/threadpool.py b/src/tashi/messaging/threadpool.py
deleted file mode 100644
index 5684ef2..0000000
--- a/src/tashi/messaging/threadpool.py
+++ /dev/null
@@ -1,305 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-# 
-#   http://www.apache.org/licenses/LICENSE-2.0
-# 
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied.  See the License for the
-# specific language governing permissions and limitations
-# under the License.    
-
-import threading
-import time
-import Queue
-import logging
-
-_log = logging.getLogger('tashi.messaging.threadpool')
-
-def threaded(func):
-	def fn(*args, **kwargs):
-		thread = threading.Thread(target=func, args=args, kwargs=kwargs)
-		thread.start()
-		return thread
-	return fn
-
-
-class ThreadPool(Queue.Queue):
-	def __init__(self, size=8, maxsize=0):
-		Queue.Queue.__init__(self, maxsize)
-		for i in range(size):
-			thread = threading.Thread(target=self._worker)
-			thread.setDaemon(True)
-			thread.start()
-	def _worker(self):
-		while True:
-			try:
-				func, args, kwargs = self.get()
-				func(*args, **kwargs)
-			except Exception, e:
-				_log.error(e)
-				# FIXME: do something smarter here, backtrace, log,
-				# allow user-defined error handling...
-				
-	def submit(self, func, *args, **kwargs):
-		self.put((func, args, kwargs))
-	def submitlist(self, func, args, kwargs):
-		self.put((func, args, kwargs))
-
-class ThreadPoolClass:
-	def __init__(self, size=8, maxsize=0):
-		self._threadpool_pool = ThreadPool(size=size, maxsize=maxsize)
-
-
-def threadpool(pool):
-	def dec(func):
-		def fn(*args, **kwargs):
-			pool.submit(func, *args, **kwargs)
-		return fn
-	return dec
-
-def threadpoolmethod(meth):
-	def fn(*args, **kwargs):
-		try:
-			pool = args[0]._threadpool_pool
-		except AttributeError:
-			pool = args[0].__dict__.setdefault('_threadpool_pool', ThreadPool())
-		# FIXME: how do we check parent class?
-#		assert args[0].__class__ == ThreadPoolClass, "Thread pool method must be in a ThreadPoolClass"
-		pool.submit(meth, *args, **kwargs)
-	return fn
-
-def synchronized(lock=None):
-	_log.debug('synchronized decorator factory called')
-	if lock==None:
-		lock = threading.RLock()
-	def dec(func):
-		_log.debug('synchronized decorator called')
-		def fn(*args, **kwargs):
-			_log.debug('getting sync lock')
-			lock.acquire()
-			_log.debug('got sync lock')
-			ex = None
-			try:
-				r = func(*args, **kwargs)
-			except Exception, e:
-				ex = e
-			_log.debug('releasing sync lock')
-			lock.release()
-			_log.debug('released sync lock')
-			if ex != None:
-				raise e
-			return r
-		return fn
-	return dec
-			
-def synchronizedmethod(func):
-	def fn(*args, **kwargs):
-		try:
-			lock = args[0]._synchronized_lock
-		except AttributeError:
-			lock = args[0].__dict__.setdefault('_synchronized_lock', threading.RLock())
-		lock.acquire()
-		ex = None
-		try:
-			func(*args, **kwargs)
-		except Exception, e:
-			ex = e
-		lock.release()
-		if ex != None:
-			raise e
-	return fn
-		
-
-##############################
-# Test Code
-##############################
-import unittest
-import sys
-import time
-
-class TestThreadPool(unittest.TestCase):
-	def setUp(self):
-		self.errmargin = 0.5
-
-	def testUnthreaded(self):
-		queue = Queue.Queue()
-		def slowfunc(sleep=1):
-			time.sleep(sleep)
-			queue.put(None)
-		tt = time.time()
-		for i in range(4):
-			slowfunc()
-		for i in range(4):
-			queue.get()
-		tt = time.time() - tt
-		self.assertAlmostEqual(tt, 4, 1) 
-
-	def testThreaded(self):
-		queue = Queue.Queue()
-		@threaded
-		def slowthreadfunc(sleep=1):
-			time.sleep(sleep)
-			queue.put(None)
-		tt = time.time()
-		for i in range(8):
-			slowthreadfunc()
-		for i in range(8):
-			queue.get()
-		tt = time.time() - tt
-		self.assertAlmostEqual(tt, 1, 1) 
-
-	def testThreadPool(self):
-		pool = ThreadPool(size=4)
-		queue = Queue.Queue()
-		@threadpool(pool)
-		def slowpoolfunc(sleep=1):
-			time.sleep(sleep)
-			queue.put(None)
-		tt = time.time()
-		for i in range(8):
-			slowpoolfunc()
-		for i in range(8):
-			queue.get()
-		tt = time.time() - tt
-		self.assertAlmostEqual(tt, 2, 1) 
-
-	def testUnthreadedMethod(self):
-		queue = Queue.Queue()
-		class slowclass:
-			def __init__(self, sleep=1):
-				self.sleep=sleep
-			def beslow(self):
-				time.sleep(self.sleep)
-				queue.put(None)
-		sc = slowclass()
-		tt = time.time()
-		for i in range(4):
-			sc.beslow()
-		for i in range(4):
-			queue.get()
-		tt = time.time() - tt
-		self.assertAlmostEqual(tt, 4, 1)
-	
-	def testThreadedMethod(self):
-		queue = Queue.Queue()
-		class slowclass:
-			def __init__(self, sleep=1):
-				self.sleep=sleep
-			@threaded
-			def beslow(self):
-				time.sleep(self.sleep)
-				queue.put(None)
-		sc = slowclass()
-		tt = time.time()
-		for i in range(4):
-			sc.beslow()
-		for i in range(4):
-			queue.get()
-		tt = time.time() - tt
-		self.assertAlmostEqual(tt, 1, 1)
-	
-	def testThreadPoolMethod(self):
-		queue = Queue.Queue()
-		class slowclass:
-			def __init__(self, sleep=1):
-				self.sleep=sleep
-			@threadpoolmethod
-			def beslow(self):
-				time.sleep(self.sleep)
-				queue.put(None)
-		sc = slowclass()
-		tt = time.time()
-		for i in range(16):
-			sc.beslow()
-		for i in range(16):
-			queue.get()
-		tt = time.time() - tt
-		self.assertAlmostEqual(tt, 2, 1)
-	
-	def testSynchronized(self):
-		queue = Queue.Queue()
-		@synchronized()
-		def addtoqueue():
-			time.sleep(1)
-			queue.put(None)
-		@threaded
-		def slowthreadfunc():
-			addtoqueue()
-		tt = time.time()
-		for i in range(4):
-			slowthreadfunc()
-		for i in range(4):
-			queue.get()
-		tt = time.time() - tt
-		self.assertAlmostEqual(tt, 4, 1) 
-
-	def testSynchronizedMethod(self):
-		queue = Queue.Queue()
-		class addtoqueue:
-			@synchronizedmethod
-			def addtoqueue1(self):
-				time.sleep(1)
-				queue.put(None)
-			@synchronizedmethod
-			def addtoqueue2(self):
-				time.sleep(1)
-				queue.put(None)
-		atc = addtoqueue()
-		@threaded
-		def slowthreadfunc1():
-			atc.addtoqueue1()
-		@threaded
-		def slowthreadfunc2():
-			atc.addtoqueue2()
-		tt = time.time()
-		for i in range(4):
-			slowthreadfunc1()
-			slowthreadfunc2()
-		for i in range(8):
-			queue.get()
-		tt = time.time() - tt
-		self.assertAlmostEqual(tt, 8, 1) 
-
-	def testUnsynchronizedMethod(self):
-		queue = Queue.Queue()
-		class addtoqueue:
-			def addtoqueue1(self):
-				time.sleep(1)
-				queue.put(None)
-			def addtoqueue2(self):
-				time.sleep(1)
-				queue.put(None)
-		atc = addtoqueue()
-		@threaded
-		def slowthreadfunc1():
-			atc.addtoqueue1()
-		@threaded
-		def slowthreadfunc2():
-			atc.addtoqueue2()
-		tt = time.time()
-		for i in range(4):
-			slowthreadfunc1()
-			slowthreadfunc2()
-		for i in range(8):
-			queue.get()
-		tt = time.time() - tt
-		self.assertAlmostEqual(tt, 1, 1) 
-
-
-
-if __name__=='__main__':
-	import sys
-
-	logging.basicConfig(level=logging.INFO,
-						format="%(asctime)s %(levelname)s:\t %(message)s",
-						stream=sys.stdout)
-
-	suite = unittest.TestLoader().loadTestsFromTestCase(TestThreadPool)
-	unittest.TextTestRunner(verbosity=2).run(suite)
diff --git a/src/tashi/messaging/thriftmessaging.py b/src/tashi/messaging/thriftmessaging.py
deleted file mode 100755
index 0c73ff0..0000000
--- a/src/tashi/messaging/thriftmessaging.py
+++ /dev/null
@@ -1,278 +0,0 @@
-#!/usr/bin/env python
-
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-# 
-#   http://www.apache.org/licenses/LICENSE-2.0
-# 
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied.  See the License for the
-# specific language governing permissions and limitations
-# under the License.    
-
-import sys
-import time
-import socket
-import traceback
-import threading
-
-sys.path.append('./gen-py')
-import tashi.messaging.messagingthrift
-import tashi.messaging.messagingthrift.MessageBrokerThrift
-import tashi.messaging.messagingthrift.SubscriberThrift
-from tashi.messaging.messagingthrift.ttypes import *
-
-from thrift import Thrift
-from thrift.transport import TSocket
-from thrift.transport import TTransport
-from thrift.protocol import TBinaryProtocol
-from thrift.server import TServer
-
-from tashi import ConnectionManager
-
-from tashi.messaging.messaging import *
-from tashi.messaging.threadpool import ThreadPoolClass, threadpool, ThreadPool, threadpoolmethod, threaded
-
-class MessageBrokerThrift(MessageBroker):
-	def __init__(self, port, daemon=True):
-		MessageBroker.__init__(self)
-		self.processor = tashi.messaging.messagingthrift.MessageBrokerThrift.Processor(self)
-		self.transport = TSocket.TServerSocket(port)
-		self.tfactory = TTransport.TBufferedTransportFactory()
-		self.pfactory = TBinaryProtocol.TBinaryProtocolFactory()
-		self.proxy = ConnectionManager(tashi.messaging.messagingthrift.SubscriberThrift.Client, 0)
-		self.ready = threading.Event()
-#		 self.server = TServer.TSimpleServer(self.processor,
-#											 self.transport,
-#											 self.tfactory,
-#											 self.pfactory)
-#		 self.server = TServer.TThreadPoolServer(self.processor,
-#												 self.transport,
-#												 self.tfactory,
-#												 self.pfactory)
-		self.server = TServer.TThreadedServer(self.processor,
-												self.transport,
-												self.tfactory,
-												self.pfactory)
-		self.publishCalls = 0
-
-		def ssvrthrd():
-			try:
-				# FIXME: Race condition, the ready event should be set after
-				# starting the server.  However, server.serve()
-				# doesn't return under normal circumstances.  This
-				# seems to work in practice, even though it's clearly
-				# wrong.
-				self.ready.set()
-				self.server.serve()
-			except Exception, e:
-				print e
-				sys.stdout.flush()
-				pass
-		svt = threading.Thread(target=ssvrthrd)
-		svt.setDaemon(daemon)
-		svt.start()
-		self.ready.wait()
-	def log(self, message):
-		MessageBroker.log(self, message)
-	@synchronizedmethod
-	def addSubscriber(self, host, port):
-		subscribers = self.getSubscribers()
-		for sub in subscribers:
-			if sub.host == host and sub.port == port:
-				return
-		subscriber = SubscriberThriftProxy(host, port, self.proxy)
-		MessageBroker.addSubscriber(self, subscriber)
-	def removeSubscriber(self, host, port):
-		subscriber = None
-		subscribers = self.getSubscribers()
-		for sub in subscribers:
-			if sub.host == host and sub.port == port:
-				subscriber = sub
-		if subscriber != None:
-			MessageBroker.removeSubscriber(self, subscriber)
-	@synchronizedmethod
-	def publish(self, message):
-		self.publishCalls  = self.publishCalls + 1
-		sys.stdout.flush()
-		MessageBroker.publish(self, message)
-
-class MessageBrokerThriftProxy:
-	def __init__(self, host, port):
-		self.host = host
-		self.port = port
-		self.proxy = ConnectionManager(tashi.messaging.messagingthrift.MessageBrokerThrift.Client,port)
-	@synchronizedmethod
-	def log(self, message):
-		self.proxy[self.host, self.port].log(message)
-	@synchronizedmethod
-	def publish(self, message):
-		self.proxy[self.host, self.port].publish(message)
-	@synchronizedmethod
-	def publishList(self, messages):
-		self.proxy[self.host, self.port].publishList(messages)
-	@synchronizedmethod
-	def addSubscriber(self, subscriber):
-		self.proxy[self.host, self.port].addSubscriber(host=subscriber.host, port=subscriber.port)
-	@synchronizedmethod
-	def removeSubscriber(self, subscriber):
-		self.proxy[self.host, self.port].removeSubscriber(host=subscriber.host, port=subscriber.port)
-
-
-
-class SubscriberThrift(Subscriber, threading.Thread):
-	def __init__(self, broker, port, synchronized=False):
-		self.host = socket.gethostname()
-		self.port = port
-		self.processor = tashi.messaging.messagingthrift.SubscriberThrift.Processor(self)
-		self.transport = TSocket.TServerSocket(port)
-		self.tfactory = TTransport.TBufferedTransportFactory()
-		self.pfactory = TBinaryProtocol.TBinaryProtocolFactory()
-		self.server = TServer.TThreadedServer(self.processor,
-											  self.transport,
-											  self.tfactory,
-											  self.pfactory)
-		def ssvrthrd():
-			try:
-				self.server.serve()
-			except Exception, e:
-				pass
-
-
-		self.thread = threading.Thread(target=ssvrthrd)
-		self.thread.setDaemon(True)
-		self.thread.start()
-
-		# We have to call this AFTER initializing our server, so that
-		# the broker can contact us
-		# Wrap this in a try/catch because the broker may not be online yet
-		try:
-			Subscriber.__init__(self, broker,  synchronized=synchronized)		
-		except:
-			pass
-		threading.Thread.__init__(self)
-		self.setDaemon(True)
-		self.start()
-
-	def stop(self):
-#		 # FIXME: this is broken, there is no clear way to stop a
-#		 # Thrift server
-		self.broker.removeSubscriber(self)
-		self.transport.close()
-	def run(self):
-		while(True):
-			# renew subscription every 5 min
-			try:
-				self.broker.addSubscriber(self)
-			except:
-				pass
-			time.sleep(5*60)
-
-class SubscriberThriftProxy:
-	def __init__(self, host, port, proxy, aggregate = 100):
-		self.host = host
-		self.port = port
-		self.proxy = proxy
-		# for some reason, thrift clients are not thread-safe, lock during send
-		self.lock = threading.Lock()
-		self.pending = []
-		self.aggregateSize = aggregate
-	def publish(self, message):
-		self.lock.acquire()
-		sys.stdout.flush()
-		if message.has_key('aggregate') and message['aggregate'] == 'True':
-			self.pending.append(message)
-			if len(self.pending) >= self.aggregateSize:
-				try:
-					self.proxy[self.host, self.port].publishList(self.pending)
-				except Exception, e:
-					print e
-					self.lock.release()
-					raise e
-				self.pending = []
-		else:
-			try:
-				self.proxy[self.host, self.port].publish(message)
-			except Exception, e:
-				sys.stdout.flush()
-				print e
-				self.lock.release()
-				raise e
-		self.lock.release()
-
-class PublisherThrift(Publisher):
-	def __init__(self, host, port):
-		self.host = host
-		self.port = port
-		self.broker = MessageBrokerThriftProxy(host, port)
-		Publisher.__init__(self, self.broker)
-		
-####################
-# Testing Code 
-####################
-
-class TestSubscriberThrift(SubscriberThrift):
-	def __init__(self, *args, **kwargs):
-		self.queue = Queue.Queue()
-		SubscriberThrift.__init__(self, *args, **kwargs)
-	def handle(self, message):
-		self.queue.put(message)
-
-portnum = 1718
-class TestThriftMessaging(unittest.TestCase):
-	def setUp(self):
-		global portnum
-		self.broker = MessageBrokerThrift(portnum)
-		self.brokerPort = portnum
-		portnum = portnum + 1 
-		self.proxy = MessageBrokerThriftProxy('localhost', self.brokerPort)
-		self.publisher = PublisherThrift('localhost', self.brokerPort)
-		self.subscriber = TestSubscriberThrift(self.proxy, portnum)
-		portnum = portnum + 1
-	def tearDown(self):
-		pass
-	def testSetUp(self):
-		pass
-	def testPublish(self):
-		self.publisher.publish( {'message':'hello world'} )
-		self.subscriber.queue.get(True, timeout=5)
-		self.assertEqual(self.subscriber.queue.qsize(), 0)
-	def testPublishList(self):
-		nrmsgs = 10
-		msgs = []
-		for i in range(nrmsgs):
-			msgs.append( {'msgnum':str(i)} )
-		self.publisher.publishList( msgs )
-		for i in range(nrmsgs):
-			self.subscriber.queue.get(True, timeout=5)
-		self.assertEqual(self.subscriber.queue.qsize(), 0)
-	def testAggregate(self):
-		nrmsgs = self.publisher.aggregateSize
-		for i in range(nrmsgs):
-			self.assertEqual(self.subscriber.queue.qsize(), 0)
-			self.publisher.aggregate( {'msgnum':str(i)} )
-		for i in range(nrmsgs):
-			self.subscriber.queue.get(True, timeout=5)
-		self.assertEqual(self.subscriber.queue.qsize(), 0)
-	def testAggregateKeyword(self):
-		nrmsgs = self.publisher.aggregateSize
-		for i in range(nrmsgs):
-			self.assertEqual(self.subscriber.queue.qsize(), 0)
-			self.publisher.publish( {'msgnum':str(i), 'aggregate':'True'} )
-		for i in range(nrmsgs):
-			self.subscriber.queue.get(True, timeout=5)
-		self.assertEqual(self.subscriber.queue.qsize(), 0)
-
-
-if __name__=='__main__':
-	suite = unittest.TestLoader().loadTestsFromTestCase(TestThriftMessaging)
-	unittest.TextTestRunner(verbosity=2).run(suite)
-
-
diff --git a/src/tashi/nodemanager/vmcontrol/xenpv.py b/src/tashi/nodemanager/vmcontrol/xenpv.py
index 7c7de75..9401df3 100644
--- a/src/tashi/nodemanager/vmcontrol/xenpv.py
+++ b/src/tashi/nodemanager/vmcontrol/xenpv.py
@@ -28,7 +28,7 @@
 from tashi.rpycservices.rpyctypes import Errors, InstanceState, TashiException
 from tashi.rpycservices.rpyctypes import Instance, Host
 from tashi import boolean, convertExceptions, ConnectionManager, version
-from tashi.util import isolatedRPC, broken
+from tashi.util import broken
 
 import tashi.parallel
 from tashi.parallel import synchronized, synchronizedmethod
diff --git a/src/tashi/thrift/build.py b/src/tashi/thrift/build.py
deleted file mode 100755
index 42b22fa..0000000
--- a/src/tashi/thrift/build.py
+++ /dev/null
@@ -1,56 +0,0 @@
-#!/usr/bin/python
-
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-# 
-#   http://www.apache.org/licenses/LICENSE-2.0
-# 
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied.  See the License for the
-# specific language governing permissions and limitations
-# under the License.    
-
-import shutil
-import os
-from os import path
-import re
-
-if __name__ == '__main__':
-	if (path.exists('gen-py')):
-		print 'Removing \'gen-py\' directory...'
-		shutil.rmtree('gen-py')
-		
-	if (path.exists('../services')):
-		print 'Removing \'../services\' directory...'
-		shutil.rmtree('../services')
-	
-	if (path.exists('../messaging/messagingthrift')):
-		print 'Removing \'../messaging/messagingthrift\' directory...'
-		shutil.rmtree('../messaging/messagingthrift')
-	
-	print 'Generating Python code for \'services.thrift\'...'
-	os.system('thrift --gen py:new_style services.thrift')
-	
-	print 'Copying generated code to \'tashi.services\' package...'
-	shutil.copytree('gen-py/services', '../services')
-	
-	print 'Generatign Python code for \'messagingthrift\'...'
-	os.system('rm -rf gen-py')
-	os.system('thrift --gen py messagingthrift.thrift')
-	
-	print 'Copying generated code to \'tashi.messaging.messagingthrift\' package...'
-	shutil.copytree(os.path.join('gen-py', 'messagingthrift'),
-			os.path.join('..', 'messaging', 'messagingthrift'))
-
-	print 'Generating Python code for \'layoutlocality.thrift\'...'
-	os.system('thrift --gen py:new_style layoutlocality.thrift')
- 
-	print 'Copying generated code to \'tashi.services\' package...'
-	shutil.copytree('gen-py/layoutlocality', '../services/layoutlocality')
diff --git a/src/tashi/thrift/layoutlocality.thrift b/src/tashi/thrift/layoutlocality.thrift
deleted file mode 100644
index e14910c..0000000
--- a/src/tashi/thrift/layoutlocality.thrift
+++ /dev/null
@@ -1,42 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-# 
-#   http://www.apache.org/licenses/LICENSE-2.0
-# 
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied.  See the License for the
-# specific language governing permissions and limitations
-# under the License.
-
-struct BlockLocation {
-       list<string> hosts,           // hostnames of data nodes
-       list<i32> ports,              // ports for data nodes
-       list<string> names,           // hostname:port of data nodes
-       i64 blocknum,
-       i64 offset,
-       i64 length
-}
-
-struct Pathname {
-       string pathname
-}
-
-exception FileNotFoundException {
-       string message
-}
-
-service layoutservice {
-       list <BlockLocation> getFileBlockLocations(1:Pathname path, 2:i64 offset, 3:i64 length)
-                            throws (1:FileNotFoundException ouch),
-}
-
-service localityservice {
-       list <list<double>> getHopCountMatrix(1:list<string> sourceHosts, 2:list<string> destHosts),
-}
diff --git a/src/tashi/thrift/messagingthrift.thrift b/src/tashi/thrift/messagingthrift.thrift
deleted file mode 100644
index 401e9a1..0000000
--- a/src/tashi/thrift/messagingthrift.thrift
+++ /dev/null
@@ -1,36 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-# 
-#   http://www.apache.org/licenses/LICENSE-2.0
-# 
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied.  See the License for the
-# specific language governing permissions and limitations
-# under the License.    
-
-typedef map<string, string> strstrmap
-
-service SubscriberThrift{
-  # the async keyword seems to slow things down in the simple
-  # tests.  However, with non-trivial subscribers it will be 
-  # necessary to use async here.
-  async void publish(strstrmap message)
-  async void publishList(list<strstrmap> messages)
-}
-
-service MessageBrokerThrift{
-  void log(strstrmap message),
-  void addSubscriber(string host, i16 port)
-  void removeSubscriber(string host, i16 port)
-  async void publish(strstrmap message)
-  async void publishList(list<strstrmap> messages)
-
-}
-
diff --git a/src/tashi/thrift/services.thrift b/src/tashi/thrift/services.thrift
deleted file mode 100644
index fa29c30..0000000
--- a/src/tashi/thrift/services.thrift
+++ /dev/null
@@ -1,166 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-# 
-#   http://www.apache.org/licenses/LICENSE-2.0
-# 
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied.  See the License for the
-# specific language governing permissions and limitations
-# under the License.    
-
-enum Errors {
-	ConvertedException = 1,
-	NoSuchInstanceId = 2,
-	NoSuchVmId = 3,
-	IncorrectVmState = 4,
-	NoSuchHost = 5,
-	NoSuchHostId = 6,
-	InstanceIdAlreadyExists = 7,
-	HostNameMismatch = 8,
-	HostNotUp = 9,
-	HostStateError = 10,
-	InvalidInstance = 11,
-	UnableToResume = 12,
-	UnableToSuspend = 13,
-}
-
-enum InstanceState {
-	Pending = 1,		// Job submitted
-	Activating = 2,		// activateVm has been called, but instantiateVm hasn't finished yet
-	Running = 3,		// Normal state
-	Pausing = 4,		// Beginning pause sequence
-	Paused = 5		// Paused
-	Unpausing = 6,		// Beginning unpause sequence
-	Suspending = 7,		// Beginning suspend sequence
-	Resuming = 8,		// Beginning resume sequence
-	MigratePrep = 9,	// Migrate state #1
-	MigrateTrans = 10,	// Migrate state #2
-	ShuttingDown = 11,	// Beginning exit sequence
-	Destroying = 12,	// Beginning exit sequence
-	Orphaned = 13,		// Host is missing
-	Held = 14,		// Activation failed
-	Exited = 15,		// VM has exited
-	Suspended = 16,		// VM is suspended
-}
-
-enum HostState {
-	Normal = 1,
-	Drained = 2,
-	VersionMismatch = 3
-}
-
-exception TashiException {
-	1: Errors errno
-	2: string msg
-}
-
-struct Host {
-	1:i32 id,
-	2:string name,
-	3:bool up,
-	4:bool decayed,
-	5:HostState state,
-	6:i32 memory,
-	7:i32 cores,
-	8:string version
-	// Other properties (disk?)
-}
-
-struct Network {
-	1:i32 id
-	2:string name
-}
-
-struct User {
-	1:i32 id,
-	2:string name
-}
-
-struct DiskConfiguration {
-	1:string uri,
-	2:bool persistent
-}
-
-struct NetworkConfiguration {
-	1:i32 network,
-	2:string mac,
-	3:string ip
-}
-
-struct Instance {
-	1:i32 id,
-	2:i32 vmId,
-	3:i32 hostId,
-	4:bool decayed,
-	5:InstanceState state,
-	6:i32 userId,
-	7:string name, // User specified
-	8:i32 cores, // User specified
-	9:i32 memory, // User specified
-	10:list<DiskConfiguration> disks, // User specified
-	11:list<NetworkConfiguration> nics // User specified
-	12:map<string, string> hints // User specified
-}
-
-service clustermanagerservice {
-	// Client-facing RPCs
-	Instance createVm(1:Instance instance) throws (1:TashiException e)
-	
-	void shutdownVm(1:i32 instanceId) throws (1:TashiException e)
-	void destroyVm(1:i32 instanceId) throws (1:TashiException e)
-	
-	void suspendVm(1:i32 instanceId) throws (1:TashiException e)
-	Instance resumeVm(1:i32 instanceId) throws (1:TashiException e)
-	
-	void migrateVm(1:i32 instanceId, 2:i32 targetHostId) throws (1:TashiException e)
-	
-	void pauseVm(1:i32 instanceId) throws (1:TashiException e)
-	void unpauseVm(1:i32 instanceId) throws (1:TashiException e)
-	
-	list<Host> getHosts() throws (1:TashiException e)
-	list<Network> getNetworks() throws (1:TashiException e)
-	list<User> getUsers() throws (1:TashiException e)
-
-	list<Instance> getInstances() throws (1:TashiException e)
-	
-	string vmmSpecificCall(1:i32 instanceId, 2:string arg) throws (1:TashiException e)
-	
-	// NodeManager-facing RPCs
-	i32 registerNodeManager(1:Host host, 2:list<Instance> instances) throws (1:TashiException e)
-	void vmUpdate(1:i32 instanceId, 2:Instance instance, 3:InstanceState old) throws (1:TashiException e)
-
-	// Agent-facing RPCs
-	void activateVm(1:i32 instanceId, 2:Host host) throws (1:TashiException e)
-}
-
-service nodemanagerservice {
-	// ClusterManager-facing RPCs
-	i32 instantiateVm(1:Instance instance) throws (1:TashiException e)
-	
-	void shutdownVm(1:i32 vmId) throws (1:TashiException e)
-	void destroyVm(1:i32 vmId) throws (1:TashiException e)
-	
-	void suspendVm(1:i32 vmId, 2:string destination) throws (1:TashiException e)
-	i32 resumeVm(1:Instance instance, 2:string source) throws (1:TashiException e)
-	
-	string prepReceiveVm(1:Instance instance, 2:Host source) throws (1:TashiException e)
-	void migrateVm(1:i32 vmId, 2:Host target, 3:string transportCookie) throws (1:TashiException e)
-	void receiveVm(1:Instance instance, 2:string transportCookie) throws (1:TashiException e)
-	
-	void pauseVm(1:i32 vmId) throws (1:TashiException e)
-	void unpauseVm(1:i32 vmId) throws (1:TashiException e)
-
-	Instance getVmInfo(1:i32 vmId) throws (1:TashiException e)
-	list<i32> listVms() throws (1:TashiException e)
-	
-	string vmmSpecificCall(1:i32 vmId, 2:string arg) throws (1:TashiException e)
-	
-	// Host getHostInfo() throws (1:TashiException e)
-}
diff --git a/src/tashi/util.py b/src/tashi/util.py
index 18b6133..dd8a28d 100644
--- a/src/tashi/util.py
+++ b/src/tashi/util.py
@@ -149,14 +149,6 @@
 	def __delattr__(self, name):
 		return delattr(self.__dict__['__real_obj__'], name)
 
-def isolatedRPC(client, method, *args, **kw):
-	"""Opens and closes a thrift transport for a single RPC call"""
-	if (not client._iprot.trans.isOpen()):
-		client._iprot.trans.open()
-	res = getattr(client, method)(*args, **kw)
-	client._iprot.trans.close()
-	return res
-
 def signalHandler(signalNumber):
 	"""Used to denote a particular function as the signal handler for a 
 	   specific signal"""
@@ -193,7 +185,7 @@
 
 def convertExceptions(oldFunc):
 	"""This converts any exception type into a TashiException so that 
-	   it can be passed over a Thrift RPC"""
+	   it can be passed over an RPC"""
 	def newFunc(*args, **kw):
 		try:
 			return oldFunc(*args, **kw)
diff --git a/src/utils/Makefile b/src/utils/Makefile
deleted file mode 100644
index aea56ee..0000000
--- a/src/utils/Makefile
+++ /dev/null
@@ -1,24 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-# 
-#   http://www.apache.org/licenses/LICENSE-2.0
-# 
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied.  See the License for the
-# specific language governing permissions and limitations
-# under the License.    
-
-all: nmd
-
-clean:
-	rm -f ./nmd
-
-nmd: nmd.c
-	${CC} $< -o $@
diff --git a/src/utils/getLocality.py b/src/utils/getLocality.py
deleted file mode 100755
index 49ecb11..0000000
--- a/src/utils/getLocality.py
+++ /dev/null
@@ -1,68 +0,0 @@
-#!/usr/bin/python
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-# 
-#   http://www.apache.org/licenses/LICENSE-2.0
-# 
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied.  See the License for the
-# specific language governing permissions and limitations
-# under the License.
-
-import sys
-import os
-from os import system
-
-import tashi.services.layoutlocality.localityservice as localityservice
-
-from thrift import Thrift
-from thrift.transport import TSocket
-from thrift.transport import TTransport
-from thrift.protocol import TBinaryProtocol
-
-from tashi.util import getConfig
-
-(config, configFiles) = getConfig(["Client"])
-host = config.get('LocalityService', 'host')
-port = int(config.get('LocalityService', 'port'))
-
-socket = TSocket.TSocket(host, port)
-transport = TTransport.TBufferedTransport(socket)
-protocol = TBinaryProtocol.TBinaryProtocol(transport)
-client = localityservice.Client(protocol)
-transport.open()
-
-while True:
-	line1 = "\n"
-	line2 = "\n"
-	while line1 != "":
-		line1 = sys.stdin.readline()
-		if line1 == "":
-			sys.exit(0)
-		if line1 != "\n":
-			break
-	line1 = line1.strip()
-	while line2 != "":
-		line2 = sys.stdin.readline()
-		if line2 == "":
-			sys.exit(0)
-		if line2 != "\n":
-			break
-	line2 = line2.strip()
-
-	sources = line1.split(" ")
-	destinations = line2.split(" ")
-
-	mat = client.getHopCountMatrix(sources, destinations)
-	for r in mat:
-		for c in r:
-			print '%f\t'%c,
-		print '\n',
-	print '\n',
diff --git a/src/utils/nmd.c b/src/utils/nmd.c
deleted file mode 100644
index effa1d2..0000000
--- a/src/utils/nmd.c
+++ /dev/null
@@ -1,123 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- * 
- *   http://www.apache.org/licenses/LICENSE-2.0
- * 
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.    
- */
-
-#include <sys/types.h>
-#include <sys/wait.h>
-#include <dirent.h>
-#include <fcntl.h>
-#include <stdio.h>
-#include <stdlib.h>
-#include <unistd.h>
-#include <assert.h>
-
-#define SLEEP_INTERVAL 10
-#define TASHI_PATH "/usr/local/tashi/"
-#define LOG_FILE "/var/log/nodemanager.log"
-
-/* This function changes (on Linux!) its oom scoring, to make it
- * unattractive to kill
- */
-
-void make_invincible()
-{
-	int oom_adj_fd;
-	int r;
-
-	oom_adj_fd = open("/proc/self/oom_adj", O_WRONLY);
-	assert(oom_adj_fd != -1);
-	r = write(oom_adj_fd, "-17\n", 4);
-	assert(r == 4);
-	close(oom_adj_fd);
-
-}
-
-/* This function resets (on Linux!) its oom scoring to default
- */
-void make_vulnerable()
-{
-	int oom_adj_fd;
-	int r;
-
-	oom_adj_fd = open("/proc/self/oom_adj", O_WRONLY);
-	assert(oom_adj_fd != -1);
-	r = write(oom_adj_fd, "0\n", 2);
-	assert(r == 2);
-	close(oom_adj_fd);
-}
-
-int main(int argc, char **argv)
-{
-	char* env[2];
-	int status;
-	DIR* d;
-	int pid;
-	int lfd;
-	int foreground=0;
-
-/* If first argument is "-f", run in foreground */
-	if ((argc > 1) && (strncmp(argv[1], "-f", 3)==0)) {
-		foreground=1;
-	}
-/* If not running in foreground, fork off and exit the parent.
- * The child closes its default file descriptors.
- */
-	if (!foreground) {
-		pid = fork();
-		if (pid != 0) {
-			exit(0);
-		}
-		close(0);
-		close(1);
-		close(2);
-	}
-/* Adjust OOM preference */
-	make_invincible();
-/* Configure environment of children */
-	env[0] = "PYTHONPATH="TASHI_PATH"/src/";
-	env[1] = NULL;
-	while (1) {
-		pid = fork();
-		if (pid == 0) {
-			/* child */
-			/* nodemanagers are vulnerable. Not the supervisor. */
-			make_vulnerable();
-			if (!foreground) {
-				/* If not running fg, open log file */
-				lfd = open(LOG_FILE, O_WRONLY|O_APPEND|O_CREAT);
-				if (lfd < 0) {
-					/* If this failed, open something? */
-					lfd = open("/dev/null", O_WRONLY);
-				}
-				/* Make this fd stdout and stderr */
-				dup2(lfd, 2);
-				dup2(lfd, 1);
-				/* close stdin */
-				close(0);
-			}
-			chdir(TASHI_PATH);
-			/* start node manager with python environment */
-			execle("./bin/nodemanager.py", "./bin/nodemanager.py", NULL, env);
-			exit(-1);
-		}
-		/* sleep before checking for child's status */
-		sleep(SLEEP_INTERVAL);
-		/* catch child exiting and go through loop again */
-		waitpid(pid, &status, 0);
-	}	 /* while (1) */
-}