60

High level
I want to call some functions with no return value in a async mode without waiting for them to finish. If I use std::async the future object doesn't destruct until the task is over, this make the call not sync in my case.

Example

void sendMail(const std::string& address, const std::string& message)
{
    //sending the e-mail which takes some time...
}

myResonseType processRequest(args...)
{
    //Do some processing and valuate the address and the message...

    //Sending the e-mail async
    auto f = std::async(std::launch::async, sendMail, address, message);

    //returning the response ASAP to the client
    return myResponseType;

} //<-- I'm stuck here until the async call finish to allow f to be destructed.
  // gaining no benefit from the async call.

My questions are

  1. Is there a way to overcome this limitation?
  2. if (1) is no, should I implement once a thread that will take those "zombie" futures and wait on them?
  3. Is (1) and (2) are no, is there any other option then just build my own thread pool?

note:
I rather not using the option of thread+detach (suggested by @galop1n) since creating a new thread have an overhead I wish to avoid. While using std::async (at least on MSVC) is using an inner thread pool.

Thanks.

6
  • 2
    You cannot do that with async by design.
    – eerorika
    Commented Feb 3, 2014 at 15:36
  • I know I can't. what I'm asking is if someone have a simple way to extend the basic std::async in order to do that. and if not what approch should I take in order to achieve that. (maybe not using std::async at all). Commented Feb 3, 2014 at 15:40
  • 4
    Well, that answered to the your title question :) If you want thread pooling, perhaps you could do that explicitly rather than rely on implementation details.
    – eerorika
    Commented Feb 3, 2014 at 15:44
  • If you need to send more than one e-mail consider using Asynchronous Agents Library (part of PPL shipped with VS). msdn.microsoft.com/en-us/library/dd492627.aspx
    – alexm
    Commented Feb 3, 2014 at 17:36
  • 1
    Just fork another process for sendMail and forget about it. :) Commented Mar 20, 2014 at 20:53

7 Answers 7

25

You can move the future into a global object, so when the local future's destructor runs it doesn't have to wait for the asynchronous thread to complete.

std::vector<std::future<void>> pending_futures;

myResonseType processRequest(args...)
{
    //Do some processing and valuate the address and the message...

    //Sending the e-mail async
    auto f = std::async(std::launch::async, sendMail, address, message);

    // transfer the future's shared state to a longer-lived future
    pending_futures.push_back(std::move(f));

    //returning the response ASAP to the client
    return myResponseType;

}

N.B. This is not safe if the asynchronous thread refers to any local variables in the processRequest function.

While using std::async (at least on MSVC) is using an inner thread pool.

That's actually non-conforming, the standard explicitly says tasks run with std::launch::async must run as if in a new thread, so any thread-local variables must not persist from one task to another. It doesn't usually matter though.

8
  • 6
    That is a bad approach, I'll end up with a constantly growing vector (pending_futuers) Commented Mar 24, 2014 at 13:46
  • 7
    So go through the vector periodically and remove the ready futures. You could add that to the processRequest function, so every time you call it you see if there are any ready futures that can be removed from the vector. That's not complicated. Commented Mar 24, 2014 at 13:50
  • 3
    Your question was how to avoid waiting in the future destructor, which I answered. If you want to create your own thread pool that's fine (although I doubt your thread pool is as efficient as the one in the Windows runtime) but that doesn't change what you originally asked. Commented Mar 24, 2014 at 16:56
  • 1
    If std::async is called in class method, you can assign future returned by it to a class member. This way waiting for future will be deferred until class is destructed.
    – Amit
    Commented Oct 28, 2015 at 10:33
  • 1
    @starfury no you can't, that won't compile. f is an lvalue, you can't construct another future from it without turning it into an rvaue. Commented Jul 12, 2016 at 13:23
21

why do you not just start a thread and detach if you do not care on joining ?

std::thread{ sendMail, address, message}.detach();   

std::async is bound to the lifetime of the std::future it returns and their is no alternative to that.

