<NME> worker.py
<BEF> #!/usr/bin/env python
# -*- coding: utf-8 -*-

# Copyright (C) 2009-2014:
#     Gabes Jean, naparuba@gmail.com
#     Gerhard Lausser, Gerhard.Lausser@consol.de
#     Gregory Starck, g.starck@gmail.com
#     Hartmut Goebel, h.goebel@goebel-consult.de
#
# This file is part of Shinken.
#
# Shinken is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Shinken is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with Shinken.  If not, see <http://www.gnu.org/licenses/>.

from Queue import Empty

# In android, we should use threads, not process
#they can die if they do not do anything (param timeout)

from Queue import Empty
from multiprocessing import Process, Queue

import time
import sys

from shinken.log import logger, BrokHandler
from shinken.misc.common import setproctitle

class Worker(object):
    """This class is used for poller and reactionner to work.
    The worker is a process launch by theses process and read Message in a Queue
    (self.s) (slave)
    They launch the Check and then send the result in the Queue self.m (master)
    they can die if they do not do anything (param timeout)

    """

    id = 0  # None
    _process = None
    _mortal = None
    _idletime = None
    _timeout = None
    _c = None

    def __init__(self, id, s, returns_queue, processes_by_worker, mortal=True, timeout=300,
                 max_plugins_output_length=8192, target=None, loaded_into='unknown',
                 http_daemon=None):
        self.id = self.__class__.id
        self.__class__.id += 1

        self._mortal = mortal
        self._idletime = 0
        self._timeout = timeout
        self.s = None
        self.processes_by_worker = processes_by_worker
        self._c = Queue()  # Private Control queue for the Worker
        # By default, take our own code
        if target is None:
            target = self.work
        self._process = Process(target=self._prework, args=(target, s, returns_queue, self._c))
        self.returns_queue = returns_queue
        self.max_plugins_output_length = max_plugins_output_length
        self.i_am_dying = False
    # AND close correctly the queues (input and output)
    # each queue got a thread, so close it too....
    def terminate(self):
        self._process.terminate()
        self._c.close()
        self._c.join_thread()
        self.input_queue.close()
            if isinstance(handler, BrokHandler):
                logger.info("Cleaning BrokHandler %r from logger.handlers..", handler)
                logger.removeHandler(handler)
        real_work(*args)

    def is_mortal(self):
        return self._mortal


    def start(self):
        self._process.start()


    # Kill the background process
    # AND close correctly the queues (input and output)
    # each queue got a thread, so close it too....
    def terminate(self):
        # We can just terminate process, not threads
        if not is_android:
            self._process.terminate()
        # Is we are with a Manager() way
        # there should be not such functions
        if hasattr(self._c, 'close'):
            self._c.close()
            self._c.join_thread()
        if hasattr(self.s, 'close'):
            self.s.close()
            self.s.join_thread()

    def join(self, timeout=None):
        self._process.join(timeout)

    def is_alive(self):
        return self._process.is_alive()

    def is_killable(self):
        return self._mortal and self._idletime > self._timeout

    def add_idletime(self, time):
        self._idletime = self._idletime + time

    def reset_idle(self):
        self._idletime = 0

    def send_message(self, msg):
        self._c.put(msg)

    # A zombie is immortal, so kill not be kill anymore
    def set_zombie(self):
        self._mortal = False

    # Get new checks if less than nb_checks_max
    # If no new checks got and no check in queue,
    # sleep for 1 sec
    # REF: doc/shinken-action-queues.png (3)
    def get_new_checks(self):
        try:
            while(len(self.checks) < self.processes_by_worker):
                # print "I", self.id, "wait for a message"
                msg = self.s.get(block=False)
                if msg is not None:
                    self.checks.append(msg.get_data())
                # print "I", self.id, "I've got a message!"
        except Empty, exp:
            if len(self.checks) == 0:
                self._idletime = self._idletime + 1
                time.sleep(1)
        # Maybe the Queue() is not available, if so, just return
        # get back to work :)
        except IOError, exp:
            return


    # Launch checks that are in status
    # REF: doc/shinken-action-queues.png (4)
    def launch_new_checks(self):
        # queue
        for chk in self.checks:
            if chk.status == 'queue':
                self._idletime = 0
                r = chk.execute()
                # Maybe we got a true big problem in the
                # action launching
                #We answer to the master
                #msg = Message(id=self.id, type='Result', data=action)
                try:
                    self.returns_queue.append(action)#msg)
                except IOError , exp:
                    print "[%d]Exiting: %s" % (self.id, exp)
                    sys.exit(2)
        wait_time = 1
        now = time.time()
        for action in self.checks:
            if action.status == 'launched' and action.last_poll < now - action.wait_time:
                action.check_finished(self.max_plugins_output_length)
                wait_time = min(wait_time, action.wait_time)
                # If action done, we can launch a new one
            if action.status in ('done', 'timeout'):
                to_del.append(action)
                # We answer to the master
                # msg = Message(id=self.id, type='Result', data=action)
                try:
                    self.returns_queue.put(action)
                except IOError, exp:
                    logger.error("[%d] Exiting: %s", self.id, exp)
                    sys.exit(2)

        # Little sleep
        self.wait_time = wait_time

        for chk in to_del:
            self.checks.remove(chk)

        # Little sleep
        time.sleep(wait_time)

    # Check if our system time change. If so, change our
    def check_for_system_time_change(self):
        now = time.time()
    #c = Control Queue for the worker
    def work(self, s, returns_queue, c):
        ## restore default signal handler for the workers:
        signal.signal(signal.SIGTERM, signal.SIG_DFL)
        timeout = 1.0
        self.checks = []
        self.returns_queue = returns_queue
            return 0


    # Wrapper function for work in order to catch the exception
    # to see the real work, look at do_work
    def work(self, s, returns_queue, c):
        try:
            self.do_work(s, returns_queue, c)
        # Catch any exception, try to print it and exit anyway
        except Exception, exp:
            output = cStringIO.StringIO()
            traceback.print_exc(file=output)
            logger.error("Worker '%d' exit with an unmanaged exception : %s",
                         self.id, output.getvalue())
            output.close()
            # Ok I die now
            raise


    # id = id of the worker
    # s = Global Queue Master->Slave
    # m = Queue Slave->Master
    # return_queue = queue managed by manager
    # c = Control Queue for the worker
    def do_work(self, s, returns_queue, c):
        # restore default signal handler for the workers:
        # but on android, we are a thread, so don't do it
        if not is_android:
            signal.signal(signal.SIGTERM, signal.SIG_DFL)

        self.set_proctitle()

        print "I STOP THE http_daemon", self.http_daemon
        if self.http_daemon:
            self.http_daemon.shutdown()

        timeout = 1.0
        self.checks = []
        self.returns_queue = returns_queue
        self.s = s
        self.t_each_loop = time.time()
        while True:
            begin = time.time()
            msg = None
            cmsg = None

            # If we are dying (big problem!) we do not
            # take new jobs, we just finished the current one
            if not self.i_am_dying:
                # REF: doc/shinken-action-queues.png (3)
                self.get_new_checks()
                # REF: doc/shinken-action-queues.png (4)
                self.launch_new_checks()
            # REF: doc/shinken-action-queues.png (5)
            self.manage_finished_checks()

            # Now get order from master
            try:
                cmsg = c.get(block=False)
                if cmsg.get_type() == 'Die':
                    logger.debug("[%d] Dad say we are dying...", self.id)
                    break
            except Exception:
                pass

            # Look if we are dying, and if we finish all current checks
            # if so, we really die, our master poller will launch a new
            # worker because we were too weak to manage our job :(
            if len(self.checks) == 0 and self.i_am_dying:
                logger.warning("[%d] I DIE because I cannot do my job as I should"
                               "(too many open files?)... forgot me please.", self.id)
                break

            # Manage a possible time change (our avant will be change with the diff)
            diff = self.check_for_system_time_change()
            begin += diff

            timeout -= time.time() - begin
            if timeout < 0:
                timeout = 1.0

    def set_proctitle(self):
        setproctitle("shinken-%s worker" % self.loaded_into)

