src/rdkafka_broker.c
static void rd_kafka_broker_ops_io_serve (rd_kafka_broker_t *rkb,
                                          rd_ts_t abs_timeout) {
        rd_ts_t now;

        if (unlikely(rd_kafka_terminating(rkb->rkb_rk)))
                abs_timeout = rd_clock() + 1000;
        else if (unlikely(rd_kafka_broker_needs_connection(rkb)))
                abs_timeout = RD_POLL_NOWAIT;
        else if (unlikely(abs_timeout == RD_POLL_INFINITE))
                abs_timeout = rd_clock() +
                        ((rd_ts_t)rd_kafka_max_block_ms * 1000);

        if (likely(rkb->rkb_transport != NULL)) {
                /* Serve IO events.
                 *
                 * If there are IO events, cut out the queue ops_serve
                 * timeout (below) since we'll probably have to perform more
                 * duties based on the IO.
                 * IO polling granularity is milliseconds while
                 * queue granularity is microseconds. */
                if (rd_kafka_transport_io_serve(
                            rkb->rkb_transport,
                            rd_timeout_remains(abs_timeout)))
                        abs_timeout = RD_POLL_NOWAIT;
        }


        /* Serve broker ops */
        rd_kafka_broker_ops_serve(rkb, rd_timeout_remains_us(abs_timeout));

        /* An op might have triggered the need for a connection, if so
         * transition to TRY_CONNECT state. */
        if (unlikely(rd_kafka_broker_needs_connection(rkb) &&
                     rkb->rkb_state == RD_KAFKA_BROKER_STATE_INIT)) {
                rd_kafka_broker_lock(rkb);
                rd_kafka_broker_set_state(
                        rkb, RD_KAFKA_BROKER_STATE_TRY_CONNECT);
                rd_kafka_broker_unlock(rkb);
        }

        /* Scan queues for timeouts. */
        now = rd_clock();
        if (rd_interval(&rkb->rkb_timeout_scan_intvl, 1000000, now) > 0)
                rd_kafka_broker_timeout_scan(rkb, now);
}
/**
 * @brief Serve broker ops and IOs.
 *
 * If a connection exists, poll IO first based on timeout.
 * Use remaining timeout for ops queue poll.
 *
 * If no connection, poll ops queue using timeout.
 *
 * Sparse connections: if there's need for a connection, set
 *                     timeout to NOWAIT.
 *
 * @param abs_timeout Maximum block time (absolute time).
 *
 * @locality broker thread
 * @locks none
 */
@brief Serve broker ops and IOs.
If a connection exists, poll IO first based on timeout.
Use remaining timeout for ops queue poll.

If no connection, poll ops queue using timeout.

Sparse connections: if there's need for a connection, set
timeout to NOWAIT.

@param abs_timeout Maximum block time (absolute time).

@locality broker thread
@locks none
@brief Serve broker ops and IOs . If a connection exists poll IO first based on timeout . Use remaining timeout for ops queue poll . If no connection poll ops queue using timeout . Sparse connections : if there ' s need for a connection set timeout to NOWAIT . @param abs_timeout Maximum block time ( absolute time ) . @locality broker thread @locks none
static void rd_kafka_broker_ops_io_serve (rd_kafka_broker_t *rkb,
                                          rd_ts_t abs_timeout) {
        rd_ts_t now;
        if (unlikely(rd_kafka_terminating(rkb->rkb_rk)))
                abs_timeout = rd_clock() + 1000;
        else if (unlikely(rd_kafka_broker_needs_connection(rkb)))
                abs_timeout = RD_POLL_NOWAIT;
        else if (unlikely(abs_timeout == RD_POLL_INFINITE))
                abs_timeout = rd_clock() +
                        ((rd_ts_t)rd_kafka_max_block_ms * 1000);
        if (likely(rkb->rkb_transport != NULL)) {
                if (rd_kafka_transport_io_serve(
                            rkb->rkb_transport,
                            rd_timeout_remains(abs_timeout)))
                        abs_timeout = RD_POLL_NOWAIT;
        }
        rd_kafka_broker_ops_serve(rkb, rd_timeout_remains_us(abs_timeout));
        if (unlikely(rd_kafka_broker_needs_connection(rkb) &&
                     rkb->rkb_state == RD_KAFKA_BROKER_STATE_INIT)) {
                rd_kafka_broker_lock(rkb);
                rd_kafka_broker_set_state(
                        rkb, RD_KAFKA_BROKER_STATE_TRY_CONNECT);
                rd_kafka_broker_unlock(rkb);
        }
        now = rd_clock();
        if (rd_interval(&rkb->rkb_timeout_scan_intvl, 1000000, now) > 0)
                rd_kafka_broker_timeout_scan(rkb, now);
}
static void rd_kafka_broker_ops_io_serve ( rd_kafka_broker_t * rkb , rd_ts_t abs_timeout ) { rd_ts_t now ; if ( unlikely ( rd_kafka_terminating ( rkb -> rkb_rk ) ) ) abs_timeout = rd_clock ( ) + 1000 ; else if ( unlikely ( rd_kafka_broker_needs_connection ( rkb ) ) ) abs_timeout = RD_POLL_NOWAIT ; else if ( unlikely ( abs_timeout == RD_POLL_INFINITE ) ) abs_timeout = rd_clock ( ) + ( ( rd_ts_t ) rd_kafka_max_block_ms * 1000 ) ; if ( likely ( rkb -> rkb_transport != NULL ) ) { if ( rd_kafka_transport_io_serve ( rkb -> rkb_transport , rd_timeout_remains ( abs_timeout ) ) ) abs_timeout = RD_POLL_NOWAIT ; } rd_kafka_broker_ops_serve ( rkb , rd_timeout_remains_us ( abs_timeout ) ) ; if ( unlikely ( rd_kafka_broker_needs_connection ( rkb ) && rkb -> rkb_state == RD_KAFKA_BROKER_STATE_INIT ) ) { rd_kafka_broker_lock ( rkb ) ; rd_kafka_broker_set_state ( rkb , RD_KAFKA_BROKER_STATE_TRY_CONNECT ) ; rd_kafka_broker_unlock ( rkb ) ; } now = rd_clock ( ) ; if ( rd_interval ( & rkb -> rkb_timeout_scan_intvl , 1000000 , now ) > 0 ) rd_kafka_broker_timeout_scan ( rkb , now ) ; }
@brief Serve broker ops and IOs.
@brief Serve broker ops and IOs .
/* Serve IO events.
                 *
                 * If there are IO events, cut out the queue ops_serve
                 * timeout (below) since we'll probably have to perform more
                 * duties based on the IO.
                 * IO polling granularity is milliseconds while
                 * queue granularity is microseconds. */ /* Serve broker ops */ /* An op might have triggered the need for a connection, if so
         * transition to TRY_CONNECT state. */ /* Scan queues for timeouts. */