Setting thread and process names properly through prctl

This commit is contained in:
Fabio Manganiello 2019-01-10 23:45:13 +01:00
parent 56c7258c74
commit fbf3600e91
12 changed files with 42 additions and 11 deletions

View file

@ -24,6 +24,7 @@ from .message.event import Event, StopEvent
from .message.event.application import ApplicationStartedEvent, ApplicationStoppedEvent from .message.event.application import ApplicationStartedEvent, ApplicationStoppedEvent
from .message.request import Request from .message.request import Request
from .message.response import Response from .message.response import Response
from .utils import set_thread_name
__author__ = 'Fabio Manganiello <blacklight86@gmail.com>' __author__ = 'Fabio Manganiello <blacklight86@gmail.com>'
@ -138,6 +139,7 @@ class Daemon:
def start(self): def start(self):
""" Start the daemon """ """ Start the daemon """
set_thread_name('platypush')
print('---- Starting platypush v.{}'.format(__version__)) print('---- Starting platypush v.{}'.format(__version__))
redis_conf = Config.get('backend.redis') redis_conf = Config.get('backend.redis')

View file

@ -18,7 +18,7 @@ from platypush.message import Message
from platypush.message.event import Event, StopEvent from platypush.message.event import Event, StopEvent
from platypush.message.request import Request from platypush.message.request import Request
from platypush.message.response import Response from platypush.message.response import Response
from platypush.utils import get_redis_queue_name_by_message from platypush.utils import get_redis_queue_name_by_message, set_thread_name
class Backend(Thread): class Backend(Thread):
@ -44,7 +44,8 @@ class Backend(Thread):
:type kwargs: dict :type kwargs: dict
""" """
super().__init__(name='PlatypushBackend_' + self.__class__.__name__) self._thread_name = 'pp-' + self.__class__.__name__
super().__init__(name=self._thread_name)
# If no bus is specified, create an internal queue where # If no bus is specified, create an internal queue where
# the received messages will be pushed # the received messages will be pushed
@ -216,6 +217,7 @@ class Backend(Thread):
def run(self): def run(self):
""" Starts the backend thread. To be implemented in the derived classes """ """ Starts the backend thread. To be implemented in the derived classes """
self.thread_id = threading.get_ident() self.thread_id = threading.get_ident()
set_thread_name(self._thread_name)
def on_stop(self): def on_stop(self):
""" Callback invoked when the process stops """ """ Callback invoked when the process stops """

View file

@ -20,7 +20,7 @@ from platypush.message import Message
from platypush.message.event import Event, StopEvent from platypush.message.event import Event, StopEvent
from platypush.message.event.web.widget import WidgetUpdateEvent from platypush.message.event.web.widget import WidgetUpdateEvent
from platypush.message.request import Request from platypush.message.request import Request
from platypush.utils import get_ssl_server_context from platypush.utils import get_ssl_server_context, set_thread_name
from .. import Backend from .. import Backend
@ -232,6 +232,8 @@ class HttpBackend(Backend):
def webserver(self): def webserver(self):
""" Web server main process """ """ Web server main process """
set_thread_name('pp-web-server')
basedir = os.path.dirname(inspect.getfile(self.__class__)) basedir = os.path.dirname(inspect.getfile(self.__class__))
template_dir = os.path.join(basedir, 'templates') template_dir = os.path.join(basedir, 'templates')
app = Flask(__name__, template_folder=template_dir) app = Flask(__name__, template_folder=template_dir)
@ -416,6 +418,7 @@ class HttpBackend(Backend):
def websocket(self): def websocket(self):
""" Websocket main server """ """ Websocket main server """
import websockets import websockets
set_thread_name('pp-websocket-server')
async def register_websocket(websocket, path): async def register_websocket(websocket, path):
address = websocket.remote_address[0] if websocket.remote_address \ address = websocket.remote_address[0] if websocket.remote_address \
@ -456,7 +459,7 @@ class HttpBackend(Backend):
webserver = self.webserver() webserver = self.webserver()
self.server_proc = Process(target=webserver.run, self.server_proc = Process(target=webserver.run,
name='PlatypushWebServer', name='pp-web-server',
kwargs=kwargs) kwargs=kwargs)
self.server_proc.start() self.server_proc.start()

