#!/usr/bin/python2

# PBR Generated from u'console_scripts'

import sys

from cinder.cmd.volume import main

if __name__ == "__main__":

    sys.exit(main())

def main():

    objects.register_all()   # import cinder/objects目录下的所有模块

    gmr_opts.set_defaults(CONF)  # oslo_reports模块，用于生成错误报告，如内存泄漏等

    CONF(sys.argv[1:], project='cinder',

         version=version.version_string())

    logging.setup(CONF, "cinder")

    python_logging.captureWarnings(True)

    priv_context.init(root_helper=shlex.split(utils.get_root_helper()))  # oslo_privsep，service的执行权限设置

    utils.monkey_patch()   # monkey-patch，替换库比如socket、thread等，不改变import的行为，改变的是操作的指向函数

    gmr.TextGuruMeditation.setup_autorun(version, conf=CONF)

    launcher = service.get_launcher()----------------第一步

    LOG = logging.getLogger(__name__)

    service_started = False

    if CONF.enabled_backends:-----------在cinder.conf文件中使能的多存储后端

        for backend in filter(None, CONF.enabled_backends):-------循环读取配置文件中enabled_backends的值

            CONF.register_opt(host_opt, group=backend)

            backend_host = getattr(CONF, backend).backend_host

            host = "%s@%s" % (backend_host or CONF.host, backend)------生成cinder-volume host的名字，使用命令行cinder service-list的时候，查到的名字

            # We also want to set cluster to None on empty strings, and we

            # ignore leading and trailing spaces.

            cluster = CONF.cluster and CONF.cluster.strip()

            cluster = (cluster or None) and '%s@%s' % (cluster, backend)

            try:

                server = service.Service.create(host=host,------------------s1 步为每一个backend生成一个服务对象

                                                service_name=backend,

                                                binary='cinder-volume',

                                                coordination=True,

                                                cluster=cluster)

            except Exception:

                msg = _('Volume service %s failed to start.') % host

                LOG.exception(msg)

            else:--------如果没有异常发生，那么走这个分支

                # Dispose of the whole DB connection pool here before

                # starting another process.  Otherwise we run into cases where

                # child processes share DB connections which results in errors.

                session.dispose_engine()

                launcher.launch_service(server)--------------第二步，调用launcher里面的launch_service

                service_started = True

    else:

        LOG.error(_LE('Configuration for cinder-volume does not specify '

                      '"enabled_backends". Using DEFAULT section to configure '

                      'drivers is not supported since Ocata.'))

    if not service_started:

        msg = _('No volume service(s) started successfully, terminating.')

        LOG.error(msg)

        sys.exit(1)

    launcher.wait()

cinder/service.py:class Service(service.Service)里面的类方法

    @classmethod

    def create(cls, host=None, binary=None, topic=None, manager=None,

               report_interval=None, periodic_interval=None,

               periodic_fuzzy_delay=None, service_name=None,

               coordination=False, cluster=None):

        """Instantiates class and passes back application object.

        :param host: defaults to CONF.host

        :param binary: defaults to basename of executable

        :param topic: defaults to bin_name - 'cinder-' part

        :param manager: defaults to CONF.<topic>_manager

        :param report_interval: defaults to CONF.report_interval

        :param periodic_interval: defaults to CONF.periodic_interval

        :param periodic_fuzzy_delay: defaults to CONF.periodic_fuzzy_delay

        :param cluster: Defaults to None, as only some services will have it

        """

        if not host:

            host = CONF.host

        if not binary:

            binary = os.path.basename(inspect.stack()[-1][1])

        if not topic:

            topic = binary

        if not manager:

            subtopic = topic.rpartition('cinder-')[2]

            manager = CONF.get('%s_manager' % subtopic, None)

        if report_interval is None:

            report_interval = CONF.report_interval-------------读取配置文件中的上报周期值

        if periodic_interval is None:

            periodic_interval = CONF.periodic_interval---------读取配置文件中的轮训周期值

        if periodic_fuzzy_delay is None:

            periodic_fuzzy_delay = CONF.periodic_fuzzy_delay

        service_obj = cls(host, binary, topic, manager,--------调用类初始化一个服务对象

                          report_interval=report_interval,

                          periodic_interval=periodic_interval,

                          periodic_fuzzy_delay=periodic_fuzzy_delay,

                          service_name=service_name,

                          coordination=coordination,

                          cluster=cluster)

        return service_obj