Putting the std::future in a waiting queue read by an other thread will require the same safety mechanism as a pool receiving new task, like mutex around the container.

Your best option, then, is a thread pool to consume tasks directly pushed in a thread safe queue. And it will not depends on a specific implementation.

Below a thread pool implementation taking any callable and arguments, the threads do poling on the queue, a better implementation should use condition variables (coliru) :

#include <iostream>
#include <queue>
#include <memory>
#include <thread>
#include <mutex>
#include <functional>
#include <string>

struct ThreadPool {
    struct Task {
        virtual void Run() const = 0;
        virtual ~Task() {};
    };   

    template < typename task_, typename... args_ >
    struct RealTask : public Task {
        RealTask( task_&& task, args_&&... args ) : fun_( std::bind( std::forward<task_>(task), std::forward<args_>(args)... ) ) {}
        void Run() const override {
            fun_();
        }
    private:
        decltype( std::bind(std::declval<task_>(), std::declval<args_>()... ) ) fun_;
    };

    template < typename task_, typename... args_ >
    void AddTask( task_&& task, args_&&... args ) {
        auto lock = std::unique_lock<std::mutex>{mtx_};
        using FinalTask = RealTask<task_, args_... >;
        q_.push( std::unique_ptr<Task>( new FinalTask( std::forward<task_>(task), std::forward<args_>(args)... ) ) );
    }

    ThreadPool() {
        for( auto & t : pool_ )
            t = std::thread( [=] {
                while ( true ) {
                    std::unique_ptr<Task> task;
                    {
                        auto lock = std::unique_lock<std::mutex>{mtx_};
                        if ( q_.empty() && stop_ ) 
                            break;
                        if ( q_.empty() )
                            continue;
                        task = std::move(q_.front());
                        q_.pop();
                    }
                    if (task)
                        task->Run();
                }
            } );
    }
    ~ThreadPool() {
        {
            auto lock = std::unique_lock<std::mutex>{mtx_};
            stop_ = true;
        }
        for( auto & t : pool_ )
            t.join();
    }
private:
    std::queue<std::unique_ptr<Task>> q_;
    std::thread pool_[8]; 
    std::mutex mtx_;
    volatile bool stop_ {};
};

void foo( int a, int b ) {
    std::cout << a << "." << b;
}
void bar( std::string const & s) {
    std::cout << s;
}

int main() {
    ThreadPool pool;
    for( int i{}; i!=42; ++i ) {
        pool.AddTask( foo, 3, 14 );    
        pool.AddTask( bar, " - " );    
    }
}
7
  • 2
    std::async when compiling with MSVC is using an inner thread pool. creating a thread myself each time is having a performance overhead I wish to avoid. Commented Feb 3, 2014 at 15:30
  • 12
    The program has a data race: ThreadPool and ~ThreadPool access stop_ potentially simultaneously. volatile has no useful (portable) semantics for multithreading: it needs to be a std::atomic or ~ThreadPool needs to access it with mtx_ held. Your threads also busy-wait, it would be nice to block on a condition variable while the queue is empty.
    – Casey
    Commented Feb 3, 2014 at 17:59
  • @Casey This is only a proof of concept for variable task queue, i tried to keep it as simple as possible. Also, I never give a serious try to the c++11 condition variables ( only native ones ) and did not want to miss something, in fact, i use that sample to test them at this time. I add a note about using condition variable in a real use case.
    – galop1n
    Commented Feb 3, 2014 at 18:07
  • 5
    Your thread in ThreadPool busy spins wasting CPU. Learn how to use condition variables. Commented Mar 20, 2014 at 20:50
  • I am personally not very fond of Microsoft's non-conforming implementation of std::async. I investigated on this for a colleague and there is only an overhead for the first time std::async is called. At the first call, all threads are initialized and put in a waiting state atleast using VS2010
    – Jens Munk
    Commented Apr 13, 2014 at 17:43
10

Rather than moving the future into a global object (and manually manage deletion of unused futures), you can actually move it into the local scope of the asynchronously called function.