<MSG> Add: reactionner can run under android.

<DFF> @@ -28,7 +28,19 @@
 #they can die if they do not do anything (param timeout)
 
 from Queue import Empty
-from multiprocessing import Process, Queue
+
+# In android, we sould use threads, not process
+is_android = True
+try:
+   import android
+except ImportError:
+   is_android = False
+
+if not is_android:
+   from multiprocessing import Process, Queue
+else:
+   from Queue import Queue
+   from threading import Thread as Process
 
 import time
 import sys
@@ -74,7 +86,9 @@ class Worker:
     # AND close correctly the queues (input and output)
     # each queue got a thread, so close it too....
     def terminate(self):
-        self._process.terminate()
+        # We can just terminate process, not threads
+        if not is_android:
+            self._process.terminate()
         self._c.close()
         self._c.join_thread()
         self.input_queue.close()
@@ -164,7 +178,11 @@ class Worker:
                 #We answer to the master
                 #msg = Message(id=self.id, type='Result', data=action)
                 try:
-                    self.returns_queue.append(action)#msg)
+                    # In android, the Queue is changed
+                    if not is_android:
+                        self.returns_queue.append(action)
+                    else:
+                        self.returns_queue.put(action)
                 except IOError , exp:
                     print "[%d]Exiting: %s" % (self.id, exp)
                     sys.exit(2)