cinder/cmd/volume.py

from cinder import service

launcher = service.get_launcher()

cinder/service.py

from oslo_service import service

def get_launcher():

    # Note(lpetrut): ProcessLauncher uses green pipes which fail on Windows

    # due to missing support of non-blocking I/O pipes. For this reason, the

    # service must be spawned differently on Windows, using the ServiceLauncher

    # class instead.

    if os.name == 'nt':

        return Launcher()--------------------windows操作系统

    else:

        return process_launcher()--------------Linux操作系统

def process_launcher():

    return service.ProcessLauncher(CONF)	

oslo_service/service.py

class ProcessLauncher(object):

    """Launch a service with a given number of workers."""

    def __init__(self, conf, wait_interval=0.01, restart_method='reload'):

        """Constructor.

        :param conf: an instance of ConfigOpts

        :param wait_interval: The interval to sleep for between checks

                              of child process exit.

        :param restart_method: If 'reload', calls reload_config_files on

            SIGHUP. If 'mutate', calls mutate_config_files on SIGHUP. Other

            values produce a ValueError.

        """

        self.conf = conf

        conf.register_opts(_options.service_opts)

        self.children = {}

        self.sigcaught = None

        self.running = True

        self.wait_interval = wait_interval

        self.launcher = None

        rfd, self.writepipe = os.pipe()

        self.readpipe = eventlet.greenio.GreenPipe(rfd, 'r')

        self.signal_handler = SignalHandler()

        self.handle_signal()

        self.restart_method = restart_method

        if restart_method not in _LAUNCHER_RESTART_METHODS:

            raise ValueError(_("Invalid restart_method: %s") % restart_method)	

   def launch_service(self, service, workers=1):-----默认是一个worker,即一个进程

        """Launch a service with a given number of workers.

       :param service: a service to launch, must be an instance of

              :class:`oslo_service.service.ServiceBase`

       :param workers: a number of processes in which a service

              will be running

        """

        _check_service_base(service)

        wrap = ServiceWrapper(service, workers)

        # Hide existing objects from the garbage collector, so that most

        # existing pages will remain in shared memory rather than being

        # duplicated between subprocesses in the GC mark-and-sweep. (Requires

        # Python 3.7 or later.)

        if hasattr(gc, 'freeze'):

            gc.freeze()

        LOG.info('Starting %d workers', wrap.workers)

        while self.running and len(wrap.children) < wrap.workers:

            self._start_child(wrap)

    def _start_child(self, wrap):

        if len(wrap.forktimes) > wrap.workers:

            # Limit ourselves to one process a second (over the period of

            # number of workers * 1 second). This will allow workers to

            # start up quickly but ensure we don't fork off children that

            # die instantly too quickly.

            if time.time() - wrap.forktimes[0] < wrap.workers:

                LOG.info('Forking too fast, sleeping')

                time.sleep(1)

            wrap.forktimes.pop(0)

        wrap.forktimes.append(time.time())

        pid = os.fork()

		使用fork创建子进程后，子进程会复制父进程的数据信息，而后，程序就分两个进程继续运行后面的程序

		在子进程内，这个方法会返回0；在父进程内，这个方法会返回子进程的编号PID

        if pid == 0:

            self.launcher = self._child_process(wrap.service)-----fork的子进程接管服务

            while True:

                self._child_process_handle_signal()

                status, signo = self._child_wait_for_exit_or_signal(

                    self.launcher)

                if not _is_sighup_and_daemon(signo):

                    self.launcher.wait()

                    break

                self.launcher.restart()------重启这个进程

            os._exit(status)

        LOG.debug('Started child %d', pid)

        wrap.children.add(pid)

        self.children[pid] = wrap

        return pid

