Persistent queues storage redesign [draft]

[andsel]

Motivation

Moquette has always used a local key value storage for its necessity, started with MapDB and then switched to H2 MVStore. The things that the broker stores are:

• session state for clients that use clean session false
  • subscriptions
  • last will message
  • retained message for topics
• queued messages for QoS1/QoS2.

This type of storage are essentially some form of BTrees, which is ideal to keep things ordered and with fast random access time.
While this characteristic is useful for session state information, it's sub-optimal for queues (FIFO in out case).
This discussion is to describe the design I've in mind for this particular aspect, shaped as a draft work in progress PR #638.

Requirements

The implementation should resort to OS features to speed up the I/O operations, permit concurrent access in read and write and have a design that's easy to grasp, because the maintenance of it is part of the Moquette project.

High level point of view

The idea is to use memory mapped files, to store parts of the queues, named pages. All the queues are contained in the same file structures, there isn't the concept of a file per queue, because that means to have an extra file descriptor for each client, and the clients could be a lot.
Each page ( 64Mb wide) is subdivided into fixed segments (4 Mb), a segment is owned exclusively by only a queue. A queue is composed of multiple segments and an index file keeps the list of all segments for each queue.
Pages are named with incremental integer name, for example page.0, page.1 and so one.
Each queue has a tail and head pointers, that refers (page, offset) inside the page.

We have the invariant, for each queue

head >= tail

The write operation to a queue move forward the head pointer, the same happens for the read against the tail pointer.

Write

The write could happen with 2 cases, respect of the payload it's saving:

• the payload + size header (4 bytes) are contained into the head segment
• the payload + size header (4 bytes) can't be fully contained into the head segment

In case 1, the write happens in the current segment, so it happens with a move forward of the head pointer (CAS operation) and then the bytes are effectively written to the mapped region. If the CAS operation on the head pointer doesn't success then simply retry, in a form of spin loop.

In case 2, it needs to write across two segments, so this time with a global lock (used by all the queues) it gets a new fresh segment, and write the payload split across the segments and release the lock.

The motivation for this, is that majority of the writes happens with CAS move of the head pointer. In particular, related to the new session event loop concept, given that the queues are per session, we are sure that there is no contention on the CAS.

Read

The read operation is symmetric to write, the tail point point to first byte to read, so conceptually it should read the 4 length bytes that prepend every payload and move forward the tail pointer with a CAS operation.
Like the write, there could be 2 cases:

• the payload + size header are contained in the segment.
• the payload + size header aren't contained in the segment.
  In the first, the reader, read 4 bytes from the ail pointer, calculate the new position and if fall in the same segment, it does CAS move operation, if it fails, restart.
  In the second case, if the pointer exceeds the tail segment limit, then the read operation enters the global lock, and retrieve the next segment, reading the data inside the lock and then releasing it.

[hylkevds]

The theory so far sounds solid. Still, I have some questions (of course :))

• In the second read-case, is the global lock needed? The queue already "owns" both segments. After the read it can return the now empty segment to the pool, does it need the lock for that?
• Will you be adding details on how segment-reuse is done, and how new pages are created and old ones cleaned up?
• With 64MB per page, and 4MB per segment, that makes 16 segments per page, and thus at least one page per 16 clients. Should those numbers be configurable?
• One theoretical problem may be the case with messages larger than 4MB, since they can spread across more than 2 segments. I'm not sure how common that is, but it would be a really good use case for mapping things to disk instead of keeping them in-memory!

[andsel]

Thanks @hylkevds for the questions, I'll try to respond. Question are always good!

In the second read-case, is the global lock needed? The queue already "owns" both segments. After the read it can return the now empty segment to the pool, does it need the lock for that?

It's needed because multiple threads could read from same queue, so switch of the last segment and the swap of the tail pointer must be done in a lock. This is the more general case, if we have only thread that can read from the queue, your assumption is correct.

Will you be adding details on how segment-reuse is done,

Sure, I'm thinking if we could compact segments in pages to avoid page level fragmentation. The idea is to not reuse segments, but once it's consumed it can be considered as trash.

and how new pages are created

when the current page has no more segments, a new page is created

and old ones cleaned up?

The problem I currently have is when a long time queue rarely consumed has a segment is in the same page of another queue which is immediately consumed. In this case we end with a Page with only one segment still active.
We could imagine to do a compaction, but this requires a stop-all-queue operation, because during compaction the pointers are moved around.
As a simpler option, the compaction could be done only during the bootstrap of the queue system.

With 64MB per page, and 4MB per segment, that makes 16 segments per page, and thus at least one page per 16 clients. Should those numbers be configurable?

I haven't thought about it, changing it would mean that the read logic is not more consistent, unless this information is associated to each page. This information could be done stored into the checkpoint file.

One theoretical problem may be the case with messages larger than 4MB

A message could be split across multiple segments, when a message cross more then one segment then a global lock needs to be entered. The idea is that the mean case a queue contains small messages, so that 4Mb can keep many small messages and favor the concurrent access, blocking only on cross segments messages.

[hylkevds]

Is there a specific reason to not reuse a segment? Once a segment is released by a queue, the next queue that needs a new segment can take it. Reusing segments would greatly reduce the problem of fragmentation and of blocks staying active even though only 1 segment is in use.

[andsel]

Is there a specific reason to not reuse a segment?

No, I thought that have that sort of immutability, one segment once written could only be read, would avoid concurrency problems.
But there's not real point in doing it. The segments allocator once notified that a segment is completely consumed, could track it as reusable.

[robinwu2008]

Is Cassandra or ScyllaDB a good persistence solution?

[andsel]

Hi @robinwu2008 no they are not good for the purpose to be embeddable in Moquette. However Moquette has already switched to the new persistent queue implementation sin version 0.17.

[robinwu2008]

Hi，Aandsel。
Thanks for your reply. I meant to use an external persistence solution, not embedd

[robinwu2008]

Recently, I plan to use mqtt broker to implement an im system. If external storage (such as hbase, ScyllaDB, etc.) can be used, it will not only realize offline messages, but also record massive historical messages.

[andsel]

Moquette has a minimal storage abstraction interface, maybe you can start from that to implement yours that wrap calls to HBase or what else.

Remember that the retention in MQTT is not intended to be an historical storage of message, but just temporary.