diff --git a/cinder/brick/local_dev/lvm.py b/cinder/brick/local_dev/lvm.py
index ac2e5672ee1..5c4c2ca1dd3 100644
--- a/cinder/brick/local_dev/lvm.py
+++ b/cinder/brick/local_dev/lvm.py
@@ -703,6 +703,44 @@ class LVM(executor.Executor):
                 return True
         return False
 
+    def lv_is_snapshot(self, name):
+        """Return True if LV is a snapshot, False otherwise."""
+        cmd = LVM.LVM_CMD_PREFIX + ['lvdisplay', '--noheading', '-C', '-o',
+                                    'Attr', '%s/%s' % (self.vg_name, name)]
+        out, _err = self._execute(*cmd,
+                                  root_helper=self._root_helper,
+                                  run_as_root=True)
+        out = out.strip()
+        if out:
+            if (out[0] == 's'):
+                return True
+        return False
+
+    def lv_is_open(self, name):
+        """Return True if LV is currently open, False otherwise."""
+        cmd = LVM.LVM_CMD_PREFIX + ['lvdisplay', '--noheading', '-C', '-o',
+                                    'Attr', '%s/%s' % (self.vg_name, name)]
+        out, _err = self._execute(*cmd,
+                                  root_helper=self._root_helper,
+                                  run_as_root=True)
+        out = out.strip()
+        if out:
+            if (out[5] == 'o'):
+                return True
+        return False
+
+    def lv_get_origin(self, name):
+        """Return the origin of an LV that is a snapshot, None otherwise."""
+        cmd = LVM.LVM_CMD_PREFIX + ['lvdisplay', '--noheading', '-C', '-o',
+                                    'Origin', '%s/%s' % (self.vg_name, name)]
+        out, _err = self._execute(*cmd,
+                                  root_helper=self._root_helper,
+                                  run_as_root=True)
+        out = out.strip()
+        if out:
+            return out
+        return None
+
     def extend_volume(self, lv_name, new_size):
         """Extend the size of an existing volume."""
         # Volumes with snaps have attributes 'o' or 'O' and will be
diff --git a/cinder/tests/unit/brick/test_brick_lvm.py b/cinder/tests/unit/brick/test_brick_lvm.py
index a8dc3f9f89c..2673abf4b53 100644
--- a/cinder/tests/unit/brick/test_brick_lvm.py
+++ b/cinder/tests/unit/brick/test_brick_lvm.py
@@ -47,7 +47,7 @@ class BrickLvmTestCase(test.TestCase):
     def fake_customised_lvm_version(obj, *cmd, **kwargs):
         return ("  LVM version:     2.02.100(2)-RHEL6 (2013-09-12)\n", "")
 
-    def fake_execute(obj, *cmd, **kwargs):
+    def fake_execute(obj, *cmd, **kwargs):  # noqa
         cmd_string = ', '.join(cmd)
         data = "\n"
 
@@ -115,8 +115,18 @@ class BrickLvmTestCase(test.TestCase):
               cmd_string):
             if 'test-volumes' in cmd_string:
                 data = '  wi-a-'
+            elif 'snapshot' in cmd_string:
+                data = '  swi-a-s--'
+            elif 'open' in cmd_string:
+                data = '  -wi-ao---'
             else:
                 data = '  owi-a-'
+        elif ('env, LC_ALL=C, lvdisplay, --noheading, -C, -o, Origin' in
+              cmd_string):
+            if 'snapshot' in cmd_string:
+                data = '  fake-volume-1'
+            else:
+                data = '       '
         elif 'env, LC_ALL=C, pvs, --noheadings' in cmd_string:
             data = "  fake-vg|/dev/sda|10.00|1.00\n"
             data += "  fake-vg|/dev/sdb|10.00|1.00\n"
@@ -322,6 +332,19 @@ class BrickLvmTestCase(test.TestCase):
         self.assertTrue(self.vg.lv_has_snapshot('fake-vg'))
         self.assertFalse(self.vg.lv_has_snapshot('test-volumes'))
 