View file

@ -11,6 +11,7 @@ from frozendict import frozendict
from threading import Thread from threading import Thread
from platypush.message.event.http import HttpEvent from platypush.message.event.http import HttpEvent
from platypush.utils import set_thread_name
class HttpRequest(object): class HttpRequest(object):
poll_seconds = 60 poll_seconds = 60
@ -53,6 +54,7 @@ class HttpRequest(object):
def execute(self): def execute(self):
def _thread_func(): def _thread_func():
set_thread_name('pp-http-poll')
is_first_call = self.last_request_timestamp == 0 is_first_call = self.last_request_timestamp == 0
self.last_request_timestamp = time.time() self.last_request_timestamp = time.time()
@ -77,7 +79,7 @@ class HttpRequest(object):
self.logger.warning('Encountered an error while retrieving {}: {}'. self.logger.warning('Encountered an error while retrieving {}: {}'.
format(self.args.url, str(e))) format(self.args.url, str(e)))
Thread(target=_thread_func, name='PlatypushHttpPoll').start() Thread(target=_thread_func, name='pp-http-poll').start()
def get_new_items(self, response): def get_new_items(self, response):

View file

@ -9,6 +9,7 @@ from platypush.backend import Backend
from platypush.context import get_plugin from platypush.context import get_plugin
from platypush.message import Message from platypush.message import Message
from platypush.message.request import Request from platypush.message.request import Request
from platypush.utils import set_thread_name
class MqttBackend(Backend): class MqttBackend(Backend):
@ -97,6 +98,7 @@ class MqttBackend(Backend):
def on_message(client, userdata, msg): def on_message(client, userdata, msg):
def response_thread(msg): def response_thread(msg):
set_thread_name('pp-mqtt-processor')
response = self.get_message_response(msg) response = self.get_message_response(msg)
if not response: if not response:
return return
@ -122,7 +124,7 @@ class MqttBackend(Backend):
if isinstance(msg, Request): if isinstance(msg, Request):
threading.Thread(target=response_thread, threading.Thread(target=response_thread,
name='PlatypushMQTTResponseProcessor', name='pp-mqtt-processor',
args=(msg,)).start() args=(msg,)).start()
super().run() super().run()

View file

@ -5,6 +5,7 @@ import time
from platypush.backend import Backend from platypush.backend import Backend
from platypush.context import get_plugin from platypush.context import get_plugin
from platypush.utils import set_thread_name
from platypush.message.event.music.snapcast import ClientVolumeChangeEvent, \ from platypush.message.event.music.snapcast import ClientVolumeChangeEvent, \
GroupMuteChangeEvent, ClientConnectedEvent, ClientDisconnectedEvent, \ GroupMuteChangeEvent, ClientConnectedEvent, ClientDisconnectedEvent, \
ClientLatencyChangeEvent, ClientNameChangeEvent, GroupStreamChangeEvent, \ ClientLatencyChangeEvent, ClientNameChangeEvent, GroupStreamChangeEvent, \
@ -129,6 +130,7 @@ class MusicSnapcastBackend(Backend):
def _client(self, host, port): def _client(self, host, port):
def _thread(): def _thread():
set_thread_name('pp-snapcast-' + host)
status = None status = None
try: try:
@ -199,7 +201,7 @@ class MusicSnapcastBackend(Backend):
port = self.ports[i] port = self.ports[i]
self._threads[host] = threading.Thread( self._threads[host] = threading.Thread(
target=self._client(host, port), target=self._client(host, port),
name='PlatypushSnapcastWorker' name='pp-snapcast'
) )
self._threads[host].start() self._threads[host].start()

View file

