"""

"""

topology_file = os.path.join(CLUSTERSHELL_CONFIG_DIR, 'topology.conf')

_std_default = { "stderr" : False,

"stdout_msgtree" : True,

"stderr_msgtree" : True,

"engine" : 'auto',

"port_qlimit" : 100,

"auto_tree" : False,

"topology_file" : topology_file }

_std_info = { "debug" : False,

"print_debug" : _task_print_debug,

"fanout" : 64,

"grooming_delay" : 0.25,

"connect_timeout" : 10,

"command_timeout" : 0 }

_tasks = {}

_taskid_max = 0

_task_lock = threading.Lock()

class _SyncMsgHandler(EventHandler):

"""Special task control port event handler.

def ev_msg(self, port, msg):

"""Message received: call appropriate task method."""

# pull out function and its arguments from message

func, (args, kwargs) = msg[0], msg[1:]

# call task method

func(port.task, *args, **kwargs)

class tasksyncmethod(object):

"""Class encapsulating a function that checks if the calling

def __call__(self, f):

def taskfunc(*args, **kwargs):

# pull out the class instance

task, fargs = args[0], args[1:]

# check if the calling task is the current thread task

if task._is_task_self():

return f(task, *fargs, **kwargs)

elif task._dispatch_port:

# no, safely call the task method by message

# through the task special dispatch port

task._dispatch_port.msg_send((f, fargs, kwargs))

else:

task.info("print_debug")(task, "%s: dropped call: %s" % \

(task, str(fargs)))

# modify the decorator meta-data for pydoc

# Note: should be later replaced by @wraps (functools)

# as of Python 2.5

taskfunc.__name__ = f.__name__

taskfunc.__doc__ = f.__doc__

taskfunc.__dict__ = f.__dict__

taskfunc.__module__ = f.__module__

return taskfunc

class _SuspendCondition(object):

"""Special class to manage task suspend condition."""

def __init__(self, lock=threading.RLock(), initial=0):

self._cond = threading.Condition(lock)

self.suspend_count = initial

def atomic_inc(self):

"""Increase suspend count."""

self._cond.acquire()

self.suspend_count += 1

self._cond.release()

def atomic_dec(self):

"""Decrease suspend count."""

self._cond.acquire()

self.suspend_count -= 1

self._cond.release()

def wait_check(self, release_lock=None):

"""Wait for condition if needed."""

self._cond.acquire()

try:

if self.suspend_count > 0:

if release_lock:

release_lock.release()

self._cond.wait()

finally:

self._cond.release()

def notify_all(self):

"""Signal all threads waiting for condition."""

self._cond.acquire()

try:

self.suspend_count = min(self.suspend_count, 0)

self._cond.notifyAll()

finally:

self._cond.release()

def __new__(cls, thread=None):

"""

if thread:

if thread not in cls._tasks:

cls._tasks[thread] = object.__new__(cls)

return cls._tasks[thread]

return object.__new__(cls)

def __init__(self, thread=None):

"""

if not getattr(self, "_engine", None):

# first time called

self._default_lock = threading.Lock()

self._default = self.__class__._std_default.copy()

self._info = self.__class__._std_info.copy()

# use factory class PreferredEngine that gives the proper

# engine instance

self._engine = PreferredEngine(self.default("engine"), self._info)

self.timeout = 0

# task synchronization objects

self._run_lock = threading.Lock() # primitive lock

self._suspend_lock = threading.RLock() # reentrant lock

# both join and suspend conditions share the same underlying lock

self._suspend_cond = Task._SuspendCondition(self._suspend_lock, 1)

self._join_cond = threading.Condition(self._suspend_lock)

self._suspended = False

self._quit = False

# Default router

self.topology = None

self.router = None

self.pwrks = {}

self.pmwkrs = {}

# STDIN tree

self._msgtree = None

# STDERR tree

self._errtree = None

# dict of sources to return codes

self._d_source_rc = {}

# dict of return codes to sources

self._d_rc_sources = {}

# keep max rc

self._max_rc = 0

# keep timeout'd sources

self._timeout_sources = set()

# allow no-op call to getters before resume()

self._reset()

# special engine port for task method dispatching

self._dispatch_port = EnginePort(self,

handler=Task._SyncMsgHandler(),

autoclose=True)

self._engine.add(self._dispatch_port)

# set taskid used as Thread name

Task._task_lock.acquire()

Task._taskid_max += 1

self._taskid = Task._taskid_max

Task._task_lock.release()

# create new thread if needed

self._thread_foreign = bool(thread)

if self._thread_foreign:

self.thread = thread

else:

self.thread = thread = \

threading.Thread(None,

Task._thread_start,

"Task-%d" % self._taskid,

args=(self,))

Task._tasks[thread] = self

thread.start()

def _is_task_self(self):

"""Private method used by the library to check if the task is

return self.thread == threading.currentThread()

def default_excepthook(self, exc_type, exc_value, tb):

"""Default excepthook for a newly Task. When an exception is