+    def test_lv_is_snapshot(self):
+        self.assertTrue(self.vg.lv_is_snapshot('fake-snapshot'))
+        self.assertFalse(self.vg.lv_is_snapshot('test-volumes'))
+
+    def test_lv_is_open(self):
+        self.assertTrue(self.vg.lv_is_open('fake-open'))
+        self.assertFalse(self.vg.lv_is_open('fake-snapshot'))
+
+    def test_lv_get_origin(self):
+        self.assertEqual('fake-volume-1',
+                         self.vg.lv_get_origin('fake-snapshot'))
+        self.assertFalse(None, self.vg.lv_get_origin('test-volumes'))
+
     def test_activate_lv(self):
         with mock.patch.object(self.vg, '_execute'):
             self.vg._supports_lvchange_ignoreskipactivation = True
diff --git a/cinder/tests/unit/test_volume_utils.py b/cinder/tests/unit/test_volume_utils.py
index 30e9a7df694..d5ae8d8e89c 100644
--- a/cinder/tests/unit/test_volume_utils.py
+++ b/cinder/tests/unit/test_volume_utils.py
@@ -758,52 +758,116 @@ class VolumeUtilsTestCase(test.TestCase):
         host_2 = 'fake_host2@backend1'
         self.assertFalse(volume_utils.hosts_are_equivalent(host_1, host_2))
 
+    @mock.patch('cinder.volume.utils.CONF')
+    def test_extract_id_from_volume_name_vol_id_pattern(self, conf_mock):
+        conf_mock.volume_name_template = 'volume-%s'
+        vol_id = 'd8cd1feb-2dcc-404d-9b15-b86fe3bec0a1'
+        vol_name = conf_mock.volume_name_template % vol_id
+        result = volume_utils.extract_id_from_volume_name(vol_name)
+        self.assertEqual(vol_id, result)
+
+    @mock.patch('cinder.volume.utils.CONF')
+    def test_extract_id_from_volume_name_vol_id_vol_pattern(self, conf_mock):
+        conf_mock.volume_name_template = 'volume-%s-volume'
+        vol_id = 'd8cd1feb-2dcc-404d-9b15-b86fe3bec0a1'
+        vol_name = conf_mock.volume_name_template % vol_id
+        result = volume_utils.extract_id_from_volume_name(vol_name)
+        self.assertEqual(vol_id, result)
+
+    @mock.patch('cinder.volume.utils.CONF')
+    def test_extract_id_from_volume_name_id_vol_pattern(self, conf_mock):
+        conf_mock.volume_name_template = '%s-volume'
+        vol_id = 'd8cd1feb-2dcc-404d-9b15-b86fe3bec0a1'
+        vol_name = conf_mock.volume_name_template % vol_id
+        result = volume_utils.extract_id_from_volume_name(vol_name)
+        self.assertEqual(vol_id, result)
+
+    @mock.patch('cinder.volume.utils.CONF')
+    def test_extract_id_from_volume_name_no_match(self, conf_mock):
+        conf_mock.volume_name_template = '%s-volume'
+        vol_name = 'd8cd1feb-2dcc-404d-9b15-b86fe3bec0a1'
+        result = volume_utils.extract_id_from_volume_name(vol_name)
+        self.assertIsNone(result)
+        vol_name = 'blahblahblah'
+        result = volume_utils.extract_id_from_volume_name(vol_name)
+        self.assertIsNone(result)
+
     @mock.patch('cinder.db.sqlalchemy.api.resource_exists', return_value=True)
     def test_check_managed_volume_already_managed(self, exists_mock):
         id_ = 'd8cd1feb-2dcc-404d-9b15-b86fe3bec0a1'
-        vol_id = 'volume-' + id_
-        result = volume_utils.check_already_managed_volume(vol_id)
-        self.assertTrue(result)
-        exists_mock.assert_called_once_with(mock.ANY, models.Volume, id_)
-
-    @mock.patch('cinder.db.sqlalchemy.api.resource_exists', return_value=True)
-    def test_check_already_managed_with_vol_id_vol_pattern(self, exists_mock):
-        template = 'volume-%s-volume'
-        self.override_config('volume_name_template', template)
-        id_ = 'd8cd1feb-2dcc-404d-9b15-b86fe3bec0a1'
-        vol_id = template % id_
-
-        result = volume_utils.check_already_managed_volume(vol_id)
-        self.assertTrue(result)
-        exists_mock.assert_called_once_with(mock.ANY, models.Volume, id_)
-
-    @mock.patch('cinder.db.sqlalchemy.api.resource_exists', return_value=True)
-    def test_check_already_managed_with_id_vol_pattern(self, exists_mock):
-        template = '%s-volume'
-        self.override_config('volume_name_template', template)
-        id_ = 'd8cd1feb-2dcc-404d-9b15-b86fe3bec0a1'
-        vol_id = template % id_
-
-        result = volume_utils.check_already_managed_volume(vol_id)
+        result = volume_utils.check_already_managed_volume(id_)
         self.assertTrue(result)
         exists_mock.assert_called_once_with(mock.ANY, models.Volume, id_)
 
     @mock.patch('cinder.db.sqlalchemy.api.resource_exists', return_value=False)
