Fixed #16649 -- Refactored save_base logic

Model.save() will use UPDATE - if not updated - INSERT instead of
SELECT - if found UPDATE else INSERT. This should save a query when
updating, but will cost a little when inserting model with PK set.

Also fixed #17341 -- made sure .save() commits transactions only after
the whole model has been saved. This wasn't the case in model
inheritance situations.

The save_base implementation was refactored into multiple methods.
A typical chain for inherited save is:
save_base()
    _save_parents(self)
        for each parent:
            _save_parents(parent)
            _save_table(parent)
    _save_table(self)

tests/basic/tests.py

@@ -1,11 +1,13 @@
 from __future__ import absolute_import, unicode_literals
 
 from datetime import datetime
+import threading
 
 from django.core.exceptions import ObjectDoesNotExist, MultipleObjectsReturned, FieldError
+from django.db import connections, DEFAULT_DB_ALIAS
 from django.db.models.fields import Field, FieldDoesNotExist
 from django.db.models.query import QuerySet, EmptyQuerySet, ValuesListQuerySet
-from django.test import TestCase, skipIfDBFeature, skipUnlessDBFeature
+from django.test import TestCase, TransactionTestCase, skipIfDBFeature, skipUnlessDBFeature
 from django.utils import six
 from django.utils.translation import ugettext_lazy
 
@@ -690,4 +692,29 @@ class ModelTest(TestCase):
     def test_invalid_qs_list(self):
         qs = Article.objects.order_by('invalid_column')
         self.assertRaises(FieldError, list, qs)
-        self.assertRaises(FieldError, list, qs)
+        self.assertRaises(FieldError, list, qs)
+
+class ConcurrentSaveTests(TransactionTestCase):
+    @skipUnlessDBFeature('test_db_allows_multiple_connections')
+    def test_concurrent_delete_with_save(self):
+        """
+        Test fetching, deleting and finally saving an object - we should get
+        an insert in this case.
+        """
+        a = Article.objects.create(headline='foo', pub_date=datetime.now())
+        exceptions = []
+        def deleter():
+            try:
+                # Do not delete a directly - doing so alters its state.
+                Article.objects.filter(pk=a.pk).delete()
+                connections[DEFAULT_DB_ALIAS].commit_unless_managed()
+            except Exception as e:
+                exceptions.append(e)
+            finally:
+                connections[DEFAULT_DB_ALIAS].close()
+        self.assertEqual(len(exceptions), 0)
+        t = threading.Thread(target=deleter)
+        t.start()
+        t.join()
+        a.save()
+        self.assertEqual(Article.objects.get(pk=a.pk).headline, 'foo')