Repository: qdrant/qdrant
File: lib/collection/src/collection_manager/collection_updater.rs (Rust)

Code before:
        tokio::task::block_in_place(|| {
            // Allow only one update at a time, ensure no data races between segments.
            // let _lock = self.update_lock.lock().unwrap();

            let scroll_lock = segments.read().scroll_read_lock.clone();
            let _scroll_lock = scroll_lock.blocking_write();

            let operation_result = match operation {
                CollectionUpdateOperations::PointOperation(point_operation) => {
                    process_point_operation(segments, op_num, point_operation, hw_counter)
                }
                CollectionUpdateOperations::VectorOperation(vector_operation) => {
                    process_vector_operation(segments, op_num, vector_operation, hw_counter)
                }
                CollectionUpdateOperations::PayloadOperation(payload_operation) => {
                    process_payload_operation(segments, op_num, payload_operation, hw_counter)
                }
                CollectionUpdateOperations::FieldIndexOperation(index_operation) => {
                    process_field_index_operation(segments, op_num, &index_operation, hw_counter)
                }
            };

            CollectionUpdater::handle_update_result(segments, op_num, &operation_result);

            operation_result
        })
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;
    use std::sync::atomic::AtomicBool;

    use common::counter::hardware_accumulator::HwMeasurementAcc;
    use itertools::Itertools;
    use parking_lot::RwLockUpgradableReadGuard;
    use segment::data_types::vectors::{
        DEFAULT_VECTOR_NAME, VectorStructInternal, only_default_vector,
    };
    use segment::entry::entry_point::SegmentEntry;
    use segment::json_path::JsonPath;
    use segment::payload_json;
    use segment::types::PayloadSchemaType::Keyword;
    use segment::types::{Payload, PayloadContainer, PayloadFieldSchema, WithPayload};
    use serde_json::json;
    use tempfile::Builder;

    use super::*;
    use crate::collection_manager::fixtures::{
        build_segment_1, build_segment_2, build_test_holder,

Review comment (performance):
`CollectionUpdate::update` is blocking, but we call it from async context in a few places. Seems reasonable to wrap the whole `update` into `block_in_place`.

`block_in_place` does have a performance hit when called from async, but
1) we already use `block_in_place` inside `update`, so we already "pay" for this perf hit, we just extend `block_in_place` scope here, which may actually improve perf when called from async
2) `update_worker_fn` (which is the most important `update` path) is a blocking task now, and other places where we still call `update` from async are less critical (e.g., `LocalShard::load_from_wal`)


Code after:
        scroll_lock: Arc<tokio::sync::RwLock<()>>,
        update_tracker: UpdateTracker,
        hw_counter: &HardwareCounterCell,
    ) -> CollectionResult<usize> {
        // Use block_in_place here to avoid blocking the current async executor
        let operation_result = tokio::task::block_in_place(|| {
            // Allow only one update at a time, ensure no data races between segments.
            // let _update_lock = self.update_lock.lock().unwrap();

            let _scroll_lock = scroll_lock.blocking_write();
            let _update_guard = update_tracker.update();

            match operation {
                CollectionUpdateOperations::PointOperation(point_operation) => {
                    process_point_operation(segments, op_num, point_operation, hw_counter)
                }
                CollectionUpdateOperations::VectorOperation(vector_operation) => {
                    process_vector_operation(segments, op_num, vector_operation, hw_counter)
                }
                CollectionUpdateOperations::PayloadOperation(payload_operation) => {
                    process_payload_operation(segments, op_num, payload_operation, hw_counter)
                }
                CollectionUpdateOperations::FieldIndexOperation(index_operation) => {
                    process_field_index_operation(segments, op_num, &index_operation, hw_counter)
                }
            }
        });

        CollectionUpdater::handle_update_result(segments, op_num, &operation_result);

        operation_result
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;
    use std::sync::atomic::AtomicBool;

    use common::counter::hardware_accumulator::HwMeasurementAcc;
    use itertools::Itertools;
    use parking_lot::RwLockUpgradableReadGuard;
    use segment::data_types::vectors::{
        DEFAULT_VECTOR_NAME, VectorStructInternal, only_default_vector,
    };
    use segment::entry::entry_point::SegmentEntry;
    use segment::json_path::JsonPath;
    use segment::payload_json;
    use segment::types::PayloadSchemaType::Keyword;
    use segment::types::{Payload, PayloadContainer, PayloadFieldSchema, WithPayload};
    use serde_json::json;