diff options
Diffstat (limited to 'synapse/util/linked_list.py')
-rw-r--r-- | synapse/util/linked_list.py | 150 |
1 files changed, 0 insertions, 150 deletions
diff --git a/synapse/util/linked_list.py b/synapse/util/linked_list.py deleted file mode 100644 index 8efbf061aa..0000000000 --- a/synapse/util/linked_list.py +++ /dev/null @@ -1,150 +0,0 @@ -# Copyright 2021 The Matrix.org Foundation C.I.C. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -"""A circular doubly linked list implementation. -""" - -import threading -from typing import Generic, Optional, Type, TypeVar - -P = TypeVar("P") -LN = TypeVar("LN", bound="ListNode") - - -class ListNode(Generic[P]): - """A node in a circular doubly linked list, with an (optional) reference to - a cache entry. - - The reference should only be `None` for the root node or if the node has - been removed from the list. - """ - - # A lock to protect mutating the list prev/next pointers. - _LOCK = threading.Lock() - - # We don't use attrs here as in py3.6 you can't have `attr.s(slots=True)` - # and inherit from `Generic` for some reason - __slots__ = [ - "cache_entry", - "prev_node", - "next_node", - ] - - def __init__(self, cache_entry: Optional[P] = None) -> None: - self.cache_entry = cache_entry - self.prev_node: Optional[ListNode[P]] = None - self.next_node: Optional[ListNode[P]] = None - - @classmethod - def create_root_node(cls: Type["ListNode[P]"]) -> "ListNode[P]": - """Create a new linked list by creating a "root" node, which is a node - that has prev_node/next_node pointing to itself and no associated cache - entry. - """ - root = cls() - root.prev_node = root - root.next_node = root - return root - - @classmethod - def insert_after( - cls: Type[LN], - cache_entry: P, - node: "ListNode[P]", - ) -> LN: - """Create a new list node that is placed after the given node. - - Args: - cache_entry: The associated cache entry. - node: The existing node in the list to insert the new entry after. - """ - new_node = cls(cache_entry) - with cls._LOCK: - new_node._refs_insert_after(node) - return new_node - - def remove_from_list(self) -> None: - """Remove this node from the list.""" - with self._LOCK: - self._refs_remove_node_from_list() - - # We drop the reference to the cache entry to break the reference cycle - # between the list node and cache entry, allowing the two to be dropped - # immediately rather than at the next GC. - self.cache_entry = None - - def move_after(self, node: "ListNode[P]") -> None: - """Move this node from its current location in the list to after the - given node. - """ - with self._LOCK: - # We assert that both this node and the target node is still "alive". - assert self.prev_node - assert self.next_node - assert node.prev_node - assert node.next_node - - assert self is not node - - # Remove self from the list - self._refs_remove_node_from_list() - - # Insert self back into the list, after target node - self._refs_insert_after(node) - - def _refs_remove_node_from_list(self) -> None: - """Internal method to *just* remove the node from the list, without - e.g. clearing out the cache entry. - """ - if self.prev_node is None or self.next_node is None: - # We've already been removed from the list. - return - - prev_node = self.prev_node - next_node = self.next_node - - prev_node.next_node = next_node - next_node.prev_node = prev_node - - # We set these to None so that we don't get circular references, - # allowing us to be dropped without having to go via the GC. - self.prev_node = None - self.next_node = None - - def _refs_insert_after(self, node: "ListNode[P]") -> None: - """Internal method to insert the node after the given node.""" - - # This method should only be called when we're not already in the list. - assert self.prev_node is None - assert self.next_node is None - - # We expect the given node to be in the list and thus have valid - # prev/next refs. - assert node.next_node - assert node.prev_node - - prev_node = node - next_node = node.next_node - - self.prev_node = prev_node - self.next_node = next_node - - prev_node.next_node = self - next_node.prev_node = self - - def get_cache_entry(self) -> Optional[P]: - """Get the cache entry, returns None if this is the root node (i.e. - cache_entry is None) or if the entry has been dropped. - """ - return self.cache_entry |