Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/urllib3/http2/probe.py: 37%

1from __future__ import annotations 

2 

3import threading 

4 

5 

6class _HTTP2ProbeCache: 

7 __slots__ = ( 

8 "_lock", 

9 "_cache_locks", 

10 "_cache_values", 

11 ) 

12 

13 def __init__(self) -> None: 

14 self._lock = threading.Lock() 

15 self._cache_locks: dict[tuple[str, int], threading.RLock] = {} 

16 self._cache_values: dict[tuple[str, int], bool | None] = {} 

17 

18 def acquire_and_get(self, host: str, port: int) -> bool | None: 

19 # By the end of this block we know that 

20 # _cache_[values,locks] is available. 

21 value = None 

22 with self._lock: 

23 key = (host, port) 

24 try: 

25 value = self._cache_values[key] 

26 # If it's a known value we return right away. 

27 if value is not None: 

28 return value 

29 except KeyError: 

30 self._cache_locks[key] = threading.RLock() 

31 self._cache_values[key] = None 

32 

33 # If the value is unknown, we acquire the lock to signal 

34 # to the requesting thread that the probe is in progress 

35 # or that the current thread needs to return their findings. 

36 key_lock = self._cache_locks[key] 

37 key_lock.acquire() 

38 try: 

39 # If the by the time we get the lock the value has been 

40 # updated we want to return the updated value. 

41 value = self._cache_values[key] 

42 

43 # In case an exception like KeyboardInterrupt is raised here. 

44 except BaseException as e: # Defensive: 

45 assert not isinstance(e, KeyError) # KeyError shouldn't be possible. 

46 key_lock.release() 

47 raise 

48 

49 return value 

50 

51 def set_and_release( 

52 self, host: str, port: int, supports_http2: bool | None 

53 ) -> None: 

54 key = (host, port) 

55 key_lock = self._cache_locks[key] 

56 with key_lock: # Uses an RLock, so can be locked again from same thread. 

57 if supports_http2 is None and self._cache_values[key] is not None: 

58 raise ValueError( 

59 "Cannot reset HTTP/2 support for origin after value has been set." 

60 ) # Defensive: not expected in normal usage 

61 

62 self._cache_values[key] = supports_http2 

63 key_lock.release() 

64 

65 def _values(self) -> dict[tuple[str, int], bool | None]: 

66 """This function is for testing purposes only. Gets the current state of the probe cache""" 

67 with self._lock: 

68 return {k: v for k, v in self._cache_values.items()} 

69 

70 def _reset(self) -> None: 

71 """This function is for testing purposes only. Reset the cache values""" 

72 with self._lock: 

73 self._cache_locks = {} 

74 self._cache_values = {} 

75 

76 

77_HTTP2_PROBE_CACHE = _HTTP2ProbeCache() 

78 

79set_and_release = _HTTP2_PROBE_CACHE.set_and_release 

80acquire_and_get = _HTTP2_PROBE_CACHE.acquire_and_get 

81_values = _HTTP2_PROBE_CACHE._values 

82_reset = _HTTP2_PROBE_CACHE._reset 

83 

84__all__ = [ 

85 "set_and_release", 

86 "acquire_and_get", 

87]