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 3 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
10 changes: 7 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,13 @@ 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));
OPENVINO_ASSERT(1 <= token.size());
OPENVINO_ASSERT(1 <= token.begin()->second.generated_ids.size());
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: 13 additions & 12 deletions src/cpp/src/lm_encoding.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,7 +126,7 @@ std::pair<EncodedResults, int32_t> get_lm_encoded_results(
get_active_sequence_groups),
active_sequence_groups.end());

while (active_sequence_groups.size() > 0) {
ilya-lavrenov marked this conversation as resolved.
Show resolved Hide resolved
do {
size_t total_num_tokens = 0;

for (auto& sequence_group : active_sequence_groups) {
Copy link
Contributor

Choose a reason for hiding this comment

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

BTW

@sbalandi should we use active_sequence_groups instead of sequence_groups when compute beam_offets ?

Copy link
Contributor

Choose a reason for hiding this comment

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

it looks like yes, we can use active_sequence_groups here, but that variant is ok too, as in non active num_running_seqs will be 0
I'll check it to be sure and fix in #1215

Expand Down Expand Up @@ -203,11 +203,18 @@ std::pair<EncodedResults, int32_t> get_lm_encoded_results(
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;
// not generated tokens like several prompt phase
if (!generations.at(0).get()->can_read()) {
continue;
}
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());
for (const auto& gen_token : token.begin()->second.generated_ids) {
if (!streamer_ptr->put(gen_token)) {
break;
}
}
}

sampler_output = sampler.sample(active_sequence_groups, m_llm.get_tensor("logits"));
Expand All @@ -216,13 +223,7 @@ std::pair<EncodedResults, int32_t> get_lm_encoded_results(
active_sequence_groups.end(),
get_active_sequence_groups),
active_sequence_groups.end());
}

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();
}
} while (active_sequence_groups.size() > 0);

size_t next_selected_beam = 0;
for (size_t i = 0; i < sequence_groups.size(); i++) {
Expand Down
Loading