talksik/mediapipe
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Project import generated by Copybara.

Browse Source 
GitOrigin-RevId: f72a0f86c2c2acdb1920973c718a9e26ed3ec4b6
This commit is contained in:
MediaPipe Team
2020-06-08 12:08:33 -04:00
committed by chuoling
parent 5d028d923b
commit cd2b69d58c
315 changed files with 8162 additions and 7502 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/framework_concepts/calculators.md
+412
View File
@@ -0,0 +1,412 @@
---
layout: default
title: Calculators
parent: Framework Concepts
nav_order: 1
---
# Calculators
{: .no_toc }
1. TOC
{:toc}
---
Each calculator is a node of a graph. We describe how to create a new
calculator, how to initialize a calculator, how to perform its calculations,
input and output streams, timestamps, and options. Each node in the graph is
implemented as a `Calculator`. The bulk of graph execution happens inside its
calculators. A calculator may receive zero or more input streams and/or side
packets and produces zero or more output streams and/or side packets.
## CalculatorBase
A calculator is created by defining a new sub-class of the
[`CalculatorBase`](https://github.com/google/mediapipe/tree/master/mediapipe/framework/calculator_base.cc)
class, implementing a number of methods, and registering the new sub-class with
Mediapipe. At a minimum, a new calculator must implement the below four methods
* `GetContract()`
* Calculator authors can specify the expected types of inputs and outputs
of a calculator in GetContract(). When a graph is initialized, the
framework calls a static method to verify if the packet types of the
connected inputs and outputs match the information in this
specification.
* `Open()`
* After a graph starts, the framework calls `Open()`. The input side
packets are available to the calculator at this point. `Open()`
interprets the node configuration operations (see [Graphs](graphs.md))
and prepares the calculator's per-graph-run state. This function may
also write packets to calculator outputs. An error during `Open()` can
terminate the graph run.
* `Process()`
* For a calculator with inputs, the framework calls `Process()` repeatedly
whenever at least one input stream has a packet available. The framework
by default guarantees that all inputs have the same timestamp (see
[Synchronization](synchronization.md) for more information). Multiple
`Process()` calls can be invoked simultaneously when parallel execution
is enabled. If an error occurs during `Process()`, the framework calls
`Close()` and the graph run terminates.
* `Close()`
* After all calls to `Process()` finish or when all input streams close,
the framework calls `Close()`. This function is always called if
`Open()` was called and succeeded and even if the graph run terminated
because of an error. No inputs are available via any input streams
during `Close()`, but it still has access to input side packets and
therefore may write outputs. After `Close()` returns, the calculator
should be considered a dead node. The calculator object is destroyed as
soon as the graph finishes running.
The following are code snippets from
[CalculatorBase.h](https://github.com/google/mediapipe/tree/master/mediapipe/framework/calculator_base.h).
```c++
class CalculatorBase {
public:
...
// The subclasses of CalculatorBase must implement GetContract.
// ...
static ::MediaPipe::Status GetContract(CalculatorContract* cc);
// Open is called before any Process() calls, on a freshly constructed
// calculator. Subclasses may override this method to perform necessary
// setup, and possibly output Packets and/or set output streams' headers.
// ...
virtual ::MediaPipe::Status Open(CalculatorContext* cc) {
return ::MediaPipe::OkStatus();
}
// Processes the incoming inputs. May call the methods on cc to access
// inputs and produce outputs.
// ...
virtual ::MediaPipe::Status Process(CalculatorContext* cc) = 0;
// Is called if Open() was called and succeeded. Is called either
// immediately after processing is complete or after a graph run has ended
// (if an error occurred in the graph). ...
virtual ::MediaPipe::Status Close(CalculatorContext* cc) {
return ::MediaPipe::OkStatus();
}
...
};
```
## Life of a calculator
During initialization of a MediaPipe graph, the framework calls a
`GetContract()` static method to determine what kinds of packets are expected.
The framework constructs and destroys the entire calculator for each graph run
(e.g. once per video or once per image). Expensive or large objects that remain
constant across graph runs should be supplied as input side packets so the
calculations are not repeated on subsequent runs.
After initialization, for each run of the graph, the following sequence occurs:
* `Open()`
* `Process()` (repeatedly)
* `Close()`
The framework calls `Open()` to initialize the calculator. `Open()` should
interpret any options and set up the calculator's per-graph-run state. `Open()`
may obtain input side packets and write packets to calculator outputs. If
appropriate, it should call `SetOffset()` to reduce potential packet buffering
of input streams.
If an error occurs during `Open()` or `Process()` (as indicated by one of them
returning a non-`Ok` status), the graph run is terminated with no further calls
to the calculator's methods, and the calculator is destroyed.
For a calculator with inputs, the framework calls `Process()` whenever at least
one input has a packet available. The framework guarantees that inputs all have
the same timestamp, that timestamps increase with each call to `Process()` and
that all packets are delivered. As a consequence, some inputs may not have any
packets when `Process()` is called. An input whose packet is missing appears to
produce an empty packet (with no timestamp).
The framework calls `Close()` after all calls to `Process()`. All inputs will
have been exhausted, but `Close()` has access to input side packets and may
write outputs. After Close returns, the calculator is destroyed.
Calculators with no inputs are referred to as sources. A source calculator
continues to have `Process()` called as long as it returns an `Ok` status. A
source calculator indicates that it is exhausted by returning a stop status
(i.e. MediaPipe::tool::StatusStop).
## Identifying inputs and outputs
The public interface to a calculator consists of a set of input streams and
output streams. In a CalculatorGraphConfiguration, the outputs from some
calculators are connected to the inputs of other calculators using named
streams. Stream names are normally lowercase, while input and output tags are
normally UPPERCASE. In the example below, the output with tag name `VIDEO` is
connected to the input with tag name `VIDEO_IN` using the stream named
`video_stream`.
```proto
# Graph describing calculator SomeAudioVideoCalculator
node {
calculator: "SomeAudioVideoCalculator"
input_stream: "INPUT:combined_input"
output_stream: "VIDEO:video_stream"
}
node {
calculator: "SomeVideoCalculator"
input_stream: "VIDEO_IN:video_stream"
output_stream: "VIDEO_OUT:processed_video"
}
```
Input and output streams can be identified by index number, by tag name, or by a
combination of tag name and index number. You can see some examples of input and
output identifiers in the example below. `SomeAudioVideoCalculator` identifies
its video output by tag and its audio outputs by the combination of tag and
index. The input with tag `VIDEO` is connected to the stream named
`video_stream`. The outputs with tag `AUDIO` and indices `0` and `1` are
connected to the streams named `audio_left` and `audio_right`.
`SomeAudioCalculator` identifies its audio inputs by index only (no tag needed).
```proto
# Graph describing calculator SomeAudioVideoCalculator
node {
calculator: "SomeAudioVideoCalculator"
input_stream: "combined_input"
output_stream: "VIDEO:video_stream"
output_stream: "AUDIO:0:audio_left"
output_stream: "AUDIO:1:audio_right"
}
node {
calculator: "SomeAudioCalculator"
input_stream: "audio_left"
input_stream: "audio_right"
output_stream: "audio_energy"
}
```
In the calculator implementation, inputs and outputs are also identified by tag
name and index number. In the function below input are output are identified:
* By index number: The combined input stream is identified simply by index
`0`.
* By tag name: The video output stream is identified by tag name "VIDEO".
* By tag name and index number: The output audio streams are identified by the
combination of the tag name `AUDIO` and the index numbers `0` and `1`.
```c++
// c++ Code snippet describing the SomeAudioVideoCalculator GetContract() method
class SomeAudioVideoCalculator : public CalculatorBase {
public:
static ::mediapipe::Status GetContract(CalculatorContract* cc) {
cc->Inputs().Index(0).SetAny();
// SetAny() is used to specify that whatever the type of the
// stream is, it's acceptable. This does not mean that any
// packet is acceptable. Packets in the stream still have a
// particular type. SetAny() has the same effect as explicitly
// setting the type to be the stream's type.
cc->Outputs().Tag("VIDEO").Set<ImageFrame>();
cc->Outputs().Get("AUDIO", 0).Set<Matrix>;
cc->Outputs().Get("AUDIO", 1).Set<Matrix>;
return ::mediapipe::OkStatus();
}
```
## Processing
`Process()` called on a non-source node must return `::mediapipe::OkStatus()` to
indicate that all went well, or any other status code to signal an error
If a non-source calculator returns `tool::StatusStop()`, then this signals the
graph is being cancelled early. In this case, all source calculators and graph
input streams will be closed (and remaining Packets will propagate through the
graph).
A source node in a graph will continue to have `Process()` called on it as long
as it returns `::mediapipe::OkStatus(`). To indicate that there is no more data
to be generated return `tool::StatusStop()`. Any other status indicates an error
has occurred.
`Close()` returns `::mediapipe::OkStatus()` to indicate success. Any other
status indicates a failure.
Here is the basic `Process()` function. It uses the `Input()` method (which can
be used only if the calculator has a single input) to request its input data. It
then uses `std::unique_ptr` to allocate the memory needed for the output packet,
and does the calculations. When done it releases the pointer when adding it to
the output stream.
```c++
::util::Status MyCalculator::Process() {
const Matrix& input = Input()->Get<Matrix>();
std::unique_ptr<Matrix> output(new Matrix(input.rows(), input.cols()));
// do your magic here....
// output->row(n) = ...
Output()->Add(output.release(), InputTimestamp());
return ::mediapipe::OkStatus();
}
```
## Example calculator
This section discusses the implementation of `PacketClonerCalculator`, which
does a relatively simple job, and is used in many calculator graphs.
`PacketClonerCalculator` simply produces a copy of its most recent input
packets on demand.
`PacketClonerCalculator` is useful when the timestamps of arriving data packets
are not aligned perfectly. Suppose we have a room with a microphone, light
sensor and a video camera that is collecting sensory data. Each of the sensors
operates independently and collects data intermittently. Suppose that the output
of each sensor is:
* microphone = loudness in decibels of sound in the room (Integer)
* light sensor = brightness of room (Integer)
* video camera = RGB image frame of room (ImageFrame)
Our simple perception pipeline is designed to process sensory data from these 3
sensors such that at any time when we have image frame data from the camera that
is synchronized with the last collected microphone loudness data and light
sensor brightness data. To do this with MediaPipe, our perception pipeline has 3
input streams:
* room_mic_signal - Each packet of data in this input stream is integer data
representing how loud audio is in a room with timestamp.
* room_lightening_sensor - Each packet of data in this input stream is integer
data representing how bright is the room illuminated with timestamp.
* room_video_tick_signal - Each packet of data in this input stream is
imageframe of video data representing video collected from camera in the
room with timestamp.
Below is the implementation of the `PacketClonerCalculator`. You can see
the `GetContract()`, `Open()`, and `Process()` methods as well as the instance
variable `current_` which holds the most recent input packets.
```c++
// This takes packets from N+1 streams, A_1, A_2, ..., A_N, B.
// For every packet that appears in B, outputs the most recent packet from each
// of the A_i on a separate stream.
#include <vector>
#include "absl/strings/str_cat.h"
#include "mediapipe/framework/calculator_framework.h"
namespace mediapipe {
// For every packet received on the last stream, output the latest packet
// obtained on all other streams. Therefore, if the last stream outputs at a
// higher rate than the others, this effectively clones the packets from the
// other streams to match the last.
//
// Example config:
// node {
// calculator: "PacketClonerCalculator"
// input_stream: "first_base_signal"
// input_stream: "second_base_signal"
// input_stream: "tick_signal"
// output_stream: "cloned_first_base_signal"
// output_stream: "cloned_second_base_signal"
// }
//
class PacketClonerCalculator : public CalculatorBase {
public:
static ::mediapipe::Status GetContract(CalculatorContract* cc) {
const int tick_signal_index = cc->Inputs().NumEntries() - 1;
// cc->Inputs().NumEntries() returns the number of input streams
// for the PacketClonerCalculator
for (int i = 0; i < tick_signal_index; ++i) {
cc->Inputs().Index(i).SetAny();
// cc->Inputs().Index(i) returns the input stream pointer by index
cc->Outputs().Index(i).SetSameAs(&cc->Inputs().Index(i));
}
cc->Inputs().Index(tick_signal_index).SetAny();
return ::mediapipe::OkStatus();
}
::mediapipe::Status Open(CalculatorContext* cc) final {
tick_signal_index_ = cc->Inputs().NumEntries() - 1;
current_.resize(tick_signal_index_);
// Pass along the header for each stream if present.
for (int i = 0; i < tick_signal_index_; ++i) {
if (!cc->Inputs().Index(i).Header().IsEmpty()) {
cc->Outputs().Index(i).SetHeader(cc->Inputs().Index(i).Header());
// Sets the output stream of index i header to be the same as
// the header for the input stream of index i
}
}
return ::mediapipe::OkStatus();
}
::mediapipe::Status Process(CalculatorContext* cc) final {
// Store input signals.
for (int i = 0; i < tick_signal_index_; ++i) {
if (!cc->Inputs().Index(i).Value().IsEmpty()) {
current_[i] = cc->Inputs().Index(i).Value();
}
}
// Output if the tick signal is non-empty.
if (!cc->Inputs().Index(tick_signal_index_).Value().IsEmpty()) {
for (int i = 0; i < tick_signal_index_; ++i) {
if (!current_[i].IsEmpty()) {
cc->Outputs().Index(i).AddPacket(
current_[i].At(cc->InputTimestamp()));
// Add a packet to output stream of index i a packet from inputstream i
// with timestamp common to all present inputs
//
} else {
cc->Outputs().Index(i).SetNextTimestampBound(
cc->InputTimestamp().NextAllowedInStream());
// if current_[i], 1 packet buffer for input stream i is empty, we will set
// next allowed timestamp for input stream i to be current timestamp + 1
}
}
}
return ::mediapipe::OkStatus();
}
private:
std::vector<Packet> current_;
int tick_signal_index_;
};
REGISTER_CALCULATOR(PacketClonerCalculator);
} // namespace mediapipe
```
Typically, a calculator has only a .cc file. No .h is required, because
mediapipe uses registration to make calculators known to it. After you have
defined your calculator class, register it with a macro invocation
REGISTER_CALCULATOR(calculator_class_name).
Below is a trivial MediaPipe graph that has 3 input streams, 1 node
(PacketClonerCalculator) and 3 output streams.
```proto
input_stream: "room_mic_signal"
input_stream: "room_lighting_sensor"
input_stream: "room_video_tick_signal"
node {
calculator: "PacketClonerCalculator"
input_stream: "room_mic_signal"
input_stream: "room_lighting_sensor"
input_stream: "room_video_tick_signal"
output_stream: "cloned_room_mic_signal"
output_stream: "cloned_lighting_sensor"
}
```
The diagram below shows how the `PacketClonerCalculator` defines its output
packets based on its series of input packets.
| ![Graph using |
: PacketClonerCalculator](../images/packet_cloner_calculator.png) :
| :--------------------------------------------------------------------------: |
| *Each time it receives a packet on its TICK input stream, the |
: PacketClonerCalculator outputs the most recent packet from each of its input :
: streams. The sequence of output packets is determined by the sequene of :
: input packets and their timestamps. The timestamps are shows along the right :
: side of the diagram.* :
docs/framework_concepts/framework_concepts.md
+112
View File
@@ -0,0 +1,112 @@
---
layout: default
title: Framework Concepts
nav_order: 5
has_children: true
has_toc: false
---
# Framework Concepts
{: .no_toc }
1. TOC
{:toc}
---
## The basics
### Packet
The basic data flow unit. A packet consists of a numeric timestamp and a shared
pointer to an **immutable** payload. The payload can be of any C++ type, and the
payload's type is also referred to as the type of the packet. Packets are value
classes and can be copied cheaply. Each copy shares ownership of the payload,
with reference-counting semantics. Each copy has its own timestamp. See also
[Packet](packets.md).
### Graph
MediaPipe processing takes place inside a graph, which defines packet flow paths
between **nodes**. A graph can have any number of inputs and outputs, and data
flow can branch and merge. Generally data flows forward, but backward loops are
possible. See [Graphs](graphs.md) for details.
### Nodes
Nodes produce and/or consume packets, and they are where the bulk of the graph’s
work takes place. They are also known as “calculators”, for historical reasons.
Each node’s interface defines a number of input and output **ports**, identified
by a tag and/or an index. See [Calculators](calculators.md) for details.
### Streams
A stream is a connection between two nodes that carries a sequence of packets,
whose timestamps must be monotonically increasing.
### Side packets
A side packet connection between nodes carries a single packet (with unspecified
timestamp). It can be used to provide some data that will remain constant,
whereas a stream represents a flow of data that changes over time.
### Packet Ports
A port has an associated type; packets transiting through the port must be of
that type. An output stream port can be connected to any number of input stream
ports of the same type; each consumer receives a separate copy of the output
packets, and has its own queue, so it can consume them at its own pace.
Similarly, a side packet output port can be connected to as many side packet
input ports as desired.
A port can be required, meaning that a connection must be made for the graph to
be valid, or optional, meaning it may remain unconnected.
Note: even if a stream connection is required, the stream may not carry a packet
for all timestamps.
## Input and output
Data flow can originate from **source nodes**, which have no input streams and
produce packets spontaneously (e.g. by reading from a file); or from **graph
input streams**, which let an application feed packets into a graph.
Similarly, there are **sink nodes** that receive data and write it to various
destinations (e.g. a file, a memory buffer, etc.), and an application can also
receive output from the graph using **callbacks**.
## Runtime behavior
### Graph lifetime
Once a graph has been initialized, it can be **started** to begin processing
data, and can process a stream of packets until each stream is closed or the
graph is **canceled**. Then the graph can be destroyed or **started** again.
### Node lifetime
There are three main lifetime methods the framework will call on a node:
- Open: called once, before the other methods. When it is called, all input
side packets required by the node will be available.
- Process: called multiple times, when a new set of inputs is available,
according to the node’s input policy.
- Close: called once, at the end.
In addition, each calculator can define constructor and destructor, which are
useful for creating and deallocating resources that are independent of the
processed data.
### Input policies
The default input policy is deterministic collation of packets by timestamp. A
node receives all inputs for the same timestamp at the same time, in an
invocation of its Process method; and successive input sets are received in
their timestamp order. This can require delaying the processing of some packets
until a packet with the same timestamp is received on all input streams, or
until it can be guaranteed that a packet with that timestamp will not be
arriving on the streams that have not received it.
Other policies are also available, implemented using a separate kind of
component known as an InputStreamHandler.
See [Synchronization](synchronization.md) for more details.
docs/framework_concepts/gpu.md
+163
View File
@@ -0,0 +1,163 @@
---
layout: default
title: GPU
parent: Framework Concepts
nav_order: 5
---
# GPU
{: .no_toc }
1. TOC
{:toc}
---
## Overview
MediaPipe supports calculator nodes for GPU compute and rendering, and allows combining multiple GPU nodes, as well as mixing them with CPU based calculator nodes. There exist several GPU APIs on mobile platforms (eg, OpenGL ES, Metal and Vulkan). MediaPipe does not attempt to offer a single cross-API GPU abstraction. Individual nodes can be written using different APIs, allowing them to take advantage of platform specific features when needed.
GPU support is essential for good performance on mobile platforms, especially for real-time video. MediaPipe enables developers to write GPU compatible calculators that support the use of GPU for:
* On-device real-time processing, not just batch processing
* Video rendering and effects, not just analysis
Below are the design principles for GPU support in MediaPipe
* GPU-based calculators should be able to occur anywhere in the graph, and not necessarily be used for on-screen rendering.
* Transfer of frame data from one GPU-based calculator to another should be fast, and not incur expensive copy operations.
* Transfer of frame data between CPU and GPU should be as efficient as the platform allows.
* Because different platforms may require different techniques for best performance, the API should allow flexibility in the way things are implemented behind the scenes.
* A calculator should be allowed maximum flexibility in using the GPU for all or part of its operation, combining it with the CPU if necessary.
## OpenGL ES Support
MediaPipe supports OpenGL ES up to version 3.2 on Android/Linux and up to ES 3.0
on iOS. In addition, MediaPipe also supports Metal on iOS.
OpenGL ES 3.1 or greater is required (on Android/Linux systems) for running
machine learning inference calculators and graphs.
MediaPipe allows graphs to run OpenGL in multiple GL contexts. For example, this
can be very useful in graphs that combine a slower GPU inference path (eg, at 10
FPS) with a faster GPU rendering path (eg, at 30 FPS): since one GL context
corresponds to one sequential command queue, using the same context for both
tasks would reduce the rendering frame rate.
One challenge MediaPipe's use of multiple contexts solves is the ability to
communicate across them. An example scenario is one with an input video that is
sent to both the rendering and inferences paths, and rendering needs to have
access to the latest output from inference.
An OpenGL context cannot be accessed by multiple threads at the same time.
Furthermore, switching the active GL context on the same thread can be slow on
some Android devices. Therefore, our approach is to have one dedicated thread
per context. Each thread issues GL commands, building up a serial command queue
on its context, which is then executed by the GPU asynchronously.
## Life of a GPU Calculator
This section presents the basic structure of the Process method of a GPU
calculator derived from base class GlSimpleCalculator. The GPU calculator
`LuminanceCalculator` is shown as an example. The method
`LuminanceCalculator::GlRender` is called from `GlSimpleCalculator::Process`.
```c++
// Converts RGB images into luminance images, still stored in RGB format.
// See GlSimpleCalculator for inputs, outputs and input side packets.
class LuminanceCalculator : public GlSimpleCalculator {
public:
::mediapipe::Status GlSetup() override;
::mediapipe::Status GlRender(const GlTexture& src,
const GlTexture& dst) override;
::mediapipe::Status GlTeardown() override;
private:
GLuint program_ = 0;
GLint frame_;
};
REGISTER_CALCULATOR(LuminanceCalculator);
::mediapipe::Status LuminanceCalculator::GlRender(const GlTexture& src,
const GlTexture& dst) {
static const GLfloat square_vertices[] = {
-1.0f, -1.0f, // bottom left
1.0f, -1.0f, // bottom right
-1.0f, 1.0f, // top left
1.0f, 1.0f, // top right
};
static const GLfloat texture_vertices[] = {
0.0f, 0.0f, // bottom left
1.0f, 0.0f, // bottom right
0.0f, 1.0f, // top left
1.0f, 1.0f, // top right
};
// program
glUseProgram(program_);
glUniform1i(frame_, 1);
// vertex storage
GLuint vbo[2];
glGenBuffers(2, vbo);
GLuint vao;
glGenVertexArrays(1, &vao);
glBindVertexArray(vao);
// vbo 0
glBindBuffer(GL_ARRAY_BUFFER, vbo[0]);
glBufferData(GL_ARRAY_BUFFER, 4 * 2 * sizeof(GLfloat), square_vertices,
GL_STATIC_DRAW);
glEnableVertexAttribArray(ATTRIB_VERTEX);
glVertexAttribPointer(ATTRIB_VERTEX, 2, GL_FLOAT, 0, 0, nullptr);
// vbo 1
glBindBuffer(GL_ARRAY_BUFFER, vbo[1]);
glBufferData(GL_ARRAY_BUFFER, 4 * 2 * sizeof(GLfloat), texture_vertices,
GL_STATIC_DRAW);
glEnableVertexAttribArray(ATTRIB_TEXTURE_POSITION);
glVertexAttribPointer(ATTRIB_TEXTURE_POSITION, 2, GL_FLOAT, 0, 0, nullptr);
// draw
glDrawArrays(GL_TRIANGLE_STRIP, 0, 4);
// cleanup
glDisableVertexAttribArray(ATTRIB_VERTEX);
glDisableVertexAttribArray(ATTRIB_TEXTURE_POSITION);
glBindBuffer(GL_ARRAY_BUFFER, 0);
glBindVertexArray(0);
glDeleteVertexArrays(1, &vao);
glDeleteBuffers(2, vbo);
return ::mediapipe::OkStatus();
}
```
The design principles mentioned above have resulted in the following design
choices for MediaPipe GPU support:
* We have a GPU data type, called `GpuBuffer`, for representing image data, optimized for GPU usage. The exact contents of this data type are opaque and platform-specific.
* A low-level API based on composition, where any calculator that wants to make use of the GPU creates and owns an instance of the `GlCalculatorHelper` class. This class offers a platform-agnostic API for managing the OpenGL context, setting up textures for inputs and outputs, etc.
* A high-level API based on subclassing, where simple calculators implementing image filters subclass from `GlSimpleCalculator` and only need to override a couple of virtual methods with their specific OpenGL code, while the superclass takes care of all the plumbing.
* Data that needs to be shared between all GPU-based calculators is provided as a external input that is implemented as a graph service and is managed by the `GlCalculatorHelper` class.
* The combination of calculator-specific helpers and a shared graph service allows us great flexibility in managing the GPU resource: we can have a separate context per calculator, share a single context, share a lock or other synchronization primitives, etc. -- and all of this is managed by the helper and hidden from the individual calculators.
## GpuBuffer to ImageFrame Converters
We provide two calculators called `GpuBufferToImageFrameCalculator` and `ImageFrameToGpuBufferCalculator`. These calculators convert between `ImageFrame` and `GpuBuffer`, allowing the construction of graphs that combine GPU and CPU calculators. They are supported on both iOS and Android
When possible, these calculators use platform-specific functionality to share data between the CPU and the GPU without copying.
The below diagram shows the data flow in a mobile application that captures video from the camera, runs it through a MediaPipe graph, and renders the output on the screen in real time. The dashed line indicates which parts are inside the MediaPipe graph proper. This application runs a Canny edge-detection filter on the CPU using OpenCV, and overlays it on top of the original video using the GPU.
| ![How GPU calculators interact](../images/gpu_example_graph.png) |
| :--------------------------------------------------------------------------: |
| *Video frames from the camera are fed into the graph as `GpuBuffer` packets. |
: The input stream is accessed by two calculators in parallel. :
: `GpuBufferToImageFrameCalculator` converts the buffer into an `ImageFrame`, :
: which is then sent through a grayscale converter and a canny filter (both :
: based on OpenCV and running on the CPU), whose output is then converted into :
: a `GpuBuffer` again. A multi-input GPU calculator, GlOverlayCalculator, :
: takes as input both the original `GpuBuffer` and the one coming out of the :
: edge detector, and overlays them using a shader. The output is then sent :
: back to the application using a callback calculator, and the application :
: renders the image to the screen using OpenGL.* :
docs/framework_concepts/graphs.md
+271
View File
@@ -0,0 +1,271 @@
---
layout: default
title: Graphs
parent: Framework Concepts
nav_order: 2
---
# Graphs
{: .no_toc }
1. TOC
{:toc}
---
## GraphConfig
A `GraphConfig` is a specification that describes the topology and functionality
of a MediaPipe graph. In the specification, a node in the graph represents an
instance of a particular calculator. All the necessary configurations of the
node, such its type, inputs and outputs must be described in the specification.
Description of the node can also include several optional fields, such as
node-specific options, input policy and executor, discussed in
[Synchronization](synchronization.md).
`GraphConfig` has several other fields to configure the global graph-level
settings, eg, graph executor configs, number of threads, and maximum queue size
of input streams. Several graph-level settings are useful for tuning the
performance of the graph on different platforms (eg, desktop v.s. mobile). For
instance, on mobile, attaching a heavy model-inference calculator to a separate
executor can improve the performance of a real-time application since this
enables thread locality.
Below is a trivial `GraphConfig` example where we have series of passthrough
calculators :
```proto
# This graph named main_pass_throughcals_nosubgraph.pbtxt contains 4
# passthrough calculators.
input_stream: "in"
node {
calculator: "PassThroughCalculator"
input_stream: "in"
output_stream: "out1"
}
node {
calculator: "PassThroughCalculator"
input_stream: "out1"
output_stream: "out2"
}
node {
calculator: "PassThroughCalculator"
input_stream: "out2"
output_stream: "out3"
}
node {
calculator: "PassThroughCalculator"
input_stream: "out3"
output_stream: "out4"
}
```
## Subgraph
To modularize a `CalculatorGraphConfig` into sub-modules and assist with re-use
of perception solutions, a MediaPipe graph can be defined as a `Subgraph`. The
public interface of a subgraph consists of a set of input and output streams
similar to a calculator's public interface. The subgraph can then be included in
an `CalculatorGraphConfig` as if it were a calculator. When a MediaPipe graph is
loaded from a `CalculatorGraphConfig`, each subgraph node is replaced by the
corresponding graph of calculators. As a result, the semantics and performance
of the subgraph is identical to the corresponding graph of calculators.
Below is an example of how to create a subgraph named `TwoPassThroughSubgraph`.
1. Defining the subgraph.
```proto
# This subgraph is defined in two_pass_through_subgraph.pbtxt
# and is registered as "TwoPassThroughSubgraph"
type: "TwoPassThroughSubgraph"
input_stream: "out1"
output_stream: "out3"
node {
calculator: "PassThroughculator"
input_stream: "out1"
output_stream: "out2"
}
node {
calculator: "PassThroughculator"
input_stream: "out2"
output_stream: "out3"
}
```
The public interface to the subgraph consists of:
* Graph input streams
* Graph output streams
* Graph input side packets
* Graph output side packets
2. Register the subgraph using BUILD rule `mediapipe_simple_subgraph`. The
parameter `register_as` defines the component name for the new subgraph.
```proto
# Small section of BUILD file for registering the "TwoPassThroughSubgraph"
# subgraph for use by main graph main_pass_throughcals.pbtxt
mediapipe_simple_subgraph(
name = "twopassthrough_subgraph",
graph = "twopassthrough_subgraph.pbtxt",
register_as = "TwoPassThroughSubgraph",
deps = [
"//mediapipe/calculators/core:pass_through_calculator",
"//mediapipe/framework:calculator_graph",
],
)
```
3. Use the subgraph in the main graph.
```proto
# This main graph is defined in main_pass_throughcals.pbtxt
# using subgraph called "TwoPassThroughSubgraph"
input_stream: "in"
node {
calculator: "PassThroughCalculator"
input_stream: "in"
output_stream: "out1"
}
node {
calculator: "TwoPassThroughSubgraph"
input_stream: "out1"
output_stream: "out3"
}
node {
calculator: "PassThroughCalculator"
input_stream: "out3"
output_stream: "out4"
}
```
## Cycles
<!-- TODO: add discussion of PreviousLoopbackCalculator -->
By default, MediaPipe requires calculator graphs to be acyclic and treats cycles
in a graph as errors. If a graph is intended to have cycles, the cycles need to
be annotated in the graph config. This page describes how to do that.
NOTE: The current approach is experimental and subject to change. We welcome
your feedback.
Please use the `CalculatorGraphTest.Cycle` unit test in
`mediapipe/framework/calculator_graph_test.cc` as sample code. Shown
below is the cyclic graph in the test. The `sum` output of the adder is the sum
of the integers generated by the integer source calculator.
![a cyclic graph that adds a stream of integers](../images/cyclic_integer_sum_graph.svg "A cyclic graph")
This simple graph illustrates all the issues in supporting cyclic graphs.
### Back Edge Annotation
We require that an edge in each cycle be annotated as a back edge. This allows
MediaPipe’s topological sort to work, after removing all the back edges.
There are usually multiple ways to select the back edges. Which edges are marked
as back edges affects which nodes are considered as upstream and which nodes are
considered as downstream, which in turn affects the priorities MediaPipe assigns
to the nodes.
For example, the `CalculatorGraphTest.Cycle` test marks the `old_sum` edge as a
back edge, so the Delay node is considered as a downstream node of the adder
node and is given a higher priority. Alternatively, we could mark the `sum`
input to the delay node as the back edge, in which case the delay node would be
considered as an upstream node of the adder node and is given a lower priority.
### Initial Packet
For the adder calculator to be runnable when the first integer from the integer
source arrives, we need an initial packet, with value 0 and with the same
timestamp, on the `old_sum` input stream to the adder. This initial packet
should be output by the delay calculator in the `Open()` method.
### Delay in a Loop
Each loop should incur a delay to align the previous `sum` output with the next
integer input. This is also done by the delay node. So the delay node needs to
know the following about the timestamps of the integer source calculator:
* The timestamp of the first output.
* The timestamp delta between successive outputs.
We plan to add an alternative scheduling policy that only cares about packet
ordering and ignores packet timestamps, which will eliminate this inconvenience.
### Early Termination of a Calculator When One Input Stream is Done
By default, MediaPipe calls the `Close()` method of a non-source calculator when
all of its input streams are done. In the example graph, we want to stop the
adder node as soon as the integer source is done. This is accomplished by
configuring the adder node with an alternative input stream handler,
`EarlyCloseInputStreamHandler`.
### Relevant Source Code
#### Delay Calculator
Note the code in `Open()` that outputs the initial packet and the code in
`Process()` that adds a (unit) delay to input packets. As noted above, this
delay node assumes that its output stream is used alongside an input stream with
packet timestamps 0, 1, 2, 3, ...
```c++
class UnitDelayCalculator : public Calculator {
public:
 static ::util::Status FillExpectations(
     const CalculatorOptions& extendable_options, PacketTypeSet* inputs,
     PacketTypeSet* outputs, PacketTypeSet* input_side_packets) {
   inputs->Index(0)->Set<int>("An integer.");
   outputs->Index(0)->Set<int>("The input delayed by one time unit.");
   return ::mediapipe::OkStatus();
 }
 ::util::Status Open() final {
   Output()->Add(new int(0), Timestamp(0));
   return ::mediapipe::OkStatus();
 }
 ::util::Status Process() final {
   const Packet& packet = Input()->Value();
   Output()->AddPacket(packet.At(packet.Timestamp().NextAllowedInStream()));
   return ::mediapipe::OkStatus();
 }
};
```
#### Graph Config
Note the `back_edge` annotation and the alternative `input_stream_handler`.
```proto
node {
  calculator: 'GlobalCountSourceCalculator'
  input_side_packet: 'global_counter'
  output_stream: 'integers'
}
node {
  calculator: 'IntAdderCalculator'
  input_stream: 'integers'
  input_stream: 'old_sum'
  input_stream_info: {
    tag_index: ':1' # 'old_sum'
    back_edge: true
  }
  output_stream: 'sum'
  input_stream_handler {
    input_stream_handler: 'EarlyCloseInputStreamHandler'
  }
}
node {
  calculator: 'UnitDelayCalculator'
  input_stream: 'sum'
  output_stream: 'old_sum'
}
```
docs/framework_concepts/packets.md
+30
View File
@@ -0,0 +1,30 @@
---
layout: default
title: Packets
parent: Framework Concepts
nav_order: 3
---
# Packets
{: .no_toc }
1. TOC
{:toc}
---
Each calculator is a node of of a graph. We describe how to create a new calculator, how to initialize a calculator, how to perform its calculations, input and output streams, timestamps, and options
## Creating a packet
Packets are generally created with `MediaPipe::Adopt()` (from packet.h).
```c++
// Create some data.
auto data = gtl::MakeUnique<MyDataClass>("constructor_argument");
// Create a packet to own the data.
Packet p = Adopt(data.release());
// Make a new packet with the same data and a different timestamp.
Packet p2 = p.At(Timestamp::PostStream());
```
Data within a packet is accessed with `Packet::Get<T>()`
docs/framework_concepts/synchronization.md
+175
View File
@@ -0,0 +1,175 @@
---
layout: default
title: Synchronization
parent: Framework Concepts
nav_order: 4
---
# Synchronization
{: .no_toc }
1. TOC
{:toc}
---
## Scheduling mechanics
Data processing in a MediaPipe graph occurs inside processing nodes defined as
[`CalculatorBase`] subclasses. The scheduling system decides when each
calculator should run.
Each graph has at least one **scheduler queue**. Each scheduler queue has
exactly one **executor**. Nodes are statically assigned to a queue (and
therefore to an executor). By default there is one queue, whose executor is a
thread pool with a number of threads based on the system’s capabilities.
Each node has a scheduling state, which can be *not ready*, *ready*, or
*running*. A readiness function determines whether a node is ready to run. This
function is invoked at graph initialization, whenever a node finishes running,
and whenever the state of a node’s inputs changes.
The readiness function used depends on the type of node. A node with no stream
inputs is known as a **source node**; source nodes are always ready to run,
until they tell the framework they have no more data to output, at which point
they are closed.
Non-source nodes are ready if they have inputs to process, and if those inputs
form a valid input set according to the conditions set by the node’s **input
policy** (discussed below). Most nodes use the default input policy, but some
nodes specify a different one.
Note: Because changing the input policy changes the guarantees the calculator’s
code can expect from its inputs, it is not generally possible to mix and match
calculators with arbitrary input policies. Thus a calculator that uses a special
input policy should be written for it, and declare it in its contract.
When a node becomes ready, a task is added to the corresponding scheduler queue,
which is a priority queue. The priority function is currently fixed, and takes
into account static properties of the nodes and their topological sorting within
the graph. For example, nodes closer to the output side of the graph have higher
priority, while source nodes have the lowest priority.
Each queue is served by an executor, which is responsible for actually running
the task by invoking the calculator’s code. Different executors can be provided
and configured; this can be used to customize the use of execution resources,
e.g. by running certain nodes on lower-priority threads.
## Timestamp Synchronization
MediaPipe graph execution is decentralized: there is no global clock, and
different nodes can process data from different timestamps at the same time.
This allows higher throughput via pipelining.
However, time information is very important for many perception workflows. Nodes
that receive multiple input streams generally need to coordinate them in some
way. For example, an object detector may output a list of boundary rectangles
from a frame, and this information may be fed into a rendering node, which
should process it together with the original frame.
Therefore, one of the key responsibilities of the MediaPipe framework is to
provide input synchronization for nodes. In terms of framework mechanics, the
primary role of a timestamp is to serve as a **synchronization key**.
Furthermore, MediaPipe is designed to support deterministic operations, which is
important in many scenarios (testing, simulation, batch processing, etc.), while
allowing graph authors to relax determinism where needed to meet real-time
constraints.
The two objectives of synchronization and determinism underlie several design
choices. Notably, the packets pushed into a given stream must have monotonically
increasing timestamps: this is not just a useful assumption for many nodes, but
it is also relied upon by the synchronization logic. Each stream has a
**timestamp bound**, which is the lowest possible timestamp allowed for a new
packet on the stream. When a packet with timestamp `T` arrives, the bound
automatically advances to `T+1`, reflecting the monotonic requirement. This
allows the framework to know for certain that no more packets with timestamp
lower than `T` will arrive.
## Input policies
Synchronization is handled locally on each node, using the input policy
specified by the node.
The default input policy, defined by [`DefaultInputStreamHandler`], provides
deterministic synchronization of inputs, with the following guarantees:
* If packets with the same timestamp are provided on multiple input streams,
they will always be processed together regardless of their arrival order in
real time.
* Input sets are processed in strictly ascending timestamp order.
* No packets are dropped, and the processing is fully deterministic.
* The node becomes ready to process data as soon as possible given the
guarantees above.
Note: An important consequence of this is that if the calculator always uses the
current input timestamp when outputting packets, the output will inherently obey
the monotonically increasing timestamp requirement.
Warning: On the other hand, it is not guaranteed that an input packet will
always be available for all streams.
To explain how it works, we need to introduce the definition of a settled
timestamp. We say that a timestamp in a stream is *settled* if it lower than the
timestamp bound. In other words, a timestamp is settled for a stream once the
state of the input at that timestamp is irrevocably known: either there is a
packet, or there is the certainty that a packet with that timestamp will not
arrive.
Note: For this reason, MediaPipe also allows a stream producer to explicitly
advance the timestamp bound farther that what the last packet implies, i.e. to
provide a tighter bound. This can allow the downstream nodes to settle their
inputs sooner.
A timestamp is settled across multiple streams if it is settled on each of those
streams. Furthermore, if a timestamp is settled it implies that all previous
timestamps are also settled. Thus settled timestamps can be processed
deterministically in ascending order.
Given this definition, a calculator with the default input policy is ready if
there is a timestamp which is settled across all input streams and contains a
packet on at least one input stream. The input policy provides all available
packets for a settled timestamp as a single *input set* to the calculator.
One consequence of this deterministic behavior is that, for nodes with multiple
input streams, there can be a theoretically unbounded wait for a timestamp to be
settled, and an unbounded number of packets can be buffered in the meantime.
(Consider a node with two input streams, one of which keeps sending packets
while the other sends nothing and does not advance the bound.)
Therefore, we also provide for custom input policies: for example, splitting the
inputs in different synchronization sets defined by
[`SyncSetInputStreamHandler`], or avoiding synchronization altogether and
processing inputs immediately as they arrive defined by
[`ImmediateInputStreamHandler`].
## Flow control
There are two main flow control mechanisms. A backpressure mechanism throttles
the execution of upstream nodes when the packets buffered on a stream reach a
(configurable) limit defined by [`CalculatorGraphConfig::max_queue_size`]. This
mechanism maintains deterministic behavior, and includes a deadlock avoidance
system that relaxes configured limits when needed.
The second system consists of inserting special nodes which can drop packets
according to real-time constraints (typically using custom input policies)
defined by [`FlowLimiterCalculator`]. For example, a common pattern places a
flow-control node at the input of a subgraph, with a loopback connection from
the final output to the flow-control node. The flow-control node is thus able to
keep track of how many timestamps are being processed in the downstream graph,
and drop packets if this count hits a (configurable) limit; and since packets
are dropped upstream, we avoid the wasted work that would result from partially
processing a timestamp and then dropping packets between intermediate stages.
This calculator-based approach gives the graph author control of where packets
can be dropped, and allows flexibility in adapting and customizing the graph’s
behavior depending on resource constraints.
[`CalculatorBase`]: https://github.com/google/mediapipe/tree/master/mediapipe/framework/calculator_base.h
[`DefaultInputStreamHandler`]: https://github.com/google/mediapipe/tree/master/mediapipe/framework/stream_handler/default_input_stream_handler.h
[`SyncSetInputStreamHandler`]: https://github.com/google/mediapipe/tree/master/mediapipe/framework/stream_handler/sync_set_input_stream_handler.h
[`ImmediateInputStreamHandler`]: https://github.com/google/mediapipe/tree/master/mediapipe/framework/stream_handler/immediate_input_stream_handler.h
[`CalculatorGraphConfig::max_queue_size`]: https://github.com/google/mediapipe/tree/master/mediapipe/framework/calculator.proto
[`FlowLimiterCalculator`]: https://github.com/google/mediapipe/tree/master/mediapipe/calculators/core/flow_limiter_calculator.cc