Databases
StackOverflow2013
Postsdbo
POGiven a container of futures, how to perform all gets in a non-blocking way?
 

Note that there are some explanatory texts on larger screens.

plurals
  1. POGiven a container of futures, how to perform all gets in a non-blocking way?
    primarykey
    data
    text
    <p>So, I'm trying to create a generic way to both create a container of futures, as well as perform all the future.get()' in a non-blocking way. </p> <p>I anticipate that the completion time the tasks will take should range from a few hundred milliseconds, up to 2 minutes typically. Some however, may not complete at all. There will be at least 10,000 tasks to perform in a typical run.</p> <p>I want the quickest-returning task results back without being held up by other, more long-running tasks in the futures container.</p> <p>Here's what I have so far just using dummy sleep times to simulate the task completion delays (design thanks in large part to good posts made here, such as <a href="https://stackoverflow.com/questions/10902230/stdasync-variant-which-works-over-a-collection" title="std::async variant which works over a collection">this</a>, and <a href="https://stackoverflow.com/questions/11910051/implement-search-using-futures-async-and-thread-in-c11?rq=1" title="Implement search using futures, async, and thread in C++11">this</a>):</p> <pre><code>#include &lt;future&gt; #include &lt;vector&gt; #include &lt;iostream&gt; #include &lt;random&gt; #include &lt;chrono&gt; #include &lt;ratio&gt; #include &lt;thread&gt; #include &lt;algorithm&gt; size_t rand_from_range(const size_t, const size_t); int rand_sleep_range(const size_t, const size_t); template&lt;class CT&gt; size_t get_async_all( CT&amp; ); // Given a function and a collection, // return a vector of futures. template&lt;class Function, class CT&gt; auto async_all( Function f, CT coll ) -&gt; std::vector&lt;decltype(std::async(f, *std::begin(coll)))&gt; { std::vector&lt;decltype(std::async(f, *std::begin(coll)))&gt; futures; futures.reserve(coll.size()); for (auto&amp; element : coll) futures.push_back(std::async(f, element)); return futures; } // Given the beginning and end of a number // range, return a random number therein. size_t rand_from_range( const size_t range_begin, const size_t range_end ) { std::uniform_int_distribution&lt;size_t&gt; distr(range_begin, range_end); std::random_device dev; return distr(dev); } // Given a shortest and longest duration, put the calling // thread to sleep for a random duration therein. // (in milliseconds) int rand_sleep_range( const size_t shortest_time, const size_t longest_time ) { std::chrono::milliseconds sleep_time(rand_from_range(shortest_time, longest_time)); std::this_thread::sleep_for(sleep_time); return (int)sleep_time.count(); } // Given a container of futures, perform all // get()'s. template&lt;class CT&gt; size_t get_async_all( CT&amp; async_coll ) { size_t get_ctr(0); const size_t future_cnt = async_coll.size(); std::vector&lt;size_t&gt; completed; completed.reserve(future_cnt); while (true) { for (size_t ndx = 0; ndx &lt; future_cnt; ++ndx) { // Check to see if this ndx' future has completed already. if (std::none_of(std::begin(completed), std::end(completed), [=](size_t x) { return (x == ndx); })) { // No, this one hasn't completed // yet, attempt to process it. auto&amp; f = async_coll[ndx]; if (f.wait_for(std::chrono::milliseconds(10)) == std::future_status::ready) { f.get(); // The future's work gets done here. ++get_ctr; completed.push_back(ndx); if (completed.size() == future_cnt) break; // for() } } } if (completed.size() == future_cnt) break; // while() } return get_ctr; } int main() { // A dummy container of ints. std::vector&lt;int&gt; my_vec(100); for (auto&amp; elem : my_vec) elem = rand_from_range(1, 100); // A dummy function lambda. auto my_func = [](int x) { int x_ = x; int sleep_time = rand_sleep_range(100, 20000); // in ms. x *= 2; std::cout &lt;&lt; " after sleeping " &lt;&lt; sleep_time &lt;&lt; "ms \t" &lt;&lt; "f(" &lt;&lt; x_ &lt;&lt; ") = " &lt;&lt; x &lt;&lt; std::endl; }; // Create and execute the container of futures. auto async_coll = async_all(my_func, my_vec); size_t count = get_async_all(async_coll); std::cout &lt;&lt; std::endl &lt;&lt; count &lt;&lt; " items completed. \n"; } </code></pre> <p>So, my questions are:</p> <ul> <li>Are there any gotchas to the approach I'm using?</li> <li>Is there a better/more elegant approach for the get_async_all() than the one I'm using? Or anything else I'm doing, for that matter.</li> </ul> <p>Thanks to anyone for taking the time to look over the code, and to give me any constructive criticism or feedback.</p>
    singulars
    1. This table or related slice is empty.
    1. This table or related slice is empty.
    1. PTQuestion
    1. USCommunity
    1. USHerpin the Derps
    plurals
    1. PL
      singulars
      1. LTLinked
    2. PL
      singulars
      1. LTLinked
    1. This table or related slice is empty.
    1. This table or related slice is empty.
    1. PO
      singulars
      1. PTAnswer
    1. VO
      singulars
      1. POGiven a container of futures, how to perform all gets in a non-blocking way?
      1. This table or related slice is empty.
      1. VTUpMod
    2. VO
      singulars
      1. POGiven a container of futures, how to perform all gets in a non-blocking way?
      1. This table or related slice is empty.
      1. VTUpMod
    3. VO
      singulars
      1. POGiven a container of futures, how to perform all gets in a non-blocking way?
      1. USsvick
      1. VTFavorite
    1. COWell, your code is not much of non-blocking, since it blocks until the last future completes. But I don't think you can do better without something like `then()` (which is not in C++11).
      singulars
      1. POGiven a container of futures, how to perform all gets in a non-blocking way?
      1. USsvick
    2. CO@svick Oh, I think I see your point. Hmm. I guess the most important thing is that when a task completes, it can be returned to the caller immediately, while the remainder complete in due course. If I created a deque that I passed by ref into get_async_all() as well, and push_back() on it with a completed future from the interior if() statement, would that approach work? Maybe I could have an output threadpool accept it? I plan to implement this algorithm as part of a library, if that makes any difference. Thanks for the input!
      singulars
      1. POGiven a container of futures, how to perform all gets in a non-blocking way?
      1. USHerpin the Derps
    3. COThe correct test if a task has *completed* is `.wait_for(std::chrono::seconds(0))`.
      singulars
      1. POGiven a container of futures, how to perform all gets in a non-blocking way?
      1. USXeo
 

Querying!

 
Guidance
A row detail

Detail views are divided into sections. All the information in the data section comes from columns in the selected row. The other sections display data from other, related rows.

Related data can be related in a to-one or a to-many fashion. Captions of data related in a to-many fashion link to a list view showing a filtered view of the table.

Try moving around until you find a non-empty to-many entry and click on the label to get to one. You can move back to the root by clicking on the database name in the header.

SQuiL has stopped working due to an internal error.

If you are curious you may find further information in the browser console, which is accessible through the devtools (F12).

Reload