src/rdkafka_broker.c
static rd_kafka_buf_t *rd_kafka_waitresp_find (rd_kafka_broker_t *rkb,
					       int32_t corrid) {
	rd_kafka_buf_t *rkbuf;
	rd_ts_t now = rd_clock();

	rd_kafka_assert(rkb->rkb_rk, thrd_is_current(rkb->rkb_thread));

	TAILQ_FOREACH(rkbuf, &rkb->rkb_waitresps.rkbq_bufs, rkbuf_link)
		if (rkbuf->rkbuf_corrid == corrid) {
			/* Convert ts_sent to RTT */
			rkbuf->rkbuf_ts_sent = now - rkbuf->rkbuf_ts_sent;
			rd_avg_add(&rkb->rkb_avg_rtt, rkbuf->rkbuf_ts_sent);

                        if (rkbuf->rkbuf_flags & RD_KAFKA_OP_F_BLOCKING &&
			    rd_atomic32_sub(&rkb->rkb_blocking_request_cnt,
					    1) == 1)
				rd_kafka_brokers_broadcast_state_change(
					rkb->rkb_rk);

			rd_kafka_bufq_deq(&rkb->rkb_waitresps, rkbuf);
			return rkbuf;
		}
	return NULL;
}
/**
 * Find a waitresp (rkbuf awaiting response) by the correlation id.
 */
Find a waitresp (rkbuf awaiting response) by the correlation id.
Find a waitresp ( rkbuf awaiting response ) by the correlation id .
static rd_kafka_buf_t *rd_kafka_waitresp_find (rd_kafka_broker_t *rkb,
					       int32_t corrid) {
	rd_kafka_buf_t *rkbuf;
	rd_ts_t now = rd_clock();
	rd_kafka_assert(rkb->rkb_rk, thrd_is_current(rkb->rkb_thread));
	TAILQ_FOREACH(rkbuf, &rkb->rkb_waitresps.rkbq_bufs, rkbuf_link)
		if (rkbuf->rkbuf_corrid == corrid) {
			rkbuf->rkbuf_ts_sent = now - rkbuf->rkbuf_ts_sent;
			rd_avg_add(&rkb->rkb_avg_rtt, rkbuf->rkbuf_ts_sent);
                        if (rkbuf->rkbuf_flags & RD_KAFKA_OP_F_BLOCKING &&
			    rd_atomic32_sub(&rkb->rkb_blocking_request_cnt,
					    1) == 1)
				rd_kafka_brokers_broadcast_state_change(
					rkb->rkb_rk);
			rd_kafka_bufq_deq(&rkb->rkb_waitresps, rkbuf);
			return rkbuf;
		}
	return NULL;
}
static rd_kafka_buf_t * rd_kafka_waitresp_find ( rd_kafka_broker_t * rkb , int32_t corrid ) { rd_kafka_buf_t * rkbuf ; rd_ts_t now = rd_clock ( ) ; rd_kafka_assert ( rkb -> rkb_rk , thrd_is_current ( rkb -> rkb_thread ) ) ; TAILQ_FOREACH ( rkbuf , & rkb -> rkb_waitresps . rkbq_bufs , rkbuf_link ) if ( rkbuf -> rkbuf_corrid == corrid ) { rkbuf -> rkbuf_ts_sent = now - rkbuf -> rkbuf_ts_sent ; rd_avg_add ( & rkb -> rkb_avg_rtt , rkbuf -> rkbuf_ts_sent ) ; if ( rkbuf -> rkbuf_flags & RD_KAFKA_OP_F_BLOCKING && rd_atomic32_sub ( & rkb -> rkb_blocking_request_cnt , 1 ) == 1 ) rd_kafka_brokers_broadcast_state_change ( rkb -> rkb_rk ) ; rd_kafka_bufq_deq ( & rkb -> rkb_waitresps , rkbuf ) ; return rkbuf ; } return NULL ; }
Find a waitresp (rkbuf awaiting response) by the correlation id.
Find a waitresp ( rkbuf awaiting response ) by the correlation id .
/* Convert ts_sent to RTT */