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

Project import generated by Copybara.

Browse Source 
GitOrigin-RevId: 50714fe28298d7b707eff7304547d89d6ec34a54
This commit is contained in:
MediaPipe Team
2019-11-21 13:20:47 -08:00
committed by mgyong
parent 9437483827
commit 48bcbb115f
60 changed files with 1662 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mediapipe/framework/stream_handler/immediate_input_stream_handler.cc
+4 -3
View File
@@ -115,9 +115,10 @@ void ImmediateInputStreamHandler::FillInputSet(Timestamp input_timestamp,
AddPacketToShard(&input_set->Get(id), std::move(current_packet),
stream_is_done);
} else {
bool empty = false;
bool is_done = stream->MinTimestampOrBound(&empty) == Timestamp::Done();
AddPacketToShard(&input_set->Get(id), Packet(), is_done);
Timestamp bound = stream->MinTimestampOrBound(nullptr);
AddPacketToShard(&input_set->Get(id),
Packet().At(bound.PreviousAllowedInStream()),
bound == Timestamp::Done());
}
}
}
mediapipe/framework/stream_handler/sync_set_input_stream_handler.cc
+24
View File
@@ -56,9 +56,16 @@ class SyncSetInputStreamHandler : public InputStreamHandler {
NodeReadiness GetNodeReadiness(Timestamp* min_stream_timestamp) override;
// Only invoked when associated GetNodeReadiness() returned kReadyForProcess.
// Populates packets for the ready sync-set, and populates timestamp bounds
// for all sync-sets.
void FillInputSet(Timestamp input_timestamp,
InputStreamShardSet* input_set) override;
// Populates timestamp bounds for streams outside the ready sync-set.
void FillInputBounds(Timestamp input_timestamp,
InputStreamShardSet* input_set)
EXCLUSIVE_LOCKS_REQUIRED(mutex_);
private:
absl::Mutex mutex_;
// The ids of each set of inputs.
@@ -181,6 +188,22 @@ NodeReadiness SyncSetInputStreamHandler::GetNodeReadiness(
return NodeReadiness::kNotReady;
}
void SyncSetInputStreamHandler::FillInputBounds(
Timestamp input_timestamp, InputStreamShardSet* input_set) {
for (int i = 0; i < sync_sets_.size(); ++i) {
if (i != ready_sync_set_index_) {
// Set the input streams for the not-ready sync sets.
for (CollectionItemId id : sync_sets_[i]) {
const auto stream = input_stream_managers_.Get(id);
Timestamp bound = stream->MinTimestampOrBound(nullptr);
AddPacketToShard(&input_set->Get(id),
Packet().At(bound.PreviousAllowedInStream()),
bound == Timestamp::Done());
}
}
}
}
void SyncSetInputStreamHandler::FillInputSet(Timestamp input_timestamp,
InputStreamShardSet* input_set) {
// Assume that all current packets are already cleared.
@@ -202,6 +225,7 @@ void SyncSetInputStreamHandler::FillInputSet(Timestamp input_timestamp,
AddPacketToShard(&input_set->Get(id), std::move(current_packet),
stream_is_done);
}
FillInputBounds(input_timestamp, input_set);
ready_sync_set_index_ = -1;
ready_timestamp_ = Timestamp::Done();
}