Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Update streaming in LM Encoding & CB #1377

Merged
merged 7 commits into from
Dec 16, 2024
Merged

Update streaming in LM Encoding & CB #1377

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
3daf0de
Update streaming in LM Encoding
iefode Dec 13, 2024
310461e
Improve code style
iefode Dec 13, 2024
be74340
CB
iefode Dec 13, 2024
61e55a6
Update continuous_batching_impl.cpp
iefode Dec 15, 2024
ad44b32
Revert
iefode Dec 15, 2024
02622c9
end
iefode Dec 16, 2024
dd5ca50
Merge branch 'streaming_lm_encoding' of github.com:iefode/openvino.ge…
iefode Dec 16, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions src/cpp/src/continuous_batching_impl.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -285,9 +285,11 @@ ContinuousBatchingPipeline::ContinuousBatchingImpl::generate(const std::vector<o
}
if (streamer_ptr && generations.at(0)->can_read()) {
std::unordered_map<uint64_t, GenerationOutput> token = generations.at(0).get()->back();
OPENVINO_ASSERT(1 == token.size());
OPENVINO_ASSERT(1 == token.begin()->second.generated_ids.size());
continue_generation = !streamer_ptr->put(token.begin()->second.generated_ids.at(0));
for (const auto& gen_token : token.begin()->second.generated_ids) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

continue_generation assignment is dropped here and hence, we can have abandoned requests with allocated block tables as drop_requests() is not called below.

if (!streamer_ptr->put(gen_token)) {
break;
}
}
}
}

Expand Down
25 changes: 15 additions & 10 deletions src/cpp/src/lm_encoding.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,6 +125,17 @@ std::pair<EncodedResults, int32_t> get_lm_encoded_results(
active_sequence_groups.end(),
get_active_sequence_groups),
active_sequence_groups.end());

auto stream_generated_tokens = [&streamer_ptr, &generations]() {
if (streamer_ptr && generations.at(0).get()->can_read()) {
std::unordered_map<uint64_t, GenerationOutput> token = generations.at(0).get()->back();
for (const auto& gen_token : token.begin()->second.generated_ids) {
if (!streamer_ptr->put(gen_token)) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

copy-pasted incorrect code?
According to documentation

openvino.genai/src/cpp/include/openvino/genai/streamer_base.hpp

Lines 18 to 20 in 9bcadf7

/// @brief put is called every time new token is decoded,
/// @return bool flag to indicate whether generation should be stopped, if return true generation stops
virtual bool put(int64_t token) = 0;

we need to break when put returns true

break;
}
}
}
};

while (active_sequence_groups.size() > 0) {
ilya-lavrenov marked this conversation as resolved.
Show resolved Hide resolved
size_t total_num_tokens = 0;
Expand Down Expand Up @@ -202,13 +213,7 @@ std::pair<EncodedResults, int32_t> get_lm_encoded_results(
raw_perf_counters.m_new_token_times.emplace_back(infer_end);
raw_perf_counters.m_batch_sizes.emplace_back(batch_size);

if (streamer_ptr) {
// stream data from first sequence
int64_t out_token = sequence_groups.at(0).get()->operator[](0)->get_generated_ids().back();
if (streamer_ptr->put(out_token)) {
break;
}
}
stream_generated_tokens();

sampler_output = sampler.sample(active_sequence_groups, m_llm.get_tensor("logits"));

Expand All @@ -218,9 +223,9 @@ std::pair<EncodedResults, int32_t> get_lm_encoded_results(
active_sequence_groups.end());
}

// to stream last token
stream_generated_tokens();
if (streamer_ptr) {
int64_t out_token = sequence_groups.at(0).get()->operator[](0)->get_generated_ids().back();
sbalandi marked this conversation as resolved.
Show resolved Hide resolved
streamer_ptr->put(out_token);
streamer_ptr->end();
}

Expand All @@ -246,4 +251,4 @@ std::pair<EncodedResults, int32_t> get_lm_encoded_results(
}

} // namespace genai
} // namespace ov
} // namespace ov
2 changes: 0 additions & 2 deletions src/cpp/src/speculative_decoding/speculative_decoding_impl.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -232,8 +232,6 @@ ContinuousBatchingPipeline::SpeculativeDecodingImpl::generate(const std::vector<
continue;
}
std::unordered_map<uint64_t, GenerationOutput> token = main_generations.at(0).get()->back();
OPENVINO_ASSERT(1 <= token.size());
OPENVINO_ASSERT(1 <= token.begin()->second.generated_ids.size());
for (const auto& gen_token : token.begin()->second.generated_ids) {
continue_generation = !streamer_ptr->put(gen_token);
if (!continue_generation) {
Expand Down
Loading