通过线程安全容器传递非线程安全对象

Question

我有一个线程安全的对象队列，它旨在模拟在线程链之间移动的工作管道。在某些情况下，我想将非线程安全对象（例如，std::vectors 或其他 STL 容器）作为这些工作项的一部分传递。

现在，如果您在线程之间有一个共享对象，那么在确保对象一致性方面存在一个明显的加载/存储顺序问题。由于线程安全队列确保只有一个线程拥有对象的所有权，因此多个线程不可能同时尝试修改或读取对象。我看到的唯一可能的问题是确保先前发出的负载中的内存一致性/stores 由以前的所有者线程存储在对象上。

队列通过在队列操作上创建lock_guard<...> 来确保线程安全。由于内存隔离和同步将由lock_guard 负责，难道不能保证在线程之间移动的对象的内存一致性吗？

我的一部分想要确保我只在线程之间传递线程安全的对象，但我觉得这种情况应该没有问题。这是真的吗？

Answer 1

队列通过在队列操作上创建 lock_guard<...> 来确保线程安全。由于内存隔离和同步将由 lock_guard 负责，所以在线程之间移动的对象的内存一致性是否得到保证？

是的。

我的一部分想要确保我只在线程之间传递线程安全的对象，但我觉得这种情况应该没有问题。这是真的吗？

是的。

这就是为什么volatile 不适合作为C++ 中并发数据访问的设备的原因；它不能解决竞争条件，一旦你将并发设备（例如互斥锁）带入解决这个问题，它们也会处理内存一致性问题，所以volatile 根本没有什么可做的。

Answer 2

我使用稍微不同的方法来阻止线程我不知道这是否会有所帮助，但我不介意分享我必须提供的不同意见或方法。

BlockThread.h

#ifndef BLOCK_THREAD_H
#define BLOCK_THREAD_H

namespace vmk {

class BlockThread sealed {
private:
    CRITICAL_SECTION* m_pCriticalSection;

public:
    explicit BlockThread( CRITICAL_SECTION& criticalSection );
    ~BlockThread();

private:
    BlockThread( const BlockThread& c ); // Not Implemented
    BlockThread& operator=( const BlockThread& c ); // Not Implemented
    // void swap( BlockThread& other ) throw();

}; // BlockThread

} // namespace vmk

#endif // BLOCK_THREAD_H

BlockThread.cpp

#include "stdafx.h"
#include "BlockThread.h"

namespace vmk {

// ----------------------------------------------------------------------------
// BlockThread()
BlockThread::BlockThread( CRITICAL_SECTION& criticalSection ) {
    m_pCriticalSection = &criticalSection;
    EnterCriticalSection( m_pCriticalSection );
} // BlockThread

// ----------------------------------------------------------------------------
// ~BlockThread()
BlockThread::~BlockThread() {
    LeaveCriticalSection( m_pCriticalSection );
} // ~BlockThread    

} // namespace vmk

VolatileLocker.h

#ifndef VOLATILE_LOCKER_H
#define VOLATILE_LOCKER_H

namespace vmk {

template <typename T>
class VolatileLocker {
private:
    T*                  m_pObject;
    CRITICAL_SECTION*   m_pCriticalSection;

public:
    VolatileLocker( volatile T& objectToLock, CRITICAL_SECTION& criticalSection );
    ~VolatileLocker();

    T* operator->();

private:
    VolatileLocker( const VolatileLocker& c ); // Not Implemented
    VolatileLocker& operator=( const VolatileLocker& c ); // Not Implemented

}; // VolatileLocker

#include "VolatileLocker.inl"

} // namespace vmk

#endif // VOLATILE_LOCKER_H

// reference: http://drdobbs.com/cpp/184403766

VolatileLocker.inl

// ----------------------------------------------------------------------------
// VolatileLocker()
// Locks A Volatile Variable So That It Can Be Used Across Multiple Threads Safely
template<typename T>
VolatileLocker<T>::VolatileLocker( volatile T& objectToLock, CRITICAL_SECTION& criticalSection ) :
m_pObject( const_cast<T*>( &objectToLock ) ),
m_pCriticalSection( &criticalSection ) {
    EnterCriticalSection( m_pCriticalSection );
} // VolatileLocker

// ----------------------------------------------------------------------------
// ~VolatileLocker()
template<typename T>
VolatileLocker<T>::~VolatileLocker() {
    LeaveCriticalSection( m_pCriticalSection );
} // ~VolatileLocker

// ----------------------------------------------------------------------------
// operator->()
// Allow The Locked Object To Be Used Like A Pointer
template <typename T>
T* VolatileLocker<T>::operator->() {
    return m_pObject;
} // operator->