"Let the async function take its own future", so to speak.

I have come up with this template wrapper which works for me (tested on Windows):

#include <future>

template<class Function, class... Args>
void async_wrapper(Function&& f, Args&&... args, std::future<void>& future,
                   std::future<void>&& is_valid, std::promise<void>&& is_moved) {
    is_valid.wait(); // Wait until the return value of std::async is written to "future"
    auto our_future = std::move(future); // Move "future" to a local variable
    is_moved.set_value(); // Only now we can leave void_async in the main thread

    // This is also used by std::async so that member function pointers work transparently
    auto functor = std::bind(f, std::forward<Args>(args)...);
    functor();
}

template<class Function, class... Args> // This is what you call instead of std::async
void void_async(Function&& f, Args&&... args) {
    std::future<void> future; // This is for std::async return value
    // This is for our synchronization of moving "future" between threads
    std::promise<void> valid;
    std::promise<void> is_moved;
    auto valid_future = valid.get_future();
    auto moved_future = is_moved.get_future();

    // Here we pass "future" as a reference, so that async_wrapper
    // can later work with std::async's return value
    future = std::async(
        async_wrapper<Function, Args...>,
        std::forward<Function>(f), std::forward<Args>(args)...,
        std::ref(future), std::move(valid_future), std::move(is_moved)
    );
    valid.set_value(); // Unblock async_wrapper waiting for "future" to become valid
    moved_future.wait(); // Wait for "future" to actually be moved
}

I am a little surprised it works because I thought that the moved future's destructor would block until we leave async_wrapper. It should wait for async_wrapper to return but it is waiting inside that very function. Logically, it should be a deadlock but it isn't.

I also tried to add a line at the end of async_wrapper to manually empty the future object:

our_future = std::future<void>();

This does not block either.

2
  • The thread enqueueing the job will block at moved_future.wait() until the threadpool starts this async job. In case it was busy working on previously enqueued jobs this stall would be significant. Commented Feb 12, 2017 at 20:21
  • Works on windows, but unfortunately does not compile with NDK
    – Robin
    Commented Mar 21, 2018 at 12:31
1

Based on hanshenrik's answer, here is a reusable class that acts as a 'garbage collector' of std::future and that does not rely on static variables which would make the solution hard to unit test. Kept the internal thread instead of trying to wait on them when calling Add() so that Add() runs in a deterministic and short amount of time.

#pragma once

#include <chrono>
#include <future>
#include <list>
#include <mutex>
#include <thread>

template <typename T>
class CFutureGC
{
   std::list<std::future<T>> m_all;
   bool m_runnerStop{false};
   mutable std::mutex m_mtx;
   std::thread m_runner;  // declare last!

   void GC(std::chrono::milliseconds gcSleep)
   {
      using namespace std::chrono;

      while (!m_runnerStop) {
         auto it = m_all.end();

         {
            std::unique_lock<std::mutex> lock(m_mtx);
            it = m_all.begin();
         }

         if (m_all.end() == it) {
            std::this_thread::sleep_for(gcSleep);
         } else {
            auto res = it->wait_for(0ms);

            // defered: used will call get() to run the future
            // timeout: job still running
            if (std::future_status::ready == res) {
               it = m_all.end();

               std::unique_lock<std::mutex> lock(m_mtx);
               m_all.pop_front();  // does not invalidate other iterators
            }
         }
      }
   }

  public:
   CFutureGC(const std::chrono::milliseconds &gcSleep)
       : m_runner([this, &gcSleep]() { GC(gcSleep); })
   {
   }

   ~CFutureGC()
   {
      m_runnerStop = true;
      if (m_runner.joinable()) {
         m_runner.join();
      }

      // Will wait until all std::future-s return
   }

   void Add(std::future<T> &&f)
   {
      std::unique_lock<std::mutex> lock(m_mtx);
      m_all.push_back(std::move(f));  // does not invalidate other iterators
   }

   size_t Count() const
   {
      std::unique_lock<std::mutex> lock(m_mtx);
      auto count = m_all.size();
      return count;
   }

