How to add an element to a memcached list atomically (in Python)

Question

Here is my simple memcached Python code below:

import memcache memcache_client = memcache.Client(['127.0.0.1:11211'], debug=True) key = "myList" obj = ["A", "B", "C"] memcache_client.set(key, obj) 

Now suppose I want to add the "D" element to a list cached as myList , how can I do this atomically?

I know this is wrong because it is not atomic:

 memcache_client.set(key, memcache_client.get(key) + ["D"]) 

The statement above contains the condition for race. If another thread executes the same instruction at the right time, one of the updates will fail.

How can I solve this race? How can I update a list or dictionary stored in memcached atomically?

Answer 1

Here's the corresponding python client API function

https://cloud.google.com/appengine/docs/python/memcache/clientclass#Client_cas

Also here is a good tutorial from Guido van Rossum. Hopefully he will explain things better in python than me;)

Here's what the code looks like in your case:

 memcache_client = memcache.Client(['127.0.0.1:11211'], debug=True) key = "myList" while True: # Retry loop, probably it should be limited to some reasonable retries obj = memcache_client.gets(key) assert obj is not None, 'Uninitialized object' if memcache_client.cas(key, obj + ["D"]): break 

The whole workflow remains the same: first you retrieve the value (with some internal information associated with the key), then modify the extracted value, and then try to update it in memcache. The only difference is that the value (in fact, the key / value pair) is checked for the fact that it was not changed simultaneously from the parallel process. In the latter case, the call fails, and you must repeat the workflow from the very beginning. Also, if you have a multi-threaded application, then each memcache_client instance should probably be thread-local.

Also, remember that there are incr () and decr () methods for simple integer counters that are atomic in nature.

Answer 2

If you do not want to receive the race condition, you must use the Lock primitive from the streaming module. for instance

 lock = threading.Lock() def thread_func(): obj = get_obj() lock.acquire() memcache_client.set(key, obj) lock.release()