-    def test_check_managed_volume_not_managed_cinder_like_name(self,
-                                                               exists_mock):
+    def test_check_managed_volume_not_managed_proper_uuid(self, exists_mock):
         id_ = 'd8cd1feb-2dcc-404d-9b15-b86fe3bec0a1'
-        vol_id = 'volume-' + id_
-        result = volume_utils.check_already_managed_volume(vol_id)
+        result = volume_utils.check_already_managed_volume(id_)
         self.assertFalse(result)
         exists_mock.assert_called_once_with(mock.ANY, models.Volume, id_)
 
-    def test_check_managed_volume_not_managed(self):
-        result = volume_utils.check_already_managed_volume('test-volume')
+    def test_check_managed_volume_not_managed_invalid_id(self):
+        result = volume_utils.check_already_managed_volume(1)
+        self.assertFalse(result)
+        result = volume_utils.check_already_managed_volume('not-a-uuid')
         self.assertFalse(result)
 
-    def test_check_managed_volume_not_managed_id_like_uuid(self):
-        result = volume_utils.check_already_managed_volume('volume-d8cd1fe')
-        self.assertFalse(result)
+    @mock.patch('cinder.volume.utils.CONF')
+    def test_extract_id_from_snapshot_name(self, conf_mock):
+        conf_mock.snapshot_name_template = '%s-snapshot'
+        snap_id = 'd8cd1feb-2dcc-404d-9b15-b86fe3bec0a1'
+        snap_name = conf_mock.snapshot_name_template % snap_id
+        result = volume_utils.extract_id_from_snapshot_name(snap_name)
+        self.assertEqual(snap_id, result)
+
+    @mock.patch('cinder.volume.utils.CONF')
+    def test_extract_id_from_snapshot_name_no_match(self, conf_mock):
+        conf_mock.snapshot_name_template = '%s-snapshot'
+        snap_name = 'd8cd1feb-2dcc-404d-9b15-b86fe3bec0a1'
+        result = volume_utils.extract_id_from_snapshot_name(snap_name)
+        self.assertIsNone(result)
+        snap_name = 'blahblahblah'
+        result = volume_utils.extract_id_from_snapshot_name(snap_name)
+        self.assertIsNone(result)
+
+    def test_paginate_entries_list_with_marker(self):
+        entries = [{'reference': {'name': 'vol03'}, 'size': 1},
+                   {'reference': {'name': 'vol01'}, 'size': 3},
+                   {'reference': {'name': 'vol02'}, 'size': 3},
+                   {'reference': {'name': 'vol04'}, 'size': 2},
+                   {'reference': {'name': 'vol06'}, 'size': 3},
+                   {'reference': {'name': 'vol07'}, 'size': 1},
+                   {'reference': {'name': 'vol05'}, 'size': 1}]
+        expected = [{'reference': {'name': 'vol04'}, 'size': 2},
+                    {'reference': {'name': 'vol03'}, 'size': 1},
+                    {'reference': {'name': 'vol05'}, 'size': 1}]
+        res = volume_utils.paginate_entries_list(entries, {'name': 'vol02'}, 3,
+                                                 1, ['size', 'reference'],
+                                                 ['desc', 'asc'])
+        self.assertEqual(expected, res)
+
+    def test_paginate_entries_list_without_marker(self):
+        entries = [{'reference': {'name': 'vol03'}, 'size': 1},
+                   {'reference': {'name': 'vol01'}, 'size': 3},
+                   {'reference': {'name': 'vol02'}, 'size': 3},
+                   {'reference': {'name': 'vol04'}, 'size': 2},
+                   {'reference': {'name': 'vol06'}, 'size': 3},
+                   {'reference': {'name': 'vol07'}, 'size': 1},
+                   {'reference': {'name': 'vol05'}, 'size': 1}]
+        expected = [{'reference': {'name': 'vol07'}, 'size': 1},
+                    {'reference': {'name': 'vol06'}, 'size': 3},
+                    {'reference': {'name': 'vol05'}, 'size': 1}]
+        res = volume_utils.paginate_entries_list(entries, None, 3, None,
+                                                 ['reference'], ['desc'])
+        self.assertEqual(expected, res)
+
+    def test_paginate_entries_list_marker_not_found(self):
+        entries = [{'reference': {'name': 'vol03'}, 'size': 1},
+                   {'reference': {'name': 'vol01'}, 'size': 3}]
+        self.assertRaises(exception.InvalidInput,
+                          volume_utils.paginate_entries_list,
+                          entries, {'name': 'vol02'}, 3, None,
+                          ['size', 'reference'], ['desc', 'asc'])
 
     def test_convert_config_string_to_dict(self):
         test_string = "{'key-1'='val-1' 'key-2'='val-2' 'key-3'='val-3'}"