这是一个使用 BlockThread 的类对象 - 该类依赖于此处未显示的其他类，我将仅包括该类中使用 BlockThread 对象的部分。

AudioManager.cpp

#include "stdafx.h"
#include "AudioManager.h"

#include "AudioBuffer.h"
#include "AudioSource.h"
#include "BlockThread.h"
#include "Logger.h"

namespace vmk {

static AudioManager*        s_pAudioManager = nullptr;
static CRITICAL_SECTION     s_csChangeSources;

// ----------------------------------------------------------------------------
// AudioManager()
AudioManager::AudioManager() :
Singleton( Singleton::TYPE_AUDIO_MANAGER ) {
    InitializeCriticalSection( &s_csChangeSources );

    if ( !alutInit( NULL, NULL ) ) {
        std::ostringstream strStream;
        strStream << __FUNCTION__ << " ALUT error: " << alutGetErrorString( alutGetError() );
        throw ExceptionHandler( strStream );
    }

    alGetError(); // Clear Errors

    alListenerf( AL_GAIN, 1.0f ); // Master Volume
    alListener3f( AL_POSITION, 0.0f, 0.0f, 0.0f );
    alListener3f( AL_VELOCITY, 0.0f, 0.0f, 0.0f );

    float f6Orient[] = { 0.0f, 0.0f, -1.0f,   // Forward(X, Y, Z)
                         0.0f, 1.0f,  0.0f }; // Up(X,Y,Z)

    alListenerfv( AL_ORIENTATION, f6Orient );
    if ( alGetError() != AL_NO_ERROR ) {
        std::ostringstream strStream;
        strStream << __FUNCTION__ << " Failed to initialize Listener";
        throw ExceptionHandler( strStream );
    }

    s_pAudioManager = this;
} // AudioManager

// ----------------------------------------------------------------------------
// ~AudioManager()
AudioManager::~AudioManager() {
    s_pAudioManager = nullptr;

    m_mAudioSources.clear();
    m_lAudioBuffers.clear();

    alutExit();

    DeleteCriticalSection( &s_csChangeSources );
} // ~AudioManager

// ----------------------------------------------------------------------------
// get()
AudioManager* const AudioManager::get() {
    if ( nullptr == s_pAudioManager ) {
        throw ExceptionHandler( __FUNCTION__ + std::string( " failed, AudioManager has not been constructed yet" ) );
    }
    return s_pAudioManager;
} // get    

// ----------------------------------------------------------------------------
// Create A Sound Source Using The Passed In Filename. If The File Is Not
// Already Loaded Into A Memory Buffer, Then A New Buffer Is Created.
void AudioManager::createSource( SoundSource eSoundSource, const std::string& strFilename, bool bAttachToListener ) {
    BlockThread blockThread( s_csChangeSources );

    if ( !isAvailable( eSoundSource ) ) {
        return;
    }

    std::shared_ptr<AudioBuffer> pAudioBuffer = nullptr;

    // Check If This File Has Already Been Loaded Into A Buffer
    for ( ListAudioBuffers::iterator itBuffer = m_lAudioBuffers.begin(); itBuffer != m_lAudioBuffers.end(); ++itBuffer ) {
        if ( (*itBuffer)->isThisFile( strFilename ) ) {
            // The Requested File Is Already Loaded Into Memory
            pAudioBuffer = (*itBuffer);
            break;
        }
    }

    try {
        if ( nullptr == pAudioBuffer ) {
            // Need To Load The Desired File Into Memory
            pAudioBuffer.reset( new AudioBuffer( strFilename ) );

            // Store The Buffer
            m_lAudioBuffers.push_back( pAudioBuffer );
        }

        // Create New Source Attached To The Desired Audio Buffer
        m_mAudioSources[eSoundSource] = std::shared_ptr<AudioSource>( new AudioSource( eSoundSource, pAudioBuffer, bAttachToListener ) );

    } catch ( ... ) {
        std::ostringstream strStream;
        strStream << __FUNCTION__ << " failed for SoundSource(" << eSoundSource << ")";
        Logger::log( strStream, Logger::TYPE_ERROR );
    }
} // createSource

// ----------------------------------------------------------------------------
// Removes Source From Map And If Buffer Is No Longer Needed,
// It Will Also Be Deleted
void AudioManager::deleteSource( SoundSource eSoundSource ) {
    BlockThread blockThread( s_csChangeSources );

    MapAudioSources::iterator it = m_mAudioSources.find( eSoundSource );
    if ( it == m_mAudioSources.end() ) {
        std::ostringstream strStream;
        strStream << __FUNCTION__ << " could not find SoundSource(" << eSoundSource << ")";
        Logger::log( strStream, Logger::TYPE_ERROR );
        return; // Nothing To Delete
    }

    // Get bufferId And Delete Source
    unsigned uBufferId = it->second->getBufferId();
    m_mAudioSources.erase( it );

    // Find Buffer In List
    if ( uBufferId != INVALID_UNSIGNED ) {
        for ( ListAudioBuffers::iterator itBuffer = m_lAudioBuffers.begin(); itBuffer != m_lAudioBuffers.end(); ++itBuffer ) {
            if ( (*itBuffer)->getId() == uBufferId ) {
                if ( (*itBuffer)->getNumberSourcesAttached() < 1 ) {
                    // Buffer No Longer Needed
                    m_lAudioBuffers.erase( itBuffer );
                    return;
                } // If Buffer Not Needed
            } // If Found Buffer
        } // For All Buffers
    } // If Buffer Is Loaded
} // deleteSource

// ----------------------------------------------------------------------------
// getAudioObject()
AudioObject* AudioManager::getAudioObject( SoundSource eSoundSource ) const {
    BlockThread blockThread( s_csChangeSources );

    MapAudioSources::const_iterator it = m_mAudioSources.find( eSoundSource );
    if ( it != m_mAudioSources.cend() ) {
        return it->second.get();
    }

    std::ostringstream strStream;
    strStream << __FUNCTION__ << " SoundSource(" << eSoundSource << ") has not been found";
    Logger::log( strStream, Logger::TYPE_ERROR );
    return nullptr;
} // getAudioObject


} // namespace vmk

