pthread - threads c++>

¿Cómo debo lidiar con mutexes en tipos móviles en C++?

Question

Por diseño, std::mutex no es móvil ni se puede copiar. Esto significa que una clase A , que contiene un mutex, no recibirá un constructor de movimiento predeterminado.

¿Cómo haría que este tipo A mueva de manera segura?

Answer 1

Comencemos con un poco de código:

class A
{
    using MutexType = std::mutex;
    using ReadLock = std::unique_lock<MutexType>;
    using WriteLock = std::unique_lock<MutexType>;

    mutable MutexType mut_;

    std::string field1_;
    std::string field2_;

public:
    ...

Puse algunos alias de tipo bastante sugerentes allí que realmente no aprovecharemos en C ++ 11, pero se volverán mucho más útiles en C ++ 14. Sé paciente, llegaremos allí.

Su pregunta se reduce a:

¿Cómo escribo el constructor de movimientos y el operador de asignación de movimientos para esta clase?

Comenzaremos con el constructor de movimientos.

Move Constructor

Tenga en cuenta que el miembro mutex se ha hecho mutable . Estrictamente hablando, esto no es necesario para los miembros del movimiento, pero supongo que también desea copiar miembros. Si ese no es el caso, no hay necesidad de hacer mutex mutable .

Al construir A , no necesita bloquear this->mut_ . Pero debe bloquear el mut_ del objeto desde el que está construyendo (mover o copiar). Esto se puede hacer así:

    A(A&& a)
    {
        WriteLock rhs_lk(a.mut_);
        field1_ = std::move(a.field1_);
        field2_ = std::move(a.field2_);
    }

Tenga en cuenta que primero tuvimos que construir por defecto los miembros de this , y luego asignarles valores solo después de que a.mut_ esté bloqueado.

Asignación de movimiento

El operador de asignación de movimiento es sustancialmente más complicado porque no sabe si algún otro subproceso está accediendo a lhs o rhs de la expresión de asignación. Y en general, debe protegerse contra el siguiente escenario:

// Thread 1
x = std::move(y);

// Thread 2
y = std::move(x);

Aquí está el operador de asignación de movimiento que protege correctamente el escenario anterior:

    A& operator=(A&& a)
    {
        if (this != &a)
        {
            WriteLock lhs_lk(mut_, std::defer_lock);
            WriteLock rhs_lk(a.mut_, std::defer_lock);
            std::lock(lhs_lk, rhs_lk);
            field1_ = std::move(a.field1_);
            field2_ = std::move(a.field2_);
        }
        return *this;
    }

Tenga en cuenta que uno debe usar std::lock(m1, m2) para bloquear los dos mutexes, en lugar de bloquearlos uno tras otro. Si los bloquea uno tras otro, cuando dos hilos asignan dos objetos en el orden opuesto como se muestra arriba, puede obtener un punto muerto. El punto de std::lock es evitar ese punto muerto.

Copiar constructor

No preguntaste sobre los miembros de la copia, pero podríamos hablar sobre ellos ahora (si no tú, alguien los necesitará).

    A(const A& a)
    {
        ReadLock  rhs_lk(a.mut_);
        field1_ = a.field1_;
        field2_ = a.field2_;
    }

El constructor de copia se parece mucho al constructor de movimiento, excepto que se ReadLock alias ReadLock lugar del WriteLock . Actualmente estos dos alias std::unique_lock<std::mutex> y, por lo tanto, realmente no hacen ninguna diferencia.

Pero en C ++ 14, tendrá la opción de decir esto:

    using MutexType = std::shared_timed_mutex;
    using ReadLock  = std::shared_lock<MutexType>;
    using WriteLock = std::unique_lock<MutexType>;

Esto puede ser una optimización, pero no definitivamente. Tendrá que medir para determinar si es así. Pero con este cambio, uno puede copiar construcciones de los mismos rhs en múltiples hilos simultáneamente. La solución C ++ 11 lo obliga a hacer que dichos subprocesos sean secuenciales, aunque el rhs no se esté modificando.

Asignación de copia

Para completar, aquí está el operador de asignación de copias, que debería explicarse bastante por sí mismo después de leer sobre todo lo demás:

