c++ - Problema de duración no obvio con std:: promise y std:: future

c++11 stdthread

Question

Esta pregunta es muy similar a la anterior aquí: condición de raza en pthread_once ()?

Es esencialmente el mismo problema: la duración de una std::promise finaliza durante una llamada a promise::set_value (es decir, después de que se marcó el futuro asociado, pero antes de que pthread_once haya ejecutado)

Entonces sé que mi uso tiene este problema y que, por lo tanto, no puedo usarlo de esta manera. Sin embargo, creo que esto no es obvio. (En las sabias palabras de Scott Meyer: Haz que las interfaces sean fáciles de usar, correctas y difíciles de usar de forma incorrecta )

Presento un ejemplo a continuación:

• Tengo un hilo ( dispatcher ) que gira en una cola, aparece un ''trabajo'' (una std::function y lo ejecuta.
• Tengo una clase de utilidad llamada synchronous_job que bloquea el hilo que llama hasta que el ''trabajo'' se haya ejecutado en el hilo del despachador
• std::promise y std::future son miembros de synchronous_job : una vez que se establece el future , el hilo de llamadas bloqueado continúa, lo que da como resultado que synchronous_job salga de la pila y se destruya.
• Desafortunadamente, en este momento el dispatcher fue cambiado de contexto mientras que dentro de promise::set_value ; el future está marcado, pero la llamada a pthread_once no se ha ejecutado, y la pila pthread está de alguna manera corrupta, es decir, la próxima vez: punto muerto

Esperaría que un llamado a promise::set_value fuera atómico; el hecho de que necesita hacer más trabajo después de haber marcado el future inevitablemente conducirá a este tipo de problema cuando se utilizan estas clases de esta manera.

Entonces mi pregunta es: ¿Cómo lograr este tipo de sincronización usando std::promise y std::future , manteniendo su tiempo de vida asociado con la clase que proporciona este mecanismo de sincronización?

@Jonathan Wakely, ¿podría usar alguna clase de estilo RAII internamente que establezca la variable condition_variable en su destructor después de marcar el future ? Esto significa que incluso si la promise se destruye en medio de una llamada a set_value , el trabajo adicional de configuración de la variable de condición se completará correctamente. Solo una idea, no estoy seguro si puedes usarla ...

Un ejemplo de trabajo completo a continuación y el seguimiento de pila de la aplicación interbloqueada después de:

#include <iostream>
#include <thread>
#include <future>
#include <queue>
struct dispatcher
{
dispatcher()
{
_thread = std::move(std::thread(&dispatcher::loop, this));
}
void post(std::function<void()> job)
{
std::unique_lock<std::mutex> l(_mtx);
_jobs.push(job);
_cnd.notify_one();
}
private:
void loop()
{
for (;;)
{
std::function<void()> job;
{
std::unique_lock<std::mutex> l(_mtx);
while (_jobs.empty())
_cnd.wait(l);
job.swap(_jobs.front());
_jobs.pop();
}
job();
}
}
std::thread _thread;
std::mutex _mtx;
std::condition_variable _cnd;
std::queue<std::function<void()>> _jobs;
};
//-------------------------------------------------------------
struct synchronous_job
{
synchronous_job(std::function<void()> job, dispatcher& d)
: _job(job)
, _d(d)
, _f(_p.get_future())
{
}
void run()
{
_d.post(std::bind(&synchronous_job::cb, this));
_f.wait();
}
private:
void cb()
{
_job();
_p.set_value();
}
std::function<void()> _job;
dispatcher& _d;
std::promise<void> _p;
std::future<void> _f;
};
//-------------------------------------------------------------
struct test
{
test()
: _count(0)
{
}
void run()
{
synchronous_job job(std::bind(&test::cb, this), _d);
job.run();
}
private:
void cb()
{
std::cout << ++_count << std::endl;
}
int _count;
dispatcher _d;
};
//-------------------------------------------------------------
int main()
{
test t;
for (;;)
{
t.run();
}
}

El rastro de la pila de la aplicación interbloqueada:

Hilo 1 (hilo principal)

#0 0x00007fa112ed750c in pthread_cond_wait@@GLIBC_2.3.2 () from /lib64/libpthread.so.0
#1 0x00007fa112a308ec in __gthread_cond_wait (__mutex=<optimized out>, __cond=<optimized out>) at /hostname/tmp/syddev/Build/gcc-4.6.2/gcc-build/x86_64-unknown-linux-gnu/libstdc++-v3/include/x86_64-unknown-linux-gnu/bits/gthr-default.h:846
#2 std::condition_variable::wait (this=<optimized out>, __lock=...) at ../../../../libstdc++-v3/src/condition_variable.cc:56
#3 0x00000000004291d9 in std::condition_variable::wait<std::__future_base::_State_base::wait()::{lambda()#1}>(std::unique_lock<std::mutex>&, std::__future_base::_State_base::wait()::{lambda()#1}) (this=0x78e050, __lock=..., __p=...) at /hostname/sdk/gcc470/suse11/x86_64/include/c++/4.7.0/condition_variable:93
#4 0x00000000004281a8 in std::__future_base::_State_base::wait (this=0x78e018) at /hostname/sdk/gcc470/suse11/x86_64/include/c++/4.7.0/future:331
#5 0x000000000042a2d6 in std::__basic_future<void>::wait (this=0x7fff0ae515c0) at /hostname/sdk/gcc470/suse11/x86_64/include/c++/4.7.0/future:576
#6 0x0000000000428dd8 in synchronous_job::run (this=0x7fff0ae51580) at /home/lorimer/p4/Main/Source/Trading/Confucius/Test/Scratch/Test1/main.cpp:60
#7 0x0000000000428f97 in test::run (this=0x7fff0ae51660) at /home/lorimer/p4/Main/Source/Trading/Confucius/Test/Scratch/Test1/main.cpp:83
#8 0x0000000000427ad6 in main () at /home/lorimer/p4/Main/Source/Trading/Confucius/Test/Scratch/Test1/main.cpp:99

Subproceso 2 (subproceso de despachador)

#0 0x00007fa112ed8b5b in pthread_once () from /lib64/libpthread.so.0
#1 0x0000000000427946 in __gthread_once (__once=0x78e084, __func=0x4272d0 <__once_proxy@plt>) at /hostname/sdk/gcc470/suse11/x86_64/bin/../lib/gcc/x86_64-unknown-linux-gnu/4.7.0/../../../../include/c++/4.7.0/x86_64-unknown-linux-gnu/bits/gthr-default.h:718
#2 0x000000000042948b in std::call_once<void (std::__future_base::_State_base::*)(std::function<std::unique_ptr<std::__future_base::_Result_base, std::__future_base::_Result_base::_Deleter> ()>&, bool&), std::__future_base::_State_base* const, std::reference_wrapper<std::function<std::unique_ptr<std::__future_base::_Result_base, std::__future_base::_Result_base::_Deleter> ()> >, std::reference_wrapper<bool> >(std::once_flag&, void (std::__future_base::_State_base::*&&)(std::function<std::unique_ptr<std::__future_base::_Result_base, std::__future_base::_Result_base::_Deleter> ()>&, bool&), std::__future_base::_State_base* const&&, std::reference_wrapper<std::function<std::unique_ptr<std::__future_base::_Result_base, std::__future_base::_Result_base::_Deleter> ()> >&&, std::reference_wrapper<bool>&&) (__once=..., __f=
@0x7fa111ff6be0: (void (std::__future_base::_State_base::*)(std::__future_base::_State_base * const, std::function<std::unique_ptr<std::__future_base::_Result_base, std::__future_base::_Result_base::_Deleter>()> &, bool &)) 0x42848a <std::__future_base::_State_base::_M_do_set(std::function<std::unique_ptr<std::__future_base::_Result_base, std::__future_base::_Result_base::_Deleter> ()>&, bool&)>) at /hostname/sdk/gcc470/suse11/x86_64/include/c++/4.7.0/mutex:819
#3 0x000000000042827d in std::__future_base::_State_base::_M_set_result(std::function<std::unique_ptr<std::__future_base::_Result_base, std::__future_base::_Result_base::_Deleter> ()>, bool) (this=0x78e018, __res=..., __ignore_failure=false) at /hostname/sdk/gcc470/suse11/x86_64/include/c++/4.7.0/future:362
#4 0x00000000004288d5 in std::promise<void>::set_value (this=0x7fff0ae515a8) at /hostname/sdk/gcc470/suse11/x86_64/include/c++/4.7.0/future:1206
#5 0x0000000000428e2a in synchronous_job::cb (this=0x7fff0ae51580) at /home/lorimer/p4/Main/Source/Trading/Confucius/Test/Scratch/Test1/main.cpp:66
#6 0x000000000042df53 in std::_Mem_fn<void (synchronous_job::*)()>::operator() (this=0x78c6e0, __object=0x7fff0ae51580) at /hostname/sdk/gcc470/suse11/x86_64/include/c++/4.7.0/functional:554
#7 0x000000000042d77c in std::_Bind<std::_Mem_fn<void (synchronous_job::*)()> (synchronous_job*)>::__call<void, , 0ul>(std::tuple<>&&, std::_Index_tuple<0ul>) (this=0x78c6e0, __args=...) at /hostname/sdk/gcc470/suse11/x86_64/include/c++/4.7.0/functional:1156
#8 0x000000000042cb28 in std::_Bind<std::_Mem_fn<void (synchronous_job::*)()> (synchronous_job*)>::operator()<, void>() (this=0x78c6e0) at /hostname/sdk/gcc470/suse11/x86_64/include/c++/4.7.0/functional:1215
#9 0x000000000042b772 in std::_Function_handler<void (), std::_Bind<std::_Mem_fn<void (synchronous_job::*)()> (synchronous_job*)> >::_M_invoke(std::_Any_data const&) (__functor=...) at /hostname/sdk/gcc470/suse11/x86_64/include/c++/4.7.0/functional:1926
#10 0x0000000000429f2c in std::function<void ()>::operator()() const (this=0x7fa111ff6da0) at /hostname/sdk/gcc470/suse11/x86_64/include/c++/4.7.0/functional:2311
#11 0x0000000000428c3c in dispatcher::loop (this=0x7fff0ae51668) at /home/lorimer/p4/Main/Source/Trading/Confucius/Test/Scratch/Test1/main.cpp:39

