Author: tmckay
Date: 2012-12-14 14:58:59 +0000 (Fri, 14 Dec 2012)
New Revision: 5594
Modified:
branches/elephant/sage/python/sage/aviary/aviaryoperations.py
Log:
Make start_name_node async only.
Add stop_name_node (also async).
Modified: branches/elephant/sage/python/sage/aviary/aviaryoperations.py
===================================================================
--- branches/elephant/sage/python/sage/aviary/aviaryoperations.py 2012-12-14 14:16:55 UTC
(rev 5593)
+++ branches/elephant/sage/python/sage/aviary/aviaryoperations.py 2012-12-14 14:58:59 UTC
(rev 5594)
@@ -474,8 +474,8 @@
subtype)
def set_job_attribute(self, scheduler, job_id, name, value, callback, submission):
- assert callback
-
+ assert callable(callback)
+
def my_callback(result):
self.client_pool.return_object(job_client)
result = self._pretty_result(result, scheduler.Machine)
@@ -507,7 +507,7 @@
t.start()
def submit_job(self, scheduler, ad, callback):
- assert callback
+ assert callable(callback)
type_to_aviary = {int: "INTEGER",
float: "FLOAT",
@@ -800,7 +800,7 @@
return res;
def get_job_summaries(self, submission, callback, machine_name):
- assert callback
+ assert callable(callback)
def to_int_seconds(dt):
# Change a datetime.datetime into int seconds since epoch
@@ -904,7 +904,8 @@
# Equivalence? We don't want no stinking QMF structural equivalence!
self.use_MethodResult_for_sync_calls = False
- def start_name_node(self, host, bin_file, owner, callback=None):
+ def start_name_node(self, host, bin_file, owner, callback):
+ assert callable(callback)
client = self.client_pool.get_object()
self._setup_client(client,
@@ -923,23 +924,57 @@
data = result.ref
return (status, data)
- if callback:
- def my_callback(result):
- self.client_pool.return_object(client)
- callback(*result_tuple(result, host))
+ def my_callback(result):
+ self.client_pool.return_object(client)
+ callback(*result_tuple(result, host))
- t = CallThread(self.call_client_retry, my_callback,
- client, "startNameNode", bin_file, owner)
- t.start()
- else:
- def my_process_results(result):
- return result_tuple(result, host)
+ t = CallThread(self.call_client_retry, my_callback,
+ client, "startNameNode", bin_file, owner)
+ t.start()
- res = self._call_sync(self.call_client_retry, my_process_results,
- client, "startNameNode", bin_file, owner)
+ def stop_name_node(self, host, ids, callback):
+ assert callable(callback)
+
+ if type(ids) not in (list, tuple):
+ ids = (ids)
+
+ client = self.client_pool.get_object()
+ self._setup_client(client,
+ self.servers,
+ host,
+ "stopNameNode")
+
+ refs = []
+ for id in ids:
+ new_ref = client.factory.create("ns1:HadoopID")
+ try:
+ if type(id) in (str,unicode):
+ new_ref.ipc = id
+ else:
+ new_ref.id = int(id)
+ except Exception:
+ raise Exception("Invalid HadoopID value")
+ refs.append(new_ref)
+
+ def result_tuple(result, host):
+ data = None
+ result = self._pretty_result(result, host)
+ if isinstance(result, Exception):
+ status = result
+ else:
+ status = _AviaryCommon._get_status(result.status)
+ if status == "OK" and hasattr(result, "results"):
+ data = result.results
+ return (status, data)
+
+ def my_callback(result):
self.client_pool.return_object(client)
- return res;
+ callback(*result_tuple(result, host))
+ t = CallThread(self.call_client_retry, my_callback,
+ client, "stopNameNode", refs)
+ t.start()
+
class AviaryOperations(Catalog):
def __init__(self, name, datadir,
locator_uri="",