¿Cómo debo lidiar con las exclusiones mutuas en tipos móviles en C ++?

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

¿Cómo puedo hacer que este tipo se Amueva de manera segura para subprocesos?

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:
    ...

He puesto algunos alias de tipo bastante sugerentes que realmente no aprovecharemos en C ++ 11, pero que serán mucho más útiles en C ++ 14. Tenga paciencia, llegaremos allí.

Tu pregunta se reduce a:

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

Empezaremos con el constructor de movimientos.

Mover constructor

Tenga en cuenta que el miembro mutexse ha creado mutable. Estrictamente hablando, esto no es necesario para los miembros de movimiento, pero supongo que también quiere copiar miembros. Si ese no es el caso, no es necesario realizar el mutex mutable.

Al construir A, no es necesario bloquear this->mut_. Pero necesita bloquear el mut_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 tuvimos que construir por defecto los miembros de thisprimero, y luego asignarles valores solo después de que a.mut_esté bloqueado.

Mover asignación

El operador de asignación de movimiento es sustancialmente más complicado porque no sabe si algún otro hilo 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 se debe usar std::lock(m1, m2)para bloquear los dos mutex, en lugar de simplemente bloquearlos uno tras otro. Si los bloquea uno tras otro, cuando dos subprocesos asignan dos objetos en orden opuesto como se muestra arriba, puede obtener un punto muerto. El punto std::lockes evitar ese punto muerto.

Copiar constructor

No preguntaste sobre los miembros de la copia, pero también podríamos hablar de ellos ahora (si no eres 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 ReadLockse usa el alias en 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 lo es. Pero con este cambio, se puede copiar la construcción de los mismos rhs en múltiples hilos simultáneamente. La solución C ++ 11 te obliga a hacer que dichos subprocesos sean secuenciales, aunque no se modifiquen los rhs.

Copiar asignación

Para completar, aquí está el operador de asignación de copia, que debería ser bastante autoexplicativo 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 Aal estado también deberá estar protegido si espera que varios subprocesos puedan llamarlos a la vez. Por ejemplo, aquí tienes 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 std::swaphacer el trabajo, el bloqueo estará en la granularidad incorrecta, bloqueando y desbloqueando entre los tres movimientos que std::swapse realizarían internamente.

De hecho, pensar en swappuede darte información sobre la API que podrías necesitar para proporcionar una API "segura para subprocesos" A, que en general será diferente de una API "no segura para subprocesos", debido al problema de "granularidad de bloqueo".

También tenga en cuenta la necesidad de protegerse contra el "autointercambio". el "autointercambio" debería ser una operación prohibida. Sin la autocomprobación, se bloquearía recursivamente el mismo mutex. Esto también podría resolverse sin la autocomprobación utilizando std::recursive_mutexfor MutexType.

Actualizar

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

• Agregue los tipos de bloqueo que necesite como miembros de datos. Estos miembros deben anteponerse a los datos que se están protegiendo:

  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 él merece crédito por impulsar este problema y obtener una solución a esta respuesta.

Actualización 2

Y a dyp se le ocurrió 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_)
    {}

Dado que no parece haber una manera agradable, limpia y fácil de responder a esto, la solución de Anton creo que es correcta, pero definitivamente es 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 cualquiera que tenga un bloqueo en el mutex interno mientras ocurre un movimiento aún está seguro, incluso si es discutible si esto es algo bueno que hacer

Si necesita copiar la semántica, simplemente use

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

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

Tratas con un objeto sincronizado de manera muy diferente. Un gran problema es que debe preocuparse por los interbloqueos (bloquear varios objetos). Básicamente, tampoco debería ser nunca su "versión predeterminada de un objeto": los objetos sincronizados son para objetos que estarán en disputa, y su objetivo debería ser minimizar la contención entre hilos, no barrerlos 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 pasar por 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 varias operaciones sobre el objeto en una o tener una operación sobre varios objetos.

Aquí hay una envoltura sincronizada 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 constoperaciones son seguras para múltiples lectores (que es lo stdque suponen los contenedores).

El uso se parece a:

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

para un intacceso sincronizado.

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

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

Salvo eso, podríamos copiar la fuente Ty 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 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 ubicación y las versiones de almacenamiento alineadas son un poco más desordenadas. La mayor parte del acceso a tsería reemplazado por una función miembro T&t()y T const&t()const, excepto en la construcción, donde tendría que pasar por algunos obstáculos.

Al hacer synchronizedun contenedor en lugar de ser parte de la clase, todo lo que tenemos que asegurarnos es que la clase respete internamente constcomo un lector múltiple y lo escriba de una manera de un solo subproceso.

En los casos raros 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 synchronizedobjetos (del mismo tipo) funcionan juntas, sin tener que codificarlas de antemano. Agregue una declaración de amigo y los synchronizedobjetos n-arios de varios tipos podrían funcionar juntos. Podría tener que dejar accessde ser un amigo en línea para lidiar con los conflictos de sobrecarga en ese caso.

ejemplo vivo

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

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

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 del rango de usos que permita.

Si solamente nunca se mueve de un objeto compartido estado 'productor' a un objeto local de subprocesos 'consumir' que podría estar OK para bloquear la única trasladó de objeto.

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, utilícelo std::lock()para adquirir bloqueos en ambos mutex de una manera sin interbloqueo.

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

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

Una vez más, mi productor solo está golpeando cuerdas y el consumidor se 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 concurrente potencial al objeto movido equivale a una escritura, es probable que esté bien. Si equivale a una lectura, piense por qué está bien leer un estado arbitrario.

En primer lugar, debe haber algún problema 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 movimiento, 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 bloquear el argumento.