Threads are processes which run in parallel to other threads. In a utopian scenario, if you split a big process in 2 threads, these threads will run in parallel so it would take half the time.<\/p>\n
This is not true in most cases. Using CPython, there is a mutex that prevents multiple native threads from executing Python byte codes at once. It’s called GIL (global interpreter lock). This lock is necessary mainly because CPython’s memory management is not thread-safe, but notice that I\/O, image processing, and other potentially blocking operations, happen outside the GIL, so it will only become a bottle neck in processes that spend a lot of time inside the GIL.<\/p>\n
In most applications nowadays, concurrency is something we all must be able to handle. Mostly in web applications, where one request usually starts a thread, we need to have concurrency and threading in mind so we can write our programs accordingly.<\/p>\n
Threading is also a good solution to optimize response times. Given a scenario in which we have to process 4 million objects, a good thing to do would be divide them in 4 groups of a million objects and process them in 4 separated threads.
\n
\n[ulp id=’R7QVpFMZmjosABLC’]<\/p>\n
The The process of spawning a thread is pretty simple. We just need to call a method called When we run this script we’ll see an output that looks like:<\/p>\n So, let’s see what is going on there.<\/p>\n We have a variable called Then we create two threads, each of them containing our The newer The most used procedure for spawning a thread using this module, is defining a subclass of the Once you’ve got your class, you just instantiate an object of it and call the method called Then we run it an see something like:<\/p>\n Of course, we don’t need to name our threads like that. Each Thread instance has a name with a default value that can be changed as the thread is created. Naming threads is useful in server processes with multiple service threads handling different operations. So, let’s change our code to avoid reinventing the wheel. In our constructor:<\/p>\n Then the output will look like:<\/p>\n But it usually isn’t so simple, the logic isn’t in the run method ouf our Our problem here is that we don’t have our thread name outside of it, and we don’t want that information going around in our function’s signature. But the By executing Now, let’s talk about daemon threads. Until now, our programs waited for every thread to end before actually terminating, but sometimes we don’t want that behaviour. If we have a thread, pushing status or metrics to a service, we usually don’t care if it has finished or not when we shut down our program, and maybe we don’t want to explicitly terminate it before exiting.<\/p>\n Daemon threads run without blocking the main thread from exiting. They are useful when we have services where there may not be an easy way to interrupt the thread or where letting the thread die in the middle of its work does not lose or corrupt data.<\/p>\n To spawn a daemon thread, we just spawn a normal thread an call its We are now grabbing the last thread we create, and making it daemon. The output will now look like:<\/p>\n As you can see, the main thread is not waiting for Thread-6 to finish before exiting, daemon threads are terminated when the main thread finished its execution.<\/p>\n Let’s write something that resembles a real-life problem solution. Let’s make a script, that given an array of URL’s, crawls them and saves the html in files.<\/p>\n site-crawler.py<\/em><\/span><\/p>\n So, what do we’ve got here? This is a script that asks the user to input a host and the absolute path of the file to which it’ll write the site’s output html, and measures the time it’ll take. The script as is gives the following input:<\/p>\n As the log shows us, threads are running in parallel, a thread runs while the others make I\/O operations (write to a file, or connect via http to a server). Now, there is a line commented on the See? First it crawls google Argentina, then Brazil, and so on. You sure are wondering why would someone do this. Well… this is not the only use case of the Well, there is another way, let’s make these threads daemon, just to experiment a little bit, and wait for all of them to finish before exiting the main thread:<\/p>\n site-crawler.py<\/em><\/span><\/p>\n Here, we are creating every thread as daemon<\/em>, but we are enumerating every active thread by calling Another thing that’s worthy of being pointed out, is the existence of the class If we execute it:<\/p>\n Here, we are creating two timers, both execute the same function after 3 seconds. Then we wait 2 seconds and cancel one of them. In the output we can see only one of the timers executed the delayed function.<\/p>\n This is useful on scenarios where we need to execute some process if something didn’t happen in an interval of time, or even for scheduling.<\/p>\n Now, we all know that the idea of using threads is making tasks independent from each other, but some times we need for a thread to wait for an event caused by another. Python provides a way of signaling between threads. To experiment with this, we’ll make a race:<\/p>\n race.py<\/em><\/span><\/p>\n Here, we created a subclass of We create a couple of threads, an then set the event, so they all try to start at the same time, the output is interesting:<\/p>\n first run output<\/em><\/span><\/p>\n second run output<\/em><\/span><\/p>\n Here we can see how the rabbit won the first race, but ended last on the second. Either he got tired, or our event behaves just as we wanted it to behave.<\/p>\n If we didn’t use the event, and start each thread in a loop, the first thread would have an advantage of milliseconds over the last one. And we all know every millisecond counts on computer times.<\/p>\n Sometimes we have a couple threads accessing the same resource, and if it’s not thread safe, we don’t want threads to access it at the same time. One solution to this problem could be locking.<\/p>\n A side note: Python\u2019s built-in data structures (lists, dictionaries, etc.) are thread-safe as a side-effect of having atomic byte-codes for manipulating them (the GIL is not released in the middle of an update). Other data structures implemented in Python, or simpler types like integers and floats, don\u2019t have that protection.<\/p>\n Let’s imagine, just to make a fun example, that dictionaries in python are not thread safe. We’ll make an on-memory repository and make a couple of threads read and write data to it:<\/p>\n locking.py<\/em><\/span><\/p>\n As you see, both resources (purchases and products) are extending a class This lock will guarantee that only one thread at a time can access one repository. One thing to notice is how the lock is released in a Now, let’s execute this code and input something like:<\/p>\n As you see, the logger was configured to output its log to a file. We don’t care about the logging of the There are our 8 Another side note here: Let’s imagine another scenario. We have a thread and a resource. The thread locks de resource to write some data, and in the middle, it needs to lock it again to read some other data without releasing the first lock. Well, we have a problem here… Normal Locks implement the context manager API and are compatible with the with statement. Using with removes the need to explicitly acquire and release the lock. Let’s modify our locking.py<\/em><\/span><\/p>\n And the behavior remains the same:<\/p>\n Using a lock like that, ensures only one thread at a time can access a resource, but imagine a data base access. Maybe you have a connection pool of, at most, 100 connections. In this case you want concurrent access, but you don’t want more than a 100 threads to access this resource at once. locking.py<\/em><\/span><\/p>\n We often need data to be only accessible from one thread (eg.: identifiers of the current process), python provides the process-identifier.py<\/em><\/span><\/p>\n It’s pretty easy to use, a very useful. If you run it you’ll see something like:<\/p>\n Every thread has to initialize This was an example on Threading in Python.<\/p>\n_thread<\/code> module is very effective for low level threading, let’s see an example to understand the concept. But keep in mind that since Python 2.4, this module is not used anymore.<\/p>\nstart_new_thread<\/code>, available in the _thread<\/code> module, which receives a function, and arguments to pass to it. It returns immediately and a thread will run in parallel. Let’s see:<\/p>\n\r\nimport _thread as thread\r\nimport time\r\n\r\nexecuted_count = 0\r\n\r\n\r\n# Define a function for the thread\r\ndef print_time(thread_name, delay):\r\n global executed_count\r\n count = 0\r\n while count < 5:\r\n time.sleep(delay)\r\n count += 1\r\n print("%s: %s" % (thread_name, time.ctime(time.time())))\r\n executed_count += 1\r\n\r\n\r\n# Create two threads as follows\r\ntry:\r\n threads = [thread.start_new_thread(print_time, ("Thread-1", 2,)),\r\n thread.start_new_thread(print_time, ("Thread-2", 4,))]\r\nexcept:\r\n print("Error: unable to start thread")\r\n\r\nwhile executed_count < 2:\r\n pass\r\n<\/pre>\n\r\nThread-1: Mon Dec 21 12:55:23 2015\r\nThread-2: Mon Dec 21 12:55:25 2015\r\nThread-1: Mon Dec 21 12:55:25 2015\r\nThread-1: Mon Dec 21 12:55:27 2015\r\nThread-2: Mon Dec 21 12:55:29 2015\r\nThread-1: Mon Dec 21 12:55:29 2015\r\nThread-1: Mon Dec 21 12:55:31 2015\r\nThread-2: Mon Dec 21 12:55:33 2015\r\nThread-2: Mon Dec 21 12:55:37 2015\r\nThread-2: Mon Dec 21 12:55:41 2015\r\n<\/pre>\n
executed_count<\/code> initialized in zero. Then there is a function called print_time<\/code> which receives the name of the thread and a delay, and every {delay} seconds it will print the date with the thread name, five times. Then it adds one to the executed_count<\/code> variable.<\/p>\nprint_time<\/code> function, with a name and a delay assigned to them. And we have a while which makes sure the program won’t exit until executed_count<\/code> is equal or greater than 2.<\/p>\n2. Python threading module<\/h2>\n
threading<\/code> module included with Python 2.4 provides much more powerful, high-level support for threads than the _thread<\/code> module.<\/p>\n2.1. Extending Thread<\/h3>\n
Thread<\/code> class. Once you’ve done it, you should override the __init__<\/code> and run<\/code> methods.<\/p>\nstart<\/code>. Let’s see an example of this:<\/p>\n\r\nimport threading\r\n\r\nimport time\r\n\r\n\r\nclass MyThread(threading.Thread):\r\n def __init__(self, name, sleep_time):\r\n threading.Thread.__init__(self)\r\n self.name = name\r\n self.sleep_time = sleep_time\r\n\r\n def run(self):\r\n print(\"{} start\".format(self.name))\r\n time.sleep(self.sleep_time)\r\n print(\"{} end\".format(self.name))\r\n\r\n\r\nthreads = [MyThread(\"Thread-{}\".format(i), i) for i in range(1, 4)]\r\nfor t in threads:\r\n t.start()\r\n<\/pre>\n\r\nThread-1 start\r\nThread-2 start\r\nThread-3 start\r\nThread-1 end\r\nThread-2 end\r\nThread-3 end\r\n<\/pre>\n
\r\ndef __init__(self, sleep_time):\r\n threading.Thread.__init__(self)\r\n threading.Thread.__init__(self)\r\n self.sleep_time = sleep_time\r\n<\/pre>\n
\r\nThread-2 start\r\nThread-4 start\r\nThread-6 start\r\nThread-2 end\r\nThread-4 end\r\nThread-6 end\r\n<\/pre>\n
Thread<\/code> class, so let’s make it a little more real and do the print<\/code> and sleep in a method outside our MyThread<\/code>.<\/p>\n2.2. Getting Current Thread Information<\/h3>\n
threading<\/code> module, has some methods that give us access to the current thread information:<\/p>\n\r\nimport threading\r\nimport time\r\n\r\n\r\ndef my_logic(sleep_time):\r\n thread_name = threading.current_thread().getName()\r\n print(\"{} start\".format(thread_name))\r\n time.sleep(sleep_time)\r\n print(\"{} end\".format(thread_name))\r\n\r\n\r\nclass MyThread(threading.Thread):\r\n def __init__(self, sleep_time):\r\n threading.Thread.__init__(self)\r\n threading.Thread.__init__(self)\r\n self.sleep_time = sleep_time\r\n\r\n def run(self):\r\n my_logic(self.sleep_time)\r\n\r\n\r\nthreads = [MyThread(i) for i in range(1, 4)]\r\nfor t in threads:\r\n t.start()\r\n\r\n<\/pre>\nthreading.current_thread()<\/code>, we gain access to the current thread information. Among that information we can find its status (is_alive()<\/code>), its daemon flag (isDaemon()<\/code>), and other useful methods.<\/p>\n2.3. Daemon Threads<\/h3>\n
setDaemon()<\/code> method with True<\/code> as a parameter. By default threads are not daemon. Let’s se how our program behaves when we make one of those threads daemon:<\/p>\n\r\nthreads = [MyThread(i) for i in range(1, 4)]\r\nthreads[2].setDaemon(True)\r\nfor t in threads:\r\n t.start()\r\n<\/pre>\n
\r\nThread-2 start\r\nThread-4 start\r\nThread-6 start\r\nThread-2 end\r\nThread-4 end\r\n<\/pre>\n
\r\nimport http.client\r\nimport threading\r\nimport logging\r\n\r\nlogging.basicConfig(level=logging.INFO, format='(%(threadName)-10s) %(message)s', )\r\n\r\n\r\ndef save(html, file_absolute_path):\r\n logging.info(\"saving {} bytes to {}\".format(len(html), file_absolute_path))\r\n with open(file_absolute_path, 'wb+') as file:\r\n file.write(html)\r\n file.flush()\r\n\r\n\r\ndef crawl(req):\r\n logging.info(\"executing get request for parameters: {}\".format(str(req)))\r\n connection = http.client.HTTPConnection(req[\"host\"], req[\"port\"])\r\n connection.request(\"GET\", req[\"path\"])\r\n response = connection.getresponse()\r\n logging.info(\"got {} response http code\".format(response.status))\r\n logging.debug(\"headers: {}\".format(str(response.headers)))\r\n response_content = response.read()\r\n logging.debug(\"actual response: {}\".format(response_content))\r\n return response_content\r\n\r\n\r\nclass MyCrawler(threading.Thread):\r\n def __init__(self, req, file_path):\r\n threading.Thread.__init__(self, name=\"Crawler-{}\".format(req[\"host\"]))\r\n self.req = req\r\n self.file_path = file_path\r\n\r\n def run(self):\r\n global executed_crawlers\r\n html = crawl(self.req)\r\n save(html, self.file_path)\r\n\r\n\r\ndef __main__():\r\n continue_input = True\r\n threads = []\r\n while continue_input:\r\n host = input(\"host: \")\r\n port = 80 # int(input(\"port: \"))\r\n path = \"\/\" # input(\"path: \")\r\n file_path = input(\"output file absolute path: \")\r\n req = {\"host\": host, \"port\": port, \"path\": path}\r\n threads.append(MyCrawler(req, file_path))\r\n continue_input = input(\"add another? (y\/N) \") == \"y\"\r\n\r\n for t in threads:\r\n t.start()\r\n # t.join()\r\n\r\n__main__()\r\n\r\n\r\n<\/pre>\n\r\nhost: www.google.com.ar\r\noutput file absolute path: \/tmp\/google-ar.html\r\nadd another? (y\/N) y\r\nhost: www.google.com.br\r\noutput file absolute path: \/tmp\/google-br.html\r\nadd another? (y\/N) y\r\nhost: www.google.com.co\r\noutput file absolute path: \/tmp\/google-co.html\r\nadd another? (y\/N) y\r\nhost: www.google.cl\r\noutput file absolute path: \/tmp\/google-cl.html\r\nadd another? (y\/N)\r\n(Crawler-www.google.com.ar) executing get request for parameters: {'path': '\/', 'host': 'www.google.com.ar', 'port': 80}\r\n(Crawler-www.google.com.br) executing get request for parameters: {'path': '\/', 'host': 'www.google.com.br', 'port': 80}\r\n(Crawler-www.google.com.co) executing get request for parameters: {'path': '\/', 'host': 'www.google.com.co', 'port': 80}\r\n(Crawler-www.google.cl) executing get request for parameters: {'path': '\/', 'host': 'www.google.cl', 'port': 80}\r\n(Crawler-www.google.com.co) got 200 response http code\r\n(Crawler-www.google.com.ar) got 200 response http code\r\n(Crawler-www.google.com.br) got 200 response http code\r\n(Crawler-www.google.com.co) saving 53181 bytes to \/tmp\/google-co.html\r\n(Crawler-www.google.com.ar) saving 53008 bytes to \/tmp\/google-ar.html\r\n(Crawler-www.google.com.br) saving 53605 bytes to \/tmp\/google-br.html\r\n(Crawler-www.google.cl) got 200 response http code\r\n(Crawler-www.google.cl) saving 53069 bytes to \/tmp\/google-cl.html\r\n<\/pre>\n__main__()<\/code> function that says t.join()<\/code>, the join method, causes the current thread<\/em> to wait for the thread it joined before continuing its execution, and the log looks like:<\/p>\n\r\nhost: www.google.com.ar\r\noutput file absolute path: \/tmp\/google-ar.html\r\nadd another? (y\/N) y\r\nhost: www.google.com.br\r\noutput file absolute path: \/tmp\/google-ar.html\r\nadd another? (y\/N) y\r\nhost: www.google.com.co\r\noutput file absolute path: \/tmp\/google-co.html\r\nadd another? (y\/N) y\r\nhost: www.google.cl\r\noutput file absolute path: \/tmp\/google-cl.html\r\nadd another? (y\/N)\r\n(Crawler-www.google.com.ar) executing get request for parameters: {'port': 80, 'path': '\/', 'host': 'www.google.com.ar'}\r\n(Crawler-www.google.com.ar) got 200 response http code\r\n(Crawler-www.google.com.ar) saving 52973 bytes to \/tmp\/google-ar.html\r\n(Crawler-www.google.com.br) executing get request for parameters: {'port': 80, 'path': '\/', 'host': 'www.google.com.br'}\r\n(Crawler-www.google.com.br) got 200 response http code\r\n(Crawler-www.google.com.br) saving 54991 bytes to \/tmp\/google-ar.html\r\n(Crawler-www.google.com.co) executing get request for parameters: {'port': 80, 'path': '\/', 'host': 'www.google.com.co'}\r\n(Crawler-www.google.com.co) got 200 response http code\r\n(Crawler-www.google.com.co) saving 53172 bytes to \/tmp\/google-co.html\r\n(Crawler-www.google.cl) executing get request for parameters: {'port': 80, 'path': '\/', 'host': 'www.google.cl'}\r\n(Crawler-www.google.cl) got 200 response http code\r\n(Crawler-www.google.cl) saving 53110 bytes to \/tmp\/google-cl.html\r\n<\/pre>\njoin<\/code> method. Imagine these threads where daemon<\/em>, and you don’t have control over that. You would have to instantiate a variable which holds the amount of threads executed and then wait for it to equal the number of threads that must be executed before exiting the main thread. It’s not very elegant.<\/p>\n2.4. Joining Threads<\/h3>\n
\r\nimport http.client\r\nimport threading\r\nimport logging\r\n\r\nlogging.basicConfig(level=logging.INFO, format='(%(threadName)-10s) %(message)s', )\r\n\r\n\r\ndef save(html, file_absolute_path):\r\n logging.info(\"saving {} bytes to {}\".format(len(html), file_absolute_path))\r\n with open(file_absolute_path, 'wb+') as file:\r\n file.write(html)\r\n file.flush()\r\n\r\n\r\ndef crawl(req):\r\n logging.info(\"executing get request for parameters: {}\".format(str(req)))\r\n connection = http.client.HTTPConnection(req[\"host\"], req[\"port\"])\r\n connection.request(\"GET\", req[\"path\"])\r\n response = connection.getresponse()\r\n logging.info(\"got {} response http code\".format(response.status))\r\n logging.debug(\"headers: {}\".format(str(response.headers)))\r\n response_content = response.read()\r\n logging.debug(\"actual response: {}\".format(response_content))\r\n return response_content\r\n\r\n\r\nclass MyCrawler(threading.Thread):\r\n def __init__(self, req, file_path):\r\n threading.Thread.__init__(self, name=\"Crawler-{}\".format(req[\"host\"]), daemon=True)\r\n self.req = req\r\n self.file_path = file_path\r\n\r\n def run(self):\r\n global executed_crawlers\r\n html = crawl(self.req)\r\n save(html, self.file_path)\r\n\r\n\r\ndef __main__():\r\n continue_input = True\r\n threads = []\r\n while continue_input:\r\n host = input(\"host: \")\r\n port = 80 # int(input(\"port: \"))\r\n path = \"\/\" # input(\"path: \")\r\n file_path = input(\"output file absolute path: \")\r\n req = {\"host\": host, \"port\": port, \"path\": path}\r\n threads.append(MyCrawler(req, file_path))\r\n continue_input = input(\"add another? (y\/N) \") == \"y\"\r\n\r\n for t in threads:\r\n t.start()\r\n\r\n current_thread = threading.currentThread()\r\n for thread in threading.enumerate():\r\n if thread is not current_thread:\r\n thread.join()\r\n\r\n\r\n__main__()\r\n\r\n<\/pre>\nthreading.enumerate()<\/code> and joining every thread which is not the main one. The behavior remains the same:<\/p>\n\r\nhost: www.google.com.ar\r\noutput file absolute path: \/tmp\/google-ar.html\r\nadd another? (y\/N) y\r\nhost: www.google.com.br\r\noutput file absolute path: \/tmp\/google-br.html\r\nadd another? (y\/N) y\r\nhost: www.google.com.co\r\noutput file absolute path: \/tmp\/google-co.html\r\nadd another? (y\/N) y\r\nhost: www.google.cl\r\noutput file absolute path: \/tmp\/google-cl.html\r\nadd another? (y\/N)\r\n(Crawler-www.google.com.ar) executing get request for parameters: {'port': 80, 'host': 'www.google.com.ar', 'path': '\/'}\r\n(Crawler-www.google.com.br) executing get request for parameters: {'port': 80, 'host': 'www.google.com.br', 'path': '\/'}\r\n(Crawler-www.google.com.co) executing get request for parameters: {'port': 80, 'host': 'www.google.com.co', 'path': '\/'}\r\n(Crawler-www.google.cl) executing get request for parameters: {'port': 80, 'host': 'www.google.cl', 'path': '\/'}\r\n(Crawler-www.google.com.ar) got 200 response http code\r\n(Crawler-www.google.cl) got 200 response http code\r\n(Crawler-www.google.com.br) got 200 response http code\r\n(Crawler-www.google.com.ar) saving 52980 bytes to \/tmp\/google-ar.html\r\n(Crawler-www.google.cl) saving 53088 bytes to \/tmp\/google-cl.html\r\n(Crawler-www.google.com.br) saving 53549 bytes to \/tmp\/google-br.html\r\n(Crawler-www.google.com.co) got 200 response http code\r\n(Crawler-www.google.com.co) saving 53117 bytes to \/tmp\/google-co.html\r\n<\/pre>\n2.5. Time Threads<\/h3>\n
threading.Timer<\/code>. It is basically a subclass of Thread<\/code> which, given a delay and a function, it executes the function after the delay has passed. Also, it can be cancelled at any point.<\/p>\n\r\nimport threading\r\nimport time\r\nimport logging\r\n\r\nlogging.basicConfig(level=logging.DEBUG, format='(%(threadName)-10s) %(message)s',)\r\n\r\ndef delayed():\r\n logging.debug('worker running')\r\n return\r\n\r\nt1 = threading.Timer(3, delayed)\r\nt1.setName('t1')\r\nt2 = threading.Timer(3, delayed)\r\nt2.setName('t2')\r\n\r\nlogging.debug('starting timers')\r\nt1.start()\r\nt2.start()\r\n\r\nlogging.debug('waiting before canceling %s', t2.getName())\r\ntime.sleep(2)\r\nlogging.debug('canceling %s', t2.getName())\r\nt2.cancel()\r\nlogging.debug('done')\r\n<\/pre>\n\r\n(MainThread) starting timers\r\n(MainThread) waiting before canceling t2\r\n(MainThread) canceling t2\r\n(MainThread) done\r\n(t1 ) worker running\r\n<\/pre>\n
2.6. Events: Communication Between Threads<\/h3>\n
\r\nimport threading\r\n\r\n\r\nclass Racer(threading.Thread):\r\n\r\n def __init__(self, name, start_signal):\r\n threading.Thread.__init__(self, name=name)\r\n self.start_signal = start_signal\r\n\r\n def run(self):\r\n self.start_signal.wait()\r\n print(\"I, {}, got to the goal!\".format(self.name))\r\n\r\n\r\nclass Race:\r\n\r\n def __init__(self, racer_names):\r\n self.start_signal = threading.Event()\r\n self.racers = [Racer(name, self.start_signal) for name in racer_names]\r\n for racer in self.racers:\r\n racer.start()\r\n\r\n def start(self):\r\n self.start_signal.set()\r\n\r\n\r\ndef __main__():\r\n race = Race([\"rabbit\", \"turtle\", \"cheetah\", \"monkey\", \"cow\", \"horse\", \"tiger\", \"lion\"])\r\n race.start()\r\n\r\n\r\n__main__()\r\n<\/pre>\nThread<\/code> called Racer<\/code>, which on its run waits for the event to be set and then prints “I, {name}, got to the goal!”<\/em>.<\/p>\n\r\nI, rabbit, got to the goal!\r\nI, lion, got to the goal!\r\nI, turtle, got to the goal!\r\nI, cheetah, got to the goal!\r\nI, monkey, got to the goal!\r\nI, cow, got to the goal!\r\nI, horse, got to the goal!\r\nI, tiger, got to the goal!\r\n<\/pre>\n
\r\nI, lion, got to the goal!\r\nI, turtle, got to the goal!\r\nI, monkey, got to the goal!\r\nI, horse, got to the goal!\r\nI, cow, got to the goal!\r\nI, tiger, got to the goal!\r\nI, cheetah, got to the goal!\r\nI, rabbit, got to the goal!\r\n<\/pre>\n
2.7. Locking Resources<\/h3>\n
\r\nimport random\r\nimport threading\r\nimport logging\r\n\r\nlogging.basicConfig(level=logging.INFO,\r\n format='[%(levelname)s] (%(threadName)-s) (%(module)-s) (%(funcName)-s) %(message)s',\r\n filename='\/tmp\/locking-py.log')\r\n\r\n\r\nclass Repository:\r\n def __init__(self):\r\n self.repo = {}\r\n self.lock = threading.Lock()\r\n\r\n def create(self, entry):\r\n logging.info(\"waiting for lock\")\r\n self.lock.acquire()\r\n try:\r\n logging.info(\"acquired lock\")\r\n new_id = len(self.repo.keys())\r\n entry[\"id\"] = new_id\r\n self.repo[new_id] = entry\r\n finally:\r\n logging.info(\"releasing lock\")\r\n self.lock.release()\r\n\r\n def find(self, entry_id):\r\n logging.info(\"waiting for lock\")\r\n self.lock.acquire()\r\n try:\r\n logging.info(\"acquired lock\")\r\n return self.repo[entry_id]\r\n except KeyError:\r\n return None\r\n finally:\r\n logging.info(\"releasing lock\")\r\n self.lock.release()\r\n\r\n def all(self):\r\n logging.info(\"waiting for lock\")\r\n self.lock.acquire()\r\n try:\r\n logging.info(\"acquired lock\")\r\n return self.repo\r\n finally:\r\n logging.info(\"releasing lock\")\r\n self.lock.release()\r\n\r\n\r\nclass ProductRepository(Repository):\r\n def __init__(self):\r\n Repository.__init__(self)\r\n\r\n def add_product(self, description, price):\r\n self.create({\"description\": description, \"price\": price})\r\n\r\n\r\nclass PurchaseRepository(Repository):\r\n def __init__(self, product_repository):\r\n Repository.__init__(self)\r\n self.product_repository = product_repository\r\n\r\n def add_purchase(self, product_id, qty):\r\n product = self.product_repository.find(product_id)\r\n if product is not None:\r\n total_amount = product[\"price\"] * qty\r\n self.create({\"product_id\": product_id, \"qty\": qty, \"total_amount\": total_amount})\r\n\r\n def sales_by_product(self, product_id):\r\n sales = {\"product_id\": product_id, \"qty\": 0, \"total_amount\": 0}\r\n all_purchases = self.all()\r\n for k in all_purchases:\r\n purchase = all_purchases[k]\r\n if purchase[\"product_id\"] == sales[\"product_id\"]:\r\n sales[\"qty\"] += purchase[\"qty\"]\r\n sales[\"total_amount\"] += purchase[\"total_amount\"]\r\n return sales\r\n\r\n\r\nclass Buyer(threading.Thread):\r\n def __init__(self, name, product_repository, purchase_repository):\r\n threading.Thread.__init__(self, name=\"Buyer-\" + name)\r\n self.product_repository = product_repository\r\n self.purchase_repository = purchase_repository\r\n\r\n def run(self):\r\n for i in range(0, 1000):\r\n max_product_id = len(self.product_repository.all().keys())\r\n product_id = random.randrange(0, max_product_id + 1, 1)\r\n qty = random.randrange(0, 100, 1)\r\n self.purchase_repository.add_purchase(product_id, qty)\r\n\r\n\r\nclass ProviderAuditor(threading.Thread):\r\n def __init__(self, product_id, purchase_repository):\r\n threading.Thread.__init__(self, name=\"Auditor-product_id=\" + str(product_id))\r\n self.product_id = product_id\r\n self.purchase_repository = purchase_repository\r\n\r\n def run(self):\r\n logging.info(str(self.purchase_repository.sales_by_product(self.product_id)))\r\n\r\n\r\ndef __main__():\r\n product_repository = ProductRepository()\r\n purchase_repository = PurchaseRepository(product_repository)\r\n\r\n input_another_product = True\r\n while input_another_product:\r\n description = input(\"product description: \")\r\n price = float(input(\"product price: \"))\r\n product_repository.add_product(description, price)\r\n input_another_product = input(\"continue (y\/N): \") == \"y\"\r\n\r\n buyers = [Buyer(\"carlos\", product_repository, purchase_repository),\r\n Buyer(\"juan\", product_repository, purchase_repository),\r\n Buyer(\"mike\", product_repository, purchase_repository),\r\n Buyer(\"sarah\", product_repository, purchase_repository)]\r\n\r\n for b in buyers:\r\n b.start()\r\n b.join()\r\n\r\n for i in product_repository.all():\r\n ProviderAuditor(i, purchase_repository).start()\r\n\r\n\r\n__main__()\r\n\r\n<\/pre>\nRepository<\/code> which has locks for every access method (let’s assume every developer will know that he mustn’t access the repository’s dictionary directly).<\/p>\nfinally<\/code> block, you should be very careful with that. If you don’t put the release in a finally block, whenever an exception is raised and interrupts the function’s execution, your lock will not be released, and there will be no way to access that resource anymore.<\/p>\n\r\nproduct description: a\r\nproduct price: 1\r\ncontinue (y\/N): y\r\nproduct description: b\r\nproduct price: 2\r\ncontinue (y\/N): y\r\nproduct description: c\r\nproduct price: 3\r\ncontinue (y\/N): y\r\nproduct description: d\r\nproduct price: 4\r\ncontinue (y\/N): y\r\nproduct description: e\r\nproduct price: 5\r\ncontinue (y\/N): y\r\nproduct description: f\r\nproduct price: 6\r\ncontinue (y\/N): y\r\nproduct description: g\r\nproduct price: 7\r\ncontinue (y\/N): y\r\nproduct description: h\r\nproduct price: 8\r\ncontinue (y\/N): y\r\nproduct description: i\r\nproduct price: 9\r\ncontinue (y\/N):\r\n<\/pre>\n
Buyer<\/code> threads, since they perform a thousand actions each. That log won’t be readable, BUT, ProviderAuditor<\/code> threads will log some very interesting information. So we run grep \"Auditor\" \/tmp\/locking-py.log<\/code> and see:<\/p>\n\r\n[INFO] (Auditor-product_id=0) (locking) (all) waiting for lock\r\n[INFO] (Auditor-product_id=0) (locking) (all) acquired lock\r\n[INFO] (Auditor-product_id=0) (locking) (all) releasing lock\r\n[INFO] (Auditor-product_id=1) (locking) (all) waiting for lock\r\n[INFO] (Auditor-product_id=0) (locking) (run) {'total_amount': 19850.0, 'product_id': 0, 'qty': 19850}\r\n[INFO] (Auditor-product_id=2) (locking) (all) waiting for lock\r\n[INFO] (Auditor-product_id=1) (locking) (all) acquired lock\r\n[INFO] (Auditor-product_id=3) (locking) (all) waiting for lock\r\n[INFO] (Auditor-product_id=4) (locking) (all) waiting for lock\r\n[INFO] (Auditor-product_id=5) (locking) (all) waiting for lock\r\n[INFO] (Auditor-product_id=1) (locking) (all) releasing lock\r\n[INFO] (Auditor-product_id=6) (locking) (all) waiting for lock\r\n[INFO] (Auditor-product_id=7) (locking) (all) waiting for lock\r\n[INFO] (Auditor-product_id=1) (locking) (run) {'total_amount': 41586.0, 'product_id': 1, 'qty': 20793}\r\n[INFO] (Auditor-product_id=2) (locking) (all) acquired lock\r\n[INFO] (Auditor-product_id=2) (locking) (all) releasing lock\r\n[INFO] (Auditor-product_id=2) (locking) (run) {'total_amount': 60294.0, 'product_id': 2, 'qty': 20098}\r\n[INFO] (Auditor-product_id=3) (locking) (all) acquired lock\r\n[INFO] (Auditor-product_id=3) (locking) (all) releasing lock\r\n[INFO] (Auditor-product_id=3) (locking) (run) {'total_amount': 86752.0, 'product_id': 3, 'qty': 21688}\r\n[INFO] (Auditor-product_id=4) (locking) (all) acquired lock\r\n[INFO] (Auditor-product_id=8) (locking) (all) waiting for lock\r\n[INFO] (Auditor-product_id=4) (locking) (all) releasing lock\r\n[INFO] (Auditor-product_id=4) (locking) (run) {'total_amount': 93960.0, 'product_id': 4, 'qty': 18792}\r\n[INFO] (Auditor-product_id=5) (locking) (all) acquired lock\r\n[INFO] (Auditor-product_id=5) (locking) (all) releasing lock\r\n[INFO] (Auditor-product_id=5) (locking) (run) {'total_amount': 109776.0, 'product_id': 5, 'qty': 18296}\r\n[INFO] (Auditor-product_id=6) (locking) (all) acquired lock\r\n[INFO] (Auditor-product_id=6) (locking) (all) releasing lock\r\n[INFO] (Auditor-product_id=6) (locking) (run) {'total_amount': 140945.0, 'product_id': 6, 'qty': 20135}\r\n[INFO] (Auditor-product_id=7) (locking) (all) acquired lock\r\n[INFO] (Auditor-product_id=7) (locking) (all) releasing lock\r\n[INFO] (Auditor-product_id=7) (locking) (run) {'total_amount': 164152.0, 'product_id': 7, 'qty': 20519}\r\n[INFO] (Auditor-product_id=8) (locking) (all) acquired lock\r\n[INFO] (Auditor-product_id=8) (locking) (all) releasing lock\r\n[INFO] (Auditor-product_id=8) (locking) (run) {'total_amount': 182475.0, 'product_id': 8, 'qty': 20275}\r\n<\/pre>\nProviderAuditor<\/code> threads, the first one to acquire the lock is the Auditor-product_id=0<\/em>, then releases it and prints our sales. Then goes Auditor-product_id=1<\/em> and 2, 3, 4 and 5<\/em> are waiting. And it goes on and on. Now (again, imagining python’s dictionaries are not thread safe) our resources are thread safe.<\/p>\nLock<\/code> objects can not be acquired more than once, even by the same thread. Changing it is easy, just substitute Lock<\/code> with RLock<\/code>, which is a Re-entrant Lock<\/em> and will provide access to a locked resource, only to the thread which performed the lock.<\/p>\nRepository<\/code> class code to make it prettier:<\/p>\n\r\nclass Repository:\r\n def __init__(self):\r\n self.repo = {}\r\n self.lock = threading.Lock()\r\n\r\n def create(self, entry):\r\n logging.info(\"waiting lock\")\r\n with self.lock:\r\n logging.info(\"acquired lock\")\r\n new_id = len(self.repo.keys())\r\n entry[\"id\"] = new_id\r\n self.repo[new_id] = entry\r\n\r\n def find(self, entry_id):\r\n logging.info(\"waiting for lock\")\r\n with self.lock:\r\n try:\r\n logging.info(\"acquired lock\")\r\n return self.repo[entry_id]\r\n except KeyError:\r\n return None\r\n\r\n def all(self):\r\n logging.info(\"waiting for lock\")\r\n with self.lock:\r\n logging.info(\"acquired lock\")\r\n return self.repo\r\n<\/pre>\n\r\n[INFO] (Auditor-product_id=0) (locking) (all) waiting for lock\r\n[INFO] (Auditor-product_id=0) (locking) (all) acquired lock\r\n[INFO] (Auditor-product_id=1) (locking) (all) waiting for lock\r\n[INFO] (Auditor-product_id=0) (locking) (run) {'product_id': 0, 'total_amount': 19098.0, 'qty': 19098}\r\n[INFO] (Auditor-product_id=2) (locking) (all) waiting for lock\r\n[INFO] (Auditor-product_id=1) (locking) (all) acquired lock\r\n[INFO] (Auditor-product_id=3) (locking) (all) waiting for lock\r\n[INFO] (Auditor-product_id=4) (locking) (all) waiting for lock\r\n[INFO] (Auditor-product_id=5) (locking) (all) waiting for lock\r\n[INFO] (Auditor-product_id=1) (locking) (run) {'product_id': 1, 'total_amount': 36344.0, 'qty': 18172}\r\n[INFO] (Auditor-product_id=6) (locking) (all) waiting for lock\r\n[INFO] (Auditor-product_id=2) (locking) (all) acquired lock\r\n[INFO] (Auditor-product_id=7) (locking) (all) waiting for lock\r\n[INFO] (Auditor-product_id=8) (locking) (all) waiting for lock\r\n[INFO] (Auditor-product_id=9) (locking) (all) waiting for lock\r\n[INFO] (Auditor-product_id=2) (locking) (run) {'product_id': 2, 'total_amount': 57555.0, 'qty': 19185}\r\n[INFO] (Auditor-product_id=3) (locking) (all) acquired lock\r\n[INFO] (Auditor-product_id=3) (locking) (run) {'product_id': 3, 'total_amount': 72292.0, 'qty': 18073}\r\n[INFO] (Auditor-product_id=4) (locking) (all) acquired lock\r\n[INFO] (Auditor-product_id=4) (locking) (run) {'product_id': 4, 'total_amount': 88835.0, 'qty': 17767}\r\n[INFO] (Auditor-product_id=5) (locking) (all) acquired lock\r\n[INFO] (Auditor-product_id=5) (locking) (run) {'product_id': 5, 'total_amount': 110754.0, 'qty': 18459}\r\n[INFO] (Auditor-product_id=6) (locking) (all) acquired lock\r\n[INFO] (Auditor-product_id=6) (locking) (run) {'product_id': 6, 'total_amount': 129766.0, 'qty': 18538}\r\n[INFO] (Auditor-product_id=7) (locking) (all) acquired lock\r\n[INFO] (Auditor-product_id=7) (locking) (run) {'product_id': 7, 'total_amount': 152576.0, 'qty': 19072}\r\n[INFO] (Auditor-product_id=8) (locking) (all) acquired lock\r\n[INFO] (Auditor-product_id=8) (locking) (run) {'product_id': 8, 'total_amount': 150210.0, 'qty': 16690}\r\n[INFO] (Auditor-product_id=9) (locking) (all) acquired lock\r\n[INFO] (Auditor-product_id=9) (locking) (run) {'product_id': 9, 'total_amount': 160150.0, 'qty': 16015}\r\n<\/pre>\n2.8. Limiting Concurrent Access to Resources<\/h3>\n
Semaphore<\/code> comes to help, it tells the Lock<\/code> how many threads can acquire lock at once. Let’s modify our ProviderAuditor<\/code> code to let at most 5 providers acquire lock at the same time:<\/p>\n\r\nclass ProviderAuditor(threading.Thread):\r\n def __init__(self, product_id, purchase_repository):\r\n threading.Thread.__init__(self, name=\"Auditor-product_id=\" + str(product_id))\r\n self.product_id = product_id\r\n self.purchase_repository = purchase_repository\r\n self.semaphore = threading.Semaphore(5)\r\n\r\n def run(self):\r\n with self.semaphore:\r\n logging.info(str(self.purchase_repository.sales_by_product(self.product_id)))\r\n<\/pre>\n
2.9. Thread-Specific Data<\/h3>\n
local()<\/code> method which returns data that is only accessible from one thread, here goes an example:<\/p>\n\r\nimport logging\r\nimport random\r\n\r\nfrom threading import Thread, local\r\n\r\nlogging.basicConfig(level=logging.INFO,\r\n format='[%(levelname)s] (%(threadName)-s) (%(module)-s) (%(funcName)-s) %(message)s',)\r\n\r\n\r\ndef my_method(data):\r\n try:\r\n logging.info(str(data.value))\r\n except AttributeError:\r\n logging.info(\"data does not have a value yet\")\r\n\r\n\r\nclass MyProcess(Thread):\r\n def __init__(self):\r\n Thread.__init__(self)\r\n\r\n def run(self):\r\n data = local()\r\n my_method(data)\r\n data.value = {\"process_id\": random.randint(0, 1000)}\r\n my_method(data)\r\n\r\nfor i in range(0, 4):\r\n MyProcess().start()\r\n\r\n<\/pre>\n\r\n[INFO] (Thread-1) (process-identifier) (my_method) data does not have a value yet\r\n[INFO] (Thread-1) (process-identifier) (my_method) {'process_id': 567}\r\n[INFO] (Thread-2) (process-identifier) (my_method) data does not have a value yet\r\n[INFO] (Thread-2) (process-identifier) (my_method) {'process_id': 477}\r\n[INFO] (Thread-3) (process-identifier) (my_method) data does not have a value yet\r\n[INFO] (Thread-3) (process-identifier) (my_method) {'process_id': 812}\r\n[INFO] (Thread-4) (process-identifier) (my_method) data does not have a value yet\r\n[INFO] (Thread-4) (process-identifier) (my_method) {'process_id': 981}\r\n<\/pre>\nlocal().value<\/code>, data from other threads will never be available there.<\/p>\n3. Download the Code Project<\/h2>\n