Quadratic behavior in timer scheduling

[travisdowns]

Description

Timer scheduling is O(n^2) in the number of timers (under a reasonable assumption about the interval distribution).

How to reproduce

Consume from any topic with 20k partitions with a non-zero statistics interval configured. Note a large amount of time spent in rd_kafka_timer_schedule_next (usually actually inside rd_kafka_timer_cmp which is inlined), both during initial creation of the topic and then even during idle (with poll being called periodically), with a typical stack for the latter case shown below:

rdk:main 905626 339264.203807:   32365636 cycles: 
            55648d5c00fc rd_kafka_timer_cmp+0x4c (inlined)
            55648d5c00fc rd_kafka_timer_schedule_next+0x4c (/home/tdowns/dev/librdkafka-travis/examples/many_consumers)
            55648d5c0908 rd_kafka_timer_schedule+0x1e8 (inlined)
            55648d5c0908 rd_kafka_timer_schedule+0x1e8 (inlined)
            55648d5c0908 rd_kafka_timers_run+0x1e8 (/home/tdowns/dev/librdkafka-travis/examples/many_consumers)
            55648d5947e8 rd_kafka_thread_main+0x228 (/home/tdowns/dev/librdkafka-travis/examples/many_consumers)
            7f8f785c2809 start_thread+0x139 (/usr/lib/x86_64-linux-gnu/libc.so.6)

The problem is that timer scheduling is quadratic, because each time a timer is popped from the front, we do a linear search through the list to find the new position, which in this case is always the end of the list (the timers all have the same interval so a just popped timer will reschedule with a time older than the others). So in one interval 20k timers will be resheduled each of which will look through 20k entries.

The overwhelming amount time is spent in this hot assembly loop:

This is just doing the linked list traversal and looking for the end of the list or the position to insert the element. At best, this loop can execute one iteration every 4 cycles, so 20k reshedules costs 20,000^2 * 4 = 1.6 billion cycles, or half a second of CPU time on a 3.2 GHz machine. With a higher partition count or more clients, the CPU can be saturated just running timers.

In this case, the timers were the statistics callbacks rktp_consumer_lag_tmr but it could apply to any timer. As a workaround for this specific high partition count + stats case, we can disable statistics.

This occurs on current master f7f527d8f2ff7f5bd86856ddc43115eb4dfbba97.

Checklist

IMPORTANT: We will close issues where the checklist has not been completed.

Please provide the following information:

