Repository: qdrant/qdrant
File: lib/gridstore/src/gridstore.rs (Rust)

Code before:
                length,
            } = pointer;

            let raw = self.read_from_pages::<true>(page_id, block_offset, length);

            hw_counter.incr_delta(raw.len());

            let decompressed = self.decompress(raw);
            let value = V::from_bytes(&decompressed);
            if !callback(point_offset, value)? {
                return Ok(());
            }
        }
        Ok(())
    }

    /// Return the storage size in bytes
    pub fn get_storage_size_bytes(&self) -> usize {
        self.bitmask.read().get_storage_size_bytes()
    }
}

impl<V> Gridstore<V> {
    /// The number of blocks needed for a given value bytes size
    #[inline]
    fn blocks_for_value(value_size: usize, block_size: usize) -> u32 {
        value_size.div_ceil(block_size).try_into().unwrap()
    }

    /// Create flusher that durably persists all pending changes when invoked
    pub fn flusher(&self) -> Flusher {
        let pending_updates = self.tracker.read().pending_updates.clone();

        let pages = Arc::downgrade(&self.pages);
        let tracker = Arc::downgrade(&self.tracker);
        let bitmask = Arc::downgrade(&self.bitmask);
        let block_size_bytes = self.config.block_size_bytes;

        let is_alive_flush_lock = self.is_alive_flush_lock.handle();

        Box::new(move || {
            let (Some(is_alive_flush_guard), Some(pages), Some(tracker), Some(bitmask)) = (
                is_alive_flush_lock.lock_if_alive(),
                pages.upgrade(),
                tracker.upgrade(),
                bitmask.upgrade(),
            ) else {
                log::trace!("Gridstore was cleared, cancelling flush");
                return Err(GridstoreError::FlushCancelled);
            };


Review comment (bug):
This looks sound 👍 

I was afraid it could data race where newly incoming data could be written to pages before we read here. But writes require a `&mut self`, while this holds a `&self`.

Code after:
                self.tracker
                    .read()
                    .iter_pointers(from)
                    .filter_map(|(point_offset, opt_pointer)| {
                        opt_pointer.map(|pointer| (point_offset, pointer))
                    })
                    .take(BUFFER_SIZE),
            );

            // If no more pointers, we're done
            if buffer.is_empty() {
                break;
            }

            // Update `from` for the next iteration
            from = buffer.last().unwrap().0 + 1;

            // Process buffer without holding the tracker lock
            for &(point_offset, pointer) in &buffer {
                let ValuePointer {
                    page_id,
                    block_offset,
                    length,
                } = pointer;

                let raw = self.read_from_pages::<true>(page_id, block_offset, length);

                hw_counter.incr_delta(raw.len());

                let decompressed = self.decompress(raw);
                let value = V::from_bytes(&decompressed);
                if !callback(point_offset, value)? {
                    return Ok(());
                }
            }
        }
        Ok(())
    }

    /// Return the storage size in bytes
    pub fn get_storage_size_bytes(&self) -> usize {
        self.bitmask.read().get_storage_size_bytes()
    }
}

impl<V> Gridstore<V> {
    /// The number of blocks needed for a given value bytes size
    #[inline]
    fn blocks_for_value(value_size: usize, block_size: usize) -> u32 {
        value_size.div_ceil(block_size).try_into().unwrap()
    }