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

Project import generated by Copybara.

Browse Source 
GitOrigin-RevId: 852dfb05d450167899c0dd5ef7c45622a12e865b
This commit is contained in:
MediaPipe Team
2020-02-10 14:13:25 -08:00
committed by Hadon Nash
parent d144e564d8
commit de4fbc10e6
100 changed files with 1664 additions and 628 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mediapipe/framework/tool/BUILD
+1
View File
@@ -244,6 +244,7 @@ cc_library(
hdrs = ["subgraph_expansion.h"],
visibility = ["//visibility:public"],
deps = [
":name_util",
":tag_map",
"//mediapipe/framework:calculator_cc_proto",
"//mediapipe/framework:packet_generator",
mediapipe/framework/tool/name_util.cc
+25
View File
@@ -68,5 +68,30 @@ std::string GetUnusedSidePacketName(
return candidate;
}
std::string CanonicalNodeName(const CalculatorGraphConfig& graph_config,
int node_id) {
const auto& node_config = graph_config.node(node_id);
std::string node_name = node_config.name().empty() ? node_config.calculator()
: node_config.name();
int count = 0;
int sequence = 0;
for (int i = 0; i < graph_config.node_size(); i++) {
const auto& current_node_config = graph_config.node(i);
std::string current_node_name = current_node_config.name().empty()
? current_node_config.calculator()
: current_node_config.name();
if (node_name == current_node_name) {
++count;
if (i < node_id) {
++sequence;
}
}
}
if (count <= 1) {
return node_name;
}
return absl::StrCat(node_name, "_", sequence + 1);
}
} // namespace tool
} // namespace mediapipe
mediapipe/framework/tool/name_util.h
+46
View File
@@ -31,7 +31,53 @@ std::string GetUnusedSidePacketName(const CalculatorGraphConfig& /*config*/,
std::string GetUnusedNodeName(const CalculatorGraphConfig& config,
const std::string& node_name_base);
// Returns a short unique name for a Node in a CalculatorGraphConfig.
// This is the Node.name (if specified) or the Node.calculator.
// If there are multiple calculators with similar name in the graph, the name
// will be postfixed by "_<COUNT>". For example, in the following graph the node
// names will be as mentiond.
//
// node { // Name will be "CalcA"
// calculator: "CalcA"
// }
// node { // Name will be "NameB"
// calculator: "CalcB"
// name: "NameB"
// }
// node { // Name will be "CalcC_1" due to duplicate "calculator" field.
// calculator: "CalcC"
// }
// node { // Name will be "CalcC_2" due to duplicate "calculator" field.
// calculator: "CalcC"
// }
// node { // Name will be "NameX".
// calculator: "CalcD"
// name: "NameX"
// }
// node { // Name will be "NameY".
// calculator: "CalcD"
// name: "NameY"
// }
// node { // Name will be "NameZ_1". due to "name" field duplicate.
// calculator: "CalcE"
// name: "NameZ"
// }
// node { // Name will be "NameZ_2". due to "name" field duplicate.
// calculator: "CalcF"
// name: "NameZ"
// }
//
// TODO: Update GraphNode.UniqueName in MediaPipe Visualizer to match
// this logic.
// TODO: Fix the edge case mentioned in the bug.
std::string CanonicalNodeName(const CalculatorGraphConfig& graph_config,
int node_id);
} // namespace tool
} // namespace mediapipe
namespace mediapipe {
using ::mediapipe::tool::CanonicalNodeName;
} // namespace mediapipe
#endif // MEDIAPIPE_FRAMEWORK_TOOL_NAME_UTIL_H_
mediapipe/framework/tool/simulation_clock.cc
+6
View File
@@ -15,10 +15,16 @@
#include "mediapipe/framework/tool/simulation_clock.h"
#include "absl/synchronization/mutex.h"
#include "absl/time/time.h"
#include "mediapipe/framework/port/logging.h"
namespace mediapipe {
SimulationClock::~SimulationClock() {
ThreadStart();
ThreadFinish();
}
absl::Time SimulationClock::TimeNow() {
absl::MutexLock l(&time_mutex_);
return time_;
mediapipe/framework/tool/simulation_clock.h
+6 -6
View File
@@ -39,7 +39,7 @@ namespace mediapipe {
class SimulationClock : public mediapipe::Clock {
public:
SimulationClock() {}
~SimulationClock() override {}
~SimulationClock() override;
// Returns the simulated time.
absl::Time TimeNow() override;
@@ -59,9 +59,9 @@ class SimulationClock : public mediapipe::Clock {
protected:
// Queue up wake up waiter.
void SleepInternal(absl::Time wakeup_time)
EXCLUSIVE_LOCKS_REQUIRED(time_mutex_);
ABSL_EXCLUSIVE_LOCKS_REQUIRED(time_mutex_);
// Advances to the next wake up time if no related threads are running.
void TryAdvanceTime() EXCLUSIVE_LOCKS_REQUIRED(time_mutex_);
void TryAdvanceTime() ABSL_EXCLUSIVE_LOCKS_REQUIRED(time_mutex_);
// Represents a thread blocked in SleepUntil.
struct Waiter {
@@ -71,9 +71,9 @@ class SimulationClock : public mediapipe::Clock {
protected:
absl::Mutex time_mutex_;
absl::Time time_ GUARDED_BY(time_mutex_);
std::multimap<absl::Time, Waiter*> waiters_ GUARDED_BY(time_mutex_);
int num_running_ GUARDED_BY(time_mutex_) = 0;
absl::Time time_ ABSL_GUARDED_BY(time_mutex_);
std::multimap<absl::Time, Waiter*> waiters_ ABSL_GUARDED_BY(time_mutex_);
int num_running_ ABSL_GUARDED_BY(time_mutex_) = 0;
};
} // namespace mediapipe
mediapipe/framework/tool/simulation_clock_test.cc
+54
View File
@@ -242,5 +242,59 @@ TEST_F(SimulationClockTest, InFlight) {
ElementsAre(10000, 20000, 40000, 60000, 70000, 100000));
}
// Shows successful destruction of CalculatorGraph, SimulationClockExecutor,
// and SimulationClock. With tsan, this test reveals a race condition unless
// the SimulationClock destructor calls ThreadFinish to waits for all threads.
TEST_F(SimulationClockTest, DestroyClock) {
auto graph_config = ParseTextProtoOrDie<CalculatorGraphConfig>(R"(
node {
calculator: "LambdaCalculator"
input_side_packet: 'callback_0'
output_stream: "input_1"
}
node {
calculator: "LambdaCalculator"
input_side_packet: 'callback_1'
input_stream: "input_1"
output_stream: "output_1"
}
)");
int input_count = 0;
ProcessFunction wait_0 = [&](const InputStreamShardSet& inputs,
OutputStreamShardSet* outputs) {
clock_->Sleep(absl::Microseconds(20000));
if (++input_count < 4) {
outputs->Index(0).AddPacket(
MakePacket<uint64>(input_count).At(Timestamp(input_count)));
return ::mediapipe::OkStatus();
} else {
return tool::StatusStop();
}
};
ProcessFunction wait_1 = [&](const InputStreamShardSet& inputs,
OutputStreamShardSet* outputs) {
clock_->Sleep(absl::Microseconds(30000));
return PassThrough(inputs, outputs);
};
std::vector<Packet> out_packets;
::mediapipe::Status status;
{
CalculatorGraph graph;
auto executor = std::make_shared<SimulationClockExecutor>(4);
clock_ = executor->GetClock().get();
MP_ASSERT_OK(graph.SetExecutor("", executor));
tool::AddVectorSink("output_1", &graph_config, &out_packets);
MP_ASSERT_OK(graph.Initialize(graph_config,
{
{"callback_0", Adopt(new auto(wait_0))},
{"callback_1", Adopt(new auto(wait_1))},
}));
MP_EXPECT_OK(graph.Run());
}
EXPECT_EQ(out_packets.size(), 3);
}
} // namespace
} // namespace mediapipe
mediapipe/framework/tool/subgraph_expansion.cc
+20 -17
View File
@@ -35,6 +35,7 @@
#include "mediapipe/framework/port/status_macros.h"
#include "mediapipe/framework/status_handler.pb.h"
#include "mediapipe/framework/subgraph.h"
#include "mediapipe/framework/tool/name_util.h"
#include "mediapipe/framework/tool/tag_map.h"
namespace mediapipe {
@@ -95,6 +96,13 @@ namespace tool {
config->mutable_output_side_packet()}) {
MP_RETURN_IF_ERROR(TransformStreamNames(streams, transform));
}
std::vector<std::string> node_names(config->node_size());
for (int node_id = 0; node_id < config->node_size(); ++node_id) {
node_names[node_id] = CanonicalNodeName(*config, node_id);
}
for (int node_id = 0; node_id < config->node_size(); ++node_id) {
config->mutable_node(node_id)->set_name(transform(node_names[node_id]));
}
for (auto& node : *config->mutable_node()) {
for (auto* streams :
{node.mutable_input_stream(), node.mutable_output_stream(),
@@ -102,9 +110,6 @@ namespace tool {
node.mutable_output_side_packet()}) {
MP_RETURN_IF_ERROR(TransformStreamNames(streams, transform));
}
if (!node.name().empty()) {
node.set_name(transform(node.name()));
}
}
for (auto& generator : *config->mutable_packet_generator()) {
for (auto* streams : {generator.mutable_input_side_packet(),
@@ -120,21 +125,18 @@ namespace tool {
}
// Adds a prefix to the name of each stream, side packet and node in the
// config. Each call to this method should use a different subgraph_index
// to produce a different numerical prefix. For example:
// 1, { foo, bar } --PrefixNames-> { __sg_1_foo, __sg_1_bar }
// 2, { foo, bar } --PrefixNames-> { __sg_2_foo, __sg_2_bar }
// config. Each call to this method should use a different prefix. For example:
// 1, { foo, bar } --PrefixNames-> { qsg__foo, qsg__bar }
// 2, { foo, bar } --PrefixNames-> { rsg__foo, rsg__bar }
// This means that two copies of the same subgraph will not interfere with
// each other.
static ::mediapipe::Status PrefixNames(int subgraph_index,
static ::mediapipe::Status PrefixNames(std::string prefix,
CalculatorGraphConfig* config) {
// TODO: prefix with subgraph name instead (see cl/157677233
// discussion).
// TODO: since we expand nested subgraphs outside-in, we should
// append the prefix to the existing prefix, if any. This is unimportant
// with the meaningless prefix we use now, but it should be considered
// when prefixing with names.
std::string prefix = absl::StrCat("__sg", subgraph_index, "_");
std::transform(prefix.begin(), prefix.end(), prefix.begin(), ::tolower);
std::replace(prefix.begin(), prefix.end(), '.', '_');
std::replace(prefix.begin(), prefix.end(), ' ', '_');
std::replace(prefix.begin(), prefix.end(), ':', '_');
absl::StrAppend(&prefix, "__");
auto add_prefix = [&prefix](absl::string_view s) {
return absl::StrCat(prefix, s);
};
@@ -271,7 +273,6 @@ static ::mediapipe::Status PrefixNames(int subgraph_index,
graph_registry ? graph_registry : &GraphRegistry::global_graph_registry;
RET_CHECK(config);
auto* nodes = config->mutable_node();
int subgraph_counter = 0;
while (1) {
auto subgraph_nodes_start = std::stable_partition(
nodes->begin(), nodes->end(),
@@ -283,11 +284,13 @@ static ::mediapipe::Status PrefixNames(int subgraph_index,
std::vector<CalculatorGraphConfig> subgraphs;
for (auto it = subgraph_nodes_start; it != nodes->end(); ++it) {
const auto& node = *it;
int node_id = it - nodes->begin();
std::string node_name = CanonicalNodeName(*config, node_id);
MP_RETURN_IF_ERROR(ValidateSubgraphFields(node));
ASSIGN_OR_RETURN(auto subgraph,
graph_registry->CreateByName(config->package(),
node.calculator(), &node));
MP_RETURN_IF_ERROR(PrefixNames(subgraph_counter++, &subgraph));
MP_RETURN_IF_ERROR(PrefixNames(node_name, &subgraph));
MP_RETURN_IF_ERROR(ConnectSubgraphStreams(node, &subgraph));
subgraphs.push_back(subgraph);
}
mediapipe/framework/tool/subgraph_expansion_test.cc
+25 -23
View File
@@ -250,6 +250,7 @@ TEST(SubgraphExpansionTest, TransformNames) {
output_stream: "__sg0_output_1"
}
node {
name: "__sg0_SomeRegularCalculator"
calculator: "SomeRegularCalculator"
input_stream: "__sg0_output_1"
output_stream: "__sg0_output_2"
@@ -438,20 +439,20 @@ TEST(SubgraphExpansionTest, ExpandSubgraphs) {
output_stream: "foo"
}
node {
name: "__sg0_regular_node"
name: "testsubgraph__regular_node"
calculator: "SomeRegularCalculator"
input_stream: "foo"
output_stream: "__sg0_stream_a"
input_side_packet: "__sg0_side"
output_stream: "testsubgraph__stream_a"
input_side_packet: "testsubgraph__side"
}
node {
name: "__sg0_simple_sink"
name: "testsubgraph__simple_sink"
calculator: "SomeSinkCalculator"
input_stream: "__sg0_stream_a"
input_stream: "testsubgraph__stream_a"
}
packet_generator {
packet_generator: "SomePacketGenerator"
output_side_packet: "__sg0_side"
output_side_packet: "testsubgraph__side"
}
)");
MP_EXPECT_OK(tool::ExpandSubgraphs(&supergraph));
@@ -503,23 +504,24 @@ TEST(SubgraphExpansionTest, ExecutorFieldOfNodeInSubgraphPreserved) {
output_stream: "OUT:output"
}
)");
CalculatorGraphConfig expected_graph =
::mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(R"(
input_stream: "input"
executor {
name: "custom_thread_pool"
type: "ThreadPoolExecutor"
options {
[mediapipe.ThreadPoolExecutorOptions.ext] { num_threads: 4 }
}
}
node {
calculator: "PassThroughCalculator"
input_stream: "input"
output_stream: "output"
executor: "custom_thread_pool"
}
)");
CalculatorGraphConfig expected_graph = ::mediapipe::ParseTextProtoOrDie<
CalculatorGraphConfig>(R"(
input_stream: "input"
executor {
name: "custom_thread_pool"
type: "ThreadPoolExecutor"
options {
[mediapipe.ThreadPoolExecutorOptions.ext] { num_threads: 4 }
}
}
node {
calculator: "PassThroughCalculator"
name: "enclosingsubgraph__nodewithexecutorsubgraph__PassThroughCalculator"
input_stream: "input"
output_stream: "output"
executor: "custom_thread_pool"
}
)");
MP_EXPECT_OK(tool::ExpandSubgraphs(&supergraph));
EXPECT_THAT(supergraph, mediapipe::EqualsProto(expected_graph));
}