.NET 작업 병렬 라이브러리

Question

TPL에 대한 문서 및 많은 자습서를 읽었지만 달성하고자하는 모델은 없습니다.

일부 알고리즘에는 항상 고정 된 수의 반복이있었습니다.

나는 항상 (가능한 한 많은으로) 스레드를 실행 필요 :

동안

• 메인 스레드에서 데이터를 얻을 (참)
• (별도의 스레드에서) 무거운 시간이 많이 걸리는 작업을 수행
• 갱신 메인 쓰레드 정보

는 Additionaly 나는 ABL 될 것이다 메커니즘이 필요 전자 시계 설정 (예 : 5 초). 5 초 후에 모든 작업을 잠시 중단 한 다음 다시 시작해야합니다.

Task.ContinueWith 같은 작업을 사용해야합니까? 그러나 이전 작업 실행 결과를 처리하지 않고 대신 MAIN 스레드에서 데이터 구조를 업데이트 한 다음 새 작업 반복의 입력 내용을 결정합니다. ...

얼마나 많은 작업을해야합니까? 효율성을 극대화하기 위해 만들어야합니까?

아니요. 좋은 RunEventCompleted 이벤트가 있기 때문에 BackgroundWorkers를 사용하고 있습니다. 메인 구조를 업데이트하고 시간 제약 조건을 확인한 다음 결국 완료된 BackgroundWorker에서 다시 StartAsync를 호출 할 수 있습니다. 그것은 훌륭하고 명확하지만, 아마도 매우 불편합니다. 멀티 프로세서, 멀티 코어 서버에서 매우 효율적으로 만들 필요가 있습니다.

하나의 문제는 계산이 항상 온라인 상태이며 중단되지 않는다는 것입니다. MAIN 구조의 현재 상태를 원격으로 묻는 네트워킹도 있습니다.

두 번째 문제는 중요한 시간 제어입니다 (정확한 타이머가 있어야합니다 - 멈출 때 스레드를 다시 시작할 수 없음). 끝나면 특별한 최우선 과제가 나오고 모든 연구가 재개됩니다.

세 번째 문제는 수행 할 작업의 상한이 없다는 것입니다.

내가 관찰 한 이러한 세 가지 제약 조건은 TPL을 잘 따라 가지 않습니다. Parallel.For와 같은 것을 사용할 수 없습니다. 실시간으로 작업 자체의 결과에 따라 컬렉션이 수정되기 때문입니다 ... I do not 결합하는 방법도 알고 TPL 한 번만

만드는 스레드 (일시 정지 및 연속 다시 시작 사이의 동기화 포인트) 스레드의 평생들이받은의 종류와

를 작성해야합니다 얼마나 많은 스레드를 결정하게하는

• 능력을 처음에는 항상 새로운 PA로 다시 시작해야합니다. rameters)

누군가가 내게 단서를 줄 수 있습니까? 나는 그것을 나쁜, 비효율적 인 방법으로하는 법을 안다. 내가 설명했던 몇 가지 작은 요구 사항이 있는데, 이는 내가이 권리를하지 못하게합니다. 나는 조금 혼란 스럽다.

Answer 1

메시징 + 액터 + 스케쥴러를 사용해야합니다.그런 다음 가능한 언어를 사용해야합니다. Azure Service Bus에서 비동기 적으로 수신하고 공유 대기열에 대기시키고 액터를 통해 런타임 상태를 관리하는 this code을 살펴보십시오.

인라인 :

내가 Task.ContinueWith 같은 작업 사용해야합니까?

아니요, ContinueWith는 각 연속 전달에서 예외 처리를 기반으로 프로그램을 종료합니다. 실패한 상태를 호출 측/주 스레드로 마샬링하는 TPL에는 좋은 방법이 없습니다.

하지만 이전 작업 실행의 결과를 처리하고 있지 않다, 그러나 ... 대신 내가 메인 스레드에서 데이터 구조를 업데이트 한 다음 새 작업 반복의 입력됩니다 결정할

문제에 많은 시간을 할애하지 않는 한 스레딩을 넘어서야합니다.

나는 많은 작업이 최고의 효율성을 생성하는 방법을 TPL의 결정을 떠날 수 있는가?

비동기 워크 플로를 실행하는 프레임 워크에서 처리합니다.

없는 그들이 좋은 RunEventCompleted 이벤트가 becase 내가 BackgroundWorkers을 사용하고 아니오 - 그 안에 BackgroundWorker에 다시 나는 내가 내 주요 구조를 업데이트 시간 제약을 확인할 수 있습니다 내 주 스레드에서 생각하고 결국 호출 StartAsync 완료되었습니다. 그것은 좋고 명확하지만, 아마도 매우 inneficient입니다. 멀티 프로세서, 멀티 코어 서버에서 매우 효율적으로이 작업을 수행해야합니다.

하나의 문제는 계산이 항상 온라인 상태이며 중단되지 않는다는 것입니다. 거기에 도 현재 MAIN 구조의 상태를 원격으로 요청할 수있는 네트워킹입니다. 두 번째 문제는 중요한 시간 제어입니다. ( 은 정확한 타이머가 있어야합니다. 스레드가 멈추었을 때 을 다시 시작할 수 없음).