因此得出，每个 backend 启动的 cinder-volume service 都是独立的子进程，即每个 launcher 都是一个子进程

oslo_service/service.py:class Launcher(object):

    def restart(self):

        """Reload config files and restart service.

        :returns: The return value from reload_config_files or

          mutate_config_files, according to the restart_method.

        """

        if self.restart_method == 'reload':

            self.conf.reload_config_files()

        elif self.restart_method == 'mutate':

            self.conf.mutate_config_files()

        self.services.restart()

oslo_service/service.py:class Services(object)

    def restart(self):

        """Reset services and start them in new threads."""

        self.stop()

        self.done = event.Event()

        for restart_service in self.services:

            restart_service.reset()

            self.tg.add_thread(self.run_service, restart_service, self.done)

    @staticmethod

    def run_service(service, done):

        """Service start wrapper.

        :param service: service to run

        :param done: event to wait on until a shutdown is triggered

        :returns: None

        """

        try:

            service.start()-------调用服务的start方法，最终调用的是cinder/service.py:class Service(service.Service):start方法，因为该方法继承oslo_service,service方法

        except Exception:

            LOG.exception('Error starting thread.')

            raise SystemExit(1)

        else:

            done.wait()		

cinder/service.py:class Service类:start方法，

   def start(self):

        version_string = version.version_string()

        LOG.info(_LI('Starting %(topic)s node (version %(version_string)s)'),

                 {'topic': self.topic, 'version_string': version_string})

        self.model_disconnected = False

        if self.coordination:

            coordination.COORDINATOR.start()

        self.manager.init_host(added_to_cluster=self.added_to_cluster,--------------------初始化驱动程序

                               service_id=Service.service_id)

        LOG.debug("Creating RPC server for service %s", self.topic)

        ctxt = context.get_admin_context()

        endpoints = [self.manager]

        endpoints.extend(self.manager.additional_endpoints)

        obj_version_cap = objects.Service.get_minimum_obj_version(ctxt)

        LOG.debug("Pinning object versions for RPC server serializer to %s",

                  obj_version_cap)

        serializer = objects_base.CinderObjectSerializer(obj_version_cap)

        target = messaging.Target(topic=self.topic, server=self.host)

        self.rpcserver = rpc.get_server(target, endpoints, serializer)

        self.rpcserver.start()

        # NOTE(dulek): Kids, don't do that at home. We're relying here on

        # oslo.messaging implementation details to keep backward compatibility

        # with pre-Ocata services. This will not matter once we drop

        # compatibility with them.

        if self.topic == constants.VOLUME_TOPIC:

            target = messaging.Target(

                topic='%(topic)s.%(host)s' % {'topic': self.topic,

                                              'host': self.host},

                server=vol_utils.extract_host(self.host, 'host'))

            self.backend_rpcserver = rpc.get_server(target, endpoints,

                                                    serializer)

            self.backend_rpcserver.start()

        # TODO(geguileo): In O - Remove the is_svc_upgrading_to_n part

        if self.cluster and not self.is_svc_upgrading_to_n(self.binary):

            LOG.info(_LI('Starting %(topic)s cluster %(cluster)s (version '

                         '%(version)s)'),

                     {'topic': self.topic, 'version': version_string,

                      'cluster': self.cluster})

            target = messaging.Target(

                topic='%s.%s' % (self.topic, self.cluster),

                server=vol_utils.extract_host(self.cluster, 'host'))

            serializer = objects_base.CinderObjectSerializer(obj_version_cap)

            self.cluster_rpcserver = rpc.get_server(target, endpoints,

                                                    serializer)

            self.cluster_rpcserver.start()

        self.manager.init_host_with_rpc()

        if self.report_interval:

            pulse = loopingcall.FixedIntervalLoopingCall(----------------启动上报给scheduler定时任务

                self.report_state)

            pulse.start(interval=self.report_interval,

                        initial_delay=self.report_interval)

            self.timers.append(pulse)

        if self.periodic_interval:

            if self.periodic_fuzzy_delay:

                initial_delay = random.randint(0, self.periodic_fuzzy_delay)

            else:

                initial_delay = None

            periodic = loopingcall.FixedIntervalLoopingCall(--------------启动轮训根性定时任务

                self.periodic_tasks)

            periodic.start(interval=self.periodic_interval,

                           initial_delay=initial_delay)

            self.timers.append(periodic)

