def create_volume(self, context, volume_id, request_spec=None, filter_properties=None, allow_reschedule=True, volume=None):
    'Creates the volume.'
    if (volume is None):
        volume = objects.Volume.get_by_id(context, volume_id)
    context_elevated = context.elevated()
    if (filter_properties is None):
        filter_properties = {}
    if (request_spec is None):
        request_spec = {}
    try:
        flow_engine = create_volume.get_flow(context_elevated, self, self.db, self.driver, self.scheduler_rpcapi, self.host, volume.id, allow_reschedule, context, request_spec, filter_properties, image_volume_cache=self.image_volume_cache)
    except Exception:
        msg = _('Create manager volume flow failed.')
        LOG.exception(msg, resource={'type': 'volume', 'id': volume.id})
        raise exception.CinderException(msg)
    snapshot_id = request_spec.get('snapshot_id')
    source_volid = request_spec.get('source_volid')
    source_replicaid = request_spec.get('source_replicaid')
    if (snapshot_id is not None):
        locked_action = ('%s-%s' % (snapshot_id, 'delete_snapshot'))
    elif (source_volid is not None):
        locked_action = ('%s-%s' % (source_volid, 'delete_volume'))
    elif (source_replicaid is not None):
        locked_action = ('%s-%s' % (source_replicaid, 'delete_volume'))
    else:
        locked_action = None

    def _run_flow():
        with flow_utils.DynamicLogListener(flow_engine, logger=LOG):
            flow_engine.run()

    @utils.synchronized(locked_action, external=True)
    def _run_flow_locked():
        _run_flow()
    rescheduled = False
    vol_ref = None
    try:
        if (locked_action is None):
            _run_flow()
        else:
            _run_flow_locked()
    finally:
        try:
            vol_ref = flow_engine.storage.fetch('volume_ref')
        except tfe.NotFound:
            try:
                rescheduled = flow_engine.storage.get_revert_result(create_volume.OnFailureRescheduleTask.make_name([create_volume.ACTION]))
            except tfe.NotFound:
                pass
        if (not rescheduled):
            if (not vol_ref):
                vol_ref = objects.Volume.get_by_id(context, volume.id)
            self._update_allocated_capacity(vol_ref)
    LOG.info(_LI('Created volume successfully.'), resource=vol_ref)
    return vol_ref.id
Creates the volume.
def create_volume(self, context, volume_id, request_spec=None, filter_properties=None, allow_reschedule=True, volume=None):
    
    if (volume is None):
        volume = objects.Volume.get_by_id(context, volume_id)
    context_elevated = context.elevated()
    if (filter_properties is None):
        filter_properties = {}
    if (request_spec is None):
        request_spec = {}
    try:
        flow_engine = create_volume.get_flow(context_elevated, self, self.db, self.driver, self.scheduler_rpcapi, self.host, volume.id, allow_reschedule, context, request_spec, filter_properties, image_volume_cache=self.image_volume_cache)
    except Exception:
        msg = _('Create manager volume flow failed.')
        LOG.exception(msg, resource={'type': 'volume', 'id': volume.id})
        raise exception.CinderException(msg)
    snapshot_id = request_spec.get('snapshot_id')
    source_volid = request_spec.get('source_volid')
    source_replicaid = request_spec.get('source_replicaid')
    if (snapshot_id is not None):
        locked_action = ('%s-%s' % (snapshot_id, 'delete_snapshot'))
    elif (source_volid is not None):
        locked_action = ('%s-%s' % (source_volid, 'delete_volume'))
    elif (source_replicaid is not None):
        locked_action = ('%s-%s' % (source_replicaid, 'delete_volume'))
    else:
        locked_action = None

    def _run_flow():
        with flow_utils.DynamicLogListener(flow_engine, logger=LOG):
            flow_engine.run()

    @utils.synchronized(locked_action, external=True)
    def _run_flow_locked():
        _run_flow()
    rescheduled = False
    vol_ref = None
    try:
        if (locked_action is None):
            _run_flow()
        else:
            _run_flow_locked()
    finally:
        try:
            vol_ref = flow_engine.storage.fetch('volume_ref')
        except tfe.NotFound:
            try:
                rescheduled = flow_engine.storage.get_revert_result(create_volume.OnFailureRescheduleTask.make_name([create_volume.ACTION]))
            except tfe.NotFound:
                pass
        if (not rescheduled):
            if (not vol_ref):
                vol_ref = objects.Volume.get_by_id(context, volume.id)
            self._update_allocated_capacity(vol_ref)
    LOG.info(_LI('Created volume successfully.'), resource=vol_ref)
    return vol_ref.id