diff --git a/csrc/engine/compiler/paged_compiler.cpp b/csrc/engine/compiler/paged_compiler.cpp index eea15750b..85dcc8605 100644 --- a/csrc/engine/compiler/paged_compiler.cpp +++ b/csrc/engine/compiler/paged_compiler.cpp @@ -3,13 +3,142 @@ #include "../../utils.hpp" #include +#include #include +#include +#include +#include #include +#include +#include #include namespace infinilm::engine { namespace { +constexpr size_t kShortDecodeBlockTableWidth = 8; +constexpr size_t kShortDecodeBlockSize = 256; +constexpr size_t kShortDecodeMaxSequenceLength = kShortDecodeBlockTableWidth * kShortDecodeBlockSize; + +constexpr char kBaichuanFixedPrefillGraphEnv[] = "INFINILM_ENABLE_BAICHUAN_PREFILL_GRAPH"; +constexpr size_t kBaichuanFixedPrefillBatchSize = 1; +constexpr size_t kBaichuanFixedPrefillSequenceLength = 10; +constexpr size_t kBaichuanFixedPrefillBlockSize = 256; + +struct ReviewedPagedGraphProfile { + std::string_view model_type; + size_t hidden_size; + size_t num_attention_heads; + size_t num_key_value_heads; + size_t head_dim; + size_t block_size; + size_t minimum_num_blocks; + std::optional exact_num_blocks; + std::optional max_batch_size; + std::optional tensor_parallel_world_size; + std::optional num_hidden_layers; + std::optional position_id_axes; + std::optional dtype; + bool require_unquantized; +}; + +struct PagedGraphProperties { + infinicore::Device::Type device_type; + backends::AttentionBackend attention_backend; + size_t tensor_parallel_world_size; + size_t block_size; + size_t num_blocks; + size_t max_batch_size; + std::string model_type; + size_t hidden_size; + size_t num_hidden_layers; + size_t num_attention_heads; + size_t num_key_value_heads; + size_t head_dim; + size_t position_id_axes; + infinicore::DataType dtype; + quantization::QuantScheme quant_scheme; + quantization::KVQuantAlgo kv_quant_scheme; +}; + +const ReviewedPagedGraphProfile kBaichuanFixedPrefillProfile{ + "baichuan", 4096, 32, 32, 128, kBaichuanFixedPrefillBlockSize, 1, + 1, 1, 2, 32, 1, std::nullopt, true}; + +const std::array kShortDecodeProfiles{{ + {"internlm3", 4096, 32, 2, 128, kShortDecodeBlockSize, + kShortDecodeBlockTableWidth, std::nullopt, std::nullopt, + std::nullopt, std::nullopt, std::nullopt, std::nullopt, false}, + {"chatglm", 4096, 32, 2, 128, kShortDecodeBlockSize, + kShortDecodeBlockTableWidth, 512, std::nullopt, 1, 28, 1, + infinicore::DataType::kFloat16, true}, +}}; + +template +bool matches_optional_constraint( + const T &actual, + const std::optional &expected) { + return !expected.has_value() || actual == expected.value(); +} + +std::optional read_paged_graph_properties( + const cache::PagedKVCacheConfig &paged_config, + const config::ModelConfig *model_config) { + if (model_config == nullptr) { + return std::nullopt; + } + + const size_t hidden_size = model_config->get_or("hidden_size", 0); + const size_t num_attention_heads = model_config->get_or("num_attention_heads", 0); + const size_t head_dim = model_config->get_or( + "head_dim", + num_attention_heads == 0 ? 0 : hidden_size / num_attention_heads); + + return PagedGraphProperties{ + infinicore::context::getDevice().type(), + infinilm::global_state::get_infinilm_config().attention_backend, + infinilm::global_state::get_tensor_model_parallel_world_size(), + paged_config.block_size(), + paged_config.num_blocks(), + paged_config.max_batch_size(), + model_config->get_or("model_type", ""), + hidden_size, + model_config->get_or("num_hidden_layers", 0), + num_attention_heads, + model_config->get_or("num_key_value_heads", 0), + head_dim, + model_config->get_or("position_id_axes", 1), + model_config->get_dtype(), + model_config->get_quant_scheme(), + model_config->get_kv_quant_scheme(), + }; +} + +bool matches_reviewed_paged_graph_profile( + const PagedGraphProperties &properties, + const ReviewedPagedGraphProfile &profile) { + return properties.device_type == infinicore::Device::Type::kNvidia + && properties.attention_backend == backends::AttentionBackend::FLASH_ATTN + && properties.model_type == profile.model_type + && properties.hidden_size == profile.hidden_size + && properties.num_attention_heads == profile.num_attention_heads + && properties.num_key_value_heads == profile.num_key_value_heads + && properties.head_dim == profile.head_dim + && properties.block_size == profile.block_size + && properties.num_blocks >= profile.minimum_num_blocks + && matches_optional_constraint(properties.num_blocks, profile.exact_num_blocks) + && matches_optional_constraint(properties.max_batch_size, profile.max_batch_size) + && matches_optional_constraint( + properties.tensor_parallel_world_size, + profile.tensor_parallel_world_size) + && matches_optional_constraint(properties.num_hidden_layers, profile.num_hidden_layers) + && matches_optional_constraint(properties.position_id_axes, profile.position_id_axes) + && matches_optional_constraint(properties.dtype, profile.dtype) + && (!profile.require_unquantized + || (properties.quant_scheme == quantization::QuantScheme::NONE + && properties.kv_quant_scheme == quantization::KVQuantAlgo::NONE)); +} + bool has_mamba_cache(const infinilm::global_state::ForwardContext &forward_context) { auto has_state = [](const std::vector &state_vec) { for (const auto &state : state_vec) { @@ -23,6 +152,155 @@ bool has_mamba_cache(const infinilm::global_state::ForwardContext &forward_conte return has_state(forward_context.conv_state_vec) || has_state(forward_context.ssm_state_vec); } +bool env_flag_enabled(const char *name) { + const char *value = std::getenv(name); + return value != nullptr && std::string_view(value) == "1"; +} + +bool supports_baichuan_fixed_prefill_graph( + const cache::PagedKVCacheConfig &paged_config, + const config::ModelConfig *model_config, + bool has_mamba_state) { + if (!env_flag_enabled(kBaichuanFixedPrefillGraphEnv) || has_mamba_state) { + return false; + } + + const auto properties = read_paged_graph_properties( + paged_config, model_config); + return properties.has_value() + && matches_reviewed_paged_graph_profile( + properties.value(), kBaichuanFixedPrefillProfile); +} + +bool is_cpu_contiguous_tensor( + const std::optional &tensor, + infinicore::DataType dtype, + const std::vector &shape) { + return tensor.has_value() + && tensor.value() + && tensor.value()->device().type() == infinicore::Device::Type::kCpu + && tensor.value()->dtype() == dtype + && tensor.value()->shape() == shape + && tensor.value()->is_contiguous(); +} + +template +bool tensor_values_equal( + const std::optional &tensor, + std::initializer_list expected) { + const auto *values = reinterpret_cast(tensor.value()->data()); + return std::equal(expected.begin(), expected.end(), values); +} + +bool is_exact_baichuan_fixed_prefill_input( + const InfinilmModel::Input &input) { + const bool tensors_match = is_cpu_contiguous_tensor( + input.input_ids, + infinicore::DataType::kInt64, + {kBaichuanFixedPrefillBatchSize, + kBaichuanFixedPrefillSequenceLength}) + && is_cpu_contiguous_tensor( + input.position_ids, + infinicore::DataType::kInt64, + {kBaichuanFixedPrefillSequenceLength}) + && is_cpu_contiguous_tensor( + input.past_sequence_lengths, + infinicore::DataType::kInt32, + {kBaichuanFixedPrefillBatchSize}) + && is_cpu_contiguous_tensor( + input.total_sequence_lengths, + infinicore::DataType::kInt32, + {kBaichuanFixedPrefillBatchSize}) + && is_cpu_contiguous_tensor( + input.input_offsets, + infinicore::DataType::kInt32, + {kBaichuanFixedPrefillBatchSize + 1}) + && is_cpu_contiguous_tensor( + input.cu_seqlens, + infinicore::DataType::kInt32, + {kBaichuanFixedPrefillBatchSize + 1}) + && is_cpu_contiguous_tensor( + input.block_tables, + infinicore::DataType::kInt32, + {kBaichuanFixedPrefillBatchSize, 1}) + && is_cpu_contiguous_tensor( + input.slot_mapping, + infinicore::DataType::kInt64, + {kBaichuanFixedPrefillSequenceLength}); + if (!tensors_match) { + return false; + } + + const bool has_unsupported_input = input.mamba_init_state_indices.has_value() + || input.mamba_final_state_indices.has_value() + || input.pixel_values.has_value() + || input.image_bound.has_value() + || input.tgt_sizes.has_value() + || input.image_grid_thw.has_value() + || input.image_req_ids.has_value() + || input.visual_token_ranges.has_value() + || input.target_hidden_states.has_value() + || input.sample_all_positions; + return !has_unsupported_input + && tensor_values_equal( + input.position_ids, + {0, 1, 2, 3, 4, 5, 6, 7, 8, 9}) + && tensor_values_equal( + input.past_sequence_lengths, {0}) + && tensor_values_equal( + input.total_sequence_lengths, {10}) + && tensor_values_equal( + input.input_offsets, {0, 10}) + && tensor_values_equal( + input.cu_seqlens, {0, 10}) + && tensor_values_equal( + input.block_tables, {0}) + && tensor_values_equal( + input.slot_mapping, + {0, 1, 2, 3, 4, 5, 6, 7, 8, 9}); +} + +bool supports_reviewed_short_decode_graph( + const cache::PagedKVCacheConfig &paged_config, + const config::ModelConfig *model_config) { + const auto properties = read_paged_graph_properties( + paged_config, model_config); + return properties.has_value() + && std::any_of( + kShortDecodeProfiles.begin(), + kShortDecodeProfiles.end(), + [&](const ReviewedPagedGraphProfile &profile) { + return matches_reviewed_paged_graph_profile( + properties.value(), profile); + }); +} + +bool tensors_compatible(const infinicore::Tensor &target, + const infinicore::Tensor &source) { + return target + && source + && target->shape() == source->shape() + && target->dtype() == source->dtype(); +} + +bool required_tensors_compatible( + const std::optional &target, + const std::optional &source) { + return target.has_value() + && source.has_value() + && tensors_compatible(target.value(), source.value()); +} + +bool optional_tensors_compatible( + const std::optional &target, + const std::optional &source) { + if (target.has_value() != source.has_value()) { + return false; + } + return !target.has_value() + || tensors_compatible(target.value(), source.value()); +} + } // namespace PagedCompiler::PagedCompiler(const std::shared_ptr &model, RankBarrier *barrier) @@ -57,9 +335,16 @@ PagedCompiler::PagedCompiler(const std::shared_ptr &model, RankBa } void PagedCompiler::compile() { - if (model_->get_cache_config() != nullptr && dynamic_cast(model_->get_cache_config())) { + const auto *paged_config = dynamic_cast( + model_->get_cache_config()); + if (paged_config != nullptr) { + compiled_short_decode_b1_.reset(); + compiled_baichuan_prefill_b1_s10_.reset(); compiled_map_decode_.clear(); - size_t nblocks = dynamic_cast(model_->get_cache_config())->num_blocks(); + block_tables_holder_.reset(); + short_block_tables_holder_.reset(); + + const size_t nblocks = paged_config->num_blocks(); auto &forward_context = infinilm::global_state::get_forward_context(); const bool has_mamba_state = has_mamba_cache(forward_context); @@ -76,7 +361,9 @@ void PagedCompiler::compile() { {nblocks * max_batch_size}, infinicore::DataType::kInt32, infinicore::context::getDevice()); set_zeros(block_tables_holder_); - auto make_decode_input = [&](size_t b) { + auto make_decode_input = [&](size_t b, + size_t block_per_req, + const infinicore::Tensor &block_tables_holder) { InfinilmModel::Input input; input.input_ids = infinicore::Tensor::empty({1, b}, infinicore::DataType::kInt64, infinicore::context::getDevice()); input.position_ids = infinicore::Tensor::empty( @@ -98,8 +385,9 @@ void PagedCompiler::compile() { infinicore::context::memcpyH2D(input.input_offsets.value()->data(), input_offsets_vec.data(), (b + 1) * sizeof(int32_t), false); input.cu_seqlens = infinicore::Tensor::empty({b + 1}, infinicore::DataType::kInt32, infinicore::context::getDevice()); infinicore::context::memcpyH2D(input.cu_seqlens.value()->data(), input_offsets_vec.data(), (b + 1) * sizeof(int32_t), false); - const size_t block_per_req = nblocks; - input.block_tables = block_tables_holder_->as_strided({b, block_per_req}, {(ptrdiff_t)block_per_req, 1}); + input.block_tables = block_tables_holder->as_strided( + {b, block_per_req}, + {static_cast(block_per_req), 1}); input.slot_mapping = infinicore::Tensor::empty({b}, infinicore::DataType::kInt64, infinicore::context::getDevice()); set_zeros(input.slot_mapping.value()); @@ -142,9 +430,154 @@ void PagedCompiler::compile() { return input; }; + auto make_baichuan_fixed_prefill_input = [&]() { + InfinilmModel::Input input; + input.input_ids = infinicore::Tensor::empty( + {kBaichuanFixedPrefillBatchSize, + kBaichuanFixedPrefillSequenceLength}, + infinicore::DataType::kInt64, + infinicore::context::getDevice()); + input.position_ids = infinicore::Tensor::empty( + {kBaichuanFixedPrefillSequenceLength}, + infinicore::DataType::kInt64, + infinicore::context::getDevice()); + input.past_sequence_lengths = infinicore::Tensor::empty( + {kBaichuanFixedPrefillBatchSize}, + infinicore::DataType::kInt32, + infinicore::context::getDevice()); + input.total_sequence_lengths = infinicore::Tensor::empty( + {kBaichuanFixedPrefillBatchSize}, + infinicore::DataType::kInt32, + infinicore::context::getDevice()); + input.input_offsets = infinicore::Tensor::empty( + {kBaichuanFixedPrefillBatchSize + 1}, + infinicore::DataType::kInt32, + infinicore::context::getDevice()); + input.cu_seqlens = infinicore::Tensor::empty( + {kBaichuanFixedPrefillBatchSize + 1}, + infinicore::DataType::kInt32, + infinicore::context::getDevice()); + input.block_tables = infinicore::Tensor::empty( + {kBaichuanFixedPrefillBatchSize, 1}, + infinicore::DataType::kInt32, + infinicore::context::getDevice()); + input.slot_mapping = infinicore::Tensor::empty( + {kBaichuanFixedPrefillSequenceLength}, + infinicore::DataType::kInt64, + infinicore::context::getDevice()); + + const std::vector position_ids{ + 0, 1, 2, 3, 4, 5, 6, 7, 8, 9}; + const std::vector past_sequence_lengths{0}; + const std::vector total_sequence_lengths{ + static_cast(kBaichuanFixedPrefillSequenceLength)}; + const std::vector packed_offsets{ + 0, + static_cast(kBaichuanFixedPrefillSequenceLength)}; + const std::vector block_tables{0}; + const std::vector slot_mapping{ + 0, 1, 2, 3, 4, 5, 6, 7, 8, 9}; + set_zeros(input.input_ids.value()); + infinicore::context::memcpyH2D( + input.position_ids.value()->data(), + position_ids.data(), + position_ids.size() * sizeof(int64_t), + false); + infinicore::context::memcpyH2D( + input.past_sequence_lengths.value()->data(), + past_sequence_lengths.data(), + past_sequence_lengths.size() * sizeof(int32_t), + false); + infinicore::context::memcpyH2D( + input.total_sequence_lengths.value()->data(), + total_sequence_lengths.data(), + total_sequence_lengths.size() * sizeof(int32_t), + false); + infinicore::context::memcpyH2D( + input.input_offsets.value()->data(), + packed_offsets.data(), + packed_offsets.size() * sizeof(int32_t), + false); + infinicore::context::memcpyH2D( + input.cu_seqlens.value()->data(), + packed_offsets.data(), + packed_offsets.size() * sizeof(int32_t), + false); + infinicore::context::memcpyH2D( + input.block_tables.value()->data(), + block_tables.data(), + block_tables.size() * sizeof(int32_t), + false); + infinicore::context::memcpyH2D( + input.slot_mapping.value()->data(), + slot_mapping.data(), + slot_mapping.size() * sizeof(int64_t), + false); + + forward_context.attn_metadata = { + input.past_sequence_lengths, + input.total_sequence_lengths, + input.input_offsets, + input.cu_seqlens, + input.block_tables, + input.slot_mapping, + kBaichuanFixedPrefillSequenceLength, + kBaichuanFixedPrefillSequenceLength, + }; + forward_context.mamba_metadata = { + input.input_offsets, + std::nullopt, + std::nullopt, + }; + return input; + }; + + auto make_compiled_result = []( + InfinilmModel::Input input, + std::shared_ptr graph, + const InfinilmModel::Output &output) { + infinicore::Tensor graph_hidden_states; + if (output.hidden_states) { + graph_hidden_states = infinicore::graph::GraphTensor( + output.hidden_states, + infinicore::graph::GraphTensor::SnapshotPolicy::kBlob); + } + auto shared_output = std::shared_ptr( + new InfinilmModel::Output{ + infinicore::graph::GraphTensor( + output.logits, + infinicore::graph::GraphTensor::SnapshotPolicy::kBlob), + graph_hidden_states}); + + return CompiledResult{ + std::move(input), + std::make_tuple(std::move(graph), std::move(shared_output))}; + }; + + auto capture_decode = [&](size_t b, + size_t block_per_req, + const infinicore::Tensor &block_tables_holder) { + auto input = make_decode_input(b, block_per_req, block_tables_holder); + + barrier_->wait(); + (void)model_->forward(input); + infinicore::context::syncStream(); + // Capture must not start with stale state from previous attempts. + model_->reset_runtime_state(); + infinicore::context::syncStream(); + GraphRecordingGuard recording; + auto output = model_->forward(input); + auto graph = recording.finish(); + barrier_->wait(); + + return make_compiled_result( + std::move(input), std::move(graph), output); + }; + { const size_t warmup_batch_size = std::min(max_batch_size, static_cast(64)); - auto input = make_decode_input(warmup_batch_size); + auto input = make_decode_input( + warmup_batch_size, nblocks, block_tables_holder_); model_->forward(input); infinicore::context::syncStream(); // Clear transient operator state before CUDA graph capture. @@ -153,12 +586,34 @@ void PagedCompiler::compile() { } for (size_t b : decode_batch_sizes_) { - auto input = make_decode_input(b); + compiled_map_decode_[b] = capture_decode(b, nblocks, block_tables_holder_); + } + if (supports_reviewed_short_decode_graph( + *paged_config, model_->get_model_config().get())) { + short_block_tables_holder_ = infinicore::Tensor::empty( + {kShortDecodeBlockTableWidth}, + infinicore::DataType::kInt32, + infinicore::context::getDevice()); + set_zeros(short_block_tables_holder_); + compiled_short_decode_b1_.emplace(capture_decode( + 1, + kShortDecodeBlockTableWidth, + short_block_tables_holder_)); + } + + if (supports_baichuan_fixed_prefill_graph( + *paged_config, model_->get_model_config().get(), has_mamba_state)) { + auto input = make_baichuan_fixed_prefill_input(); + + if (std::getenv("INFINICORE_GRAPH_DEBUG") != nullptr) { + spdlog::info( + "fixed Baichuan prefill graph compile: rank={}, batch=1, seq=10", + infinilm::global_state::get_tensor_model_parallel_rank()); + } barrier_->wait(); (void)model_->forward(input); infinicore::context::syncStream(); - // Capture must not start with stale state from previous attempts. model_->reset_runtime_state(); infinicore::context::syncStream(); GraphRecordingGuard recording; @@ -166,74 +621,183 @@ void PagedCompiler::compile() { auto graph = recording.finish(); barrier_->wait(); - auto shared_output = std::shared_ptr( - new InfinilmModel::Output{infinicore::graph::GraphTensor(output.logits)}); - - compiled_map_decode_[b] = CompiledResult{std::move(input), std::make_tuple(graph, shared_output)}; + compiled_baichuan_prefill_b1_s10_.emplace( + make_compiled_result( + std::move(input), std::move(graph), output)); } } } PagedCompiler::Compiled PagedCompiler::get_compiled(const InfinilmModel::Input &input) { - if (model_->get_cache_config() != nullptr && dynamic_cast(model_->get_cache_config())) { - size_t batch_size = input.block_tables.value()->size(0); - size_t block_per_req = input.block_tables.value()->size(1); + const auto *paged_config = dynamic_cast( + model_->get_cache_config()); + if (paged_config == nullptr + || !input.block_tables.has_value() + || !input.block_tables.value() + || input.block_tables.value()->ndim() != 2 + || !input.input_ids.has_value() + || !input.input_ids.value() + || input.input_ids.value()->ndim() != 2) { + return {nullptr, nullptr}; + } + + const auto &runtime_block_tables = input.block_tables.value(); + const size_t batch_size = runtime_block_tables->size(0); + const size_t block_per_req = runtime_block_tables->size(1); - // only support decode only batch + const bool use_baichuan_fixed_prefill_graph = compiled_baichuan_prefill_b1_s10_.has_value() + && is_exact_baichuan_fixed_prefill_input(input); + CompiledResult *selected_result = nullptr; + bool use_short_decode_graph = false; + size_t required_pages = 0; + if (use_baichuan_fixed_prefill_graph) { + selected_result = &compiled_baichuan_prefill_b1_s10_.value(); + } else { + // Every other compiled paged graph is decode-only. if (batch_size != input.input_ids.value()->size(1)) { return {nullptr, nullptr}; - } else { - auto result = compiled_map_decode_.find(batch_size); - if (result == compiled_map_decode_.end()) { - return {nullptr, nullptr}; - } - auto &graph_input = result->second.input; - - graph_input.input_ids.value()->copy_from(input.input_ids.value()); - graph_input.position_ids.value()->copy_from(input.position_ids.value()); - graph_input.total_sequence_lengths.value()->copy_from(input.total_sequence_lengths.value()); - graph_input.input_offsets.value()->copy_from(input.input_offsets.value()); - graph_input.cu_seqlens.value()->copy_from(input.cu_seqlens.value()); - - const size_t compiled_block_per_req = graph_input.block_tables.value()->size(1); - if (block_per_req > compiled_block_per_req) { - // Runtime width exceeds compiled graph slot; fall back to eager path. - return {nullptr, nullptr}; - } + } - // Initialize only the active graph rows to -1, then overwrite the - // runtime logical region. Avoid clearing the full preallocated - // holder on every decode token. - auto &graph_block_tables = graph_input.block_tables.value(); - set_minus_one_device_async(graph_block_tables); - graph_block_tables->narrow({{1, 0, block_per_req}})->copy_from(input.block_tables.value()); - graph_input.slot_mapping.value()->copy_from(input.slot_mapping.value()); - const bool graph_has_mamba_indices = graph_input.mamba_init_state_indices.has_value() && graph_input.mamba_final_state_indices.has_value(); - const bool input_has_mamba_indices = input.mamba_init_state_indices.has_value() && input.mamba_final_state_indices.has_value(); - if (graph_has_mamba_indices != input_has_mamba_indices) { - return {nullptr, nullptr}; - } - if (graph_has_mamba_indices) { - graph_input.mamba_init_state_indices.value()->copy_from( - input.mamba_init_state_indices.value()); - graph_input.mamba_final_state_indices.value()->copy_from( - input.mamba_final_state_indices.value()); + auto general_result = compiled_map_decode_.find(batch_size); + if (general_result == compiled_map_decode_.end()) { + return {nullptr, nullptr}; + } + selected_result = &general_result->second; + + if (batch_size == 1 && compiled_short_decode_b1_.has_value()) { + const auto &total_sequence_lengths = input.total_sequence_lengths; + const bool valid_short_metadata = total_sequence_lengths.has_value() + && total_sequence_lengths.value() + && total_sequence_lengths.value()->device().type() + == infinicore::Device::Type::kCpu + && total_sequence_lengths.value()->dtype() + == infinicore::DataType::kInt32 + && total_sequence_lengths.value()->ndim() == 1 + && total_sequence_lengths.value()->size(0) == 1 + && total_sequence_lengths.value()->is_contiguous() + && runtime_block_tables->device().type() + == infinicore::Device::Type::kCpu + && runtime_block_tables->dtype() == infinicore::DataType::kInt32 + && runtime_block_tables->is_contiguous(); + if (valid_short_metadata) { + const int32_t total_sequence_length = reinterpret_cast( + total_sequence_lengths.value()->data())[0]; + if (total_sequence_length > 0 + && static_cast(total_sequence_length) + <= kShortDecodeMaxSequenceLength) { + required_pages = 1 + + (static_cast(total_sequence_length) - 1) + / paged_config->block_size(); + if (required_pages <= kShortDecodeBlockTableWidth + && block_per_req >= required_pages) { + selected_result = &compiled_short_decode_b1_.value(); + use_short_decode_graph = true; + } + } } - // CUDA graph replay reuses the same per-layer Marlin workspaces. - // The graph itself does not contain a workspace reset, so enqueue - // one on the same stream before launch. This is correct but costs - // decode latency; the intended follow-up is a reusable global - // zero workspace/lock buffer shared by all Marlin layers. - model_->reset_runtime_state(); + } + } - auto graph = std::get<0>(result->second.compiled); - auto shared_output = std::shared_ptr(new InfinilmModel::Output{std::get<1>(result->second.compiled)->logits->resume_from_blob_()}); + auto &graph_input = selected_result->input; + const size_t compiled_block_per_req = graph_input.block_tables.value()->size(1); + if ((!use_short_decode_graph && block_per_req > compiled_block_per_req) + || graph_input.block_tables.value()->dtype() + != runtime_block_tables->dtype() + || graph_input.block_tables.value()->size(0) != batch_size + || !required_tensors_compatible(graph_input.input_ids, input.input_ids) + || !required_tensors_compatible( + graph_input.position_ids, input.position_ids) + || (use_baichuan_fixed_prefill_graph + && !required_tensors_compatible( + graph_input.past_sequence_lengths, + input.past_sequence_lengths)) + || !required_tensors_compatible( + graph_input.total_sequence_lengths, + input.total_sequence_lengths) + || !required_tensors_compatible( + graph_input.input_offsets, input.input_offsets) + || !required_tensors_compatible( + graph_input.cu_seqlens, input.cu_seqlens) + || !required_tensors_compatible( + graph_input.slot_mapping, input.slot_mapping) + || !optional_tensors_compatible( + graph_input.mamba_init_state_indices, + input.mamba_init_state_indices) + || !optional_tensors_compatible( + graph_input.mamba_final_state_indices, + input.mamba_final_state_indices)) { + // Validate every input before mutating storage shared by a graph. + return {nullptr, nullptr}; + } - return std::make_tuple(graph, shared_output); - } + if (use_baichuan_fixed_prefill_graph + && std::getenv("INFINICORE_GRAPH_DEBUG") != nullptr) { + spdlog::info( + "fixed Baichuan prefill graph hit: rank={}, batch=1, seq=10", + infinilm::global_state::get_tensor_model_parallel_rank()); + } + + graph_input.input_ids.value()->copy_from(input.input_ids.value()); + graph_input.position_ids.value()->copy_from(input.position_ids.value()); + if (use_baichuan_fixed_prefill_graph) { + graph_input.past_sequence_lengths.value()->copy_from( + input.past_sequence_lengths.value()); + } + graph_input.total_sequence_lengths.value()->copy_from(input.total_sequence_lengths.value()); + graph_input.input_offsets.value()->copy_from(input.input_offsets.value()); + graph_input.cu_seqlens.value()->copy_from(input.cu_seqlens.value()); + + auto &graph_block_tables = graph_input.block_tables.value(); + if (use_baichuan_fixed_prefill_graph) { + graph_block_tables->copy_from(runtime_block_tables); + } else if (use_short_decode_graph) { + infinicore::context::setDevice(graph_block_tables->device()); + set_minus_one_device_async(graph_block_tables); + infinicore::context::memcpyH2D( + graph_block_tables->data(), + runtime_block_tables->data(), + required_pages * sizeof(int32_t), + false); } else { - return {nullptr, nullptr}; + // Initialize only the active graph rows to -1, then overwrite the + // runtime logical region. Avoid clearing the full preallocated holder + // on every decode token. + set_minus_one_device_async(graph_block_tables); + graph_block_tables->narrow({{1, 0, block_per_req}}) + ->copy_from(runtime_block_tables); } + graph_input.slot_mapping.value()->copy_from(input.slot_mapping.value()); + + const bool graph_has_mamba_indices = graph_input.mamba_init_state_indices.has_value() + && graph_input.mamba_final_state_indices.has_value(); + if (graph_has_mamba_indices) { + graph_input.mamba_init_state_indices.value()->copy_from( + input.mamba_init_state_indices.value()); + graph_input.mamba_final_state_indices.value()->copy_from( + input.mamba_final_state_indices.value()); + } + + // CUDA graph replay reuses the same per-layer Marlin workspaces. + // The graph itself does not contain a workspace reset, so enqueue + // one on the same stream before launch. This is correct but costs + // decode latency; the intended follow-up is a reusable global + // zero workspace/lock buffer shared by all Marlin layers. + if (!use_baichuan_fixed_prefill_graph) { + model_->reset_runtime_state(); + } + + auto graph = std::get<0>(selected_result->compiled); + const auto &compiled_output = std::get<1>(selected_result->compiled); + infinicore::Tensor hidden_states; + if (compiled_output->hidden_states) { + hidden_states = compiled_output->hidden_states->resume_from_blob_(); + } + auto shared_output = std::shared_ptr( + new InfinilmModel::Output{ + compiled_output->logits->resume_from_blob_(), + hidden_states}); + + return std::make_tuple(graph, shared_output); } } // namespace infinilm::engine diff --git a/csrc/engine/compiler/paged_compiler.hpp b/csrc/engine/compiler/paged_compiler.hpp index a1125864d..a0d280e84 100644 --- a/csrc/engine/compiler/paged_compiler.hpp +++ b/csrc/engine/compiler/paged_compiler.hpp @@ -2,6 +2,7 @@ #include "graph_compiler.hpp" +#include #include namespace infinilm::engine { @@ -17,12 +18,15 @@ class PagedCompiler : public GraphCompiler { std::vector decode_batch_sizes_; infinicore::Tensor block_tables_holder_; + infinicore::Tensor short_block_tables_holder_; struct CompiledResult { InfinilmModel::Input input; Compiled compiled; }; + std::optional compiled_short_decode_b1_; + std::optional compiled_baichuan_prefill_b1_s10_; std::unordered_map< size_t, // num_requests CompiledResult> diff --git a/csrc/engine/compiler/static_batching_compiler.cpp b/csrc/engine/compiler/static_batching_compiler.cpp index d3858a1e2..ae86b61b5 100644 --- a/csrc/engine/compiler/static_batching_compiler.cpp +++ b/csrc/engine/compiler/static_batching_compiler.cpp @@ -5,6 +5,7 @@ #include #include +#include #include namespace { @@ -64,6 +65,9 @@ void StaticBatchingCompiler::compile() { return; } const size_t cache_page_size = *static_graph_cache_page_size(b); + const auto &model_config = model_->get_model_config(); + const bool uses_target_hidden_states = model_config + && model_config->get_or("model_type", "") == "minicpm_eagle"; { InfinilmModel::Input input; input.input_ids = infinicore::Tensor::empty({b, 1}, infinicore::DataType::kInt64, infinicore::context::getDevice()); @@ -87,6 +91,13 @@ void StaticBatchingCompiler::compile() { slot_mapping_vec[i] = static_cast(i * cache_page_size); } infinicore::context::memcpyH2D(input.slot_mapping.value()->data(), slot_mapping_vec.data(), b * sizeof(int64_t), false); + if (uses_target_hidden_states) { + input.target_hidden_states = infinicore::Tensor::empty( + {b, 1, model_config->get("hidden_size")}, + model_config->get_dtype(), + infinicore::context::getDevice()); + set_zeros(input.target_hidden_states.value()); + } // Attention reads attn_metadata from thread-local forward context. infinilm::global_state::get_forward_context().attn_metadata = { @@ -97,6 +108,8 @@ void StaticBatchingCompiler::compile() { input.block_tables, input.slot_mapping, }; + infinilm::global_state::get_forward_context().attn_metadata.first_past_sequence_length = 0; + infinilm::global_state::get_forward_context().attn_metadata.first_total_sequence_length = 1; barrier_->wait(); (void)model_->forward(input); @@ -107,7 +120,18 @@ void StaticBatchingCompiler::compile() { auto graph = recording.finish(); barrier_->wait(); - auto shared_output = std::shared_ptr(new InfinilmModel::Output{infinicore::graph::GraphTensor(output.logits)}); + infinicore::Tensor graph_hidden_states; + if (output.hidden_states) { + graph_hidden_states = infinicore::graph::GraphTensor( + output.hidden_states, + infinicore::graph::GraphTensor::SnapshotPolicy::kBlob); + } + auto shared_output = std::shared_ptr( + new InfinilmModel::Output{ + infinicore::graph::GraphTensor( + output.logits, + infinicore::graph::GraphTensor::SnapshotPolicy::kBlob), + graph_hidden_states}); compiled_map_[std::make_tuple(b, 1)] = CompiledResult{ std::move(input), std::make_tuple(graph, shared_output), cache_page_size}; @@ -124,10 +148,26 @@ StaticBatchingCompiler::Compiled StaticBatchingCompiler::get_compiled( return std::make_tuple(nullptr, nullptr); } else { auto &graph_input = result->second.input; + const bool graph_has_target_hidden_states = graph_input.target_hidden_states.has_value(); + const bool input_has_target_hidden_states = input.target_hidden_states.has_value(); + if (graph_has_target_hidden_states != input_has_target_hidden_states) { + return std::make_tuple(nullptr, nullptr); + } + if (graph_has_target_hidden_states + && (graph_input.target_hidden_states.value()->shape() + != input.target_hidden_states.value()->shape() + || graph_input.target_hidden_states.value()->dtype() + != input.target_hidden_states.value()->dtype())) { + return std::make_tuple(nullptr, nullptr); + } graph_input.input_ids.value()->copy_from(input.input_ids.value()); graph_input.position_ids.value()->copy_from(input.position_ids.value()); graph_input.past_sequence_lengths.value()->copy_from(input.past_sequence_lengths.value()); graph_input.total_sequence_lengths.value()->copy_from(input.total_sequence_lengths.value()); + if (graph_has_target_hidden_states) { + graph_input.target_hidden_states.value()->copy_from( + input.target_hidden_states.value()); + } ASSERT(input.past_sequence_lengths.value()->device().type() == infinicore::Device::Type::kCpu); ASSERT(input.past_sequence_lengths.value()->dtype() == infinicore::DataType::kInt32); @@ -148,7 +188,15 @@ StaticBatchingCompiler::Compiled StaticBatchingCompiler::get_compiled( false); auto graph = std::get<0>(result->second.compiled); - auto shared_output = std::shared_ptr(new InfinilmModel::Output{std::get<1>(result->second.compiled)->logits->resume_from_blob_()}); + const auto &compiled_output = std::get<1>(result->second.compiled); + infinicore::Tensor hidden_states; + if (compiled_output->hidden_states) { + hidden_states = compiled_output->hidden_states->resume_from_blob_(); + } + auto shared_output = std::shared_ptr( + new InfinilmModel::Output{ + compiled_output->logits->resume_from_blob_(), + hidden_states}); return std::make_tuple(graph, shared_output); } } else { diff --git a/csrc/engine/infer_engine.cpp b/csrc/engine/infer_engine.cpp index 563e92439..2c627a370 100644 --- a/csrc/engine/infer_engine.cpp +++ b/csrc/engine/infer_engine.cpp @@ -45,6 +45,33 @@ size_t max_length_from_offsets( return max_length; } +std::optional first_sequence_length( + const std::optional &lengths, + const char *name) { + if (!lengths.has_value()) { + return std::nullopt; + } + + auto cpu_lengths = lengths.value(); + if (cpu_lengths->device().type() != infinicore::Device::Type::kCpu) { + cpu_lengths = cpu_lengths->to( + infinicore::Device{infinicore::Device::Type::kCpu}); + } + + if (cpu_lengths->dtype() != infinicore::DataType::kInt32 + || cpu_lengths->shape().size() != 1 + || cpu_lengths->shape()[0] == 0) { + throw std::invalid_argument( + std::string(name) + " must be a non-empty one-dimensional int32 tensor"); + } + + const auto value = reinterpret_cast(cpu_lengths->data())[0]; + if (value < 0) { + throw std::invalid_argument(std::string(name) + " must contain nonnegative lengths"); + } + return static_cast(value); +} + } // namespace //------------------------------------------------------ @@ -238,7 +265,10 @@ std::vector InferEngine::state_dict_keys() { // forward //------------------------------------------------------ infinilm::InfinilmModel::Input -InferEngine::Input::to_model_input(infinicore::Device device) const { +InferEngine::Input::to_model_input( + infinicore::Device device, + bool snapshot_static_sequence_lengths, + bool preserve_target_hidden_device) const { auto to_device = [&](const std::optional &t) -> std::optional { @@ -264,6 +294,15 @@ InferEngine::Input::to_model_input(infinicore::Device device) const { const size_t max_query_length = is_prefill ? max_length_from_offsets(input_offsets, "input_offsets") : 0; const size_t max_sequence_length = is_prefill ? max_length_from_offsets(cu_seqlens, "cu_seqlens") : 0; + std::optional first_past_sequence_length; + std::optional first_total_sequence_length; + if (snapshot_static_sequence_lengths) { + first_past_sequence_length = first_sequence_length( + past_sequence_lengths, "past_sequence_lengths"); + first_total_sequence_length = first_sequence_length( + total_sequence_lengths, "total_sequence_lengths"); + } + infinilm::InfinilmModel::Input input = { to_device(input_ids), // @todo: on device in the future to_device(position_ids), @@ -281,10 +320,13 @@ InferEngine::Input::to_model_input(infinicore::Device device) const { to_device_vec(image_grid_thw), image_req_ids, visual_token_ranges, - to_device(target_hidden_states), + preserve_target_hidden_device + ? target_hidden_states + : to_device(target_hidden_states), sample_all_positions}; - infinilm::global_state::get_forward_context().attn_metadata = { + auto &attn_metadata = infinilm::global_state::get_forward_context().attn_metadata; + attn_metadata = { input.past_sequence_lengths, input.total_sequence_lengths, input.input_offsets, @@ -293,6 +335,8 @@ InferEngine::Input::to_model_input(infinicore::Device device) const { input.slot_mapping, max_query_length, max_sequence_length}; + attn_metadata.first_past_sequence_length = first_past_sequence_length; + attn_metadata.first_total_sequence_length = first_total_sequence_length; infinilm::global_state::get_forward_context().mamba_metadata = { input.input_offsets, diff --git a/csrc/engine/rank_worker.cpp b/csrc/engine/rank_worker.cpp index eab78c4c3..ddfe1ee4e 100644 --- a/csrc/engine/rank_worker.cpp +++ b/csrc/engine/rank_worker.cpp @@ -418,18 +418,25 @@ void RankWorker::thread_loop() { infinicore::Tensor logits; infinicore::Tensor hidden_states; - // All-position speculative/MTP runs need eager mode because - // hidden states are not part of compiled graph outputs. + // Packed all-position runs do not match the one-token-per-request + // output shape captured by the compiled decode graphs. if (!local_args.sample_all_positions && compiler_ != nullptr && rank_info_.pp_size == 1) { - auto [graph, output] = compiler_->get_compiled(local_args.to_model_input(infinicore::Device{infinicore::Device::Type::kCpu})); + auto [graph, output] = compiler_->get_compiled( + local_args.to_model_input( + infinicore::Device{infinicore::Device::Type::kCpu}, + false, + true)); if (graph != nullptr && output != nullptr) { graph->run(); logits = output->logits; + hidden_states = output->hidden_states; } } // Fall back to eager mode if (!logits) { - auto model_args = local_args.to_model_input(rank_info_.device); + auto model_args = local_args.to_model_input( + rank_info_.device, + attention_backend_ == backends::AttentionBackend::STATIC_ATTN); auto model_output = model_->forward(model_args); logits = model_output.logits; hidden_states = model_output.hidden_states; @@ -484,16 +491,41 @@ void RankWorker::thread_loop() { const size_t n_out = sample_all_positions ? static_cast(input_offsets[n_req]) : n_req; auto output_ids{infinicore::Tensor::empty({n_out}, infinicore::DataType::kInt64, rank_info_.device)}; - for (size_t i{0}; i < n_out; ++i) { - size_t score_idx = i; - if (!sample_all_positions && !logits_are_last_token_only) { - score_idx = static_cast(input_offsets[i + 1] - 1); + const bool parameter_greedy = top_p == 0.0f || top_k == 1 || temperature == 0.0f; + const auto logits_dtype = logits->dtype(); + const bool batch_greedy = rank_info_.device.type() == infinicore::Device::Type::kNvidia + && parameter_greedy + && n_out > 0 + && logits_positions == n_out + && logits->is_contiguous() + && (logits_dtype == infinicore::DataType::kFloat16 + || logits_dtype == infinicore::DataType::kBFloat16 + || logits_dtype == infinicore::DataType::kFloat32) + && (sample_all_positions || logits_are_last_token_only); + if (batch_greedy) { + float random_val = 0.0f; + for (size_t i{0}; i < n_out; ++i) { + random_val = std::uniform_real_distribution(0, 1)(rng_); } - auto score{logits->view({logits_positions, vocab_size})->narrow({{0, score_idx, 1}})->view({vocab_size})}; - auto out{output_ids->narrow({{0, i, 1}})->view({})}; - float random_val = std::uniform_real_distribution(0, 1)(rng_); infinicore::op::random_sample_( - out, score, random_val, top_p, top_k, temperature); + output_ids, + logits->view({logits_positions, vocab_size}), + random_val, + top_p, + top_k, + temperature); + } else { + for (size_t i{0}; i < n_out; ++i) { + size_t score_idx = i; + if (!sample_all_positions && !logits_are_last_token_only) { + score_idx = static_cast(input_offsets[i + 1] - 1); + } + auto score{logits->view({logits_positions, vocab_size})->narrow({{0, score_idx, 1}})->view({vocab_size})}; + auto out{output_ids->narrow({{0, i, 1}})->view({})}; + float random_val = std::uniform_real_distribution(0, 1)(rng_); + infinicore::op::random_sample_( + out, score, random_val, top_p, top_k, temperature); + } } if (rank_info_.pp_size > 1) { diff --git a/csrc/engine/rank_worker.hpp b/csrc/engine/rank_worker.hpp index d396ef6f1..ac8a9d0d4 100644 --- a/csrc/engine/rank_worker.hpp +++ b/csrc/engine/rank_worker.hpp @@ -79,7 +79,10 @@ class RankWorker { float top_p{1}; - infinilm::InfinilmModel::Input to_model_input(infinicore::Device device) const; + infinilm::InfinilmModel::Input to_model_input( + infinicore::Device device, + bool snapshot_static_sequence_lengths = false, + bool preserve_target_hidden_device = false) const; }; struct Output { diff --git a/csrc/global_state/forward_context.hpp b/csrc/global_state/forward_context.hpp index 6d02de4c5..1bc608424 100644 --- a/csrc/global_state/forward_context.hpp +++ b/csrc/global_state/forward_context.hpp @@ -21,6 +21,10 @@ struct AttentionMetadata { size_t max_query_length{0}; /// Maximum total sequence length in the current batch. size_t max_sequence_length{0}; + /// Element 0 of past_sequence_lengths, snapshotted for static eager attention. + std::optional first_past_sequence_length; + /// Element 0 of total_sequence_lengths, snapshotted for static eager attention. + std::optional first_total_sequence_length; AttentionMetadata() = default; diff --git a/csrc/infinicore/include/infinicore/graph/graph.hpp b/csrc/infinicore/include/infinicore/graph/graph.hpp index db1ba5b37..f0e63cb2d 100644 --- a/csrc/infinicore/include/infinicore/graph/graph.hpp +++ b/csrc/infinicore/include/infinicore/graph/graph.hpp @@ -15,7 +15,13 @@ class GraphManager; class GraphTensor : public Tensor { public: + enum class SnapshotPolicy { + kRecordingAware, + kBlob, + }; + GraphTensor(const Tensor &); + GraphTensor(const Tensor &, SnapshotPolicy policy); }; class GraphOperator { diff --git a/csrc/infinicore/include/infinicore/ops/mha_kvcache.hpp b/csrc/infinicore/include/infinicore/ops/mha_kvcache.hpp index 0d4709b17..bd969da38 100644 --- a/csrc/infinicore/include/infinicore/ops/mha_kvcache.hpp +++ b/csrc/infinicore/include/infinicore/ops/mha_kvcache.hpp @@ -61,9 +61,21 @@ class MhaKVCache : public graph::DispatchableGraphOperator { std::optional alibi_slopes, float scale); - // Some FlashAttention providers allocate temporary storage outside the - // graph lease, so decode remains a conservative host segment. - bool is_device_graph_capture_safe() const override { return false; } + static bool supports_device_graph_capture( + const Tensor &out, + const Tensor &q, + const Tensor &k_cache, + const Tensor &v_cache, + const Tensor &seqlens_k, + const Tensor &block_table, + const std::optional &alibi_slopes); + + bool is_device_graph_capture_safe() const override { + return device_graph_capture_safe_; + } + +private: + bool device_graph_capture_safe_; }; Tensor mha_kvcache(const Tensor &q, diff --git a/csrc/infinicore/include/infinicore/ops/mha_varlen.hpp b/csrc/infinicore/include/infinicore/ops/mha_varlen.hpp index 4d226f383..96b06be6a 100644 --- a/csrc/infinicore/include/infinicore/ops/mha_varlen.hpp +++ b/csrc/infinicore/include/infinicore/ops/mha_varlen.hpp @@ -6,19 +6,66 @@ namespace infinicore::op { -INFINICORE_GRAPH_OP_CLASS( - MultiheadAttentionVarlen, - Tensor, - const Tensor &, - const Tensor &, - const Tensor &, - const Tensor &, - const Tensor &, - std::optional, - int, - int, - std::optional, - float); +class MultiheadAttentionVarlen : public graph::DispatchableGraphOperator { +public: + using schema = void (*)(Tensor, + const Tensor &, + const Tensor &, + const Tensor &, + const Tensor &, + const Tensor &, + std::optional, + int, + int, + std::optional, + float); + using plan_schema = void *(*)(Tensor, + const Tensor &, + const Tensor &, + const Tensor &, + const Tensor &, + const Tensor &, + std::optional, + int, + int, + std::optional, + float); + + static common::OpDispatcher &plan_dispatcher(); + static common::OpDispatcher &run_dispatcher(); + static common::OpDispatcher &cleanup_dispatcher(); + + MultiheadAttentionVarlen(Tensor out, + const Tensor &q, + const Tensor &k, + const Tensor &v, + const Tensor &cum_seqlens_q, + const Tensor &cum_seqlens_kv, + std::optional block_table, + int max_seqlen_q, + int max_seqlen_k, + std::optional alibi_slopes, + float scale); + + static void execute(Tensor out, + const Tensor &q, + const Tensor &k, + const Tensor &v, + const Tensor &cum_seqlens_q, + const Tensor &cum_seqlens_kv, + std::optional block_table, + int max_seqlen_q, + int max_seqlen_k, + std::optional alibi_slopes, + float scale); + + bool is_device_graph_capture_safe() const override { + return device_graph_capture_safe_; + } + +private: + bool device_graph_capture_safe_; +}; Tensor mha_varlen(const Tensor &q, const Tensor &k, diff --git a/csrc/infinicore/src/context/context_impl.cc b/csrc/infinicore/src/context/context_impl.cc index 919ff07b8..03d359599 100644 --- a/csrc/infinicore/src/context/context_impl.cc +++ b/csrc/infinicore/src/context/context_impl.cc @@ -22,9 +22,9 @@ constexpr std::array(Device::Type::kCount)> kD Device::Type::kCpu, }; -void warn_graph_cleanup_failure(const char *operation, const char *detail) noexcept { +void warn_context_cleanup_failure(const char *operation, const char *detail) noexcept { try { - spdlog::warn("{} failed during graph cleanup: {}", operation, detail); + spdlog::warn("{} failed during context cleanup: {}", operation, detail); } catch (...) { } } @@ -48,7 +48,26 @@ std::shared_ptr ContextImpl::getOrCreateRuntimeLocked(Device device, co thread_runtimes.erase(found); } - auto runtime = std::shared_ptr(new Runtime(device)); + std::shared_ptr runtime; + try { + runtime = std::shared_ptr(new Runtime(device)); + } catch (...) { + const auto original_error = std::current_exception(); + if (current_runtime_ != nullptr) { + try { + current_runtime_->activate(); + } catch (const std::exception &error) { + warn_context_cleanup_failure( + "restoring the current runtime after construction failure", + error.what()); + } catch (...) { + warn_context_cleanup_failure( + "restoring the current runtime after construction failure", + "unknown error"); + } + } + std::rethrow_exception(original_error); + } thread_runtimes.emplace(thread_id, runtime); return runtime; } @@ -165,9 +184,9 @@ std::shared_ptr ContextImpl::stopGraphRecording() { try { current_runtime_->activate(); } catch (const std::exception &error) { - warn_graph_cleanup_failure("restoring the previous runtime", error.what()); + warn_context_cleanup_failure("restoring the previous runtime", error.what()); } catch (...) { - warn_graph_cleanup_failure("restoring the previous runtime", "unknown error"); + warn_context_cleanup_failure("restoring the previous runtime", "unknown error"); } std::rethrow_exception(original_error); } @@ -184,9 +203,9 @@ void ContextImpl::cancelGraphRecording() noexcept { try { owner->activate(); } catch (const std::exception &error) { - warn_graph_cleanup_failure("activating the graph runtime", error.what()); + warn_context_cleanup_failure("activating the graph runtime", error.what()); } catch (...) { - warn_graph_cleanup_failure("activating the graph runtime", "unknown error"); + warn_context_cleanup_failure("activating the graph runtime", "unknown error"); } owner->cancelGraphRecording(); @@ -195,9 +214,9 @@ void ContextImpl::cancelGraphRecording() noexcept { try { current_runtime_->activate(); } catch (const std::exception &error) { - warn_graph_cleanup_failure("restoring the previous runtime", error.what()); + warn_context_cleanup_failure("restoring the previous runtime", error.what()); } catch (...) { - warn_graph_cleanup_failure("restoring the previous runtime", "unknown error"); + warn_context_cleanup_failure("restoring the previous runtime", "unknown error"); } } } diff --git a/csrc/infinicore/src/context/runtime/runtime.cc b/csrc/infinicore/src/context/runtime/runtime.cc index e12983764..c967d0ecb 100644 --- a/csrc/infinicore/src/context/runtime/runtime.cc +++ b/csrc/infinicore/src/context/runtime/runtime.cc @@ -42,6 +42,15 @@ Runtime::Runtime(Device device) : device_(device), graph_manager_(std::make_uniq device_memory_allocator_ = std::make_unique(device); pinned_host_memory_allocator_ = std::make_unique(device); } + + const auto stream_status = infini::rt::runtime::StreamCreate(&stream_); + if (stream_status != infini::rt::runtime::kSuccess && stream_ != nullptr) { + warn_runtime_cleanup_failure( + "destroying a partially-created runtime stream", + infini::rt::runtime::StreamDestroy(stream_)); + stream_ = nullptr; + } + INFINICORE_CHECK_ERROR(stream_status); } Runtime::~Runtime() noexcept { Runtime *restore_runtime = ContextImpl::current_runtime_.get(); @@ -50,7 +59,6 @@ Runtime::~Runtime() noexcept { warn_runtime_cleanup_failure("selecting the runtime device", set_device_status); try { - std::lock_guard lock{stream_mutex_}; if (stream_ != nullptr) { const auto synchronize_status = infini::rt::runtime::StreamSynchronize(stream_); warn_runtime_cleanup_failure("synchronizing the runtime stream", synchronize_status); @@ -65,7 +73,6 @@ Runtime::~Runtime() noexcept { pinned_host_memory_allocator_.reset(); device_memory_allocator_.reset(); try { - std::lock_guard lock{stream_mutex_}; if (stream_ != nullptr) { const auto destroy_status = infini::rt::runtime::StreamDestroy(stream_); warn_runtime_cleanup_failure("destroying the runtime stream", destroy_status); @@ -97,13 +104,6 @@ Device Runtime::device() const { } infini::rt::runtime::Stream Runtime::stream() const { - infini::rt::set_runtime_device_type(device_.type()); - INFINICORE_CHECK_ERROR(infini::rt::runtime::SetDevice(device_.index())); - - std::lock_guard lock{stream_mutex_}; - if (stream_ == nullptr) { - INFINICORE_CHECK_ERROR(infini::rt::runtime::StreamCreate(&stream_)); - } return stream_; } @@ -123,7 +123,6 @@ void Runtime::syncStreamForCleanup() noexcept { if (set_device_status == infini::rt::runtime::kSuccess) { try { - std::lock_guard lock{stream_mutex_}; if (stream_ != nullptr) { const auto synchronize_status = infini::rt::runtime::StreamSynchronize(stream_); warn_runtime_cleanup_failure("synchronizing the graph runtime stream", synchronize_status); diff --git a/csrc/infinicore/src/context/runtime/runtime.hpp b/csrc/infinicore/src/context/runtime/runtime.hpp index 5df5eb0e2..af07cfc53 100644 --- a/csrc/infinicore/src/context/runtime/runtime.hpp +++ b/csrc/infinicore/src/context/runtime/runtime.hpp @@ -15,8 +15,7 @@ class ContextImpl; class Runtime : public std::enable_shared_from_this { private: Device device_; - mutable std::mutex stream_mutex_; - mutable infini::rt::runtime::Stream stream_ = nullptr; + infini::rt::runtime::Stream stream_ = nullptr; std::unique_ptr device_memory_allocator_; std::unique_ptr pinned_host_memory_allocator_; std::unique_ptr graph_manager_; diff --git a/csrc/infinicore/src/graph/graph.cc b/csrc/infinicore/src/graph/graph.cc index e60e989a3..85e9c6e95 100644 --- a/csrc/infinicore/src/graph/graph.cc +++ b/csrc/infinicore/src/graph/graph.cc @@ -132,7 +132,14 @@ class StreamCaptureGuard { * GraphTensor * ========================= */ -GraphTensor::GraphTensor(const Tensor &tensor) : Tensor(tensor->to_blob_()) { +GraphTensor::GraphTensor(const Tensor &tensor) + : Tensor(context::isGraphRecording() ? tensor->to_blob_() : tensor) { +} + +GraphTensor::GraphTensor(const Tensor &tensor, SnapshotPolicy policy) + : Tensor(policy == SnapshotPolicy::kBlob || context::isGraphRecording() + ? tensor->to_blob_() + : tensor) { } /* ========================= diff --git a/csrc/infinicore/src/ops/infiniops_impl.hpp b/csrc/infinicore/src/ops/infiniops_impl.hpp index fcf0de543..8ade3606f 100644 --- a/csrc/infinicore/src/ops/infiniops_impl.hpp +++ b/csrc/infinicore/src/ops/infiniops_impl.hpp @@ -103,6 +103,25 @@ infini::ops::Config configForImplementation( template infini::ops::Config defaultConfigForDevice(infini::ops::Device::Type device_type) { + if (device_type == infini::ops::Device::Type::kNvidia) { + static const std::size_t implementation_index = [] { + const auto implementation_indices = Operator::active_implementation_indices( + infini::ops::Device::Type::kNvidia); + if (implementation_indices.empty()) { + throw std::runtime_error( + "InfiniOps operator has no active implementation for device '" + + std::string(infini::ops::Device::StringFromType( + infini::ops::Device::Type::kNvidia)) + + "'."); + } + return implementation_indices.front(); + }(); + + infini::ops::Config config; + config.set_implementation_index(implementation_index); + return config; + } + const auto implementation_indices = Operator::active_implementation_indices(device_type); if (implementation_indices.empty()) { throw std::runtime_error( diff --git a/csrc/infinicore/src/ops/mha_kvcache/mha_kvcache.cc b/csrc/infinicore/src/ops/mha_kvcache/mha_kvcache.cc index 5d24d8e09..12fdf2b74 100644 --- a/csrc/infinicore/src/ops/mha_kvcache/mha_kvcache.cc +++ b/csrc/infinicore/src/ops/mha_kvcache/mha_kvcache.cc @@ -5,6 +5,113 @@ namespace infinicore::op { INFINICORE_GRAPH_OP_DISPATCHERS_IMPL(MhaKVCache); +bool MhaKVCache::supports_device_graph_capture( + const Tensor &out, + const Tensor &q, + const Tensor &k_cache, + const Tensor &v_cache, + const Tensor &seqlens_k, + const Tensor &block_table, + const std::optional &alibi_slopes) { + if (!out + || !q + || !k_cache + || !v_cache + || !seqlens_k + || !block_table + || alibi_slopes.has_value()) { + return false; + } + + const auto device = out->device(); + const auto same_device = [&](const Tensor &tensor) { + return tensor->device() == device; + }; + if (out->ndim() != 4 + || q->ndim() != 4 + || k_cache->ndim() != 4 + || v_cache->ndim() != 4 + || seqlens_k->ndim() != 1 + || block_table->ndim() != 2) { + return false; + } + + // Keep graph capture limited to shapes validated with the persistent + // InfiniOps provider. Other decode shapes retain the eager fallback. + const bool p12_shape = q->shape() == Shape({1, 1, 32, 128}) + && out->shape() == Shape({1, 1, 32, 128}) + && k_cache->shape() == Shape({512, 256, 2, 128}) + && seqlens_k->shape() == Shape({1}) + && block_table->shape() == Shape({1, 8}); + const bool p13_shape = q->shape() == Shape({1, 1, 32, 128}) + && k_cache->size(1) == 256 + && k_cache->size(2) == 2 + && k_cache->size(3) == 128 + && seqlens_k->shape() == Shape({1}) + && block_table->shape() == Shape({1, 8}); + const bool p09_shape = q->shape() == Shape({16, 1, 24, 128}) + && k_cache->size(0) == 128 + && k_cache->size(1) == 256 + && k_cache->size(2) == 8 + && k_cache->size(3) == 128 + && seqlens_k->shape() == Shape({16}) + && block_table->shape() == Shape({16, 128}); + const bool p11_shape = q->shape() == Shape({1, 1, 16, 128}) + && k_cache->size(0) == 1 + && k_cache->size(1) == 256 + && k_cache->size(2) == 16 + && k_cache->size(3) == 128 + && seqlens_k->shape() == Shape({1}) + && block_table->shape() == Shape({1, 1}); + const bool p14_shape = q->shape() == Shape({16, 1, 32, 128}) + && k_cache->size(0) == 512 + && k_cache->size(1) == 256 + && k_cache->size(2) == 2 + && k_cache->size(3) == 128 + && seqlens_k->shape() == Shape({16}) + && block_table->shape() == Shape({16, 512}); + const bool p13_layout = q->is_contiguous() && out->is_contiguous(); + // P09 reads Q directly from the fused QKV projection while output is dense. + const bool p09_layout = q->strides() == Strides({5120, 5120, 128, 1}) + && out->strides() == Strides({3072, 3072, 128, 1}); + // P11 has the same fused-QKV view pattern at batch size one. + const bool p11_layout = q->strides() == Strides({6144, 6144, 128, 1}) + && out->strides() == Strides({2048, 2048, 128, 1}); + // P14 reads Q from MiniCPM4's fused QKV projection at batch size 16. + const bool p14_layout = q->strides() == Strides({4608, 4608, 128, 1}) + && out->strides() == Strides({4096, 4096, 128, 1}); + // P12 has the same fused QKV projection layout at batch size one. + const bool p12_layout = q->strides() == Strides({4608, 4608, 128, 1}) + && out->strides() == Strides({4096, 4096, 128, 1}); + const bool reviewed_shape_and_layout = (p12_shape && p12_layout) + || (p13_shape && p13_layout) + || (p09_shape && p09_layout) + || (p11_shape && p11_layout) + || (p14_shape && p14_layout); + const auto dtype = q->dtype(); + return device.type() == Device::Type::kNvidia + && same_device(q) + && same_device(k_cache) + && same_device(v_cache) + && same_device(seqlens_k) + && same_device(block_table) + && reviewed_shape_and_layout + && out->shape() == q->shape() + && k_cache->size(0) > 0 + && v_cache->shape() == k_cache->shape() + && (dtype == DataType::kFloat16 + || dtype == DataType::kBFloat16) + && out->dtype() == dtype + && k_cache->dtype() == dtype + && v_cache->dtype() == dtype + && seqlens_k->dtype() == DataType::kInt32 + && block_table->dtype() == DataType::kInt32 + && k_cache->is_contiguous() + && v_cache->is_contiguous() + && seqlens_k->is_contiguous() + && block_table->is_contiguous(); +} + MhaKVCache::MhaKVCache(Tensor out, const Tensor &q, const Tensor &k_cache, @@ -12,7 +119,17 @@ MhaKVCache::MhaKVCache(Tensor out, const Tensor &seqlens_k, const Tensor &block_table, std::optional alibi_slopes, - float scale) { + float scale) + : device_graph_capture_safe_( + context::isGraphRecording() + && supports_device_graph_capture( + out, + q, + k_cache, + v_cache, + seqlens_k, + block_table, + alibi_slopes)) { INFINICORE_ASSERT_TENSORS_SAME_DEVICE(out, q, k_cache, v_cache, seqlens_k, block_table); INFINICORE_GRAPH_OP_DISPATCH(out->device().type(), out, q, k_cache, v_cache, seqlens_k, block_table, alibi_slopes, scale); diff --git a/csrc/infinicore/src/ops/mha_kvcache/mha_kvcache_infiniops.cc b/csrc/infinicore/src/ops/mha_kvcache/mha_kvcache_infiniops.cc index 62ee56e38..1f00ef4ce 100644 --- a/csrc/infinicore/src/ops/mha_kvcache/mha_kvcache_infiniops.cc +++ b/csrc/infinicore/src/ops/mha_kvcache/mha_kvcache_infiniops.cc @@ -7,12 +7,14 @@ #include #include +#include #include namespace infinicore::op::mha_kvcache_impl::infiniops { namespace { using TensorMeta = ::infinicore::op::infiniops::TensorMeta; +using FlashAttnOperator = infini::ops::Operator; // TODO: Remove backend-specific implementation indices from InfiniLM once // InfiniOps provides device-aware default selection for these operators. @@ -109,8 +111,57 @@ struct PlannedMeta { seqlens_k_tensor, block_table_tensor; std::optional alibi_slopes_tensor; float scale; + std::unique_ptr graph_safe_provider; }; +std::unique_ptr make_graph_safe_provider( + PlannedMeta &planned) { + const auto config = ::infinicore::op::infiniops::configForImplementation< + infini::ops::FlashAttnWithKvcache>( + infini::ops::Device::Type::kNvidia, 17); + + const auto q = planned.q.tensor(planned.q_tensor); + const auto k_cache = planned.k_cache.tensor(planned.k_cache_tensor); + const auto v_cache = planned.v_cache.tensor(planned.v_cache_tensor); + const std::optional no_tensor; + const std::optional cache_seqlens{ + planned.seqlens_k.tensor(planned.seqlens_k_tensor)}; + const std::optional block_table{ + planned.block_table.tensor(planned.block_table_tensor)}; + const std::optional softmax_scale{planned.scale}; + const bool causal = true; + const std::vector window_size{-1, -1}; + const double softcap = 0.0; + const bool rotary_interleaved = true; + const std::int64_t num_splits = 0; + const bool return_softmax_lse = false; + const auto out = planned.out.tensor(planned.out_tensor); + + return FlashAttnOperator::Make( + config, + q, + k_cache, + v_cache, + no_tensor, + no_tensor, + no_tensor, + no_tensor, + cache_seqlens, + no_tensor, + no_tensor, + block_table, + no_tensor, + softmax_scale, + causal, + window_size, + softcap, + rotary_interleaved, + num_splits, + return_softmax_lse, + out, + no_tensor); +} + } // namespace void *plan(Tensor out, @@ -123,7 +174,7 @@ void *plan(Tensor out, float scale) { INFINICORE_ASSERT(is_supported( out, q, k_cache, v_cache, seqlens_k, block_table, alibi_slopes)); - return new PlannedMeta{ + auto planned = std::unique_ptr{new PlannedMeta{ TensorMeta(out), TensorMeta(q), TensorMeta(k_cache), @@ -141,18 +192,30 @@ void *plan(Tensor out, alibi_slopes ? std::optional{graph::GraphTensor(*alibi_slopes)} : std::nullopt, - scale}; + scale, + nullptr}}; + if (context::isGraphRecording() + && MhaKVCache::supports_device_graph_capture( + out, + q, + k_cache, + v_cache, + seqlens_k, + block_table, + alibi_slopes)) { + planned->graph_safe_provider = make_graph_safe_provider(*planned); + } + return planned.release(); } void run(void *planned_meta) { auto *planned = reinterpret_cast(planned_meta); infini::ops::Handle handle; handle.set_stream(context::getStream()); - const auto device_type = planned->q.device.type(); - const auto implementation_index = implementation_index_for_device(device_type); - auto config = ::infinicore::op::infiniops::configForImplementation< - infini::ops::FlashAttnWithKvcache>(device_type, implementation_index); + const auto q = planned->q.tensor(planned->q_tensor); + const auto k_cache = planned->k_cache.tensor(planned->k_cache_tensor); + const auto v_cache = planned->v_cache.tensor(planned->v_cache_tensor); const std::optional no_tensor; const std::optional cache_seqlens{ planned->seqlens_k.tensor(planned->seqlens_k_tensor)}; @@ -162,13 +225,54 @@ void run(void *planned_meta) { ? std::optional{ planned->alibi_slopes->tensor(*planned->alibi_slopes_tensor)} : std::nullopt; + const std::optional softmax_scale{planned->scale}; + const bool causal = true; + const std::vector window_size{-1, -1}; + const double softcap = 0.0; + const bool rotary_interleaved = true; + const std::int64_t num_splits = 0; + const bool return_softmax_lse = false; + const auto out = planned->out.tensor(planned->out_tensor); + + if (planned->graph_safe_provider) { + (*planned->graph_safe_provider)( + handle, + q, + k_cache, + v_cache, + no_tensor, + no_tensor, + no_tensor, + no_tensor, + cache_seqlens, + no_tensor, + no_tensor, + block_table, + no_tensor, + softmax_scale, + causal, + window_size, + softcap, + rotary_interleaved, + num_splits, + return_softmax_lse, + out, + no_tensor); + return; + } + + const auto device_type = planned->q.device.type(); + const auto implementation_index = implementation_index_for_device(device_type); + const auto config = ::infinicore::op::infiniops::configForImplementation< + infini::ops::FlashAttnWithKvcache>( + device_type, implementation_index); infini::ops::FlashAttnWithKvcache::Call( handle, config, - planned->q.tensor(planned->q_tensor), - planned->k_cache.tensor(planned->k_cache_tensor), - planned->v_cache.tensor(planned->v_cache_tensor), + q, + k_cache, + v_cache, no_tensor, no_tensor, no_tensor, @@ -178,14 +282,14 @@ void run(void *planned_meta) { no_tensor, block_table, alibi_slopes, - std::optional{planned->scale}, - true, - std::vector{-1, -1}, - 0.0, - true, - std::int64_t{0}, - false, - planned->out.tensor(planned->out_tensor), + softmax_scale, + causal, + window_size, + softcap, + rotary_interleaved, + num_splits, + return_softmax_lse, + out, no_tensor); } diff --git a/csrc/infinicore/src/ops/multi_head_attention_varlen/mha_varlen.cc b/csrc/infinicore/src/ops/multi_head_attention_varlen/mha_varlen.cc index f6d4612bd..c06cbc40e 100644 --- a/csrc/infinicore/src/ops/multi_head_attention_varlen/mha_varlen.cc +++ b/csrc/infinicore/src/ops/multi_head_attention_varlen/mha_varlen.cc @@ -15,7 +15,9 @@ MultiheadAttentionVarlen::MultiheadAttentionVarlen(Tensor out, int max_seqlen_q, int max_seqlen_k, std::optional alibi_slopes, - float scale) { + float scale) + : device_graph_capture_safe_( + out->device().type() != Device::Type::kNvidia) { INFINICORE_ASSERT_TENSORS_SAME_DEVICE(out, q, k, v, cum_seqlens_q, cum_seqlens_kv); if (block_table.has_value()) { INFINICORE_ASSERT_TENSORS_SAME_DEVICE(out, block_table.value()); diff --git a/csrc/infinicore/src/ops/paged_caching/paged_caching_infiniops.cc b/csrc/infinicore/src/ops/paged_caching/paged_caching_infiniops.cc index 593e21c8d..04b7fb945 100644 --- a/csrc/infinicore/src/ops/paged_caching/paged_caching_infiniops.cc +++ b/csrc/infinicore/src/ops/paged_caching/paged_caching_infiniops.cc @@ -21,11 +21,9 @@ void *plan(Tensor k_cache, Tensor v_cache, const Tensor &k, const Tensor &v, con INFINICORE_ASSERT(::infinicore::op::infiniops::isSupportedDevice(k_cache->device().type())); INFINICORE_ASSERT_TENSORS_SAME_DEVICE(k_cache, v_cache, k, v, slot_mapping); - // The canonical API requires valid scales even though the "auto" path does - // not apply quantization. + // The canonical "auto" path requires valid scale metadata, but its + // backends do not read scale values. auto scale = Tensor::empty({1}, DataType::kFloat32, k_cache->device()); - constexpr float one = 1.0f; - context::memcpyH2D(scale->data(), &one, sizeof(one), false); return new PlannedMeta{ TensorMeta(k), TensorMeta(v), TensorMeta(slot_mapping), TensorMeta(scale), TensorMeta(k_cache), TensorMeta(v_cache), graph::GraphTensor(k), graph::GraphTensor(v), graph::GraphTensor(slot_mapping), graph::GraphTensor(scale), graph::GraphTensor(k_cache), graph::GraphTensor(v_cache), diff --git a/csrc/infinicore/src/ops/random_sample/random_sample.cc b/csrc/infinicore/src/ops/random_sample/random_sample.cc index 300a783a9..8a496db89 100644 --- a/csrc/infinicore/src/ops/random_sample/random_sample.cc +++ b/csrc/infinicore/src/ops/random_sample/random_sample.cc @@ -17,6 +17,8 @@ bool tryGreedyWithInfiniOps( float random_value, float top_p, int top_k, float temperature) { const auto dtype = logits->dtype(); const auto device_type = logits->device().type(); + const bool batched = logits->ndim() == 2; + const size_t num_rows = batched ? logits->size(0) : 1; if ((device_type != Device::Type::kNvidia && device_type != Device::Type::kMetax && device_type != Device::Type::kIluvatar @@ -27,11 +29,13 @@ bool tryGreedyWithInfiniOps( && top_p != 0.0f && top_k != 1 && temperature != 0.0f) - || logits->ndim() != 1 + || (logits->ndim() != 1 + && !(device_type == Device::Type::kNvidia && batched)) || logits->numel() == 0 || !logits->is_contiguous() || (dtype != DataType::kFloat16 && dtype != DataType::kBFloat16 && dtype != DataType::kFloat32) - || indices->numel() != 1 + || indices->numel() != num_rows + || (batched && (indices->ndim() != 1 || indices->size(0) != num_rows)) || indices->dtype() != DataType::kInt64 || !indices->is_contiguous()) { return false; @@ -43,12 +47,14 @@ bool tryGreedyWithInfiniOps( const infiniops::TensorMeta indices_meta(indices); auto argmax_config = infiniops::defaultConfigForDevice( logits_meta.device.type()); - const std::optional no_dim; + const std::optional dim = batched + ? std::optional{1} + : std::nullopt; infini::ops::Argmax::Call( handle, argmax_config, logits_meta.tensor(logits), - no_dim, + dim, false, indices_meta.tensor(indices)); return true; diff --git a/csrc/infinicore/src/ops/silu_and_mul/silu_and_mul.cc b/csrc/infinicore/src/ops/silu_and_mul/silu_and_mul.cc index 04bc975cd..ff7a674b1 100644 --- a/csrc/infinicore/src/ops/silu_and_mul/silu_and_mul.cc +++ b/csrc/infinicore/src/ops/silu_and_mul/silu_and_mul.cc @@ -29,7 +29,32 @@ Tensor silu_and_mul(const Tensor &x) { } void silu_and_mul_(Tensor out, const Tensor &x) { - SiluAndMul::execute(out, x); + constexpr Size MAX_ELEMENTS_PER_LAUNCH = Size{1} << 30; + if (out->numel() <= MAX_ELEMENTS_PER_LAUNCH + || !out->is_contiguous() + || !x->is_contiguous()) { + SiluAndMul::execute(out, x); + return; + } + + const Size output_row_width = out->size(out->ndim() - 1); + const Size input_row_width = x->size(x->ndim() - 1); + INFINICORE_ASSERT(output_row_width > 0 + && output_row_width <= MAX_ELEMENTS_PER_LAUNCH); + INFINICORE_ASSERT(input_row_width == output_row_width * 2); + INFINICORE_ASSERT(x->numel() == out->numel() * 2); + + const Size num_rows = out->numel() / output_row_width; + const Size max_rows = MAX_ELEMENTS_PER_LAUNCH / output_row_width; + auto output_rows = out->view({num_rows, output_row_width}); + auto input_rows = x->view({num_rows, input_row_width}); + + for (Size start = 0; start < num_rows; start += max_rows) { + const Size remaining = num_rows - start; + const Size rows = remaining < max_rows ? remaining : max_rows; + SiluAndMul::execute(output_rows->narrow({{0, start, rows}}), + input_rows->narrow({{0, start, rows}})); + } } } // namespace infinicore::op diff --git a/csrc/layers/attention/backends/static_attn.cpp b/csrc/layers/attention/backends/static_attn.cpp index dab4476b9..3f24405b9 100644 --- a/csrc/layers/attention/backends/static_attn.cpp +++ b/csrc/layers/attention/backends/static_attn.cpp @@ -6,6 +6,23 @@ #include "infinicore/ops/per_tensor_quant_i8.hpp" namespace infinilm::layers::attention::backends { +namespace { + +size_t sequence_length_from_metadata( + const std::optional &snapshot, + const std::optional &lengths) { + if (snapshot.has_value()) { + return snapshot.value(); + } + + ASSERT(lengths.has_value()); + return reinterpret_cast( + lengths.value() + ->to(infinicore::Device{infinicore::Device::Type::kCpu}) + ->data())[0]; +} + +} // namespace StaticAttentionImpl::StaticAttentionImpl(size_t num_heads, size_t head_size, @@ -48,20 +65,23 @@ infinicore::Tensor StaticAttentionImpl::forward(const AttentionLayer &layer, size_t seq_len = shape[2]; size_t value_head_dim = v_reshaped->size(3); - auto past_sequence_lengths = attn_metadata.past_sequence_lengths; - auto total_sequence_lengths = attn_metadata.total_sequence_lengths; - if (infinicore::context::isGraphRecording()) { ASSERT(this->kv_quant_scheme_ == infinilm::quantization::KVQuantAlgo::NONE); return forward_graph_(q_reshaped, k_permuted, v_permuted, kv_cache, attn_metadata); } + const size_t cache_pos = sequence_length_from_metadata( + attn_metadata.first_past_sequence_length, + attn_metadata.past_sequence_lengths); + const size_t total_seq_len = sequence_length_from_metadata( + attn_metadata.first_total_sequence_length, + attn_metadata.total_sequence_lengths); + // update static kv cache // k_total: [bs, n_kv_head, max_seq_len, head_dim] // v_total : [bs, n_kv_head, max_seq_len, head_dim] - auto [k_total, v_total] = do_kv_cache_update(layer, k_permuted, v_permuted, kv_cache, past_sequence_lengths.value()); - - size_t total_seq_len = reinterpret_cast(total_sequence_lengths.value()->to(infinicore::Device{infinicore::Device::Type::kCpu})->data())[0]; + auto [k_total, v_total] = do_kv_cache_update( + layer, k_permuted, v_permuted, kv_cache, cache_pos); if (infinilm::quantization::KVQuantAlgo::NONE != this->kv_quant_scheme_) { infinilm::KVQuantUtils::dequantize( @@ -138,7 +158,7 @@ std::tuple StaticAttentionImpl::do_kv_ca const infinicore::Tensor key, const infinicore::Tensor value, infinicore::Tensor &kv_cache, - const infinicore::Tensor past_sequence_lengths) const { + size_t cache_pos) const { auto batch_size = key->size(0); auto update_len = key->size(2); @@ -152,7 +172,6 @@ std::tuple StaticAttentionImpl::do_kv_ca ASSERT_EQ(batch_size, max_batch_size); - size_t cache_pos = reinterpret_cast(past_sequence_lengths->to(infinicore::Device{infinicore::Device::Type::kCpu})->data())[0]; auto result_len = cache_pos + update_len; ASSERT(result_len <= max_seq_len); diff --git a/csrc/layers/attention/backends/static_attn.hpp b/csrc/layers/attention/backends/static_attn.hpp index cb9e50436..c2ca4ecc3 100644 --- a/csrc/layers/attention/backends/static_attn.hpp +++ b/csrc/layers/attention/backends/static_attn.hpp @@ -31,7 +31,7 @@ class StaticAttentionImpl { const infinicore::Tensor key, const infinicore::Tensor value, infinicore::Tensor &kv_cache, - const infinicore::Tensor past_sequence_lengths) const; + size_t cache_pos) const; private: infinicore::Tensor forward_graph_( diff --git a/csrc/layers/mlp/mlp.cpp b/csrc/layers/mlp/mlp.cpp index f7604c505..68cc9ffef 100644 --- a/csrc/layers/mlp/mlp.cpp +++ b/csrc/layers/mlp/mlp.cpp @@ -27,12 +27,10 @@ MLP::MLP(std::shared_ptr model_config, } infinicore::Tensor MLP::forward(const infinicore::Tensor &hidden_states) const { - // 1. Project to gate and up auto hidden_states_mutable = hidden_states; - auto [gate, up] = gate_up_proj_->forward_split(hidden_states_mutable); - // 2. Apply SwiGLU: silu(gate) * up - auto intermediate = infinicore::op::swiglu(up, gate); - // 3. Project down + // GateUpParallelLinear produces the packed [gate, up] layout expected here. + auto gate_up = gate_up_proj_->forward(hidden_states_mutable); + auto intermediate = infinicore::op::silu_and_mul(gate_up); auto output = down_proj_->forward(intermediate); return output; } diff --git a/csrc/models/qwen3_next/qwen3_next_sparse_moe_block.cpp b/csrc/models/qwen3_next/qwen3_next_sparse_moe_block.cpp index d3548c9fa..db51db46e 100644 --- a/csrc/models/qwen3_next/qwen3_next_sparse_moe_block.cpp +++ b/csrc/models/qwen3_next/qwen3_next_sparse_moe_block.cpp @@ -7,7 +7,7 @@ #include #include #include -#include +#include #include @@ -48,8 +48,8 @@ Qwen3NextSharedExpert::Qwen3NextSharedExpert(std::shared_ptrforward_split(hidden_states_mutable); - auto intermediate = infinicore::op::swiglu(up, gate); + auto gate_up = gate_up_proj_->forward(hidden_states_mutable); + auto intermediate = infinicore::op::silu_and_mul(gate_up); return down_proj_->forward(intermediate); } diff --git a/python/infinilm/infer_engine.py b/python/infinilm/infer_engine.py index 117d82f9a..d73d20b82 100644 --- a/python/infinilm/infer_engine.py +++ b/python/infinilm/infer_engine.py @@ -18,6 +18,68 @@ } +class _PagedDecodeMetadataBuffers: + def __init__(self, batch_size: int, slot_stride: int): + import numpy as np + + self._np = np + self._storage_views = [] + self.position_ids, self._position_ids_values = self._empty( + [batch_size], infinicore.int64, np.int64 + ) + self.slot_mapping, self._slot_mapping_values = self._empty( + [batch_size], infinicore.int64, np.int64 + ) + self.past_kv_lengths, self._past_kv_lengths_values = self._empty( + [batch_size], infinicore.int32, np.int32 + ) + self.total_kv_lengths, self._total_kv_lengths_values = self._empty( + [batch_size], infinicore.int32, np.int32 + ) + self.cu_seqlens, self._cu_seqlens_values = self._empty( + [batch_size + 1], infinicore.int32, np.int32 + ) + self.input_offsets, self._input_offsets_values = self._empty( + [batch_size + 1], infinicore.int32, np.int32 + ) + + self._slot_bases = np.arange(batch_size, dtype=np.int64) * slot_stride + self._sequence_indices = np.arange(batch_size + 1, dtype=np.int32) + self._input_offsets_values[:] = self._sequence_indices + + def _empty(self, shape, dtype, numpy_dtype): + tensor = infinicore.empty(shape, dtype=dtype) + scalar_type = self._np.ctypeslib.as_ctypes_type(self._np.dtype(numpy_dtype)) + storage = (scalar_type * tensor.numel()).from_address(tensor.data_ptr()) + values = self._np.ctypeslib.as_array(storage).reshape(shape) + self._storage_views.append(storage) + return tensor, values + + def update(self, past_seq_len: int): + total_seq_len = past_seq_len + 1 + self._position_ids_values.fill(past_seq_len) + self._np.add( + self._slot_bases, + past_seq_len, + out=self._slot_mapping_values, + ) + self._past_kv_lengths_values.fill(past_seq_len) + self._total_kv_lengths_values.fill(total_seq_len) + self._np.multiply( + self._sequence_indices, + total_seq_len, + out=self._cu_seqlens_values, + ) + return ( + self.position_ids, + self.slot_mapping, + self.past_kv_lengths, + self.total_kv_lengths, + self.cu_seqlens, + self.input_offsets, + ) + + def _apply_torch_dtype_defaults(config: dict) -> dict: if config.get("torch_dtype") is None: config["torch_dtype"] = config.get("dtype") or _MODEL_DEFAULTS.get( @@ -571,15 +633,38 @@ def generate( dtype=infinicore.int32, ) + decode_metadata = None for iter in range(0, generation_config.max_new_tokens): if _measure_and_log_time: start_time = time.perf_counter() batch_size, seq_len = input_ids.shape[:2] + reuse_decode_metadata = ( + self.enable_paged_attn + and iter > 0 + and seq_len == 1 + and prompt_position_ids is None + and self.position_id_axes == 1 + and mamba_state_indices is None + ) if self.enable_paged_attn: input_ids = input_ids.view([1, batch_size * seq_len]) - if prompt_position_ids is not None: + if reuse_decode_metadata: + if decode_metadata is None: + decode_metadata = _PagedDecodeMetadataBuffers( + batch_size, + max_blocks_per_batch * paged_block_size, + ) + ( + position_ids, + slot_mapping, + past_kv_lengths, + total_kv_lengths, + cu_seqlens, + input_offsets, + ) = decode_metadata.update(past_seq_len) + elif prompt_position_ids is not None: if iter == 0: position_ids_list = [ list(axis) * batch_size for axis in prompt_position_ids @@ -605,35 +690,36 @@ def generate( position_ids_list = [ position_ids_list for _ in range(self.position_id_axes) ] - position_ids = infinicore.from_list( - position_ids_list, dtype=infinicore.int64 - ) + if not reuse_decode_metadata: + position_ids = infinicore.from_list( + position_ids_list, dtype=infinicore.int64 + ) - if iter == 0: - slot_mapping_list = [] - for b in range(batch_size): - slot_mapping_list.extend( - [ - b * max_blocks_per_batch * paged_block_size + i - for i in range(seq_len) - ] - ) - else: - slot_mapping_list = [ - i - for i in range( - past_seq_len, - max_blocks_per_batch - * paged_block_size - * initial_batch_size, - max_blocks_per_batch * paged_block_size, - ) - ] + if iter == 0: + slot_mapping_list = [] + for b in range(batch_size): + slot_mapping_list.extend( + [ + b * max_blocks_per_batch * paged_block_size + i + for i in range(seq_len) + ] + ) + else: + slot_mapping_list = [ + i + for i in range( + past_seq_len, + max_blocks_per_batch + * paged_block_size + * initial_batch_size, + max_blocks_per_batch * paged_block_size, + ) + ] - slot_mapping = infinicore.from_list( - slot_mapping_list, - dtype=infinicore.int64, - ) + slot_mapping = infinicore.from_list( + slot_mapping_list, + dtype=infinicore.int64, + ) else: position_ids = infinicore.from_list( [ @@ -645,19 +731,22 @@ def generate( slot_mapping = None - past_kv_lengths = infinicore.from_list( - [past_seq_len] * batch_size, dtype=infinicore.int32 - ) - total_kv_lengths = infinicore.from_list( - [past_seq_len + seq_len] * batch_size, dtype=infinicore.int32 - ) - cu_seqlens = infinicore.from_list( - [(past_seq_len + seq_len) * i for i in range(batch_size + 1)], - dtype=infinicore.int32, - ) - input_offsets = infinicore.from_list( - [seq_len * i for i in range(batch_size + 1)], dtype=infinicore.int32 - ) + if not reuse_decode_metadata: + past_kv_lengths = infinicore.from_list( + [past_seq_len] * batch_size, dtype=infinicore.int32 + ) + total_kv_lengths = infinicore.from_list( + [past_seq_len + seq_len] * batch_size, + dtype=infinicore.int32, + ) + cu_seqlens = infinicore.from_list( + [(past_seq_len + seq_len) * i for i in range(batch_size + 1)], + dtype=infinicore.int32, + ) + input_offsets = infinicore.from_list( + [seq_len * i for i in range(batch_size + 1)], + dtype=infinicore.int32, + ) mamba_init_state_indices = None mamba_final_state_indices = None diff --git a/python/infinilm/llm/model_runner/speculative_runner.py b/python/infinilm/llm/model_runner/speculative_runner.py index d3c5211d4..d36f26d9f 100644 --- a/python/infinilm/llm/model_runner/speculative_runner.py +++ b/python/infinilm/llm/model_runner/speculative_runner.py @@ -1,4 +1,5 @@ import infinicore + from infinilm.cache.cache import StaticKVCacheConfig from infinilm.distributed import DistConfig from infinilm.infer_engine import InferEngine @@ -56,7 +57,9 @@ def forward(self, scheduler_output, model_input): if not requests: return [] - target_output = self.target_model_engine.forward_raw(**model_input) + target_model_input = dict(model_input) + target_model_input["sample_all_positions"] = scheduler_output.is_prefill + target_output = self.target_model_engine.forward_raw(**target_model_input) target_token_ids = target_output["output_ids"].to_numpy().tolist() if not target_token_ids: return target_token_ids @@ -253,6 +256,7 @@ def _draft_eagle_tokens_batch(self, jobs: list[dict]) -> list[list[int]]: temperature=1.0, top_k=1, top_p=1.0, + sample_all_positions=False, ) token_ids = draft_output["output_ids"].to_numpy().tolist() draft_hidden = draft_output["hidden_states"] @@ -277,6 +281,12 @@ def _build_paged_verify_batch_input(self, candidates: list[dict]) -> dict: max_block_table_len = max( len(candidate["req"].block_table) for candidate in candidates ) + # A one-token candidate only needs its correction token. Selecting the + # last position keeps that case eligible for the compiled decode graph; + # longer candidates still need every intermediate prediction. + sample_all_positions = any( + len(candidate["draft_tokens"]) != 1 for candidate in candidates + ) for candidate in candidates: req = candidate["req"] @@ -307,4 +317,5 @@ def _build_paged_verify_batch_input(self, candidates: list[dict]) -> dict: "temperature": 1.0, "top_k": 1, "top_p": 1.0, + "sample_all_positions": sample_all_positions, } diff --git a/test/bench/bench_context_runtime.py b/test/bench/bench_context_runtime.py new file mode 100644 index 000000000..8de74757a --- /dev/null +++ b/test/bench/bench_context_runtime.py @@ -0,0 +1,59 @@ +import argparse +import json +import statistics +import time + +import infinicore + + +def measure_get_stream(iterations): + get_stream = infinicore.get_stream + start = time.perf_counter_ns() + for _ in range(iterations): + get_stream() + return (time.perf_counter_ns() - start) / iterations + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--device", default="cuda:0") + parser.add_argument("--warmup", type=int, default=10_000) + parser.add_argument("--iterations", type=int, default=1_000_000) + parser.add_argument("--repeat", type=int, default=7) + args = parser.parse_args() + + if args.warmup < 0 or args.iterations <= 0 or args.repeat <= 0: + parser.error( + "warmup must be non-negative; iterations and repeat must be positive" + ) + + infinicore.set_device(args.device) + expected_stream = infinicore.get_stream() + for _ in range(args.warmup): + if infinicore.get_stream() != expected_stream: + raise RuntimeError("runtime stream changed during warmup") + + samples = [measure_get_stream(args.iterations) for _ in range(args.repeat)] + if infinicore.get_stream() != expected_stream: + raise RuntimeError("runtime stream changed during benchmark") + + print( + json.dumps( + { + "benchmark": "infinicore.get_stream", + "device": args.device, + "iterations": args.iterations, + "repeat": args.repeat, + "unit": "ns/call", + "min": min(samples), + "median": statistics.median(samples), + "max": max(samples), + "samples": samples, + }, + sort_keys=True, + ) + ) + + +if __name__ == "__main__": + main() diff --git a/test/runtime/test_context_runtime.py b/test/runtime/test_context_runtime.py new file mode 100644 index 000000000..e0e656e87 --- /dev/null +++ b/test/runtime/test_context_runtime.py @@ -0,0 +1,117 @@ +import threading +import unittest + +import infinicore +import torch + + +class RuntimeStreamTest(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.device_count = infinicore.get_device_count("cuda") + if cls.device_count == 0: + raise unittest.SkipTest("NVIDIA device is required") + + def tearDown(self): + infinicore.set_device("cuda:0") + + def test_stream_is_stable_on_one_device(self): + infinicore.set_device("cuda:0") + + first = infinicore.get_stream() + second = infinicore.get_stream() + infinicore.zeros((4,), device="cuda:0") + infinicore.sync_stream() + + self.assertNotEqual(first, 0) + self.assertEqual(first, second) + self.assertEqual(torch.cuda.current_device(), 0) + + def test_switching_devices_reuses_each_runtime_stream(self): + if self.device_count < 2: + self.skipTest("two NVIDIA devices are required") + + infinicore.set_device("cuda:0") + stream_0 = infinicore.get_stream() + keep_runtime_0 = infinicore.zeros((4,), device="cuda:0") + infinicore.sync_stream() + + infinicore.set_device("cuda:1") + stream_1 = infinicore.get_stream() + keep_runtime_1 = infinicore.zeros((4,), device="cuda:1") + infinicore.sync_stream() + + infinicore.set_device("cuda:0") + self.assertEqual(infinicore.get_stream(), stream_0) + self.assertNotEqual(stream_0, stream_1) + self.assertEqual(torch.cuda.current_device(), 0) + self.assertIsNotNone(keep_runtime_0) + self.assertIsNotNone(keep_runtime_1) + + def test_threads_own_distinct_streams_on_one_device(self): + results = self._collect_thread_streams((0, 0)) + + self.assertNotEqual(results[0][0], results[1][0]) + self.assertEqual(results[0][1], 0) + self.assertEqual(results[1][1], 0) + + def test_threads_select_devices_independently(self): + if self.device_count < 2: + self.skipTest("two NVIDIA devices are required") + + infinicore.set_device("cuda:0") + results = self._collect_thread_streams((0, 1)) + + self.assertEqual(results[0][1], 0) + self.assertEqual(results[1][1], 1) + self.assertEqual(torch.cuda.current_device(), 0) + + def test_explicit_set_device_restores_external_cuda_switch(self): + if self.device_count < 2: + self.skipTest("two NVIDIA devices are required") + + infinicore.set_device("cuda:0") + torch.cuda.set_device(1) + self.assertEqual(torch.cuda.current_device(), 1) + + infinicore.set_device("cuda:0") + self.assertEqual(torch.cuda.current_device(), 0) + + def _collect_thread_streams(self, device_indices): + barrier = threading.Barrier(len(device_indices)) + results = [None] * len(device_indices) + errors = [] + + def worker(slot, device_index): + try: + target = f"cuda:{device_index}" + infinicore.set_device(target) + first = infinicore.get_stream() + keep_runtime = infinicore.zeros((4,), device=target) + infinicore.sync_stream() + barrier.wait(timeout=30) + second = infinicore.get_stream() + results[slot] = (second, torch.cuda.current_device()) + self.assertEqual(first, second) + self.assertIsNotNone(keep_runtime) + except BaseException as error: + errors.append(error) + barrier.abort() + + threads = [ + threading.Thread(target=worker, args=(slot, device_index)) + for slot, device_index in enumerate(device_indices) + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=40) + + self.assertFalse(any(thread.is_alive() for thread in threads)) + if errors: + raise errors[0] + return results + + +if __name__ == "__main__": + unittest.main() diff --git a/test/runtime/test_packed_swiglu.py b/test/runtime/test_packed_swiglu.py new file mode 100644 index 000000000..c8f315a2a --- /dev/null +++ b/test/runtime/test_packed_swiglu.py @@ -0,0 +1,102 @@ +import ctypes +import unittest + +import infinicore +import torch +import torch.nn.functional as torch_functional +from infinicore import ops + + +def to_torch_bfloat16(tensor): + source = tensor.to("cpu").contiguous() + result = torch.empty(source.shape, dtype=torch.bfloat16) + ctypes.memmove(result.data_ptr(), source.data_ptr(), source.numel() * 2) + return result + + +class PackedSwiGLUTest(unittest.TestCase): + @classmethod + def setUpClass(cls): + if infinicore.get_device_count("cuda") == 0: + raise unittest.SkipTest("NVIDIA device is required") + + def setUp(self): + infinicore.set_device("cuda:0") + gate = torch.tensor( + [[[-3.0, -0.5, 0.0, 2.0], [0.25, 1.0, 3.0, -2.0]]], + device="cuda:0", + dtype=torch.bfloat16, + ) + up = torch.tensor( + [[[0.5, -2.0, 4.0, 1.5], [-3.0, 0.75, -0.5, 2.0]]], + device="cuda:0", + dtype=torch.bfloat16, + ) + packed = torch.cat((gate, up), dim=-1) + self.input = infinicore.from_torch(packed) + self.expected = ( + (torch_functional.silu(gate.float()) * up.float()).to(torch.bfloat16).cpu() + ) + + def assert_matches_reference(self, output): + infinicore.sync_stream() + actual = to_torch_bfloat16(output) + torch.testing.assert_close( + actual.float(), self.expected.float(), rtol=2e-2, atol=2e-2 + ) + + def test_eager_packed_gate_up_matches_torch(self): + output = ops.silu_and_mul(self.input) + + self.assertEqual(output.shape, [1, 2, 4]) + self.assert_matches_reference(output) + + def test_direct_packed_path_matches_legacy_repack(self): + gate = self.input.narrow(2, 0, 4) + up = self.input.narrow(2, 4, 4) + legacy_packed = infinicore.empty( + self.input.shape, dtype=infinicore.bfloat16, device="cuda:0" + ) + legacy_packed.narrow(2, 0, 4).copy_(gate) + legacy_packed.narrow(2, 4, 4).copy_(up) + + direct = ops.silu_and_mul(self.input) + legacy = ops.silu_and_mul(legacy_packed) + infinicore.sync_stream() + + self.assertTrue( + torch.equal(to_torch_bfloat16(direct), to_torch_bfloat16(legacy)) + ) + self.assert_matches_reference(direct) + + def test_graph_packed_gate_up_matches_torch(self): + zero = infinicore.zeros( + self.input.shape, dtype=infinicore.bfloat16, device="cuda:0" + ) + infinicore.sync_stream() + infinicore.start_graph_recording() + try: + packed_intermediate = ops.add(self.input, zero) + output = ops.silu_and_mul(packed_intermediate) + graph = infinicore.stop_graph_recording() + except BaseException: + if infinicore.is_graph_recording(): + infinicore.cancel_graph_recording() + raise + + del packed_intermediate + churn = [ + infinicore.empty( + self.input.shape, dtype=infinicore.bfloat16, device="cuda:0" + ) + for _ in range(4) + ] + del churn + + for _ in range(2): + graph.run() + self.assert_matches_reference(output) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/static/test_core_chunking_sync.py b/test/static/test_core_chunking_sync.py index 02aa53e04..ce0eee062 100644 --- a/test/static/test_core_chunking_sync.py +++ b/test/static/test_core_chunking_sync.py @@ -44,6 +44,28 @@ def test_swiglu_chunks_only_at_row_boundaries(self) -> None: with self.subTest(tensor=tensor): self.assertIn(f"{tensor}->narrow({{{{0, start, rows}}}})", body) + def test_silu_and_mul_chunks_only_at_row_boundaries(self) -> None: + source = read_source("csrc/infinicore/src/ops/silu_and_mul/silu_and_mul.cc") + body = function_body(source, "void silu_and_mul_(") + + self.assertIn("MAX_ELEMENTS_PER_LAUNCH = Size{1} << 30", body) + self.assertIn("!out->is_contiguous()", body) + self.assertIn("!x->is_contiguous()", body) + self.assertIn( + "output_row_width = out->size(out->ndim() - 1)", + body, + ) + self.assertIn( + "max_rows = MAX_ELEMENTS_PER_LAUNCH / output_row_width", + body, + ) + for tensor in ("output_rows", "input_rows"): + with self.subTest(tensor=tensor): + self.assertIn( + f"{tensor}->narrow({{{{0, start, rows}}}})", + body, + ) + def test_static_attention_keeps_value_head_dimension(self) -> None: source = read_source("csrc/layers/attention/backends/static_attn.cpp") forward = function_body( diff --git a/test/static/test_infinicore_python_contracts.py b/test/static/test_infinicore_python_contracts.py index 377d37d12..36c6f2e7f 100644 --- a/test/static/test_infinicore_python_contracts.py +++ b/test/static/test_infinicore_python_contracts.py @@ -1,5 +1,7 @@ +import ast import unittest from pathlib import Path +from types import SimpleNamespace ROOT = Path(__file__).resolve().parents[2] @@ -9,6 +11,90 @@ def read_source(relative_path: str) -> str: class InfiniCorePythonContractsTest(unittest.TestCase): + def test_paged_decode_reuses_cpu_metadata_storage(self) -> None: + import numpy as np + + source = read_source("python/infinilm/infer_engine.py") + + self.assertIn("class _PagedDecodeMetadataBuffers:", source) + self.assertIn("self._storage_views.append(storage)", source) + self.assertIn("self._input_offsets_values[:] = self._sequence_indices", source) + self.assertIn("out=self._slot_mapping_values", source) + self.assertIn("out=self._cu_seqlens_values", source) + + generate_start = source.index(" def generate(") + generate_end = source.index(" def reset_cache(", generate_start) + generate = source[generate_start:generate_end] + self.assertIn("decode_metadata = None", generate) + self.assertIn("and iter > 0", generate) + self.assertIn("and seq_len == 1", generate) + self.assertIn("and prompt_position_ids is None", generate) + self.assertIn("and self.position_id_axes == 1", generate) + self.assertIn("and mamba_state_indices is None", generate) + self.assertIn(") = decode_metadata.update(past_seq_len)", generate) + self.assertIn("if not reuse_decode_metadata:", generate) + + class FakeTensor: + def __init__(self, shape, dtype): + self.values = np.empty(shape, dtype=dtype) + + def data_ptr(self): + return self.values.ctypes.data + + def numel(self): + return self.values.size + + fake_infinicore = SimpleNamespace( + int32=np.int32, + int64=np.int64, + empty=lambda shape, *, dtype: FakeTensor(shape, dtype), + ) + helper_node = next( + node + for node in ast.parse(source).body + if isinstance(node, ast.ClassDef) + and node.name == "_PagedDecodeMetadataBuffers" + ) + namespace = {"infinicore": fake_infinicore} + exec( + compile( + ast.Module(body=[helper_node], type_ignores=[]), + "infer_engine.py", + "exec", + ), + namespace, + ) + + buffers = namespace["_PagedDecodeMetadataBuffers"](3, 2048) + first = buffers.update(1024) + first_pointers = [tensor.data_ptr() for tensor in first] + self.assertEqual( + [tensor.values.shape for tensor in first], + [(3,), (3,), (3,), (3,), (4,), (4,)], + ) + self.assertEqual( + [tensor.values.dtype for tensor in first], + [np.int64, np.int64, np.int32, np.int32, np.int32, np.int32], + ) + self.assertEqual(first[0].values.tolist(), [1024, 1024, 1024]) + self.assertEqual(first[1].values.tolist(), [1024, 3072, 5120]) + self.assertEqual(first[2].values.tolist(), [1024, 1024, 1024]) + self.assertEqual(first[3].values.tolist(), [1025, 1025, 1025]) + self.assertEqual(first[4].values.tolist(), [0, 1025, 2050, 3075]) + self.assertEqual(first[5].values.tolist(), [0, 1, 2, 3]) + + second = buffers.update(1025) + self.assertEqual( + [tensor.data_ptr() for tensor in second], + first_pointers, + ) + self.assertEqual(second[0].values.tolist(), [1025, 1025, 1025]) + self.assertEqual(second[1].values.tolist(), [1025, 3073, 5121]) + self.assertEqual(second[2].values.tolist(), [1025, 1025, 1025]) + self.assertEqual(second[3].values.tolist(), [1026, 1026, 1026]) + self.assertEqual(second[4].values.tolist(), [0, 1026, 2052, 3078]) + self.assertEqual(second[5].values.tolist(), [0, 1, 2, 3]) + def test_build_installs_one_shared_runtime_for_both_extensions(self) -> None: xmake = read_source("xmake.lua") setup = read_source("setup.py") diff --git a/test/static/test_infinicore_runtime_contracts.py b/test/static/test_infinicore_runtime_contracts.py index e785fd01a..209256925 100644 --- a/test/static/test_infinicore_runtime_contracts.py +++ b/test/static/test_infinicore_runtime_contracts.py @@ -24,6 +24,72 @@ def function_body(source: str, signature: str) -> str: class InfiniCoreRuntimeContractsTest(unittest.TestCase): + def test_eager_graph_tensors_alias_while_compiled_outputs_force_blobs( + self, + ) -> None: + header = read_source("csrc/infinicore/include/infinicore/graph/graph.hpp") + graph = read_source("csrc/infinicore/src/graph/graph.cc") + + self.assertIn("enum class SnapshotPolicy", header) + self.assertIn("kRecordingAware", header) + self.assertIn("GraphTensor(const Tensor &);", header) + self.assertIn("GraphTensor(const Tensor &, SnapshotPolicy policy);", header) + constructors_start = graph.index("GraphTensor::GraphTensor(") + constructors_end = graph.index( + "/* =========================", constructors_start + ) + constructors = graph[constructors_start:constructors_end] + self.assertEqual(constructors.count("GraphTensor::GraphTensor("), 2) + self.assertIn("policy == SnapshotPolicy::kBlob", constructors) + self.assertIn("context::isGraphRecording()", constructors) + self.assertIn("? tensor->to_blob_()", constructors) + self.assertIn(": tensor", constructors) + + expected_snapshots = { + "csrc/engine/compiler/static_batching_compiler.cpp": 2, + "csrc/engine/compiler/paged_compiler.cpp": 2, + } + for compiler, count in expected_snapshots.items(): + with self.subTest(compiler=compiler): + source = read_source(compiler) + self.assertEqual( + source.count("GraphTensor::SnapshotPolicy::kBlob"), count + ) + + def test_standard_mlp_consumes_packed_gate_up_without_repacking(self) -> None: + consumers = ( + ( + "csrc/layers/mlp/mlp.cpp", + "infinicore::Tensor MLP::forward(", + ), + ( + "csrc/models/qwen3_next/qwen3_next_sparse_moe_block.cpp", + "infinicore::Tensor Qwen3NextSharedExpert::forward(", + ), + ) + for relative_path, signature in consumers: + with self.subTest(path=relative_path): + body = function_body(read_source(relative_path), signature) + self.assertIn( + "auto gate_up = gate_up_proj_->forward(hidden_states_mutable);", + body, + ) + self.assertIn("infinicore::op::silu_and_mul(gate_up)", body) + self.assertNotIn("forward_split", body) + self.assertNotIn("infinicore::op::swiglu", body) + + fused_linear = read_source("csrc/layers/linear/fused_linear.cpp") + split = function_body( + fused_linear, + "GateUpParallelLinear::forward_split(infinicore::Tensor &input)", + ) + self.assertIn("output->narrow({{2, 0, cols / 2}})", split) + self.assertIn("output->narrow({{2, cols / 2, cols / 2}})", split) + self.assertIn( + "{gate_name, 0, half_size},\n {up_name, half_size, half_size},", + fused_linear, + ) + def test_retained_op_headers_do_not_depend_on_legacy_public_abi(self) -> None: headers = ( "bitwise_right_shift.hpp", @@ -138,6 +204,38 @@ def test_infiniops_adapters_select_an_active_implementation(self) -> None: ) self.assertNotIn(".set_implementation_index(", source) + def test_nvidia_default_infiniops_config_caches_only_the_index(self) -> None: + bridge = read_source("csrc/infinicore/src/ops/infiniops_impl.hpp") + default_config = function_body( + bridge, + "infini::ops::Config defaultConfigForDevice(", + ) + + self.assertIn( + "device_type == infini::ops::Device::Type::kNvidia", + default_config, + ) + self.assertIn( + "static const std::size_t implementation_index", + default_config, + ) + self.assertIn( + "Operator::active_implementation_indices(\n" + " infini::ops::Device::Type::kNvidia)", + default_config, + ) + self.assertNotIn("static const infini::ops::Config", default_config) + self.assertIn("infini::ops::Config config;", default_config) + self.assertIn( + "config.set_implementation_index(implementation_index)", + default_config, + ) + self.assertIn( + "return configForImplementation(device_type, " + "implementation_indices.front())", + default_config, + ) + def test_causal_softmax_uses_composed_infiniops_operators(self) -> None: source = read_source( "csrc/infinicore/src/ops/causal_softmax/causal_softmax_infiniops.cc" @@ -166,10 +264,7 @@ def test_infiniops_adapter_temporaries_keep_owning_tensors(self) -> None: "causal_softmax/causal_softmax_infiniops.cc": ( "std::optional mask_owner;", ), - "paged_caching/paged_caching_infiniops.cc": ( - "Tensor scale_owner;", - "context::memcpyH2D(scale->data(), &one, sizeof(one), false)", - ), + "paged_caching/paged_caching_infiniops.cc": ("Tensor scale_owner;",), "swiglu/swiglu_infiniops.cc": ("Tensor packed_owner;",), "topksoftmax/topksoftmax_infiniops.cc": ( "Tensor token_expert_indices_owner;", @@ -181,6 +276,34 @@ def test_infiniops_adapter_temporaries_keep_owning_tensors(self) -> None: for token in tokens: self.assertIn(token, source) + def test_paged_caching_auto_scale_avoids_unused_value_copy(self) -> None: + source = read_source( + "csrc/infinicore/src/ops/paged_caching/paged_caching_infiniops.cc" + ) + plan = function_body( + source, + "void *plan(Tensor k_cache, Tensor v_cache, const Tensor &k,", + ) + run = function_body(source, "void run(void *planned_meta)") + + self.assertIn( + "Tensor::empty({1}, DataType::kFloat32, k_cache->device())", + plan, + ) + self.assertNotIn("memcpyH2D", plan) + self.assertNotIn("constexpr float one", plan) + self.assertEqual(plan.count("TensorMeta(scale)"), 1) + self.assertEqual(plan.count("graph::GraphTensor(scale)"), 1) + self.assertRegex( + plan, + re.compile(r"graph::GraphTensor\(v_cache\),\s*scale\s*\};"), + ) + self.assertEqual( + run.count("planned->scale.tensor(planned->scale_tensor)"), + 2, + ) + self.assertIn('std::string{"auto"}', run) + def test_gemm_call_matches_canonical_infiniops_schema(self) -> None: source = read_source("csrc/infinicore/src/ops/gemm/gemm_infiniops.cc") call = function_body(source, "void run(void *planned_meta)") @@ -740,9 +863,13 @@ def test_graph_compilers_cancel_capture_on_exception(self) -> None: self.assertNotIn("context::startGraphRecording()", source) self.assertNotIn("context::stopGraphRecording()", source) self.assertLess( - paged_source.index("for (size_t b : decode_batch_sizes_)"), + paged_source.index("auto capture_decode ="), paged_source.index("GraphRecordingGuard recording;"), ) + self.assertLess( + paged_source.index("for (size_t b : decode_batch_sizes_)"), + paged_source.index("capture_decode(b, nblocks, block_tables_holder_)"), + ) graph_manager = function_body( read_source("csrc/infinicore/src/graph/graph.cc"), @@ -797,6 +924,384 @@ def test_paged_graph_replay_updates_sequence_lengths_on_device(self) -> None: self.assertNotIn("Device::Type::CPU", compiler) self.assertNotIn("DataType::I32", compiler) + def test_p12_p13_short_decode_graph_is_bounded_and_selected_before_copy( + self, + ) -> None: + header = read_source("csrc/engine/compiler/paged_compiler.hpp") + compiler = read_source("csrc/engine/compiler/paged_compiler.cpp") + compile_body = function_body(compiler, "void PagedCompiler::compile()") + replay_body = function_body( + compiler, + "PagedCompiler::Compiled PagedCompiler::get_compiled(", + ) + support = function_body(compiler, "bool supports_reviewed_short_decode_graph(") + property_reader = function_body( + compiler, "std::optional read_paged_graph_properties(" + ) + profile_matcher = function_body( + compiler, "bool matches_reviewed_paged_graph_profile(" + ) + normalized_compiler = " ".join(compiler.split()) + + self.assertLess( + header.index("short_block_tables_holder_"), + header.index("compiled_short_decode_b1_"), + ) + self.assertLess( + header.index("compiled_short_decode_b1_"), + header.index("compiled_map_decode_"), + ) + for token in ( + "kShortDecodeBlockTableWidth = 8", + "kShortDecodeBlockSize = 256", + "kShortDecodeMaxSequenceLength", + ): + self.assertIn(token, compiler) + for token in ( + "read_paged_graph_properties(", + "kShortDecodeProfiles.begin()", + "kShortDecodeProfiles.end()", + "matches_reviewed_paged_graph_profile(", + ): + self.assertIn(token, support) + self.assertNotIn("get_or<", support) + self.assertEqual(support.count("read_paged_graph_properties("), 1) + + for token in ( + 'model_config->get_or("model_type", "")', + 'model_config->get_or("hidden_size", 0)', + 'model_config->get_or("num_hidden_layers", 0)', + 'model_config->get_or("num_attention_heads", 0)', + 'model_config->get_or("num_key_value_heads", 0)', + 'model_config->get_or("position_id_axes", 1)', + "get_tensor_model_parallel_world_size()", + "paged_config.num_blocks()", + "model_config->get_dtype()", + "model_config->get_quant_scheme()", + "model_config->get_kv_quant_scheme()", + ): + self.assertIn(token, property_reader) + + for token in ( + "properties.device_type == infinicore::Device::Type::kNvidia", + "properties.attention_backend == backends::AttentionBackend::FLASH_ATTN", + "properties.num_blocks >= profile.minimum_num_blocks", + "profile.exact_num_blocks", + "profile.tensor_parallel_world_size", + "profile.num_hidden_layers", + "profile.position_id_axes", + "profile.dtype", + "profile.require_unquantized", + "quantization::QuantScheme::NONE", + "quantization::KVQuantAlgo::NONE", + ): + self.assertIn(token, profile_matcher) + self.assertIn( + '{"internlm3", 4096, 32, 2, 128, kShortDecodeBlockSize, ' + "kShortDecodeBlockTableWidth, std::nullopt, std::nullopt, " + "std::nullopt, std::nullopt, std::nullopt, std::nullopt, false}", + normalized_compiler, + ) + self.assertIn( + '{"chatglm", 4096, 32, 2, 128, kShortDecodeBlockSize, ' + "kShortDecodeBlockTableWidth, 512, std::nullopt, 1, 28, 1, " + "infinicore::DataType::kFloat16, true}", + normalized_compiler, + ) + + self.assertLess( + compile_body.index("compiled_short_decode_b1_.reset()"), + compile_body.index("compiled_map_decode_.clear()"), + ) + self.assertLess( + compile_body.index("for (size_t b : decode_batch_sizes_)"), + compile_body.index("compiled_short_decode_b1_.emplace("), + ) + self.assertIn("capture_decode(b, nblocks, block_tables_holder_)", compile_body) + self.assertIn("short_block_tables_holder_", compile_body) + + selection = replay_body.index( + "selected_result = &compiled_short_decode_b1_.value()" + ) + validation = replay_body.index( + "// Validate every input before mutating storage shared by a graph." + ) + first_copy = replay_body.index("->copy_from(") + self.assertLess(selection, validation) + self.assertLess(validation, first_copy) + for token in ( + "batch_size == 1", + "Device::Type::kCpu", + "DataType::kInt32", + "total_sequence_length > 0", + "<= kShortDecodeMaxSequenceLength", + "required_pages <= kShortDecodeBlockTableWidth", + "block_per_req >= required_pages", + "required_pages * sizeof(int32_t)", + ): + self.assertIn(token, replay_body) + normalized_replay = " ".join(replay_body.split()) + self.assertIn( + "required_pages = 1 + " + "(static_cast(total_sequence_length) - 1) / " + "paged_config->block_size()", + normalized_replay, + ) + self.assertIn( + "graph_block_tables->narrow({{1, 0, block_per_req}})", + replay_body, + ) + self.assertIn( + "auto graph = std::get<0>(selected_result->compiled)", replay_body + ) + self.assertIn("std::get<1>(selected_result->compiled)", replay_body) + + def test_baichuan_fixed_prefill_graph_is_opt_in_and_exactly_bounded( + self, + ) -> None: + header = read_source("csrc/engine/compiler/paged_compiler.hpp") + compiler = read_source("csrc/engine/compiler/paged_compiler.cpp") + compile_body = function_body(compiler, "void PagedCompiler::compile()") + replay_body = function_body( + compiler, + "PagedCompiler::Compiled PagedCompiler::get_compiled(", + ) + support = function_body(compiler, "bool supports_baichuan_fixed_prefill_graph(") + profile_matcher = function_body( + compiler, "bool matches_reviewed_paged_graph_profile(" + ) + exact_input = function_body( + compiler, "bool is_exact_baichuan_fixed_prefill_input(" + ) + normalized_compiler = " ".join(compiler.split()) + + self.assertIn("compiled_baichuan_prefill_b1_s10_", header) + for token in ( + "!env_flag_enabled(kBaichuanFixedPrefillGraphEnv)", + "has_mamba_state", + "read_paged_graph_properties(", + "kBaichuanFixedPrefillProfile", + "matches_reviewed_paged_graph_profile(", + ): + self.assertIn(token, support) + self.assertNotIn("get_or<", support) + self.assertEqual(support.count("read_paged_graph_properties("), 1) + self.assertIn( + 'kBaichuanFixedPrefillProfile{ "baichuan", 4096, 32, 32, 128, ' + "kBaichuanFixedPrefillBlockSize, 1, 1, 1, 2, 32, 1, " + "std::nullopt, true}", + normalized_compiler, + ) + for token in ( + "properties.device_type == infinicore::Device::Type::kNvidia", + "properties.attention_backend == backends::AttentionBackend::FLASH_ATTN", + "properties.max_batch_size", + "profile.max_batch_size", + "properties.quant_scheme == quantization::QuantScheme::NONE", + "properties.kv_quant_scheme == quantization::KVQuantAlgo::NONE", + ): + self.assertIn(token, profile_matcher) + self.assertIn('"INFINILM_ENABLE_BAICHUAN_PREFILL_GRAPH"', compiler) + self.assertIn('std::string_view(value) == "1"', compiler) + + for field in ( + "input_ids", + "position_ids", + "past_sequence_lengths", + "total_sequence_lengths", + "input_offsets", + "cu_seqlens", + "block_tables", + "slot_mapping", + ): + self.assertIn(f"input.{field}", exact_input) + for unsupported in ( + "mamba_init_state_indices", + "mamba_final_state_indices", + "pixel_values", + "image_bound", + "tgt_sizes", + "image_grid_thw", + "image_req_ids", + "visual_token_ranges", + "target_hidden_states", + "sample_all_positions", + ): + self.assertIn(f"input.{unsupported}", exact_input) + self.assertNotIn( + "tensor_values_equal(\n input.input_ids", + exact_input, + ) + for values in ( + "{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}", + "input.past_sequence_lengths, {0}", + "input.total_sequence_lengths, {10}", + "input.input_offsets, {0, 10}", + "input.cu_seqlens, {0, 10}", + "input.block_tables, {0}", + ): + self.assertIn(values, exact_input) + + self.assertLess( + compile_body.index("compiled_baichuan_prefill_b1_s10_.reset()"), + compile_body.index("compiled_map_decode_.clear()"), + ) + self.assertIn("make_baichuan_fixed_prefill_input", compile_body) + normalized_compile = " ".join(compile_body.split()) + self.assertIn( + "input.slot_mapping, kBaichuanFixedPrefillSequenceLength, " + "kBaichuanFixedPrefillSequenceLength", + normalized_compile, + ) + self.assertIn("compiled_baichuan_prefill_b1_s10_.emplace(", compile_body) + self.assertIn( + '"fixed Baichuan prefill graph compile: rank={}, batch=1, seq=10"', + compile_body, + ) + + exact_selection = replay_body.index( + "is_exact_baichuan_fixed_prefill_input(input)" + ) + validation = replay_body.index( + "// Validate every input before mutating storage shared by a graph." + ) + first_copy = replay_body.index("->copy_from(") + self.assertLess(exact_selection, validation) + self.assertLess(validation, first_copy) + self.assertIn( + '"fixed Baichuan prefill graph hit: rank={}, batch=1, seq=10"', + replay_body, + ) + self.assertIn( + "graph_input.past_sequence_lengths.value()->copy_from(", + replay_body, + ) + self.assertIn( + "if (!use_baichuan_fixed_prefill_graph) {\n" + " model_->reset_runtime_state();", + replay_body, + ) + + def test_nvidia_varlen_attention_replays_as_host_graph_segments(self) -> None: + header = read_source("csrc/infinicore/include/infinicore/ops/mha_varlen.hpp") + source = read_source( + "csrc/infinicore/src/ops/multi_head_attention_varlen/mha_varlen.cc" + ) + capture_safety = function_body( + header, "bool is_device_graph_capture_safe() const override" + ) + + self.assertIn( + "class MultiheadAttentionVarlen : public graph::DispatchableGraphOperator", + header, + ) + self.assertNotIn( + "INFINICORE_GRAPH_OP_CLASS(\n MultiheadAttentionVarlen", + header, + ) + self.assertIn("return device_graph_capture_safe_;", capture_safety) + self.assertIn("bool device_graph_capture_safe_;", header) + self.assertIn("out->device().type() != Device::Type::kNvidia", source) + + def test_paged_graph_restores_hidden_states_for_speculative_decode(self) -> None: + compiler = read_source("csrc/engine/compiler/paged_compiler.cpp") + compile_body = function_body(compiler, "void PagedCompiler::compile()") + replay_body = function_body( + compiler, + "PagedCompiler::Compiled PagedCompiler::get_compiled(", + ) + worker = function_body( + read_source("csrc/engine/rank_worker.cpp"), + "void RankWorker::thread_loop()", + ) + + self.assertIn("if (output.hidden_states)", compile_body) + self.assertIn("output.hidden_states", compile_body) + self.assertIn("if (compiled_output->hidden_states)", replay_body) + self.assertIn( + "compiled_output->hidden_states->resume_from_blob_()", replay_body + ) + self.assertIn("hidden_states = output->hidden_states;", worker) + + def test_static_graph_carries_eagle_hidden_input_and_output(self) -> None: + compiler = read_source("csrc/engine/compiler/static_batching_compiler.cpp") + compile_body = function_body(compiler, "void StaticBatchingCompiler::compile()") + replay_body = function_body( + compiler, + "StaticBatchingCompiler::Compiled StaticBatchingCompiler::get_compiled(", + ) + + self.assertIn('"model_type", "") == "minicpm_eagle"', compile_body) + self.assertIn( + "input.target_hidden_states = infinicore::Tensor::empty", compile_body + ) + self.assertIn('{b, 1, model_config->get("hidden_size")}', compile_body) + self.assertIn("model_config->get_dtype()", compile_body) + self.assertIn("set_zeros(input.target_hidden_states.value())", compile_body) + self.assertIn("if (output.hidden_states)", compile_body) + self.assertIn("output.hidden_states", compile_body) + hidden_input_start = compile_body.index("if (uses_target_hidden_states)") + hidden_input_end_marker = "set_zeros(input.target_hidden_states.value())" + hidden_input_end = compile_body.index( + hidden_input_end_marker, hidden_input_start + ) + len(hidden_input_end_marker) + hidden_input_block = compile_body[hidden_input_start:hidden_input_end] + for token in ( + "input.target_hidden_states = infinicore::Tensor::empty", + "model_config->get_dtype()", + "set_zeros(input.target_hidden_states.value())", + ): + self.assertIn(token, hidden_input_block) + + normalized_compile = " ".join(compile_body.split()) + self.assertIn( + "GraphTensor::SnapshotPolicy::kBlob), graph_hidden_states}", + normalized_compile, + ) + + self.assertIn("graph_input.target_hidden_states.has_value()", replay_body) + self.assertIn("input.target_hidden_states.has_value()", replay_body) + self.assertIn("graph_has_target_hidden_states !=", replay_body) + self.assertIn("input_has_target_hidden_states", replay_body) + self.assertIn("target_hidden_states.value()->shape()", replay_body) + self.assertIn("target_hidden_states.value()->dtype()", replay_body) + self.assertIn( + "graph_input.target_hidden_states.value()->copy_from(", replay_body + ) + self.assertIn("if (compiled_output->hidden_states)", replay_body) + self.assertIn( + "compiled_output->hidden_states->resume_from_blob_()", replay_body + ) + normalized_replay = " ".join(replay_body.split()) + self.assertIn( + "new InfinilmModel::Output{ " + "compiled_output->logits->resume_from_blob_(), hidden_states}", + normalized_replay, + ) + + def test_graph_lookup_preserves_target_hidden_on_source_device(self) -> None: + header = read_source("csrc/engine/rank_worker.hpp") + engine_source = read_source("csrc/engine/infer_engine.cpp") + engine = function_body( + engine_source, + "InferEngine::Input::to_model_input(", + ) + worker = function_body( + read_source("csrc/engine/rank_worker.cpp"), + "void RankWorker::thread_loop()", + ) + + self.assertIn("bool preserve_target_hidden_device = false", header) + self.assertIn("bool preserve_target_hidden_device)", engine_source) + self.assertIn("? target_hidden_states", engine) + self.assertIn(": to_device(target_hidden_states)", engine) + normalized_worker = " ".join(worker.split()) + self.assertIn( + "compiler_->get_compiled( local_args.to_model_input( " + "infinicore::Device{infinicore::Device::Type::kCpu}, false, true))", + normalized_worker, + ) + def test_paged_decode_graph_is_enabled_under_tensor_parallelism(self) -> None: source = read_source("csrc/engine/compiler/paged_compiler.cpp") compile_body = function_body(source, "void PagedCompiler::compile()") @@ -808,18 +1313,128 @@ def test_paged_decode_graph_is_enabled_under_tensor_parallelism(self) -> None: self.assertNotIn("get_tensor_model_parallel_world_size", replay_body) self.assertIn("GraphRecordingGuard recording", compile_body) - def test_flash_attention_decode_uses_host_graph_segments(self) -> None: + def test_reviewed_flash_attention_shapes_own_graph_safe_provider(self) -> None: mha_header = read_source( "csrc/infinicore/include/infinicore/ops/mha_kvcache.hpp" ) + mha_source = read_source("csrc/infinicore/src/ops/mha_kvcache/mha_kvcache.cc") + adapter = read_source( + "csrc/infinicore/src/ops/mha_kvcache/mha_kvcache_infiniops.cc" + ) paged_attention_source = read_source( "csrc/infinicore/src/ops/paged_attention/paged_attention.cc" ) capture_safety = function_body( mha_header, "bool is_device_graph_capture_safe() const override" ) - self.assertIn("return false;", capture_safety) - self.assertNotIn("device_graph_capture_safe_", mha_header) + predicate = function_body( + mha_source, "bool MhaKVCache::supports_device_graph_capture(" + ) + + p12_shape_start = predicate.index("const bool p12_shape =") + p12_shape_end = predicate.index("const bool p13_shape =", p12_shape_start) + p12_shape = predicate[p12_shape_start:p12_shape_end] + for token in ( + "q->shape() == Shape({1, 1, 32, 128})", + "out->shape() == Shape({1, 1, 32, 128})", + "k_cache->shape() == Shape({512, 256, 2, 128})", + "seqlens_k->shape() == Shape({1})", + "block_table->shape() == Shape({1, 8})", + ): + self.assertIn(token, p12_shape) + + p12_layout_start = predicate.index("const bool p12_layout =") + p12_layout_end = predicate.index( + "const bool reviewed_shape_and_layout =", p12_layout_start + ) + p12_layout = predicate[p12_layout_start:p12_layout_end] + for token in ( + "q->strides() == Strides({4608, 4608, 128, 1})", + "out->strides() == Strides({4096, 4096, 128, 1})", + ): + self.assertIn(token, p12_layout) + self.assertEqual(predicate.count("(p12_shape && p12_layout)"), 1) + + p14_shape_start = predicate.index("const bool p14_shape =") + p14_shape_end = predicate.index("const bool p13_layout =", p14_shape_start) + p14_shape = predicate[p14_shape_start:p14_shape_end] + for token in ( + "q->shape() == Shape({16, 1, 32, 128})", + "k_cache->size(0) == 512", + "k_cache->size(1) == 256", + "k_cache->size(2) == 2", + "k_cache->size(3) == 128", + "seqlens_k->shape() == Shape({16})", + "block_table->shape() == Shape({16, 512})", + ): + self.assertIn(token, p14_shape) + + p14_layout_start = predicate.index("const bool p14_layout =") + p14_layout_end = predicate.index( + "const bool reviewed_shape_and_layout =", p14_layout_start + ) + p14_layout = predicate[p14_layout_start:p14_layout_end] + for token in ( + "q->strides() == Strides({4608, 4608, 128, 1})", + "out->strides() == Strides({4096, 4096, 128, 1})", + ): + self.assertIn(token, p14_layout) + self.assertEqual(predicate.count("(p14_shape && p14_layout)"), 1) + + self.assertIn("return device_graph_capture_safe_;", capture_safety) + self.assertIn("bool device_graph_capture_safe_;", mha_header) + self.assertIn("context::isGraphRecording()", mha_source) + for token in ( + "Device::Type::kNvidia", + "Shape({1, 1, 32, 128})", + "Shape({16, 1, 24, 128})", + "Shape({1, 1, 16, 128})", + "Shape({16, 1, 32, 128})", + "k_cache->size(0) == 128", + "k_cache->size(0) == 1", + "k_cache->size(0) == 512", + "k_cache->size(1) == 256", + "k_cache->size(2) == 2", + "k_cache->size(2) == 8", + "k_cache->size(2) == 16", + "block_table->shape() == Shape({1, 8})", + "seqlens_k->shape() == Shape({16})", + "block_table->shape() == Shape({16, 128})", + "block_table->shape() == Shape({1, 1})", + "block_table->shape() == Shape({16, 512})", + "q->strides() == Strides({5120, 5120, 128, 1})", + "out->strides() == Strides({3072, 3072, 128, 1})", + "q->strides() == Strides({6144, 6144, 128, 1})", + "out->strides() == Strides({2048, 2048, 128, 1})", + "q->strides() == Strides({4608, 4608, 128, 1})", + "out->strides() == Strides({4096, 4096, 128, 1})", + "(p13_shape && p13_layout)", + "(p12_shape && p12_layout)", + "(p09_shape && p09_layout)", + "(p11_shape && p11_layout)", + "(p14_shape && p14_layout)", + "reviewed_shape_and_layout", + "alibi_slopes.has_value()", + "q->is_contiguous()", + "block_table->is_contiguous()", + ): + self.assertIn(token, predicate) + self.assertNotIn("Strides({3072, 0, 128, 1})", predicate) + self.assertNotIn("INFINICORE_GRAPH_CAPTURE_DEBUG", mha_source) + + self.assertIn("std::unique_ptr graph_safe_provider", adapter) + self.assertIn("make_graph_safe_provider(*planned)", adapter) + self.assertIn("FlashAttnOperator::Make(", adapter) + self.assertIn("Device::Type::kNvidia, 17", adapter) + self.assertIn("(*planned->graph_safe_provider)(", adapter) + self.assertIn("implementation_index_for_device(device_type)", adapter) + for implementation_index in (0, 8, 16): + with self.subTest(implementation_index=implementation_index): + self.assertIn( + f"return {implementation_index};", + adapter, + ) + self.assertIn("infini::ops::FlashAttnWithKvcache::Call(", adapter) paged_attention = function_body( paged_attention_source, "void PagedAttention::execute(" ) @@ -863,6 +1478,52 @@ def test_static_graph_keeps_dynamic_cache_metadata_on_device(self) -> None: self.assertNotIn("infinicore::op::kv_caching_", graph_forward) self.assertNotIn("Device::Type::kCpu", graph_forward) + def test_static_eager_snapshots_sequence_lengths_once_per_forward(self) -> None: + metadata = read_source("csrc/global_state/forward_context.hpp") + engine = read_source("csrc/engine/infer_engine.cpp") + worker_header = read_source("csrc/engine/rank_worker.hpp") + worker = read_source("csrc/engine/rank_worker.cpp") + attention = read_source("csrc/layers/attention/backends/static_attn.cpp") + compiler = read_source("csrc/engine/compiler/static_batching_compiler.cpp") + + self.assertIn("std::optional first_past_sequence_length", metadata) + self.assertIn("std::optional first_total_sequence_length", metadata) + self.assertIn("bool snapshot_static_sequence_lengths = false", worker_header) + + conversion = function_body(engine, "InferEngine::Input::to_model_input(") + for field in ("past_sequence_lengths", "total_sequence_lengths"): + snapshot = f"first_sequence_length(\n {field}" + transfer = f"to_device({field})" + with self.subTest(field=field): + self.assertIn(snapshot, conversion) + self.assertIn(transfer, conversion) + self.assertLess(conversion.index(snapshot), conversion.index(transfer)) + self.assertIn(f"attn_metadata.first_{field[:-1]}", conversion) + + run = function_body(worker, "void RankWorker::thread_loop()") + self.assertIn( + "attention_backend_ == backends::AttentionBackend::STATIC_ATTN", run + ) + + eager = function_body( + attention, "infinicore::Tensor StaticAttentionImpl::forward(" + ) + update = function_body(attention, "StaticAttentionImpl::do_kv_cache_update(") + fallback = function_body(attention, "size_t sequence_length_from_metadata(") + for body in (eager, update): + self.assertNotIn("Device::Type::kCpu", body) + self.assertNotIn("->data()", body) + self.assertIn("attn_metadata.first_past_sequence_length", eager) + self.assertIn("attn_metadata.first_total_sequence_length", eager) + self.assertIn("size_t cache_pos", attention) + self.assertIn("if (snapshot.has_value())", fallback) + self.assertIn("Device::Type::kCpu", fallback) + self.assertNotIn("value_or", fallback) + + compile_body = function_body(compiler, "void StaticBatchingCompiler::compile()") + self.assertIn("first_past_sequence_length = 0", compile_body) + self.assertIn("first_total_sequence_length = 1", compile_body) + def test_static_graph_uses_dynamic_canonical_cache_slots(self) -> None: compiler = read_source("csrc/engine/compiler/static_batching_compiler.cpp") compile_body = function_body(compiler, "void StaticBatchingCompiler::compile()") @@ -898,6 +1559,7 @@ def test_random_sampling_uses_only_canonical_infiniops(self) -> None: adapter = read_source( "csrc/infinicore/src/ops/random_sample/random_sample_infiniops.cc" ) + worker = read_source("csrc/engine/rank_worker.cpp") self.assertIn("infini::ops::Argmax::Call", wrapper) for sentinel in ( @@ -939,6 +1601,34 @@ def test_random_sampling_uses_only_canonical_infiniops(self) -> None: self.assertNotIn("RandomSampleInfinilm", adapter) self.assertNotIn("random_sample_infinilm.h", adapter) + self.assertIn("const bool batched = logits->ndim() == 2", wrapper) + self.assertIn("device_type == Device::Type::kNvidia && batched", wrapper) + self.assertIn("indices->numel() != num_rows", wrapper) + self.assertIn("indices->ndim() != 1 || indices->size(0) != num_rows", wrapper) + self.assertIn("std::optional{1}", wrapper) + + run = function_body(worker, "void RankWorker::thread_loop()") + self.assertIn( + "const bool parameter_greedy = top_p == 0.0f || top_k == 1 " + "|| temperature == 0.0f", + run, + ) + self.assertIn( + "rank_info_.device.type() == infinicore::Device::Type::kNvidia", + run, + ) + self.assertIn("sample_all_positions || logits_are_last_token_only", run) + self.assertIn("logits_positions == n_out", run) + self.assertIn("logits->is_contiguous()", run) + for dtype in ("kFloat16", "kBFloat16", "kFloat32"): + self.assertIn(f"logits_dtype == infinicore::DataType::{dtype}", run) + self.assertIn("logits->view({logits_positions, vocab_size})", run) + self.assertIn("for (size_t i{0}; i < n_out; ++i)", run) + self.assertIn( + "score_idx = static_cast(input_offsets[i + 1] - 1)", + run, + ) + def test_canonical_rope_preserves_static_batch_positions(self) -> None: static_calls = ( ( @@ -1111,6 +1801,12 @@ def test_context_owns_one_runtime_per_thread_and_device(self) -> None: self.assertIn("std::shared_ptr(new Runtime(device))", create_runtime) self.assertIn("found->second.lock()", create_runtime) self.assertIn("thread_id", create_runtime) + create_failure = create_runtime.index("catch (...)") + self.assertIn("current_runtime_->activate()", create_runtime[create_failure:]) + self.assertIn( + "std::rethrow_exception(original_error)", + create_runtime[create_failure:], + ) get_current = function_body( context_source, "Runtime *ContextImpl::getCurrentRuntime()" ) @@ -1142,9 +1838,9 @@ def test_context_owns_one_runtime_per_thread_and_device(self) -> None: f"initializeDeviceType()", constructor ) - self.assertIn("mutable std::mutex stream_mutex_;", runtime_header) + self.assertNotIn("stream_mutex_", runtime_header) self.assertIn( - "mutable infini::rt::runtime::Stream stream_ = nullptr;", + "infini::rt::runtime::Stream stream_ = nullptr;", runtime_header, ) self.assertNotIn("std::unordered_map None: runtime_constructor = function_body( runtime_source, "Runtime::Runtime(Device device)" ) - self.assertNotIn("StreamCreate", runtime_constructor) + self.assertIn("StreamCreate(&stream_)", runtime_constructor) + self.assertGreater( + runtime_constructor.index("StreamCreate(&stream_)"), + runtime_constructor.rindex("make_unique"), + ) + self.assertIn( + "stream_status != infini::rt::runtime::kSuccess", + runtime_constructor, + ) + self.assertIn( + "StreamDestroy(stream_)", + runtime_constructor, + ) stream = function_body( runtime_source, "infini::rt::runtime::Stream Runtime::stream() const" ) - for token in ("stream_mutex_", "StreamCreate", "stream_"): - self.assertIn(token, stream) - self.assertIn("if (stream_ == nullptr)", stream) self.assertIn("return stream_", stream) + for token in ("stream_mutex_", "StreamCreate", "SetDevice"): + self.assertNotIn(token, stream) destructor = function_body(runtime_source, "Runtime::~Runtime() noexcept") - self.assertIn("stream_mutex_", destructor) + self.assertNotIn("stream_mutex_", destructor) self.assertIn("stream_", destructor) self.assertIn("StreamDestroy", destructor) diff --git a/test/static/test_speculative_runner_contracts.py b/test/static/test_speculative_runner_contracts.py new file mode 100644 index 000000000..6f6b45b55 --- /dev/null +++ b/test/static/test_speculative_runner_contracts.py @@ -0,0 +1,148 @@ +import ast +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] + + +class SpeculativeRunnerContractsTest(unittest.TestCase): + def test_primary_target_only_requests_all_positions_during_prefill(self) -> None: + source = ( + ROOT / "python/infinilm/llm/model_runner/speculative_runner.py" + ).read_text(encoding="utf-8") + tree = ast.parse(source) + runner = next( + node + for node in tree.body + if isinstance(node, ast.ClassDef) and node.name == "SpeculativeRunner" + ) + method = next( + node + for node in runner.body + if isinstance(node, ast.FunctionDef) and node.name == "forward" + ) + + assignment = next( + node + for node in ast.walk(method) + if isinstance(node, ast.Assign) + and any( + isinstance(target, ast.Subscript) + and isinstance(target.value, ast.Name) + and target.value.id == "target_model_input" + and isinstance(target.slice, ast.Constant) + and target.slice.value == "sample_all_positions" + for target in node.targets + ) + ) + self.assertIsInstance(assignment.value, ast.Attribute) + self.assertEqual(assignment.value.attr, "is_prefill") + self.assertIsInstance(assignment.value.value, ast.Name) + self.assertEqual(assignment.value.value.id, "scheduler_output") + + target_call = next( + node + for node in ast.walk(method) + if isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr == "forward_raw" + and any( + keyword.arg is None + and isinstance(keyword.value, ast.Name) + and keyword.value.id == "target_model_input" + for keyword in node.keywords + ) + ) + self.assertIsInstance(target_call.func.value, ast.Attribute) + self.assertEqual(target_call.func.value.attr, "target_model_engine") + + def test_single_token_verification_uses_last_position_graph_path(self) -> None: + source = ( + ROOT / "python/infinilm/llm/model_runner/speculative_runner.py" + ).read_text(encoding="utf-8") + tree = ast.parse(source) + runner = next( + node + for node in tree.body + if isinstance(node, ast.ClassDef) and node.name == "SpeculativeRunner" + ) + method = next( + node + for node in runner.body + if isinstance(node, ast.FunctionDef) + and node.name == "_build_paged_verify_batch_input" + ) + assignment = next( + node + for node in method.body + if isinstance(node, ast.Assign) + and any( + isinstance(target, ast.Name) and target.id == "sample_all_positions" + for target in node.targets + ) + ) + expression = compile( + ast.Expression(assignment.value), + str(ROOT / "speculative_runner.py"), + "eval", + ) + + def requires_all_positions(lengths: list[int]) -> bool: + namespace = { + "any": any, + "len": len, + "candidates": [{"draft_tokens": [0] * length} for length in lengths], + } + return bool(eval(expression, namespace)) + + self.assertFalse(requires_all_positions([1])) + self.assertFalse(requires_all_positions([1, 1])) + self.assertTrue(requires_all_positions([2])) + self.assertTrue(requires_all_positions([1, 2])) + + returned = next( + node for node in reversed(method.body) if isinstance(node, ast.Return) + ) + self.assertIsInstance(returned.value, ast.Dict) + entries = { + key.value: value + for key, value in zip(returned.value.keys, returned.value.values) + if isinstance(key, ast.Constant) + } + self.assertIn("sample_all_positions", entries) + self.assertIsInstance(entries["sample_all_positions"], ast.Name) + self.assertEqual(entries["sample_all_positions"].id, "sample_all_positions") + + def test_draft_forward_disables_all_position_sampling(self) -> None: + source = ( + ROOT / "python/infinilm/llm/model_runner/speculative_runner.py" + ).read_text(encoding="utf-8") + tree = ast.parse(source) + runner = next( + node + for node in tree.body + if isinstance(node, ast.ClassDef) and node.name == "SpeculativeRunner" + ) + method = next( + node + for node in runner.body + if isinstance(node, ast.FunctionDef) + and node.name == "_draft_eagle_tokens_batch" + ) + draft_call = next( + node + for node in ast.walk(method) + if isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr == "forward_raw" + and isinstance(node.func.value, ast.Attribute) + and node.func.value.attr == "draft_model_engine" + ) + keywords = {keyword.arg: keyword.value for keyword in draft_call.keywords} + sample_all_positions = keywords["sample_all_positions"] + self.assertIsInstance(sample_all_positions, ast.Constant) + self.assertIs(sample_all_positions.value, False) + + +if __name__ == "__main__": + unittest.main()