cinder/volume/manager.py:VolumeManager

init_host 过程分析

    def init_host(self, added_to_cluster=None, **kwargs):

        """Perform any required initialization."""

        ctxt = context.get_admin_context()

        if not self.driver.supported:

            utils.log_unsupported_driver_warning(self.driver)

            if not self.configuration.enable_unsupported_driver:

                LOG.error(_LE("Unsupported drivers are disabled."

                              " You can re-enable by adding "

                              "enable_unsupported_driver=True to the "

                              "driver section in cinder.conf"),

                          resource={'type': 'driver',

                                    'id': self.__class__.__name__})

                return

        # If we have just added this host to a cluster we have to include all

        # our resources in that cluster.

        if added_to_cluster:

            self._include_resources_in_cluster(ctxt)

        LOG.info(_LI("Starting volume driver %(driver_name)s (%(version)s)"),

                 {'driver_name': self.driver.__class__.__name__,

                  'version': self.driver.get_version()})

        try:

            self.driver.do_setup(ctxt)---------------------存储的初始化操作放到了 do_setup 中，当 do_setup 失败时，并不会导致服务启动失败，也不会影响 multi backend 的其他 backend。

            self.driver.check_for_setup_error()

        except Exception:

            LOG.exception(_LE("Failed to initialize driver."),

                          resource={'type': 'driver',

                                    'id': self.__class__.__name__})

            # we don't want to continue since we failed

            # to initialize the driver correctly.

            return

        # Initialize backend capabilities list

        self.driver.init_capabilities()

        volumes = self._get_my_volumes(ctxt)

        snapshots = self._get_my_snapshots(ctxt)

        self._sync_provider_info(ctxt, volumes, snapshots)

        # FIXME volume count for exporting is wrong

        self.stats['pools'] = {}

        self.stats.update({'allocated_capacity_gb': 0})

        try:

            for volume in volumes:

                # available volume should also be counted into allocated

                if volume['status'] in ['in-use', 'available']:

                    # calculate allocated capacity for driver

                    self._count_allocated_capacity(ctxt, volume)

                    try:

                        if volume['status'] in ['in-use']:

                            self.driver.ensure_export(ctxt, volume)

                    except Exception:

                        LOG.exception(_LE("Failed to re-export volume, "

                                          "setting to ERROR."),

                                      resource=volume)

                        volume.conditional_update({'status': 'error'},

                                                  {'status': 'in-use'})

            # All other cleanups are processed by parent class CleanableManager

        except Exception:

            LOG.exception(_LE("Error during re-export on driver init."),

                          resource=volume)

            return

        self.driver.set_throttle()

        # at this point the driver is considered initialized.

        # NOTE(jdg): Careful though because that doesn't mean

        # that an entry exists in the service table

        self.driver.set_initialized()-------------set_initialized 将驱动的 initialized 属性设定为 true，标志驱动成功启动并初始化完成。

        # Keep the image tmp file clean when init host.

        backend_name = vol_utils.extract_host(self.service_topic_queue)

        image_utils.cleanup_temporary_file(backend_name)

        # collect and publish service capabilities

        self.publish_service_capabilities(ctxt)

        LOG.info(_LI("Driver initialization completed successfully."),

                 resource={'type': 'driver',

                           'id': self.driver.__class__.__name__})

        # Make sure to call CleanableManager to do the cleanup

        super(VolumeManager, self).init_host(added_to_cluster=added_to_cluster,

                                             **kwargs)