Framework Concepts - Real-time Streams

단순히 문서 해석과 궁금한 점을 덧붙여 작성하였습니다.

시리즈는 아래와 같습니다.

Framework Concepts

1. Calculators

2. Graphs

3. Packets

4. Synchronization

5. GPU

6. Real-time Streams

 

Framework Concepts - Real-time Streams

 

Real-time timestamps

Mediapipe calculator graphs 는 주로 비디오의 스트림을 처리하거나 상호작용 애플리케이션을 위한 오디오 프레임들을 처리하는 데에 사용됩니다. Mediapipe 프레임워크는 단일적으로 증가하는 timestamps 를 할당하기 위한 연달은 패킷들을 오직 필요로 합니다. 관례에 따르면, 실시간 calculators와 graphs 는 레코딩 시간이나 그 timestamps 로써 각 프레임의 프레젠테이션 타임에, Jan/1/1970:00:00:00 이후부터 마이크로세컨드를 가리키는 각 timestamp와 같이 사용합니다. 이것은 여러 소스로부터 오는 패킷이 글로벌하게 일관성있는 시퀀스 안에서 처리되는 것을 허용합니다.

Real-time scheduling

보통, 각 Calculator는 주어진 timestamp의 모든 input packets을 사용할 수 있게 되면 바로 실행됩니다. 보통, 이것은 calculator가 이전 프레임을 프로세싱하는 것을 끝냈을 때, 그리고 calculaotrs의 각각이 그것의 inputs을 각 프레임을 처리하는 것을 끝냈을 때 발생합니다. Mediapipe scheduler는 각 calculaotr가 이 조건이 충족되는대로 각 calculator를 발생시킵니다. 더 자세한 것은 Synchronization을 참고하세요.

Timestamp bounds

calculator가 주어진 timestamp에 대해 어떤 output packets을 생산하지 않을 때, 대신에 그 timestamp에 대해 패킷이 생성되지 않음을 나타내는 "timestamp bound" 를 출력할 수 있습니다. 이 나타냄은 해당 timestamp에 downstream calculators를 허용하기 위해 필요합니다, 심지어 어떤 패킷도 그 timestamp를 위한 특정한 streams를 위해 도착하지 않는다 해도 말입니다. 이것은 각 calculator가 가능한 즉시 처리하는 것을 시작하는 것이 중요한 상호작용하는 애플리케이션에서 실시간 그래프에 특별히 중요합니다.

따라오는 것과 같이 그래프를 고려하세요:

node {
   calculator: "A"
   input_stream: "alpha_in"
   output_stream: "alpha"
}
node {
   calculator: "B"
   input_stream: "alpha"
   input_stream: "foo"
   output_stream: "beta"
}

Suppose: timestamp T에서, node A가 그것의 output stream alpha 에서 패킷을 전달하지 않습니다. Node B는 timestamp T 에서 foo안의 패킷을 가집니다. 그리고 timestamp T에서 alpha의 패킷을 위해 기다립니다. 만약 alpha를 위해 timestamp bound 에서 A가 B를 전달하지 않는다면, B는 alpha에서 도착하기위한 패킷을 계속 기다릴 것입니다. 한편, foo의 패킷 큐는 아마도 T에서 패킷들을 쌓고 있고, T+1에도 그 이후에도 그럴 것입니다.

stream 에서 packet을 출력하기 위하여, calculaotr는 API 함수 CalculatorContext::Outputs 와 OutputStream::Add 를 사용합니다. stream 에서 timestamp bound의 출력을 대신하기 위해, calculator는 API 함수 CalculatorContext::Outputs와 CalculatorContext::SetNextTimestampBound 를 사용할 수 있습니다. 그 특정한 bound는 특정한 output stream 에서 다음 packet을 위한 제일 낮게 허용할 수 있는 timestamp입니다. 어떤 packet도 출력이 없을 때, calculator는 보통 이런 것을 합니다:

cc->Outputs().Tag("output_frame").SetNextTimestampBound(
  cc->InputTimestamp().NextAllowedInStream());

이 함수 Timestamp::NextAllowedInStream은 연속적인 timestamp를 반환합니다. 예를 들어, Timestamp(1).NEtAllowedInStream() == Timestamp(2).

Propagating timestamp bounds

실시간 그래프에서 사용될 Calculators 는 즉시 예정되기 위한 downstream calculators를 허용하기 위하여 input timestmamp bounds를 기반으로 한 output timestamp bounds를 정의할 필요가 있습니다. 일반적인 패턴은 calculators를 위해 그들의 input packets로써 같은 timestamps와 output packets를 출력하는 것입니다. 이 경우에, Calculator::Process 를 향한 모든 호출에서 간단하게 출력된 packet은 output stamp bounds로 정의되기에 충분합니다.