    A& operator=(const A& a)
    {
        if (this != &a)
        {
            WriteLock lhs_lk(mut_, std::defer_lock);
            ReadLock  rhs_lk(a.mut_, std::defer_lock);
            std::lock(lhs_lk, rhs_lk);
            field1_ = a.field1_;
            field2_ = a.field2_;
        }
        return *this;
    }

Y etc.

Cualquier otro miembro o función gratuita que acceda al estado de A también necesitará protección si espera que varios hilos puedan llamarlos a la vez. Por ejemplo, aquí está el swap :

    friend void swap(A& x, A& y)
    {
        if (&x != &y)
        {
            WriteLock lhs_lk(x.mut_, std::defer_lock);
            WriteLock rhs_lk(y.mut_, std::defer_lock);
            std::lock(lhs_lk, rhs_lk);
            using std::swap;
            swap(x.field1_, y.field1_);
            swap(x.field2_, y.field2_);
        }
    }

Tenga en cuenta que si solo depende de que std::swap haga el trabajo, el bloqueo tendrá una granularidad incorrecta, bloqueando y desbloqueando entre los tres movimientos que std::swap realizaría internamente.

De hecho, pensar en el swap puede darle una idea de la API que puede necesitar para proporcionar una A "segura para subprocesos", que en general será diferente de una API "no segura para subprocesos", debido a la "granularidad de bloqueo" problema.

También tenga en cuenta la necesidad de protegerse contra el "auto-intercambio". "auto-intercambio" debería ser un no-op. Sin la autocomprobación, uno recursivamente bloquearía el mismo mutex. Esto también podría resolverse sin la autocomprobación mediante std::recursive_mutex para MutexType .

Actualizar

En los comentarios a continuación, Yakk está bastante descontento por tener que construir cosas por defecto en la copia y mover los constructores (y tiene un punto). Si se siente lo suficientemente fuerte sobre este problema, tanto que está dispuesto a gastar memoria en él, puede evitarlo así:

• Agregue los tipos de bloqueo que necesite como miembros de datos. Estos miembros deben presentarse antes de los datos que están siendo protegidos:

  mutable MutexType mut_;
  ReadLock  read_lock_;
  WriteLock write_lock_;
  // ... other data members ...

• Y luego en los constructores (por ejemplo, el constructor de copia) haga esto:

  A(const A& a)
      : read_lock_(a.mut_)
      , field1_(a.field1_)
      , field2_(a.field2_)
  {
      read_lock_.unlock();
  }

Vaya, Yakk borró su comentario antes de que tuviera la oportunidad de completar esta actualización. Pero merece crédito por impulsar este problema y obtener una solución a esta respuesta.

Actualización 2

Y dyp surgió con esta buena sugerencia:

    A(const A& a)
        : A(a, ReadLock(a.mut_))
    {}
private:
    A(const A& a, ReadLock rhs_lk)
        : field1_(a.field1_)
        , field2_(a.field2_)
    {}

Answer 2

Dado que no parece ser una forma agradable, limpia y fácil de responder a esto: la solución de Anton creo que es correcta pero es definitivamente discutible, a menos que surja una mejor respuesta, recomendaría poner esa clase en el montón y cuidarla a través de un std::unique_ptr :

auto a = std::make_unique<A>();

Ahora es un tipo completamente móvil y cualquier persona que tenga un bloqueo en el mutex interno mientras ocurre un movimiento todavía es seguro, incluso si es discutible si es algo bueno hacer

Si necesita semántica de copia, simplemente use

auto a2 = std::make_shared<A>();

Answer 3

El uso de mutexes y la semántica de movimiento de C ++ es una excelente manera de transferir datos de manera segura y eficiente entre subprocesos.

Imagine un hilo ''productor'' que crea lotes de cadenas y las proporciona a (uno o más) consumidores. Esos lotes podrían estar representados por un objeto que contiene objetos (potencialmente grandes) std::vector<std::string> . Queremos absolutamente ''mover'' el estado interno de esos vectores a sus consumidores sin duplicación innecesaria.

Simplemente reconoce el mutex como parte del objeto, no como parte del estado del objeto. Es decir, no desea mover el mutex.

El bloqueo que necesita depende de su algoritmo o de la generalización de sus objetos y de la gama de usos que permita.

Si alguna vez se mueve de un objeto ''productor'' de estado compartido a un objeto ''consumidor'' de subprocesos locales, es posible que solo bloquee el objeto movido.

Si se trata de un diseño más general, deberá bloquear ambos. En tal caso, debe considerar el bloqueo muerto.

Si ese es un problema potencial, use std::lock() para adquirir bloqueos en ambos mutexes de una manera libre de punto muerto.

http://en.cppreference.com/w/cpp/thread/lock

Como nota final, debe asegurarse de comprender la semántica del movimiento. Recuerde que el objeto movido del objeto se deja en un estado válido pero desconocido. Es completamente posible que un hilo que no realiza el movimiento tenga una razón válida para intentar acceder al objeto movido desde el objeto cuando puede encontrar ese estado válido pero desconocido.

Una vez más, mi productor solo está golpeando las cuerdas y el consumidor está quitando toda la carga. En ese caso, cada vez que el productor intenta agregar al vector, puede encontrar el vector no vacío o vacío.

En resumen, si el acceso simultáneo potencial al objeto movido desde el objeto equivale a una escritura, es probable que esté bien. Si equivale a una lectura, piense por qué está bien leer un estado arbitrario.

Answer 4

En primer lugar, debe haber algo mal con su diseño si desea mover un objeto que contiene un mutex.

Pero si decide hacerlo de todos modos, debe crear un nuevo mutex en el constructor de movimientos, es decir, por ejemplo:

// movable
struct B{};

class A {
    B b;
    std::mutex m;
public:
    A(A&& a)
        : b(std::move(a.b))
        // m is default-initialized.
    {
    }
};

Esto es seguro para subprocesos, porque el constructor de movimiento puede asumir con seguridad que su argumento no se usa en ningún otro lugar, por lo que no es necesario bloquearlo.

Answer 5

Esta es una respuesta al revés. En lugar de incrustar "estos objetos deben sincronizarse" como una base del tipo, en su lugar, inyéctelo debajo de cualquier tipo.

Tratas con un objeto sincronizado de manera muy diferente. Un gran problema es que debe preocuparse por los puntos muertos (bloqueo de varios objetos). Básicamente, nunca debería ser su "versión predeterminada de un objeto": los objetos sincronizados son para objetos que estarán en contención, y su objetivo debe ser minimizar la contención entre hilos, no barrerla debajo de la alfombra.

Pero sincronizar objetos sigue siendo útil. En lugar de heredar de un sincronizador, podemos escribir una clase que envuelva un tipo arbitrario en sincronización. Los usuarios tienen que saltar algunos aros para realizar operaciones en el objeto ahora que está sincronizado, pero no están limitados a un conjunto limitado de operaciones codificadas a mano en el objeto. Pueden componer múltiples operaciones en el objeto en uno, o tener una operación en múltiples objetos.

Aquí hay un contenedor sincronizado alrededor de un tipo arbitrario T :

template<class T>
struct synchronized {
  template<class F>
  auto read(F&& f) const&->std::result_of_t<F(T const&)> {
    return access(std::forward<F>(f), *this);
  }
  template<class F>
  auto read(F&& f) &&->std::result_of_t<F(T&&)> {
    return access(std::forward<F>(f), std::move(*this));
  }
  template<class F>
  auto write(F&& f)->std::result_of_t<F(T&)> {
    return access(std::forward<F>(f), *this);
  }
  // uses `const` ness of Syncs to determine access:
  template<class F, class... Syncs>
  friend auto access( F&& f, Syncs&&... syncs )->
  std::result_of_t< F(decltype(std::forward<Syncs>(syncs).t)...) >
  {
    return access2( std::index_sequence_for<Syncs...>{}, std::forward<F>(f), std::forward<Syncs>(syncs)... );
  };
  synchronized(synchronized const& o):t(o.read([](T const&o){return o;})){}
  synchronized(synchronized && o):t(std::move(o).read([](T&&o){return std::move(o);})){}  
  // special member functions:
  synchronized( T & o ):t(o) {}
  synchronized( T const& o ):t(o) {}
  synchronized( T && o ):t(std::move(o)) {}
  synchronized( T const&& o ):t(std::move(o)) {}
  synchronized& operator=(T const& o) {
    write([&](T& t){
      t=o;
    });
    return *this;
  }
  synchronized& operator=(T && o) {
    write([&](T& t){
      t=std::move(o);
    });
    return *this;
  }
private:
  template<class X, class S>
  static auto smart_lock(S const& s) {
    return std::shared_lock< std::shared_timed_mutex >(s.m, X{});
  }
  template<class X, class S>
  static auto smart_lock(S& s) {
    return std::unique_lock< std::shared_timed_mutex >(s.m, X{});
  }
  template<class L>
  static void lock(L& lockable) {
      lockable.lock();
  }
  template<class...Ls>
  static void lock(Ls&... lockable) {
      std::lock( lockable... );
  }
  template<size_t...Is, class F, class...Syncs>
  friend auto access2( std::index_sequence<Is...>, F&&f, Syncs&&...syncs)->
  std::result_of_t< F(decltype(std::forward<Syncs>(syncs).t)...) >
  {
    auto locks = std::make_tuple( smart_lock<std::defer_lock_t>(syncs)... );
    lock( std::get<Is>(locks)... );
    return std::forward<F>(f)(std::forward<Syncs>(syncs).t ...);
  }

