博客

C++11多线程编程 第十章: 使用packaged_task优雅的让同步函数异步执行

  • Post author:
  • Post category:其他


C++11 Multithreading – Part 10: packaged_task<> Example and Tutorial


Varun

July 2, 2017

C++11 Multithreading – Part 10: packaged_task Example and Tutorial

2018-08-18T15:23:07+00:00

C++ 11

,

c++11 Threads

,

Multithreading

,

packaged_task


2 Comments

In this example we will discuss c++11 std::packaged_task feature and its uses.

std::packaged_task<>


std::packaged_task<>

is a class template and represents a asynchronous task. It encapsulates,

  1. A callable entity i.e either function, lambda function or function object.
  2. A shared state that stores the value returned or thrown exception by associated callback.

Need of std::packaged_task<>

Suppose we have an existing function that fetches the data from DB and return i.e.

1

2

3

4

5

6

7

// Fetch some data from DB

std::string getDataFromDB( std::string token)

{

// Do some stuff to fetch the data

std::string data = “Data fetched from DB by Filter :: ” + token;

return data;

}

Now we want to execute this function in a separate thread. But how we will fetch the result or exception back in main thread after other thread is finished ?

One way is to change the declaration of function and pass a std::promise<> in the function. Before passing the std::promise<> object in thread function, fetch the associated std::future<> out of it and keep that in main thread. Now, before thread function returns its value, it should set that in passed std::promise<> argument, so that it can be available in associated std::future<> object in main thread. Check below tutorial for this approach i.e,


C++11 Multithreading – Part 8: std::future , std::promise and Returning values from Thread

But creating this std::promise<> and changing function code can be prevented if we use std::packaged_task<>.

Using packaged_task<> with function to create Asynchronous tasks


std::packaged_task<>

can wrap around a normal function and make it applicable to run as asynchronous function.

When

std::packaged_task<>

is called in a separate thread, it calls the associated callback and stores the return  value/exception in its internal shared state. This value can be accessed in other thread or main function through

std::future<>

object.

Let’s create a std::packaged_task<> from above mentioned function, execute in separate thread and fetch result from its future<> object.

Creating std::packaged_task<> object

std::package_task<> is a class template, therefore we need to pass template parameter to packaged_task<> i.e. type of callable function

1

2

// Create a packaged_task<> that encapsulated the callback i.e. a function

std::packaged_task<std::string (std::string)> task(getDataFromDB);

Fetch the future object from it,

1

2

// Fetch the associated future<> from packaged_task<>

std::future<std::string> result = task.get_future();

Passing packaged_task<> to a thread,

std::packaged_task<> is movable but not copy-able, so we need to move it to thread i.e.

1

2

// Pass the packaged_task to thread to run asynchronously

std::thread th(std::move(task), “Arg”);

As packaged_task was only movable and not copy-able, therefore we fetched the

std::future<>

object from it before moving it to thread.

Thread will execute this task, which internally calls associated callable entity i.e. our function

getDataFromDB()

.

Now when this function returns the value, std::packaged_task<> sets it to associated shared state and result or exception returned by

getDataFromDB()

will eventually be available in associated future object.

In main function, fetch result from future<> object i.e.

1

2

// Fetch the result of packaged_task<> i.e. value returned by getDataFromDB()

std::string data =  result.get();

get() function will block the calling thread until the callable entity returns and std::packaged_task<> set the data in its shareable state.

Complete example is as follows,

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

#include <iostream>

#include <thread>

#include <future>

#include <string>

// Fetch some data from DB

std::string getDataFromDB( std::string token)

{

// Do some stuff to fetch the data

std::string data = “Data fetched from DB by Filter :: ” + token;

return data;

}

int main()

{

// Create a packaged_task<> that encapsulated the callback i.e. a function

std::packaged_task<std::string (std::string)> task(getDataFromDB);

// Fetch the associated future<> from packaged_task<>

std::future<std::string> result = task.get_future();

// Pass the packaged_task to thread to run asynchronously

std::thread th(std::move(task), “Arg”);

// Join the thread. Its blocking and returns when thread is finished.

th.join();

// Fetch the result of packaged_task<> i.e. value returned by getDataFromDB()

std::string data =  result.get();

std::cout <<  data << std::endl;

return 0;

}


Output:

1

Data fetched from DB by Filter :: Arg

On similar lines we can create a packaged_task with lambda function and function objects too i.e.

Creating packaged_task with Lambda Function

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

#include <iostream>

#include <thread>

#include <future>

#include <string>

int main()

{

// Create a packaged_task<> that encapsulated a lambda function

std::packaged_task<std::string (std::string)> task([](std::string token){

// Do some stuff to fetch the data

std::string data = “Data From ” + token;

return data;

});

// Fetch the associated future<> from packaged_task<>

std::future<std::string> result = task.get_future();

// Pass the packaged_task to thread to run asynchronously

std::thread th(std::move(task), “Arg”);

// Join the thread. Its blocking and returns when thread is finished.

th.join();

// Fetch the result of packaged_task<> i.e. value returned by getDataFromDB()

std::string data =  result.get();

std::cout <<  data << std::endl;

return 0;

}


Output:

1

Data fetched from DB by Filter :: Arg

Creating packaged_task with Function Object

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

#include <iostream>

#include <thread>

#include <future>

#include <string>

/*

* Function Object to Fetch Data from DB

*/

struct DBDataFetcher

{

std::string operator()(std::string token)

{

// Do some stuff to fetch the data

std::string data = “Data From ” + token;

return data;

}

};

int main()

{

// Create a packaged_task<> that encapsulated a lambda function

std::packaged_task<std::string (std::string)> task(std::move(DBDataFetcher()));

// Fetch the associated future<> from packaged_task<>

std::future<std::string> result = task.get_future();

// Pass the packaged_task to thread to run asynchronously

std::thread th(std::move(task), “Arg”);

// Join the thread. Its blocking and returns when thread is finished.

th.join();

// Fetch the result of packaged_task<> i.e. value returned by getDataFromDB()

std::string data =  result.get();

std::cout <<  data << std::endl;

return 0;

}


Output:

1

Data fetched from DB by Filter :: Arg

ps:

就packaged_task来说, 其能完成的, 通过future/promise都能完成.  但是对于已有代码来说, 使用packaged_task能有更少的浸入性, 不用为了异步执行函数修改现有代码, 也不用为了异步执行函数在封装一个函数传入promise调用已有函数.