Ocata Neutron代码分析——Neutron RPC启动过程分析

Posted 2020-10-18

RPC启动跟Neutron API的启动在同一个函数中执行，neutron.server.wsgi_eventlet.py中的eventlet_wsgi_server。

def eventlet_wsgi_server():
    neutron_api = service.serve_wsgi(service.NeutronApiService)
    start_api_and_rpc_workers(neutron_api)

def start_api_and_rpc_workers(neutron_api):
    try:
        worker_launcher = service.start_all_workers()

        pool = eventlet.GreenPool()
        api_thread = pool.spawn(neutron_api.wait)
        plugin_workers_thread = pool.spawn(worker_launcher.wait)

        # api and other workers should die together. When one dies,
        # kill the other.
        api_thread.link(lambda gt: plugin_workers_thread.kill())
        plugin_workers_thread.link(lambda gt: api_thread.kill())

        pool.waitall()
    except NotImplementedError:
        LOG.info(_LI("RPC was already started in parent process by "
                     "plugin."))

        neutron_api.wait()

start_api_and_rpc_workers函数中使用start_all_workers函数来启动RPC相关的workers，返回worker_launcher对象（实际上是oslo_service.service::ProcessLauncher的实例）。接着，主进程使用GreenPool分别spawn出neutron_api和worker_launcher的wait函数，并调用waitall函数来等待这两个GreenThread结束。其中的两个link函数是为了保证其中一个服务挂掉，另外一个服务也要随之结束。

下面重点分析neutron.service的start_all_workers函数：

def start_all_workers():
    workers = _get_rpc_workers() + _get_plugins_workers()
    return _start_workers(workers)

start_all_workers主要完成获取workers并启动的工作，这里的workers分为两类：

• _get_rpc_workers返回每个plugin（包括core plugin和service plugin）所必须的rpc worker；
• _get_plugins_workers返回每一种（跟第一点中的每个plugin不同）plugin中用于实现自己特殊需求（如果有）的rpc workers。

先分析_get_rpc_workers：

def _get_rpc_workers():
    plugin = directory.get_plugin()                         # 获取core plugin
    service_plugins = directory.get_plugins().values()      # 获取plugin，包括core plugin和service plugin

    if cfg.CONF.rpc_workers < 1:                            # 检查配置文件中的rpc_workers的值，必须大于或等于1
        cfg.CONF.set_override(‘rpc_workers‘, 1)

    # If 0 < rpc_workers then start_rpc_listeners would be called in a
    # subprocess and we cannot simply catch the NotImplementedError.  It is
    # simpler to check this up front by testing whether the plugin supports
    # multiple RPC workers.
    if not plugin.rpc_workers_supported():
        LOG.debug("Active plugin doesn‘t implement start_rpc_listeners")
        if 0 < cfg.CONF.rpc_workers:
            LOG.error(_LE("‘rpc_workers = %d‘ ignored because "
                          "start_rpc_listeners is not implemented."),
                      cfg.CONF.rpc_workers)
        raise NotImplementedError()

    # passing service plugins only, because core plugin is among them
    rpc_workers = [RpcWorker(service_plugins,
                             worker_process_count=cfg.CONF.rpc_workers)]

    if (cfg.CONF.rpc_state_report_workers > 0 and
            plugin.rpc_state_report_workers_supported()):
        rpc_workers.append(
            RpcReportsWorker(
                [plugin],
                worker_process_count=cfg.CONF.rpc_state_report_workers
            )
        )
    return rpc_workers

plugin.rpc_workers_supported实际上是检查core plugin是否实现了start_rpc_listeners方法。

接着，实例化了RpcWorker类，这是这个函数中最关键的步骤。RpcWorker类和Neutron API部分的WorkerService类一样，是继承于neutron_worker.NeutronWorker类的子类，同样实现了start函数。

最后，判断配置文件中的rpc_state_report_workers和core plugin是否支持对rpc state的report。如果支持，则再启动rpc_state_report_workers这么多个子进程来运行 RpcReportsWorker。这个Worker是RpcWorker的子类，只是其start_listeners_method跟RpcWorker不同。

class RpcReportsWorker(RpcWorker):
    start_listeners_method = ‘start_rpc_state_reports_listener‘

 下面分析RpcWorker的构造函数和其start函数：

