From 8532acea53513638d86f285b131734dda50dfd2e Mon Sep 17 00:00:00 2001 From: chrislu Date: Wed, 15 Oct 2025 23:29:49 -0700 Subject: [PATCH] fix: add cache to LookupTopicBrokers to eliminate 26% CPU overhead MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CRITICAL: LookupTopicBrokers was bypassing cache, causing 26% CPU overhead! Problem (from CPU profile): - LookupTopicBrokers: 35.74% CPU (9s out of 25.18s) - ReadTopicConfFromFiler: 26.41% CPU (6.65s) - protojson.Unmarshal: 16.64% CPU (4.19s) - LookupTopicBrokers called b.fca.ReadTopicConfFromFiler() directly on line 35 - Completely bypassed our unified topicCache! Root Cause: LookupTopicBrokers is called VERY frequently by clients (every fetch request needs to know partition assignments). It was calling ReadTopicConfFromFiler directly instead of using the cache, causing: 1. Expensive gRPC calls to filer on every lookup 2. Expensive JSON unmarshaling on every lookup 3. 26%+ CPU overhead on hot path 4. Our cache optimization was useless for this critical path Solution: Created getTopicConfFromCache() helper and updated all callers: Changes to broker_topic_conf_read_write.go: - Added getTopicConfFromCache() - public API for cached topic config reads - Implements same caching logic: check cache -> read filer -> cache result - Handles both positive (conf != nil) and negative (conf == nil) caching - Refactored GetOrGenerateLocalPartition() to use new helper (code dedup) - Now only 14 lines instead of 60 lines (removed duplication) Changes to broker_grpc_lookup.go: - Modified LookupTopicBrokers() to call getTopicConfFromCache() - Changed from: b.fca.ReadTopicConfFromFiler(t) (no cache) - Changed to: b.getTopicConfFromCache(t) (with cache) - Added comment explaining this fixes 26% CPU overhead Cache Strategy: - First call: Cache MISS -> read filer + unmarshal JSON -> cache for 30s - Next 1000+ calls in 30s: Cache HIT -> return cached config immediately - No filer gRPC, no JSON unmarshaling, near-zero CPU - Cache invalidated on topic create/update/delete Expected CPU Reduction: - Before: 26.41% on ReadTopicConfFromFiler + 16.64% on JSON unmarshal = 43% CPU - After: <0.1% (only on cache miss every 30s) - Expected total broker CPU: 25.18s -> ~8s (67% reduction!) Performance Impact (with 250 lookups/sec): - Before: 250 filer reads/sec + 250 JSON unmarshals/sec - After: 0.17 filer reads/sec (5 topics / 30s TTL) - Reduction: 99.93% fewer expensive operations Code Quality: - Eliminated code duplication (60 lines -> 14 lines in GetOrGenerateLocalPartition) - Single source of truth for cached reads (getTopicConfFromCache) - Clear API: "Always use getTopicConfFromCache, never ReadTopicConfFromFiler directly" Testing: - ✅ Compiles successfully - Ready to deploy and measure CPU improvement Priority: CRITICAL - Completes the cache optimization to achieve full performance fix --- weed/mq/broker/broker_grpc_lookup.go | 13 ++-- .../mq/broker/broker_topic_conf_read_write.go | 59 +++++++++++-------- 2 files changed, 41 insertions(+), 31 deletions(-) diff --git a/weed/mq/broker/broker_grpc_lookup.go b/weed/mq/broker/broker_grpc_lookup.go index 17187e526..0cb12017a 100644 --- a/weed/mq/broker/broker_grpc_lookup.go +++ b/weed/mq/broker/broker_grpc_lookup.go @@ -30,14 +30,17 @@ func (b *MessageQueueBroker) LookupTopicBrokers(ctx context.Context, request *mq t := topic.FromPbTopic(request.Topic) ret := &mq_pb.LookupTopicBrokersResponse{} - conf := &mq_pb.ConfigureTopicResponse{} ret.Topic = request.Topic - if conf, err = b.fca.ReadTopicConfFromFiler(t); err != nil { + + // Use cached topic config to avoid expensive filer reads (26% CPU overhead!) + conf, err := b.getTopicConfFromCache(t) + if err != nil { glog.V(0).Infof("lookup topic %s conf: %v", request.Topic, err) - } else { - err = b.ensureTopicActiveAssignments(t, conf) - ret.BrokerPartitionAssignments = conf.BrokerPartitionAssignments + return ret, err } + + err = b.ensureTopicActiveAssignments(t, conf) + ret.BrokerPartitionAssignments = conf.BrokerPartitionAssignments return ret, err } diff --git a/weed/mq/broker/broker_topic_conf_read_write.go b/weed/mq/broker/broker_topic_conf_read_write.go index f926afd01..601199223 100644 --- a/weed/mq/broker/broker_topic_conf_read_write.go +++ b/weed/mq/broker/broker_topic_conf_read_write.go @@ -18,10 +18,38 @@ import ( ) func (b *MessageQueueBroker) GetOrGenerateLocalPartition(t topic.Topic, partition topic.Partition) (localTopicPartition *topic.LocalPartition, getOrGenError error) { - // get or generate a local partition + // get or generate a local partition using cached topic config + conf, err := b.getTopicConfFromCache(t) + if err != nil { + glog.Errorf("topic %v not found: %v", t, err) + return nil, fmt.Errorf("topic %v not found: %w", t, err) + } + + localTopicPartition, _, getOrGenError = b.doGetOrGenLocalPartition(t, partition, conf) + if getOrGenError != nil { + glog.Errorf("topic %v partition %v not setup: %v", t, partition, getOrGenError) + return nil, fmt.Errorf("topic %v partition %v not setup: %w", t, partition, getOrGenError) + } + return localTopicPartition, nil +} + +// invalidateTopicCache removes a topic from the unified cache +// Should be called when a topic is created, deleted, or config is updated +func (b *MessageQueueBroker) invalidateTopicCache(t topic.Topic) { + topicKey := t.String() + b.topicCacheMu.Lock() + delete(b.topicCache, topicKey) + b.topicCacheMu.Unlock() + glog.V(4).Infof("Invalidated topic cache for %s", topicKey) +} + +// getTopicConfFromCache reads topic configuration with caching +// Returns the config or error if not found. Uses unified cache to avoid expensive filer reads. +// This is the public API for reading topic config - always use this instead of direct filer reads. +func (b *MessageQueueBroker) getTopicConfFromCache(t topic.Topic) (*mq_pb.ConfigureTopicResponse, error) { topicKey := t.String() - // Check unified cache first to avoid expensive filer reads (60% CPU overhead!) + // Check unified cache first b.topicCacheMu.RLock() if entry, found := b.topicCache[topicKey]; found { if time.Now().Before(entry.expiresAt) { @@ -35,12 +63,7 @@ func (b *MessageQueueBroker) GetOrGenerateLocalPartition(t topic.Topic, partitio } glog.V(4).Infof("Topic cache HIT for %s", topicKey) - localTopicPartition, _, getOrGenError = b.doGetOrGenLocalPartition(t, partition, conf) - if getOrGenError != nil { - glog.Errorf("topic %v partition %v not setup: %v", t, partition, getOrGenError) - return nil, fmt.Errorf("topic %v partition %v not setup: %w", t, partition, getOrGenError) - } - return localTopicPartition, nil + return conf, nil } } b.topicCacheMu.RUnlock() @@ -59,7 +82,6 @@ func (b *MessageQueueBroker) GetOrGenerateLocalPartition(t topic.Topic, partitio } b.topicCacheMu.Unlock() glog.V(4).Infof("Topic cached as non-existent: %s", topicKey) - glog.Errorf("topic %v not found: %v", t, readConfErr) return nil, fmt.Errorf("topic %v not found: %w", t, readConfErr) } @@ -70,23 +92,8 @@ func (b *MessageQueueBroker) GetOrGenerateLocalPartition(t topic.Topic, partitio } b.topicCacheMu.Unlock() glog.V(4).Infof("Topic config cached for %s", topicKey) - - localTopicPartition, _, getOrGenError = b.doGetOrGenLocalPartition(t, partition, conf) - if getOrGenError != nil { - glog.Errorf("topic %v partition %v not setup: %v", t, partition, getOrGenError) - return nil, fmt.Errorf("topic %v partition %v not setup: %w", t, partition, getOrGenError) - } - return localTopicPartition, nil -} - -// invalidateTopicCache removes a topic from the unified cache -// Should be called when a topic is created, deleted, or config is updated -func (b *MessageQueueBroker) invalidateTopicCache(t topic.Topic) { - topicKey := t.String() - b.topicCacheMu.Lock() - delete(b.topicCache, topicKey) - b.topicCacheMu.Unlock() - glog.V(4).Infof("Invalidated topic cache for %s", topicKey) + + return conf, nil } func (b *MessageQueueBroker) doGetOrGenLocalPartition(t topic.Topic, partition topic.Partition, conf *mq_pb.ConfigureTopicResponse) (localPartition *topic.LocalPartition, isGenerated bool, err error) {