diff --git a/cinder/tests/unit/volume/drivers/dell_emc/unity/fake_exception.py b/cinder/tests/unit/volume/drivers/dell_emc/unity/fake_exception.py
index cd415decde6..cdcd40230fe 100644
--- a/cinder/tests/unit/volume/drivers/dell_emc/unity/fake_exception.py
+++ b/cinder/tests/unit/volume/drivers/dell_emc/unity/fake_exception.py
@@ -80,3 +80,7 @@ class UnexpectedLunDeletion(Exception):
 
 class AdapterSetupError(Exception):
     pass
+
+
+class HostDeleteIsCalled(Exception):
+    pass
diff --git a/cinder/tests/unit/volume/drivers/dell_emc/unity/test_adapter.py b/cinder/tests/unit/volume/drivers/dell_emc/unity/test_adapter.py
index 7475f13a4df..b2b487e9e73 100644
--- a/cinder/tests/unit/volume/drivers/dell_emc/unity/test_adapter.py
+++ b/cinder/tests/unit/volume/drivers/dell_emc/unity/test_adapter.py
@@ -45,6 +45,7 @@ class MockConfig(object):
         self.san_password = 'pass'
         self.driver_ssl_cert_verify = True
         self.driver_ssl_cert_path = None
+        self.remove_empty_host = False
 
     def safe_get(self, name):
         return getattr(self, name)
@@ -70,6 +71,7 @@ class MockDriver(object):
 class MockClient(object):
     def __init__(self):
         self._system = test_client.MockSystem()
+        self.host = '10.10.10.10'  # fake unity IP
 
     @staticmethod
     def get_pools():
@@ -130,6 +132,15 @@ class MockClient(object):
     def create_host(name):
         return test_client.MockResource(name=name)
 
+    @staticmethod
+    def create_host_wo_lock(name):
+        return test_client.MockResource(name=name)
+
+    @staticmethod
+    def delete_host_wo_lock(host):
+        if host.name == 'empty-host':
+            raise ex.HostDeleteIsCalled()
+
     @staticmethod
     def attach(host, lun_or_snap):
         return 10
@@ -467,6 +478,16 @@ class CommonAdapterTest(test.TestCase):
 
         self.assertRaises(ex.DetachIsCalled, f)
 
+    def test_terminate_connection_remove_empty_host(self):
+        self.adapter.remove_empty_host = True
+
+        def f():
+            connector = {'host': 'empty-host'}
+            vol = MockOSResource(provider_location='id^lun_45', id='id_45')
+            self.adapter.terminate_connection(vol, connector)
+
+        self.assertRaises(ex.HostDeleteIsCalled, f)
+
     def test_manage_existing_by_name(self):
         ref = {'source-id': 12}
         volume = MockOSResource(name='lun1')
@@ -808,6 +829,20 @@ class FCAdapterTest(test.TestCase):
         target_wwn = ['100000051e55a100', '100000051e55a121']
         self.assertListEqual(target_wwn, data['target_wwn'])
 
+    def test_terminate_connection_remove_empty_host_return_data(self):
+        self.adapter.remove_empty_host = True
+        connector = {'host': 'empty-host-return-data', 'wwpns': 'abcdefg'}
+        volume = MockOSResource(provider_location='id^lun_41', id='id_41')
+        ret = self.adapter.terminate_connection(volume, connector)
+        self.assertEqual('fibre_channel', ret['driver_volume_type'])
+        data = ret['data']
+        target_map = {
+            '200000051e55a100': ('100000051e55a100', '100000051e55a121'),
+            '200000051e55a121': ('100000051e55a100', '100000051e55a121')}
+        self.assertDictEqual(target_map, data['initiator_target_map'])
+        target_wwn = ['100000051e55a100', '100000051e55a121']
+        self.assertListEqual(target_wwn, data['target_wwn'])
+
     def test_validate_ports_whitelist_none(self):
         ports = self.adapter.validate_ports(None)
         self.assertEqual(set(('spa_iom_0_fc0', 'spa_iom_0_fc1')), set(ports))
diff --git a/cinder/tests/unit/volume/drivers/dell_emc/unity/test_client.py b/cinder/tests/unit/volume/drivers/dell_emc/unity/test_client.py
index db7d78df5a1..9a444dd5fc4 100644
--- a/cinder/tests/unit/volume/drivers/dell_emc/unity/test_client.py
+++ b/cinder/tests/unit/volume/drivers/dell_emc/unity/test_client.py
@@ -63,6 +63,8 @@ class MockResource(object):
             raise ex.UnityResourceNotFoundError()
         elif self.get_id() == 'snap_in_use':
             raise ex.UnityDeleteAttachedSnapError()
+        elif self.name == 'empty_host':
+            raise ex.HostDeleteIsCalled()
 
     @property
     def pool(self):
@@ -545,3 +547,15 @@ class ClientTest(unittest.TestCase):
     def test_restore_snapshot(self):
         back_snap = self.client.restore_snapshot('snap1')
         self.assertEqual("internal_snap", back_snap.name)
