{ "cells": [ { "cell_type": "markdown", "id": "f64b009a", "metadata": {}, "source": [ "\n", "\n", "### Programming large-scale parallel systems\n", "\n", "\n", "# Asynchronous programming in Julia\n" ] }, { "cell_type": "markdown", "id": "bf68ad38", "metadata": {}, "source": [ "## Contents\n", "\n", "In this notebook, we will learn the basics of asynchronous programming in Julia. In particular, we will learn about:\n", "\n", "- Tasks\n", "- Channels\n", "\n", "Understanding these concepts is important to learn distributed computing later." ] }, { "cell_type": "markdown", "id": "cde5ee75", "metadata": {}, "source": [ "
\n", "Note: Do not forget to execute the next cell before starting this notebook! \n", "
" ] }, { "cell_type": "code", "execution_count": null, "id": "0b0496c7", "metadata": {}, "outputs": [], "source": [ "function why_q1()\n", " msg = \"\"\"\n", " Evaluating compute_π(100_000_000) takes about 0.25 seconds on the teacher's laptop. Thus, the loop would take about 2.5 seconds since we are calling the function 10 times.\n", " \"\"\"\n", " println(msg)\n", "end\n", "function why_q2()\n", " msg = \"\"\"\n", " The time in doing the loop will be O(1) since the loop just schedules 10 tasks, which should be a (small) constant time independent of n.\n", " \"\"\"\n", " println(msg)\n", "end\n", "function why_q3()\n", " msg = \"\"\"\n", " It will take 2.5 seconds, like in question 1. The @sync macro forces to wait for all tasks we have generated with the @async macro. Since we have created 10 tasks and each of them takes about 0.25 seconds, the total time will be about 2.5 seconds.\n", " \"\"\"\n", " println(msg)\n", "end\n", "function why_q4()\n", " msg = \"\"\"\n", " It will take about 3 seconds. The channel has buffer size 4, thus the call to put!will not block. The call to take! will not block neither since there is a value stored in the channel. The taken value is 3 and therefore we will wait for 3 seconds.\n", " \"\"\"\n", " println(msg)\n", "end\n", "function why_q5()\n", " msg = \"\"\"\n", " The channel is not buffered and therefore the call to put! will block. The cell will run forever, since there is no other task that calls take! on this channel.\n", " \"\"\"\n", " println(msg)\n", "end\n", "println(\"🥳 Well done! \")" ] }, { "cell_type": "markdown", "id": "caf64254", "metadata": {}, "source": [ "## Tasks\n", "\n", "### Creating a task\n", "\n", "Technically, a task in Julia is a *symmetric* [*co-routine*](https://en.wikipedia.org/wiki/Coroutine). More informally, a task is a piece of computational work that can be started (scheduled) at some point in the future, and that can be interrupted and resumed. To create a task, we first need to create a function that represents the work to be done in the task. In next cell, we generate a task that generates and sums two matrices." ] }, { "cell_type": "code", "execution_count": null, "id": "fe668cb1", "metadata": {}, "outputs": [], "source": [ "function work()\n", " println(\"Starting work\")\n", " sleep(7)\n", " a = rand(3,3)\n", " b = rand(3,3)\n", " r = a + b\n", " println(\"Finishing work\")\n", " r\n", "end" ] }, { "cell_type": "code", "execution_count": null, "id": "67ee0328", "metadata": {}, "outputs": [], "source": [ "t = Task(work)" ] }, { "cell_type": "markdown", "id": "e459c5c2", "metadata": {}, "source": [ "### Scheduling a task\n", "\n", "The task has been created, but the corresponding work has not started. Note that we do not see any output from function `work` yet. To run the task we need to schedule it." ] }, { "cell_type": "code", "execution_count": null, "id": "8778c199", "metadata": {}, "outputs": [], "source": [ "schedule(t)" ] }, { "cell_type": "markdown", "id": "f1fb9283", "metadata": {}, "source": [ "### Fetching the task result\n", "\n", "The task has been executed, but we do not see the result. To get the result we need to fetch it." ] }, { "cell_type": "code", "execution_count": null, "id": "0c7b626e", "metadata": {}, "outputs": [], "source": [ "fetch(t)" ] }, { "cell_type": "markdown", "id": "fedbbd71", "metadata": {}, "source": [ "### Tasks run asynchronously\n", "\n", "It is important to note that tasks run asynchronously. To illustrate this let's create and schedule a new task." ] }, { "cell_type": "code", "execution_count": null, "id": "4ccc996c", "metadata": {}, "outputs": [], "source": [ "t = Task(work)" ] }, { "cell_type": "code", "execution_count": null, "id": "015bea27", "metadata": {}, "outputs": [], "source": [ "schedule(t)" ] }, { "cell_type": "markdown", "id": "5ec0718e", "metadata": {}, "source": [ "Note that while the task is running we can execute Julia code. To check this, execute the next two cells while the task is running." ] }, { "cell_type": "code", "execution_count": null, "id": "a70fcbe8", "metadata": {}, "outputs": [], "source": [ "sin(4π)*exp(-0.1)" ] }, { "cell_type": "code", "execution_count": null, "id": "6def444b", "metadata": {}, "outputs": [], "source": [ "1 + 1" ] }, { "cell_type": "markdown", "id": "d483d4d0", "metadata": {}, "source": [ "How is this possible? Tasks run in the background and this particular task is sleeping for most of the time. Thus, it is possible to use the current Julia process for other operations while the task is sleeping." ] }, { "cell_type": "markdown", "id": "910323fd", "metadata": {}, "source": [ "### Tasks do not run in parallel\n", "\n", "It is also important to note that tasks do not run in parallel. We were able to run code while previous tasks was running because the task was idling most of the time in the sleep function. If the task does actual work, the current process will be busy running this task and preventing to run other tasks. Let's illustrate this with an example. The following code computes an approximation of $\\pi$ using [Leibniz formula](https://en.wikipedia.org/wiki/Leibniz_formula_for_pi). The quality of the approximation increases with the value of `n`." ] }, { "cell_type": "code", "execution_count": null, "id": "b53ac640", "metadata": {}, "outputs": [], "source": [ "function compute_π(n)\n", " s = 1.0\n", " for i in 1:n\n", " s += (isodd(i) ? -1 : 1) / (i*2+1)\n", " end\n", " 4*s\n", "end" ] }, { "cell_type": "markdown", "id": "7614df94", "metadata": {}, "source": [ " Call this function with a large number. Note that it will take some time." ] }, { "cell_type": "code", "execution_count": null, "id": "63f2aec3", "metadata": {}, "outputs": [], "source": [ "compute_π(4_000_000_000)" ] }, { "cell_type": "markdown", "id": "840ff590", "metadata": {}, "source": [ "Create a task that performs this computation." ] }, { "cell_type": "code", "execution_count": null, "id": "f16ade6c", "metadata": {}, "outputs": [], "source": [ "fun = () -> @show compute_π(4_000_000_000)\n", "t = Task(fun)" ] }, { "cell_type": "markdown", "id": "60c2567c", "metadata": {}, "source": [ "Schedule the tasks and then try to execute the 2nd cell bellow. Note that the current process will be busy running the task." ] }, { "cell_type": "code", "execution_count": null, "id": "1f28a388", "metadata": {}, "outputs": [], "source": [ "schedule(t)" ] }, { "cell_type": "code", "execution_count": null, "id": "52e060e7", "metadata": {}, "outputs": [], "source": [ "1+1" ] }, { "cell_type": "markdown", "id": "a700fe69", "metadata": {}, "source": [ "### `yield`\n", "\n", "If tasks do not run in parallel, what is the purpose of tasks? Tasks are handy since they can be interrupted and to switch control to other tasks. This is achieved via function `yield`. When we call yield, we provide the opportunity to switch to another task. The function below is a variation of function `compute_π` in which we yield every 1000 iterations. At the call to yield we allow other tasks to take over. Without this call to yield, once we start function `compute_π` we cannot start any other tasks until this function finishes." ] }, { "cell_type": "code", "execution_count": null, "id": "9260c065", "metadata": {}, "outputs": [], "source": [ "function compute_π_yield(n)\n", " s = 1.0\n", " for i in 1:n\n", " s += (isodd(i) ? -1 : 1) / (i*2+1)\n", " if mod(i,1000) == 0\n", " yield()\n", " end\n", " end\n", " 4*s\n", "end" ] }, { "cell_type": "markdown", "id": "69fd4131", "metadata": {}, "source": [ "You can check this behavior experimentally with the two following cells. The next one creates and schedules a task that computes pi with the function `compute_π_yield`. Note that you can run the 2nd cell bellow while this task is running since we call to yield often inside `compute_π_yield`." ] }, { "cell_type": "code", "execution_count": null, "id": "a85f3f39", "metadata": {}, "outputs": [], "source": [ "fun = () -> @show compute_π_yield(3_000_000_000)\n", "t = Task(fun)\n", "schedule(t)" ] }, { "cell_type": "code", "execution_count": null, "id": "24e23e88", "metadata": {}, "outputs": [], "source": [ "1+1" ] }, { "cell_type": "markdown", "id": "66ca10d6", "metadata": {}, "source": [ "### Example: Implementing function sleep\n", "\n", "Using yield, we can implement our own version of the sleep function as follows:" ] }, { "cell_type": "code", "execution_count": null, "id": "beed2b29", "metadata": {}, "outputs": [], "source": [ "function mysleep(secs)\n", " final_time = time() + secs\n", " while time() < final_time\n", " yield()\n", " end\n", " nothing\n", "end" ] }, { "cell_type": "markdown", "id": "5c44f6c1", "metadata": {}, "source": [ "You can check that it behaves as expected." ] }, { "cell_type": "code", "execution_count": null, "id": "73c13bfb", "metadata": {}, "outputs": [], "source": [ "@time mysleep(3)" ] }, { "cell_type": "markdown", "id": "25048665", "metadata": {}, "source": [ "### Tasks take a function with no arguments\n", "\n", "This function needs to have zero arguments, but it can capture variables if needed. If we try to create a task with a function that has arguments, it will result in an error when we schedule it." ] }, { "cell_type": "code", "execution_count": null, "id": "87397749", "metadata": {}, "outputs": [], "source": [ "add(a,b) = a + b" ] }, { "cell_type": "code", "execution_count": null, "id": "e99766c7", "metadata": {}, "outputs": [], "source": [ "t = Task(add)" ] }, { "cell_type": "code", "execution_count": null, "id": "1a785bae", "metadata": {}, "outputs": [], "source": [ "schedule(t)" ] }, { "cell_type": "markdown", "id": "ebfce725", "metadata": {}, "source": [ "If we need, we can capture variables in the function to be run by the task as shown in the next cells." ] }, { "cell_type": "code", "execution_count": null, "id": "06f54fa0", "metadata": {}, "outputs": [], "source": [ "a = rand(3,3)\n", "b = rand(3,3);" ] }, { "cell_type": "code", "execution_count": null, "id": "4c9e586e", "metadata": {}, "outputs": [], "source": [ "fun = () -> add(a, b)\n", "t = Task(fun)\n", "schedule(t)" ] }, { "cell_type": "markdown", "id": "cd829a64", "metadata": {}, "source": [ "### Useful macro: `@async`\n", "\n", "So far, we have created tasks using low-level functions, but there are more convenient ways of creating and scheduling tasks. For instance using the `@async` macro. This macro is used to run a piece of code asynchronously. Under the hood it puts the code in an anonymous function, creates a task, and schedules it. For instance, the next cell is equivalent to the previous one." ] }, { "cell_type": "code", "execution_count": null, "id": "9b5152d1", "metadata": {}, "outputs": [], "source": [ "@async a + b" ] }, { "cell_type": "markdown", "id": "ac3262d8", "metadata": {}, "source": [ "### Another useful macro: `@sync`\n", "\n", "This macro is used to wait for all the tasks created with `@async` in a given block of code. " ] }, { "cell_type": "code", "execution_count": null, "id": "ef2e49f9", "metadata": {}, "outputs": [], "source": [ "@sync begin\n", " @async sleep(3)\n", " @async sleep(4)\n", "end" ] }, { "cell_type": "markdown", "id": "98f0d685", "metadata": {}, "source": [ "## Channels\n", "\n", "### Sending data between tasks\n", "\n", "Julia provides channels as a way to send data between tasks. A channel is like a FIFO queue which tasks can put values into and take values from. In the next example, we create a channel and a task that puts five values into the channel. Finally, the task closes the channel." ] }, { "cell_type": "code", "execution_count": null, "id": "b88d5308", "metadata": {}, "outputs": [], "source": [ "chnl = Channel{Int}()" ] }, { "cell_type": "code", "execution_count": null, "id": "3421c7a6", "metadata": { "scrolled": true }, "outputs": [], "source": [ "@async begin\n", " for i in 1:5\n", " put!(chnl,i)\n", " end\n", " close(chnl)\n", "end" ] }, { "cell_type": "markdown", "id": "1df0508c", "metadata": {}, "source": [ "By executing next cell several times, we will get the values from the channel. We are indeed communicating values from two different tasks. If we execute the cell more than 5 times, it will raise an error since the channel is closed. " ] }, { "cell_type": "code", "execution_count": null, "id": "64b9436e", "metadata": {}, "outputs": [], "source": [ "take!(chnl)" ] }, { "cell_type": "markdown", "id": "757a1b07", "metadata": {}, "source": [ "### Channels are iterable\n", "\n", "Instead of taking values from a channel until an error occurs, we can also iterate over the channel in a for loop until the channel is closed." ] }, { "cell_type": "code", "execution_count": null, "id": "2fc22dfa", "metadata": {}, "outputs": [], "source": [ "chnl = Channel{Int}()" ] }, { "cell_type": "code", "execution_count": null, "id": "fa62a4df", "metadata": {}, "outputs": [], "source": [ "@async begin\n", " for i in 1:5\n", " put!(chnl,i)\n", " end\n", " close(chnl)\n", "end" ] }, { "cell_type": "code", "execution_count": null, "id": "e511e19b", "metadata": {}, "outputs": [], "source": [ "for i in chnl\n", " @show i\n", "end" ] }, { "cell_type": "markdown", "id": "b1a2a557", "metadata": {}, "source": [ "### Calls to `put!` and `take!` are blocking\n", "\n", "Note that `put!` and `take!` are blocking operations. Calling `put!` blocks the tasks until another task calls `take!` and viceversa. Thus, we need at least 2 tasks for this to work. If we call `put!` and `take!` from the same task, it will result in a dead lock. We have added a print statement to the previous example. Run it again and note how `put!` blocks until we call `take!`. " ] }, { "cell_type": "code", "execution_count": null, "id": "f34373ca", "metadata": {}, "outputs": [], "source": [ "chnl = Channel{Int}()" ] }, { "cell_type": "code", "execution_count": null, "id": "bfde2ecd", "metadata": {}, "outputs": [], "source": [ "@async begin\n", " for i in 1:5\n", " put!(chnl,i)\n", " println(\"I have put $i\")\n", " end\n", " close(chnl)\n", "end" ] }, { "cell_type": "code", "execution_count": null, "id": "5830659d", "metadata": {}, "outputs": [], "source": [ "take!(chnl)" ] }, { "cell_type": "markdown", "id": "006140bd", "metadata": {}, "source": [ "### Buffered channels\n", "\n", "We can be a bit more flexible and use a buffered channel. In this case, `put!` will block only if the channel is full and `take!` will block if the channel is empty. We repeat the previous example, but with a buffered channel of size 2. Note that we can call `put!` until the channel is full. At this point, we need to wait to until we call `take!` which removes an item from the channel, making room for a new item." ] }, { "cell_type": "code", "execution_count": null, "id": "dfe06b5f", "metadata": {}, "outputs": [], "source": [ "buffer_size = 2\n", "chnl = Channel{Int}(buffer_size)" ] }, { "cell_type": "code", "execution_count": null, "id": "6289bc2e", "metadata": {}, "outputs": [], "source": [ "@async begin\n", " for i in 1:5\n", " put!(chnl,i)\n", " println(\"I have put $i\")\n", " end\n", " close(chnl)\n", "end" ] }, { "cell_type": "code", "execution_count": null, "id": "2a87cd5f", "metadata": {}, "outputs": [], "source": [ "take!(chnl)" ] }, { "cell_type": "markdown", "id": "8a1ef849", "metadata": {}, "source": [ "In summary:\n", "\n", "- `put!` will wait for a `take!` if there is not space left in the channel's buffer.\n", "- `take!` will wait for a `put!` if there is no data to be consumed in the channel.\n", "- Both `put!` and `take!` will raise an error if the channel is closed." ] }, { "cell_type": "markdown", "id": "9ddd66ca", "metadata": {}, "source": [ "## Questions\n", "\n", "In the next questions, `t` is the value of the variable `t` defined in the next cell." ] }, { "cell_type": "code", "execution_count": null, "id": "dd45ae08", "metadata": {}, "outputs": [], "source": [ "n = 140_000_000\n", "t = @elapsed @show compute_π(n)" ] }, { "cell_type": "markdown", "id": "2a23b1c3", "metadata": {}, "source": [ "
\n", "Question (NB2-Q1): How long will the compute time of next cell be? \n", "
\n", "\n", " a) 10*t\n", " b) t\n", " c) 0.1*t\n", " d) O(1), i.e. time independent from n \n" ] }, { "cell_type": "code", "execution_count": null, "id": "18d6cfe3", "metadata": {}, "outputs": [], "source": [ "n = 140_000_000\n", "@time for i in 1:10\n", " @show compute_π(n)\n", "end" ] }, { "cell_type": "code", "execution_count": null, "id": "d6b8382e", "metadata": {}, "outputs": [], "source": [ "why_q1()" ] }, { "cell_type": "markdown", "id": "5f19d38c", "metadata": {}, "source": [ "
\n", "Question (NB2-Q2): How long will the compute time of next cell be? \n", "
\n", "\n", " a) 10*t\n", " b) t\n", " c) 0.1*t\n", " d) O(1) \n", "\n" ] }, { "cell_type": "code", "execution_count": null, "id": "dac0c92a", "metadata": {}, "outputs": [], "source": [ "n = 140_000_000\n", "@time for i in 1:10\n", " @async @show compute_π(n)\n", "end" ] }, { "cell_type": "code", "execution_count": null, "id": "edff9747", "metadata": {}, "outputs": [], "source": [ "why_q2()" ] }, { "cell_type": "markdown", "id": "5041c355", "metadata": {}, "source": [ "
\n", "Question (NB2-Q3): How long will the compute time of next cell be? \n", "
\n", "\n", " a) 10*t\n", " b) t\n", " c) 0.1*t\n", " d) O(1) \n" ] }, { "cell_type": "code", "execution_count": null, "id": "c06dc4a5", "metadata": {}, "outputs": [], "source": [ "n = 140_000_000\n", "@time @sync for i in 1:10\n", " @async @show compute_π(n)\n", "end" ] }, { "cell_type": "code", "execution_count": null, "id": "87bc7c5c", "metadata": {}, "outputs": [], "source": [ "why_q3()" ] }, { "cell_type": "markdown", "id": "841b690e", "metadata": {}, "source": [ "
\n", "Question (NB2-Q4): How long will the compute time of the 2nd cell be? \n", "
\n", "\n", " a) infinity\n", " b) 1 second\n", " c) less than 1 seconds\n", " d) 3 seconds" ] }, { "cell_type": "code", "execution_count": null, "id": "6ac116bb", "metadata": {}, "outputs": [], "source": [ "buffer_size = 4\n", "chnl = Channel{Int}(buffer_size)" ] }, { "cell_type": "code", "execution_count": null, "id": "25363a90", "metadata": {}, "outputs": [], "source": [ "@time begin\n", " put!(chnl,3)\n", " i = take!(chnl)\n", " sleep(i)\n", "end" ] }, { "cell_type": "code", "execution_count": null, "id": "a18a0a7d", "metadata": {}, "outputs": [], "source": [ "why_q4()" ] }, { "cell_type": "markdown", "id": "df663f11", "metadata": {}, "source": [ "
\n", "Question (NB2-Q5): How long will the compute time of the 2nd cell be? \n", "
\n", "\n", " a) infinity\n", " b) 1 second\n", " c) less than 1 seconds\n", " d) 3 seconds" ] }, { "cell_type": "code", "execution_count": null, "id": "9abeed40", "metadata": {}, "outputs": [], "source": [ "chnl = Channel{Int}()" ] }, { "cell_type": "code", "execution_count": null, "id": "8e8428ce", "metadata": {}, "outputs": [], "source": [ "@time begin\n", " put!(chnl,3)\n", " i = take!(chnl)\n", " sleep(i)\n", "end" ] }, { "cell_type": "code", "execution_count": null, "id": "d8923fae", "metadata": {}, "outputs": [], "source": [ "why_q5()" ] }, { "cell_type": "markdown", "id": "0ee77abe", "metadata": {}, "source": [ "
\n", "Note: If for some reason a cell keeps running forever, we can stop it with Kernel > Interrupt or Kernel > Restart (see tabs above).\n", "
" ] }, { "cell_type": "markdown", "id": "dfab0c90", "metadata": {}, "source": [ "## Summary\n", "\n", "In order to start \"thinking in parallel\" you first need to be familiar with concepts of asynchronous programming, in particular tasks. In this notebook, we have seen the basics of working with tasks. Some key points to remember:\n", "\n", "- How to create, schedule, and fetch from a task.\n", "- Tasks run asynchronously, but not in parallel. You can have a single core CPU and still be able to work with several tasks.\n", "- Channels are used to communicate data between tasks.\n", "- Adding data (`put!`) or taking data (`take!`) from a channel might wait depending on the channel state. Be careful to avoid dead locks.\n", "\n" ] }, { "cell_type": "markdown", "id": "a5d3730b", "metadata": {}, "source": [ "# License\n", "\n", "This notebook is part of the course [Programming Large Scale Parallel Systems](https://www.francescverdugo.com/XM_40017/) at Vrije Universiteit Amsterdam and may be used under a [CC BY 4.0](https://creativecommons.org/licenses/by/4.0/) license." ] } ], "metadata": { "kernelspec": { "display_name": "Julia 1.10.0", "language": "julia", "name": "julia-1.10" }, "language_info": { "file_extension": ".jl", "mimetype": "application/julia", "name": "julia", "version": "1.10.0" } }, "nbformat": 4, "nbformat_minor": 5 }