class RpcWorker(neutron_worker.NeutronWorker):
    """Wraps a worker to be handled by ProcessLauncher"""
    start_listeners_method = ‘start_rpc_listeners‘

    def __init__(self, plugins, worker_process_count=1):                         # 构造函数只是进行了变量的简单赋值
        super(RpcWorker, self).__init__(
            worker_process_count=worker_process_count
        )

        self._plugins = plugins
        self._servers = []

    def start(self):
        super(RpcWorker, self).start()
        for plugin in self._plugins:
            if hasattr(plugin, self.start_listeners_method):
                try:
                    servers = getattr(plugin, self.start_listeners_method)()
                except NotImplementedError:
                    continue
                self._servers.extend(servers)

start函数会遍历所有的plugins（包括core plugin和service plugins），查看各个plugin是否实现了start_listeners_method（即start_rpc_listeners）方法，如果实现了则调用之，如果没有就跳过。

这就是RpcWorker的作用。各个plugin的start_rpc_listeners方法中就完成了rpc的功能，主要是通过消费特定名称的mq队列消息来提供服务。

下面分析_get_plugins_workers函数：

def _get_plugins_workers():
    # NOTE(twilson) get_plugins also returns the core plugin
    plugins = directory.get_unique_plugins()

    # TODO(twilson) Instead of defaulting here, come up with a good way to
    # share a common get_workers default between NeutronPluginBaseV2 and
    # ServicePluginBase
    return [
        plugin_worker
        for plugin in plugins if hasattr(plugin, ‘get_workers‘)
        for plugin_worker in plugin.get_workers()
    ]

_get_plugins_workers函数检查了每种plugin（不是每一个）中是否实现了get_workers方法，并将该方法返回的所有workers收集后返回。这里的get_workesr方法返回plugin用于实现自己的特殊需求或提供个性化服务的workers。

两个收集workers的函数执行完后，下面就是启动各个workers的函数：

def start_all_workers():
    workers = _get_rpc_workers() + _get_plugins_workers()
    return _start_workers(workers)

def _start_workers(workers):
    process_workers = [
        plugin_worker for plugin_worker in workers
        if plugin_worker.worker_process_count > 0
    ]                                                                 # 筛选出worker_process_count大于0的workers

    try:
        if process_workers:                                           # 如果存在worker_process_count大于0的workers
            worker_launcher = common_service.ProcessLauncher(
                cfg.CONF, wait_interval=1.0
            )

            # add extra process worker and spawn there all workers with
            # worker_process_count == 0
            thread_workers = [
                plugin_worker for plugin_worker in workers
                if plugin_worker.worker_process_count < 1
            ]
            if thread_workers:
                process_workers.append(
                    AllServicesNeutronWorker(thread_workers)
                )

            # dispose the whole pool before os.fork, otherwise there will
            # be shared DB connections in child processes which may cause
            # DB errors.
            session.context_manager.dispose_pool()

            for worker in process_workers:
                worker_launcher.launch_service(worker,
                                               worker.worker_process_count)
        else:                                                          # 如果workers中的所有worker的worker_process_count都为0
            worker_launcher = common_service.ServiceLauncher(cfg.CONF)
            for worker in workers:
                worker_launcher.launch_service(worker)
        return worker_launcher
    except Exception:
        with excutils.save_and_reraise_exception():
            LOG.exception(_LE(‘Unrecoverable error: please check log for ‘
                              ‘details.‘))

_start_workers首先判断workers中是否存在worker_process_count大于0的workers。

• 如果不存在（else分支）：实例化ServiceLauncher，并调用其launch_service方法依次在当前的进程中启动各个worker；
• 如果存在（if分支）：
  • 首先实例化ProcessLauncher；

  • 接着对workers中可能存在的worker_process_count为0的worker进行处理，将这样的worker形成thread_workers列表，并将这些worker作为参数实例化AllServicesNeutronWorker（AllServicesNeutronWorker也是NeutronWorker的子类并重写了start方法，直接调用Launcher.launch_service启动服务）。最后将AllServicesNeutronWorker实例append到process_workers列表中；

  • 最后，调用ProcessLauncher.launch_service方法依次启动各个worker（与Neutron API的启动一样）。

ServiceLauncher和ProcessLauncher均实现了launch_service方法。但ServiceLauncher是Launcher的子类，而ProcessLauncher的父类是Object。其launch_service存在的区别是：

• ServiceLauncher是将任务放到greenthread中运行；
• ProcessLauncher是将任务放到os fork出来的子进程中运行。