VolatileLocker 的使用如下： 注意：我还有一个 OpenglThread 类对象，此处未显示，以允许 OpenGL 使用多个线程。

Game.cpp

#include "Game.h"
#include "OpenglThread.h"
#include "VolatileLocker.h"

... other class object includes

namespace vmk {

static CRITICAL_SECTION s_criticalSection; 

// ----------------------------------------------------------------------------
// Game()
Game::Game() :
Engine( glm::uvec2( 3, 3 ) ),
m_maxSpeechArea( 250, INVALID_UNSIGNED ),
m_eLastSpeechSound( SS_INTRO_SHOT ),
m_eSoundSourceToPlay( SS_INTRO_SHOT ) {
    InitializeCriticalSection( &s_criticalSection );

    const glm::uvec2 gamePixelSize = m_pSettings->getGameSize();

    // (0,0) In Top Left Corner Not Bottom Left Which Is The Default
    m_m4Projection = glm::ortho( 0.0f, static_cast<float>( gamePixelSize.x ), static_cast<float>( gamePixelSize.y ), 0.0f, -10.0f, 10.0f );

    // Set Background Color
    glClearColor( 22.0f / 255.0f, 22.0f / 255.0f, 22.0f / 255.0f, 1.0f );

    // Turn Transparencies On
    glEnable( GL_BLEND );

    // Initialize Shaders
    std::string strVertexShader( "Shaders/gui.vert" );
    std::string strFragmentShader( "Shaders/gui.frag" );

    ShaderProgramSettings shaderProgramSettings( P_MAIN, strVertexShader, strFragmentShader );
    shaderProgramSettings.addVariable( ShaderAttribute( A_POSITION,             AT_FLOAT_VEC2 ), "inPosition" );
    shaderProgramSettings.addVariable( ShaderAttribute( A_COLOR,                AT_FLOAT_VEC4 ), "inColor" );
    shaderProgramSettings.addVariable( ShaderAttribute( A_TEXTURE_COORD0,       AT_FLOAT_VEC2 ), "inTextureCoord0" );

    shaderProgramSettings.addVariable( ShaderUniform( U_MVP_MATRIX,             UT_FLOAT_MAT4 ), "modelViewProjectionMatrix" );
    shaderProgramSettings.addVariable( ShaderUniform( U_TEXTURE0_SAMPLER_2D,    UT_SAMPLER_2D ), "texture0Sampler2d" ); 
    shaderProgramSettings.addVariable( ShaderUniform( U_USING_TEXTURE,          UT_BOOL ),       "usingTexture" );  
    shaderProgramSettings.addVariable( ShaderUniform( U_ALPHA,                  UT_FLOAT ),      "inAlpha" );

    m_pShaderManager->create( shaderProgramSettings );
    m_pShaderManager->enable( P_MAIN );

    m_pBatchManager.reset( new BatchManager( 10, 10000 ) );

    // Must Be Called Before Any GuiElements Are Loaded
    GuiElement::initialize();   

    // Load Game Logo - Title Screen
    m_pTitleScreen = new GuiScreen( std::string( "TitleScreen" ) );

    TextureFileReader titleTextureFileReader( "Assets/images/titleScreen.png" );
    m_titleTextureInfo =  titleTextureFileReader.getOrCreateTextureInfo( TextureInfo::FILTER_NONE, false, false );

    // Start Worker Thread
    _beginthread( loadAssets, 0, this );

    // Game Logo
    GuiLayoutAbsolute* pTitleCoverLayout = new GuiLayoutAbsolute( glm::ivec2(), Gui::LEFT, m_pSettings->getGameSize(), "title cover" );
    pTitleCoverLayout->setColor( glm::vec4( 0.0862745, 0.0862745, 0.0862745, 1.0 ) );
    m_pTitleScreen->addChild( pTitleCoverLayout );

    m_pTitleLayout = new GuiLayoutAbsolute( glm::ivec2( 0, 200 ), Gui::LEFT, glm::uvec2( 955, 400 ), "title" );
    m_pTitleLayout->setBackgroundImage( m_titleTextureInfo, glm::uvec2( 0, 359 ), glm::uvec2( 955, 400 ) );
    m_pTitleLayout->changePriority( 1 );
    m_pTitleScreen->addChild( m_pTitleLayout );

    // Flying Bullet
    m_pFlyingBulletLayout = new GuiLayoutAbsolute( glm::ivec2( 40, -100 ), "flying bullet" );
    m_pTitleLayout->addChild( m_pFlyingBulletLayout );

    // Intro Sound Effect
    m_pAudioManager->createSource( SS_INTRO_SHOT, "Assets/audio/introShot.ogg" );
    m_pAudioManager->play( SS_INTRO_SHOT );

    // Set Timer For How Long Title Screen Should Be Visible
    m_pAnimationManager->addFunction( 5.0, Animation::LINEAR, splashScreenUpdate, this, splashScreenDone, this );

    /*
    int debugLogging = m_pSettings->getDebugLogging() | Settings::DEBUG_RENDER;
    m_pSettings->setDebugLogging( debugLogging );
    */

} // Game

// ----------------------------------------------------------------------------
// ~Game()
Game::~Game() { 
    DeleteCriticalSection( &s_criticalSection );
} // ~Game


// ----------------------------------------------------------------------------
// splashScreenDone()
// Defined Outside Of Game But Is Declared As A Friend Function To Game And It Utilizes The VolatileLocker
void splashScreenDone( void* pParameter ) {
    Game* pGame = reinterpret_cast<Game*>( pParameter );
    if ( nullptr == pGame ) {
        throw ExceptionHandler( __FUNCTION__ + std::string( " Invalid pParameter passed in" ) );
    }

    VolatileLocker<GameState>( pGame->m_gameState, s_criticalSection )->timerDone();

    if ( VolatileLocker<GameState>( pGame->m_gameState, s_criticalSection )->is( GameState::PLAYING ) ) {
        pGame->m_pAudioManager->play( SS_CHOOSE_LETTER );
    }

} // splashScreenDone

// ----------------------------------------------------------------------------
// keyboardInput()
// A Member Function Of Game That Utilizes The VolatileLocker
void Game::keyboardInput( unsigned vkCode, bool isPressed ) {
    if ( isPressed || VolatileLocker<GameState>( m_gameState, s_criticalSection )->isSplashScreen() ) {
        // Wait For Splash Screen To Be Finished
        // Only React To Key Release Events
        return;
    }

    static unsigned lastKey = 0;

    if ( (VK_ESCAPE == lastKey && VK_ESCAPE == vkCode) ||
        (VK_ESCAPE == vkCode && m_bGameOver) ) {
        // TODO: Show Credits

        quitGame( nullptr, nullptr );
        return;
    }
    lastKey = vkCode;

    if ( m_bGameOver ) {
        if ( VK_SPACE == vkCode ) {
            restart();
        }
        return;
    }

    if ( VK_ESCAPE == vkCode ) {
        updateSpeech( "To qui the game, press ESC again.", COLOR_YELLOW );
        speak( SS_QUIT_GAME );

    } else if ( vkCode >= VK_KEYA && vkCode <= VK_KEYZ ) {
        // Play Sound
        m_pAudioManager->play( SS_GUN_SHOT );

        updatePuzzle( static_cast<char>( vkCode ) );

        // Show Gun Fire & Start Timer To Reset Graphics Back To Normal
        showGunFire( true );
        m_pAnimationManager->addTimer( 0.2, resetGunGraphics, this );

    } else {
        updateSpeech( "Choose a letter." );
        speak( SS_CHOOSE_LETTER );
    }

} // keyboardInput

} // namespace vmk

现在我知道关于在线程池或队列中工作的 OP，但我认为出于安全原因能够锁定线程的整体概念也可以在这里展示。这也可以作为任何可能阅读此内容的人的指南。