123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150 |
- # 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
|