Skip to content

Instantly share code, notes, and snippets.

@orlp
Created December 14, 2011 10:42
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save orlp/1476083 to your computer and use it in GitHub Desktop.
Save orlp/1476083 to your computer and use it in GitHub Desktop.
# set-like class with automatic expiration of entries.
import time
import sys
# high precision timer function, cross-platform
if sys.platform == "win32":
timer_func = time.clock # on Windows, the best timer is time.clock()
else:
timer_func = time.time # on most other platforms the best timer is time.time()
class ExpirationSet:
def __init__(self, expiration_time, callback=None):
self.expiration_time = expiration_time
self.callback = callback
self.data = {}
# adds key, overwriting previous time if the key existed
def add(self, key):
self.data[key] = timer_func()
# removes key, throwing error if key wasn't found and calls the callback
def remove(self, key):
del self.data[key]
if self.callback:
self.callback(key, False)
# removes key, if the key wasn't found no error is thrown and calls the callback
def discard(self, key):
if key in self.data:
del self.data[key]
if self.callback:
self.callback(key, False)
# remove all keys that were expired and call the callback for them, if set
def expire_keys(self, expiration_time = None):
now = timer_func()
if not expiration_time:
expiration_time = self.expiration_time
# first loop over the dict, collecting our data
delete_keys = [key for key, starttime in self.data.items() if (now - starttime) > self.expiration_time]
# then delete the keys
for key in delete_keys:
del self.data[key]
# and finally call the callbacks
if self.callback:
for key in delete_keys:
self.callback(key, True)
# some general functions
def clear(self):
self.data.clear()
def copy(self):
expirationset = ExpirationSet(self.expiration_time, self.callback)
expirationset.data = self.data.copy()
return self
# and some internal functions
def __contains__(self, key):
return key in self.data
def __iter__(self):
return self.data.iterkeys()
def __len__(self):
return len(self.data)
@Medo42
Copy link

Medo42 commented Dec 14, 2011

Apparently, you didn't like my expirationset for different reasons than me. The only real problem I see with mine, having checked it again, is that the time at which the callbacks are called can be a bit surprising. Having the entries expire automagically does not seem bad practice to me as such, because I only included access functions where this would not cause surprising behavior (e.g. no remove - because you wouldn't have an intuitive way to know whether the entry actually still exists)

As for efficiency, by using the OrderedDict cleanup_stale was able to work by only looking at the keys that actually needed removing, plus one. That means calling it often is not a big problem, since it will then usually just have to check one or two entries, while your implementation always checks all entries. In its use as an anti-flooding measure (see RECENT_ENDPOINTS), this makes a huge (theoretical) difference, because with my old code, receiving and rejecting a quick burst of UDP datagrams on the lobby will have O(n+m) complexity (where m is the number of entries which need to be expired and n the number of datagrams in the burst), while using your implementation would be O(n*s) (where s = number of entries in the expirationset. Notice that s>m).

Of course none of the efficiency questions actually have any impact on the GG2 lobby given its low traffic, but you did bring it up and suggested this was an improvement.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment