{"id":12515,"date":"2017-02-02T14:30:09","date_gmt":"2017-02-02T21:30:09","guid":{"rendered":"https:\/\/blogs.msdn.microsoft.com\/vcblog\/?p=12515"},"modified":"2019-02-18T17:48:44","modified_gmt":"2019-02-18T17:48:44","slug":"using-ibuv-with-c-resumable-functions","status":"publish","type":"post","link":"https:\/\/devblogs.microsoft.com\/cppblog\/using-ibuv-with-c-resumable-functions\/","title":{"rendered":"Using C++ Resumable Functions with Libuv"},"content":{"rendered":"

Previously on this blog we have talked about Resumable Functions<\/a>, and even recently we touched on the renaming of the yield keyword to co_yield<\/a> in our implementation in Visual Studio 2017<\/a>. I am very excited about this potential C++ standards feature, so in this blog post I wanted to share with you a real world use of it by adapting it to the libuv library. You can use the code with Microsoft\u2019s compiler or even with other compilers that have an implementation of resumable functions. Before we jump into code, let\u2019s recap the problem space and why you should care.<\/p>\n

Problem Space<\/h2>\n

Waiting for disks or data over a network is inherently slow and we have all learned (or been told) by now that writing software that blocks is bad, right? For client side programs, doing I\/O or blocking on the UI thread is a great way to create a poor user experience as the app glitches or appears to hang.\u00a0 For server side programs, new requests can usually just create a new thread if all others are blocked, but that can cause inefficient resource usage as threads are often not a cheap resource.<\/span>\u00a0<\/span><\/p>\n

However, it is still remarkably difficult to write code that is efficient and truly asynchronous. Different platforms provide different mechanisms and APIs for doing asynchronous I\/O. Many APIs don\u2019t have any asynchronous equivalent at all. Often, the solution is to make the call from a worker thread, which calls a blocking API, and then return the result back to the main thread. This can be difficult as well and requires using synchronization mechanisms to avoid concurrency problems.<\/span>\u00a0<\/span>There are libraries that provide abstractions over these disparate mechanisms, however. Examples of this include Boost ASIO, the C++ Rest SDK, and libuv. Boost ASIO and the Rest SDK are C++ libraries and libuv is a C library. They have some overlap between them but each has its own strengths as well.<\/span><\/p>\n

Libuv<\/span><\/a> is a C library that provides the asynchronous I\/O in Node.js. While it was explicitly designed for use by Node.js, it can be used on its own and provides a common cross-platform API, abstracting away the various platform-specific asynchronous APIs. Also, the API exposes a UTF8-only API even on Windows, which is convenient. Every API that can block takes a pointer to a callback function which will be called when the requested operation has completed. An event loop runs and waits for various requests to complete and calls the specified callbacks. For me, writing libuv code was straightforward but it isn’t easy to follow the logic of a program.\u00a0 Using C++ lambdas for the callback functions can help somewhat, but passing data along the chain of callbacks requires a lot of boilerplate code. For more information on libuv, there is plenty of information on their <\/span>website<\/span><\/a>.<\/span>\u00a0<\/span><\/p>\n

There has been a lot of interest in coroutines lately. Many languages have added support for them, and there have been several coroutine proposals submitted to the C++ committee. None have been approved as of yet, but there will likely be coroutine support at some point. One of the coroutine proposals for C++ standardization is resumable functions and the current version of that proposal is <\/span>N4402<\/span><\/a>, although there are some newer changes as well. It proposes new language syntax for stackless coroutines, and does not define an actual implementation but instead specifies how the language syntax binds to a library implementation. This allows a lot of flexibility and allows supporting different runtime mechanisms.\u00a0<\/span>\u00a0<\/span><\/p>\n

Adapting libuv to resumable functions<\/h2>\n

When I started looking at this, I had never used libuv before, so I initially just wrote some code using straight libuv calls and started thinking about how I would like to be able to write the code. With resumable functions, you can write code that looks very sequential but executes asynchronously.\u00a0 Whenever the co_await keyword is encountered in a resumable function, the function will “return” if the result of the await expression is not available.<\/span>\u00a0<\/span><\/p>\n

I had several goals in creating this library.<\/span>\u00a0<\/span><\/p>\n

    \n
  1. Performance should be very good.<\/span>\u00a0<\/span><\/li>\n
  2. Avoid creating a thick C++ wrapper library.<\/span>\u00a0<\/span><\/li>\n
  3. Provide a model that should feel familiar to existing libuv users.<\/span>\u00a0<\/span><\/li>\n
  4. Allow mixing of straight libuv calls with resumable functions.<\/span><\/li>\n<\/ol>\n

    All of the code I show here and the actual library code as well as a couple of samples is available on <\/span>github<\/span><\/a> and can be compiled using Visual Studio 2015, Visual Studio 2017, or in this branch of <\/span>Clang<\/a>\u00a0and\u00a0<\/span>LLVM<\/span><\/a> that implements this proposal. You will also need CMake and libuv installed. I used version 1.8 of libuv on Linux and 1.10.1 on Windows.\u00a0<\/span>If you want to\u00a0use Clang\/LLVM, follow these standard <\/span>instructions<\/span><\/a>\u00a0to build it.<\/span><\/p>\n

    I experimented with several different ways to bind libuv to resumable functions, and I show two of these in my library. \u00a0The first (and the one I use in the following examples) uses something similar to std::promise and std::future. There is awaituv::promise_t and awaituv::future_t, which point to a shared state object that holds the \u201creturn value\u201d from the libuv call. I put \u201creturn value\u201d in quotes because the value is provided asynchronously through a callback in libuv. This mechanism requires a heap allocation to hold the shared state. The second mechanism lets the developer put the shared state on the stack of the calling function, which avoids a separate heap allocation and associated shared_ptr machinery. It isn’t as transparent as the first mechanism, but it\u00a0can be useful for performance.<\/span>\u00a0<\/span><\/p>\n

    Examples<\/h2>\n

    Let\u2019s look at a simple example that writes out “hello world” 1000 times asynchronously.<\/span><\/p>\n

    [code lang=”cpp”]\nfuture_t<void> start_hello_world()\n{\n for (int i = 0; i < 1000; ++i)\n {\n string_buf_t buf("\\nhello world\\n");\n fs_t req;\n (void) co_await fs_write(uv_default_loop(), &req, 1 \/*stdout*\/, &buf, 1, -1);\n }\n}\n<\/pre>\n

    \u00a0<\/span>A function that uses co_await must have a return type that is an awaitable type, so this function returns a future_t<void>, which implements the methods necessary for the compiler to generate code for a resumable function. This function will loop one thousand times and asynchronously write out \u201chello world\u201d. The \u201cfs_write\u201d function is in the awaituv namespace and is a thin wrapper over libuv’s uv_fs_write.\u00a0 Its return type is future_t<int>, which is awaitable. In this case, I am ignoring the actual value but still awaiting the completion. The start_hello_world function \u201creturns\u201d if the result of the await expression is not immediately available, and a pointer to resume the function is stored such that when the write completes the function is resumed. The string_buf_t type is a thin wrapper over the uv_buf_t type, although\u00a0the raw uv_buf_t type could be used as well. The fs_t type is also a thin wrapper over uv_fs_t and has a destructor that calls uv_fs_cleanup.\u00a0 This is also not required to be used but does make the code a little cleaner.<\/span>\u00a0<\/span><\/p>\n

    Note: unlike std::future, future_t does not provide a \u201cget\u201d method as that would need to actually block. In the case of libuv, this would essentially hang the program as no callbacks can run unless the event loop is processing. For this to work, you can only await on a future.<\/span>\u00a0<\/span><\/p>\n

    Now let\u2019s look at a slightly more complicated example which reads a file and dumps it to stdout.<\/span>\u00a0<\/span><\/p>\n

    [code lang=”cpp”]\nfuture_t<void> start_dump_file(const std::string& str)\n{\n \/\/ We can use the same request object for all file operations as they don’t overlap.\n static_buf_t<1024> buffer;<\/p>\n

    fs_t openreq;\n uv_file file = co_await fs_open(uv_default_loop(), &openreq, str.c_str(), O_RDONLY, 0);\n if (file > 0)\n {\n while (1)\n {\n fs_t readreq;\n int result = co_await fs_read(uv_default_loop(), &readreq, file, &buffer, 1, -1);\n if (result <= 0)\n break;\n buffer.len = result;\n fs_t req;\n (void) co_await fs_write(uv_default_loop(), &req, 1 \/*stdout*\/, &buffer, 1, -1);\n }\n fs_t closereq;\n (void) co_await fs_close(uv_default_loop(), &closereq, file);\n }\n}\n<\/pre>\n

    This function should be pretty easy to understand as it is written very much like a synchronous version would be written. The static_buf_t type is another simple C++ wrapper over uv_buf_t that provides a fixed size buffer. This function opens a file, reads a chunk into a buffer, writes it to stdout, iterates until no more data, and then closes the file.\u00a0 In this case, you can see we are using the result of the await expression when opening the file and when reading data.<\/span>\u00a0<\/span>\nNext, let\u2019s look at a function that will change the text color of stdout on a timer.<\/span>\u00a0<\/span><\/p>\n

    [code lang=”cpp”]\nbool run_timer = true;\nuv_timer_t color_timer;\nfuture_t<void> start_color_changer()\n{\n static string_buf_t normal = "\033[40;37m";\n static string_buf_t red = "\033[41;37m";<\/p>\n

    uv_timer_init(uv_default_loop(), &color_timer);<\/p>\n

    uv_write_t writereq;\n uv_tty_t tty;\n uv_tty_init(uv_default_loop(), &tty, 1, 0);\n uv_tty_set_mode(&tty, UV_TTY_MODE_NORMAL);<\/p>\n

    int cnt = 0;\n unref(&color_timer);<\/p>\n

    auto timer = timer_start(&color_timer, 1, 1);<\/p>\n

    while (run_timer)\n {\n (void) co_await timer.next_future();<\/p>\n

    if (++cnt % 2 == 0)\n (void) co_await write(&writereq, reinterpret_cast<uv_stream_t*>(&tty), &normal, 1);\n else\n (void) co_await write(&writereq, reinterpret_cast<uv_stream_t*>(&tty), &red, 1);\n }<\/p>\n

    \/\/reset back to normal\n (void) co_await write(&writereq, reinterpret_cast<uv_stream_t*>(&tty), &normal, 1);<\/p>\n

    uv_tty_reset_mode();\n co_await close(&tty);\n co_await close(&color_timer); \/\/ close handle\n}\n<\/pre>\n

    Much of this function is straightforward libuv code, which includes support for processing ANSI escape sequences to set colors. The new concept in this function is that a timer can be recurring and doesn\u2019t have a single completion. The timer_start function (wraps uv_timer_start) returns a promise_t rather than a future_t. To get an awaitable object, you must call \u201cnext_future\u201d on the timer. This resets the internal state such that it can be awaited on again. The color_timer variable is a global so that the stop_color_changer function (not shown) can stop the timer.<\/span>\u00a0<\/span><\/p>\n

    Finally, here is a function that opens a socket and sends an http request to google.com.<\/span>\u00a0<\/span><\/p>\n

    [code lang=”cpp”]\nfuture_t<void> start_http_google()\n{\n uv_tcp_t socket;\n if (uv_tcp_init(uv_default_loop(), &socket) == 0)\n {\n \/\/ Use HTTP\/1.0 rather than 1.1 so that socket is closed by server when done sending data.\n \/\/ Makes it easier than figuring it out on our end…\n const char* httpget =\n "GET \/ HTTP\/1.0\\r\\n"\n "Host: www.google.com\\r\\n"\n "Cache-Control: max-age=0\\r\\n"\n "Accept: text\/html,application\/xhtml+xml,application\/xml;q=0.9,image\/webp,*\/*;q=0.8\\r\\n"\n "\\r\\n";\n const char* host = "www.google.com";<\/p>\n

    uv_getaddrinfo_t req;\n addrinfo_state addrstate;\n if (co_await getaddrinfo(addrstate, uv_default_loop(), &req, host, "http", nullptr) == 0)\n {\n uv_connect_t connectreq;\n awaitable_state<int> connectstate;\n if (co_await tcp_connect(connectstate, &connectreq, &socket, addrstate._addrinfo->ai_addr) == 0)\n {\n string_buf_t buffer{ httpget };\n ::uv_write_t writereq;\n awaitable_state<int> writestate;\n if (co_await write(writestate, &writereq, connectreq.handle, &buffer, 1) == 0)\n {\n read_request_t reader;\n if (read_start(connectreq.handle, &reader) == 0)\n {\n while (1)\n {\n auto state = co_await reader.read_next();\n if (state->_nread <= 0)\n break;\n uv_buf_t buf = uv_buf_init(state->_buf.base, state->_nread);\n fs_t writereq;\n awaitable_state<int> writestate;\n (void) co_await fs_write(writestate, uv_default_loop(), &writereq, 1 \/*stdout*\/, &buf, 1, -1);\n }\n }\n }\n }\n }\n awaitable_state<void> closestate;\n co_await close(closestate, &socket);\n }\n}\n<\/pre>\n

    Again, a couple of new concepts show up in this example.\u00a0 First, we don\u2019t directly await on getaddrinfo. The getaddrinfo function returns a future_t<addrinfo_state>, which contains two pieces of information. The result of awaiting on future_t<addrinfo_state> gives an integer which indicates success or failure, but there is also a addrinfo pointer, which is used in the tcp_connect call. Finally, reading data on a socket potentially results in multiple callbacks as data arrives.\u00a0 This requires a different mechanism than just await\u2019ing the read. For this, there is the read_request_t type. As data arrives on a socket, it will pass the data on if there is an outstanding await.\u00a0 Otherwise, it holds onto that data until the next time an await occurs on it.<\/span>\u00a0<\/span><\/p>\n

    Finally, let\u2019s look at using these functions in combination.<\/span>\u00a0<\/span><\/p>\n

    [code lang=”cpp”]\nint main(int argc, char* argv[])\n{\n \/\/ Process command line\n if (argc == 1)\n {\n printf("testuv [–sequential] <file1> <file2> …");\n return -1;\n }<\/p>\n

    bool fRunSequentially = false;\n vector<string> files;\n for (int i = 1; i < argc; ++i)\n {\n string str = argv[i];\n if (str == "–sequential")\n fRunSequentially = true;\n else\n files.push_back(str);\n }<\/p>\n

    \/\/ start async color changer\n start_color_changer();<\/p>\n

    start_hello_world();\n if (fRunSequentially)\n uv_run(uv_default_loop(), UV_RUN_DEFAULT);<\/p>\n

    for (auto& file : files)\n {\n start_dump_file(file.c_str());\n if (fRunSequentially)\n uv_run(uv_default_loop(), UV_RUN_DEFAULT);\n }<\/p>\n

    start_http_google();\n if (fRunSequentially)\n uv_run(uv_default_loop(), UV_RUN_DEFAULT);<\/p>\n

    if (!fRunSequentially)\n uv_run(uv_default_loop(), UV_RUN_DEFAULT);<\/p>\n

    \/\/ stop the color changer and let it get cleaned up\n stop_color_changer();\n uv_run(uv_default_loop(), UV_RUN_DEFAULT);<\/p>\n

    uv_loop_close(uv_default_loop());<\/p>\n

    return 0;\n}\n<\/pre>\n

    \u00a0<\/span>This function supports two modes: the default parallel mode and a sequential mode. In sequential mode, we will run the libuv event loop after each task is started, allowing it to complete before starting the next. In parallel mode, all tasks (resumabled functions) are started and then resumed as awaits are completed.<\/span><\/p>\n

    Implementation<\/h2>\n

    This library is currently header only. Let\u2019s look at one of the wrapper functions.<\/span>\u00a0<\/span><\/p>\n

    [code lang=”cpp”]\nauto fs_open(uv_loop_t* loop, uv_fs_t* req, const char* path, int flags, int mode)\n{\n promise_t<uv_file> awaitable;\n auto state = awaitable._state->lock();\n req->data = state;<\/p>\n

    auto ret = uv_fs_open(loop, req, path, flags, mode,\n [](uv_fs_t* req) -> void\n {\n auto state = static_cast<promise_t<uv_file>::state_type*>(req->data);\n state->set_value(req->result);\n state->unlock();\n });<\/p>\n

    if (ret != 0)\n {\n state->set_value(ret);\n state->unlock();\n }\n return awaitable.get_future();\n}\n<\/pre>\n

    This function wraps the uv_fs_open function and the signature is almost identical to it. \u00a0It doesn’t take a callback and it returns future<int> rather than int. Internally, the promise_t<int> holds a reference counted state object, which contains an int and some other housekeeping information. Libuv provides a \u201cdata\u201d member to hold implementation specific information, which for us is a raw pointer to the state object. The actual callback passed to the uv_fs_open function is a lambda which will cast \u201cdata\u201d back to a state object and call its set_value method. If uv_fs_open returned a failure (which means the callback <\/span>will never be invoked), we directly set the value of the promise. Finally, we return a future that also has a reference counted pointer to the state. The returned future implements the necessary methods for co_await to work with it.<\/span>\u00a0<\/span><\/p>\n

    I currently have wrappers for the following libuv functions:<\/p>\n