From af0b13b6d919c8c9ddf3a803eef21cd1a00a36ce Mon Sep 17 00:00:00 2001 From: Mike Bayer Date: Mon, 30 Nov 2020 12:50:51 -0500 Subject: Send deterministic ordering into unit of work topological Improved the unit of work topological sorting system such that the toplogical sort is now deterministic based on the sorting of the input set, which itself is now sorted at the level of mappers, so that the same inputs of affected mappers should produce the same output every time, among mappers / tables that don't have any dependency on each other. This further reduces the chance of deadlocks as can be observed in a flush that UPDATEs among multiple, unrelated tables such that row locks are generated. topological.sort() has been made "deterministic" in all cases by using a separate list + set. Fixes: #5735 Change-Id: I073103df414dba549e46605b394f8ccae6e80d0e --- lib/sqlalchemy/orm/dependency.py | 1 + lib/sqlalchemy/orm/mapper.py | 4 ++++ lib/sqlalchemy/orm/unitofwork.py | 32 ++++++++++++++++++++++++-------- 3 files changed, 29 insertions(+), 8 deletions(-) (limited to 'lib/sqlalchemy/orm') diff --git a/lib/sqlalchemy/orm/dependency.py b/lib/sqlalchemy/orm/dependency.py index d4680e394..9c2c5ade3 100644 --- a/lib/sqlalchemy/orm/dependency.py +++ b/lib/sqlalchemy/orm/dependency.py @@ -43,6 +43,7 @@ class DependencyProcessor(object): else: self._passive_update_flag = attributes.PASSIVE_OFF + self.sort_key = "%s_%s" % (self.parent._sort_key, prop.key) self.key = prop.key if not self.prop.synchronize_pairs: raise sa_exc.ArgumentError( diff --git a/lib/sqlalchemy/orm/mapper.py b/lib/sqlalchemy/orm/mapper.py index e73993e50..e50183894 100644 --- a/lib/sqlalchemy/orm/mapper.py +++ b/lib/sqlalchemy/orm/mapper.py @@ -574,6 +574,10 @@ class Mapper( """ self.class_ = util.assert_arg_type(class_, type, "class_") + self._sort_key = "%s.%s" % ( + self.class_.__module__, + self.class_.__name__, + ) self.class_manager = None diff --git a/lib/sqlalchemy/orm/unitofwork.py b/lib/sqlalchemy/orm/unitofwork.py index 9c67130ce..868f8e087 100644 --- a/lib/sqlalchemy/orm/unitofwork.py +++ b/lib/sqlalchemy/orm/unitofwork.py @@ -422,6 +422,10 @@ class UOWTransaction(object): def execute(self): postsort_actions = self._generate_actions() + postsort_actions = sorted( + postsort_actions, + key=lambda item: item.sort_key, + ) # sort = topological.sort(self.dependencies, postsort_actions) # print "--------------" # print "\ndependencies:", self.dependencies @@ -431,9 +435,10 @@ class UOWTransaction(object): # execute if self.cycles: - for set_ in topological.sort_as_subsets( + for subset in topological.sort_as_subsets( self.dependencies, postsort_actions ): + set_ = set(subset) while set_: n = set_.pop() n.execute_aggregate(self, set_) @@ -542,10 +547,15 @@ class PostSortRec(object): class ProcessAll(IterateMappersMixin, PostSortRec): - __slots__ = "dependency_processor", "isdelete", "fromparent" + __slots__ = "dependency_processor", "isdelete", "fromparent", "sort_key" def __init__(self, uow, dependency_processor, isdelete, fromparent): self.dependency_processor = dependency_processor + self.sort_key = ( + "ProcessAll", + self.dependency_processor.sort_key, + isdelete, + ) self.isdelete = isdelete self.fromparent = fromparent uow.deps[dependency_processor.parent.base_mapper].add( @@ -582,11 +592,12 @@ class ProcessAll(IterateMappersMixin, PostSortRec): class PostUpdateAll(PostSortRec): - __slots__ = "mapper", "isdelete" + __slots__ = "mapper", "isdelete", "sort_key" def __init__(self, uow, mapper, isdelete): self.mapper = mapper self.isdelete = isdelete + self.sort_key = ("PostUpdateAll", mapper._sort_key, isdelete) @util.preload_module("sqlalchemy.orm.persistence") def execute(self, uow): @@ -598,10 +609,11 @@ class PostUpdateAll(PostSortRec): class SaveUpdateAll(PostSortRec): - __slots__ = ("mapper",) + __slots__ = ("mapper", "sort_key") def __init__(self, uow, mapper): self.mapper = mapper + self.sort_key = ("SaveUpdateAll", mapper._sort_key) assert mapper is mapper.base_mapper @util.preload_module("sqlalchemy.orm.persistence") @@ -634,10 +646,11 @@ class SaveUpdateAll(PostSortRec): class DeleteAll(PostSortRec): - __slots__ = ("mapper",) + __slots__ = ("mapper", "sort_key") def __init__(self, uow, mapper): self.mapper = mapper + self.sort_key = ("DeleteAll", mapper._sort_key) assert mapper is mapper.base_mapper @util.preload_module("sqlalchemy.orm.persistence") @@ -670,10 +683,11 @@ class DeleteAll(PostSortRec): class ProcessState(PostSortRec): - __slots__ = "dependency_processor", "isdelete", "state" + __slots__ = "dependency_processor", "isdelete", "state", "sort_key" def __init__(self, uow, dependency_processor, isdelete, state): self.dependency_processor = dependency_processor + self.sort_key = ("ProcessState", dependency_processor.sort_key) self.isdelete = isdelete self.state = state @@ -705,11 +719,12 @@ class ProcessState(PostSortRec): class SaveUpdateState(PostSortRec): - __slots__ = "state", "mapper" + __slots__ = "state", "mapper", "sort_key" def __init__(self, uow, state): self.state = state self.mapper = state.mapper.base_mapper + self.sort_key = ("ProcessState", self.mapper._sort_key) @util.preload_module("sqlalchemy.orm.persistence") def execute_aggregate(self, uow, recs): @@ -732,11 +747,12 @@ class SaveUpdateState(PostSortRec): class DeleteState(PostSortRec): - __slots__ = "state", "mapper" + __slots__ = "state", "mapper", "sort_key" def __init__(self, uow, state): self.state = state self.mapper = state.mapper.base_mapper + self.sort_key = ("DeleteState", self.mapper._sort_key) @util.preload_module("sqlalchemy.orm.persistence") def execute_aggregate(self, uow, recs): -- cgit v1.2.1