Answer 1

En respuesta directa a su pregunta, la respuesta correcta es dar la std::promise al hilo. De esta forma, se garantiza que existirá mientras el hilo lo quiera.

Bajo el capó, std::future y std::promise tienen un estado compartido que ambos apuntan a, y se garantiza que permanecerán disponibles hasta que ambos lados se destruyan. Conceptualmente, esto es similar tanto a la promesa como al futuro, ya que ambos tienen copias individuales de un shared_ptr para el mismo objeto. Este objeto contiene los mecanismos subyacentes necesarios para pasar el estado, bloque y otras operaciones.

En cuanto a intentar señalar la destrucción, el problema es dónde existiría esta variable de condición. El área compartida se destruye una vez que se destruyen todos los futuros y promesas asociados. El punto muerto está ocurriendo porque el área se está destruyendo mientras todavía se está utilizando (porque el compilador no sabe que otro hilo todavía está accediendo a la promesa, ya que está siendo destruida). Agregar variables de condición adicionales a cualquier estado compartido no ayudaría, ya que también se destruirían.

Answer 2

Respondiendo mi propia pregunta, para ofrecer una solución viable. No usa std::promise o std::future , pero logra la sincronización que estoy buscando.

Actualice synchronous_job para usar std::condition_variable y std::mutex lugar:

