krishvishal commented on code in PR #4502:
URL: https://github.com/apache/iggy/pull/4502#discussion_r4259828422
##########
gateways/kafka/src/protocol/handlers/produce.rs:
##########
@@ -206,6 +209,71 @@ fn stub_outcome(api_version: i16, request:
&ProduceRequest) -> HandleOutcome {
respond_or_close(encode_response(api_version, request), "Produce")
}
+/// Answers 7 for every partition. `acks=0` has no response to carry it, so
that closes instead.
+fn timed_out(api_version: i16, request: &ProduceRequest) -> HandleOutcome {
+ if request.acks == ACKS_NONE {
+ HandleOutcome::Close
+ } else {
+ respond_or_close(
+ encode_uniform_response(api_version, request,
ERROR_REQUEST_TIMED_OUT),
+ "Produce",
+ )
+ }
+}
+
+/// What the request may hold decoded, in frame bytes, at most one partition
cap.
+///
+/// Partitions decode one at a time, so the largest counts. Uncompressed, a
partition holds its own
+/// bytes or a record slot per declared record, whichever is more. Compressed,
it may fill the cap.
+fn decode_cost(request: &ProduceRequest, max_frame_size: usize) -> usize {
+ let largest = request
+ .topic_data
+ .iter()
+ .flat_map(|topic| &topic.partition_data)
+ .filter_map(|partition| partition.records.as_deref())
+ .map(|blob| {
+ if is_compressed(blob) {
+ max_frame_size
+ } else {
+ let slots = declared_records(blob).unwrap_or(0);
+ blob.len().max(slots.saturating_mul(FRAME_BYTES_PER_RECORD))
+ }
+ })
+ .max()
+ .unwrap_or(0);
+ largest.min(max_frame_size)
+}
+
+/// A large slot when the request is large, and its decode memory.
+type Admission<'a> = (Option<SemaphorePermit<'a>>, SemaphorePermit<'a>);
+
+/// Waits for the request's decode memory. A large request,
`CODEC_OFF_WORKER_BYTES` or more,
+/// first waits for a large slot, so while it waits its turn it holds no
memory.
+async fn admit(
+ state: &GatewayState,
+ cost: usize,
+) -> std::result::Result<Admission<'_>, AcquireError> {
+ let large = if cost >= CODEC_OFF_WORKER_BYTES {
+ Some(state.produce_large.acquire().await?)
+ } else {
+ None
+ };
+ let permits = u32::try_from(cost).unwrap_or(u32::MAX);
+ let memory = state.produce_memory.acquire_many(permits).await?;
+ Ok((large, memory))
+}
+
+/// The producer id of the request's first batch, or the connection when that
batch has none.
+fn producer_of(request: &ProduceRequest, connection: &ConnectionState) ->
Producer {
Review Comment:
Done, Producer::Id is used only for ids this gateway gave out, together with
the sasl user, and everything else uses the connection.
##########
gateways/kafka/src/protocol/api.rs:
##########
@@ -343,9 +347,15 @@ pub struct GatewayState {
pub(crate) probe_board: ProbeBoard,
}
-/// Each holds one decoded partition at a time, so about 160 MB at the default
8 MiB frame. Sends
-/// run one at a time anyway.
-const PRODUCE_SLOTS: usize = 4;
+/// Partition caps Produce may hold decoded at once, about 160 MB at the
default 8 MiB frame.
+///
+/// Counted in bytes, not requests: a small request takes a small share, so it
does not queue
+/// behind a few large ones.
+const PRODUCE_CAPS: usize = 4;
+
+/// Large Produce requests that decode and send at once. More at once only
contend for the
+/// gateway's workers and for Iggy: at full load they cost CPU and gave no
throughput.
+const PRODUCE_LARGE: usize = 4;
Review Comment:
Made it 5 caps with 4 large slots, and the test now admits 4 full frame
requests and then a small one.
##########
gateways/kafka/src/bridge/iggy_bridge/produce.rs:
##########
@@ -19,58 +19,171 @@
//!
//! Separate from `fetch.rs` so the two never edit one file.
+use std::collections::HashMap;
use std::future::Future;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex, PoisonError};
+use std::time::Duration;
use iggy::prelude::{IggyError, IggyMessage, MessageClient, Partitioning};
use tokio::sync::Semaphore;
-use tokio::time::{Instant, timeout};
+use tokio::time::{Instant, timeout, timeout_at};
+use super::fetch::LazyClient;
use super::{IggyBridge, SLOT_LIMIT, TopicTarget, in_slot, permit_by};
use crate::bridge::error::BridgeError;
/// Frame bytes besides the messages: request header, batch header and ids.
const SEND_FRAME_ROOM: u64 = 4 * 1024;
+/// Produce sessions open at once, each an Iggy connection.
+const PRODUCE_SESSIONS: usize = 256;
+
+/// A free session unused this long closes when a new producer opens one.
+const SESSION_IDLE: Duration = Duration::from_secs(600);
+
+/// Who sends a Produce request. Each producer gets an Iggy session of its own.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum Producer {
+ /// The producer id of the request's record batches.
+ Id(i64),
+ /// The Kafka connection, for a producer without an id.
+ Connection(u64),
+}
+
+/// One producer's Iggy client, and one send at a time on it, so its sends
reach Iggy in order.
+pub struct ProduceSession {
+ client: LazyClient,
+ turn: Arc<Semaphore>,
+}
+
+/// The Produce sessions, one per producer, so different producers send at
once.
+pub(super) struct ProducePool {
+ sessions: Mutex<HashMap<Producer, Held>>,
+ capacity: usize,
+}
+
+/// A session, and when a request last took it.
+struct Held {
+ session: Arc<ProduceSession>,
+ used: Instant,
+}
+
+impl ProducePool {
+ pub(super) fn new() -> Self {
+ Self::with_capacity(PRODUCE_SESSIONS)
+ }
+
+ fn with_capacity(capacity: usize) -> Self {
+ Self {
+ sessions: Mutex::new(HashMap::new()),
+ capacity,
+ }
+ }
+
+ /// The session of `producer`, opened if it has none. `None` when the pool
is full and every
+ /// session is busy.
+ ///
+ /// Opening one first drops the free sessions idle past [`SESSION_IDLE`],
then the least
+ /// recently used free one if the pool is still full.
+ fn take(&self, producer: Producer) -> Option<Arc<ProduceSession>> {
+ let now = Instant::now();
+ let mut sessions =
self.sessions.lock().unwrap_or_else(PoisonError::into_inner);
+ if let Some(held) = sessions.get_mut(&producer) {
+ held.used = now;
+ return Some(Arc::clone(&held.session));
+ }
+ sessions
+ .retain(|_, held| !(held.is_free() &&
now.duration_since(held.used) >= SESSION_IDLE));
Review Comment:
Added both, the session closes and logs out when the kafka connection ends,
and there is a sweep every 30s.
##########
gateways/kafka/README.md:
##########
@@ -507,13 +512,16 @@ cut before that threshold is reached - worth knowing
rather than discovering lat
### Concurrency ceiling
-- One `IggyClient` serves Produce, Metadata, CreateTopics, DescribeConfigs and
AlterConfigs for
- every Kafka connection, one Iggy request at a time. Fetch polls use 4 more,
one per read slot.
- Topic probes use 1 more. Offset calls use 8 more: a commit client and a
read client per offset slot.
+- Each producer sends on an `IggyClient` of its own, one send at a time, so
its batches land in
+ order. A producer is its producer id, or its connection when its batches
carry none. Up to 256
+ of these stay open. Opening one closes the free ones idle for 10 minutes,
then the least
+ recently used free one if 256 are open. A Produce request that finds all 256
busy answers 7.
+- Metadata, CreateTopics, DeleteTopics, DescribeConfigs and AlterConfigs share
1 more client, one
+ Iggy request at a time. Fetch polls use 4 more, one per read slot. Topic
probes use 1 more.
+ Offset calls use 8 more: a commit client and a read client per offset slot.
- OffsetCommit and OffsetFetch cost one Iggy call per partition, spread over
the offset slots. In a
cluster, an offset commit client can also open one connection per node for
stores.
-- `IGGY_KAFKA_MAX_CONNECTIONS` does not change that. A Produce client pool is
a TODO in
- [docs/SCOPE.md](docs/SCOPE.md).
+- `IGGY_KAFKA_MAX_CONNECTIONS` does not change that.
Review Comment:
Updated all four.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
