Concurrency in Python - concepts, frameworks and best practices

Speaker 1 [00:04]

Good morning, everyone. Thanks for coming to my talk. My name is Stefan Schwarzer. Something about me. I'm a Python, I'm using Python since 1999. I'm a software developer since 2000, freelancer, became a freelancer in 2005. I wrote a book in ancient times, as you see, Python 2.2. I think this is the 15th talk or so. and I'm the maintainer of the FTP client library, which sits on top of FTP lib, but is more high level. Okay, in this talk, first I talk a bit about some basics on concurrency, some terms, for example. I discuss some concurrency approaches that are available with the standard library, or actually they are conceptually, they're different concepts that are independent not the standard library. Yeah, then two problems that might come up in concurrent code are race conditions and deadlocks. And after that, I discuss queues, and then even higher level concurrency approaches. And finally, I will talk a bit about some best practices or what I think are best practices. This talk, I mean, concurrency is a very complex subject, And I can, yeah, this is a bit like scratching the surface, but yeah, still, I think it's worth more. Okay. Reasons for concurrency are using, or speeding up CPU-intensive tasks, either by running the same calculations, or kind of the same calculation on different cores in parallel, or executing a single algorithms algorithm on multiple cores, which requires that the algorithm can be parallelized. Then, another big reason for using concurrency is input-output, so that if one part of a program runs, for example, is waiting for a request data on a socket, another part of the program can continue and maybe work on another request in parallel. And finally, reactivity, this is mostly relevant for GUI applications, so if you start a lengthy operation, you don't want to sit in front of the computer and twiddling your thumbs, but if you want to continue working on the computer, not switching to another application or something, it's nice if this task, if this operation runs in the background. so concurrent with the actions in the user interface. Some terms, resource, this is a very generic term. Anything that's used by an execution thread, and execution threads don't mean necessarily operating system thread, but they could be operating system threads. And resources can be simple variables, data structures, files, networks, sockets, but also, for example, a screen. If you output data to a screen, to a terminal or something, or a GUI window, and you write from different threads, yeah, you don't want to mix the output up. So you need to serialize there as well. Concurrency, yeah, the subject of the talk, just means that there are multiple execution threads, but they don't have to run in parallel. So it could also be you run a bit of one thread, switch to another thread, run a bit of this, and so on. Parallelism is a special case of concurrency, which means that the tasks actually run in parallel, so at the same time, for example, on different CPU cores. And anatomic operation is a task that can't be interrupted by another execution thread, or you could see it as must not be interrupted by another execution thread, depending on how you look at it. Okay, the concurrency approaches in Python. Multi-threading, running operating system threads in a single process. That's the module threading in the standard library, which allows you to create threads and start threads. Also provide some APIs to deal with concurrency to deal with concurrency or with shared access, but we will come to this. Threads can share data and process memory, and for CPython, it's very relevant that there's a so-called global interpreter lock, the GIL. Maybe you heard of this. And this prevents the parallel execution of Python code. The GIL is released during IO operations, so that's the good part. So multi-threading is usable for IO limited concurrency, but it's not so, it's not really usable for CPU limited concurrency. But on the other hand, C extensions can release the GIL, for example, NumPy does this. For some operations which are independent of Python code, don't need to access attributes or whatever, don't interact directly with the interpreter, you can temporarily release the GIL and other threats can continue. The multiprocessing module deals with concurrency for operating system processes. And if you want to transfer data or use data from the main process, this is usually via messages. So, for example, pickling data and sending this to another process. or you could initialize the data and once you start the process, it takes on all the data from the parent process so you don't need to copy anything. Okay, and there's also the possibility of using shared memory. So part of the memory that is seen by all processes or by the selected processes. Okay, I already talked about serialization. So if you send data to another process and you must serialize this, this means convert this to a stream of bytes. And the receiver must decode these bytes. I usually use the pickle module in Python for this. And the big advantage of multiprocessing compared to multithreading in Python is that there's no limitation of parallel execution. also not for CPU-limited work because the GIL is still there, but it's per process. So you can execute Python code in different processes at the same time. Event loops are quite a different approach. Usually they run in a single thread, and this loop, usually also called the main loop, detects events like mouse clicks or incoming network data. That variance, one is using handlers. This means that, for example, if the event loop detects a mouse click, a handler is called with some information about where, for example, in which position the mouse was clicked or the pointer was, and maybe other data. And after the handler has processed this event, control returns to the main loop. The other variant is that code looks sequential, but execution is switched to other code. If you went to pass to wait for I.O., and this is much used, or the typical use case in the async I.O. package that was added to Python 3.4. Race conditions, definition is, or kind of a definition, actually I had some, when I looked this up on Wikipedia, or how they define it, it's really convoluted and not even, in my opinion, a useful definition. So I made up this definition. While a resource is modified by an execution thread, another execution thread tries to modify or read the resource. So processes or execution threads get in each other's way, so to speak. Okay, and this is an example. So we have a simple function which increases a global counter by 100, counting up one at a time. And here we create, we use the threading module to create 100 threads. Here the target argument is the function. So this is what should be started when the thread starts. but if you create the thread, it doesn't start yet, so you have to call the start method. And since we need the thread a bit later, just here, I append this to the threads list here. And yeah, then once you call start, the thread starts to run. And after they are done, I check that they are done by calling join. and this means wait until the thread is finished. So now we have 100 threads and each of them counts up to 100 or from zero to 99. So you would expect that the counter is increased by 10,000 overall. So when I print the counter and this is what you're actually seeing when you execute this program. So you get a different result every time. So it could come out at 10,000, but usually it doesn't. It's a bit less. So it's even not deterministic. You get a different result all the time. And here's an explanation for this, why this happens. And this is just a simplification. Here I discuss only two threads. And the start, when the thread starts, sorry. So the counter initially is zero, and thread one reads the counter zero. Then shortly after, thread two reads the counter, sees also zero, adds one internally, doesn't store it yet. And shortly after, thread one increases the counter, stores one, and thread two also stores one. It doesn't know anything about the change from thread one. so you end up with even though both threads kind of think they increase the counter by one and we should get two, actually it's only incremented by one there's a simple way to avoid this problem by using a lock and by using it this way, this makes the increment an atomic operation and using a thread has two meanings or could look at it in different ways one is I want to when I say with lock I say okay I want this to be atomic but on the other hand it means if anyone else holds the lock I want to wait or I need to wait so this means that if you have a data structure and you want to avoid that you see inconsistent data also the reader has also to use the log, not only when you write to the data structure. Okay, other than that, no changes. Dead logs happen if an execution thread mutually claim resources that the other execution thread needs. This is kind of abstract, so here's an example. Imagine both thread one and two need resources A and B to finish a task, and thread one uses a lock to get hold of resource A and once resource B, and thread two already holds resource B and once resource A, and so none of them can continue. So this is called a deadlock. This is a somewhat maybe contrived example, but I think it brings the point across. Imagine you have two threads, which work both with an input file and output file, and you don't want to mix up the output and then the output file for example. So you use two locks and the first thread gets the input lock first and the other thread directly after gets the output lock and at that point none of them can continue because they need the other lock and this is already taken. Could also be a different order or maybe sometimes it just works. If you get this lock and this lock then this can finish and after that thread two gets this lock and this lock and continues two. So this could even work, but you can't rely on it. Queues for this example, there's a queue which is fed with data and we have different workout threads which get data from the queue. And an important point is that put and get are atomic operations, so you don't need any explicit locks or something. if you want to put something in the queue or take anything out of the queue. And this is the code example for this. We use 10 worker threads. We want to execute 100 jobs. We need this later to signal that the work is done. And we define a queue, which is in the queue module. The job could be more complex. Here I just saw the job number. And worker thread looks like this here in this example. You just have a kind of infinite loop. You get an item from the queue and then you check if it's the stop token, the special object. We will see later why this is necessary. But if it's a real job, so to speak, I call process job. This just sleeps a bit and blocks the job number. I use this random call so that in the output, you don't see it counting up, but the order's a bit random. Okay, and this is how you call it, or how you use these threads. As before, you create different worker threads, start them, store them, then schedule the jobs, up to job count, put the jobs in the queue, and meanwhile, since you started the worker threads, the workers run, take items from the queue, And after that, you put the stop token in the queue. So every worker sees, takes the stop token out of the queue and sees, okay, I'm done now, so we can shut down. And this is how the loop, why the loop is exited here. And after that, join the workers. High-level concurrency approaches are concurrent futures, active objects, and process networks, which I really like. This is very similar to the previous example, as you see. This is the same code, only you don't have the stop token. Process job is a function here. And this is, yeah, a higher level API. You create a thread pool executor. There's also, by the way, a process pool executor also in concurrent future, so if you need to parallelize execution over different processes. You can use a process pool executor. And we can directly use the work account as an argument for the executor. We get an executor object back. And these submit calls say I want to execute the process job function here on job I. And what the submit returns is a so-called future object. This is kind of a handle to access the result once it's available. You can wait for the result, but you could also just keep a reference to the future and request the data later. And this is actually, so once I created the futures, the function starts to run, all the threads start to run, and what I'm doing here is I iterate over the futures and every time one of the futures is done or has actually executed the job, I continue with the loop. This is very similar to the thread join that we saw previously. And if we compare this with the queue example, this is, so we see some benefits. Defining the thread is easier, or actually we don't even have to define an explicit thread. So process job is now a function and you don't need to inherit from the thread class and you don't need to implement run. You don't even need the queue and you don't need this stop token thing. So because we use this as completed and the loop will also be done completely if every process job thread has finished. so we don't need to explicitly stop the workers. Yeah, so, yeah, use concurrent futures if you can. Active objects, yeah, in my opinion, a nice design. Principle is that you can use things like blocks and queues and other synchronization mechanisms, but you don't show them on the outside of an object. They are not part of the API. And synchronization, so, is hidden in high-level methods. This is an example. You have this Ada class, which defines several private attributes. Yeah, it creates an in queue, input queue, output queue, creates a worker thread, and immediately starts it. Yeah, so nothing of this is visible on the outside. We have, in this Ada class, We have an underscore work function or a method. And we get an item from the queue. We again check for the stop token and if it's a stop token, break out of the loop. Otherwise, this is the work that the workers do. So it's just using a work item which is a number and increase it by 1,000 and put this in the queue. And this is the public API. Yeah, again, nothing is visible of the queues and threads. You can say submit a work item, and internally this is put in the input queue. You can ask for a result, which is read from the output queue. And you can stop the adder instance by putting the stop token in the input queue and call join on the worker thread. This is how I use it. Again, here I use 100 jobs, create an adder instance, submit this work item to the adder. Then I could do other things because the adder, the thread in the adder runs, yeah, meanwhile. And after that, yeah, I collect the results. In this case, I just print them. The thread puts the, thread that is started here, puts, oops, puts, gets the input data from the input queue and process and puts it in the output queue, so once next result is called and there's already an item available, it's returned. Otherwise, next result will block. Okay, and after that I call add a stop, So, again, you see nothing of the threads, or the thread in there. Also, nothing of the stop token or anything. Process networks receive input data and or send output data. This could be, normally these are operating system processes. Data transfer between process, again, by message passing. The process can use different programming languages. If you don't use pickle but some format like json or message pack or google protocol buffers or something that can be seen by all these processes. And you again like with multiprocessing you have some overhead due to data serialization and protocols. Protocol could mean MQP or HTTP. I will talk a bit about this later. There are two kinds of process networks, or two common kinds. One with using a broker, where you have a special process, which is this broker. And this is the process that is the center of the data exchange. So, processes don't talk to each other here, but they only talk or send data or receive data from the broker. And very common broker protocol examples are MQP and MQTT, you have a nice, yeah, usually a nice configuration for the broker. Also, you have message persistence, so for example, if you shut down the broker, but a process which needs some data doesn't have fetched the data yet. It can be stored when shutting down the broker, and when you restart the broker, the process can fetch the data. Also, an advantage of the broker approach is that each process only needs to know the broker. So if we have these processes distributed on different machines they only need the ip address and port for for the broker machine but they don't need any other ip addresses or ports for other processes where to find them also yeah you could risk with the firewall configuration restrict access to the broker machine and avoid that the processes here can talk to each other Now, this is, yeah, without a broker. This has, yeah, for simple process networks, this can be easier, yeah. But, yeah, okay, you don't, of course, you don't have the advantages of the broker. And an example for this is ZeroMQ. I don't know if you heard of this. This is a framework for, to simplify, yeah, designs like that. Then, some best practices, or what I think are best practices. These are not necessarily written down in books or online. Some of them are, certainly, and from my experience with concurrent code, I have created a list or several lists here, and one thing to keep in mind is that different advice, so these These are just rules of thumb. Different advice can apply to different areas of your code. For example, depending on the problem you're trying to solve, you could have a process network where one process internally uses multi-threading or doesn't use concurrency internally at all, or another process uses an event loop. This is really up to you. and maybe in some cases you want to use locks or whatever. Some general advice, concurrency is an optimization, or I certainly see this as an optimization, and like other optimizations, use it only if necessary, not just because it would be cool to have this running faster, but only when you really think you need it. So it can make code more complex. If you're not careful, you can have race conditions and you avoid these problems all together by not using concurrency, yeah? But of course, this talk is so that you can use concurrency. Okay. Try to keep code simple and easy to understand. So if you find yourself looking at your code and yeah, you use a lock here, you use a lock there, and you think about, yeah, if this thread gets the lock and this other one and what could go wrong, maybe you're bucking up the wrong tree, so maybe you could take a step back and just use cues or something, or come up with a design which is simpler and easier to reason about. So you could use our active objects. And if you use low-level APIs, hide them if possible. Don't make logs or queues also part of the public interface. Logs, using logs outside of a class is a really bad idea. This is just too low-level and you have to, there's more that can go wrong. So if one client that calls into your API forgets to acquire the lock, you have a problem. But if you work with the lock inside your class, like we saw in this active object example, you only need to get the lock, in this case not even a lock, but work with the queues in one place. Regarding the currency approaches, This is kind of a repetition from what we had above, but just sorted differently. So above, this was sorted by concurrency approach, and then I said what it's good for, and this is kind of a, maybe not directly checklist, but something similar, ordered by task. So if you have IO-limited concurrency, you can use multi-threading, because GIL is released during I-O. You can use the async I-O library, which was added in Python 3.4 and revised or got additional or special syntax in Python 3.5. This is used for many concurrent tasks, so if you want to handle many, many requests at the same time. I've seen numbers like talking about several thousand requests per second or so, or 10,000 or whatever, because the problem with multi-threading is that each thread, or also with processes, each thread or process you create has its own stack, which takes memory, and there's also some overhead when switching between threads or processes. And so, SNKIO programs usually run in a single thread, or the main work is done in a single thread, So for these special cases, AsyncIO is better. And then, of course, process networks, yeah, this is kind of universal, but yeah, of course, you have more process, so you don't want to use process networks if your problem is relatively simple. So CPU-limited concurrency, yeah, multi-threading works only with limitations, yeah, as I said, only if extensions can release the GIL. Otherwise, you would probably use multi-processing or process networks. GUI frameworks, so this use case of reactivity in GUIs, GUI frameworks usually come with their own event loop. So this is documented in the API and how you define handlers and how you process events like clicking a button or so. And if you have maybe developed by different teams, if you have processes in different languages, yeah, I have to use process networks. Okay, I mean, maybe sometimes you have layers to interface one language with another. Okay, I mean, maybe Python and C would be a simple example, but if you have something like talking a Python program to a Java program, yeah, okay, I know there are also solutions for this, but they are somewhat cumbersome. Now, shared state is really tricky. As I described, you could have race conditions or deadlocks if you are not careful. So avoid reading shared state while it may be written or even modify shared state while it's modified by another thread or process. Even query methods could be problematic if they implicitly update internal cache of an object. So if something looks really harmless, you just read an attribute, but internally it's defined as a property, so it looks like an attribute access, but internally, for some optimization, it updates an internal cache in the object. So you can't see this from the outside. I mean, you could look at the source code, but the source code could change or whatever, or maybe you haven't looked at the source code. So, yeah, make sure the APIs used for multiple threads, which could be execution threads, are thread safe. And you can only count on the documentation because even if you look at the source code and everything seems to be fine, the source code could be different in the next version. So, okay, if you rely on the source code, you could fix the version of the library use to this version. but probably it's better to either use some logging or mechanisms to make this thread safe or contact the maintainers of the project and ask them to clarify in the documentation whether something is thread safe or not. A good thing is that even if you don't know if something is thread safe, you can always use logs or find some way to serialize the accesses. Thread safe doesn't mean you can't use this from different threads. It just means that you can't use an API from different threads at the same time. So from the point of view of the used object, if everything just seems to be one after another, it works fine. I should have had graphics on this, I guess. I don't know how difficult this is to understand. Yeah, and also, yeah, try to, yeah, if you can, try to avoid shared state in the first place. Pass immutable objects like, for example, I've seen users where instead of passing some object where you don't know whether the attribute access is really a property or something, you just pass primitive data, you pass tuple maybe of a string at an integer. It's not nice to have this on the outside of an API. So this is, again, very nice for active objects because you could, if we look at this submit method we had earlier, we could pass a complex object into submit, and submit could get data from this object and just put in the queue not the object itself, but maybe only some string attribute value and an integer or something. So you can hide this again. Another approach is setting up the state before you start threads, because in that case, it's no longer, you have only read access. Yeah, okay, which would imply, yeah, okay, again, related to the point above, if you know it's thread safe, yeah, so, yeah. Or, I mean, or you really know maybe the data you are accessing is a list of integers or something, then in this case, you don't know. you know that it's thread safe to, I mean thread safe to access data in the list, not modify it from different threads at the same time without a lock. Okay, one, finally one very important point, concurrency involving shared state is difficult to write, but it's also difficult to test. It's difficult to write tests that actually, that would fail if you, for example, forgot to use a lock i mean there are approaches but they are really cumbersome yeah it's really tricky sometimes you need sleep calls or something to kind of um yeah provoke that there's a race condition and want to check that it doesn't happen so this means even i mean you in general you shouldn't just run your program and and think okay everything seems to work so so it has no bugs, and it doesn't work that way, and especially doesn't work that way when you use concurrency, because you can run a program a hundred times, everything seems fine, and then you run it on a computer under high load, and then it fails, for example, runs into a deadlock. I have seen an example where, this was not even Python, but it was a make file, where upgrading to a faster build machine caused race conditions. So different parts of the make file but then creating a directory, this was executed at the same time and this led to a race condition. Okay, actually in Python code then. Okay, so invest some time to create a solid design and also really take the time and think about and as I said, if thinking about the problem it's hard, try to take a step back and see if you can solve the problem in a more abstract way. And also, along the same lines, have your code reviewed, look for others, for peers who have more experience with concurrency, or maybe even the same level of concurrency, but because they might see a problem where you don't. Okay, thank you for your attention. Is there any question? Yes. Ambassador? Yeah, hello. Thank you for your great talk. Thank you. what I was wondering about is, is there any chance to use GPU parallelism from Python? There are libraries. Sometimes these libraries are implicit. For example, yesterday there was a talk on PyTorch, which can run code on the GPU just by specifying a device where the code should run or where an array or a tensor, as they call it, is associated with. Also some, I think Chaos, a machine learning library also can run code on a GPU. I mean, there's nothing in the standard library as far as I know, so this is restricted to some libraries. I don't know if there are, okay, I can really imagine, I don't know, but I can really imagine that there are third party libraries on PyPI, on PyP, on a part vector index to make this easier to run code on the GPU. But running code on the GPU is, I mean I haven't looked much into it, but from what I know, or from the little I know, is you have to run a kernel or something. Yeah, you really need to apply this or adapt this to the architecture of the GPU if you want to run code in parallel. because every thread in the GPU can only execute mostly the same code, so you can't say run this code and run something completely different on another GPU call. Hi, here in the front. Coming from the C-sharp world, there are some issues with tasks, also concurrency programming, for example, so that task can die off without showing any exception to the main loop, is that also an issue in Python? It's a problem if you don't request the result. So, I mean, in Python, a problem that you might run into is that a thread raises an exception, but you never know if it is very similar to what you described. And you need to make sure that the exception arrives in the main thread or in the calling thread. It's relatively easy with futures because if you call the result method on a future, it will raise the exception that occurred on the thread. But the caveat, of course, is if you never request the result, you will never know. But, yeah, you don't see the exception. Futures have a method which is, I think it's called result, and if the result is available, you get the result back. If the thread raises an exception, it raises the exception. If the thread is not done yet, I think it raises an exception. But you can also specify a timeout if you want to. Could also use this in a polling way. Okay, thank you. thanks for the talk um so do you know of like any library um that might support actors like in erlang or aka for python good question i yeah i think there are libraries i once looked for this i mean i mean it's some time ago um the problem or yeah in a previous version of this talk i had process networks in a more generic sense like actors, concurrent sequential processes and so on. But when I reworked the talk, I looked at these libraries on PyPion. Many of them haven't been maintained for years. So probably there is something, but I don't know how reliable it is. You could do something similar with active objects. maybe you already thought about this when you heard this so you have an input queue which is the input queue for the actor so to speak Is there any other question? Yes You mentioned race conditions as a cause of non-deterministic results. Off the top of your head is there a similar list for of causes for non-deterministic performance for non what performance non-deterministic performance of code yes i can imagine something if you um yeah if you use logs yeah to restrict access to a resource a lock means I want to have explicit access for something. And if you have multiple threads or processes which have to wait for this lock, if you have high load and high lock congestion, so to speak, then in the end, in this critical section, this atomic section, you can't run this in parallel, because you said I want to use a lock for this. Vasilis, about to answer your question? None that I can think of right now. Hi, thank you. Thank you for the talk. I was wondering, last time I used concurrent futures, it was very hard to chain futures. for example, get something like a future, process the result, and then get a new future. Are there any third party libraries that support this, or how would you use this? I don't know of any libraries in particular, but what I would probably use is, yeah, okay, I mean the kind of primitive way, or what you might first think of would be use nesting loops, yeah, but this is cumbersome, but you could use, could extract the inner loop the inner loop to another function, for example. So I think you could deal with it this way. But I haven't tried it. Okay, actually I wrote a program which at the moment uses worker threads before concurrent futures was in the standard library. And I also thought about this. So it would be a bit more tricky if you have chained thread pools or something. is there any other question we have time just for the last question I think no so thank you very much