  mutable std::shared_timed_mutex m;
  T t;
};
template<class T>
synchronized< T > sync( T&& t ) {
  return {std::forward<T>(t)};
}

Características de C ++ 14 y C ++ 1z incluidas.

Esto supone que las operaciones const son seguras para múltiples lectores (que es lo que suponen los contenedores estándar).

El uso se parece a:

synchronized<int> x = 7;
x.read([&](auto&& v){
  std::cout << v << ''/n'';
});

para un int con acceso sincronizado.

Aconsejaría no haberse synchronized(synchronized const&) . Rara vez se necesita.

Si necesita synchronized(synchronized const&) , estaría tentado a reemplazar T t; con std::aligned_storage , lo que permite la construcción de colocación manual y la destrucción manual. Eso permite una gestión adecuada de por vida.

Salvo eso, podríamos copiar la fuente T y luego leerla:

synchronized(synchronized const& o):
  t(o.read(
    [](T const&o){return o;})
  )
{}
synchronized(synchronized && o):
  t(std::move(o).read(
    [](T&&o){return std::move(o);})
  )
{}

para la asignación:

synchronized& operator=(synchronized const& o) {
  access([](T& lhs, T const& rhs){
    lhs = rhs;
  }, *this, o);
  return *this;
}
synchronized& operator=(synchronized && o) {
  access([](T& lhs, T&& rhs){
    lhs = std::move(rhs);
  }, *this, std::move(o));
  return *this;
}
friend void swap(synchronized& lhs, synchronized& rhs) {
  access([](T& lhs, T& rhs){
    using std::swap;
    swap(lhs, rhs);
  }, *this, o);
}

la colocación y las versiones de almacenamiento alineadas son un poco más desordenadas. La mayoría del acceso a t sería reemplazado por una función miembro T&t() y T const&t()const , excepto en la construcción donde tendrías que saltar a través de algunos aros.

Al hacer un contenedor synchronized en lugar de parte de la clase, todo lo que tenemos que asegurar es que la clase respete internamente a const como un lector múltiple y lo escriba de una sola hebra.

En los raros casos que necesitamos una instancia sincronizada, saltamos a través de aros como el anterior.

Disculpas por cualquier error tipográfico en lo anterior. Probablemente hay algunos.

Un beneficio adicional de lo anterior es que las operaciones arbitrarias n-arias en objetos synchronized (del mismo tipo) funcionan juntas, sin tener que codificarlas de antemano. Agregue una declaración de amigo y los objetos synchronized n-ary de varios tipos podrían funcionar juntos. Podría tener que mover el access de ser un amigo en línea para lidiar con los conflictos de sobrecarga en ese caso.

ejemplo en vivo