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

Project import generated by Copybara.

Browse Source 
GitOrigin-RevId: 19a829ffd755edb43e54d20c0e7b9348512d5108
This commit is contained in:
MediaPipe Team
2022-05-05 19:57:20 +00:00
committed by schmidt-sebastian
parent c6c80c3745
commit 7fb37c80e8
136 changed files with 2572 additions and 555 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mediapipe/framework/BUILD
+7 -1
View File
@@ -234,7 +234,9 @@ cc_library(
"//mediapipe/framework/tool:options_map",
"//mediapipe/framework/tool:packet_generator_wrapper_calculator_cc_proto",
"//mediapipe/framework/tool:tag_map",
"@com_google_absl//absl/container:flat_hash_map",
"@com_google_absl//absl/memory",
"@com_google_absl//absl/strings",
],
)
@@ -348,6 +350,7 @@ cc_library(
"//mediapipe/framework/tool:validate",
"//mediapipe/framework/tool:validate_name",
"//mediapipe/gpu:graph_support",
"//mediapipe/gpu:gpu_service",
"//mediapipe/util:cpu_util",
] + select({
"//conditions:default": ["//mediapipe/gpu:gpu_shared_data_internal"],
@@ -416,7 +419,6 @@ cc_library(
"//mediapipe/framework/tool:status_util",
"//mediapipe/framework/tool:tag_map",
"//mediapipe/framework/tool:validate_name",
"//mediapipe/gpu:graph_support",
"@com_google_absl//absl/base:core_headers",
"@com_google_absl//absl/memory",
"@com_google_absl//absl/status",
@@ -613,7 +615,11 @@ cc_library(
hdrs = ["graph_service.h"],
visibility = [":mediapipe_internal"],
deps = [
":packet",
"//mediapipe/framework/port:status",
"//mediapipe/framework/port:statusor",
"@com_google_absl//absl/base:core_headers",
"@com_google_absl//absl/strings",
],
)
mediapipe/framework/api2/packet.h
+2 -3
View File
@@ -167,7 +167,6 @@ struct IsCompatibleType<V, OneOf<U...>>
template <typename T>
inline Packet<T> PacketBase::As() const {
if (!payload_) return Packet<T>().At(timestamp_);
packet_internal::Holder<T>* typed_payload = payload_->As<T>();
internal::CheckCompatibleType(*payload_, internal::Wrap<T>{});
return Packet<T>(payload_).At(timestamp_);
}
@@ -217,8 +216,8 @@ class Packet : public Packet<internal::Generic> {
const T& operator*() const { return Get(); }
const T* operator->() const { return &Get(); }
template <typename U>
T GetOr(U&& v) const {
template <typename U, typename TT = T>
std::enable_if_t<!std::is_abstract_v<TT>, TT> GetOr(U&& v) const {
return IsEmpty() ? static_cast<T>(absl::forward<U>(v)) : **this;
}
mediapipe/framework/api2/packet_nc.cc
+6 -2
View File
@@ -4,11 +4,15 @@ namespace api2 {
namespace {
#if defined(TEST_NO_ASSIGN_WRONG_PACKET_TYPE)
void AssignWrongPacketType() { Packet<int> p = MakePacket<float>(1.0); }
int AssignWrongPacketType() {
Packet<int> p = MakePacket<float>(1.0);
return *p;
}
#elif defined(TEST_NO_ASSIGN_GENERIC_TO_SPECIFIC)
void AssignWrongPacketType() {
int AssignWrongPacketType() {
Packet<> p = MakePacket<float>(1.0);
Packet<int> p2 = p;
return *p2;
}
#endif
mediapipe/framework/api2/packet_test.cc
+17
View File
@@ -264,6 +264,23 @@ TEST(PacketTest, Polymorphism) {
EXPECT_EQ((**mutable_base).name(), "Derived");
}
class AbstractBase {
public:
virtual ~AbstractBase() = default;
virtual absl::string_view name() const = 0;
};
class ConcreteDerived : public AbstractBase {
public:
absl::string_view name() const override { return "ConcreteDerived"; }
};
TEST(PacketTest, PolymorphismAbstract) {
Packet<AbstractBase> base =
PacketAdopting<AbstractBase>(absl::make_unique<ConcreteDerived>());
EXPECT_EQ(base->name(), "ConcreteDerived");
}
} // namespace
} // namespace api2
} // namespace mediapipe
mediapipe/framework/api2/port_test.cc
+11
View File
@@ -40,6 +40,17 @@ TEST(PortTest, DeletedCopyConstructorInput) {
EXPECT_EQ(std::string(kSideOutputPort.Tag()), "SIDE_OUTPUT");
}
class AbstractBase {
public:
virtual ~AbstractBase() = default;
virtual absl::string_view name() const = 0;
};
TEST(PortTest, Abstract) {
static constexpr Input<AbstractBase> kInputPort{"INPUT"};
EXPECT_EQ(std::string(kInputPort.Tag()), "INPUT");
}
} // namespace
} // namespace api2
} // namespace mediapipe
mediapipe/framework/calculator_contract.h
+10 -5
View File
@@ -21,6 +21,8 @@
#include <typeindex>
// TODO: Move protos in another CL after the C++ code migration.
#include "absl/container/flat_hash_map.h"
#include "absl/strings/string_view.h"
#include "mediapipe/framework/calculator.pb.h"
#include "mediapipe/framework/graph_service.h"
#include "mediapipe/framework/mediapipe_options.pb.h"
@@ -147,7 +149,7 @@ class CalculatorContract {
bool IsOptional() const { return optional_; }
private:
GraphServiceBase service_;
const GraphServiceBase& service_;
bool optional_ = false;
};
@@ -156,9 +158,12 @@ class CalculatorContract {
return it->second;
}
const std::map<std::string, GraphServiceRequest>& ServiceRequests() const {
return service_requests_;
}
// A GraphService's key is always a static constant, so we can use string_view
// as the key type without lifetime issues.
using ServiceReqMap =
absl::flat_hash_map<absl::string_view, GraphServiceRequest>;
const ServiceReqMap& ServiceRequests() const { return service_requests_; }
private:
template <class T>
@@ -180,7 +185,7 @@ class CalculatorContract {
std::string input_stream_handler_;
MediaPipeOptions input_stream_handler_options_;
std::string node_name_;
std::map<std::string, GraphServiceRequest> service_requests_;
ServiceReqMap service_requests_;
bool process_timestamps_ = false;
TimestampDiff timestamp_offset_ = TimestampDiff::Unset();
mediapipe/framework/calculator_graph.cc
+97 -58
View File
@@ -226,6 +226,16 @@ absl::Status CalculatorGraph::InitializeStreams() {
return absl::OkStatus();
}
// Hack for backwards compatibility with ancient GPU calculators. Can it
// be retired yet?
static void MaybeFixupLegacyGpuNodeContract(CalculatorNode& node) {
#if !MEDIAPIPE_DISABLE_GPU
if (node.Contract().InputSidePackets().HasTag(kGpuSharedTagName)) {
const_cast<CalculatorContract&>(node.Contract()).UseService(kGpuService);
}
#endif // !MEDIAPIPE_DISABLE_GPU
}
absl::Status CalculatorGraph::InitializeCalculatorNodes() {
// Check if the user has specified a maximum queue size for an input stream.
max_queue_size_ = validated_graph_->Config().max_queue_size();
@@ -246,6 +256,7 @@ absl::Status CalculatorGraph::InitializeCalculatorNodes() {
validated_graph_.get(), node_ref, input_stream_managers_.get(),
output_stream_managers_.get(), output_side_packets_.get(),
&buffer_size_hint, profiler_);
MaybeFixupLegacyGpuNodeContract(*nodes_.back());
if (buffer_size_hint > 0) {
max_queue_size_ = std::max(max_queue_size_, buffer_size_hint);
}
@@ -283,6 +294,7 @@ absl::Status CalculatorGraph::InitializePacketGeneratorNodes(
validated_graph_.get(), node_ref, input_stream_managers_.get(),
output_stream_managers_.get(), output_side_packets_.get(),
&buffer_size_hint, profiler_);
MaybeFixupLegacyGpuNodeContract(*nodes_.back());
if (!result.ok()) {
// Collect as many errors as we can before failing.
errors.push_back(result);
@@ -495,9 +507,8 @@ absl::StatusOr<Packet> CalculatorGraph::GetOutputSidePacket(
<< "\" because it doesn't exist.";
}
Packet output_packet;
if (scheduler_.IsTerminated()) {
// Side-packets from calculators can be retrieved only after the graph is
// done.
if (!output_side_packets_[side_packet_index].GetPacket().IsEmpty() ||
scheduler_.IsTerminated()) {
output_packet = output_side_packets_[side_packet_index].GetPacket();
}
if (output_packet.IsEmpty()) {
@@ -546,6 +557,7 @@ absl::Status CalculatorGraph::StartRun(
#if !MEDIAPIPE_DISABLE_GPU
absl::Status CalculatorGraph::SetGpuResources(
std::shared_ptr<::mediapipe::GpuResources> resources) {
RET_CHECK_NE(resources, nullptr);
auto gpu_service = service_manager_.GetServiceObject(kGpuService);
RET_CHECK_EQ(gpu_service, nullptr)
<< "The GPU resources have already been configured.";
@@ -557,68 +569,89 @@ std::shared_ptr<::mediapipe::GpuResources> CalculatorGraph::GetGpuResources()
return service_manager_.GetServiceObject(kGpuService);
}
absl::StatusOr<std::map<std::string, Packet>> CalculatorGraph::PrepareGpu(
static Packet GetLegacyGpuSharedSidePacket(
const std::map<std::string, Packet>& side_packets) {
std::map<std::string, Packet> additional_side_packets;
bool update_sp = false;
bool uses_gpu = false;
for (const auto& node : nodes_) {
if (node->UsesGpu()) {
uses_gpu = true;
break;
}
auto legacy_sp_iter = side_packets.find(kGpuSharedSidePacketName);
if (legacy_sp_iter == side_packets.end()) return {};
// Note that, because of b/116875321, the legacy side packet may be set but
// empty. But it's ok, because here we return an empty packet to indicate the
// missing case anyway.
return legacy_sp_iter->second;
}
absl::Status CalculatorGraph::MaybeSetUpGpuServiceFromLegacySidePacket(
Packet legacy_sp) {
if (legacy_sp.IsEmpty()) return absl::OkStatus();
auto gpu_resources = service_manager_.GetServiceObject(kGpuService);
if (gpu_resources) {
LOG(WARNING)
<< "::mediapipe::GpuSharedData provided as a side packet while the "
<< "graph already had one; ignoring side packet";
return absl::OkStatus();
}
if (uses_gpu) {
auto gpu_resources = service_manager_.GetServiceObject(kGpuService);
gpu_resources = legacy_sp.Get<::mediapipe::GpuSharedData*>()->gpu_resources;
return service_manager_.SetServiceObject(kGpuService, gpu_resources);
}
auto legacy_sp_iter = side_packets.find(kGpuSharedSidePacketName);
// Workaround for b/116875321: CalculatorRunner provides an empty packet,
// instead of just leaving it unset.
bool has_legacy_sp = legacy_sp_iter != side_packets.end() &&
!legacy_sp_iter->second.IsEmpty();
if (gpu_resources) {
if (has_legacy_sp) {
LOG(WARNING)
<< "::mediapipe::GpuSharedData provided as a side packet while the "
<< "graph already had one; ignoring side packet";
}
update_sp = true;
} else {
if (has_legacy_sp) {
gpu_resources =
legacy_sp_iter->second.Get<::mediapipe::GpuSharedData*>()
->gpu_resources;
} else {
ASSIGN_OR_RETURN(gpu_resources, ::mediapipe::GpuResources::Create());
update_sp = true;
}
MP_RETURN_IF_ERROR(
service_manager_.SetServiceObject(kGpuService, gpu_resources));
}
// Create or replace the legacy side packet if needed.
if (update_sp) {
legacy_gpu_shared_.reset(new ::mediapipe::GpuSharedData(gpu_resources));
additional_side_packets[kGpuSharedSidePacketName] =
MakePacket<::mediapipe::GpuSharedData*>(legacy_gpu_shared_.get());
}
// Set up executors.
for (auto& node : nodes_) {
if (node->UsesGpu()) {
MP_RETURN_IF_ERROR(gpu_resources->PrepareGpuNode(node.get()));
}
}
for (const auto& name_executor : gpu_resources->GetGpuExecutors()) {
MP_RETURN_IF_ERROR(
SetExecutorInternal(name_executor.first, name_executor.second));
}
std::map<std::string, Packet> CalculatorGraph::MaybeCreateLegacyGpuSidePacket(
Packet legacy_sp) {
std::map<std::string, Packet> additional_side_packets;
auto gpu_resources = service_manager_.GetServiceObject(kGpuService);
if (gpu_resources &&
(legacy_sp.IsEmpty() ||
legacy_sp.Get<::mediapipe::GpuSharedData*>()->gpu_resources !=
gpu_resources)) {
legacy_gpu_shared_ =
absl::make_unique<mediapipe::GpuSharedData>(gpu_resources);
additional_side_packets[kGpuSharedSidePacketName] =
MakePacket<::mediapipe::GpuSharedData*>(legacy_gpu_shared_.get());
}
return additional_side_packets;
}
static bool UsesGpu(const CalculatorNode& node) {
return node.Contract().ServiceRequests().contains(kGpuService.key);
}
absl::Status CalculatorGraph::PrepareGpu() {
auto gpu_resources = service_manager_.GetServiceObject(kGpuService);
if (!gpu_resources) return absl::OkStatus();
// Set up executors.
for (auto& node : nodes_) {
if (UsesGpu(*node)) {
MP_RETURN_IF_ERROR(gpu_resources->PrepareGpuNode(node.get()));
}
}
for (const auto& name_executor : gpu_resources->GetGpuExecutors()) {
MP_RETURN_IF_ERROR(
SetExecutorInternal(name_executor.first, name_executor.second));
}
return absl::OkStatus();
}
#endif // !MEDIAPIPE_DISABLE_GPU
absl::Status CalculatorGraph::PrepareServices() {
for (const auto& node : nodes_) {
for (const auto& [key, request] : node->Contract().ServiceRequests()) {
auto packet = service_manager_.GetServicePacket(request.Service());
if (!packet.IsEmpty()) continue;
auto packet_or = request.Service().CreateDefaultObject();
if (packet_or.ok()) {
MP_RETURN_IF_ERROR(service_manager_.SetServicePacket(
request.Service(), std::move(packet_or).value()));
} else if (request.IsOptional()) {
continue;
} else {
return absl::InternalError(absl::StrCat(
"Service \"", request.Service().key, "\", required by node ",
node->DebugName(), ", was not provided and cannot be created: ",
std::move(packet_or).status().message()));
}
}
}
return absl::OkStatus();
}
absl::Status CalculatorGraph::PrepareForRun(
const std::map<std::string, Packet>& extra_side_packets,
const std::map<std::string, Packet>& stream_headers) {
@@ -637,7 +670,13 @@ absl::Status CalculatorGraph::PrepareForRun(
std::map<std::string, Packet> additional_side_packets;
#if !MEDIAPIPE_DISABLE_GPU
ASSIGN_OR_RETURN(additional_side_packets, PrepareGpu(extra_side_packets));
auto legacy_sp = GetLegacyGpuSharedSidePacket(extra_side_packets);
MP_RETURN_IF_ERROR(MaybeSetUpGpuServiceFromLegacySidePacket(legacy_sp));
#endif // !MEDIAPIPE_DISABLE_GPU
MP_RETURN_IF_ERROR(PrepareServices());
#if !MEDIAPIPE_DISABLE_GPU
MP_RETURN_IF_ERROR(PrepareGpu());
additional_side_packets = MaybeCreateLegacyGpuSidePacket(legacy_sp);
#endif // !MEDIAPIPE_DISABLE_GPU
const std::map<std::string, Packet>* input_side_packets;
mediapipe/framework/calculator_graph.h
+20 -10
View File
@@ -165,10 +165,13 @@ class CalculatorGraph {
StatusOrPoller AddOutputStreamPoller(const std::string& stream_name,
bool observe_timestamp_bounds = false);
// Gets output side packet by name after the graph is done. However, base
// packets (generated by PacketGenerators) can be retrieved before
// graph is done. Returns error if the graph is still running (for non-base
// packets) or the output side packet is not found or empty.
// Gets output side packet by name. The output side packet can be successfully
// retrevied in one of the following situations:
// - The graph is done.
// - The output side packet has been generated by a calculator and the graph
// is currently idle.
// - The side packet is a base packet generated by a PacketGenerator.
// Returns error if the the output side packet is not found or empty.
absl::StatusOr<Packet> GetOutputSidePacket(const std::string& packet_name);
// Runs the graph after adding the given extra input side packets. All
@@ -367,13 +370,8 @@ class CalculatorGraph {
std::shared_ptr<GpuResources> GetGpuResources() const;
absl::Status SetGpuResources(std::shared_ptr<GpuResources> resources);
// Helper for PrepareForRun. If it returns a non-empty map, those packets
// must be added to the existing side packets, replacing existing values
// that have the same key.
absl::StatusOr<std::map<std::string, Packet>> PrepareGpu(
const std::map<std::string, Packet>& side_packets);
#endif // !MEDIAPIPE_DISABLE_GPU
template <typename T>
absl::Status SetServiceObject(const GraphService<T>& service,
std::shared_ptr<T> object) {
@@ -495,6 +493,18 @@ class CalculatorGraph {
const std::map<std::string, Packet>& extra_side_packets,
const std::map<std::string, Packet>& stream_headers);
absl::Status PrepareServices();
#if !MEDIAPIPE_DISABLE_GPU
absl::Status MaybeSetUpGpuServiceFromLegacySidePacket(Packet legacy_sp);
// Helper for PrepareForRun. If it returns a non-empty map, those packets
// must be added to the existing side packets, replacing existing values
// that have the same key.
std::map<std::string, Packet> MaybeCreateLegacyGpuSidePacket(
Packet legacy_sp);
absl::Status PrepareGpu();
#endif // !MEDIAPIPE_DISABLE_GPU
// Cleans up any remaining state after the run and returns any errors that may
// have occurred during the run. Called after the scheduler has terminated.
absl::Status FinishRun();
mediapipe/framework/calculator_graph_side_packet_test.cc
+23 -4
View File
@@ -732,11 +732,12 @@ TEST(CalculatorGraph, GetOutputSidePacket) {
status_or_packet = graph.GetOutputSidePacket("unknown");
EXPECT_FALSE(status_or_packet.ok());
EXPECT_EQ(absl::StatusCode::kNotFound, status_or_packet.status().code());
// Should return UNAVAILABLE before graph is done for valid non-base
// packets.
// Should return the packet after the graph becomes idle.
MP_ASSERT_OK(graph.WaitUntilIdle());
status_or_packet = graph.GetOutputSidePacket("num_of_packets");
EXPECT_FALSE(status_or_packet.ok());
EXPECT_EQ(absl::StatusCode::kUnavailable, status_or_packet.status().code());
MP_ASSERT_OK(status_or_packet);
EXPECT_EQ(max_count, status_or_packet.value().Get<int>());
EXPECT_EQ(Timestamp::Unset(), status_or_packet.value().Timestamp());
// Should stil return a base even before graph is done.
status_or_packet = graph.GetOutputSidePacket("output_uint64");
MP_ASSERT_OK(status_or_packet);
@@ -896,5 +897,23 @@ TEST(CalculatorGraph, GeneratorAfterCalculatorProcess) {
}
}
TEST(CalculatorGraph, GetOutputSidePacketAfterCalculatorIsOpened) {
CalculatorGraph graph;
CalculatorGraphConfig config =
mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(R"pb(
node {
calculator: "IntegerOutputSidePacketCalculator"
output_side_packet: "offset"
}
)pb");
MP_ASSERT_OK(graph.Initialize(config));
MP_ASSERT_OK(graph.StartRun({}));
// Must be called to ensure that the calculator is opened.
MP_ASSERT_OK(graph.WaitUntilIdle());
absl::StatusOr<Packet> status_or_packet = graph.GetOutputSidePacket("offset");
MP_ASSERT_OK(status_or_packet);
EXPECT_EQ(1, status_or_packet.value().Get<int>());
}
} // namespace
} // namespace mediapipe
mediapipe/framework/calculator_node.cc
+1 -7
View File
@@ -46,7 +46,6 @@
#include "mediapipe/framework/tool/status_util.h"
#include "mediapipe/framework/tool/tag_map.h"
#include "mediapipe/framework/tool/validate_name.h"
#include "mediapipe/gpu/graph_support.h"
namespace mediapipe {
@@ -155,11 +154,6 @@ absl::Status CalculatorNode::Initialize(
const CalculatorContract& contract = node_type_info_->Contract();
uses_gpu_ =
node_type_info_->InputSidePacketTypes().HasTag(kGpuSharedTagName) ||
ContainsKey(node_type_info_->Contract().ServiceRequests(),
kGpuService.key);
// TODO Propagate types between calculators when SetAny is used.
MP_RETURN_IF_ERROR(InitializeOutputSidePackets(
@@ -397,7 +391,7 @@ absl::Status CalculatorNode::PrepareForRun(
std::move(schedule_callback), error_callback);
output_stream_handler_->PrepareForRun(error_callback);
const auto& contract = node_type_info_->Contract();
const auto& contract = Contract();
input_side_packet_types_ = RemoveOmittedPacketTypes(
contract.InputSidePackets(), all_side_packets, validated_graph_);
MP_RETURN_IF_ERROR(input_side_packet_handler_.PrepareForRun(
mediapipe/framework/calculator_node.h
+6 -6
View File
@@ -195,9 +195,6 @@ class CalculatorNode {
// Called by SchedulerQueue when a node is opened.
void NodeOpened() ABSL_LOCKS_EXCLUDED(status_mutex_);
// Returns whether this is a GPU calculator node.
bool UsesGpu() const { return uses_gpu_; }
// Returns the scheduler queue the node is assigned to.
internal::SchedulerQueue* GetSchedulerQueue() const {
return scheduler_queue_;
@@ -234,6 +231,12 @@ class CalculatorNode {
return *calculator_state_;
}
// Returns the node's contract.
// Must not be called before the CalculatorNode is initialized.
const CalculatorContract& Contract() const {
return node_type_info_->Contract();
}
private:
// Sets up the output side packets from the main flat array.
absl::Status InitializeOutputSidePackets(
@@ -363,9 +366,6 @@ class CalculatorNode {
std::unique_ptr<OutputStreamHandler> output_stream_handler_;
// Whether this is a GPU calculator.
bool uses_gpu_ = false;
// True if CleanupAfterRun() needs to call CloseNode().
bool needs_to_close_ = false;
mediapipe/framework/formats/BUILD
+17
View File
@@ -187,6 +187,21 @@ cc_library(
],
)
config_setting(
name = "opencv",
define_values = {
"use_opencv": "true",
},
)
config_setting(
name = "portable_opencv",
define_values = {
"use_portable_opencv": "true",
"use_opencv": "false",
},
)
cc_library(
name = "location",
srcs = ["location.cc"],
@@ -194,6 +209,8 @@ cc_library(
defines = select({
"//conditions:default": [],
"//mediapipe:android": ["MEDIAPIPE_ANDROID_OPENCV"],
":portable_opencv": ["MEDIAPIPE_ANDROID_OPENCV"],
":opencv": [],
}),
visibility = ["//visibility:public"],
deps = [
mediapipe/framework/formats/tensor.h
+3 -1
View File
@@ -76,7 +76,7 @@ class Tensor {
public:
// No resources are allocated here.
enum class ElementType { kNone, kFloat16, kFloat32, kUInt8 };
enum class ElementType { kNone, kFloat16, kFloat32, kUInt8, kInt8 };
struct Shape {
Shape() = default;
Shape(std::initializer_list<int> dimensions) : dims(dimensions) {}
@@ -217,6 +217,8 @@ class Tensor {
return sizeof(float);
case ElementType::kUInt8:
return 1;
case ElementType::kInt8:
return 1;
}
}
int bytes() const { return shape_.num_elements() * element_size(); }
mediapipe/framework/graph_service.h
+65 -3
View File
@@ -16,6 +16,12 @@
#define MEDIAPIPE_FRAMEWORK_GRAPH_SERVICE_H_
#include <memory>
#include <type_traits>
#include <utility>
#include "absl/strings/str_cat.h"
#include "mediapipe/framework/packet.h"
#include "mediapipe/framework/port/status.h"
namespace mediapipe {
@@ -27,18 +33,74 @@ namespace mediapipe {
// IMPORTANT: this is an experimental API. Get in touch with the MediaPipe team
// if you want to use it. In most cases, you should use a side packet instead.
struct GraphServiceBase {
class GraphServiceBase {
public:
// TODO: fix services for which default init is broken, remove
// this setting.
enum DefaultInitSupport {
kAllowDefaultInitialization,
kDisallowDefaultInitialization
};
constexpr GraphServiceBase(const char* key) : key(key) {}
virtual ~GraphServiceBase() = default;
inline virtual absl::StatusOr<Packet> CreateDefaultObject() const {
return DefaultInitializationUnsupported();
}
const char* key;
protected:
absl::Status DefaultInitializationUnsupported() const {
return absl::UnimplementedError(absl::StrCat(
"Graph service '", key, "' does not support default initialization"));
}
};
template <typename T>
struct GraphService : public GraphServiceBase {
class GraphService : public GraphServiceBase {
public:
using type = T;
using packet_type = std::shared_ptr<T>;
constexpr GraphService(const char* key) : GraphServiceBase(key) {}
constexpr GraphService(const char* my_key, DefaultInitSupport default_init =
kDisallowDefaultInitialization)
: GraphServiceBase(my_key), default_init_(default_init) {}
absl::StatusOr<Packet> CreateDefaultObject() const override {
if (default_init_ != kAllowDefaultInitialization) {
return DefaultInitializationUnsupported();
}
auto packet_or = CreateDefaultObjectInternal();
if (packet_or.ok()) {
return MakePacket<std::shared_ptr<T>>(std::move(packet_or).value());
} else {
return packet_or.status();
}
}
private:
absl::StatusOr<std::shared_ptr<T>> CreateDefaultObjectInternal() const {
auto call_create = [](auto x) -> decltype(decltype(x)::type::Create()) {
return decltype(x)::type::Create();
};
if constexpr (std::is_invocable_r_v<absl::StatusOr<std::shared_ptr<T>>,
decltype(call_create), type_tag<T>>) {
return T::Create();
}
if constexpr (std::is_default_constructible_v<T>) {
return std::make_shared<T>();
}
return DefaultInitializationUnsupported();
}
template <class U>
struct type_tag {
using type = U;
};
DefaultInitSupport default_init_;
};
template <typename T>
mediapipe/framework/graph_service_manager.h
+2
View File
@@ -35,6 +35,8 @@ class GraphServiceManager {
Packet GetServicePacket(const GraphServiceBase& service) const;
std::map<std::string, Packet> service_packets_;
friend class CalculatorGraph;
};
} // namespace mediapipe
mediapipe/framework/graph_service_manager_test.cc
+3 -5
View File
@@ -6,11 +6,13 @@
#include "mediapipe/framework/port/status_matchers.h"
namespace mediapipe {
namespace {
const GraphService<int> kIntService("mediapipe::IntService");
} // namespace
TEST(GraphServiceManager, SetGetServiceObject) {
GraphServiceManager service_manager;
constexpr GraphService<int> kIntService("mediapipe::IntService");
EXPECT_EQ(service_manager.GetServiceObject(kIntService), nullptr);
MP_EXPECT_OK(service_manager.SetServiceObject(kIntService,
@@ -22,8 +24,6 @@ TEST(GraphServiceManager, SetGetServiceObject) {
TEST(GraphServiceManager, SetServicePacket) {
GraphServiceManager service_manager;
constexpr GraphService<int> kIntService("mediapipe::IntService");
MP_EXPECT_OK(service_manager.SetServicePacket(
kIntService,
mediapipe::MakePacket<std::shared_ptr<int>>(std::make_shared<int>(100))));
@@ -36,8 +36,6 @@ TEST(GraphServiceManager, ServicePackets) {
EXPECT_TRUE(service_manager.ServicePackets().empty());
constexpr GraphService<int> kIntService("mediapipe::IntService");
MP_EXPECT_OK(service_manager.SetServiceObject(kIntService,
std::make_shared<int>(100)));
mediapipe/framework/graph_service_test.cc
+7
View File
@@ -150,5 +150,12 @@ TEST_F(GraphServiceTest, OptionalIsAvailable) {
EXPECT_EQ(PacketValues<int>(output_packets_), (std::vector<int>{108}));
}
TEST_F(GraphServiceTest, CreateDefault) {
EXPECT_FALSE(kTestService.CreateDefaultObject().ok());
MP_EXPECT_OK(kAnotherService.CreateDefaultObject());
EXPECT_FALSE(kNoDefaultService.CreateDefaultObject().ok());
MP_EXPECT_OK(kNeedsCreateService.CreateDefaultObject());
}
} // namespace
} // namespace mediapipe
mediapipe/framework/input_stream_handler.cc
+6 -3
View File
@@ -50,15 +50,18 @@ absl::Status InputStreamHandler::SetupInputShards(
return absl::OkStatus();
}
std::vector<std::pair<std::string, int>>
std::vector<std::tuple<std::string, int, int, Timestamp>>
InputStreamHandler::GetMonitoringInfo() {
std::vector<std::pair<std::string, int>> monitoring_info_vector;
std::vector<std::tuple<std::string, int, int, Timestamp>>
monitoring_info_vector;
for (auto& stream : input_stream_managers_) {
if (!stream) {
continue;
}
monitoring_info_vector.emplace_back(
std::pair<std::string, int>(stream->Name(), stream->QueueSize()));
std::tuple<std::string, int, int, Timestamp>(
stream->Name(), stream->QueueSize(), stream->NumPacketsAdded(),
stream->MinTimestampOrBound(nullptr)));
}
return monitoring_info_vector;
}
mediapipe/framework/input_stream_handler.h
+1 -1
View File
@@ -94,7 +94,7 @@ class InputStreamHandler {
// Returns a vector of pairs of stream name and queue size for monitoring
// purpose.
std::vector<std::pair<std::string, int>> GetMonitoringInfo();
std::vector<std::tuple<std::string, int, int, Timestamp>> GetMonitoringInfo();
// Resets the input stream handler and its underlying input streams for
// another run of the graph.
mediapipe/framework/input_stream_manager.cc
+5
View File
@@ -329,6 +329,11 @@ Packet InputStreamManager::PopQueueHead(bool* stream_is_done) {
return packet;
}
int InputStreamManager::NumPacketsAdded() const {
absl::MutexLock lock(&stream_mutex_);
return num_packets_added_;
}
int InputStreamManager::QueueSize() const {
absl::MutexLock lock(&stream_mutex_);
return static_cast<int>(queue_.size());
mediapipe/framework/input_stream_manager.h
+7 -2
View File
@@ -87,12 +87,14 @@ class InputStreamManager {
// Timestamp::PostStream(), the packet must be the only packet in the
// stream.
// Violation of any of these conditions causes an error status.
absl::Status AddPackets(const std::list<Packet>& container, bool* notify);
absl::Status AddPackets(const std::list<Packet>& container, bool* notify)
ABSL_LOCKS_EXCLUDED(stream_mutex_);
// Move a list of timestamped packets. Sets "notify" to true if the queue
// becomes non-empty. Does nothing if the input stream is closed. After the
// move, all packets in the container must be empty.
absl::Status MovePackets(std::list<Packet>* container, bool* notify);
absl::Status MovePackets(std::list<Packet>* container, bool* notify)
ABSL_LOCKS_EXCLUDED(stream_mutex_);
// Closes the input stream. This function can be called multiple times.
void Close() ABSL_LOCKS_EXCLUDED(stream_mutex_);
@@ -140,6 +142,9 @@ class InputStreamManager {
// Timestamp::Done() after the pop.
Packet PopQueueHead(bool* stream_is_done) ABSL_LOCKS_EXCLUDED(stream_mutex_);
// Returns the number of packets in the queue.
int NumPacketsAdded() const ABSL_LOCKS_EXCLUDED(stream_mutex_);
// Returns the number of packets in the queue.
int QueueSize() const ABSL_LOCKS_EXCLUDED(stream_mutex_);
mediapipe/framework/input_stream_manager_test.cc
+2
View File
@@ -767,6 +767,7 @@ TEST_F(InputStreamManagerTest, QueueSizeTest) {
EXPECT_EQ(3, num_packets_dropped_);
EXPECT_TRUE(input_stream_manager_->IsEmpty());
EXPECT_FALSE(stream_is_done_);
EXPECT_EQ(3, input_stream_manager_->NumPacketsAdded());
packets.clear();
packets.push_back(MakePacket<std::string>("packet 4").At(Timestamp(60)));
@@ -776,6 +777,7 @@ TEST_F(InputStreamManagerTest, QueueSizeTest) {
input_stream_manager_->AddPackets(packets, &notify_)); // Notification
EXPECT_FALSE(input_stream_manager_->IsEmpty());
EXPECT_TRUE(notify_);
EXPECT_EQ(5, input_stream_manager_->NumPacketsAdded());
expected_queue_becomes_full_count_ = 2;
expected_queue_becomes_not_full_count_ = 1;
mediapipe/framework/mediapipe_cc_test.bzl
+2
View File
@@ -12,6 +12,8 @@ def mediapipe_cc_test(
timeout = None,
args = [],
additional_deps = DEFAULT_ADDITIONAL_TEST_DEPS,
platforms = ["linux", "android", "ios", "wasm"],
exclude_platforms = None,
# ios_unit_test arguments
ios_minimum_os_version = "9.0",
# android_cc_test arguments
mediapipe/framework/port/BUILD
+1 -2
View File
@@ -412,8 +412,7 @@ cc_library(
name = "status_matchers",
testonly = 1,
hdrs = ["status_matchers.h"],
# Use this library through "mediapipe/framework/port:gtest_main".
visibility = ["//mediapipe/framework/port:__pkg__"],
visibility = ["//visibility:private"],
deps = [
":status",
"@com_google_googletest//:gtest",
mediapipe/framework/test_service.cc
+8 -2
View File
@@ -16,8 +16,14 @@
namespace mediapipe {
const GraphService<TestServiceObject> kTestService("test_service");
const GraphService<int> kAnotherService("another_service");
const GraphService<TestServiceObject> kTestService(
"test_service", GraphServiceBase::kDisallowDefaultInitialization);
const GraphService<int> kAnotherService(
"another_service", GraphServiceBase::kAllowDefaultInitialization);
const GraphService<NoDefaultConstructor> kNoDefaultService(
"no_default_service", GraphServiceBase::kAllowDefaultInitialization);
const GraphService<NeedsCreateMethod> kNeedsCreateService(
"needs_create_service", GraphServiceBase::kAllowDefaultInitialization);
absl::Status TestServiceCalculator::GetContract(CalculatorContract* cc) {
cc->Inputs().Index(0).Set<int>();
mediapipe/framework/test_service.h
+18
View File
@@ -16,6 +16,7 @@
#define MEDIAPIPE_FRAMEWORK_TEST_SERVICE_H_
#include "mediapipe/framework/calculator_framework.h"
#include "mediapipe/framework/graph_service.h"
namespace mediapipe {
@@ -24,6 +25,23 @@ using TestServiceObject = std::map<std::string, int>;
extern const GraphService<TestServiceObject> kTestService;
extern const GraphService<int> kAnotherService;
class NoDefaultConstructor {
public:
NoDefaultConstructor() = delete;
};
extern const GraphService<NoDefaultConstructor> kNoDefaultService;
class NeedsCreateMethod {
public:
static absl::StatusOr<std::shared_ptr<NeedsCreateMethod>> Create() {
return std::shared_ptr<NeedsCreateMethod>(new NeedsCreateMethod());
}
private:
NeedsCreateMethod() = default;
};
extern const GraphService<NeedsCreateMethod> kNeedsCreateService;
// Use a service.
class TestServiceCalculator : public CalculatorBase {
public:
mediapipe/framework/tool/BUILD
+1 -1
View File
@@ -134,7 +134,7 @@ cc_library(
name = "name_util",
srcs = ["name_util.cc"],
hdrs = ["name_util.h"],
visibility = ["//mediapipe/framework:mediapipe_internal"],
visibility = ["//visibility:public"],
deps = [
":validate_name",
"//mediapipe/framework:calculator_cc_proto",
mediapipe/framework/tool/test_util.cc
+2 -2
View File
@@ -225,7 +225,7 @@ std::string GetTestOutputsDir() {
return output_dir;
}
std::string GetTestDataDir(const std::string& package_base_path) {
std::string GetTestDataDir(absl::string_view package_base_path) {
return file::JoinPath(GetTestRootDir(), package_base_path, "testdata/");
}
@@ -270,7 +270,7 @@ absl::StatusOr<std::unique_ptr<ImageFrame>> LoadTestImage(
format, width, height, width * output_channels, data, stbi_image_free);
}
std::unique_ptr<ImageFrame> LoadTestPng(const std::string& path,
std::unique_ptr<ImageFrame> LoadTestPng(absl::string_view path,
ImageFormat::Format format) {
return nullptr;
}
mediapipe/framework/tool/test_util.h
+2 -2
View File
@@ -63,7 +63,7 @@ std::string GetTestFilePath(absl::string_view relative_path);
// directory.
// This handles the different paths where test data ends up when using
// ion_cc_test on various platforms.
std::string GetTestDataDir(const std::string& package_base_path);
std::string GetTestDataDir(absl::string_view package_base_path);
// Loads a binary graph from path. Returns true iff successful.
bool LoadTestGraph(CalculatorGraphConfig* proto, const std::string& path);
@@ -75,7 +75,7 @@ absl::StatusOr<std::unique_ptr<ImageFrame>> LoadTestImage(
// Loads a PNG image from path using the given ImageFormat. Returns nullptr in
// case of failure.
std::unique_ptr<ImageFrame> LoadTestPng(
const std::string& path, ImageFormat::Format format = ImageFormat::SRGBA);
absl::string_view path, ImageFormat::Format format = ImageFormat::SRGBA);
// Returns the luminance image of |original_image|.
// The format of |original_image| must be sRGB or sRGBA.