diff --git a/cinder/tests/unit/volume/drivers/test_lvm_driver.py b/cinder/tests/unit/volume/drivers/test_lvm_driver.py
index ea9aa42e72c..e63acfb9e52 100644
--- a/cinder/tests/unit/volume/drivers/test_lvm_driver.py
+++ b/cinder/tests/unit/volume/drivers/test_lvm_driver.py
@@ -791,6 +791,84 @@ class LVMVolumeDriverTestCase(DriverTestCase):
         ret = self.volume.driver.unmanage(volume)
         self.assertIsNone(ret)
 
+    def test_lvm_get_manageable_volumes(self):
+        cinder_vols = [{'id': '00000000-0000-0000-0000-000000000000'}]
+        lvs = [{'name': 'volume-00000000-0000-0000-0000-000000000000',
+                'size': '1.75'},
+               {'name': 'volume-00000000-0000-0000-0000-000000000001',
+                'size': '3.0'},
+               {'name': 'snapshot-00000000-0000-0000-0000-000000000002',
+                'size': '2.2'},
+               {'name': 'myvol', 'size': '4.0'}]
+        self.volume.driver.vg = mock.Mock()
+        self.volume.driver.vg.get_volumes.return_value = lvs
+        self.volume.driver.vg.lv_is_snapshot.side_effect = [False, False,
+                                                            True, False]
+        self.volume.driver.vg.lv_is_open.side_effect = [True, False]
+        res = self.volume.driver.get_manageable_volumes(cinder_vols, None,
+                                                        1000, 0,
+                                                        ['size'], ['asc'])
+        exp = [{'size': 2, 'reason_not_safe': None, 'extra_info': None,
+                'reference': {'source-name':
+                              'volume-00000000-0000-0000-0000-000000000000'},
+                'cinder_id': '00000000-0000-0000-0000-000000000000',
+                'safe_to_manage': False, 'reason_not_safe': 'already managed'},
+               {'size': 3, 'reason_not_safe': 'volume in use',
+                'reference': {'source-name':
+                              'volume-00000000-0000-0000-0000-000000000001'},
+                'safe_to_manage': False, 'cinder_id': None,
+                'extra_info': None},
+               {'size': 4, 'reason_not_safe': None,
+                'safe_to_manage': True, 'reference': {'source-name': 'myvol'},
+                'cinder_id': None, 'extra_info': None}]
+        self.assertEqual(exp, res)
+
+    def test_lvm_get_manageable_snapshots(self):
+        cinder_snaps = [{'id': '00000000-0000-0000-0000-000000000000'}]
+        lvs = [{'name': 'snapshot-00000000-0000-0000-0000-000000000000',
+                'size': '1.75'},
+               {'name': 'volume-00000000-0000-0000-0000-000000000001',
+                'size': '3.0'},
+               {'name': 'snapshot-00000000-0000-0000-0000-000000000002',
+                'size': '2.2'},
+               {'name': 'mysnap', 'size': '4.0'}]
+        self.volume.driver.vg = mock.Mock()
+        self.volume.driver.vg.get_volumes.return_value = lvs
+        self.volume.driver.vg.lv_is_snapshot.side_effect = [True, False, True,
+                                                            True]
+        self.volume.driver.vg.lv_is_open.side_effect = [True, False]
+        self.volume.driver.vg.lv_get_origin.side_effect = [
+            'volume-00000000-0000-0000-0000-000000000000',
+            'volume-00000000-0000-0000-0000-000000000002',
+            'myvol']
+        res = self.volume.driver.get_manageable_snapshots(cinder_snaps, None,
+                                                          1000, 0,
+                                                          ['size'], ['asc'])
+        exp = [{'size': 2, 'reason_not_safe': 'already managed',
+                'reference':
+                {'source-name':
+                 'snapshot-00000000-0000-0000-0000-000000000000'},
+                'safe_to_manage': False, 'extra_info': None,
+                'cinder_id': '00000000-0000-0000-0000-000000000000',
+                'source_reference':
+                {'source-name':
+                 'volume-00000000-0000-0000-0000-000000000000'}},
+               {'size': 3, 'reason_not_safe': 'snapshot in use',
+                'reference':
+                {'source-name':
+                 'snapshot-00000000-0000-0000-0000-000000000002'},
+                'safe_to_manage': False, 'extra_info': None,
+                'cinder_id': None,
+                'source_reference':
+                {'source-name':
+                 'volume-00000000-0000-0000-0000-000000000002'}},
+               {'size': 4, 'reason_not_safe': None,
+                'reference': {'source-name': 'mysnap'},
+                'safe_to_manage': True, 'cinder_id': None,
+                'source_reference': {'source-name': 'myvol'},
+                'extra_info': None}]
+        self.assertEqual(exp, res)
+
     # Global setting, LVM setting, expected outcome
     @ddt.data((10.0, 2.0, 2.0))
     @ddt.data((10.0, None, 10.0))
