oslo-service-reproducer

oslo-service-reproducer.py

#!/usr/bin/env python
#

import logging
from oslo_service import service
from oslo_config import cfg
import pdb
import traceback
import os
import time

LOG = logging.getLogger()
logging.basicConfig(level=logging.DEBUG,
                    format='%(asctime)s %(thread)d %(message)s')

from threading import Lock, Event, current_thread, Thread

class TaskServer3(service.Service):
    def __init__(self):
        super(TaskServer3, self).__init__()
        self._server = Thread(target=self._test_service)
        self._done = False
        LOG.info("TaskServer Created (%s) [%s] %s", os.getpid(),
                 current_thread(), self)

    def _test_service(self):
        # background thread
        while not self._done:
            print("Hi %d" % os.getpid())
            time.sleep(1.0)
        print("Bye %d" % os.getpid())

    def start(self):
        LOG.info("TaskServer start() called (%s) [%s] %s", os.getpid(),
                 current_thread(), self)
        self._server.start()
        #
        # Issue: if an exception is raised _after_ the server thread is
        # spawned, then Service.stop() will not be invoked if the
        # ProcessLauncher.stop() method is called.
        # HOWEVER if ProcessLauncher.wait() is called, then Service.wait() _is_
        # called, which leads to a hang (wait() called without stop() called)
        #
        raise Exception("Exception thrown!!!")
        LOG.info("TaskServer server start done (%s) [%s] %s", os.getpid(),
                 current_thread(), self)

    def stop(self):
        LOG.info("TaskServer stop() called (%s) [%s] %s", os.getpid(),
                 current_thread(), self)
        self._done = True
        super(TaskServer3, self).stop(graceful=True)
        LOG.info("TaskServer stop done (%s) [%s] %s", os.getpid(),
                 current_thread(), self)

    def reset(self):
        LOG.info("TaskServer reset() called (%s) %s", os.getpid(), self)
        super(TaskServer3, self).reset()
        LOG.info("TaskServer reset() done (%s) %s", os.getpid(), self)

    def wait(self):
        LOG.info("TaskServer wait called (%s) [%s] %s", os.getpid(),
                 current_thread(), self)
        if self._server and self._server.is_alive():
            self._server.join()
        else:
            LOG.info("TaskServer wait() before server started (%s) [%s] %s",
                     os.getpid(),
                     current_thread(), self)
        super(TaskServer3, self).wait()
        LOG.info("TaskServer wait done (%s) [%s] %s", os.getpid(),
                 current_thread(), self)


def stop_me(ts):
    LOG.info("stop_me called... sleeping...")
    time.sleep(10)
    LOG.info("stop_me stopping...")
    ts.stop()
    LOG.info("stop_me done.")


def main():

    W = 2  # control the # of workers

    LOG.info("Creating a launcher %s", os.getpid())
    ll = service.ProcessLauncher(cfg.CONF, restart_method='mutate')
    LOG.info("Launcher Created %s", os.getpid())
    ll.launch_service(TaskServer3(), workers=W)
    LOG.info("Launcher Launched %s", os.getpid())
    LOG.info("Launcher WAITING.... %s", os.getpid())
    Thread(target=stop_me, args=(ll,)).start()
    ll.wait()
    LOG.info("Launcher WAITING Done!  %s", os.getpid())
    LOG.info("Main thread exiting %s", os.getpid())

if __name__ == '__main__':
    main()