+
+    def test_delete_host_wo_lock(self):
+        host = MockResource(name='empty-host')
+        self.client.host_cache['empty-host'] = host
+        self.assertRaises(ex.HostDeleteIsCalled,
+                          self.client.delete_host_wo_lock(host))
+
+    def test_delete_host_wo_lock_remove_from_cache(self):
+        host = MockResource(name='empty-host-in-cache')
+        self.client.host_cache['empty-host-in-cache'] = host
+        self.client.delete_host_wo_lock(host)
+        self.assertNotIn(host.name, self.client.host_cache)
diff --git a/cinder/volume/drivers/dell_emc/unity/adapter.py b/cinder/volume/drivers/dell_emc/unity/adapter.py
index 50ad66c4880..45ed0fef6bf 100644
--- a/cinder/volume/drivers/dell_emc/unity/adapter.py
+++ b/cinder/volume/drivers/dell_emc/unity/adapter.py
@@ -139,6 +139,8 @@ class CommonAdapter(object):
         self.storage_pools_map = None
         self._client = None
         self.allowed_ports = None
+        self.remove_empty_host = False
+        self.to_lock_host = False
 
     def do_setup(self, driver, conf):
         self.driver = driver
@@ -166,6 +168,9 @@ class CommonAdapter(object):
 
         self.allowed_ports = self.validate_ports(self.config.unity_io_ports)
 
+        self.remove_empty_host = self.config.remove_empty_host
+        self.to_lock_host = self.remove_empty_host
+
         group_name = (self.config.config_group if self.config.config_group
                       else 'DEFAULT')
         folder_name = '%(group)s.%(sys_name)s' % {
@@ -291,11 +296,25 @@ class CommonAdapter(object):
         else:
             self.client.delete_lun(lun_id)
 
+    def _create_host_and_attach(self, host_name, lun_or_snap):
+        @utils.lock_if(self.to_lock_host, '{lock_name}')
+        def _lock_helper(lock_name):
+            if not self.to_lock_host:
+                host = self.client.create_host(host_name)
+            else:
+                # Use the lock in the decorator
+                host = self.client.create_host_wo_lock(host_name)
+            hlu = self.client.attach(host, lun_or_snap)
+            return host, hlu
+
+        return _lock_helper('{unity}-{host}'.format(unity=self.client.host,
+                                                    host=host_name))
+
     def _initialize_connection(self, lun_or_snap, connector, vol_id):
-        host = self.client.create_host(connector['host'])
+        host, hlu = self._create_host_and_attach(connector['host'],
+                                                 lun_or_snap)
         self.client.update_host_initiators(
             host, self.get_connector_uids(connector))
-        hlu = self.client.attach(host, lun_or_snap)
         data = self.get_connection_info(hlu, host, connector)
         data['target_discovered'] = True
         if vol_id is not None:
@@ -311,13 +330,44 @@ class CommonAdapter(object):
         lun = self.client.get_lun(lun_id=self.get_lun_id(volume))
         return self._initialize_connection(lun, connector, volume.id)
 
+    @staticmethod
+    def filter_targets_by_host(host):
+        # No target info for iSCSI driver
+        return []
+
+    def _detach_and_delete_host(self, host_name, lun_or_snap):
+        @utils.lock_if(self.to_lock_host, '{lock_name}')
+        def _lock_helper(lock_name):
+            # Only get the host from cache here
+            host = self.client.create_host_wo_lock(host_name)
+            self.client.detach(host, lun_or_snap)
+            host.update()  # need update to get the latest `host_luns`
+            targets = self.filter_targets_by_host(host)
+            if self.remove_empty_host and not host.host_luns:
+                self.client.delete_host_wo_lock(host)
+            return targets
+
+        return _lock_helper('{unity}-{host}'.format(unity=self.client.host,
+                                                    host=host_name))
+
+    @staticmethod
+    def get_terminate_connection_info(connector, targets):
+        # No return data from terminate_connection for iSCSI driver
+        return {}
+
     def _terminate_connection(self, lun_or_snap, connector):
         is_force_detach = connector is None
+        data = {}
         if is_force_detach:
             self.client.detach_all(lun_or_snap)
         else:
-            host = self.client.create_host(connector['host'])
-            self.client.detach(host, lun_or_snap)
+            targets = self._detach_and_delete_host(connector['host'],
+                                                   lun_or_snap)
+            data = self.get_terminate_connection_info(connector, targets)
+        return {
+            'driver_volume_type': self.driver_volume_type,
+            'data': data,
+        }
 
     @cinder_utils.trace
     def terminate_connection(self, volume, connector):
@@ -742,24 +792,19 @@ class FCAdapter(CommonAdapter):
         data['target_lun'] = hlu
         return data
 
-    def _terminate_connection(self, lun_or_snap, connector):
+    def filter_targets_by_host(self, host):
+        if self.auto_zone_enabled and not host.host_luns:
+            return self.client.get_fc_target_info(
+                host=host, logged_in_only=True,
+                allowed_ports=self.allowed_ports)
+        return []
+
+    def get_terminate_connection_info(self, connector, targets):
         # For FC, terminate_connection needs to return data to zone manager
         # which would clean the zone based on the data.
-        super(FCAdapter, self)._terminate_connection(lun_or_snap, connector)
-
-        ret = None
-        if self.auto_zone_enabled:
-            ret = {
-                'driver_volume_type': self.driver_volume_type,
-                'data': {}
-            }
-            host = self.client.create_host(connector['host'])
-            if not host.host_luns:
-                targets = self.client.get_fc_target_info(
-                    logged_in_only=True, allowed_ports=self.allowed_ports)
-                ret['data'] = self._get_fc_zone_info(connector['wwpns'],
-                                                     targets)
-        return ret
+        if targets:
+            return self._get_fc_zone_info(connector['wwpns'], targets)
+        return {}
 
     def _get_fc_zone_info(self, initiator_wwns, target_wwns):
         mapping = self.lookup_service.get_device_mapping_from_network(
diff --git a/cinder/volume/drivers/dell_emc/unity/client.py b/cinder/volume/drivers/dell_emc/unity/client.py
index 64c74e53538..fcd8783995b 100644
--- a/cinder/volume/drivers/dell_emc/unity/client.py
+++ b/cinder/volume/drivers/dell_emc/unity/client.py
@@ -188,6 +188,9 @@ class UnityClient(object):
 
     @coordination.synchronized('{self.host}-{name}')
     def create_host(self, name):
+        return self.create_host_wo_lock(name)
+
+    def create_host_wo_lock(self, name):
         """Provides existing host if exists else create one."""
         if name not in self.host_cache:
             try:
@@ -202,6 +205,10 @@ class UnityClient(object):
             host = self.host_cache[name]
         return host
 
+    def delete_host_wo_lock(self, host):
+        host.delete()
+        del self.host_cache[host.name]
+
     def update_host_initiators(self, host, uids):
         """Updates host with the supplied uids."""
         host_initiators_ids = self.get_host_initiator_ids(host)
diff --git a/cinder/volume/drivers/dell_emc/unity/driver.py b/cinder/volume/drivers/dell_emc/unity/driver.py
index 6783b862541..a663784b809 100644
--- a/cinder/volume/drivers/dell_emc/unity/driver.py
+++ b/cinder/volume/drivers/dell_emc/unity/driver.py
@@ -37,7 +37,11 @@ UNITY_OPTS = [
     cfg.ListOpt('unity_io_ports',
                 default=None,
                 help='A comma-separated list of iSCSI or FC ports to be used. '
-                     'Each port can be Unix-style glob expressions.')]
+                     'Each port can be Unix-style glob expressions.'),
+    cfg.BoolOpt('remove_empty_host',
+                default=False,
+                help='To remove the host from Unity when the last LUN is '
+                     'detached from it. By default, it is False.')]
 
 CONF.register_opts(UNITY_OPTS, group=configuration.SHARED_CONF_GROUP)
 