print >> sys.stderr, 'Exception in thread %s:' % self.thread

traceback.print_exception(exc_type, exc_value, tb, file=sys.stderr)

_excepthook = default_excepthook

def _getexcepthook(self):

return self._excepthook

def _setexcepthook(self, hook):

self._excepthook = hook

# If thread has not been created by us, install sys.excepthook which

# might handle uncaught exception.

if self._thread_foreign:

sys.excepthook = self._excepthook

# When an exception is raised and uncaught on Task's thread,

# excepthook is called. You may want to override this three

# arguments method (very similar of what you can do with

# sys.excepthook)."""

excepthook = property(_getexcepthook, _setexcepthook)

def _thread_start(self):

"""Task-managed thread entry point"""

while not self._quit:

self._suspend_cond.wait_check()

if self._quit:

break

try:

self._resume()

except:

self.excepthook(*sys.exc_info())

self._quit = True

self._terminate(kill=True)

def _run(self, timeout):

"""Run task (always called from its self thread)."""

# check if task is already running

if self._run_lock.locked():

raise AlreadyRunningError("task is already running")

# use with statement later

try:

self._run_lock.acquire()

self._engine.run(timeout)

finally:

self._run_lock.release()

def set_topology(self, topology_file):

"""Set new propagation topology from provided file."""

self.set_default("topology_file", topology_file)

self.topology = self._default_topology()

def _default_topology(self):

try:

parser = TopologyParser()

parser.load(self.default("topology_file"))

return parser.tree(_getshorthostname())

except TopologyError, exc:

logging.getLogger(__name__).exception("_default_topology(): %s", \

str(exc))

raise

return None

def _default_router(self):

if self.router is None:

topology = self.topology

self.router = PropagationTreeRouter(str(topology.root.nodeset), \

topology)

return self.router

def default(self, default_key, def_val=None):

"""

self._default_lock.acquire()

try:

return self._default.get(default_key, def_val)

finally:

self._default_lock.release()

def set_default(self, default_key, value):

"""

self._default_lock.acquire()

try:

self._default[default_key] = value

finally:

self._default_lock.release()

def info(self, info_key, def_val=None):

"""

return self._info.get(info_key, def_val)

@tasksyncmethod()

def set_info(self, info_key, value):

"""

self._info[info_key] = value

def shell(self, command, **kwargs):

"""

handler = kwargs.get("handler", None)

timeo = kwargs.get("timeout", None)

autoclose = kwargs.get("autoclose", False)

stderr = kwargs.get("stderr", self.default("stderr"))

if kwargs.get("nodes", None):

assert kwargs.get("key", None) is None, \

"'key' argument not supported for distant command"

tree = kwargs.get("tree")

if tree and self.topology is None:

raise TaskError("tree mode required for distant shell command" \

" with unknown topology!")

if tree is None: # means auto

tree = self.default("auto_tree") and (self.topology is not None)

if tree:

# create tree of ssh worker

worker = WorkerTree(NodeSet(kwargs["nodes"]), command=command,

handler=handler, stderr=stderr,

timeout=timeo, autoclose=autoclose)

else:

# create ssh-based worker

worker = WorkerSsh(NodeSet(kwargs["nodes"]), command=command,

handler=handler, stderr=stderr,

timeout=timeo, autoclose=autoclose)

else:

# create (local) worker

worker = WorkerPopen(command, key=kwargs.get("key", None),

handler=handler, stderr=stderr,

timeout=timeo, autoclose=autoclose)

# schedule worker for execution in this task

self.schedule(worker)

return worker

def copy(self, source, dest, nodes, **kwargs):

"""

assert nodes != None, "local copy not supported"

handler = kwargs.get("handler", None)

stderr = kwargs.get("stderr", self.default("stderr"))

timeo = kwargs.get("timeout", None)

preserve = kwargs.get("preserve", None)

reverse = kwargs.get("reverse", False)

# create a new copy worker

worker = WorkerSsh(nodes, source=source, dest=dest, handler=handler,

stderr=stderr, timeout=timeo, preserve=preserve,

reverse=reverse)

self.schedule(worker)

return worker

def rcopy(self, source, dest, nodes, **kwargs):

"""

kwargs['reverse'] = True

return self.copy(source, dest, nodes, **kwargs)

@tasksyncmethod()

def _add_port(self, port):

"""Add an EnginePort instance to Engine (private method)."""

self._engine.add(port)

@tasksyncmethod()

def _remove_port(self, port):

"""Remove a port from Engine (private method)."""

self._engine.remove(port)

def port(self, handler=None, autoclose=False):

"""

port = EnginePort(self, handler, autoclose)

self._add_port(port)

return port

def timer(self, fire, handler, interval=-1.0, autoclose=False):

"""

assert fire >= 0.0, \

"timer's relative fire time must be a positive floating number"

timer = EngineTimer(fire, interval, autoclose, handler)

# The following method may be sent through msg port (async

# call) if called from another task.

self._add_timer(timer)

# always return new timer (sync)

return timer

@tasksyncmethod()

def _add_timer(self, timer):

"""Add a timer to task engine (thread-safe)."""

self._engine.add_timer(timer)

@tasksyncmethod()

def schedule(self, worker):

"""