@ -5,6 +5,7 @@ import threading
from platypush.backend import Backend from platypush.backend import Backend
from platypush.message import Message from platypush.message import Message
from platypush.message.request import Request from platypush.message.request import Request
from platypush.utils import set_thread_name
class TcpBackend(Backend): class TcpBackend(Backend):
@ -83,12 +84,13 @@ class TcpBackend(Backend):
sock.send(str(response).encode()) sock.send(str(response).encode())
def _f_wrapper(): def _f_wrapper():
set_thread_name('pp-tcp-listen')
try: try:
_f() _f()
finally: finally:
sock.close() sock.close()
threading.Thread(target=_f_wrapper, name='PlatypushTCPListener').start() threading.Thread(target=_f_wrapper, name='pp-tcp-listen').start()
def run(self): def run(self):
super().run() super().run()

View file

@ -8,7 +8,7 @@ from platypush.config import Config
from platypush.message.event import Event, EventMatchResult from platypush.message.event import Event, EventMatchResult
from platypush.message.request import Request from platypush.message.request import Request
from platypush.procedure import Procedure from platypush.procedure import Procedure
from platypush.utils import get_event_class_by_type from platypush.utils import get_event_class_by_type, set_thread_name
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -150,6 +150,7 @@ class EventHook(object):
runs the hook actions if the condition is met """ runs the hook actions if the condition is met """
def _thread_func(result): def _thread_func(result):
set_thread_name('pp-event-' + self.name)
self.actions.execute(event=event, **result.parsed_args) self.actions.execute(event=event, **result.parsed_args)
result = self.matches_event(event) result = self.matches_event(event)
@ -158,7 +159,7 @@ class EventHook(object):
if result.is_match: if result.is_match:
logger.info('Running hook {} triggered by an event'.format(self.name)) logger.info('Running hook {} triggered by an event'.format(self.name))
threading.Thread(target=_thread_func, threading.Thread(target=_thread_func,
name='PlatypushEventHook_' + self.name, name='pp-event-' + self.name,
args=(result,)).start() args=(result,)).start()

View file

@ -11,6 +11,7 @@ from phue import Bridge
from platypush.context import get_backend from platypush.context import get_backend
from platypush.plugins import action from platypush.plugins import action
from platypush.plugins.light import LightPlugin from platypush.plugins.light import LightPlugin
from platypush.utils import set_thread_name
class LightHuePlugin(LightPlugin): class LightHuePlugin(LightPlugin):
@ -666,6 +667,7 @@ class LightHuePlugin(LightPlugin):
def _animate_thread(lights): def _animate_thread(lights):
set_thread_name('pp-hue-animate')
self.logger.info('Starting {} animation'.format( self.logger.info('Starting {} animation'.format(
animation, (lights or groups))) animation, (lights or groups)))
@ -720,7 +722,7 @@ class LightHuePlugin(LightPlugin):
self.stop_animation() self.stop_animation()
self.animation_thread = Thread(target=_animate_thread, self.animation_thread = Thread(target=_animate_thread,
name='PlatypushLightHueAnimate', name='pp-hue-animate',
args=(lights,)) args=(lights,))
self.animation_thread.start() self.animation_thread.start()

View file

@ -163,6 +163,15 @@ def get_ssl_client_context(ssl_cert=None, ssl_key=None, ssl_cafile=None,
ssl_cert=ssl_cert, ssl_key=ssl_key, ssl_cert=ssl_cert, ssl_key=ssl_key,
ssl_cafile=ssl_cafile, ssl_capath=ssl_capath) ssl_cafile=ssl_cafile, ssl_capath=ssl_capath)
def set_thread_name(name):
global logger
try:
import prctl
prctl.set_name(name)
except ImportError:
logger.debug('Unable to set thread name: prctl module is missing')
# vim:sw=4:ts=4:et: # vim:sw=4:ts=4:et:

View file

@ -8,6 +8,9 @@
# YAML configuration support # YAML configuration support
pyyaml pyyaml
# Support for setting thread/process name
prctl
# Apache Kafka backend support # Apache Kafka backend support
kafka-python kafka-python

View file

@ -63,6 +63,7 @@ setup(
'pyyaml', 'pyyaml',
'redis', 'redis',
'requests', 'requests',
'prctl',
], ],
extras_require = { extras_require = {
'Support for Apache Kafka backend': ['kafka-python'], 'Support for Apache Kafka backend': ['kafka-python'],