모든 항목을 비동기 적으로 실행하면 메시지를 일시 중단 한 액터로 메시지를 전달할 수 있습니다. 액티비티를 스케줄하는 사용자는 모든 구독자에게 스케줄 된 메시지를 호출해야합니다. 연결된 코드에서 paused 상태를 살펴보십시오. 미해결 요청이있는 경우 취소 토큰을 전달하고 '하드'취소/소켓 중단을 처리 할 수 ​​있습니다. 그것은 종료 후

는 모든 작업을 재개, 특별한 우선 순위가 높은 작업을 온다. 이 두 제약, 내가 관찰 된 것과, 잘 TPL을 따라 가지 마세요 - 나는 컬렉션이 실시간으로 작업 자체의 결과에 의해 수정 때문에 Parallel.For 같은 것을 사용할 수 없습니다 ...

에게 당신을 아마도 파이프 앤 필터 (pipe-and-filters)라는 패턴이 필요할 것입니다. 입력 내용을 근로자 체인 (배우)에 연결합니다. 각 근로자는 다른 근로자의 결과물을 소비합니다.시그널링은 제어 채널을 사용하여 수행됩니다 (제 경우에는 액터의 인 박스 임). 여러 소비자를 소비하고 다른 소비자에게 보내하기로 결정하면서, 하나 개의 생산 생산 품목 :

Answer 2

난 당신이 저도 같은 문제를 겪고

MSDN: How to implement a producer/consumer dataflow pattern

읽어한다고 생각합니다. 각 소비자는 다른 소비자와 독립적으로 비동기 적으로 작업하고있었습니다.

주요 작업은 생산자입니다. 다른 작업에서 처리해야하는 항목을 생성합니다.

public async Task ProduceOutputAsync(...) 

메인 프로그램은 사용하여이 작업을 시작합니다 :

var producerTask = Task.Run(() => MyProducer.ProduceOutputAsync(...) 

이 생산자 작업이 출력을 생성하기 시작이라고하면 주요 작업의 코드 클래스는 기능이 있습니다. 한편 주 프로그램은 예를 들어 소비자를 시작하는 것과 같은 다른 일을 계속할 수 있습니다.

먼저 Producer 작업에 초점을 맞추어 보겠습니다.

생산자 작업은 다른 작업에서 처리 할 T 유형의 항목을 생성합니다. 그것들은 ITargetBlock '을 구현하는 객체를 사용하여 다른 작업으로 이월됩니다.

제작자 태스크가 ITargetBlock.Post를 사용하여 대상 블록 또는 바람직 비동기 버전로 전송 타입 T의 객체를 생성 완료마다 :

while (continueProducing()) 
{ 
    T product = await CreateProduct(...) 
    bool accepted = await this.TargetBlock(product) 
    // process the return value 
} 
// if here, nothing to produce anymore. Notify the consumers: 
this.TargetBlock.Complete(); 

프로듀서>은 ITargetBlock <T 필요하다. 내 응용 프로그램에서는 BufferBlock <T>으로 충분했습니다. MSDN에서 가능한 다른 대상을 확인하십시오.

어쨌든 데이터 흐름 블록은 ISourceBlock <T>을 구현해야합니다. 수신자는 입력이 소스에 도달 할 때까지 기다리고, 소스를 가져 와서 처리합니다. 일단 끝나면 결과를 자체 대상 블록에 보내고 더 이상 입력이 없을 때까지 다음 입력을 기다릴 수 있습니다. 물론 소비자가 생산량을 산출하지 못하면 목표물에 아무 것도 보낼 필요가 없습니다. 다음과 같이 입력을 기다리는

이 수행됩니다

ISourceBlock`<T`> mySource = ...; 
while (await mySource.ReceiveAsync()) 
{ // a object of type T is available at the source 
    T objectToProcess = await mySource.ReceiveAsync(); 
    // keep in mind that someone else might have fetched your object 
    // so only process it if you've got it. 
    if (objectToProcess != null) 
    { 
     await ProcessAsync(objectToProcess); 

     // if your processing produces output send the output to your target: 
     var myOutput = await ProduceOutput(objectToprocess); 
     await myTarget.SendAsync(myOutput); 
    } 
} 
// if here, no input expected anymore, notify my consumers: 
myTarget.Complete(); 

• 이 프로듀서
• 구조를 구성하는 모든 소비자
• 시작 생산자 MyProducer로 출력을 보내도록 생산자에게 BufferBlock을 제공 .ProduceOutputAsync (...)
• 생산자가 출력을 생성하고 버퍼 블록으로 보냅니다.
• 는
• 이 Task.WhenAll (...)이 완료 될 때까지 모든 작업을 기다릴 기다리고 별도의 작업으로 같은 BufferBlock
• 시작을 소비자에게 소비자를 제공합니다.

더 이상 입력이 필요 없다는 소식을 듣 자마자 중단됩니다. 모든 작업이 끝나면 주 기능이 결과를 읽고 반환 할 수 있습니다.