ronantakizawa/codereview-bench huggingface dataset:
Repository: tiangolo/fastapi
File: tests/test_list_bytes_file_order_preserved_issue_14811.py (Python)

Code before:
"""
Regression test: preserve order when using list[bytes] + File()
See https://github.com/fastapi/fastapi/discussions/14811
Related: PR #3372
"""

from typing import Annotated

import anyio
import pytest
from fastapi import FastAPI, File
from fastapi.testclient import TestClient
from starlette.datastructures import UploadFile as StarletteUploadFile


def test_list_bytes_file_preserves_order_issue_14811(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    app = FastAPI()

    @app.post("/upload")
    async def upload(files: Annotated[list[bytes], File()]):
        # return something that makes order obvious
        return [b[0] for b in files]

    original_read = StarletteUploadFile.read

    async def patched_read(self: StarletteUploadFile, size: int = -1) -> bytes:
        # Make the FIRST file slower *deterministically*
        if self.filename == "slow.txt":
            await anyio.sleep(0.05)
        return await original_read(self, size)

    monkeypatch.setattr(StarletteUploadFile, "read", patched_read)

    client = TestClient(app)

    files = [
        ("files", ("slow.txt", b"A" * 10, "text/plain")),
        ("files", ("fast.txt", b"B" * 10, "text/plain")),
    ]

Review comment (suggestion):
```suggestion
def test_list_bytes_file_preserves_order(
```

Code after:
"""
Regression test: preserve order when using list[bytes] + File()
See https://github.com/fastapi/fastapi/discussions/14811
Related: PR #3372
"""

from typing import Annotated

import anyio
import pytest
from fastapi import FastAPI, File
from fastapi.testclient import TestClient
from starlette.datastructures import UploadFile as StarletteUploadFile


def test_list_bytes_file_preserves_order(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    app = FastAPI()

    @app.post("/upload")
    async def upload(files: Annotated[list[bytes], File()]):
        # return something that makes order obvious
        return [b[0] for b in files]

    original_read = StarletteUploadFile.read

    async def patched_read(self: StarletteUploadFile, size: int = -1) -> bytes:
        # Make the FIRST file slower *deterministically*
        if self.filename == "slow.txt":
            await anyio.sleep(0.05)
        return await original_read(self, size)

    monkeypatch.setattr(StarletteUploadFile, "read", patched_read)

    client = TestClient(app)

    files = [
        ("files", ("slow.txt", b"A" * 10, "text/plain")),
        ("files", ("fast.txt", b"B" * 10, "text/plain")),
    ]
Repository: tiangolo/fastapi
File: tests/test_tutorial/test_websockets/test_tutorial003.py (Python)

Code before:
def get_html(mod: ModuleType):
    return mod.html


@pytest.fixture(name="client")
def get_client(mod: ModuleType):
    client = TestClient(mod.app)

    return client


def test_get(client: TestClient, html: str):
    response = client.get("/")
    assert response.text == html


def test_websocket_handle_disconnection(client: TestClient):
    with (
        client.websocket_connect("/ws/1234") as connection,
        client.websocket_connect("/ws/5678") as connection_two,
    ):
        connection.send_text("Hello from 1234")
        data1 = connection.receive_text()
        assert data1 == "You wrote: Hello from 1234"
        data2 = connection_two.receive_text()
        client1_says = "Client #1234 says: Hello from 1234"
        assert data2 == client1_says
        data1 = connection.receive_text()
        assert data1 == client1_says
        connection_two.close()
        data1 = connection.receive_text()
        assert data1 == "Client #5678 left the chat"


Review comment (bug):
Without this timeout test **sometimes** froze and failed with timeout.
Not sure if we can handle this in a different way..

Code after:
@pytest.fixture(name="html")
def get_html(mod: ModuleType):
    return mod.html


@pytest.fixture(name="client")
def get_client(mod: ModuleType):
    client = TestClient(mod.app)

    return client


def test_get(client: TestClient, html: str):
    response = client.get("/")
    assert response.text == html


def test_websocket_handle_disconnection(client: TestClient):
    with (
        client.websocket_connect("/ws/1234") as connection,
        client.websocket_connect("/ws/5678") as connection_two,
    ):
        connection.send_text("Hello from 1234")
        data1 = connection.receive_text()
        assert data1 == "You wrote: Hello from 1234"
        time.sleep(0.01)  # Give server time to process broadcast
        data2 = connection_two.receive_text()
        client1_says = "Client #1234 says: Hello from 1234"
        assert data2 == client1_says
        data1 = connection.receive_text()
        assert data1 == client1_says
        connection_two.close()
        time.sleep(0.01)  # Give server time to process broadcast
        data1 = connection.receive_text()
        assert data1 == "Client #5678 left the chat"
Repository: tikv/tikv
File: components/cdc/src/txn_source.rs (Rust)

Code before:
// Copyright 2023 TiKV Project Authors. Licensed under Apache-2.0.

// TiCDC uses 1 - 255 to indicate the source of TiDB.
// For now, 1 - 15 are reserved for TiCDC to implement BDR synchronization.
// 16 - 255 are reserved for extendability.
const CDC_WRITE_SOURCE_BITS: u64 = 8;
const CDC_WRITE_SOURCE_MAX: u64 = (1 << CDC_WRITE_SOURCE_BITS) - 1;

// TiCDC uses 1-15 to indicate the change from a lossy DDL reorg Backfill job.
// For now, we only use 1 for column reorg backfill job.
#[cfg(test)]
const LOSSY_DDL_REORG_SOURCE_BITS: u64 = 4;
#[cfg(test)]
const LOSSY_DDL_COLUMN_REORG_SOURCE: u64 = 1;
#[cfg(test)]
const LOSSY_DDL_REORG_SOURCE_MAX: u64 = (1 << LOSSY_DDL_REORG_SOURCE_BITS) - 1;
const LOSSY_DDL_REORG_SOURCE_SHIFT: u64 = CDC_WRITE_SOURCE_BITS;

/// For kv.TxnSource
/// We use an uint64 to represent the source of a transaction.
/// The first 8 bits are reserved for TiCDC, and the next 4 bits are reserved
/// for Lossy DDL reorg Backfill job. The remaining 52 bits are reserved for
/// extendability.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Default)]
pub(crate) struct TxnSource(u64);

impl TxnSource {
    #[cfg(test)]
    pub(crate) fn set_cdc_write_source(&mut self, value: u64) {
        if value > CDC_WRITE_SOURCE_MAX {
            unreachable!("Only use it in tests")
        }
        self.0 |= value;
    }

    #[cfg(test)]
    pub(crate) fn get_cdc_write_source(&self) -> u64 {

Review comment (suggestion):
As you described, 4 bits are for all write sources used by TiCDC (for BDR replication and DDL reorg). And DDL reorg only uses 1 bit exactly.

Code after:
// Copyright 2023 TiKV Project Authors. Licensed under Apache-2.0.

// The bitmap:
// |RESERVED|LOSSY_DDL_REORG_SOURCE_BITS|CDC_WRITE_SOURCE_BITS|
// |  48    |             8             | 4(RESERVED) |  4    |
//
// TiCDC uses 1 - 255 to indicate the source of TiDB.
// For now, 1 - 15 are reserved for TiCDC to implement BDR synchronization.
// 16 - 255 are reserved for extendability.
const CDC_WRITE_SOURCE_BITS: u64 = 8;
const CDC_WRITE_SOURCE_MAX: u64 = (1 << CDC_WRITE_SOURCE_BITS) - 1;

// TiCDC uses 1-255 to indicate the change from a lossy DDL reorg Backfill job.
// For now, we only use 1 for column reorg backfill job.
#[cfg(test)]
const LOSSY_DDL_REORG_SOURCE_BITS: u64 = 8;
#[cfg(test)]
const LOSSY_DDL_COLUMN_REORG_SOURCE: u64 = 1;
#[cfg(test)]
const LOSSY_DDL_REORG_SOURCE_MAX: u64 = (1 << LOSSY_DDL_REORG_SOURCE_BITS) - 1;
const LOSSY_DDL_REORG_SOURCE_SHIFT: u64 = CDC_WRITE_SOURCE_BITS;

/// For kv.TxnSource
/// We use an uint64 to represent the source of a transaction.
/// The first 8 bits are reserved for TiCDC, and the next 8 bits are reserved
/// for Lossy DDL reorg Backfill job. The remaining 48 bits are reserved for
/// extendability.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Default)]
pub(crate) struct TxnSource(u64);

impl TxnSource {
    #[cfg(test)]
    pub(crate) fn set_cdc_write_source(&mut self, value: u64) {
        if value > CDC_WRITE_SOURCE_MAX {
            unreachable!("Only use it in tests")
        }
        self.0 |= value;
Repository: tikv/tikv
File: components/cdc/src/txn_source.rs (Rust)

Code before:
// Copyright 2023 TiKV Project Authors. Licensed under Apache-2.0.

// The bitmap:
// |RESERVED|LOSSY_DDL_REORG_SOURCE_BITS|CDC_WRITE_SOURCE_BITS|
// |  52    |             4             |          8          |
//
// TiCDC uses 1 - 255 to indicate the source of TiDB.
// For now, 1 - 15 are reserved for TiCDC to implement BDR synchronization.
// 16 - 255 are reserved for extendability.
const CDC_WRITE_SOURCE_BITS: u64 = 8;
const CDC_WRITE_SOURCE_MAX: u64 = (1 << CDC_WRITE_SOURCE_BITS) - 1;

// TiCDC uses 1-15 to indicate the change from a lossy DDL reorg Backfill job.
// For now, we only use 1 for column reorg backfill job.
#[cfg(test)]
const LOSSY_DDL_REORG_SOURCE_BITS: u64 = 4;
#[cfg(test)]
const LOSSY_DDL_COLUMN_REORG_SOURCE: u64 = 1;
#[cfg(test)]
const LOSSY_DDL_REORG_SOURCE_MAX: u64 = (1 << LOSSY_DDL_REORG_SOURCE_BITS) - 1;
const LOSSY_DDL_REORG_SOURCE_SHIFT: u64 = CDC_WRITE_SOURCE_BITS;

/// For kv.TxnSource
/// We use an uint64 to represent the source of a transaction.
/// The first 8 bits are reserved for TiCDC, and the next 4 bits are reserved
/// for Lossy DDL reorg Backfill job. The remaining 52 bits are reserved for
/// extendability.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Default)]
pub(crate) struct TxnSource(u64);


Review comment (suggestion):
```suggestion
// |RESERVED|LOSSY_DDL_REORG_SOURCE_BITS|RESERVED|CDC_WRITE_SOURCE_BITS|
// |  52    |             4             |   4    |          4          |
```


Code after:
// Copyright 2023 TiKV Project Authors. Licensed under Apache-2.0.

// The bitmap:
// |RESERVED|LOSSY_DDL_REORG_SOURCE_BITS|CDC_WRITE_SOURCE_BITS|
// |  48    |             8             | 4(RESERVED) |  4    |
//
// TiCDC uses 1 - 255 to indicate the source of TiDB.
// For now, 1 - 15 are reserved for TiCDC to implement BDR synchronization.
// 16 - 255 are reserved for extendability.
const CDC_WRITE_SOURCE_BITS: u64 = 8;
const CDC_WRITE_SOURCE_MAX: u64 = (1 << CDC_WRITE_SOURCE_BITS) - 1;

// TiCDC uses 1-255 to indicate the change from a lossy DDL reorg Backfill job.
// For now, we only use 1 for column reorg backfill job.
#[cfg(test)]
const LOSSY_DDL_REORG_SOURCE_BITS: u64 = 8;
#[cfg(test)]
const LOSSY_DDL_COLUMN_REORG_SOURCE: u64 = 1;
#[cfg(test)]
const LOSSY_DDL_REORG_SOURCE_MAX: u64 = (1 << LOSSY_DDL_REORG_SOURCE_BITS) - 1;
const LOSSY_DDL_REORG_SOURCE_SHIFT: u64 = CDC_WRITE_SOURCE_BITS;

/// For kv.TxnSource
/// We use an uint64 to represent the source of a transaction.
/// The first 8 bits are reserved for TiCDC, and the next 8 bits are reserved
/// for Lossy DDL reorg Backfill job. The remaining 48 bits are reserved for
/// extendability.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Default)]
pub(crate) struct TxnSource(u64);
Repository: tikv/tikv
File: components/raftstore/src/store/worker/metrics.rs (Rust)

Code before:
// Copyright 2016 TiKV Project Authors. Licensed under Apache-2.0.

use std::{cell::RefCell, time::Duration};

use lazy_static::lazy_static;
use prometheus::{local::LocalIntCounter, *};
use prometheus_static_metric::*;
use tikv_util::time::Instant;

make_auto_flush_static_metric! {
    pub label_enum SnapType {
       generate,
       apply
    }

    // snapshot task status
    // |all---------start--------------|
    //               |
    //               |
    //               V
    // |success|abort|fail|delay|ignore|
    //
    // ingest_delay is used to records occurrences of delayed ingestions
    // during snapshot apply (e.g. due to high L0 file count).
    pub label_enum SnapStatus {
        all,
        start,
        success,
        abort,
        fail,
        delay,
        ignore,
        ingest_delay,
    }

    pub struct SnapCounter: LocalIntCounter {
        "type" => SnapType,
        "status" => SnapStatus,
    }

    pub struct CheckSplitCounter : LocalIntCounter {
        "type" => SnapStatus,
    }

    pub struct SnapHistogram : LocalHistogram {
        "type" => SnapType,
    }
}

Review comment (suggestion):
```suggestion
    // ingest_delay is used to record occurrences of delayed ingestions
```

Code after:
// Copyright 2016 TiKV Project Authors. Licensed under Apache-2.0.

use std::{cell::RefCell, time::Duration};

use lazy_static::lazy_static;
use prometheus::{local::LocalIntCounter, *};
use prometheus_static_metric::*;
use tikv_util::time::Instant;

make_auto_flush_static_metric! {
    pub label_enum SnapType {
       generate,
       apply
    }

    // snapshot task status
    // |all---------start--------------|
    //               |
    //               |
    //               V
    // |success|abort|fail|delay|ignore|
    //
    // ingest_delay is used to record occurrences of delayed ingestions
    // during snapshot apply (e.g. due to high L0 file count).
    pub label_enum SnapStatus {
        all,
        start,
        success,
        abort,
        fail,
        delay,
        ignore,
        ingest_delay,
    }

    pub struct SnapCounter: LocalIntCounter {
        "type" => SnapType,
        "status" => SnapStatus,
    }

    pub struct CheckSplitCounter : LocalIntCounter {
        "type" => SnapStatus,
    }

    pub struct SnapHistogram : LocalHistogram {
        "type" => SnapType,
    }
}
Repository: tikv/tikv
File: components/cdc/src/channel.rs (Rust)

Code before:
    }

    pub fn build(self) -> Vec<ChangeDataEvent> {
        self.buffer
    }

    // Return the total bytes of event and resolved ts.
    pub fn statistics(&self) -> (usize, usize) {
        (self.total_event_bytes, self.total_resolved_ts_bytes)
    }
}

pub fn channel(buffer: usize, memory_quota: Arc<MemoryQuota>) -> (Sink, Drain) {
    let (unbounded_sender, unbounded_receiver) = unbounded();
    let (bounded_sender, bounded_receiver) = bounded(buffer);
    (
        Sink {
            unbounded_sender,
            bounded_sender,
            memory_quota: memory_quota.clone(),
        },
        Drain {
            unbounded_receiver,
            bounded_receiver,
            memory_quota,
            conn_id: Default::default(),
        },
    )
}

#[derive(Clone, Debug, PartialEq)]
pub enum SendError {
    Full,
    Disconnected,
    Congested,
}

impl std::error::Error for SendError {}

impl_display_as_debug!(SendError);

macro_rules! impl_from_future_send_error {
    ($($f:ty,)+) => {
        $(
            impl From<$f> for SendError {
                fn from(e: $f) -> Self {
                    if e.is_disconnected() {
                        SendError::Disconnected
                    } else if e.is_full() {
                        Self::Full
                    } else {

Review comment (suggestion):
If a `conn_id` is needed, try:

```rust
pub fn channel(conn_id: ConnID, buffer: usize, memory_quota: Arc<MemoryQuota>) -> (Sink, Drain)
```

Code after:
    }

    pub fn build(self) -> Vec<ChangeDataEvent> {
        self.buffer
    }

    // Return the total bytes of event and resolved ts.
    pub fn statistics(&self) -> (usize, usize) {
        (self.total_event_bytes, self.total_resolved_ts_bytes)
    }
}

pub fn channel(conn_id: ConnId, buffer: usize, memory_quota: Arc<MemoryQuota>) -> (Sink, Drain) {
    let (unbounded_sender, unbounded_receiver) = unbounded();
    let (bounded_sender, bounded_receiver) = bounded(buffer);
    (
        Sink {
            unbounded_sender,
            bounded_sender,
            memory_quota: memory_quota.clone(),
        },
        Drain {
            unbounded_receiver,
            bounded_receiver,
            memory_quota,
            conn_id,
        },
    )
}

#[derive(Clone, Debug, PartialEq)]
pub enum SendError {
    Full,
    Disconnected,
    Congested,
}

impl std::error::Error for SendError {}

impl_display_as_debug!(SendError);

macro_rules! impl_from_future_send_error {
    ($($f:ty,)+) => {
        $(
            impl From<$f> for SendError {
                fn from(e: $f) -> Self {
                    if e.is_disconnected() {
                        SendError::Disconnected
                    } else if e.is_full() {
                        Self::Full
                    } else {
Repository: tikv/tikv
File: components/raftstore/src/store/fsm/apply.rs (Rust)

Code before:
            // open files in rocksdb.
            kv_snap,
            to_store_id: self.to_peer.store_id,
            use_plain_file: self.use_plain_file,
        };
        box_try!(region_sched.schedule(snapshot));
        Ok(())
    }

    pub fn set_use_plain_file(
        &mut self,
        snap_min_approximate_size: ReadableSize,
        region_approximate_size: Option<u64>,
        region_approximate_keys: Option<u64>,
    ) {
        fail_point!("on_snap_use_plain_file", |_| {
            self.use_plain_file = true;
        });
        if snap_min_approximate_size.0 == 0 {
            return;
        }
        let snap_min_approximate_keys = std::cmp::max(
            10000,
            (snap_min_approximate_size.as_mb_f64() * 10000.0) as u64,
        );
        // If the size or the count of keys of cf is relatively small, it's recommended
        // to use plain file for mitigating performance issue when ingesting snapshot.
        // TODO: should consider the size of each column family rather than the total
        // size.
        self.use_plain_file = region_approximate_size.unwrap_or(u64::MAX)
            <= snap_min_approximate_size.0
            && region_approximate_keys.unwrap_or(u64::MAX) <= snap_min_approximate_keys;
    }
}

impl Debug for GenSnapTask {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("GenSnapTask")
            .field("region_id", &self.region_id)
            .finish()
    }
}

#[derive(Debug)]
pub enum ObserverType {
    Cdc(ObserveHandle),
    Rts(ObserveHandle),
    Pitr(ObserveHandle),
}

impl ObserverType {

Review comment (suggestion):
```suggestion
        // If the size and the count of keys of cf are relatively small, it's recommended
```
The code seems to indicate `and` rather than `or`

Code after:
            // This snapshot may be held for a long time, which may cause too many
            // open files in rocksdb.
            kv_snap,
            to_store_id: self.to_peer.store_id,
        };
        box_try!(region_sched.schedule(snapshot));
        Ok(())
    }
}

impl Debug for GenSnapTask {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("GenSnapTask")
            .field("region_id", &self.region_id)
            .finish()
    }
}

#[derive(Debug)]
pub enum ObserverType {
    Cdc(ObserveHandle),
    Rts(ObserveHandle),
    Pitr(ObserveHandle),
}

impl ObserverType {
    pub fn handle(&self) -> &ObserveHandle {
        match self {
            ObserverType::Cdc(h) => h,
            ObserverType::Rts(h) => h,
            ObserverType::Pitr(h) => h,
        }
    }
}

#[derive(Debug)]
pub struct ChangeObserver {
    pub ty: ObserverType,
    pub region_id: u64,
}

impl ChangeObserver {
    pub fn from_cdc(region_id: u64, id: ObserveHandle) -> Self {
        Self {
            ty: ObserverType::Cdc(id),
            region_id,
        }
    }

    pub fn from_rts(region_id: u64, id: ObserveHandle) -> Self {
        Self {
Repository: tikv/tikv
File: components/raftstore/src/store/metrics.rs (Rust)

Code before:
        no_enough_sampled_key,
        // The number of sampled keys located on left and right does not meet the threshold.
        no_enough_lr_key,
        // The number of balanced keys does not meet the score.
        no_balance_key,
        // The number of contained keys does not meet the score.
        no_uncross_key,
        // Split info for the top hot CPU region has been collected, ready to split.
        ready_to_split_cpu_top,
        // Hottest key range for the top hot CPU region could not be found.
        empty_hottest_key_range,
        // The top hot CPU region could not be split.
        unable_to_split_cpu_top,
    }

    pub label_enum SnapshotBrWaitApplyEventType {
        sent,
        trivial,
        accepted,
        term_not_match,
        epoch_not_match,
        duplicated,
        finished,
    }

    pub label_enum SnapshotGeneraterBytesType {
        kv,
        sst,
        plain,
    }

    pub struct SnapshotBrWaitApplyEvent : IntCounter {
        "event" => SnapshotBrWaitApplyEventType
    }

    pub label_enum SnapshotBrLeaseEventType {
        create,
        renew,
        expired,
        reset,
    }

    pub struct SnapshotBrLeaseEvent : IntCounter {
        "event" => SnapshotBrLeaseEventType
    }

    pub struct HibernatedPeerStateGauge: IntGauge {
        "state" => {
            awaken,
            hibernated,
        },

Review comment (suggestion):
```suggestion
    pub label_enum SnapshotGenerateBytesType {
```

Code after:
        no_enough_sampled_key,
        // The number of sampled keys located on left and right does not meet the threshold.
        no_enough_lr_key,
        // The number of balanced keys does not meet the score.
        no_balance_key,
        // The number of contained keys does not meet the score.
        no_uncross_key,
        // Split info for the top hot CPU region has been collected, ready to split.
        ready_to_split_cpu_top,
        // Hottest key range for the top hot CPU region could not be found.
        empty_hottest_key_range,
        // The top hot CPU region could not be split.
        unable_to_split_cpu_top,
    }

    pub label_enum SnapshotBrWaitApplyEventType {
        sent,
        trivial,
        accepted,
        term_not_match,
        epoch_not_match,
        duplicated,
        finished,
    }

    pub label_enum SnapshotGenerateBytesType {
        kv,
        sst,
        plain,
    }

    pub struct SnapshotBrWaitApplyEvent : IntCounter {
        "event" => SnapshotBrWaitApplyEventType
    }

    pub label_enum SnapshotBrLeaseEventType {
        create,
        renew,
        expired,
        reset,
    }

    pub struct SnapshotBrLeaseEvent : IntCounter {
        "event" => SnapshotBrLeaseEventType
    }

    pub struct HibernatedPeerStateGauge: IntGauge {
        "state" => {
            awaken,
            hibernated,
        },