perf: use `hscan` and `scan` for retained messages

[robertsLando]

actually we are using keys method to retrieve retained messages from redis. This is generally a bad approach as stated here.

I also think we could someway improving scan to do some pattern matching (when possible).

Example:

async getRetainedMessages(patterns) {
    const qOpts = qlobberOptions;

    const packets = [];

    const wildcardPatterns = [];

    const qlobber = new QlobberTrue(qlobberOptions);

    for (const p of patterns) {
      if (!p.includes(qOpts.wildcard_some) || !p.includes(qOpts.wildcard_one)) {
        const packet = await this.getRetainedMessage(p);
        if (packet) {
          packets.push(packet);
        }
      } else {
        qlobber.add(p);
        wildcardPatterns.push(p);
      }
    }

    patterns = wildcardPatterns;

    if (patterns.length === 0) return packets;

    // use hscan to iterate over all RETAINEDKEY keys
    let cursor = '0';

    const patternsKey = this.getPatternsKey(patterns);

    do {
      // in NON-CLUSTER mode we store using `HSET` so we have to use `HSCAN` to iterate over the keys
      const [newCursor, keys] = this.isCluster
        ? await this.redisClient.scanBuffer(cursor, 'MATCH', patternsKey)
        : await this.redisClient.hscanBuffer(
            RETAINEDKEY,
            cursor,
            'MATCH',
            patternsKey
          );

      cursor = newCursor.toString();

      for (let i = 0; i < keys.length; i += 2) {
        const topic = this.isCluster
          ? decodeURIComponent(keys[i].toString().split(':')[1])
          : keys[i];
        if (qlobber.test(topic.toString())) {
          const p = this.decodeRetainedPacket(keys[i + 1]);
          if (p) {
            packets.push(p);
          }
        }
      }
    } while (cursor !== '0');

    return packets;
  }

/**
   * Returns an optimized redis key to look for when matching provided patterns
   * prevents to iterate over ALL retained keys when looking for a pattern
   */
  getPatternsKey(patterns): string {
    let topic = '';

    let stop = false;
    let charIndex = 0;

    while (!stop) {
      const char = patterns[0][charIndex];
      // ensure all patterns have the same char at the same index
      // stop if pattern ends or char is different or is a wildcard
      for (const pattern of patterns) {
        if (
          pattern.length <= charIndex ||
          pattern[charIndex] !== char ||
          [
            qlobberOptions.wildcard_one,
            qlobberOptions.wildcard_some,
          ].includes(char)
        ) {
          stop = true;
          break;
        }
      }

      if (!stop) {
        topic += char;
        charIndex++;
      }
    }

    return (this.isCluster ? this.retainedKey(topic) : topic) + '*';
  }

@seriousme thoughts on this?

[seriousme]

Redis indeed advises to use SCAN instead of KEYS to avoid performance spikes. So we should do that anyway.

As for the performance optimization:

The naive solution as currently used is of course to fetch all retained keys and match them in the persistence layer, discarding the ones that are not needed and fetching the values of the remaining ones. This works great if there is a relative small number of topics that contain retained keys.

Optimizing this by directly fetching non-wildcard topics and using the common prefix of wildcard topics might help with performance.

We might skip a scan if no wildcards are used, but we might also ask for values that do not exist.
Even with the common prefix the whole table will need to be SCAN-ed. However we save on keys that need to cross the wire.

So it really depends on:

• the total amount of retained keys
  • that determines if a SCAN is expensive on the Redis side and if the transfer is expensive
• how many non-existent topics are asked for
  • my guess would be that this number is small
• the depth of the topic tree versus the width of the tree, combined with the access pattern.

I have no hard data to base a decision on but the additional effort and additional complexity seems small.
At worst we have only added a bit of complexity, at best we might skip a table scan and all others are in between.

The algorithm for finding the shortest prefix might be simplified by sorting the patterns first and then only comparing first and last item, e.g. something like:

function longestCommonPrefixSorted(strs) {
  if (!strs || strs.length === 0) {
    return "";
  }

  strs.sort(); // Sort the array lexicographically
  const first = strs[0];
  const last = strs[strs.length - 1];
  let i = 0;

  while (i < first.length && i < last.length && first[i] === last[i] ) {
    i++;
  }
  return first.substring(0, i);
}

Btw: this does not look at wildcards yet but that is easy to add.

I'm happy to add this performance optimization if you want.

Kind regards,
Hans

[robertsLando]

I have no hard data to base a decision on but the additional effort and additional complexity seems small.
At worst we have only added a bit of complexity, at best we might skip a table scan and all others are in between.

I agree with this, I think the complexity added is worth the benefits in terms of performance.

I'm happy to add this performance optimization if you want.

Perfect, thanks 🙏🏼

[getlarge]

Another option would be to implement a trie to lookup exactly for the topic parts you need. It (probably) involves extra Redis requests but avoid fetching too many keys. After testing this approach for RabbitMQ to implement restoration of retained messages using filter patterns, it was more efficient this way and it scaled better when the amount of keys grows.

If you are interested, you could draw inspiration from this PR and of course, Qlobber could be a good source.

[seriousme]

Building the trie in Redis is technically possible but add quite some complexity (coding Redis Functions in Lua, testing it all etc) and I'm not sure that it's worth it, because performance will be depending on usage as explained above.

Raw speed comes from the stuff that you don't have to do. And in the latency pyramid network and disk (even SSD) will be slower than RAM which in turn is slower than CPU cache.

If we would only have one topic then we could probably create a superfast MQTT server with lots of stuff hardcoded where we just could read the subscribe and put the mostly preencoded message on the NIC. Superfast, but a veeerrry specific use case.

As far as I know we don't know how most of the users use Aedes + Redis, so we need to maintain a solution that works for most of the use cases.

The challenge of course is that RAM more expensive than SSD + network, and is not as persistent.
(Btw: the assumption here is that price is a constraint, for some use cases it might not be!)

So if we can fit all topics+messages in RAM then we are faster then when we need to roundtrip to Redis.

But if we want to reduce RAM consumption then the first thing to drop is messages.
This assumes that messages are bigger than topics, but if the usecase only exchanges small numbers or booleans then this assumption is already off!

By keeping topics in memory we can very efficiently query Redis for the messages.
We already use HLRU, so we could use that as well to keep the most used messages in memory.

But this is also an assumption. If messages are only exchanged between 2 clients then caching them is useless.
However if the same message is retained and picked up later by 1000 clients we just might have saved a 1000 table scans.

So, what is a wise thing to do to avoid overcomplicating stuff and at the same time offer a good solution to many usecases?

The way I understand (but I could be wrong) Aedes-cached-persistence it is more a cluster facilty to manage subscriptions across a cluster of brokers and less of a cache. It would be possible to create one or more configurable cache modules which apply various cache strategies (e.g. a plain writethrough cache for topics and retained messages, a cache which retains all topics but uses a LRU to retain messages etc). By separating this from the storage backend they can be plugged on top of various backends. I think we do need to cater for clusters of brokers as well (although that is also an assumption!). Currently that is done by sharing the database, but it could also be done differently.

Seems this message is way longer than I initially thought, sorry for that. Hope this sparks some ideas.

Kind regards,
Hans

[robertsLando]

@seriousme Thanks for your detailed explaination, I agree with you and I think that for now it would be better to keep things easier then possible, your idea about creating different cache strategies based on use case is good but I think a bit overkill and hard to maintain giving the number of contributors Aedes have.

I would keep going with the suggestion proposed at start

[robertsLando]

Closed by #122