{"id":8779,"date":"2023-12-11T11:30:56","date_gmt":"2023-12-11T11:30:56","guid":{"rendered":"https:\/\/www.modernescpp.com\/?p=8779"},"modified":"2023-12-11T11:42:16","modified_gmt":"2023-12-11T11:42:16","slug":"a-bug-in-the-priority-scheduler-for-coroutines","status":"publish","type":"post","link":"https:\/\/www.modernescpp.com\/index.php\/a-bug-in-the-priority-scheduler-for-coroutines\/","title":{"rendered":"A Bug in the Priority Scheduler for Coroutines"},"content":{"rendered":"\n

In my last two posts, I presented priority schedulers for coroutines. The schedulers had a bug.<\/p>\n\n\n\n

\"\"<\/figure>\n\n\n\n

First of all, here is the erroneous scheduler.<\/p>\n\n\n\n

\/\/ priority_queueSchedulerPriority.cpp<\/span>\n\n#include <concepts><\/span>\n#include <coroutine><\/span>\n#include <functional><\/span>\n#include <iostream><\/span>\n#include <queue><\/span>\n#include <utility><\/span>\n\n\nstruct<\/span> Task {\n\n  struct<\/span> promise_type {\n    std::<\/span>suspend_always initial_suspend() noexcept { return<\/span> {}; }\n    std::<\/span>suspend_always final_suspend() noexcept { return<\/span> {}; }\n\n    Task get_return_object() { \n        return<\/span> std::<\/span>coroutine_handle<<\/span>promise_type>::<\/span>from_promise(*<\/span>this<\/span>); \n    }\n    void<\/span> return_void() {}\n    void<\/span> unhandled_exception() {}\n  };\n\n  Task(std::<\/span>coroutine_handle<<\/span>promise_type><\/span> handle):<\/span> handle{handle}{}\n\n  auto<\/span> get_handle() { return<\/span> handle; }\n\n  std::<\/span>coroutine_handle<<\/span>promise_type><\/span> handle;\n};\n\nusing<\/span> job =<\/span> std::<\/span>pair<<\/span>int<\/span>, std::<\/span>coroutine_handle<>><\/span>;\n\ntemplate<\/span> <<\/span>typename<\/span> Updater =<\/span> std::<\/span>identity,                         \/\/ (1)<\/span>\n          typename<\/span> Comperator =<\/span> std::<\/span>ranges::<\/span>less><\/span>                   \nrequires std::<\/span>invocable<<\/span>decltype(Updater()), int<\/span>><\/span> &&<\/span>                \/\/ (2)<\/span>\n         std::<\/span>predicate<<\/span>decltype(Comperator()), job, job><\/span>             \nclass<\/span> Scheduler<\/span> {\n\n  std::<\/span>priority_queue<<\/span>job, std::<\/span>vector<<\/span>job><\/span>, Comperator><\/span> _prioTasks;\n\n  public:<\/span> \n    void<\/span> emplace<\/span>(int<\/span> prio, std::<\/span>coroutine_handle<><\/span> task) {\n      _prioTasks.push(std::<\/span>make_pair(prio, task));\n    }\n\n    void<\/span> schedule<\/span>() {\n      Updater upd =<\/span> {};                                            \/\/ (3)<\/span>\n      while<\/span>(!<\/span>_prioTasks.empty()) {\n        auto<\/span> [prio, task] =<\/span> _prioTasks.top();\n        _prioTasks.pop();\n        task.resume();\n\n        if<\/span>(!<\/span>task.done()) { \n          _prioTasks.push(std::<\/span>make_pair(upd(prio), task));          \/\/ (4)<\/span>\n        }\n        else<\/span> {\n          task.destroy();\n        }\n      }\n    }\n\n};\n\n\nTask createTask<\/span>(const<\/span> std::<\/span>string&<\/span> name) {\n  std::<\/span>cout <<<\/span> name <<<\/span> " start<\/span>\\n<\/span>"<\/span>;\n  co_await std::<\/span>suspend_always();\n  for<\/span> (int<\/span> i =<\/span> 0<\/span>; i <=<\/span> 3<\/span>; ++<\/span>i ) { \n    std::<\/span>cout <<<\/span> name <<<\/span> " execute "<\/span> <<<\/span> i <<<\/span> "<\/span>\\n<\/span>"<\/span>;                  \/\/ (5)<\/span>\n    co_await std::<\/span>suspend_always();\n  }\n  co_await std::<\/span>suspend_always();\n  std::<\/span>cout <<<\/span> name <<<\/span> " finish<\/span>\\n<\/span>"<\/span>;\n}\n\n\nint<\/span> main<\/span>() {\n\n  std::<\/span>cout <<<\/span> '\\n'<\/span>;\n\n  Scheduler scheduler1;                                               \/\/ (6)<\/span>\n\n  scheduler1.emplace(0<\/span>, createTask("TaskA"<\/span>).get_handle());\n  scheduler1.emplace(1<\/span>, createTask("  TaskB"<\/span>).get_handle());\n  scheduler1.emplace(2<\/span>, createTask("    TaskC"<\/span>).get_handle());\n\n  scheduler1.schedule();\n\n  std::<\/span>cout <<<\/span> '\\n'<\/span>;\n\n  Scheduler<<\/span>decltype([](int<\/span> a) { return<\/span> a -<\/span> 1<\/span>; })><\/span> scheduler2;        \/\/ (7)<\/span>\n\n  scheduler2.emplace(0<\/span>, createTask("TaskA"<\/span>).get_handle());\n  scheduler2.emplace(1<\/span>, createTask("  TaskB"<\/span>).get_handle());\n  scheduler2.emplace(2<\/span>, createTask("    TaskC"<\/span>).get_handle());\n\n  scheduler2.schedule();\n\n  std::<\/span>cout <<<\/span> '\\n'<\/span>;\n\n}\n<\/pre><\/div>\n\n\n\n\n
This is the output of the program I got.<\/p>\n\n\n\n
\"\"<\/figure>\n\n\n\n
Christof Meerwald got another output with GCC. Thanks a lot for letting me know. Here is the GCC output with optimization enabled. <\/p>\n\n\n\n
\"\"<\/figure>\n\n\n\n
Similarly, the Windows output was also broken:<\/p>\n\n\n\n
\"\"<\/figure>\n\n\n\n
Can you spot the error? Here are the crucial lines:<\/p>\n\n\n\n
Task createTask<\/span>(const<\/span> std::<\/span>string&<\/span> name) {                      \/\/ (1)<\/span>\n  std::<\/span>cout <<<\/span> name <<<\/span> " start<\/span>\\n<\/span>"<\/span>;\n  co_await std::<\/span>suspend_always();\n  for<\/span> (int<\/span> i =<\/span> 0<\/span>; i <=<\/span> 3<\/span>; ++<\/span>i ) { \n    std::<\/span>cout <<<\/span> name <<<\/span> " execute "<\/span> <<<\/span> i <<<\/span> "<\/span>\\n<\/span>"<\/span>;                  \n    co_await std::<\/span>suspend_always();\n  }\n  co_await std::<\/span>suspend_always();\n  std::<\/span>cout <<<\/span> name <<<\/span> " finish<\/span>\\n<\/span>"<\/span>;\n}\n\n\nint<\/span> main<\/span>() {\n\n  std::<\/span>cout <<<\/span> '\\n'<\/span>;\n\n  Scheduler scheduler1;                                               \n\n  scheduler1.emplace(0<\/span>, createTask("TaskA"<\/span>).get_handle());      \/\/ (2)<\/span>\n  scheduler1.emplace(1<\/span>, createTask("  TaskB"<\/span>).get_handle());    \/\/ (3)<\/span>\n  scheduler1.emplace(2<\/span>, createTask("    TaskC"<\/span>).get_handle());  \/\/ (4)<\/span>\n\n  scheduler1.schedule();\n\n  std::<\/span>cout <<<\/span> '\\n'<\/span>;\n\n  Scheduler<<\/span>decltype([](int<\/span> a) { return<\/span> a -<\/span> 1<\/span>; })><\/span> scheduler2;        \n\n  scheduler2.emplace(0<\/span>, createTask("TaskA"<\/span>).get_handle());     \/\/ (5)<\/span>\n  scheduler2.emplace(1<\/span>, createTask("  TaskB"<\/span>).get_handle());   \/\/ (6)<\/span>\n  scheduler2.emplace(2<\/span>, createTask("    TaskC"<\/span>).get_handle()); \/\/ (7)<\/span>\n\n  scheduler2.schedule();\n\n  std::<\/span>cout <<<\/span> '\\n'<\/span>;\n\n}\n<\/pre><\/div>\n\n\n\n\n\n\n
The coroutine createTask<\/code> takes its string by const lvalue reference (1), but its arguments \"TaskA\" - \"TaskC\"<\/code>  are rvalues (lines 2 – 7).  Using the reference to the temporaries is undefined behavior. The additional schedulers priority_SchedulerSimplified<\/code> and priority_queueSchedulerComparator<\/code> in the posts “A Priority Scheduler for Coroutines<\/a>” and  “An Advanced Priority Scheduler for Coroutines<\/a>” suffer the same issue.<\/p>\n\n\n\n
Fixing the issue is straightforward. Either the coroutine createTask <\/code>takes its argument by value (Task createTask(std::string name<\/code>)) or its arguments become lvalues. Here is the second approach in lines (1) – (3):<\/p>\n\n\n\n
\/\/ priority_queueSchedulerPriority.cpp<\/span>\n\n#include <concepts><\/span>\n#include <coroutine><\/span>\n#include <functional><\/span>\n#include <iostream><\/span>\n#include <queue><\/span>\n#include <utility><\/span>\n\n\nstruct<\/span> Task {\n\n  struct<\/span> promise_type {\n    std::<\/span>suspend_always initial_suspend() noexcept { return<\/span> {}; }\n    std::<\/span>suspend_always final_suspend() noexcept { return<\/span> {}; }\n\n    Task get_return_object() { \n        return<\/span> std::<\/span>coroutine_handle<<\/span>promise_type>::<\/span>from_promise(*<\/span>this<\/span>); \n    }\n    void<\/span> return_void() {}\n    void<\/span> unhandled_exception() {}\n  };\n\n  Task(std::<\/span>coroutine_handle<<\/span>promise_type><\/span> handle):<\/span> handle{handle}{}\n\n  auto<\/span> get_handle() { return<\/span> handle; }\n\n  std::<\/span>coroutine_handle<<\/span>promise_type><\/span> handle;\n};\n\nusing<\/span> job =<\/span> std::<\/span>pair<<\/span>int<\/span>, std::<\/span>coroutine_handle<>><\/span>;\n\ntemplate<\/span> <<\/span>typename<\/span> Updater =<\/span> std::<\/span>identity,                         \n          typename<\/span> Comperator =<\/span> std::<\/span>ranges::<\/span>less><\/span>                   \nrequires std::<\/span>invocable<<\/span>decltype(Updater()), int<\/span>><\/span> &&<\/span>                \n         std::<\/span>predicate<<\/span>decltype(Comperator()), job, job><\/span>             \nclass<\/span> Scheduler<\/span> {\n\n  std::<\/span>priority_queue<<\/span>job, std::<\/span>vector<<\/span>job><\/span>, Comperator><\/span> _prioTasks;\n\n  public:<\/span> \n    void<\/span> emplace<\/span>(int<\/span> prio, std::<\/span>coroutine_handle<><\/span> task) {\n      _prioTasks.push(std::<\/span>make_pair(prio, task));\n    }\n\n    void<\/span> schedule<\/span>() {\n      Updater upd =<\/span> {};                                            \n      while<\/span>(!<\/span>_prioTasks.empty()) {\n        auto<\/span> [prio, task] =<\/span> _prioTasks.top();\n        _prioTasks.pop();\n        task.resume();\n\n        if<\/span>(!<\/span>task.done()) { \n          _prioTasks.push(std::<\/span>make_pair(upd(prio), task));          \n        }\n        else<\/span> {\n          task.destroy();\n        }\n      }\n    }\n\n};\n\n\nTask createTask<\/span>(const<\/span> std::<\/span>string&<\/span> name) {\n  std::<\/span>cout <<<\/span> name <<<\/span> " start<\/span>\\n<\/span>"<\/span>;\n  co_await std::<\/span>suspend_always();\n  for<\/span> (int<\/span> i =<\/span> 0<\/span>; i <=<\/span> 3<\/span>; ++<\/span>i ) { \n    std::<\/span>cout <<<\/span> name <<<\/span> " execute "<\/span> <<<\/span> i <<<\/span> "<\/span>\\n<\/span>"<\/span>;                  \n    co_await std::<\/span>suspend_always();\n  }\n  co_await std::<\/span>suspend_always();\n  std::<\/span>cout <<<\/span> name <<<\/span> " finish<\/span>\\n<\/span>"<\/span>;\n}\n\n\nint<\/span> main<\/span>() {\n\n  std::<\/span>cout <<<\/span> '\\n'<\/span>;\n\n  std::<\/span>string taskA =<\/span> "TaskA"<\/span>;                    \/\/ (1)<\/span>\n  std::<\/span>string taskB =<\/span> "  TaskB"<\/span>;                  \/\/ (2)<\/span>\n  std::<\/span>string taskC =<\/span> "    TaskC"<\/span>;                \/\/ (3)<\/span>\n\n  Scheduler scheduler1;                                          \n\n  scheduler1.emplace(0<\/span>, createTask(taskA).get_handle());\n  scheduler1.emplace(1<\/span>, createTask(taskB).get_handle());\n  scheduler1.emplace(2<\/span>, createTask(taskC).get_handle());\n\n  scheduler1.schedule();\n\n  std::<\/span>cout <<<\/span> '\\n'<\/span>;\n\n  Scheduler<<\/span>decltype([](int<\/span> a) { return<\/span> a -<\/span> 1<\/span>; })><\/span> scheduler2;        \n\n  scheduler2.emplace(0<\/span>, createTask(taskA).get_handle());\n  scheduler2.emplace(1<\/span>, createTask(taskB).get_handle());\n  scheduler2.emplace(2<\/span>, createTask(taskC).get_handle());\n\n  scheduler2.schedule();\n\n  std::<\/span>cout <<<\/span> '\\n'<\/span>;\n\n}\n<\/pre><\/div>\n\n\n\n\n\n
What’s Next? <\/h2>\n\n\n\n
Coroutines provide an intuitive way of writing asynchronous code. My next post will be a guest post from Ljubic Damir, presenting a single producer \u2013 single consumer workflow based on coroutines.<\/p>\n","protected":false},"excerpt":{"rendered":"
In my last two posts, I presented priority schedulers for coroutines. The schedulers had a bug. First of all, here is the erroneous scheduler. \/\/ priority_queueSchedulerPriority.cpp #include <concepts> #include <coroutine> #include <functional> #include <iostream> #include <queue> #include <utility> struct Task { struct promise_type { std::suspend_always initial_suspend() noexcept { return {}; } std::suspend_always final_suspend() noexcept { […]<\/p>\n","protected":false},"author":21,"featured_media":8780,"comment_status":"closed","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[375],"tags":[445],"class_list":["post-8779","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-c-20","tag-coroutines"],"_links":{"self":[{"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/posts\/8779","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/users\/21"}],"replies":[{"embeddable":true,"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/comments?post=8779"}],"version-history":[{"count":12,"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/posts\/8779\/revisions"}],"predecessor-version":[{"id":8794,"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/posts\/8779\/revisions\/8794"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/media\/8780"}],"wp:attachment":[{"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/media?parent=8779"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/categories?post=8779"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/tags?post=8779"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}