Thread-safe LRU cache in Python

Question

I've written a simple LRU cache class and I am trying to make it thread-safe. My thoughts are that I just need to wrap the code that updates the ordered dict in a lock so that if any thread is writing to the ordered dict, all other writes/reads need to wait. Is that correct?

Also, what is the best way to test that something is thread-safe?

import threading
from collections import OrderedDict

class Cache:
  def __init__(self, capacity):
    self._count = 0
    self._capacity = capacity
    self._store = OrderedDict()
    self._lock = threading.Lock()
    
  def count(self):
    return self._count

  def capacity(self):
    return self._capacity

  def get(self, key):
    if key in self._store:
      self._store.move_to_end(key)
    return self._store[key]

  def set(self, key, value):
    with self._lock: # this was the only place i thought that needed locking
      if key not in self._store:
        if self._count >= self._capacity:
          self._store.popitem(last=False)
          self._count -= 1
        self._store[key] = value
        self._count += 1
      else:
        self._store[key] = value
        self._store.move_to_end(key)

  def containsKey(self, key):
    return key in self._store

Answer 1

The use of self._store isn't thread-safe, because we are only locking its writers. We need to ensure that readers don't see it in an inconsistent state, too.

Answer 2

I've experienced that key in self._store is not a thread-safe way to check if a key exists in a dictionary, even when wrapped in a with lock.

Replacing key in self._store with key in self._store.keys(), and wrapping all accessors and setters on self._store with with self._lock should be sufficient.