Editar: se actualizó para incluir un indicador booleano según lo sugerido por Dave S

struct synchronous_job
{
synchronous_job(std::function<void()> job, dispatcher& d)
: _job(job)
, _d(d)
, _done(false)
{
}
void run()
{
_d.post(std::bind(&synchronous_job::cb, this));
std::unique_lock<std::mutex> l(_mtx);
if (!_done)
_cnd.wait(l);
}
private:
void cb()
{
_job();
std::unique_lock<std::mutex> l(_mtx);
_done = true;
_cnd.notify_all();
}
std::function<void()> _job;
dispatcher& _d;
std::condition_variable _cnd;
std::mutex _mtx;
bool _done;
};

Answer 3

std::promise es como cualquier otro objeto: solo puede acceder a él de un hilo a la vez. En este caso, está llamando a set_value() y destruyendo el objeto de hilos separados sin suficiente sincronización: en ninguna parte de la especificación dice que set_value no tocará el objeto de la promise después de preparar el future .

Sin embargo, dado que este futuro se utiliza para una sincronización de una sola vez, no es necesario que haga eso de todos modos: cree la promesa / futuro par justo en run() , y pase la promesa al hilo:

struct synchronous_job
{
synchronous_job(std::function<void()> job, dispatcher& d)
: _job(job)
, _d(d)
{
}
void run(){
std::promise<void> p;
std::future<void> f=p.get_future();
_d.post(
[&]{
cb(std::move(p));
});
f.wait();
}
private:
void cb(std::promise<void> p)
{
_job();
p.set_value();
}
std::function<void()> _job;
dispatcher& _d;
};

Answer 4

La respuesta canónica es nunca std :: bind a esto, sino a std :: weak_ptr. Cuando reciba la devolución de llamada, bloquee () y verifique NULL antes de invocar la devolución de llamada.

O, re-declarado, nunca llame a una función miembro (desde afuera) desde un alcance que no contenga un shared_ptr para el objeto.