Repository: moquette-io/moquette
Pull request #691 (2022-11-23 12:59:11)

Persisted unsafe queues implementation
## Release notes
Implements a disk persistent queues system that is not thread safe.

## What it does?
Create the implementation of segmented queues. The data is stored in mmapped memory pages, where each page is divided in segments. Each queue own a list of segments. The access to the each segment is intended to be done by the same thread, so no serialization is needed. The only critical section, guarded by a lock, is the interaction to require or release segment.

- close #638

hylkevds: This work looks really interesting. How far do you think you are till the first integration with Moquette?
andsel: I think we are close, just want to change the API of `dequeue` to return `Optional.empty()` instead of null to signal the empty queue condition.
andsel: Then integrate into Moquette, with feature flag to enable it and create a new release.
hylkevds: I'm looking forward to performance-testing it in combination with the work from #608 (and #648). That should greatly increase the stability of long-running servers.

Patch:
diff --git a/broker/src/main/java/io/moquette/broker/unsafequeues/PagedFilesAllocator.java b/broker/src/main/java/io/moquette/broker/unsafequeues/PagedFilesAllocator.java
new file mode 100644
index 00000000..d259eaf2
--- /dev/null
+++ b/broker/src/main/java/io/moquette/broker/unsafequeues/PagedFilesAllocator.java
@@ -0,0 +1,110 @@
+package io.moquette.broker.unsafequeues;
+
+import java.io.IOException;
+import java.nio.MappedByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.Properties;
+
+/**
+ * Default implementation of SegmentAllocator. It uses a series of files (named pages) and split them in segments.
+ *
+ * This class is not thread safe.
+ * */
+class PagedFilesAllocator implements SegmentAllocator {
+
+    interface AllocationListener {
+        void segmentedCreated(String name, Segment segment);
+    }
+
+    public static final int MB = 1024 * 1024;
+    public static final int PAGE_SIZE = 64 * MB;
+    private final Path pagesFolder;
+    private final int segmentSize;
+    private int lastSegmentAllocated;
+    private int lastPage;
+    private MappedByteBuffer currentPage;
+    private FileChannel currentPageFile;
+
+    PagedFilesAllocator(Path pagesFolder, int segmentSize, int lastPage, int lastSegmentAllocated) throws QueueException {
+        this.pagesFolder = pagesFolder;
+        this.segmentSize = segmentSize;
+        this.lastPage = lastPage;
+        this.lastSegmentAllocated = lastSegmentAllocated;
+        this.currentPage = openRWPageFile(this.pagesFolder, this.lastPage);
+    }
+
+    private MappedByteBuffer openRWPageFile(Path pagesFolder, int pageId) throws QueueException {
+        final Path pageFile = pagesFolder.resolve(String.format("%d.page", pageId));
+        boolean createNew = false;
+        if (!Files.exists(pageFile)) {
+            try {
+                pageFile.toFile().createNewFile();
+                createNew = true;
+            } catch (IOException ex) {
+                throw new QueueException("Reached an IO error during the bootstrapping of empty 'checkpoint.properties'", ex);
+            }
+        }
+
+        try (FileChannel fileChannel = FileChannel.open(pageFile, StandardOpenOption.READ, StandardOpenOption.WRITE)) {
+            this.currentPageFile = fileChannel;
+            final MappedByteBuffer mappedPage = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, PAGE_SIZE);
+            // DBG
+            if (createNew && QueuePool.queueDebug) {
+                for (int i = 0; i < PAGE_SIZE; i++){
+                    mappedPage.put(i, (byte) 'C');
+                }
+            }
+            // DBG
+            return mappedPage;
+        } catch (IOException e) {
+            throw new QueueException("Can't open page file " + pageFile, e);
+        }
+    }
+
+    @Override
+    public Segment nextFreeSegment() throws QueueException {
+        if (currentPageIsExhausted()) {
+            lastPage ++;
+            currentPage = openRWPageFile(pagesFolder, lastPage);
+            lastSegmentAllocated = 0;
+        }
+
+        final int beginOffset = lastSegmentAllocated * segmentSize;
+        final int endOffset = ((lastSegmentAllocated + 1) * segmentSize) - 1;
+
+        lastSegmentAllocated += 1;
+        return new Segment(currentPage, new SegmentPointer(lastPage, beginOffset), new SegmentPointer(lastPage, endOffset));
+    }
+
+    @Override
+    public Segment reopenSegment(int pageId, int beginOffset) throws QueueException {
+        final MappedByteBuffer page = openRWPageFile(pagesFolder, pageId);
+        final SegmentPointer begin = new SegmentPointer(pageId, beginOffset);
+        final SegmentPointer end = new SegmentPointer(pageId, beginOffset + segmentSize - 1);
+        return new Segment(page, begin, end);
+    }
+
+    @Override
+    public void close() throws QueueException {
+        if (currentPageFile != null) {
+            try {
+                currentPageFile.close();
+            } catch (IOException ex) {
+                throw new QueueException("Problem closing current page file", ex);
+            }
+        }
+    }
+
+    @Override
+    public void dumpState(Properties checkpoint) {
+        checkpoint.setProperty("segments.last_page", String.valueOf(this.lastPage));
+        checkpoint.setProperty("segments.last_segment", String.valueOf(this.lastSegmentAllocated));
+    }
+
+    private boolean currentPageIsExhausted() {
+        return lastSegmentAllocated * segmentSize == PAGE_SIZE;
+    }
+}
diff --git a/broker/src/main/java/io/moquette/broker/unsafequeues/Queue.java b/broker/src/main/java/io/moquette/broker/unsafequeues/Queue.java
new file mode 100644
index 00000000..3ad779a9
--- /dev/null
+++ b/broker/src/main/java/io/moquette/broker/unsafequeues/Queue.java
@@ -0,0 +1,308 @@
+package io.moquette.broker.unsafequeues;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.Buffer;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * Not thread safe disk persisted queue.
+ * */
+public class Queue {
+    private static final Logger LOG = LoggerFactory.getLogger(Queue.class);
+
+    public static final int LENGTH_HEADER_SIZE = 4;
+    private final String name;
+    /* Last wrote byte, point to head byte */
+    private VirtualPointer currentHeadPtr;
+    private Segment headSegment;
+
+    /* First readable byte, point to the last occupied byte */
+    private VirtualPointer currentTailPtr;
+    private Segment tailSegment;
+
+    private final QueuePool queuePool;
+    private final PagedFilesAllocator.AllocationListener allocationListener;
+//    private final ReentrantLock lock = new ReentrantLock();
+
+    Queue(String name, Segment headSegment, VirtualPointer currentHeadPtr,
+          Segment tailSegment, VirtualPointer currentTailPtr,
+          SegmentAllocator allocator, PagedFilesAllocator.AllocationListener allocationListener, QueuePool queuePool) {
+        this.name = name;
+        this.headSegment = headSegment;
+        this.currentHeadPtr = currentHeadPtr;
+        this.currentTailPtr = currentTailPtr;
+        this.tailSegment = tailSegment;
+        this.allocationListener = allocationListener;
+        this.queuePool = queuePool;
+    }
+
+    /**
+     * @throws QueueException if an error happens during access to file.
+     * */
+    public void enqueue(ByteBuffer payload) throws QueueException {
+        final int messageSize = LENGTH_HEADER_SIZE + payload.remaining();
+        if (headSegment.hasSpace(currentHeadPtr, messageSize)) {
+            LOG.debug("Head segment has sufficient space for message length {}", LENGTH_HEADER_SIZE + payload.remaining());
+            writeData(headSegment, currentHeadPtr.plus(1), payload);
+            // move head segment
+            currentHeadPtr = currentHeadPtr.moveForward(messageSize);
+            return;
+        }
+
+        LOG.debug("Head segment doesn't have enough space");
+        // the payload can't be fully contained into the current head segment and needs to be splitted
+        // with another segment.
+
+
+        final int dataSize = payload.remaining();
+        final ByteBuffer rawData = (ByteBuffer) ByteBuffer.allocate(LENGTH_HEADER_SIZE + dataSize)
+            .putInt(dataSize)
+            .put(payload)
+            .flip();
+
+        // the bytes written from the payload input
+        long bytesRemainingInHeaderSegment = Math.min(rawData.remaining(), headSegment.bytesAfter(currentHeadPtr));
+        LOG.trace("Writing partial payload to offset {} for {} bytes", currentHeadPtr, bytesRemainingInHeaderSegment);
+
+        if (bytesRemainingInHeaderSegment > 0) {
+            int copySize = (int) bytesRemainingInHeaderSegment;
+            ByteBuffer slice = rawData.slice();
+            slice.limit(copySize);
+            writeDataNoHeader(headSegment, currentHeadPtr.plus(1), slice);
+            currentHeadPtr = currentHeadPtr.moveForward(bytesRemainingInHeaderSegment);
+            // No need to move newSegmentPointer the pointer because the last spinningMove has already moved it
+
+            // shift forward the consumption point
+            rawData.position(rawData.position() + copySize);
+        }
+
+        Segment newSegment = null;
+
+        // till the payload is not completely stored,
+        // save the remaining part into a new segment.
+        while (rawData.hasRemaining()) {
+            // To request the next segment, it's needed to be done in global lock.
+            newSegment = queuePool.nextFreeSegment();
+            //notify segment creation for queue in queue pool
+            allocationListener.segmentedCreated(name, newSegment);
+
+            int copySize = (int) Math.min(rawData.remaining(), Segment.SIZE);
+            ByteBuffer slice = rawData.slice();
+            slice.limit(copySize);
+
+            currentHeadPtr = currentHeadPtr.moveForward(copySize);
+            writeDataNoHeader(newSegment, newSegment.begin, slice);
+            headSegment = newSegment;
+
+            // shift forward the consumption point
+            rawData.position(rawData.position() + copySize);
+        }
+    }
+
+    private void writeDataNoHeader(Segment segment, SegmentPointer start, ByteBuffer data) {
+        segment.write(start, data);
+    }
+
+    private void writeDataNoHeader(Segment segment, VirtualPointer start, ByteBuffer data) {
+        segment.write(start, data);
+    }
+
+    /**
+     * Writes data and size to the current Head segment starting from start pointer.
+     * */
+    private void writeData(Segment segment, VirtualPointer start, ByteBuffer data) {
+        writeData(segment, start, data.remaining(), data);
+    }
+
+    /**
+     * @param segment the target segment.
+     * @param start where start writing.
+     * @param size the length of the data to write on the segment.
+     * @param data the data to write.
+     * */
+    private void writeData(Segment segment, VirtualPointer start, int size, ByteBuffer data) {
+        ByteBuffer length = (ByteBuffer) ByteBuffer.allocate(LENGTH_HEADER_SIZE).putInt(size).flip();
+        segment.write(start, length); // write 4 bytes header
+        segment.write(start.plus(LENGTH_HEADER_SIZE), data); // write the payload
+    }
+
+    /**
+     * Used in test
+     * */
+    void force() {
+        headSegment.force();
+    }
+
+    VirtualPointer currentHead() {
+        return currentHeadPtr;
+    }
+
+    VirtualPointer currentTail() {
+        return currentTailPtr;
+    }
+
+    public boolean isEmpty() {
+        if (isTailFirstUsage(currentTailPtr)) {
+            return currentHeadPtr.compareTo(currentTailPtr) == 0;
+        } else {
+            return currentHeadPtr.moveForward(1).compareTo(currentTailPtr) == 0;
+        }
+    }
+
+    /**
+     * Read next message or return null if the queue has no data.
+     * */
+    public Optional<ByteBuffer> dequeue() throws QueueException {
+        if (!currentHeadPtr.isGreaterThan(currentTailPtr)) {
+            if (currentTailPtr.isGreaterThan(currentHeadPtr)) {
+                // sanity check
+                throw new QueueException("Current tail " + currentTailPtr + " is forward head " + currentHeadPtr);
+            }
+            // head and tail pointer are the same, the queue is empty
+            return Optional.empty();
+        }
+
+        LOG.debug("currentTail is {}", currentTailPtr);
+        if (containsHeader(tailSegment, currentTailPtr)) {
+            // currentSegment contains at least the header (payload length)
+            final VirtualPointer existingTail;
+            if (isTailFirstUsage(currentTailPtr)) {
+                // move to the first readable byte
+                existingTail = currentTailPtr.plus(1);
+            } else {
+                existingTail = currentTailPtr.copy();
+            }
+            final int payloadLength = tailSegment.readHeader(existingTail);
+            // tail must be moved to the next byte to read, so has to move to
+            // header size + payload size + 1
+            final int fullMessageSize = payloadLength + LENGTH_HEADER_SIZE;
+            if (tailSegment.hasSpace(existingTail, fullMessageSize + 1)) {
+                // currentSegments contains fully the payload
+                currentTailPtr = existingTail.moveForward(fullMessageSize);
+                // read data from currentTail + 4 bytes(the length)
+                final VirtualPointer dataStart = existingTail.moveForward(LENGTH_HEADER_SIZE);
+
+                return Optional.of(readData(tailSegment, dataStart, payloadLength));
+            } else {
+                // payload is split across currentSegment and next ones
+                VirtualPointer dataStart = existingTail.moveForward(LENGTH_HEADER_SIZE);
+
+                LOG.debug("Loading payload size {}", payloadLength);
+                return Optional.of(loadPayloadFromSegments(payloadLength, tailSegment, dataStart));
+            }
+        } else {
+            // header is split across 2 segments
+            // the currentSegment is still the tailSegment
+            // read the length header that's crossing 2 segments
+            final CrossSegmentHeaderResult result = decodeCrossHeader(tailSegment, currentTailPtr);
+
+            // load all payload parts from the segments
+            LOG.debug("Loading payload size {}", result.payloadLength);
+            return Optional.of(loadPayloadFromSegments(result.payloadLength, result.segment, result.pointer));
+        }
+    }
+
+    private static boolean containsHeader(Segment segment, VirtualPointer tail) {
+        return segment.bytesAfter(tail) + 1 >= LENGTH_HEADER_SIZE;
+    }
+
+    private static class CrossSegmentHeaderResult {
+        private final Segment segment;
+        private final VirtualPointer pointer;
+        private final int payloadLength;
+
+        private CrossSegmentHeaderResult(Segment segment, VirtualPointer pointer, int payloadLength) {
+            this.segment = segment;
+            this.pointer = pointer;
+            this.payloadLength = payloadLength;
+        }
+    }
+
+    // TO BE called owning the lock
+    private CrossSegmentHeaderResult decodeCrossHeader(Segment segment, VirtualPointer pointer) throws QueueException {
+        // read first part
+        ByteBuffer lengthBuffer = ByteBuffer.allocate(LENGTH_HEADER_SIZE);
+        final ByteBuffer partialHeader = segment.readAllBytesAfter(pointer);
+        final int consumedHeaderSize = partialHeader.remaining();
+        lengthBuffer.put(partialHeader);
+        queuePool.consumedTailSegment(name);
+
+        if (QueuePool.queueDebug) {
+            segment.fillWith((byte) 'D');
+        }
+
+        // read second part
+        final int remainingHeaderSize =  LENGTH_HEADER_SIZE - consumedHeaderSize;
+        Segment nextTailSegment = queuePool.openNextTailSegment(name);
+        lengthBuffer.put(nextTailSegment.read(nextTailSegment.begin, remainingHeaderSize));
+        final VirtualPointer dataStart = pointer.moveForward(LENGTH_HEADER_SIZE);
+        int payloadLength = ((ByteBuffer) lengthBuffer.flip()).getInt();
+
+        return new CrossSegmentHeaderResult(nextTailSegment, dataStart, payloadLength);
+    }
+
+    // TO BE called owning the lock on segments allocator
+    private ByteBuffer loadPayloadFromSegments(int remaining, Segment segment, VirtualPointer tail) throws QueueException {
+        List<ByteBuffer> createdBuffers = new ArrayList<>(segmentCountFromSize(remaining));
+        VirtualPointer scan = tail;
+
+        do {
+            LOG.debug("Looping remaining {}", remaining);
+            int availableDataLength = Math.min(remaining, (int) segment.bytesAfter(scan) + 1);
+            final ByteBuffer buffer = segment.read(scan, availableDataLength);
+            createdBuffers.add(buffer);
+            final boolean segmentCompletelyConsumed = (segment.bytesAfter(scan) + 1) == availableDataLength;
+            scan = scan.moveForward(availableDataLength);
+            final boolean consumedQueue = scan.isGreaterThan(currentHead());
+            remaining -= buffer.remaining();
+
+            if (remaining > 0 || (segmentCompletelyConsumed && !consumedQueue)) {
+                queuePool.consumedTailSegment(name);
+                if (QueuePool.queueDebug) {
+                    segment.fillWith((byte) 'D');
+                }
+                segment = queuePool.openNextTailSegment(name);
+            }
+        } while (remaining > 0);
+
+        // assign to tailSegment without CAS because we are in lock
+        tailSegment = segment;
+        currentTailPtr = scan;
+        LOG.debug("Moved currentTailPointer to {} from {}", scan, tail);
+
+        return joinBuffers(createdBuffers);
+    }
+
+    private int segmentCountFromSize(int remaining) {
+        return (int) Math.ceil((double) remaining / Segment.SIZE);
+    }
+
+    private boolean isTailFirstUsage(VirtualPointer tail) {
+        return tail.isUntouched();
+    }
+
+    /**
+     * @return a ByteBuffer that's a composition of all buffers
+     * */
+    private ByteBuffer joinBuffers(List<ByteBuffer> buffers) {
+        final int neededSpace = buffers.stream().mapToInt(Buffer::remaining).sum();
+        byte[] heapBuffer = new byte[neededSpace];
+        int offset = 0;
+        for (ByteBuffer buffer : buffers) {
+            final int readBytes = buffer.remaining();
+            buffer.get(heapBuffer, offset, readBytes);
+            offset += readBytes;
+        }
+
+        return ByteBuffer.wrap(heapBuffer);
+    }
+
+    private ByteBuffer readData(Segment source, VirtualPointer start, int length) {
+        return source.read(start, length);
+    }
+}
diff --git a/broker/src/main/java/io/moquette/broker/unsafequeues/QueueException.java b/broker/src/main/java/io/moquette/broker/unsafequeues/QueueException.java
new file mode 100644
index 00000000..bf233169
--- /dev/null
+++ b/broker/src/main/java/io/moquette/broker/unsafequeues/QueueException.java
@@ -0,0 +1,14 @@
+package io.moquette.broker.unsafequeues;
+
+public class QueueException extends Exception {
+
+    private static final long serialVersionUID = -4782799401089093829L;
+
+    public QueueException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public QueueException(String message) {
+        super(message);
+    }
+}
diff --git a/broker/src/main/java/io/moquette/broker/unsafequeues/QueuePool.java b/broker/src/main/java/io/moquette/broker/unsafequeues/QueuePool.java
new file mode 100644
index 00000000..4cec190f
--- /dev/null
+++ b/broker/src/main/java/io/moquette/broker/unsafequeues/QueuePool.java
@@ -0,0 +1,486 @@
+package io.moquette.broker.unsafequeues;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileReader;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.nio.MappedByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.TreeSet;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
+
+import static io.moquette.broker.unsafequeues.PagedFilesAllocator.PAGE_SIZE;
+
+public class QueuePool {
+
+    private static final Logger LOG = LoggerFactory.getLogger(QueuePool.class);
+
+    static final boolean queueDebug = Boolean.parseBoolean(System.getProperty("moquette.queue.debug", "false"));
+
+    private static SegmentAllocationCallback callback;
+
+    // visible for testing
+    static class SegmentRef implements Comparable<SegmentRef> {
+        final int pageId;
+        final int offset;
+
+        // visible for testing
+        SegmentRef(int pageId, int offset) {
+            this.pageId = pageId;
+            this.offset = offset;
+        }
+
+        public SegmentRef(Segment segment) {
+            this.pageId = segment.begin.pageId();
+            this.offset = segment.begin.offset();
+        }
+
+        @Override
+        public String toString() {
+            return String.format("(%d, %d)", pageId, offset);
+        }
+
+        @Override
+        public int compareTo(SegmentRef o) {
+            final int pageCompare = Integer.compare(pageId, o.pageId);
+            if (pageCompare != 0) {
+                return pageCompare;
+            }
+            return Integer.compare(offset, o.offset);
+        }
+    }
+
+    private static class QueueName {
+        final String name;
+
+        private QueueName(String name) {
+            this.name = name;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            QueueName queueName = (QueueName) o;
+            return Objects.equals(name, queueName.name);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(name);
+        }
+
+        @Override
+        public String toString() {
+            return "QueueName{name='" + name + '\'' + '}';
+        }
+    }
+
+    private final SegmentAllocator allocator;
+    private final Path dataPath;
+    private final int segmentSize;
+    private final ConcurrentMap<QueueName, LinkedList<SegmentRef>> queueSegments = new ConcurrentHashMap<>();
+    private final ConcurrentMap<QueueName, Queue> queues = new ConcurrentHashMap<>();
+    private final ConcurrentSkipListSet<SegmentRef> recycledSegments = new ConcurrentSkipListSet<>();
+    private final ReentrantLock segmentsAllocationLock = new ReentrantLock();
+
+    private QueuePool(SegmentAllocator allocator, Path dataPath, int segmentSize) {
+        this.allocator = allocator;
+        this.dataPath = dataPath;
+        this.segmentSize = segmentSize;
+    }
+
+    private static class SegmentAllocationCallback implements PagedFilesAllocator.AllocationListener {
+
+        private final QueuePool queuePool;
+
+        private SegmentAllocationCallback(QueuePool queuePool) {
+            this.queuePool = queuePool;
+        }
+
+        @Override
+        public void segmentedCreated(String name, Segment segment) {
+            queuePool.segmentedCreated(name, segment);
+        }
+    }
+
+    private void segmentedCreated(String name, Segment segment) {
+        LOG.debug("Registering new segment {} for queue {}", segment, name);
+        final QueueName queueName = new QueueName(name);
+        List<SegmentRef> segmentRefs = this.queueSegments.computeIfAbsent(queueName, k -> new LinkedList<>());
+
+        // adds in head
+        segmentRefs.add(0, new SegmentRef(segment));
+
+        LOG.debug("queueSegments for queue {} after insertion {}", queueName, queueSegments.get(queueName));
+    }
+
+    public static QueuePool loadQueues(Path dataPath) throws QueueException {
+        // read in checkpoint.properties
+        final Properties checkpointProps = createOrLoadCheckpointFile(dataPath);
+
+        // load last references to segment and instantiate the allocator
+        final int lastPage = Integer.parseInt(checkpointProps.getProperty("segments.last_page", "0"));
+        final int lastSegment = Integer.parseInt(checkpointProps.getProperty("segments.last_segment", "0"));
+
+        final PagedFilesAllocator allocator = new PagedFilesAllocator(dataPath, Segment.SIZE, lastPage, lastSegment);
+
+        final QueuePool queuePool = new QueuePool(allocator, dataPath, Segment.SIZE);
+        callback = new SegmentAllocationCallback(queuePool);
+        queuePool.loadQueueDefinitions(checkpointProps);
+        LOG.debug("Loaded queues definitions: {}", queuePool.queueSegments);
+
+        queuePool.loadRecycledSegments(checkpointProps);
+        LOG.debug("Recyclable segments are: {}", queuePool.recycledSegments);
+        return queuePool;
+    }
+
+    private static Properties createOrLoadCheckpointFile(Path dataPath) throws QueueException {
+        final Path checkpointPath = dataPath.resolve("checkpoint.properties");
+        if (!Files.exists(checkpointPath)) {
+            LOG.info("Can't find any file named 'checkpoint.properties' in path: {}, creating new one", dataPath);
+            final boolean notExisted;
+            try {
+                notExisted = checkpointPath.toFile().createNewFile();
+            } catch (IOException e) {
+                throw new QueueException("Reached an IO error during the bootstrapping of empty 'checkpoint.properties'", e);
+            }
+            if (!notExisted) {
+                LOG.warn("Found a checkpoint file while bootstrapping {}", checkpointPath);
+            }
+        }
+
+        final FileReader fileReader;
+        try {
+            fileReader = new FileReader(checkpointPath.toFile());
+        } catch (FileNotFoundException e) {
+            throw new QueueException("Can't find any file named 'checkpoint.properties' in path: " + dataPath, e);
+        }
+        final Properties checkpointProps = new Properties();
+        try {
+            checkpointProps.load(fileReader);
+        } catch (IOException e) {
+            throw new QueueException("if an error occurred when reading from: " + checkpointPath, e);
+        }
+        return checkpointProps;
+    }
+
+    private void loadQueueDefinitions(Properties checkpointProps) throws QueueException {
+        // structure of queues definitions in properties file:
+        // queues.0.name = bla bla
+        // queues.0.segments = head (id_page, offset), (id_page, offset), ... tail
+        // queues.0.head_offset = bytes offset from the start of the page where last data was written
+        // queues.0.tail_offset = bytes offset from the start of the page where first data could be read
+        boolean noMoreQueues = false;
+        int queueId = 0;
+        while (!noMoreQueues) {
+            final String queueKey = String.format("queues.%d.name", queueId);
+            if (!checkpointProps.containsKey(queueKey)) {
+                noMoreQueues = true;
+                continue;
+            }
+            final QueueName queueName = new QueueName(checkpointProps.getProperty(queueKey));
+            LinkedList<SegmentRef> segmentRefs = decodeSegments(checkpointProps.getProperty(String.format("queues.%d.segments", queueId)));
+            final int numSegments = segmentRefs.size();
+            queueSegments.put(queueName, segmentRefs);
+
+            final long headOffset = Long.parseLong(checkpointProps.getProperty(String.format("queues.%d.head_offset", queueId)));
+            final SegmentRef headSegmentRef = segmentRefs.get(0);
+            final SegmentPointer currentHead = new SegmentPointer(headSegmentRef.pageId, headOffset);
+            // TODO this reopen could be done in lazy way during getOrCreate method.
+            Segment headSegment = allocator.reopenSegment(headSegmentRef.pageId, headSegmentRef.offset);
+
+            final long tailOffset = Long.parseLong(checkpointProps.getProperty(String.format("queues.%d.tail_offset", queueId)));
+            final SegmentRef tailSegmentRef = segmentRefs.getLast();
+            final SegmentPointer currentTail = new SegmentPointer(tailSegmentRef.pageId, tailOffset);
+            Segment tailSegment = allocator.reopenSegment(tailSegmentRef.pageId, tailSegmentRef.offset);
+
+            // Create relative positioned head and tail pointers
+            // Tail is an offset relative to start of the first segment in the list
+            // Head is n-1 full segments plus the offset of the physical head
+            final VirtualPointer logicalTail = new VirtualPointer(currentTail.offset());
+            final VirtualPointer logicalHead = new VirtualPointer((long)(numSegments - 1) * Segment.SIZE + currentHead.offset());
+            final Queue queue = new Queue(queueName.name, headSegment, logicalHead, tailSegment, logicalTail,
+                allocator, callback, this);
+            queues.put(queueName, queue);
+
+            queueId++;
+        }
+    }
+
+    private void loadRecycledSegments(Properties checkpointProps) throws QueueException {
+        TreeSet<SegmentRef> usedSegments = new TreeSet<>();
+
+        boolean noMoreQueues = false;
+        int queueId = 0;
+
+        // load all queues definitions from checkpoint file
+        // TODO second use of this, extract as an iterator
+        while (!noMoreQueues) {
+            final String queueKey = String.format("queues.%d.name", queueId);
+            if (!checkpointProps.containsKey(queueKey)) {
+                noMoreQueues = true;
+                continue;
+            }
+            LinkedList<SegmentRef> segmentRefs = decodeSegments(checkpointProps.getProperty(String.format("queues.%d.segments", queueId)));
+            usedSegments.addAll(segmentRefs);
+
+            queueId++;
+        }
+
+        if (usedSegments.isEmpty()) {
+            // no queue definitions were loaded
+            return;
+        }
+
+        final List<SegmentRef> recreatedSegments = recreateSegmentHoles(usedSegments);
+
+        segmentsAllocationLock.lock();
+        try {
+            recycledSegments.addAll(recreatedSegments);
+        } finally {
+            segmentsAllocationLock.unlock();
+        }
+    }
+
+    /**
+     * @param usedSegments sorted set of used segments
+     * */
+    // package-private for testing
+    List<SegmentRef> recreateSegmentHoles(TreeSet<SegmentRef> usedSegments) throws QueueException {
+        // find the holes in the list of used segments
+        if (usedSegments.isEmpty()) {
+            throw new QueueException("Status error, expected to find at least one segment");
+        }
+
+        // prev point to the last examined segment.
+        // recreates segments on left of current segment.
+        SegmentRef prev = null;
+        final List<SegmentRef> recreatedSegments = new LinkedList<>();
+        for (SegmentRef current : usedSegments) {
+            if (prev == null) {
+                // recreate recycled segments before first used segment
+                recreatedSegments.addAll(recreateRecycledSegmentsBetween(current));
+                prev = current;
+                continue;
+            }
+            if (isAdjacent(prev, current)) {
+                // contiguous, skip it
+                prev = current;
+                continue;
+            }
+            if (prev.pageId == current.pageId) {
+                recreatedSegments.addAll(recreateRecycledSegments(prev.offset + segmentSize, current.offset, prev.pageId));
+            } else {
+                // recreate recycled segments between 2 used segments
+                recreatedSegments.addAll(recreateRecycledSegmentsBetween(prev, current));
+            }
+        }
+        return recreatedSegments;
+    }
+
+    private boolean isAdjacent(SegmentRef prev, SegmentRef segment) {
+        if (prev.pageId == segment.pageId) {
+            // same page
+            if (prev.offset + segmentSize == segment.offset) {
+                // contiguous, skip it
+                return true;
+            }
+        } else if (prev.pageId + 1 == segment.pageId) {
+            // adjacent pages, last segment in one and first in the other
+            if (prev.offset == PagedFilesAllocator.PAGE_SIZE - Segment.SIZE && segment.offset == 0) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    private List<SegmentRef> recreateRecycledSegmentsBetween(SegmentRef toSegment) {
+        return recreateRecycledSegmentsBetween(null, toSegment);
+    }
+
+    private List<SegmentRef> recreateRecycledSegmentsBetween(SegmentRef fromSegment, SegmentRef toSegment) {
+        final List<SegmentRef> recreatedSegments = new LinkedList<>();
+        int prevPageId = 0;
+        if (fromSegment != null) {
+            prevPageId = fromSegment.pageId;
+            // holes after previous segment, to complete the page
+            recreatedSegments.addAll(recreateRecycledSegments(fromSegment.offset + segmentSize, PAGE_SIZE, fromSegment.pageId));
+            prevPageId++;
+        }
+
+        // all the intermediate pages
+        for (; prevPageId < toSegment.pageId; prevPageId++) {
+            recreatedSegments.addAll(recreateRecycledSegments(0, PAGE_SIZE, prevPageId));
+        }
+
+        // holes before the current segment
+        recreatedSegments.addAll(recreateRecycledSegments(0, toSegment.offset, toSegment.pageId));
+        return recreatedSegments;
+    }
+
+    private List<SegmentRef> recreateRecycledSegments(int fromOffset, int toOffset, int pageId) {
+        final List<SegmentRef> recreatedSegments = new LinkedList<>();
+        while (fromOffset != toOffset) {
+            recreatedSegments.add(new SegmentRef(pageId, fromOffset));
+            fromOffset = fromOffset + segmentSize;
+        }
+        return recreatedSegments;
+    }
+
+    private LinkedList<SegmentRef> decodeSegments(String s) {
+        final String[] segments = s.substring(s.indexOf("(") + 1, s.lastIndexOf(")"))
+                .split("\\), \\(");
+
+        LinkedList<SegmentRef> acc = new LinkedList<>();
+        for (String segment : segments) {
+            final String[] split = segment.split(",");
+            final int idPage = Integer.parseInt(split[0].trim());
+            final int offset = Integer.parseInt(split[1].trim());
+
+            acc.offer(new SegmentRef(idPage, offset));
+        }
+        return acc;
+    }
+
+    public Queue getOrCreate(String queueName) throws QueueException {
+        final QueueName queueN = new QueueName(queueName);
+        if (queues.containsKey(queueN)) {
+            return queues.get(queueN);
+        } else {
+            // create new queue with first empty segment
+            final Segment segment = nextFreeSegment();
+            //notify segment creation for queue in queue pool
+            segmentedCreated(queueName, segment);
+
+            // When a segment is freshly created the head must the last occupied byte,
+            // so can't be the start of a segment, but one position before, or in case
+            // of a new page, -1
+            final Queue queue = new Queue(queueName, segment, VirtualPointer.buildUntouched(), segment, VirtualPointer.buildUntouched(),
+                this.allocator, callback, this);
+            queues.put(queueN, queue);
+            return queue;
+        }
+    }
+
+    /**
+     * Free mapped files
+     * */
+    public void close() throws QueueException {
+        allocator.close();
+
+        //save all into the checkpoint file
+        Properties checkpoint = new Properties();
+        allocator.dumpState(checkpoint);
+
+        int queueCounter = 0;
+        for (Map.Entry<QueueName, LinkedList<SegmentRef>> entry : queueSegments.entrySet()) {
+            // queues.0.name = bla bla
+            final QueueName queueName = entry.getKey();
+            checkpoint.setProperty("queues." + queueCounter + ".name", queueName.name);
+
+            // queues.0.segments = head (id_page, offset), (id_page, offset), ... tail
+            final LinkedList<SegmentRef> segmentRefs = entry.getValue();
+            final String segmentsDef = segmentRefs.stream()
+                .map(SegmentRef::toString)
+                .collect(Collectors.joining(", "));
+            checkpoint.setProperty("queues." + queueCounter + ".segments", segmentsDef);
+
+            // queues.0.head_offset = bytes offset from the start of the page where last data was written
+            final Queue queue = queues.get(queueName);
+            checkpoint.setProperty("queues." + queueCounter + ".head_offset", String.valueOf(queue.currentHead().segmentOffset()));
+            checkpoint.setProperty("queues." + queueCounter + ".tail_offset", String.valueOf(queue.currentTail().segmentOffset()));
+        }
+
+        final File propertiesFile = dataPath.resolve("checkpoint.properties").toFile();
+        final FileWriter fileWriter;
+        try {
+            fileWriter = new FileWriter(propertiesFile);
+        } catch (IOException ex) {
+            throw new QueueException("Problem opening checkpoint.properties file", ex);
+        }
+        try {
+            checkpoint.store(fileWriter, "DON'T EDIT, AUTOGENERATED");
+        } catch (IOException ex) {
+            throw new QueueException("Problem writing checkpoint.properties file", ex);
+        }
+    }
+
+    Segment openNextTailSegment(String name) throws QueueException {
+        // definition from QueuePool.queueSegments
+        final QueueName queueName = new QueueName(name);
+        final LinkedList<SegmentRef> segmentRefs = queueSegments.get(queueName);
+
+        final SegmentRef pollSegment = segmentRefs.peekLast();
+        if (pollSegment == null) {
+            LOG.error("Queue segments is empty for queue {}, segment references: {}", queueName, segmentRefs);
+            throw new IllegalStateException("Opening tail segment can't never go in empty queue, because it's checked upfront in Queue");
+        }
+
+        final Path pageFile = dataPath.resolve(String.format("%d.page", pollSegment.pageId));
+        if (!Files.exists(pageFile)) {
+            throw new QueueException("Can't find file for page file" + pageFile);
+        }
+
+        final MappedByteBuffer tailPage;
+        try (FileChannel fileChannel = FileChannel.open(pageFile, StandardOpenOption.READ, StandardOpenOption.WRITE)) {
+            tailPage = fileChannel.map(FileChannel.MapMode.READ_WRITE/*READ_ONLY*/, 0, PAGE_SIZE);
+        } catch (IOException ex) {
+            throw new QueueException("Can't open page file " + pageFile, ex);
+        }
+
+        final SegmentPointer begin = new SegmentPointer(pollSegment.pageId, pollSegment.offset);
+        final SegmentPointer end = new SegmentPointer(pollSegment.pageId, pollSegment.offset + segmentSize - 1);
+        return new Segment(tailPage, begin, end);
+    }
+
+    /**
+     * Notify the actual tail segment was completely read
+     * */
+    void consumedTailSegment(String name) {
+        final QueueName queueName = new QueueName(name);
+        final LinkedList<SegmentRef> segmentRefs = queueSegments.get(queueName);
+        final SegmentRef segmentRef = segmentRefs.pollLast();
+        LOG.debug("Consumed tail segment {} from queue {}", segmentRef, queueName);
+        segmentsAllocationLock.lock();
+        try {
+            recycledSegments.add(segmentRef);
+        } finally {
+            segmentsAllocationLock.unlock();
+        }
+    }
+
+    Segment nextFreeSegment() throws QueueException {
+        segmentsAllocationLock.lock();
+        try {
+            if (recycledSegments.isEmpty()) {
+                LOG.debug("no recycled segments available, request the creation of new one");
+                return allocator.nextFreeSegment();
+            }
+            final SegmentRef recycledSegment = recycledSegments.pollFirst();
+            if (recycledSegment == null) {
+                throw new QueueException("Invalid state, expected available recycled segment");
+            }
+            LOG.debug("Reusing recycled segment from page: {} at page offset: {}", recycledSegment.pageId, recycledSegment.offset);
+            return allocator.reopenSegment(recycledSegment.pageId, recycledSegment.offset);
+        } finally {
+            segmentsAllocationLock.unlock();
+        }
+    }
+}
diff --git a/broker/src/main/java/io/moquette/broker/unsafequeues/Segment.java b/broker/src/main/java/io/moquette/broker/unsafequeues/Segment.java
new file mode 100644
index 00000000..19d0887e
--- /dev/null
+++ b/broker/src/main/java/io/moquette/broker/unsafequeues/Segment.java
@@ -0,0 +1,155 @@
+package io.moquette.broker.unsafequeues;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.nio.MappedByteBuffer;
+
+final class Segment {
+    private static final Logger LOG = LoggerFactory.getLogger(Segment.class);
+
+    final static int SIZE = 4 * 1024 * 1024;
+
+    final SegmentPointer begin;
+    final SegmentPointer end;
+
+    private final MappedByteBuffer mappedBuffer;
+
+    Segment(MappedByteBuffer page, SegmentPointer begin, SegmentPointer end) {
+        assert begin.samePage(end);
+        this.begin = begin;
+        this.end = end;
+        this.mappedBuffer = page;
+    }
+
+    boolean hasSpace(VirtualPointer mark, long length) {
+        return bytesAfter(mark) >= length;
+    }
+
+    /**
+     * @return number of bytes in segment after the pointer.
+     * The pointer slot is not counted.
+     * */
+    public long bytesAfter(SegmentPointer mark) {
+        assert mark.samePage(this.end);
+        return end.distance(mark);
+    }
+
+    public long bytesAfter(VirtualPointer mark) {
+        final int pageOffset = rebasedOffset(mark);
+        final SegmentPointer physicalMark = new SegmentPointer(this.end.pageId(), pageOffset);
+        return end.distance(physicalMark);
+    }
+
+    void write(SegmentPointer offset, ByteBuffer content) {
+        final ByteBuffer buffer = (ByteBuffer) mappedBuffer.position(offset.offset());
+        checkContentStartWith(content);
+        buffer.put(content);
+    }
+
+    // fill the segment with value bytes
+    void fillWith(byte value) {
+        LOG.debug("Wipe segment {}", this);
+        final ByteBuffer buffer = (ByteBuffer) mappedBuffer.position(this.begin.offset());
+        for (int i = 0; i < Segment.SIZE; i++) {
+            buffer.put(i, value);
+        }
+    }
+
+    // debug method
+    private void checkContentStartWith(ByteBuffer content) {
+        if (content.get(0) == 0 && content.get(1) == 0 && content.get(2) == 0 && content.get(3) == 0) {
+            System.out.println("DNADBG content starts with 4 zero");
+        }
+    }
+
+    void write(VirtualPointer offset, ByteBuffer content) {
+        final int pageOffset = rebasedOffset(offset);
+        final ByteBuffer buffer = (ByteBuffer) mappedBuffer.position(pageOffset);
+        buffer.put(content);
+    }
+
+    /**
+     * Force flush of memory mapper buffer to disk
+     * */
+    void force() {
+        mappedBuffer.force();
+    }
+
+    /**
+     * return the int value contained in the 4 bytes after the pointer.
+     *
+     * @param pointer virtual pointer to start read from.
+     * */
+    int readHeader(VirtualPointer pointer) {
+        final int rebasedIndex = rebasedOffset(pointer);
+        LOG.debug(" {} {} {} {} at {}", Integer.toHexString(mappedBuffer.get(rebasedIndex)),
+            Integer.toHexString(mappedBuffer.get(rebasedIndex + 1)),
+            Integer.toHexString(mappedBuffer.get(rebasedIndex + 2)),
+            Integer.toHexString(mappedBuffer.get(rebasedIndex + 3)),
+            pointer
+        );
+        return mappedBuffer.getInt(rebasedIndex);
+    }
+
+    /*private*/ int rebasedOffset(VirtualPointer virtualPtr) {
+        return this.begin.plus((int) virtualPtr.segmentOffset()).offset();
+    }
+
+    public ByteBuffer read(VirtualPointer start, int length) {
+        final int pageOffset = rebasedOffset(start);
+        byte[] dst = new byte[length];
+
+        int sourceIdx = pageOffset;
+        for (int dstIndex = 0; dstIndex < length; dstIndex++, sourceIdx++) {
+            dst[dstIndex] = mappedBuffer.get(sourceIdx);
+        }
+
+        return ByteBuffer.wrap(dst);
+    }
+
+    public ByteBuffer read(SegmentPointer start, int length) {
+        byte[] dst = new byte[length];
+
+        if (length > mappedBuffer.remaining() - start.offset())
+            throw new BufferUnderflowException();
+
+        int sourceIdx = start.offset();
+        for (int dstIndex = 0; dstIndex < length; dstIndex++, sourceIdx++) {
+            dst[dstIndex] = mappedBuffer.get(sourceIdx);
+        }
+
+        return ByteBuffer.wrap(dst);
+    }
+
+    private long size() {
+        return end.distance(begin) + 1;
+    }
+
+    @Override
+    public String toString() {
+        return "Segment{page=" + begin.pageId() + ", begin=" + begin.offset() + ", end=" + end.offset() + ", size=" + size() + "}";
+    }
+
+    ByteBuffer readAllBytesAfter(SegmentPointer start) {
+        // WARN, dataStart points to a byte position to read
+        // if currentSegment.end is at offset 1023, and data start is 1020, the bytes after are 4 and
+        // not 1023 - 1020.
+        final long availableDataLength = bytesAfter(start) + 1;
+        final ByteBuffer buffer = read(start, (int) availableDataLength);
+        buffer.rewind();
+        return buffer;
+    }
+
+    ByteBuffer readAllBytesAfter(VirtualPointer start) {
+        // WARN, dataStart points to a byte position to read
+        // if currentSegment.end is at offset 1023, and data start is 1020, the bytes after are 4 and
+        // not 1023 - 1020.
+        final long availableDataLength = bytesAfter(start) + 1;
+        final ByteBuffer buffer = read(start, (int) availableDataLength);
+        buffer.rewind();
+        return buffer;
+    }
+}
diff --git a/broker/src/main/java/io/moquette/broker/unsafequeues/SegmentAllocator.java b/broker/src/main/java/io/moquette/broker/unsafequeues/SegmentAllocator.java
new file mode 100644
index 00000000..18bd7f80
--- /dev/null
+++ b/broker/src/main/java/io/moquette/broker/unsafequeues/SegmentAllocator.java
@@ -0,0 +1,21 @@
+package io.moquette.broker.unsafequeues;
+
+import java.util.Properties;
+
+interface SegmentAllocator {
+
+    /**
+     * Return the next free segment in the current page, or create a new Page if necessary.
+     *
+     * This method has to be invoked inside a lock, it's not thread safe.
+     *
+     * @throws QueueException if any IO error happens on the filesystem.
+     * */
+    Segment nextFreeSegment() throws QueueException;
+
+    Segment reopenSegment(int pageId, int beginOffset) throws QueueException;
+
+    void close() throws QueueException;
+
+    void dumpState(Properties checkpoint);
+}
diff --git a/broker/src/main/java/io/moquette/broker/unsafequeues/SegmentPointer.java b/broker/src/main/java/io/moquette/broker/unsafequeues/SegmentPointer.java
new file mode 100644
index 00000000..86a456dd
--- /dev/null
+++ b/broker/src/main/java/io/moquette/broker/unsafequeues/SegmentPointer.java
@@ -0,0 +1,83 @@
+package io.moquette.broker.unsafequeues;
+
+import java.util.Objects;
+
+final class SegmentPointer implements Comparable<SegmentPointer> {
+    private final int idPage;
+    private final long offset;
+
+    public SegmentPointer(int idPage, long offset) {
+        this.idPage = idPage;
+        this.offset = offset;
+    }
+
+    /**
+     * Construct using the segment, but changing the offset.
+     * */
+    public SegmentPointer(Segment segment, long offset) {
+        this.idPage = segment.begin.idPage;
+        this.offset = offset;
+    }
+
+    /**
+     * Copy constructor
+     * */
+    public SegmentPointer copy() {
+        return new SegmentPointer(idPage, offset);
+    }
+
+    @Override
+    public int compareTo(SegmentPointer other) {
+        if (idPage == other.idPage) {
+            return Long.compare(offset, other.offset);
+        } else {
+            return Integer.compare(idPage, other.idPage);
+        }
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        SegmentPointer that = (SegmentPointer) o;
+        return idPage == that.idPage && offset == that.offset;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(idPage, offset);
+    }
+
+    boolean samePage(SegmentPointer other) {
+        return idPage == other.idPage;
+    }
+
+    SegmentPointer moveForward(long length) {
+        return new SegmentPointer(idPage, offset + length);
+    }
+
+    @Override
+    public String toString() {
+        return "SegmentPointer{idPage=" + idPage + ", offset=" + offset + '}';
+    }
+
+    /**
+     * Calculate the distance in bytes inside the same segment
+     * */
+    public long distance(SegmentPointer other) {
+        assert idPage == other.idPage;
+        return offset - other.offset;
+    }
+
+    int offset() {
+        return (int) offset;
+    }
+
+    public SegmentPointer plus(int delta) {
+        return moveForward(delta);
+    }
+
+    int pageId() {
+        return this.idPage;
+    }
+}
diff --git a/broker/src/main/java/io/moquette/broker/unsafequeues/VirtualPointer.java b/broker/src/main/java/io/moquette/broker/unsafequeues/VirtualPointer.java
new file mode 100644
index 00000000..c49e6c17
--- /dev/null
+++ b/broker/src/main/java/io/moquette/broker/unsafequeues/VirtualPointer.java
@@ -0,0 +1,51 @@
+package io.moquette.broker.unsafequeues;
+
+public class VirtualPointer implements Comparable<VirtualPointer> {
+    final long logicalOffset;
+
+    public static VirtualPointer buildUntouched() {
+        return new VirtualPointer(-1);
+    }
+
+    public VirtualPointer(long logicalOffset) {
+        this.logicalOffset = logicalOffset;
+    }
+
+    @Override
+    public int compareTo(VirtualPointer other) {
+        return Long.compare(logicalOffset, other.logicalOffset);
+    }
+
+    public long segmentOffset() {
+        return logicalOffset % Segment.SIZE;
+    }
+
+    public long logicalOffset() {
+        return logicalOffset;
+    }
+
+    public VirtualPointer moveForward(long delta) {
+        return new VirtualPointer(logicalOffset + delta);
+    }
+
+    public VirtualPointer plus(int i) {
+        return new VirtualPointer(logicalOffset + i);
+    }
+
+    public boolean isGreaterThan(VirtualPointer other) {
+        return this.compareTo(other) > 0;
+    }
+
+    public boolean isUntouched() {
+        return logicalOffset == -1;
+    }
+
+    public VirtualPointer copy() {
+        return new VirtualPointer(this.logicalOffset);
+    }
+
+    @Override
+    public String toString() {
+        return "VirtualPointer{logicalOffset=" + logicalOffset + '}';
+    }
+}


Test patch:
diff --git a/broker/src/test/java/io/moquette/broker/unsafequeues/DummySegmentAllocator.java b/broker/src/test/java/io/moquette/broker/unsafequeues/DummySegmentAllocator.java
new file mode 100644
index 00000000..e9598ba1
--- /dev/null
+++ b/broker/src/test/java/io/moquette/broker/unsafequeues/DummySegmentAllocator.java
@@ -0,0 +1,44 @@
+package io.moquette.broker.unsafequeues;
+
+import java.io.IOException;
+import java.nio.MappedByteBuffer;
+import java.util.Properties;
+
+public class DummySegmentAllocator implements SegmentAllocator {
+
+    @Override
+    public Segment nextFreeSegment() {
+        final MappedByteBuffer pageBuffer = createFreshPageTmpTile();
+        final SegmentPointer begin = new SegmentPointer(0, 0);
+        final SegmentPointer end = new SegmentPointer(0, Segment.SIZE);
+        return new Segment(pageBuffer, begin, end);
+    }
+
+    @Override
+    public Segment reopenSegment(int pageId, int beginOffset) throws QueueException {
+        final MappedByteBuffer pageBuffer = createFreshPageTmpTile();
+        final SegmentPointer begin = new SegmentPointer(pageId, beginOffset);
+        final SegmentPointer end = new SegmentPointer(pageId, beginOffset + Segment.SIZE);
+        return new Segment(pageBuffer, begin, end);
+    }
+
+    private MappedByteBuffer createFreshPageTmpTile() {
+        final MappedByteBuffer pageBuffer;
+        try {
+            pageBuffer = Utils.createPageFile();
+        } catch (IOException ex) {
+            // used only in tests, so it's safe to fail the test with an untyped exception
+            throw new RuntimeException(ex);
+        }
+        return pageBuffer;
+    }
+
+    @Override
+    public void close() throws QueueException {
+        // TODO, maybe
+    }
+
+    @Override
+    public void dumpState(Properties checkpoint) {
+    }
+}
diff --git a/broker/src/test/java/io/moquette/broker/unsafequeues/QueuePoolTest.java b/broker/src/test/java/io/moquette/broker/unsafequeues/QueuePoolTest.java
new file mode 100644
index 00000000..96608b29
--- /dev/null
+++ b/broker/src/test/java/io/moquette/broker/unsafequeues/QueuePoolTest.java
@@ -0,0 +1,355 @@
+package io.moquette.broker.unsafequeues;
+
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.FileReader;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.security.NoSuchAlgorithmException;
+import java.security.SecureRandom;
+import java.util.*;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import static io.moquette.broker.unsafequeues.Queue.LENGTH_HEADER_SIZE;
+import static io.moquette.broker.unsafequeues.QueueTest.generatePayload;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class QueuePoolTest {
+
+    private static final Logger LOG = LoggerFactory.getLogger(QueuePoolTest.class);
+
+    @TempDir
+    Path tempQueueFolder;
+
+    @Test
+    public void checkpointFileContainsCorrectReferences() throws QueueException, IOException {
+        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder);
+        final Queue queue = queuePool.getOrCreate("test");
+        queue.enqueue((ByteBuffer)ByteBuffer.wrap("AAAA".getBytes(StandardCharsets.UTF_8)));
+        queue.force();
+        queuePool.close();
+
+        // verify
+        final Path checkpointPath = tempQueueFolder.resolve("checkpoint.properties");
+        final File checkpointFile = checkpointPath.toFile();
+        assertTrue(checkpointFile.exists(), "Checkpoint file must be created");
+
+        final Properties checkpoint = loadCheckpoint(checkpointPath);
+        final int lastPage = Integer.parseInt(checkpoint.get("segments.last_page").toString());
+        assertEquals(0, lastPage);
+        final int lastSegment = Integer.parseInt(checkpoint.get("segments.last_segment").toString());
+        assertEquals(1, lastSegment);
+
+        assertEquals("test", checkpoint.get("queues.0.name"), "Queue name must match");
+    }
+
+    private Properties loadCheckpoint(Path checkpointPath) throws IOException {
+        final FileReader fileReader;
+        fileReader = new FileReader(checkpointPath.toFile());
+        final Properties checkpointProps = new Properties();
+        checkpointProps.load(fileReader);
+        return checkpointProps;
+    }
+
+    @Test
+    public void reloadQueuePoolAndCheckRestartFromWhereItLeft() throws QueueException, IOException {
+        QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder);
+        Queue queue = queuePool.getOrCreate("test");
+        queue.enqueue(ByteBuffer.wrap("AAAA".getBytes(StandardCharsets.UTF_8)));
+        queue.force();
+        queuePool.close();
+
+        // reload
+        queuePool = QueuePool.loadQueues(tempQueueFolder);
+        queue = queuePool.getOrCreate("test");
+        queue.enqueue(ByteBuffer.wrap("BBBB".getBytes(StandardCharsets.UTF_8)));
+        queue.force();
+        queuePool.close();
+
+        // verify
+        final Path checkpointPath = tempQueueFolder.resolve("checkpoint.properties");
+        final File checkpointFile = checkpointPath.toFile();
+        assertTrue(checkpointFile.exists(), "Checkpoint file must be created");
+
+        final Properties checkpoint = loadCheckpoint(checkpointPath);
+        final int lastPage = Integer.parseInt(checkpoint.get("segments.last_page").toString());
+        assertEquals(0, lastPage);
+        final int lastSegment = Integer.parseInt(checkpoint.get("segments.last_segment").toString());
+        assertEquals(1, lastSegment);
+
+        assertEquals("test", checkpoint.get("queues.0.name"), "Queue name must match");
+        assertEquals("15", checkpoint.get("queues.0.head_offset"), "Queue head must be 16 bytes over the start");
+    }
+
+    private TreeSet<QueuePool.SegmentRef> asTreeSet(QueuePool.SegmentRef... segments) {
+        final TreeSet<QueuePool.SegmentRef> usedSegments = new TreeSet<>();
+        usedSegments.addAll(Arrays.asList(segments));
+        return usedSegments;
+    }
+
+    @Test
+    public void checkRecreateHolesAtTheStartOfThePage() throws QueueException {
+        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder);
+        final QueuePool.SegmentRef middleSegment = new QueuePool.SegmentRef(0, Segment.SIZE);
+
+        // Exercise
+        final List<QueuePool.SegmentRef> holes = queuePool.recreateSegmentHoles(asTreeSet(middleSegment));
+
+        // Verify
+        assertEquals(1, holes.size(), "Only the preceding segment should be created");
+        QueuePool.SegmentRef singleHole = holes.get(0);
+        assertEquals(0, singleHole.pageId);
+        assertEquals(0, singleHole.offset);
+    }
+
+    @Test
+    public void checkRecreateHolesAtTheStartOfThePageWith2OccupiedContiguousSegments() throws QueueException {
+        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder);
+        final QueuePool.SegmentRef firstSegment = new QueuePool.SegmentRef(0, Segment.SIZE);
+        final QueuePool.SegmentRef secondSegment = new QueuePool.SegmentRef(0, 2 * Segment.SIZE);
+
+        // Exercise
+        final List<QueuePool.SegmentRef> holes = queuePool.recreateSegmentHoles(asTreeSet(firstSegment, secondSegment));
+
+        // Verify
+        assertEquals(1, holes.size(), "Only the preceding segment should be created");
+        QueuePool.SegmentRef singleHole = holes.get(0);
+        assertEquals(0, singleHole.pageId);
+        assertEquals(0, singleHole.offset);
+    }
+
+    @Test
+    public void checkRecreateHolesBeforeSecondPage() throws QueueException {
+        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder);
+        final QueuePool.SegmentRef middleSegment = new QueuePool.SegmentRef(1, Segment.SIZE);
+
+        // Exercise
+        final List<QueuePool.SegmentRef> holes = queuePool.recreateSegmentHoles(asTreeSet(middleSegment));
+
+        // Verify
+        final int expectedHolesCount = (int) (PagedFilesAllocator.PAGE_SIZE / Segment.SIZE) + 1;
+        assertEquals(expectedHolesCount, holes.size(), "The previous empty page is full of holes");
+        for (int i = 0; i < expectedHolesCount - 1; i++) {
+            final QueuePool.SegmentRef hole = holes.get(i);
+            assertEquals(0, hole.pageId);
+            assertEquals(i * Segment.SIZE, hole.offset);
+        }
+        QueuePool.SegmentRef singleHole = holes.get(expectedHolesCount - 1);
+        assertEquals(1, singleHole.pageId);
+        assertEquals(0, singleHole.offset);
+    }
+
+    @Test
+    public void checkRecreateHolesBetweenUsedSegmentsOnSamePage() throws QueueException {
+        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder);
+        final QueuePool.SegmentRef initialSegment = new QueuePool.SegmentRef(0, 0);
+        final QueuePool.SegmentRef middleSegment = new QueuePool.SegmentRef(0, 3 * Segment.SIZE);
+
+        // Exercise
+        final List<QueuePool.SegmentRef> holes = queuePool.recreateSegmentHoles(asTreeSet(initialSegment, middleSegment));
+
+        // Verify
+        assertEquals(2, holes.size());
+
+        // first hole
+        assertEquals(0, holes.get(0).pageId);
+        assertEquals(Segment.SIZE, holes.get(0).offset);
+        // second hole
+        assertEquals(0, holes.get(1).pageId);
+        assertEquals(2 * Segment.SIZE, holes.get(1).offset);
+    }
+
+    @Test
+    public void checkRecreateHolesSpanningMultiplePages() throws QueueException {
+        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder);
+        final QueuePool.SegmentRef initialSegment = new QueuePool.SegmentRef(0, 0);
+        final QueuePool.SegmentRef middleSegment = new QueuePool.SegmentRef(2, 3 * Segment.SIZE);
+
+        // Exercise
+        final List<QueuePool.SegmentRef> holes = queuePool.recreateSegmentHoles(asTreeSet(initialSegment, middleSegment));
+
+        // Verify
+        final int holesInEmptyPage = (PagedFilesAllocator.PAGE_SIZE / Segment.SIZE);
+        final int holesInFirstPage = holesInEmptyPage - 1;
+        final int holesInLastPage = 3;
+        final int expectedHolesCount = holesInFirstPage + holesInEmptyPage + holesInLastPage;
+        assertEquals(expectedHolesCount, holes.size());
+
+        // first page hole
+        int i = 0;
+        int expectedOffset = Segment.SIZE;
+        for (; i < holesInFirstPage; i++) {
+            final QueuePool.SegmentRef hole = holes.get(i);
+            assertEquals(0, hole.pageId);
+            assertEquals(expectedOffset, hole.offset);
+            expectedOffset += Segment.SIZE;
+        }
+
+        // central empty pages
+        expectedOffset = 0;
+        for (; i < holesInFirstPage + holesInEmptyPage; i++) {
+            final QueuePool.SegmentRef hole = holes.get(i);
+            assertEquals(1, hole.pageId);
+            assertEquals(expectedOffset, hole.offset);
+            expectedOffset += Segment.SIZE;
+        }
+
+        // tail page hole
+        expectedOffset = 0;
+        for (; i < expectedHolesCount; i++) {
+            final QueuePool.SegmentRef hole = holes.get(i);
+            assertEquals(2, hole.pageId);
+            assertEquals(expectedOffset, hole.offset);
+            expectedOffset += Segment.SIZE;
+        }
+    }
+
+    @Test
+    public void checkRecreateHolesWhenSegmentAreAdjacentAndSpanningMultiplePages() throws QueueException {
+        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder);
+        final QueuePool.SegmentRef initialSegment = new QueuePool.SegmentRef(0, PagedFilesAllocator.PAGE_SIZE - Segment.SIZE);
+        final QueuePool.SegmentRef adjacentSegment = new QueuePool.SegmentRef(1, 0);
+
+        // Exercise
+        final List<QueuePool.SegmentRef> holes = queuePool.recreateSegmentHoles(asTreeSet(initialSegment, adjacentSegment));
+
+        // Verify
+        assertEquals((PagedFilesAllocator.PAGE_SIZE - Segment.SIZE) / Segment.SIZE, holes.size());
+    }
+
+    @Test
+    @Disabled
+    public void verifySingleReaderSingleWriterOnSingleQueuePool_with_955157_size_packet() throws QueueException, ExecutionException, InterruptedException, TimeoutException {
+        templateSingleReaderSingleWriterOnSingleQueuePool(955157);
+    }
+
+    @Test
+    @Disabled
+    public void verifySingleReaderSingleWriterOnSingleQueuePool_with_random_size_packet() throws QueueException, ExecutionException, InterruptedException, TimeoutException, NoSuchAlgorithmException {
+        final int payloadSize = SecureRandom.getInstanceStrong().nextInt(Segment.SIZE / 2 - LENGTH_HEADER_SIZE);
+        templateSingleReaderSingleWriterOnSingleQueuePool(payloadSize);
+
+    }
+
+    private void templateSingleReaderSingleWriterOnSingleQueuePool(int payloadSize) throws QueueException, InterruptedException, ExecutionException, TimeoutException {
+        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder);
+        Queue queue = queuePool.getOrCreate("single_writer_single_reader");
+        ExecutorService pool = Executors.newCachedThreadPool();
+        int messagesToSend = 1000;
+
+        LOG.info("Payload size: " + payloadSize);
+        ByteBuffer payload = ByteBuffer.wrap(generatePayload(payloadSize, (byte) 'a'));
+
+        Future<?> senderFuture = pool.submit(createMessageSender(queue, payload, messagesToSend));
+        Future<Integer> readerFuture = pool.submit(createMessageReader(queue, messagesToSend));
+
+        senderFuture.get(60, TimeUnit.SECONDS);
+        final int readBytes = readerFuture.get(60, TimeUnit.SECONDS);
+        assertEquals(messagesToSend * payloadSize, readBytes);
+    }
+
+    private Runnable createMessageSender(Queue queue, ByteBuffer payload, int count) {
+        return () -> {
+            int payloadSize = payload.remaining();
+            int sentBytes = 0;
+            for (int i = 0; i < count; i++) {
+                try {
+                    ByteBuffer duplicate = payload.duplicate();
+                    sentBytes += duplicate.remaining();
+                    queue.enqueue(duplicate);
+                } catch (QueueException e) {
+                    throw new RuntimeException(e);
+                }
+            }
+            LOG.debug("Finish to send data, " + count + " messages of size: " + payloadSize + " for a total of: " + sentBytes);
+        };
+    }
+
+    private Callable<Integer> createMessageReader(Queue queue, int expectedMessages) {
+        return () -> {
+            try {
+                int readBytes = 0;
+                int receivedMessages = 0;
+                do {
+                    LOG.debug("receivedMessages {} ({} expected)", receivedMessages, expectedMessages);
+                    Optional<ByteBuffer> readPayload = queue.dequeue();
+                    if (readPayload.isPresent()) {
+                        readBytes += readPayload.get().remaining();
+                        receivedMessages++;
+                    }
+                } while (receivedMessages < expectedMessages);
+                LOG.debug("Received messages: " + receivedMessages);
+                return readBytes;
+            } catch (QueueException e) {
+                throw new RuntimeException(e);
+            }
+        };
+    }
+
+    int result = 0;
+
+    @Test
+    @Disabled
+    public void testMultipleWritersSingleReader() throws QueueException, NoSuchAlgorithmException, ExecutionException, InterruptedException, TimeoutException {
+        final int payloadSize = 152433/*SecureRandom.getInstanceStrong().nextInt(Segment.SIZE / 2 - LENGTH_HEADER_SIZE)*/;
+        LOG.info("Payload size: " + payloadSize);
+        final QueuePool queuePool = QueuePool.loadQueues(/*tempQueueFolder*/Paths.get("/tmp/test_dir/"));
+        Queue queue = queuePool.getOrCreate("multiple_writers_single_reader");
+        ExecutorService pool = Executors.newCachedThreadPool();
+        int messagesToSend = 1000;
+
+        ByteBuffer payloadA = ByteBuffer.wrap(generatePayload(payloadSize, (byte) 'a'));
+        ByteBuffer payloadB = ByteBuffer.wrap(generatePayload(payloadSize, (byte) 'b'));
+
+        final Thread threadWriterA = new Thread(createMessageSender(queue, payloadA, messagesToSend / 2));
+        threadWriterA.setName("Writer-A");
+
+        final Thread threadWriterB = new Thread(createMessageSender(queue, payloadB, messagesToSend / 2));
+        threadWriterB.setName("Writer-B");
+
+        final Thread threadReader = new Thread(new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    result = createMessageReader(queue, messagesToSend).call();
+                } catch (Exception e) {
+                    throw new RuntimeException(e);
+                }
+            }
+        });
+        threadReader.setName("Reader");
+
+        threadWriterA.start();
+        threadWriterB.start();
+        threadReader.start();
+        threadWriterA.join(60_000);
+        threadWriterB.join(60_000);
+        threadReader.join(60_000);
+
+
+//        Future<?> senderFutureA = pool.submit(createMessageSender(queue, payloadA, messagesToSend / 2));
+//        Future<?> senderFutureB = pool.submit(createMessageSender(queue, payloadB, messagesToSend / 2));
+//        Future<Integer> readerFuture = pool.submit(createMessageReader(queue, messagesToSend));
+
+//        senderFutureA.get(60, TimeUnit.SECONDS);
+//        senderFutureB.get(60, TimeUnit.SECONDS);
+//        final int readBytes = readerFuture.get(60, TimeUnit.SECONDS);
+        assertEquals(messagesToSend * payloadSize, result);
+    }
+}
diff --git a/broker/src/test/java/io/moquette/broker/unsafequeues/QueueTest.java b/broker/src/test/java/io/moquette/broker/unsafequeues/QueueTest.java
new file mode 100644
index 00000000..8773dbf0
--- /dev/null
+++ b/broker/src/test/java/io/moquette/broker/unsafequeues/QueueTest.java
@@ -0,0 +1,563 @@
+package io.moquette.broker.unsafequeues;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.io.FileReader;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.MappedByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.OpenOption;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.*;
+import java.util.function.Consumer;
+
+import static io.moquette.broker.unsafequeues.Queue.LENGTH_HEADER_SIZE;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+class QueueTest {
+
+    @TempDir
+    Path tempQueueFolder;
+
+    private void assertContainsOnly(char expectedChar, byte[] verify) {
+        for (int i = 0; i < verify.length; i++) {
+            if (verify[i] != expectedChar) {
+                fail(String.format("Expected %c but found %c in %c%c%c", expectedChar, verify[i], verify[i-1], verify[i], verify[i+1]));
+            }
+        }
+    }
+
+    private void assertContainsOnly(char expectedChar, ByteBuffer verify) {
+        int pos = verify.position();
+        while (verify.hasRemaining()) {
+            final byte readChar = verify.get();
+            pos++;
+            if (readChar != expectedChar) {
+                fail(String.format("Expected %c but found %c at position %d", expectedChar, readChar, pos));
+            }
+        }
+    }
+
+    private void assertContainsOnly(char expectedChar, ByteBuffer verify, int expectedSize) {
+        int pos = verify.position();
+        int countChars = 0;
+        while (verify.hasRemaining()) {
+            final byte readChar = verify.get();
+            pos++;
+            countChars++;
+            if (readChar != expectedChar) {
+                fail(String.format("Expected %c but found %c at position %d", expectedChar, readChar, pos));
+            }
+        }
+        assertEquals(expectedSize, countChars);
+    }
+
+
+    static byte[] generatePayload(int numBytes) {
+        return generatePayload(numBytes, (byte) 'A');
+    }
+
+    static byte[] generatePayload(int numBytes, byte c) {
+        final byte[] payload = new byte[numBytes];
+        for (int i = 0; i < numBytes; i++) {
+            payload[i] = c;
+        }
+        return payload;
+    }
+
+    @Test
+    public void basicNoBlockEnqueue() throws QueueException, IOException {
+        final MappedByteBuffer pageBuffer = Utils.createPageFile();
+
+        final Segment head = new Segment(pageBuffer, new SegmentPointer(0,0), new SegmentPointer(0, 1024));
+        final VirtualPointer currentHead = VirtualPointer.buildUntouched();
+        final Queue queue = new Queue("test", head, currentHead, head, currentHead, new DummySegmentAllocator(), (name, segment) -> {
+            // NOOP
+        }, null);
+
+        // generate byte array to insert.
+        ByteBuffer payload = randomPayload(128);
+
+        queue.enqueue(payload);
+    }
+
+    private ByteBuffer randomPayload(int dataSize) {
+        byte[] payload = new byte[dataSize];
+        new Random().nextBytes(payload);
+
+        return (ByteBuffer) ByteBuffer.wrap(payload);
+    }
+
+    @Test
+    public void newlyQueueIsEmpty() throws QueueException {
+        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder);
+        final Queue queue = queuePool.getOrCreate("test");
+
+        assertTrue(queue.isEmpty(), "Freshly created queue must be empty");
+    }
+
+    @Test
+    public void consumedQueueIsEmpty() throws QueueException {
+        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder);
+        final Queue queue = queuePool.getOrCreate("test");
+        queue.enqueue(ByteBuffer.wrap("AAAA".getBytes(StandardCharsets.UTF_8)));
+        Optional<ByteBuffer> data = queue.dequeue();
+        assertTrue(data.isPresent(), "Some payload is retrieved");
+
+        assertEquals(4, data.get().remaining(), "Payload contains what's expected");
+
+        assertTrue(queue.isEmpty(), "Queue must be empty after consuming it");
+    }
+
+    @Test
+    public void insertSomeDataIntoNewQueue() throws QueueException, IOException {
+        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder);
+        final Queue queue = queuePool.getOrCreate("test");
+        queue.enqueue(ByteBuffer.wrap("AAAA".getBytes(StandardCharsets.UTF_8)));
+
+        // verify
+        final HashSet<String> fileset = new HashSet<>(Arrays.asList(tempQueueFolder.toFile().list()));
+        assertEquals(2, fileset.size());
+        assertTrue(fileset.contains("checkpoint.properties"), "Checkpoint file must be created");
+        assertTrue(fileset.contains("0.page"), "One page file must be created");
+
+        final Path pageFile = tempQueueFolder.resolve("0.page");
+        verifyFile(pageFile, 9, rawContent -> {
+            assertEquals(4, rawContent.getInt(), "First 4 bytes contains the length");
+            assertEquals('A', rawContent.get());
+            assertEquals('A', rawContent.get());
+            assertEquals('A', rawContent.get());
+            assertEquals('A', rawContent.get());
+            assertEquals(0, rawContent.get());
+        });
+    }
+
+    private void verifyFile(Path file, int bytesToRead, Consumer<ByteBuffer> verifier) throws IOException {
+        final FileChannel fc = FileChannel.open(file, StandardOpenOption.READ);
+        final ByteBuffer rawContent = ByteBuffer.allocate(bytesToRead);
+        final int read = fc.read(rawContent);
+        assertEquals(bytesToRead, read);
+        rawContent.flip();
+
+        verifier.accept(rawContent);
+    }
+
+    @Test
+    public void insertDataTriggerCreationOfNewPageFile() throws QueueException {
+        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder);
+        final Queue queue = queuePool.getOrCreate("test");
+
+        // one page is 64 MB so the loop count to fill it is 64 * 1024
+
+       // 4 bytes are left for length so that each time are inserted 1024 bytes, 4 header and 1020 payload
+        ByteBuffer payload = ByteBuffer.wrap(generatePayload(1024 - LENGTH_HEADER_SIZE));
+        for (int i = 0; i < 64; i++) {
+            writeMessages(queue, payload, 1024);
+        }
+
+        // check the 2 files are created
+        HashSet<String> fileset = new HashSet<>(Arrays.asList(tempQueueFolder.toFile().list()));
+        assertEquals(2, fileset.size());
+        assertTrue(fileset.contains("checkpoint.properties"), "Checkpoint file must be created");
+        assertTrue(fileset.contains("0.page"),
+             "One page file must be created");
+
+        // Exercise
+        // some data to force create a new page
+        final ByteBuffer crossingPayload = ByteBuffer.wrap(generatePayload(10, (byte) 'B'));
+        queue.enqueue(crossingPayload);
+
+        // Verify
+        fileset = new HashSet<>(Arrays.asList(tempQueueFolder.toFile().list()));
+        assertEquals(3, fileset.size());
+        assertTrue(fileset.contains("checkpoint.properties"), "Checkpoint file must be created");
+        assertTrue(fileset.contains("0.page"), "First page file must be created");
+        assertTrue(fileset.contains("1.page"), "Second page file must be created");
+    }
+
+    @Test
+    public void insertWithAnHeaderThatCrossSegments() throws QueueException, IOException {
+        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder);
+        final Queue queue = queuePool.getOrCreate("test");
+
+        // fill the segment, inserting last message crossing the boundary
+        ByteBuffer payload = ByteBuffer.wrap(generatePayload(1024 - LENGTH_HEADER_SIZE));
+        writeMessages(queue, payload, (4 * 1024) - 1);
+        // at the end we have 1024 bytes free, so fill only 1022 bytes of that
+        payload = ByteBuffer.wrap(generatePayload(1022 - LENGTH_HEADER_SIZE));
+        payload.rewind();
+        queue.enqueue(payload);
+
+        // Exercise
+        ByteBuffer crossingPayload = ByteBuffer.wrap(generatePayload(1024 - LENGTH_HEADER_SIZE, (byte) 'B'));
+        queue.enqueue(crossingPayload);
+
+        // Verify
+        final MappedByteBuffer page = Utils.openPageFile(tempQueueFolder.resolve("0.page"));
+        final int beforeLastMessagePayload = 4 * 1024 * 1024 + 2;
+        final ByteBuffer crossingSegment = (ByteBuffer) page.position(beforeLastMessagePayload);
+//        final int msgLength = crossingSegment.getInt();
+
+//        assertEquals(1028 - LENGTH_HEADER_SIZE, msgLength);
+//        byte[] probe = new byte[msgLength];
+        byte[] probe = new byte[1020];
+        crossingSegment.get(probe);
+        assertContainsOnly('B', probe);
+    }
+
+    @Test
+    public void insertDataCrossingSegmentBoundary() throws QueueException, IOException {
+        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder);
+        final Queue queue = queuePool.getOrCreate("test");
+
+        // one segment is 4MB 4 * 1024 * payload
+        // so send (4 * 1024) - 1 payloads of 1024 and then send
+        // a payload of 1028 (4 bytes over remaining space)
+
+        // 4 bytes are left for length so that each time are inserted 1024 bytes, 4 header and 1020 payload
+        ByteBuffer payload = ByteBuffer.wrap(generatePayload(1024 - LENGTH_HEADER_SIZE));
+        writeMessages(queue, payload, (4 * 1024) - 1);
+
+        // Experiment
+        ByteBuffer crossingPayload = ByteBuffer.wrap(generatePayload(1028 - LENGTH_HEADER_SIZE, (byte) 'B'));
+        queue.enqueue(crossingPayload);
+        queue.force();
+        queuePool.close();
+
+        // Verify
+        final MappedByteBuffer page = Utils.openPageFile(tempQueueFolder.resolve("0.page"));
+        final int beforeLastMessage = ((4 * 1024) - 1) * 1024;
+        final ByteBuffer crossingSegment = (ByteBuffer) page.position(beforeLastMessage);
+        final int msgLength = crossingSegment.getInt();
+
+        assertEquals(1028 - LENGTH_HEADER_SIZE, msgLength);
+        byte[] probe = new byte[msgLength];
+        crossingSegment.get(probe);
+        assertContainsOnly('B', probe);
+    }
+
+    @Test
+    public void insertDataBiggerThanASegment() throws QueueException, IOException {
+        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder);
+        final Queue queue = queuePool.getOrCreate("test");
+
+        // one segment is 4MB 4 * 1024 * payload
+        // so send (4 * 1024) - 1 payloads of 1024 and then send
+        // a payload of 1028 (4 bytes over remaining space)
+
+        // 4 bytes are left for length so that each time are inserted 1024 bytes, 4 header and 1020 payload
+        ByteBuffer payload = ByteBuffer.wrap(generatePayload(1024 - LENGTH_HEADER_SIZE));
+        writeMessages(queue, payload, (4 * 1024) - 1);
+
+        // Experiment
+        // 1024 + 4 * 1024 * 1024 + 16 bytes
+        int moreThanOneSegment = 1024 + 4*1024*1024 + 16;
+        ByteBuffer crossingMultipleSegmentPayload = ByteBuffer.wrap(generatePayload(moreThanOneSegment, (byte) 'B'));
+        queue.enqueue(crossingMultipleSegmentPayload);
+        queue.force();
+        queuePool.close();
+
+        // Verify
+        final MappedByteBuffer page = Utils.openPageFile(tempQueueFolder.resolve("0.page"));
+        final int beforeLastMessage = ((4 * 1024) - 1) * 1024;
+        final ByteBuffer crossingSegment = (ByteBuffer) page.position(beforeLastMessage);
+        final int msgLength = crossingSegment.getInt();
+
+        assertEquals(moreThanOneSegment, msgLength);
+        byte[] probe = new byte[msgLength];
+        crossingSegment.get(probe);
+        assertContainsOnly('B', probe);
+    }
+
+    @Test
+    public void readFromEmptyQueue() throws QueueException {
+        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder);
+        final Queue queue = queuePool.getOrCreate("test");
+
+        assertFalse(queue.dequeue().isPresent(), "Pulling from empty queue MUST return null value");
+    }
+
+    @Test
+    public void readInSameSegment() throws QueueException {
+        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder);
+        final Queue queue = queuePool.getOrCreate("test");
+
+        final ByteBuffer message = ByteBuffer.wrap("Hello World!".getBytes(StandardCharsets.UTF_8));
+        queue.enqueue(message);
+
+        //Exercise
+        final ByteBuffer result = queue.dequeue().get();
+        final String readMessage = Utils.bufferToString(result);
+        assertEquals("Hello World!", readMessage, "Read the same message tha was enqueued");
+    }
+
+    @Test
+    public void readCrossingSegment() throws QueueException {
+        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder);
+        final Queue queue = queuePool.getOrCreate("test");
+
+        // fill the segment, inserting last message crossing the boundary
+        ByteBuffer payload = ByteBuffer.wrap(generatePayload(1024 - LENGTH_HEADER_SIZE));
+        for (int i = 0; i < (4 * 1024) - 1; i++) {
+            payload.rewind();
+            queue.enqueue(payload);
+            queue.dequeue();
+        }
+
+        ByteBuffer crossingPayload = ByteBuffer.wrap(generatePayload(1028 - LENGTH_HEADER_SIZE, (byte) 'B'));
+        queue.enqueue(crossingPayload);
+
+        //Exercise
+        final ByteBuffer message = queue.dequeue().get();
+        assertEquals(1028 - LENGTH_HEADER_SIZE, message.remaining(), "There must be 1024 'B' letters");
+        assertContainsOnly('B', message);
+    }
+
+    @Test
+    public void readWithHeaderCrossingSegments() throws QueueException {
+        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder);
+        final Queue queue = queuePool.getOrCreate("test");
+
+        // fill the segment, inserting last message crossing the boundary
+        ByteBuffer payload = ByteBuffer.wrap(generatePayload(1024 - LENGTH_HEADER_SIZE));
+        for (int i = 0; i < (4 * 1024) - 1; i++) {
+            payload.rewind();
+            queue.enqueue(payload);
+            queue.dequeue();
+        }
+        // at the end we have 1024 bytes free, so fill only 1022 bytes of that
+        payload = ByteBuffer.wrap(generatePayload(1022 - LENGTH_HEADER_SIZE));
+        payload.rewind();
+        queue.enqueue(payload);
+        queue.dequeue();
+
+        // write a payload's header with 2 bytes in previous and 2 in next segment
+        ByteBuffer crossingPayload = ByteBuffer.wrap(generatePayload(1024 - LENGTH_HEADER_SIZE, (byte) 'B'));
+        queue.enqueue(crossingPayload);
+
+        //Exercise
+        final ByteBuffer message = queue.dequeue().get();
+        assertEquals(1024 - LENGTH_HEADER_SIZE, message.remaining(), "There must be 1020 'B' letters");
+        assertContainsOnly('B', message);
+    }
+
+    @Test
+    public void readCrossingPages() throws QueueException, IOException {
+        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder);
+        final Queue queue = queuePool.getOrCreate("test");
+
+        // fill all segments less one in a page
+        ByteBuffer payload = ByteBuffer.wrap(generatePayload(1024 - LENGTH_HEADER_SIZE));
+        int messageSize = payload.remaining() + LENGTH_HEADER_SIZE;
+        final int loopToFill = PagedFilesAllocator.PAGE_SIZE / messageSize;
+        writeMessages(queue, payload, loopToFill - 1);
+
+        assertEquals(1, countPages(tempQueueFolder));
+        assertEquals(PagedFilesAllocator.PAGE_SIZE - messageSize, queue.currentHead().logicalOffset() + 1,
+            "head must be one message size (1024) from the end of the segment");
+
+        // Exercise
+        payload = ByteBuffer.wrap(generatePayload(2048 - LENGTH_HEADER_SIZE, (byte) 'B'));
+        queue.enqueue(payload);
+
+        // Verify
+        assertEquals(2, countPages(tempQueueFolder));
+    }
+
+    private long countPages(Path tempQueueFolder) throws IOException {
+        return Files.list(tempQueueFolder).filter(p -> p.toString().endsWith(".page")).count();
+    }
+
+    @Test
+    public void interleavedQueueSegments() throws QueueException, IOException {
+        // first segment queue A, second segment queue B and so on, in stripped fashion.
+        // writes and read pass single page borders, checking everything is fine
+        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder);
+        final Queue queueA = queuePool.getOrCreate("testA");
+        final Queue queueB = queuePool.getOrCreate("testB");
+
+        ByteBuffer payloadQueueA = ByteBuffer.wrap(generatePayload(1024 - LENGTH_HEADER_SIZE, (byte)'A'));
+        ByteBuffer payloadQueueB = ByteBuffer.wrap(generatePayload(1024 - LENGTH_HEADER_SIZE, (byte)'B'));
+        int messageSize = payloadQueueA.remaining() + LENGTH_HEADER_SIZE;
+
+        // Exercise
+        final int numPages = 2;
+        final int segmentsToFill = numPages * PagedFilesAllocator.PAGE_SIZE / Segment.SIZE;
+        final int messagesInSegment = Segment.SIZE / messageSize;
+        for (int i = 0; i < segmentsToFill; i++) {
+            if (isEven(i)) {
+                writeMessages(queueA, payloadQueueA, messagesInSegment);
+            } else {
+                writeMessages(queueB, payloadQueueB, messagesInSegment);
+            }
+        }
+
+        // Verify
+        assertEquals(numPages, countPages(tempQueueFolder));
+        final int numMessagesInQueue = PagedFilesAllocator.PAGE_SIZE / messageSize;
+        verifyReadingFromQueue(numMessagesInQueue, queueA, 'A', 1024 - LENGTH_HEADER_SIZE);
+        verifyReadingFromQueue(numMessagesInQueue, queueB, 'B', 1024 - LENGTH_HEADER_SIZE);
+    }
+
+    private void verifyReadingFromQueue(int numMessagesInQueue, Queue queue, char ch, int expectedPayloadSize) throws QueueException {
+        for (int i = 0; i < numMessagesInQueue; i++) {
+            final ByteBuffer payload = queue.dequeue().get();
+            assertContainsOnly(ch, payload, expectedPayloadSize);
+        }
+    }
+
+    private void writeMessages(Queue targetQueue, ByteBuffer payload, int messagesToWrite) throws QueueException {
+        for (int i = 0; i < messagesToWrite; i++) {
+            payload.rewind();
+            targetQueue.enqueue(payload);
+        }
+    }
+
+    private boolean isEven(int i) {
+        return i % 2 == 0;
+    }
+
+    @Test
+    public void physicalBackwardSegment() throws IOException, QueueException {
+        // Artificially create a queue composed of segment(2) and segment(1), inverted in order, verify
+        // if read and write properly.pageBuffer.
+        final Path pageFile = this.tempQueueFolder.resolve("0.page");
+        final OpenOption[] openOptions = {StandardOpenOption.CREATE_NEW, StandardOpenOption.READ, StandardOpenOption.WRITE, StandardOpenOption.TRUNCATE_EXISTING};
+        FileChannel fileChannel = FileChannel.open(pageFile, openOptions);
+        final MappedByteBuffer pageBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, PagedFilesAllocator.PAGE_SIZE);
+        // segment with only one message of B letters
+        pageBuffer.putInt(Segment.SIZE - LENGTH_HEADER_SIZE);
+        pageBuffer.put(generatePayload(Segment.SIZE - LENGTH_HEADER_SIZE, (byte)'B'));
+        // segment with only one message of A letters
+        pageBuffer.putInt(Segment.SIZE - LENGTH_HEADER_SIZE);
+        pageBuffer.put(generatePayload(Segment.SIZE - LENGTH_HEADER_SIZE, (byte)'A'));
+        pageBuffer.force();
+        fileChannel.close();
+
+        // create the checkpoint file with the inverted segments
+        Properties checkpoint = new Properties();
+//        checkpoint.properties
+        checkpoint.put("queues.0.name", "test_inverted");
+        checkpoint.put("queues.0.segments", "(0, 0), (0, " + Segment.SIZE + ")");
+        checkpoint.put("queues.0.head_offset", Integer.toString(Segment.SIZE - 1));
+        checkpoint.put("queues.0.tail_offset", Integer.toString(-1));
+        checkpoint.put("segments.last_page", Integer.toString(0));
+        checkpoint.put("segments.last_segment", Integer.toString(2));
+        File propsFile = this.tempQueueFolder.resolve("checkpoint.properties").toFile();
+        final FileWriter propsWriter = new FileWriter(propsFile);
+        checkpoint.store(propsWriter, "test checkpoint file to verify loading of inverted segments");
+
+        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder);
+        final Queue queue = queuePool.getOrCreate("test_inverted");
+
+        assertContainsOnly('A', queue.dequeue().get(), Segment.SIZE - LENGTH_HEADER_SIZE);
+        assertContainsOnly('B', queue.dequeue().get(), Segment.SIZE - LENGTH_HEADER_SIZE);
+    }
+
+    @Test
+    public void reopenQueueWithSomeDataInto() throws QueueException {
+        // given a queue wth some data split across multiple segments
+        final QueuePool queuePoolA = QueuePool.loadQueues(tempQueueFolder);
+        final Queue queueA = queuePoolA.getOrCreate("testA");
+        queueA.enqueue(ByteBuffer.wrap(generatePayload(Segment.SIZE / 2 - LENGTH_HEADER_SIZE, (byte)'a')));
+        queueA.enqueue(ByteBuffer.wrap(generatePayload(Segment.SIZE / 2 - LENGTH_HEADER_SIZE, (byte)'A')));
+        queueA.enqueue(ByteBuffer.wrap(generatePayload(Segment.SIZE / 2 - LENGTH_HEADER_SIZE, (byte)'b')));
+        queueA.enqueue(ByteBuffer.wrap(generatePayload(Segment.SIZE / 2 - LENGTH_HEADER_SIZE, (byte)'B')));
+        // when it's closed and reopened
+        queueA.force();
+        queuePoolA.close();
+
+        // then the consumption must happen in the same order
+        final Queue reopened = queuePoolA.getOrCreate("testA");
+        assertContainsOnly('a', reopened.dequeue().get(), Segment.SIZE / 2 - LENGTH_HEADER_SIZE);
+        assertContainsOnly('A', reopened.dequeue().get(), Segment.SIZE / 2 - LENGTH_HEADER_SIZE);
+        assertContainsOnly('b', reopened.dequeue().get(), Segment.SIZE / 2 - LENGTH_HEADER_SIZE);
+        assertContainsOnly('B', reopened.dequeue().get(), Segment.SIZE / 2 - LENGTH_HEADER_SIZE);
+    }
+
+    @Test
+    public void writeTestSuiteToVerifyPagedFilesAllocatorDoesntCreateExternalFragmentation() throws QueueException, IOException {
+        // write 2 segments, consume one segment, next segment allocated should be one just freed.0
+        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder);
+        final Queue queue = queuePool.getOrCreate("test_external_fragmentation");
+
+        // fill first segment (0, 0)
+        queue.enqueue(ByteBuffer.wrap(generatePayload(Segment.SIZE / 2 - LENGTH_HEADER_SIZE, (byte)'a')));
+        queue.enqueue(ByteBuffer.wrap(generatePayload(Segment.SIZE / 2 - LENGTH_HEADER_SIZE, (byte)'A')));
+
+        // fill second segment (0, 4194304)
+        queue.enqueue(ByteBuffer.wrap(generatePayload(Segment.SIZE / 2 - LENGTH_HEADER_SIZE, (byte)'b')));
+        queue.enqueue(ByteBuffer.wrap(generatePayload(Segment.SIZE / 2 - LENGTH_HEADER_SIZE, (byte)'B')));
+
+        // consume first segment
+        assertContainsOnly('a', queue.dequeue().get(), Segment.SIZE / 2 - LENGTH_HEADER_SIZE);
+        assertContainsOnly('A', queue.dequeue().get(), Segment.SIZE / 2 - LENGTH_HEADER_SIZE);
+
+        // Exercise
+        // write new data, should go in first freed segment
+        queue.enqueue(ByteBuffer.wrap(generatePayload(Segment.SIZE / 2 - LENGTH_HEADER_SIZE, (byte)'c')));
+        queuePool.close();
+
+        // Verify
+        // checkpoint contains che correct order, (0,0), (0, 4194304)
+        final Properties checkpointProps = loadCheckpointFile(tempQueueFolder);
+
+        final String segmentRefs = checkpointProps.getProperty("queues.0.segments");
+        assertEquals("(0, 0), (0, 4194304)", segmentRefs);
+    }
+
+    @Test
+    public void reopenQueueWithFragmentation() throws QueueException, IOException {
+        // write 2 segments, consume one segment, next segment allocated should be one just freed.0
+        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder);
+        final Queue queue = queuePool.getOrCreate("test_external_fragmentation");
+
+        // fill first segment (0, 0)
+        queue.enqueue(ByteBuffer.wrap(generatePayload(Segment.SIZE / 2 - LENGTH_HEADER_SIZE, (byte)'a')));
+        queue.enqueue(ByteBuffer.wrap(generatePayload(Segment.SIZE / 2 - LENGTH_HEADER_SIZE, (byte)'A')));
+
+        // fill second segment (0, 4194304)
+        queue.enqueue(ByteBuffer.wrap(generatePayload(Segment.SIZE / 2 - LENGTH_HEADER_SIZE, (byte)'b')));
+        queue.enqueue(ByteBuffer.wrap(generatePayload(Segment.SIZE / 2 - LENGTH_HEADER_SIZE, (byte)'B')));
+
+        // consume first segment
+        assertContainsOnly('a', queue.dequeue().get(), Segment.SIZE / 2 - LENGTH_HEADER_SIZE);
+        assertContainsOnly('A', queue.dequeue().get(), Segment.SIZE / 2 - LENGTH_HEADER_SIZE);
+
+        queue.force();
+        queuePool.close();
+
+        // Exercise
+        // reopen the queue
+        final QueuePool recreatedQueuePool = QueuePool.loadQueues(tempQueueFolder);
+        final Queue reopened = recreatedQueuePool.getOrCreate("test_external_fragmentation");
+        // write new data, should go in first freed segment
+        reopened.enqueue(ByteBuffer.wrap(generatePayload(Segment.SIZE / 2 - LENGTH_HEADER_SIZE, (byte)'c')));
+        recreatedQueuePool.close();
+
+        // Verify
+        // checkpoint contains che correct order, (0,0), (0, 4194304)
+        final Properties checkpointProps = loadCheckpointFile(tempQueueFolder);
+
+        final String segmentRefs = checkpointProps.getProperty("queues.0.segments");
+        assertEquals("(0, 0), (0, 4194304)", segmentRefs);
+    }
+
+    private Properties loadCheckpointFile(Path dir) throws IOException {
+        final Path checkpointPath = dir.resolve("checkpoint.properties");
+        final FileReader fileReader = new FileReader(checkpointPath.toFile());
+        final Properties checkpointProps = new Properties();
+        checkpointProps.load(fileReader);
+        return checkpointProps;
+    }
+}
diff --git a/broker/src/test/java/io/moquette/broker/unsafequeues/SegmentPointerTest.java b/broker/src/test/java/io/moquette/broker/unsafequeues/SegmentPointerTest.java
new file mode 100644
index 00000000..f5afdbe3
--- /dev/null
+++ b/broker/src/test/java/io/moquette/broker/unsafequeues/SegmentPointerTest.java
@@ -0,0 +1,30 @@
+package io.moquette.broker.unsafequeues;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class SegmentPointerTest {
+
+    @Test
+    public void testCompareInSameSegment() {
+        final SegmentPointer minor = new SegmentPointer(1, 10);
+        final SegmentPointer otherMinor = new SegmentPointer(1, 10);
+        final SegmentPointer major = new SegmentPointer(1, 12);
+
+        assertEquals(-1, minor.compareTo(major), "minor is less than major");
+        assertEquals(1, major.compareTo(minor), "major is greater than minor");
+        assertEquals(0, minor.compareTo(otherMinor), "minor equals itself");
+    }
+
+    @Test
+    public void testCompareInDifferentSegments() {
+        final SegmentPointer minor = new SegmentPointer(1, 10);
+        final SegmentPointer otherMinor = new SegmentPointer(1, 10);
+        final SegmentPointer major = new SegmentPointer(2, 4);
+
+        assertEquals(-1, minor.compareTo(major), "minor is less than major");
+        assertEquals(1, major.compareTo(minor), "major is greater than minor");
+        assertEquals(0, minor.compareTo(otherMinor), "minor equals itself");
+    }
+}
diff --git a/broker/src/test/java/io/moquette/broker/unsafequeues/SegmentTest.java b/broker/src/test/java/io/moquette/broker/unsafequeues/SegmentTest.java
new file mode 100644
index 00000000..4113010b
--- /dev/null
+++ b/broker/src/test/java/io/moquette/broker/unsafequeues/SegmentTest.java
@@ -0,0 +1,39 @@
+package io.moquette.broker.unsafequeues;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.nio.MappedByteBuffer;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class SegmentTest {
+
+    @Test
+    public void testHasSpace() throws IOException {
+        final MappedByteBuffer pageBuffer = Utils.createPageFile();
+
+        final Segment segment = new Segment(pageBuffer, new SegmentPointer(0, 0), new SegmentPointer(0, 1023));
+
+        final VirtualPointer current = new VirtualPointer(511);
+        final VirtualPointer otherCurrent = current.moveForward(pageBuffer.capacity()); // pointer in next page
+
+        assertTrue(segment.hasSpace(current, 512));
+        assertFalse(segment.hasSpace(current, 513));
+        assertFalse(segment.hasSpace(otherCurrent, 513));
+    }
+
+    @Test
+    public void testBytesAfter() throws IOException {
+        final MappedByteBuffer pageBuffer = Utils.createPageFile();
+
+        final SegmentPointer begin = new SegmentPointer(0, 0);
+        final SegmentPointer end = new SegmentPointer(0, 1023);
+        final Segment segment = new Segment(pageBuffer, begin, end);
+
+        assertEquals(0, segment.bytesAfter(end));
+        assertEquals(1023, segment.bytesAfter(begin));
+    }
+}
diff --git a/broker/src/test/java/io/moquette/broker/unsafequeues/Utils.java b/broker/src/test/java/io/moquette/broker/unsafequeues/Utils.java
new file mode 100644
index 00000000..6890a19d
--- /dev/null
+++ b/broker/src/test/java/io/moquette/broker/unsafequeues/Utils.java
@@ -0,0 +1,37 @@
+package io.moquette.broker.unsafequeues;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.MappedByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.file.OpenOption;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+
+class Utils {
+
+    static MappedByteBuffer createPageFile() throws IOException {
+        return createPageFile(1024);
+    }
+
+    static MappedByteBuffer createPageFile(int size) throws IOException {
+        final Path pageFile = File.createTempFile("test_queue", ".page").toPath();
+        final OpenOption[] openOptions = {StandardOpenOption.READ, StandardOpenOption.WRITE, StandardOpenOption.TRUNCATE_EXISTING};
+        FileChannel fileChannel = FileChannel.open(pageFile, openOptions);
+        return fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, size);
+    }
+
+    static MappedByteBuffer openPageFile(Path pageFile) throws IOException {
+        final OpenOption[] openOptions = {StandardOpenOption.READ, StandardOpenOption.TRUNCATE_EXISTING};
+        FileChannel fileChannel = FileChannel.open(pageFile, openOptions);
+        return fileChannel.map(FileChannel.MapMode.READ_ONLY, 0, PagedFilesAllocator.PAGE_SIZE);
+    }
+
+    static String bufferToString(ByteBuffer buffer) {
+        byte[] data = new byte[buffer.remaining()];
+        buffer.get(data);
+
+        return new String(data);
+    }
+}
diff --git a/broker/src/test/java/io/moquette/broker/unsafequeues/VirtualPointerTest.java b/broker/src/test/java/io/moquette/broker/unsafequeues/VirtualPointerTest.java
new file mode 100644
index 00000000..80babada
--- /dev/null
+++ b/broker/src/test/java/io/moquette/broker/unsafequeues/VirtualPointerTest.java
@@ -0,0 +1,18 @@
+package io.moquette.broker.unsafequeues;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class VirtualPointerTest {
+
+    @Test
+    public void testCompare() {
+        final VirtualPointer lower = new VirtualPointer(100L);
+        final VirtualPointer higher = new VirtualPointer(200L);
+
+        assertEquals(1, higher.compareTo(lower), higher.logicalOffset + " must be greater than " + lower.logicalOffset);
+        assertEquals(-1, lower.compareTo(higher), lower.logicalOffset + " must be lower than " + higher.logicalOffset);
+        assertEquals(0, lower.compareTo(lower), "identity must be equal");
+    }
+}