diff --git a/cinder/volume/drivers/lvm.py b/cinder/volume/drivers/lvm.py
index c7141646ada..f10ca6fb66c 100644
--- a/cinder/volume/drivers/lvm.py
+++ b/cinder/volume/drivers/lvm.py
@@ -178,6 +178,12 @@ class LVMVolumeDriver(driver.VolumeDriver):
             return snapshot_name
         return '_' + snapshot_name
 
+    def _unescape_snapshot(self, snapshot_name):
+        # Undo snapshot name change done by _escape_snapshot()
+        if not snapshot_name.startswith('_snapshot'):
+            return snapshot_name
+        return snapshot_name[1:]
+
     def _create_volume(self, name, size, lvm_type, mirror_count, vg=None):
         vg_ref = self.vg
         if vg is not None:
@@ -586,7 +592,8 @@ class LVMVolumeDriver(driver.VolumeDriver):
         lv_name = existing_ref['source-name']
         self.vg.get_volume(lv_name)
 
-        if volutils.check_already_managed_volume(lv_name):
+        vol_id = volutils.extract_id_from_volume_name(lv_name)
+        if volutils.check_already_managed_volume(vol_id):
             raise exception.ManageExistingAlreadyManaged(volume_ref=lv_name)
 
         # Attempt to rename the LV to match the OpenStack internal name.
@@ -654,6 +661,61 @@ class LVMVolumeDriver(driver.VolumeDriver):
             existing_ref = {"source-name": existing_ref}
         return self.manage_existing(snapshot_temp, existing_ref)
 
+    def _get_manageable_resource_info(self, cinder_resources, resource_type,
+                                      marker, limit, offset, sort_keys,
+                                      sort_dirs):
+        entries = []
+        lvs = self.vg.get_volumes()
+        cinder_ids = [resource['id'] for resource in cinder_resources]
+
+        for lv in lvs:
+            is_snap = self.vg.lv_is_snapshot(lv['name'])
+            if ((resource_type == 'volume' and is_snap) or
+                    (resource_type == 'snapshot' and not is_snap)):
+                continue
+
+            if resource_type == 'volume':
+                potential_id = volutils.extract_id_from_volume_name(lv['name'])
+            else:
+                unescape = self._unescape_snapshot(lv['name'])
+                potential_id = volutils.extract_id_from_snapshot_name(unescape)
+            lv_info = {'reference': {'source-name': lv['name']},
+                       'size': int(math.ceil(float(lv['size']))),
+                       'cinder_id': None,
+                       'extra_info': None}
+
+            if potential_id in cinder_ids:
+                lv_info['safe_to_manage'] = False
+                lv_info['reason_not_safe'] = 'already managed'
+                lv_info['cinder_id'] = potential_id
+            elif self.vg.lv_is_open(lv['name']):
+                lv_info['safe_to_manage'] = False
+                lv_info['reason_not_safe'] = '%s in use' % resource_type
+            else:
+                lv_info['safe_to_manage'] = True
+                lv_info['reason_not_safe'] = None
+
+            if resource_type == 'snapshot':
+                origin = self.vg.lv_get_origin(lv['name'])
+                lv_info['source_reference'] = {'source-name': origin}
+
+            entries.append(lv_info)
+
+        return volutils.paginate_entries_list(entries, marker, limit, offset,
+                                              sort_keys, sort_dirs)
+
+    def get_manageable_volumes(self, cinder_volumes, marker, limit, offset,
+                               sort_keys, sort_dirs):
+        return self._get_manageable_resource_info(cinder_volumes, 'volume',
+                                                  marker, limit,
+                                                  offset, sort_keys, sort_dirs)
+
+    def get_manageable_snapshots(self, cinder_snapshots, marker, limit, offset,
+                                 sort_keys, sort_dirs):
+        return self._get_manageable_resource_info(cinder_snapshots, 'snapshot',
+                                                  marker, limit,
+                                                  offset, sort_keys, sort_dirs)
+
     def retype(self, context, volume, new_type, diff, host):
         """Retypes a volume, allow QoS and extra_specs change."""
 
