diff --git a/neutron/db/securitygroups_db.py b/neutron/db/securitygroups_db.py
index a610d6d4899..40b63c43598 100644
--- a/neutron/db/securitygroups_db.py
+++ b/neutron/db/securitygroups_db.py
@@ -83,6 +83,11 @@ class SecurityGroupDbMixin(ext_sg.SecurityGroupPluginBase):
 
         if not default_sg:
             self._ensure_default_security_group(context, tenant_id)
+        else:
+            existing_def_sg_id = self._get_default_sg_id(context, tenant_id)
+            if existing_def_sg_id is not None:
+                # default already exists, return it
+                return self.get_security_group(context, existing_def_sg_id)
 
         with db_api.autonested_transaction(context.session):
             security_group_db = sg_models.SecurityGroup(id=s.get('id') or (
@@ -661,27 +666,30 @@ class SecurityGroupDbMixin(ext_sg.SecurityGroupPluginBase):
         port[ext_sg.SECURITYGROUPS] = (security_group_ids and
                                        list(security_group_ids) or [])
 
-    def _ensure_default_security_group(self, context, tenant_id):
-        """Create a default security group if one doesn't exist.
-
-        :returns: the default security group id for given tenant.
-        """
+    def _get_default_sg_id(self, context, tenant_id):
         try:
             query = self._model_query(context, sg_models.DefaultSecurityGroup)
             default_group = query.filter_by(tenant_id=tenant_id).one()
             return default_group['security_group_id']
         except exc.NoResultFound:
-            security_group = {
-                'security_group':
-                    {'name': 'default',
-                     'tenant_id': tenant_id,
-                     'description': _('Default security group')}
-            }
-            # starting a transaction before create to avoid db retries
-            with context.session.begin(subtransactions=True):
-                sg_id = self.create_security_group(
-                    context, security_group, default_sg=True)['id']
-            return sg_id
+            pass
+
+    def _ensure_default_security_group(self, context, tenant_id):
+        """Create a default security group if one doesn't exist.
+
+        :returns: the default security group id for given tenant.
+        """
+        existing = self._get_default_sg_id(context, tenant_id)
+        if existing is not None:
+            return existing
+        security_group = {
+            'security_group':
+                {'name': 'default',
+                 'tenant_id': tenant_id,
+                 'description': _('Default security group')}
+        }
+        return self.create_security_group(context, security_group,
+                                          default_sg=True)['id']
 
     def _get_security_groups_on_port(self, context, port):
         """Check that all security groups on port belong to tenant.
diff --git a/neutron/plugins/ml2/plugin.py b/neutron/plugins/ml2/plugin.py
index aa076850487..14c45aa020d 100644
--- a/neutron/plugins/ml2/plugin.py
+++ b/neutron/plugins/ml2/plugin.py
@@ -735,7 +735,6 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
         tenant_id = net_data['tenant_id']
         session = context.session
         with session.begin(subtransactions=True):
-            self._ensure_default_security_group(context, tenant_id)
             net_db = self.create_network_db(context, network)
             result = self._make_network_dict(net_db, process_extensions=False,
                                              context=context)
@@ -771,6 +770,8 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
     @utils.transaction_guard
     @db_api.retry_if_session_inactive()
     def create_network(self, context, network):
+        self._ensure_default_security_group(context,
+                                            network['network']['tenant_id'])
         result, mech_context = self._create_network_db(context, network)
         kwargs = {'context': context, 'network': result}
         registry.notify(resources.NETWORK, events.AFTER_CREATE, self, **kwargs)
@@ -787,6 +788,8 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
     @utils.transaction_guard
     @db_api.retry_if_session_inactive()
     def create_network_bulk(self, context, networks):
+        tenants = {n['network']['tenant_id'] for n in networks['networks']}
+        map(lambda t: self._ensure_default_security_group(context, t), tenants)
         objects = self._create_bulk_ml2(attributes.NETWORK, context, networks)
         return [obj['result'] for obj in objects]
 
@@ -1240,6 +1243,9 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
         if not attrs.get('status'):
             attrs['status'] = const.PORT_STATUS_DOWN
 
+        # NOTE(kevinbenton): triggered outside of transaction since it
+        # emits 'AFTER' events if it creates.
+        self._ensure_default_security_group(context, attrs['tenant_id'])
         session = context.session
         with session.begin(subtransactions=True):
             dhcp_opts = attrs.get(edo_ext.EXTRADHCPOPTS, [])
diff --git a/neutron/tests/unit/plugins/ml2/test_security_group.py b/neutron/tests/unit/plugins/ml2/test_security_group.py
index d3dfd1c3237..c0cdb7c1d85 100644
--- a/neutron/tests/unit/plugins/ml2/test_security_group.py
+++ b/neutron/tests/unit/plugins/ml2/test_security_group.py
@@ -20,6 +20,9 @@ import mock
 from neutron_lib import constants as const
 from neutron_lib.plugins import directory
 
+from neutron.callbacks import events
+from neutron.callbacks import registry
+from neutron.callbacks import resources
 from neutron import context
 from neutron.extensions import securitygroup as ext_sg
 from neutron.tests import tools
@@ -148,6 +151,20 @@ class TestMl2SecurityGroups(Ml2SecurityGroupsTestCase,
             # the or_ function should only have one argument
             or_mock.assert_called_once_with(mock.ANY)
 
+    def test_security_groups_created_outside_transaction(self):
+        def record_after_state(r, e, t, context, *args, **kwargs):
+            self.was_active = context.session.is_active
+
+        registry.subscribe(record_after_state, resources.SECURITY_GROUP,
+                           events.AFTER_CREATE)
+        with self.subnet() as s:
+            self.assertFalse(self.was_active)
+            self._delete(
+                'security-groups',
+                self._list('security-groups')['security_groups'][0]['id'])
+            with self.port(subnet=s):
+                self.assertFalse(self.was_active)
+
 
 class TestMl2SGServerRpcCallBack(
     Ml2SecurityGroupsTestCase,