Comment jobs (3fbee17b) · Commits · Mirror / cc-node

cloudcontrol/node/jobs.py

+53 −4

Original line number	Diff line number	Diff line
		@@ -114,6 +114,7 @@ class BaseThreadedJob(Thread):
		self.job_manager.notify(self)

		def start(self):
		"""Start job in a background thread."""
		# first we run pre_job as it could raise an exception
		try:
		self.pre_job()
		@@ -160,6 +161,9 @@ class ForkedJob(object):
		"""

		def __init__(self, job_manager):
		"""
		:param job_manager: :class:`JobManager` instance
		"""
		self.job_manager = job_manager

		#: job id
		@@ -172,15 +176,23 @@ class ForkedJob(object):
		self.fork_pid = None

		def pre_job(self):
		# overide in sub class
		"""This method represents any preparation job that must be done in the
		parent before the fork.
		"""
		pass

		def run_job(self):
		# overide in sub class
		"""This represent the work that will be done in the forked child.

		This method MUST be redefined in subclasses.
		"""
		pass

		def after_fork(self):
		# overide in sub class
		"""This method will be called just after fork in the child.

		It does nothing by default and it can be redifined in subclasses.
		"""
		pass

		def fatal(self, fmt, args, *kwargs):
		@@ -225,7 +237,12 @@ class ForkedJob(object):
		self.fatal(fmt, *args, status=kwargs.get('status', 1))

		def close_fds(self):
		"""Close all fds uneeded fds in children."""
		"""Close all fds uneeded fds in children.

		If global debug variable in configuration is set to True, then it will
		prevent standard input/output from being closed thus allowing logging on
		stderr.
		"""
		# get max fd
		limit = resource.getrlimit(resource.RLIMIT_NOFILE)[1]
		if limit == resource.RLIM_INFINITY:
		@@ -259,6 +276,15 @@ class ForkedJob(object):
		signal.signal(signal.SIGINT, signal.SIG_IGN)

		def run(self):
		"""This method performs all the hard work by doing the actual fork.

		It catches all possible exceptions in the child, as this would prevents
		the latter from going back in the stack and doing nasty things with
		libev loop or sjRPC.
		Thus you do not need to capture all exceptions in your code,
		furthermore, if you need to exit from the child, you'd better use
		`os._exit <http://docs.python.org/library/os.html#os._exit>`_ function.
		"""
		try:
		self.fork_pid = os.fork()
		except OSError as exc:
		@@ -295,10 +321,20 @@ class ForkedJob(object):
		raise

		def start(self):
		"""This will start the job by executing :py:meth:`pre_job` method and
		:py:meth:`run`.

		"""
		self.pre_job()
		self.run()

		def stop(self):
		"""This would be called to stop the job.

		It will kill the child but it will not call `os.waitpid
		<http://docs.python.org/library/os.html#os.waitpid>`_ to get return
		status from zombie process. :py:meth:`wait` MUST be called anyway.
		"""
		try:
		os.kill(self.fork_pid, signal.SIGKILL)
		except OSError as exc:
		@@ -314,6 +350,12 @@ class ForkedJob(object):
		self.job_manager.notify(self)

		def wait(self):
		"""This will wait for the fork to end and raise exception depending on
		child return status.

		.. warning::
		This method MUST be called.
		"""
		if self.fork_pid is None:
		return
		try:
		@@ -345,12 +387,19 @@ class ForkedJob(object):
		self.notify()

		def join(self):
		"""This provides an API similar to `threading.Thread.join
		<http://docs.python.org/library/threading.html#threading.Thread.join>`_
		, you can wait
		for the job termination from multiple points in your program but one of
		these and only one MUST be `wait` method.
		"""
		self.job_done.wait()


		class BaseIOJob(ForkedJob):
		"""Fork job that set ionice on the child."""

		#: level of io nice that will be set (see :manpage:`ionice(1)`)
		IO_NICE = 7

		def after_fork(self):