diff --git a/cinder/volume/utils.py b/cinder/volume/utils.py
index 53fd2bdb7cb..6629b002a07 100644
--- a/cinder/volume/utils.py
+++ b/cinder/volume/utils.py
@@ -16,7 +16,9 @@
 
 
 import ast
+import functools
 import math
+import operator
 import re
 import time
 import uuid
@@ -664,28 +666,85 @@ def read_proc_mounts():
         return mounts.readlines()
 
 
-def _extract_id(vol_name):
+def extract_id_from_volume_name(vol_name):
     regex = re.compile(
         CONF.volume_name_template.replace('%s', '(?P<uuid>.+)'))
     match = regex.match(vol_name)
     return match.group('uuid') if match else None
 
 
-def check_already_managed_volume(vol_name):
+def check_already_managed_volume(vol_id):
     """Check cinder db for already managed volume.
 
-    :param vol_name: volume name parameter
+    :param vol_id: volume id parameter
     :returns: bool -- return True, if db entry with specified
-                      volume name exist, otherwise return False
+                      volume id exists, otherwise return False
     """
-    vol_id = _extract_id(vol_name)
     try:
-        return (vol_id and uuid.UUID(vol_id, version=4) and
+        return (vol_id and isinstance(vol_id, six.string_types) and
+                uuid.UUID(vol_id, version=4) and
                 objects.Volume.exists(context.get_admin_context(), vol_id))
     except ValueError:
         return False
 
 
+def extract_id_from_snapshot_name(snap_name):
+    """Return a snapshot's ID from its name on the backend."""
+    regex = re.compile(
+        CONF.snapshot_name_template.replace('%s', '(?P<uuid>.+)'))
+    match = regex.match(snap_name)
+    return match.group('uuid') if match else None
+
+
+def paginate_entries_list(entries, marker, limit, offset, sort_keys,
+                          sort_dirs):
+    """Paginate a list of entries.
+
+    :param entries: list of dictionaries
+    :marker: The last element previously returned
+    :limit: The maximum number of items to return
+    :offset: The number of items to skip from the marker or from the first
+             element.
+    :sort_keys: A list of keys in the dictionaries to sort by
+    :sort_dirs: A list of sort directions, where each is either 'asc' or 'dec'
+    """
+    comparers = [(operator.itemgetter(key.strip()), multiplier)
+                 for (key, multiplier) in zip(sort_keys, sort_dirs)]
+
+    def comparer(left, right):
+        for fn, d in comparers:
+            left_val = fn(left)
+            right_val = fn(right)
+            if isinstance(left_val, dict):
+                left_val = sorted(left_val.values())[0]
+            if isinstance(right_val, dict):
+                right_val = sorted(right_val.values())[0]
+            if left_val == right_val:
+                continue
+            if d == 'asc':
+                return -1 if left_val < right_val else 1
+            else:
+                return -1 if left_val > right_val else 1
+        else:
+            return 0
+    sorted_entries = sorted(entries, key=functools.cmp_to_key(comparer))
+
+    start_index = 0
+    if offset is None:
+        offset = 0
+    if marker:
+        start_index = -1
+        for i, entry in enumerate(sorted_entries):
+            if entry['reference'] == marker:
+                start_index = i + 1
+                break
+        if start_index < 0:
+            msg = _('marker not found: %s') % marker
+            raise exception.InvalidInput(reason=msg)
+    range_end = start_index + limit
+    return sorted_entries[start_index + offset:range_end + offset]
+
+
 def convert_config_string_to_dict(config_string):
     """Convert config file replication string to a dict.