浏览代码

celeryd now includes the node name of the worker in ps listings

Ask Solem 14 年之前
父节点
当前提交
50e50dfd84
共有 2 个文件被更改,包括 7 次插入4 次删除
  1. 3 2
      celery/worker/__init__.py
  2. 4 2
      celery/worker/job.py

+ 3 - 2
celery/worker/__init__.py

@@ -30,7 +30,7 @@ WORKER_SIGRESET = frozenset(["SIGTERM",
 WORKER_SIGIGNORE = frozenset(["SIGINT"])
 
 
-def process_initializer():
+def process_initializer(hostname):
     """Initializes the process so it can be used to process tasks.
 
     Used for multiprocessing environments.
@@ -38,7 +38,7 @@ def process_initializer():
     """
     map(platforms.reset_signal, WORKER_SIGRESET)
     map(platforms.ignore_signal, WORKER_SIGIGNORE)
-    platforms.set_mp_process_title("celeryd")
+    platforms.set_mp_process_title("celeryd", hostname=hostname)
 
     # This is for windows and other platforms not supporting
     # fork(). Note that init_worker makes sure it's only
@@ -161,6 +161,7 @@ class WorkController(object):
         self.pool = instantiate(pool_cls, self.concurrency,
                                 logger=self.logger,
                                 initializer=process_initializer,
+                                initargs=(self.hostname, ),
                                 maxtasksperchild=self.max_tasks_per_child,
                                 timeout=self.task_time_limit,
                                 soft_timeout=self.task_soft_time_limit,

+ 4 - 2
celery/worker/job.py

@@ -140,11 +140,13 @@ def execute_and_trace(task_name, *args, **kwargs):
         >>> WorkerTaskTrace(task_name, *args, **kwargs).execute_safe()
 
     """
-    platforms.set_mp_process_title("celeryd", info=task_name)
+    hostname = kwargs.get("hostname")
+    platforms.set_mp_process_title("celeryd", info=task_name,
+                                   hostname=hostname)
     try:
         return WorkerTaskTrace(task_name, *args, **kwargs).execute_safe()
     finally:
-        platforms.set_mp_process_title("celeryd")
+        platforms.set_mp_process_title("celeryd", hostname=hostname)
 
 
 class TaskRequest(object):