   CFutureGC(const CFutureGC &) = delete;
   CFutureGC &operator=(const CFutureGC &) = delete;
};

Usage:

CFutureGC<void> futures;
futures.Add(std::move(std::async(std::launch::async, [](){}));
0

I got myself into the same situation and applied a dirty fix in which I just push all the futures onto a global list:

auto processRequest() -> void {
    static std::forward_list<std::future<void>> requests;
    requests.emplace_front(std::async(std::launch::async, sendMail, "address", "message"));
}

Of course you have to think about what you eventually want to do with all these futures or if you even accept more than one ongoing request. I think the solution would be not to use std::async which is what I'm planning to do in the long run.

Full example:

#include <thread>
#include <future>
#include <chrono>
#include <iostream>
#include <forward_list>

using namespace std::chrono_literals;

auto sendMail(const std::string& address, const std::string& message) -> void {
    std::cout << "before send!" << std::endl;
    std::this_thread::sleep_for(1000ms);
    std::cout << "after send!" << std::endl;
}

auto processRequest() -> void {
    static std::forward_list<std::future<void>> requests;
    requests.emplace_front(std::async(std::launch::async, sendMail, "address", "message"));
}

auto main() -> int {
    std::cout << "Before call" << std::endl;
    processRequest();
    processRequest();
    std::cout << "After call" << std::endl;
}

Prints (one variant):

Before call
After call
before send!
before send!
after send!
after send!
0

You need to make your future a pointer. Below is exactly what you are looking for:

std::make_unique<std::future<void>>((std::async(std::launch::async, sendMail, address, message))).reset();
4
  • Thanks for the idea, but it won't work. it's true that static variables are only visible to the block they are defined in but they are also singeltons in a sense that two calls to this function will use the same f, which will make the second call blocked on this line until the first call future will be back. Commented Feb 18, 2021 at 10:35
  • @RoeeGavirel I just edited my answer. No more blocks with the new code. Have fun..
    – Kitiara
    Commented Feb 19, 2021 at 7:35
  • I'm long gone since the days I was programming in C/C++, but wouldn't it leave a dangling pointer and will leas to memory leak. Commented Mar 7, 2021 at 6:59
  • @RoeeGavirel unique_ptr is a smart pointer, which provides the additional feature of automatic memory management. Besides i'm destroying the object by reset(); as soon as i created the unique_ptr so there is absolutely no possible way to cause a memory leak.
    – Kitiara
    Commented Mar 7, 2021 at 10:03
-1

i have no idea what i'm doing, but this seem to work:

// :( http://www.open-std.org/jtc1/sc22/wg21/docs/papers/2012/n3451.pdf
template<typename T>
void noget(T&& in)
{
    static std::mutex vmut;
    static std::vector<T> vec;
    static std::thread getter;
    static std::mutex single_getter;
    if (single_getter.try_lock())
    {
        getter = std::thread([&]()->void
        {
            size_t size;
            for(;;)
            {
                do
                {
                    vmut.lock();
                    size=vec.size();
                    if(size>0)
                    {
                        T target=std::move(vec[size-1]);
                        vec.pop_back();
                        vmut.unlock();
                        // cerr << "getting!" << endl;
                        target.get();
                    }
                    else
                    {
                        vmut.unlock();
                    }
                }while(size>0);
                // ¯\_(ツ)_/¯
                std::this_thread::sleep_for(std::chrono::milliseconds(100));
            }
        });
        getter.detach();
    }
    vmut.lock();
    vec.push_back(std::move(in));
    vmut.unlock();
}

it creates a dedicated getter thread for each type of future you throw at it (eg. if you give a future and future, you'll have 2 threads. if you give it 100x future, you'll still only have 2 threads), and when there's a future you don't want to deal with, just do notget(fut); - you can also noget(std::async([]()->void{...})); works just fine, no block, it seems. warning, do not try to get the value from a future after using noget() on it. that's probably UB and asking for trouble.

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.