Persisted unsafe queues implementation #691
Conversation
|
This work looks really interesting. How far do you think you are till the first integration with Moquette? |
|
I think we are close, just want to change the API of |
|
Then integrate into Moquette, with feature flag to enable it and create a new release. |
| * */ | ||
| public Optional<ByteBuffer> dequeue() throws QueueException { | ||
| if (!currentHeadPtr.isGreaterThan(currentTailPtr)) { | ||
| if (currentTailPtr.compareTo(currentHeadPtr) > 0) { |
There was a problem hiding this comment.
For symmetry, could use currentTailPtr.isGreaterThan(currentHeadPtr) instead of a direct compare.
| 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); |
There was a problem hiding this comment.
Would it be significantly more efficient to make the segment read methods accept an existing Buffer to put the data into, instead of having those read methods create new buffers that are then copied into the already existing buffer?
|
|
||
| private final QueuePool queuePool; | ||
| private final PagedFilesAllocator.AllocationListener allocationListener; | ||
| private final ReentrantLock lock = new ReentrantLock(); |
There was a problem hiding this comment.
If the queue is explicitly not thread-safe, then what is the function of this lock? It's only used in the (de)queue methods, when reading cross-segment data. The comments call it a global lock, but this seems to be a queue-local lock... I would expect a global lock to be instantiated in the QueuePool. But there it should be enough to shortly lock the QueuePool in the nextFreeSegment() method.
There was a problem hiding this comment.
I think that to correctly manage the handling of the queue segments the data structures used to track those has to be accessed serially.
The data structures that manage the segment allocation, free and reassignment are:
queueSegmentsrecycledSegments
inQueuePoolclass.
The access should be serialized only operating phase, when various session threads ask for next free segment.
However we can't leave uncovered by lock in other cases, like queue definition loads, during QueuePool instantiation, during close or during the retrieval of next tail segment of a queue.
There was a problem hiding this comment.
Maybe I'm little wrong. The lock should guard only the creation or reuse of a Segment, so only recycledSegments are part of the critical section, not queueSegments.
There was a problem hiding this comment.
Do I understand the following correctly:
- Each single Queue is only accessed by a single Thread
- The QueuePool assigns segments to Queues
- Once a Segment is assigned to a Queue, only this queue touches the segment
- A queue can return a segment to the Pool. It is then added to recycledSegments.
I suspect the lock is only needed when:
- a Queue requests a segment from the pool, since it would be bad if two Queues get the same segment.
- closing/deleting an entire page (if that ever happens) since it would be bad if a page is deleted with a segment that a Queue just received.
There was a problem hiding this comment.
Yes you understand right
andsel
force-pushed
the
feature/unsafe_queues
branch
from
f987236
to
3ad5cc6
Compare
|
|
||
| private static final Logger LOG = LoggerFactory.getLogger(QueuePool.class); | ||
|
|
||
| static final boolean queueDebug = Boolean.parseBoolean(System.getProperty("moquette.queue.debug", "false")); |
There was a problem hiding this comment.
Ah, no, it's used in two other classes.
…ith idiomatic asci chars
andsel
force-pushed
the
feature/unsafe_queues
branch
from
bf1dfe8
to
8043765
Compare
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.