Repository: MystenLabs/sui
File: crates/sui-indexer-alt-framework/src/pipeline/concurrent/collector.rs (Rust)

Code before:
                    batch,
                    batch_len,
                    watermark,
                };

                if tx.send(batched_rows).await.is_err() {
                    info!(
                        pipeline = H::NAME,
                        "Committer closed channel, stopping collector"
                    );
                    return Ok(());
                }

                if batch_len == 0 {
                    break;
                }
            }

            // Fall back to select! for waiting.
            tokio::select! {
                _ = poll.tick() => {
                    let (batch, watermark, batch_len) = build_batch::<H>(
                        &*handler,
                        &mut pending,
                        &checkpoint_lag_reporter,
                        &metrics,
                    );
                    pending_rows -= batch_len;

                    let batched_rows = BatchedRows {
                        batch,
                        batch_len,
                        watermark,
                    };

                    if tx.send(batched_rows).await.is_err() {
                        info!(
                            pipeline = H::NAME,
                            "Committer closed channel, stopping collector"
                        );
                        break;
                    }

                    if pending_rows > 0 {
                        poll.reset_immediately();
                    } else if rx.is_closed() && rx.is_empty() {
                        info!(
                            pipeline = H::NAME,
                            "Processor closed channel, pending rows empty, stopping collector",
                        );
                        break;

Review comment (performance):
Factoring out the functions does help a bit, but I'm not really seeing the value of still having this piece at the end here -- why not have a `loop` with `poll.tick()` at the top, followed by a `try_recv` loop, followed by batch construction? It would be a simpler implementation (could probably get rid of the helper functions again).

Code after:
                        &checkpoint_lag_reporter,
                        &metrics,
                    );
                    pending_rows -= batch_len;

                    let batched_rows = BatchedRows {
                        batch,
                        batch_len,
                        watermark,
                    };

                    if tx.send(batched_rows).await.is_err() {
                        info!(
                            pipeline = H::NAME,
                            "Committer closed channel, stopping collector"
                        );
                        return Ok(());
                    }

                    if batch_len == 0 {
                        break;
                    }
                }
            }

            // Fall back to select! for waiting.
            tokio::select! {
                _ = poll.tick() => {
                    let (batch, watermark, batch_len) = build_batch::<H>(
                        &*handler,
                        &mut pending,
                        &checkpoint_lag_reporter,
                        &metrics,
                    );
                    pending_rows -= batch_len;

                    let batched_rows = BatchedRows {
                        batch,
                        batch_len,
                        watermark,
                    };

                    if tx.send(batched_rows).await.is_err() {
                        info!(
                            pipeline = H::NAME,
                            "Committer closed channel, stopping collector"
                        );
                        break;
                    }

                    if pending_rows > 0 {