@@ -200,7 +218,9 @@ class Worker:
     #c = Control Queue for the worker
     def work(self, s, returns_queue, c):
         ## restore default signal handler for the workers:
-        signal.signal(signal.SIGTERM, signal.SIG_DFL)
+        # but on android, we are a thread, so don't do it
+        if not is_android:
+           signal.signal(signal.SIGTERM, signal.SIG_DFL)
         timeout = 1.0
         self.checks = []
         self.returns_queue = returns_queue

{'commit': 'diff --git a/bin/shinken-reactionner b/bin/shinken-reactionner\nindex 8e4c75040..fe5aab49a 100755\n--- a/bin/shinken-reactionner\n+++ b/bin/shinken-reactionner\n@@ -38,6 +38,18 @@ import sys\n import os\n import optparse\n \n+# Try to see if we are in an android device or not\n+is_android = True\n+try:\n+   import android\n+   # Add our main script dir\n+   if os.path.exists(\'/sdcard/sl4a/scripts/\'):\n+       sys.path.append(\'/sdcard/sl4a/scripts/\')\n+       os.chdir(\'/sdcard/sl4a/scripts/\')\n+except ImportError:\n+   is_android = False\n+\n+\n try:\n     from shinken.bin import VERSION\n     import shinken\ndiff --git a/shinken/daemon.py b/shinken/daemon.py\nindex 6beaf0d7c..9547c514f 100644\n--- a/shinken/daemon.py\n+++ b/shinken/daemon.py\n@@ -39,7 +39,7 @@ from shinken.modulesmanager import ModulesManager\n from shinken.property import StringProp, BoolProp, PathProp, ConfigPathProp, IntegerProp\n \n \n-if os.name != \'nt\':\n+try:\n     import pwd, grp\n     from pwd import getpwnam\n     from grp import getgrnam\n@@ -48,13 +48,21 @@ if os.name != \'nt\':\n \n     def get_cur_group():\n         return grp.getgrgid( os.getgid() ).gr_name\n-else:\n+except ImportError, exp: # Like in nt system or Android\n     # temporary workarround :\n     def get_cur_user():\n         return "shinken"\n     def get_cur_group():\n         return "shinken"\n \n+# Try to see if we are in an android device or not\n+is_android = True\n+try:\n+   import android\n+except ImportError:\n+   is_android = False\n+\n+\n ##########################   DAEMON PART    ###############################\n # The standard I/O file descriptors are redirected to /dev/null by default.\n REDIRECT_TO = getattr(os, "devnull", "/dev/null")\n@@ -94,6 +102,12 @@ class Interface(Pyro.core.ObjBase, object):\n         return self.app.cur_conf is not None\n \n \n+# If we are under android, we can\'t give parameters\n+if is_android:\n+  DEFAULT_WORK_DIR = \'/sdcard/sl4a/scripts/\'\n+else:\n+  DEFAULT_WORK_DIR = \'var\'\n+\n \n class Daemon(object):\n \n@@ -105,7 +119,7 @@ class Daemon(object):\n         #  os.path.join( os.getcwd(), sys.argv[0] )\n         # \n         # as returned once the daemon is started.\n-        \'workdir\':       PathProp(default=\'var\'),\n+        \'workdir\':       PathProp(default=DEFAULT_WORK_DIR),\n         \'host\':          StringProp(default=\'0.0.0.0\'),\n         \'user\':          StringProp(default=get_cur_user()),\n         \'group\':         StringProp(default=get_cur_group()),\n@@ -542,6 +556,10 @@ If change failed we sys.exit(2) """\n         if insane is None:\n             insane = not self.idontcareaboutsecurity\n \n+        if is_android:\n+           print "Sorry, you can\'t change user on this system"\n+           return\n+\n         # TODO: change user on nt\n         if os.name == \'nt\':\n             print("Sorry, you can\'t change user on this system")\ndiff --git a/shinken/satellite.py b/shinken/satellite.py\nindex 10ca3883b..c489a27da 100644\n--- a/shinken/satellite.py\n+++ b/shinken/satellite.py\n@@ -34,8 +34,18 @@ if arbiter want it to have a new conf, satellite forgot old schedulers\n (and actions into) take new ones and do the (new) job.\n """\n \n+# Try to see if we are in an android device or not\n+is_android = True\n+try:\n+   import android\n+except ImportError:\n+   is_android = False\n+\n+if not is_android:\n+   from multiprocessing import Queue, Manager, active_children, cpu_count\n+else:\n+   from Queue import Queue\n \n-from multiprocessing import Queue, Manager, active_children, cpu_count\n import os\n import copy\n import time\n@@ -430,8 +440,10 @@ class Satellite(BaseSatellite):\n     # *0.005% : alien attack\n     # So they need to be detected, and restart if need\n     def check_and_del_zombie_workers(self):\n-        # Active children make a join with every one, useful :)\n-        active_children()\n+        # In android, we are using threads, so there is not active_children call\n+        if not is_android:\n+           # Active children make a join with every one, useful :)\n+           active_children()\n \n         w_to_del = []\n         for w in self.workers.values():\n@@ -585,6 +597,20 @@ class Satellite(BaseSatellite):\n                 sys.exit(0)\n \n \n+    # In android we got a Queue, and a manager list for others\n+    def get_returns_queue_len(self):\n+        if not is_android:\n+            return len(self.returns_queue)\n+        return self.returns_queue.qsize()\n+        \n+        \n+    # In android we got a Queue, and a manager list for others\n+    def get_returns_queue_item(self):\n+        if not is_android:\n+            return self.returns_queue.pop()\n+        return self.returns_queue.get()\n+\n+\n     def do_loop_turn(self):\n         print "Loop turn"\n         # Maybe the arbiter ask us to wait for a new conf\n@@ -628,7 +654,7 @@ class Satellite(BaseSatellite):\n                 # In workers we\'ve got actions send to queue - queue size\n                 for (i, q) in self.q_by_mod[mod].items():\n                     print \'[%d][%s][%s]Stats : Workers:%d (Queued:%d TotalReturnWait:%d)\' % \\\n-                        (sched_id, sched[\'name\'], mod, i, q.qsize(), len(self.returns_queue))\n+                        (sched_id, sched[\'name\'], mod, i, q.qsize(), self.get_returns_queue_len())\n \n \n         # Before return or get new actions, see how we manage\n@@ -663,8 +689,8 @@ class Satellite(BaseSatellite):\n \n         # Manage all messages we\'ve got in the last timeout\n         # for queue in self.return_messages:\n-        while len(self.returns_queue) != 0:\n-            self.manage_action_return(self.returns_queue.pop())\n+        while self.get_returns_queue_len() != 0:\n+            self.manage_action_return(self.get_returns_queue_item())\n             \n         # If we are passive, we do not initiate the check getting\n         # and return\n@@ -688,8 +714,14 @@ we must register our interfaces for 3 possible callers: arbiter, schedulers or b\n         # self.s = Queue() # Global Master -> Slave\n         # We can open the Queeu for fork AFTER\n         self.q_by_mod[\'fork\'] = {}\n-        self.manager = Manager()\n-        self.returns_queue = self.manager.list()\n+        \n+        # Under Android, we do not have multiprocessing lib\n+        # so use standard threads things\n+        if not is_android:\n+            self.manager = Manager()\n+            self.returns_queue = self.manager.list()\n+        else:\n+            self.returns_queue = Queue()\n \n         import socket\n         socket.setdefaulttimeout(None)\n@@ -749,14 +781,14 @@ we must register our interfaces for 3 possible callers: arbiter, schedulers or b\n         # Now the limit part, 0 mean : number of cpu of this machine :)\n         # if not available, use 4 (modern hardware)\n         self.max_workers = g_conf[\'max_workers\']\n-        if self.max_workers == 0:\n+        if self.max_workers == 0 and not is_android:\n             try:\n                 self.max_workers = cpu_count()\n             except NotImplementedError:\n                 self.max_workers =4\n             logger.log("Using max workers : %s" % self.max_workers)\n         self.min_workers = g_conf[\'min_workers\']\n-        if self.min_workers == 0:\n+        if self.min_workers == 0 and not is_android:\n             try:\n                 self.min_workers = cpu_count()\n             except NotImplementedError:\ndiff --git a/shinken/worker.py b/shinken/worker.py\nindex 0a1149021..f30fd1fd1 100644\n--- a/shinken/worker.py\n+++ b/shinken/worker.py\n@@ -28,7 +28,19 @@\n #they can die if they do not do anything (param timeout)\n \n from Queue import Empty\n-from multiprocessing import Process, Queue\n+\n+# In android, we sould use threads, not process\n+is_android = True\n+try:\n+   import android\n+except ImportError:\n+   is_android = False\n+\n+if not is_android:\n+   from multiprocessing import Process, Queue\n+else:\n+   from Queue import Queue\n+   from threading import Thread as Process\n \n import time\n import sys\n@@ -74,7 +86,9 @@ class Worker:\n     # AND close correctly the queues (input and output)\n     # each queue got a thread, so close it too....\n     def terminate(self):\n-        self._process.terminate()\n+        # We can just terminate process, not threads\n+        if not is_android:\n+            self._process.terminate()\n         self._c.close()\n         self._c.join_thread()\n         self.input_queue.close()\n@@ -164,7 +178,11 @@ class Worker:\n                 #We answer to the master\n                 #msg = Message(id=self.id, type=\'Result\', data=action)\n                 try:\n-                    self.returns_queue.append(action)#msg)\n+                    # In android, the Queue is changed\n+                    if not is_android:\n+                        self.returns_queue.append(action)\n+                    else:\n+                        self.returns_queue.put(action)\n                 except IOError , exp:\n                     print "[%d]Exiting: %s" % (self.id, exp)\n                     sys.exit(2)\n@@ -200,7 +218,9 @@ class Worker:\n     #c = Control Queue for the worker\n     def work(self, s, returns_queue, c):\n         ## restore default signal handler for the workers:\n-        signal.signal(signal.SIGTERM, signal.SIG_DFL)\n+        # but on android, we are a thread, so don\'t do it\n+        if not is_android:\n+           signal.signal(signal.SIGTERM, signal.SIG_DFL)\n         timeout = 1.0\n         self.checks = []\n         self.returns_queue = returns_queue\n', 'commit_subject': 'Add: reactionner can run under android.', 'repo_name': 'naparuba/shinken', 'committer': {'name': 'nap', 'email': 'c8e019fb2d4dd0d355144f4fa37aa9b8f4b57451@gmail.com', 'time_sec': '1319983730', 'tz_offset': '60', 'date': {'seconds': '1319983730'}}, 'commit_tree': 'f950a97b2864fa35252249a931bb6a19d5e48f27', 'commit_message': 'Add: reactionner can run under android.\n', 'language_name': 'Ruby', 'license': 'agpl-3.0', 'hunks': '@@ -28,7 +28,19 @@\n #they can die if they do not do anything (param timeout)\n \n from Queue import Empty\n-from multiprocessing import Process, Queue\n+\n+# In android, we sould use threads, not process\n+is_android = True\n+try:\n+   import android\n+except ImportError:\n+   is_android = False\n+\n+if not is_android:\n+   from multiprocessing import Process, Queue\n+else:\n+   from Queue import Queue\n+   from threading import Thread as Process\n \n import time\n import sys\n@@ -74,7 +86,9 @@ class Worker:\n     # AND close correctly the queues (input and output)\n     # each queue got a thread, so close it too....\n     def terminate(self):\n-        self._process.terminate()\n+        # We can just terminate process, not threads\n+        if not is_android:\n+            self._process.terminate()\n         self._c.close()\n         self._c.join_thread()\n         self.input_queue.close()\n@@ -164,7 +178,11 @@ class Worker:\n                 #We answer to the master\n                 #msg = Message(id=self.id, type=\'Result\', data=action)\n                 try:\n-                    self.returns_queue.append(action)#msg)\n+                    # In android, the Queue is changed\n+                    if not is_android:\n+                        self.returns_queue.append(action)\n+                    else:\n+                        self.returns_queue.put(action)\n                 except IOError , exp:\n                     print "[%d]Exiting: %s" % (self.id, exp)\n                     sys.exit(2)\n@@ -200,7 +218,9 @@ class Worker:\n     #c = Control Queue for the worker\n     def work(self, s, returns_queue, c):\n         ## restore default signal handler for the workers:\n-        signal.signal(signal.SIGTERM, signal.SIG_DFL)\n+        # but on android, we are a thread, so don\'t do it\n+        if not is_android:\n+           signal.signal(signal.SIGTERM, signal.SIG_DFL)\n         timeout = 1.0\n         self.checks = []\n         self.returns_queue = returns_queue\n', 'addition_count': 24, 'deletion_count': 4, 'src_file': 'a/shinken/worker.py', 'tgt_file': 'b/shinken/worker.py', 'file_extension': '.py', 'before_file': '#!/usr/bin/env python\n# -*- coding: utf-8 -*-\n\n# Copyright (C) 2009-2014:\n#     Gabes Jean, naparuba@gmail.com\n#     Gerhard Lausser, Gerhard.Lausser@consol.de\n#     Gregory Starck, g.starck@gmail.com\n#     Hartmut Goebel, h.goebel@goebel-consult.de\n#\n# This file is part of Shinken.\n#\n# Shinken is free software: you can redistribute it and/or modify\n# it under the terms of the GNU Affero General Public License as published by\n# the Free Software Foundation, either version 3 of the License, or\n# (at your option) any later version.\n#\n# Shinken is distributed in the hope that it will be useful,\n# but WITHOUT ANY WARRANTY; without even the implied warranty of\n# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the\n# GNU Affero General Public License for more details.\n#\n# You should have received a copy of the GNU Affero General Public License\n# along with Shinken.  If not, see <http://www.gnu.org/licenses/>.\n\nfrom Queue import Empty\n\n# In android, we should use threads, not process\n#they can die if they do not do anything (param timeout)\n\nfrom Queue import Empty\nfrom multiprocessing import Process, Queue\n\nimport time\nimport sys\n\nfrom shinken.log import logger, BrokHandler\nfrom shinken.misc.common import setproctitle\n\nclass Worker(object):\n    """This class is used for poller and reactionner to work.\n    The worker is a process launch by theses process and read Message in a Queue\n    (self.s) (slave)\n    They launch the Check and then send the result in the Queue self.m (master)\n    they can die if they do not do anything (param timeout)\n\n    """\n\n    id = 0  # None\n    _process = None\n    _mortal = None\n    _idletime = None\n    _timeout = None\n    _c = None\n\n    def __init__(self, id, s, returns_queue, processes_by_worker, mortal=True, timeout=300,\n                 max_plugins_output_length=8192, target=None, loaded_into=\'unknown\',\n                 http_daemon=None):\n        self.id = self.__class__.id\n        self.__class__.id += 1\n\n        self._mortal = mortal\n        self._idletime = 0\n        self._timeout = timeout\n        self.s = None\n        self.processes_by_worker = processes_by_worker\n        self._c = Queue()  # Private Control queue for the Worker\n        # By default, take our own code\n        if target is None:\n            target = self.work\n        self._process = Process(target=self._prework, args=(target, s, returns_queue, self._c))\n        self.returns_queue = returns_queue\n        self.max_plugins_output_length = max_plugins_output_length\n        self.i_am_dying = False\n    # AND close correctly the queues (input and output)\n    # each queue got a thread, so close it too....\n    def terminate(self):\n        self._process.terminate()\n        self._c.close()\n        self._c.join_thread()\n        self.input_queue.close()\n            if isinstance(handler, BrokHandler):\n                logger.info("Cleaning BrokHandler %r from logger.handlers..", handler)\n                logger.removeHandler(handler)\n        real_work(*args)\n\n    def is_mortal(self):\n        return self._mortal\n\n\n    def start(self):\n        self._process.start()\n\n\n    # Kill the background process\n    # AND close correctly the queues (input and output)\n    # each queue got a thread, so close it too....\n    def terminate(self):\n        # We can just terminate process, not threads\n        if not is_android:\n            self._process.terminate()\n        # Is we are with a Manager() way\n        # there should be not such functions\n        if hasattr(self._c, \'close\'):\n            self._c.close()\n            self._c.join_thread()\n        if hasattr(self.s, \'close\'):\n            self.s.close()\n            self.s.join_thread()\n\n    def join(self, timeout=None):\n        self._process.join(timeout)\n\n    def is_alive(self):\n        return self._process.is_alive()\n\n    def is_killable(self):\n        return self._mortal and self._idletime > self._timeout\n\n    def add_idletime(self, time):\n        self._idletime = self._idletime + time\n\n    def reset_idle(self):\n        self._idletime = 0\n\n    def send_message(self, msg):\n        self._c.put(msg)\n\n    # A zombie is immortal, so kill not be kill anymore\n    def set_zombie(self):\n        self._mortal = False\n\n    # Get new checks if less than nb_checks_max\n    # If no new checks got and no check in queue,\n    # sleep for 1 sec\n    # REF: doc/shinken-action-queues.png (3)\n    def get_new_checks(self):\n        try:\n            while(len(self.checks) < self.processes_by_worker):\n                # print "I", self.id, "wait for a message"\n                msg = self.s.get(block=False)\n                if msg is not None:\n                    self.checks.append(msg.get_data())\n                # print "I", self.id, "I\'ve got a message!"\n        except Empty, exp:\n            if len(self.checks) == 0:\n                self._idletime = self._idletime + 1\n                time.sleep(1)\n        # Maybe the Queue() is not available, if so, just return\n        # get back to work :)\n        except IOError, exp:\n            return\n\n\n    # Launch checks that are in status\n    # REF: doc/shinken-action-queues.png (4)\n    def launch_new_checks(self):\n        # queue\n        for chk in self.checks:\n            if chk.status == \'queue\':\n                self._idletime = 0\n                r = chk.execute()\n                # Maybe we got a true big problem in the\n                # action launching\n                #We answer to the master\n                #msg = Message(id=self.id, type=\'Result\', data=action)\n                try:\n                    self.returns_queue.append(action)#msg)\n                except IOError , exp:\n                    print "[%d]Exiting: %s" % (self.id, exp)\n                    sys.exit(2)\n        wait_time = 1\n        now = time.time()\n        for action in self.checks:\n            if action.status == \'launched\' and action.last_poll < now - action.wait_time:\n                action.check_finished(self.max_plugins_output_length)\n                wait_time = min(wait_time, action.wait_time)\n                # If action done, we can launch a new one\n            if action.status in (\'done\', \'timeout\'):\n                to_del.append(action)\n                # We answer to the master\n                # msg = Message(id=self.id, type=\'Result\', data=action)\n                try:\n                    self.returns_queue.put(action)\n                except IOError, exp:\n                    logger.error("[%d] Exiting: %s", self.id, exp)\n                    sys.exit(2)\n\n        # Little sleep\n        self.wait_time = wait_time\n\n        for chk in to_del:\n            self.checks.remove(chk)\n\n        # Little sleep\n        time.sleep(wait_time)\n\n    # Check if our system time change. If so, change our\n    def check_for_system_time_change(self):\n        now = time.time()\n    #c = Control Queue for the worker\n    def work(self, s, returns_queue, c):\n        ## restore default signal handler for the workers:\n        signal.signal(signal.SIGTERM, signal.SIG_DFL)\n        timeout = 1.0\n        self.checks = []\n        self.returns_queue = returns_queue\n            return 0\n\n\n    # Wrapper function for work in order to catch the exception\n    # to see the real work, look at do_work\n    def work(self, s, returns_queue, c):\n        try:\n            self.do_work(s, returns_queue, c)\n        # Catch any exception, try to print it and exit anyway\n        except Exception, exp:\n            output = cStringIO.StringIO()\n            traceback.print_exc(file=output)\n            logger.error("Worker \'%d\' exit with an unmanaged exception : %s",\n                         self.id, output.getvalue())\n            output.close()\n            # Ok I die now\n            raise\n\n\n    # id = id of the worker\n    # s = Global Queue Master->Slave\n    # m = Queue Slave->Master\n    # return_queue = queue managed by manager\n    # c = Control Queue for the worker\n    def do_work(self, s, returns_queue, c):\n        # restore default signal handler for the workers:\n        # but on android, we are a thread, so don\'t do it\n        if not is_android:\n            signal.signal(signal.SIGTERM, signal.SIG_DFL)\n\n        self.set_proctitle()\n\n        print "I STOP THE http_daemon", self.http_daemon\n        if self.http_daemon:\n            self.http_daemon.shutdown()\n\n        timeout = 1.0\n        self.checks = []\n        self.returns_queue = returns_queue\n        self.s = s\n        self.t_each_loop = time.time()\n        while True:\n            begin = time.time()\n            msg = None\n            cmsg = None\n\n            # If we are dying (big problem!) we do not\n            # take new jobs, we just finished the current one\n            if not self.i_am_dying:\n                # REF: doc/shinken-action-queues.png (3)\n                self.get_new_checks()\n                # REF: doc/shinken-action-queues.png (4)\n                self.launch_new_checks()\n            # REF: doc/shinken-action-queues.png (5)\n            self.manage_finished_checks()\n\n            # Now get order from master\n            try:\n                cmsg = c.get(block=False)\n                if cmsg.get_type() == \'Die\':\n                    logger.debug("[%d] Dad say we are dying...", self.id)\n                    break\n            except Exception:\n                pass\n\n            # Look if we are dying, and if we finish all current checks\n            # if so, we really die, our master poller will launch a new\n            # worker because we were too weak to manage our job :(\n            if len(self.checks) == 0 and self.i_am_dying:\n                logger.warning("[%d] I DIE because I cannot do my job as I should"\n                               "(too many open files?)... forgot me please.", self.id)\n                break\n\n            # Manage a possible time change (our avant will be change with the diff)\n            diff = self.check_for_system_time_change()\n            begin += diff\n\n            timeout -= time.time() - begin\n            if timeout < 0:\n                timeout = 1.0\n\n    def set_proctitle(self):\n        setproctitle("shinken-%s worker" % self.loaded_into)\n', 'message': 'Add: reactionner can run under android.\n', 'author_name': 'nap', 'lang': 'py'}