• [x] librdkafka version (release number or git tag): f7f527d8f2ff7f5bd86856ddc43115eb4dfbba97
• [x] Redpanda 22
• [x] librdkafka client configuration:

  
  # Global config
  builtin.features = gzip,snappy,sasl,regex,lz4,sasl_plain,plugins
  client.id = rdkafka
  client.software.name = librdkafka
  message.max.bytes = 1000000
  message.copy.max.bytes = 65535
  receive.message.max.bytes = 100000000
  max.in.flight.requests.per.connection = 1000000
  metadata.request.timeout.ms = 10
  topic.metadata.refresh.interval.ms = 300000
  metadata.max.age.ms = 900000
  topic.metadata.refresh.fast.interval.ms = 250
  topic.metadata.refresh.fast.cnt = 10
  topic.metadata.refresh.sparse = true
  topic.metadata.propagation.max.ms = 30000
  debug = 
  socket.timeout.ms = 60000
  socket.blocking.max.ms = 1000
  socket.send.buffer.bytes = 0
  socket.receive.buffer.bytes = 0
  socket.keepalive.enable = false
  socket.nagle.disable = false
  socket.max.fails = 1
  broker.address.ttl = 1000
  broker.address.family = any
  socket.connection.setup.timeout.ms = 30000
  connections.max.idle.ms = 0
  enable.sparse.connections = true
  reconnect.backoff.jitter.ms = 0
  reconnect.backoff.ms = 100
  reconnect.backoff.max.ms = 10000
  statistics.interval.ms = 1000
  enabled_events = 0
  error_cb = 0x5643d7876600
  throttle_cb = 0x5643d7876250
  stats_cb = 0x5643d78767d0
  log_cb = 0x5643d7878fd0
  log_level = 6
  log.queue = false
  log.thread.name = true
  enable.random.seed = true
  log.connection.close = true
  socket_cb = 0x5643d788e800
  open_cb = 0x5643d78b1770
  default_topic_conf = 0x5643d85407d0
  internal.termination.signal = 29
  api.version.request = true
  api.version.request.timeout.ms = 10000
  api.version.fallback.ms = 0
  broker.version.fallback = 0.10.0
  security.protocol = plaintext
  ssl.ca.certificate.stores = Root
  ssl.engine.id = dynamic
  enable.ssl.certificate.verification = true
  ssl.endpoint.identification.algorithm = none
  sasl.mechanisms = GSSAPI
  sasl.kerberos.service.name = kafka
  sasl.kerberos.principal = kafkaclient
  sasl.kerberos.kinit.cmd = kinit -R -t "%{sasl.kerberos.keytab}" -k %{sasl.kerberos.principal} || kinit -t "%{sasl.kerberos.keytab}" -k %{sasl.kerberos.principal}
  sasl.kerberos.min.time.before.relogin = 60000
  enable.sasl.oauthbearer.unsecure.jwt = false
  enable_sasl_queue = false
  sasl.oauthbearer.method = default
  test.mock.num.brokers = 0
  test.mock.broker.rtt = 0
  group.id = g2
  partition.assignment.strategy = roundrobin
  session.timeout.ms = 45000
  heartbeat.interval.ms = 3000
  group.protocol.type = consumer
  coordinator.query.interval.ms = 600000
  max.poll.interval.ms = 300000
  enable.auto.commit = true
  auto.commit.interval.ms = 5000
  enable.auto.offset.store = false
  queued.min.messages = 100000
  queued.max.messages.kbytes = 65536
  fetch.wait.max.ms = 500
  fetch.message.max.bytes = 1048576
  fetch.max.bytes = 52428800
  fetch.min.bytes = 1
  fetch.error.backoff.ms = 500
  offset.store.method = broker
  isolation.level = read_committed
  offset_commit_cb = 0x5643d7876290
  enable.partition.eof = false
  check.crcs = false
  allow.auto.create.topics = false
  client.rack = 
  transaction.timeout.ms = 60000
  enable.idempotence = false
  enable.gapless.guarantee = false
  queue.buffering.max.messages = 100000
  queue.buffering.max.kbytes = 1048576
  queue.buffering.max.ms = 5
  message.send.max.retries = 2147483647
  retry.backoff.ms = 100
  queue.buffering.backpressure.threshold = 1
  compression.codec = none
  batch.num.messages = 10000
  batch.size = 1000000
  delivery.report.only.error = false
  sticky.partitioning.linger.ms = 10

Topic config

request.required.acks = -1 request.timeout.ms = 30000 message.timeout.ms = 300000 queuing.strategy = fifo produce.offset.report = false partitioner = consistent_random compression.codec = inherit compression.level = -1 auto.commit.enable = true auto.commit.interval.ms = 60000 auto.offset.reset = smallest offset.store.path = . offset.store.sync.interval.ms = -1 offset.store.method = broker consume.callback.max.messages = 0

 - [x] Operating system: Ubuntu 21.10

[edenhill]

This is a great bug report, @travisdowns!

Having that many partitions for a small number of topics is a bit of a special use-case which librdkafka is not optimized for (obviously), but this should be fixed nontheless. Another problem with consumer lag monitoring is that it creates one OffsetsRequest per partition, even though the same broker may be leader for multiple partitions, so in your case that's also a very large amount of protocol requests that will be sent - which in turn may cause head of line blocking for more necessary requests.

So, two action points:

1. reimplement timers, perhaps using a timer wheel like https://25thandclement.com/~william/projects/timeout.c.html
2. try to consolidate consumer_lag requests

[travisdowns]

Having that many partitions for a small number of topics is a bit of a special use-case which librdkafka is not optimized for (obviously), but this should be fixed nontheless.

Yes, it is a bit special, but unfortunately we cannot always control the number of partitions someone chooses to use and there may be some valid reasons for such a high number (real world case: need to distribute among 10,000s of clients, need at least one partition per client if using a dynamic consumer group).

Your suggestions sound great. I also think even something O(log(n)) like a priority heap would be fine, but timer wheel looks even more specialized to this problem (not sure about the overhead at very small counts, priority heap has similar overhead to a linked list for small counts, which is nice).