그러나, calculators는 output timestamps를 위한 이런 일반적인 패턴을 따르는 것을 요청하지 않습니다, 그들은 오직 단일적으로 output stamps가 증가하는 것을 선택하는 것을 요청합니다. 그 결과, 특정한 calculators는 반드시 timestamp bounds를 명시적으로 계산해야만 합니다. Mediapipe는 각 calculator를 위한 적절한 timestamp bound를 계산하기 위한 몇 가지의 도구를 제공합니다.

1. SetNextTimestampBound() 를 사용하여 output stream의 timestamp bound, t + 1을 지정할 수 있습니다

cc->Outputs.Tag("OUT").SetNextTimestampBound(t.NextAllowedInStream());

그 대신에, timestamp t와 같이 있는 빈 packet은 timestamp bound t + 1를 지정하도록 생산해낼 수 있습니다.

cc->Outputs.Tag("OUT").Add(Packet(), t);

input stream의 timestamp bound는 input stream에서 packet 혹은 빈 packet에 의해 나타내집니다.

Timestamp bound = cc->Inputs().Tag("IN").Value().Timestamp();

2. TimestampOffset() 은 input streams 로부터 output streams까지 timestamp bound를 자동적으로 복사하기 위하여 지정될 수 있습니다.

cc->SetTimestampOffset(0);

이 설정은 자동적으로 timestamp bounds를 전파하는 장점을 가지고 있고, 심지어 timestamp bound가 도착하고 Calculator::Process가 일어나지 않을 때도 그러합니다.

3. ProcessTimestampBounds()는 각 새로운 "고정된 timestamp"를 위해 Calculator::Process를 일으키기 위하여 지정될 수 있습니다. 여기서 "고정된 timestamp"는 현재의 timestamp bounds아래의 새롭게 최고 높은 timestamp 입니다. ProcessTimestampBounds() 없이, Calculator::Process 는 오직 하나 혹은 더 많은 도착하는 패킷들과 함께 일어납니다.

cc->SetProcessTimestampBounds(true);

이 설정은 calculator가 그가 가진 timestamp bounds 계산과 전파를 구현하기 위한 것을 허용합니다, 심지어 오직 input timestamps만 업데이트 되는 때에도 말입니다. 이것은 TimestampOffset()의 영향을 복제하는 것으로 사용될 수 있지만, 이것은 또한 추가적인 요소들을 고려하는 timestamp bound 를 계산하는데에도 사용할 수 있습니다.

예를 들어, SetTimestampOffset(0) 을 복제하기 위하여, calculator는 따라오는 일들을 할 수 있습니다:

absl::Status Open(CalculatorContext* cc) {
  cc->SetProcessTimestampBounds(true);
}

absl::Status Process(CalculatorContext* cc) {
  cc->Outputs.Tag("OUT").SetNextTimestampBound(
      cc->InputTimestamp().NextAllowedInStream());
}

Scheduling of Calculator::Open and Calculator::Close

Calculator::Open 은 모든 필요한 input side-packets이 생산될 때 발생합니다. input side-packets는 애플리케이션을 둘러싸거나 그래프 안에서 "side-packet calculators"에 의해 제공될 수 있습니다. Side-packets는 API의 CalculatorGraph::Initialize와 CalculatorGraph::StartRun 를 사용하는 그래프의 바깥쪽으로부터 지정될 수 있습니다. side packets는 CalculatorGraphConfig::OutputSidePackets과 OutputSidePacket::Set을 사용하는 그래프 안에서 calculators에 의해 지정될 수 있습니다.

Calculator::Close는 input streams 모두가 닫히거나 Timestamp::Done timestamp bound에 도달함에 의해서 Done 이 될 때 발생합니다.

Note: 만약 그래프가 모든 진행되는 calculator 실행을 끝마치고 Done이 되면, 일부 streams가 Done이 되기 전에, 그러면 Mediapipe는 Calculator::Close를 위해 남은 호출을 발생시킬 겁니다, 결국 모든 calculator는 그것의 마지막 출력들을 생산해낼 수 있게 됩니다.

TimestampOffset의 사용은 Calculator::Close를 위한 영향을 가지고 있습니다. SetTimestampOffset(0) 로 지정된 calculator는 모든 input stream 이 timestamp에 도달했을 때 모든 출력 스트림이 timestamp에 도달했다는 설계신호에 의해집니다. 그리고 그래서 더 가능한 출력이 없습니다. 이것은 Calculator::Close동안 어떤 packet도 송신하는 calculator를 예방합니다. 만약 calculator가 Calculator::Close동안 summary packet을 생산할 필요가 있다면, Calculator::Proces는 Caclulator::Close동안 남을 수 있는 최소한 하나의 timestamp(Timestamp::Max 와 같은) 와 같은 timestamp bounds를 지정해야만 합니다. 이것은 보통 calculator가 SEtTimestampOffset(0)에 의지할 수 없고 반드시 SetNextTimestampBounds()를 사용하여 명시적으로 timestamp bounds를 지정하는 것을 대신해야만 하는 calculator를 의미합니다.

 

 

References

https://google.github.io/mediapipe/framework_concepts/realtime_streams.html