Add a Fair lock recipe?

[grantjenks]

Some prototyping for a fair lock recipe.

[grantjenks]

The tricky part is what to do when one of the cache key expires but not both of them. Maybe there needs to be a single cache key to track tickets and serving. It still seems the cache key could expire while someone is holding the lock. I'm not sure how they later discover that the lock was lost.

I'm sure there's research on this problem. Need to look there rather than re-implementing.

[grantjenks]

Brainstorming, came up with:

classFairRLock:
"""Recipe for cross-process and cross-thread re-entrant lock. Assumes the key will not be evicted. Set the eviction policy to 'none' on the cache to guarantee the key is not evicted. >>> import diskcache >>> cache = diskcache.Cache() >>> rlock = FairRLock(cache, 'user-123') >>> rlock.acquire() >>> rlock.acquire() >>> rlock.release() >>> with rlock: ... pass >>> rlock.release() """def__init__(self, cache, key, expire=None, tag=None):
self._cache=cacheself._key=keyself._expire=expireself._tag=tagvalue= {
'tickets': 0,
'current': 1,
'pid_tid': '',
'time': 0,
'count': 0,
}
self._cache.add(self._key, value, tag=self._tag)
def_acquire_ticket(self):
withself._cache.transact():
state=self._cache.get(self._key)
state['tickets'] +=1self._cache.set(self._key, state, tag=self._tag)
returnstate['tickets']
def_release_ticket(self, state):
state['current'] +=1state['pid_tid'] =''state['time'] =0state['count'] =0self._cache.set(self._key, state, tag=self._tag)
def_get_pid_tid(self):
pid_tid=f'{os.getpid()}-{threading.get_ident()}'returnpid_tiddefacquire(self):
pid_tid=self._get_pid_tid()
ticket=self._acquire_ticket()
whileTrue:
withself._cache.transact():
state=self._cache[self._key]
ifstate['current'] ==ticket:
state['pid_tid'] =pid_tidstate['time'] =time.monotonic()
state['count'] +=1self._cache.set(self._key, state, tag=self._tag)
breakexpired= (
self._expireisnotNoneandtime.monotonic() -state['time'] >self._expire
)
ifexpired:
self._release_ticket(state)
time.sleep(0.01)
returnticketdefrelease(self, ticket):
pid_tid=self._get_pid_tid()
withself._cache.transact():
state=self._cache.get(self._key)
ifpid_tid!=state['pid_tid']:
returnifstate['count'] >1:
state['time'] =time.monotonic()
state['count'] -=1self._cache.set(self._key, state, tag=self._tag)
assertstate['count'] ==1self._release_ticket(state)
def__enter__(self):
self.acquire()
returnselfdef__exit__(self, exc_type, exc_val, exc_tb):
self.release()

But I think this'll break because the current ticket number is only incremented by release(). If one of the release() calls doesn't happen, then everything will be stuck.

[grantjenks]

Change the code such that if the lock expires then it is immediately reacquired by another thread or process in the acquire() method rather than by first releasing it.

Update the release ticket code to set time correctly. This will allow the acquire method to skip tickets if they are not acquired after a long time.

[grantjenks]

When first acquiring a ticket, if the pid_tid already holds the lock, then don't increment the tickets count.