Argument-based methods synchronization in Python

I've been implementing yet another client for a REST-like service and I've spotted a problem I typically haven't faced while working with REST services. I needed to modify some resource -- meaning issuing a PUT request on resource /api/resource/id -- for which changes didn't apply immediately but it rather went into 'building' state or something like that. I had a method like modify_bar(self, resource_id) and things worked great for a single thread, but caused problems with multiple active threads. Obviously, I checked that resource_id is ready for operations, but in the middle of the end of the check and starting of actual work another thread could get in and cause troubles.

Let's check sample code:

	#!/usr/bin/env python
	import threading
	import time
	class Foo(object):
	def modify_bar(self, **kwargs):
	bar_id = kwargs['bar_id']
	print "working on bar id = %s" % bar_id
	time.sleep(3)
	print "done with bar id = %s" % bar_id
	if __name__ == "__main__":
	class ModifyBarThread(threading.Thread):
	def __init__(self, foo, bar_id):
	self.foo = foo
	self.bar_id = bar_id
	threading.Thread.__init__(self)
	def run(self):
	self.foo.modify_bar(bar_id=self.bar_id)
	foo = Foo()
	threads = []
	for i in range(3):
	thread = ModifyBarThread(foo, 1)
	threads.append(thread)
	thread.start()
	[thread.join() for thread in threads]

view raw example.py hosted with ❤ by GitHub

And its output:


(14:46) novel@nov-testing2:~/sync %> ./sync.py
working on bar id = 1
working on bar id = 1
working on bar id = 1
done with bar id = 1
done with bar id = 1
done with bar id = 1
(14:47) novel@nov-testing2:~/sync %>


But the result I wanted to achieve:


(14:46) novel@nov-testing2:~/sync %> ./sync.py
working on bar id = 1
done with bar id = 1
working on bar id = 1
done with bar id = 1
working on bar id = 1
done with bar id = 1
(14:47) novel@nov-testing2:~/sync %>


I've been thinking how to do it without class redesign and API changes and after some time came up with a synchronization solution for methods based on their arguments.

Implementation looks like this:

	import threading
	import time
	def synchronize(field_name):
	def inner_decorator(func):
	def wrapper(self, *args, **kwargs):
	object_id = kwargs[field_name]
	ready_to_start = False
	while not ready_to_start:
	with self._active_bars_lock:
	if object_id not in self._active_bars:
	self._active_bars.append(object_id)
	ready_to_start = True
	if not ready_to_start:
	time.sleep(1.0)
	try:
	ret = func(self, *args, **kwargs)
	finally:
	with self._active_bars_lock:
	self._active_bars.remove(object_id)
	return ret
	return wrapper
	return inner_decorator
	class Foo(object):
	def __init__(self):
	self._active_bars = []
	self._active_bars_lock = threading.Lock()
	@synchronize('bar_id')
	def modify_bar(self, **kwargs):
	bar_id = kwargs['bar_id']
	print "working on bar id = %s" % bar_id
	time.sleep(3)
	print "done with bar id = %s" % bar_id
	if __name__ == "__main__":
	class ModifyBarThread(threading.Thread):
	def __init__(self, foo, bar_id):
	self.foo = foo
	self.bar_id = bar_id
	threading.Thread.__init__(self)
	def run(self):
	self.foo.modify_bar(bar_id=self.bar_id)
	foo = Foo()
	threads = []
	for i in range(3):
	thread = ModifyBarThread(foo, 1)
	threads.append(thread)
	thread.start()
	[thread.join() for thread in threads]

view raw example2.py hosted with ❤ by GitHub

Basically, we keep a list of ids of objects we're working with. When we want to start working with some object we check the list first, if its id in the list it means it's currently being worked on and we have to wait, otherwise we push the id to the list and start working.

And the result is:


(15:09) novel@nov-testing2:~/sync %> ./sync2.py
working on bar id = 1
done with bar id = 1
working on bar id = 1
done with bar id = 1
working on bar id = 1
done with bar id = 1
(15:10) novel@nov-testing2:~/sync %>


For the sake of experiment let's start another thread with different bar_id value to make sure it's not blocked:

	if __name__ == "__main__":
	class ModifyBarThread(threading.Thread):
	def __init__(self, foo, bar_id):
	self.foo = foo
	self.bar_id = bar_id
	threading.Thread.__init__(self)
	def run(self):
	self.foo.modify_bar(bar_id=self.bar_id)
	foo = Foo()
	threads = []
	for i in range(5):
	thread = ModifyBarThread(foo, 0 if i == 2 else 1)
	threads.append(thread)
	thread.start()
	[thread.join() for thread in threads]

view raw example3.py hosted with ❤ by GitHub

And the output now:


(15:13) novel@nov-testing2:~/sync %> ./sync2.py
working on bar id = 1
working on bar id = 0
done with bar id = 1
working on bar id = 1
done with bar id = 0
done with bar id = 1
working on bar id = 1
done with bar id = 1
working on bar id = 1
done with bar id = 1
(15:14) novel@nov-testing2:~/sync %>


As you can see, things are now working as expected.