@@ -53,9 +57,10 @@ class UnityDriver(driver.ManageableVD,
         2.0.0 - Add thin clone support
         3.0.0 - Add IPv6 support
         3.1.0 - Support revert to snapshot API
+        4.0.0 - Support remove empty host
     """
 
-    VERSION = '03.01.00'
+    VERSION = '04.00.00'
     VENDOR = 'Dell EMC'
     # ThirdPartySystems wiki page
     CI_WIKI_NAME = "EMC_UNITY_CI"
diff --git a/cinder/volume/drivers/dell_emc/unity/utils.py b/cinder/volume/drivers/dell_emc/unity/utils.py
index 8d22d01129f..ca164bf4aa6 100644
--- a/cinder/volume/drivers/dell_emc/unity/utils.py
+++ b/cinder/volume/drivers/dell_emc/unity/utils.py
@@ -23,6 +23,7 @@ from oslo_utils import fnmatch
 from oslo_utils import units
 import six
 
+from cinder import coordination
 from cinder import exception
 from cinder.i18n import _
 from cinder.volume import utils as vol_utils
@@ -295,3 +296,10 @@ def match_any(full, patterns):
 
 def is_before_4_1(ver):
     return version.LooseVersion(ver) < version.LooseVersion('4.1')
+
+
+def lock_if(condition, lock_name):
+    if condition:
+        return coordination.synchronized(lock_name)
+    else:
+        return functools.partial
diff --git a/releasenotes/notes/unity-remove-empty-host-17d567dbb6738e4e.yaml b/releasenotes/notes/unity-remove-empty-host-17d567dbb6738e4e.yaml
new file mode 100644
index 00000000000..8faabc32c04
--- /dev/null
+++ b/releasenotes/notes/unity-remove-empty-host-17d567dbb6738e4e.yaml
@@ -0,0 +1,6 @@
+---
+features:
+  - |
+    Dell EMC Unity Driver: Adds support for removing empty host. The new option
+    named `remove_empty_host` could be configured as `True` to notify Unity
+    driver to remove the host after the last LUN is detached from it.