assert self in Task._tasks.values(), "deleted task"

# bind worker to task self

worker._set_task(self)

# add worker clients to engine

for client in worker._engine_clients():

self._engine.add(client)

def _resume_thread(self):

"""Resume task - called from another thread."""

self._suspend_cond.notify_all()

def _resume(self):

"""Resume task - called from self thread."""

assert self.thread == threading.currentThread()

try:

try:

self._reset()

self._run(self.timeout)

except EngineTimeoutException:

raise TimeoutError()

except EngineAbortException, exc:

self._terminate(exc.kill)

except EngineAlreadyRunningError:

raise AlreadyRunningError("task engine is already running")

finally:

# task becomes joinable

self._join_cond.acquire()

self._suspend_cond.suspend_count += 1

self._join_cond.notifyAll()

self._join_cond.release()

def resume(self, timeout=0):

"""

# If you change options here, check Task.run() compatibility.

self.timeout = timeout

self._suspend_cond.atomic_dec()

if self._is_task_self():

self._resume()

else:

self._resume_thread()

def run(self, command=None, **kwargs):

"""

worker = None

timeout = 0

# Both resume() and shell() support a 'timeout' parameter. We need a

# trick to behave correctly for both cases.

#

# Here, we mock: task.resume(10)

if type(command) in (int, float):

timeout = command

command = None

# Here, we mock: task.resume(timeout=10)

elif 'timeout' in kwargs and command is None:

timeout = kwargs.pop('timeout')

# All other cases mean a classical: shell(...)

# we mock: task.shell("mycommand", [timeout=..., ...])

elif command is not None:

worker = self.shell(command, **kwargs)

self.resume(timeout)

return worker

@tasksyncmethod()

def _suspend_wait(self):

"""Suspend request received."""

assert task_self() == self

# atomically set suspend state

self._suspend_lock.acquire()

self._suspended = True

self._suspend_lock.release()

# wait for special suspend condition, while releasing l_run

self._suspend_cond.wait_check(self._run_lock)

# waking up, atomically unset suspend state

self._suspend_lock.acquire()

self._suspended = False

self._suspend_lock.release()

def suspend(self):

"""

# first of all, increase suspend count

self._suspend_cond.atomic_inc()

# call synchronized suspend method

self._suspend_wait()

# wait for stopped task

self._run_lock.acquire() # run_lock ownership transfer

# get result: are we really suspended or just stopped?

result = True

self._suspend_lock.acquire()

if not self._suspended:

# not acknowledging suspend state, task is stopped

result = False

self._run_lock.release()

self._suspend_lock.release()

return result

@tasksyncmethod()

def _abort(self, kill=False):

"""Abort request received."""

assert task_self() == self

# raise an EngineAbortException when task is running

self._engine.abort(kill)

def abort(self, kill=False):

"""

if self._run_lock.acquire(0):

self._quit = True

self._run_lock.release()

if self._is_task_self():

self._terminate(kill)

else:

# abort on stopped/suspended task

self.resume()

else:

# self._run_lock is locked, call synchronized method

self._abort(kill)

def _terminate(self, kill):

"""

if kill:

# invalidate dispatch port

self._dispatch_port = None

# clear engine

self._engine.clear(clear_ports=kill)

# clear result objects

self._reset()

# destroy task if needed

if kill:

Task._task_lock.acquire()

try:

del Task._tasks[threading.currentThread()]

finally:

Task._task_lock.release()

def join(self):

"""

self._join_cond.acquire()

try:

if self._suspend_cond.suspend_count > 0:

if not self._suspended:

# ignore stopped task

return

self._join_cond.wait()

finally:

self._join_cond.release()

def running(self):

"""

return self._engine.running

def _reset(self):

"""

# check and reset stdout MsgTree

if self.default("stdout_msgtree"):

if not self._msgtree:

self._msgtree = MsgTree()

self._msgtree.clear()

else:

self._msgtree = None

# check and reset stderr MsgTree

if self.default("stderr_msgtree"):

if not self._errtree:

self._errtree = MsgTree()

self._errtree.clear()

else:

self._errtree = None

# other re-init's

self._d_source_rc = {}

self._d_rc_sources = {}

self._max_rc = 0

self._timeout_sources.clear()

def _msg_add(self, source, msg):

"""

msgtree = self._msgtree

if msgtree is not None:

msgtree.add(source, msg)

def _errmsg_add(self, source, msg):

"""

errtree = self._errtree

if errtree is not None:

errtree.add(source, msg)

def _rc_set(self, source, rc, override=True):

"""

if not override and self._d_source_rc.has_key(source):

return

# store rc by source

self._d_source_rc[source] = rc

# store source by rc

self._d_rc_sources.setdefault(rc, set()).add(source)

# update max rc

if rc > self._max_rc:

self._max_rc = rc