생산자 - 소비자 시나리오에서 부스트 조건 변수를 어떻게 사용할 수 있습니까?

Question

편집 : 아래

버퍼의 장치에서 데이터를 스트리밍하는 스레드가 하나 있습니다. 또한, 그 데이터에 대한 처리를 N 개의 스레드가 있습니다. 설치 과정에서 스 트리머 스레드가 장치에서 데이터를 가져오고 새 데이터를 페치하거나 제한 시간에 도달하기 전에 N 스레드가 처리를 마칠 때까지 기다려야합니다. N 스레드는 처리를 계속하기 전에 새 데이터를 가져올 때까지 대기해야합니다. 나는이 프레임 워크가 N 스레드가 버퍼에서 프로세싱을 반복하는 것을 원하지 않는다면 모든 버퍼가 건너 뛰지 않고 처리되기를 원한다면 작동해야한다고 생각한다.

주의 깊게 읽은 후에 조건 변수가 필요한 것임을 알았습니다. 나는 자습서 및 다른 스택 오버플로 질문을 따랐다, 이것은 내가 무엇을 가지고 :

전역 변수 :

boost::condition_variable cond; 
boost::mutex mut; 

멤버 변수 :

이

std::vector<double> buffer 
std::vector<bool> data_ready  // Size equal to number of threads 

데이터 수신기 루프 (1 개 실이 실행을) :

while (!gotExitSignal()) 
{ 
    { 
     boost::unique_lock<boost::mutex> ll(mut); 
     while(any(data_ready)) 
      cond.wait(ll); 
    } 

    receive_data(buffer); 

    { 
     boost::lock_guard<boost::mutex> ll(mut); 
     set_true(data_ready); 
    } 

    cond.notify_all(); 
} 

데이터 처리 루프 (N 개의 스레드에서 실행)

while (!gotExitSignal()) 
{ 
    { 
     boost::unique_lock<boost::mutex> ll(mut); 
     while(!data_ready[thread_id]) 
      cond.wait(ll); 
    } 

    process_data(buffer); 

    { 
     boost::lock_guard<boost::mutex> ll(mut); 
     data_ready[thread_id] = false; 
    } 
    cond.notify_all(); 
} 

이 두 루프는 같은 클래스의 자체 멤버 함수에 있습니다. 가변 버퍼는 멤버 변수이므로 스레드간에 공유 될 수 있습니다.

수신 스레드가 먼저 실행됩니다. data_ready 변수는 크기가 N 인 bool의 벡터입니다. data_ready [i]는 데이터를 처리 할 준비가되면 true이고, 스레드가 이미 데이터를 처리 한 경우 false입니다. any (data_ready) 함수는 data_ready의 요소 중 하나가 true이면 true를 출력하고 그렇지 않으면 false를 출력합니다. set_true (data_ready) 함수는 data_ready의 모든 요소를 ​​true로 설정합니다. 수신 쓰레드는 처리중인 쓰레드가 여전히 처리 중인지 확인합니다. 그렇지 않다면, 데이터를 가져오고, data_ready 플래그를 설정하고, 스레드에 알리고, 처리가 완료 될 때까지 시작 부분에서 멈출 루프를 계속할 것입니다. 처리 스레드는 각각의 data_ready 플래그가 true인지 확인합니다. 일단 그것이 사실이면, 처리 스레드는 일부 계산을 수행하고, 각각의 data_ready 플래그를 0으로 설정하고, 루프를 계속할 것이다.

처리 스레드가 하나 뿐인 경우 프로그램이 정상적으로 실행됩니다. 더 많은 스레드를 추가하고 나면 처리 결과가 불필요한 문제가 발생합니다. 또한 처리 스레드의 순서는 어떤 이유로 중요합니다. 다시 말하면, 내가 시작한 LAST 스레드는 올바른 데이터를 출력하지만, 입력 매개 변수가 처리에 관계없이 (유효한 매개 변수를 가정 할 때) 이전 스레드는 가비지를 출력합니다. 문제가 내 스레딩 코드로 인한 것인지 또는 장치 또는 데이터 처리 설정에 문제가 있는지 알 수 없습니다. I는 처리에 couts을 사용하고 단계를 수신하고, 정상적으로 N 프로세싱 스레드, I 출력 표시보십시오

receive data 
process 1 
process 2 
... 
process N 
receive data 
process 1 
process 2 
... 

는 조건 변수 정확한의 용도는? 무엇이 문제 일 수 있습니까?

편집 : (N 스레드가이 실행)

while (!gotExitSignal()) 
{ 
    // boost::unique_lock<boost::mutex> ll(mut); 
    boost::mutex::scoped_lock ll(mut); 
    cond.wait(ll); 

    process_data(buffer); 

    data_ready[thread_id] = false; 
} 

while (!gotExitSignal()) 
{ 
    if(!any(data_ready)) 
    { 
     receive_data(buffer); 
     boost::lock_guard<boost::mutex> ll(mut); 
     set_true(data_ready); 
     cond.notify_all(); 
    }  
} 

데이터 처리 루프 : 나는 포크의 제안을 따라에 코드를 변경 :

이

데이터 수신기 루프 (1 개 실이 실행됩니다)

다소 개선되었습니다. 올바른 잠금 장치를 사용하고 있습니까?

Answer 1

나는 당신의 모든 이야기를 읽지 못했지만, 코드를 빨리 보면 내가 틀린 조건을 사용하는 것을 볼 수 있습니다. 조건은 상태와 비슷합니다. 대기 조건에서 스레드를 설정하면 CPU가 멀리 떨어져 버립니다. 따라서 스레드는 다른 프로세스/스레드가이를 알릴 때까지 실행을 중지합니다.

코드에는 while 루프가 있으며 기다릴 데이터를 확인할 때마다 기다립니다. 그것이 틀렸을 때, 대신에해야합니다. 그러나 다시는 거기에 있으면 안됩니다. 데이터 확인은 다른 곳에서해야합니다. 그리고 작업자 스레드는 작업을 마친 후에 대기 상태에 있어야합니다.

작업자 스레드는 소비자입니다. 그리고 생산자는 데이터를 전달하는 것입니다. 데이터가 있는지 여부를 스레드가 확인하고 작업자에게 알리는 것이 더 나은 구조라고 생각합니다.

의사 코드 : 또한 일부 스레드가 교착 상태를 피하기 위해 살아있을 필요가 있다는 사실을주의해야

//producer 
while (true) { 

    1. lock mutex 
    2. is data available 
    3. unlock mutex 

    if (dataAvailableVariable) { 
     4. notify a worker 
     5. set waiting condition 
    } 
} 


//consumer 
while (true) { 
    1. lock mutex 
    2. do some work 
    3. unlock mutex 
    4. notify producer that work is done 
    5. set wait condition 
} 

는 조건을 기다리는 모든 스레드를 의미한다.

조금 도움이 되었기를 바랍니다.