[
https://issues.apache.org/jira/browse/QPID-3079?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13046936#comment-13046936
]
[email protected] commented on QPID-3079:
-----------------------------------------------------
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/860/#review797
-----------------------------------------------------------
/branches/qpid-3079/qpid/cpp/src/qpid/broker/DeliveryRecord.h
<https://reviews.apache.org/r/860/#comment1724>
See reply in SemanticState.cpp comments.
/branches/qpid-3079/qpid/cpp/src/qpid/broker/PersistableMessage.cpp
<https://reviews.apache.org/r/860/#comment1728>
You're correct - I'll give it a try.
/branches/qpid-3079/qpid/cpp/src/qpid/broker/PersistableMessage.cpp
<https://reviews.apache.org/r/860/#comment1729>
The old code _may_ be in error - it added the queue on asyncEnqueue *and*
asyncDequeue, and nowhere do I see it -remove- it!
I *think* we need to add on enqueue, and remove when dequeueComplete(), but
I could be wrong....
/branches/qpid-3079/qpid/cpp/src/qpid/broker/PersistableQueue.h
<https://reviews.apache.org/r/860/#comment1730>
The pointer is actually _shared_ between the MessageStoreImpl and the
Queue. My changes caused a bug where, if the queue was deleted _before_ a
pending dequeue completed, then the queue was destroyed during the
MessageStoreImpl destructor, which was using the pointer to do the flush. I
think explicitly making it shared is the right thing to do.
/branches/qpid-3079/qpid/cpp/src/qpid/broker/PersistableQueue.h
<https://reviews.apache.org/r/860/#comment1731>
That's a good point - let me investigate making the store api a bit more
symmetrical with this change.
Will definitely get input from Steve - thanks for the head's up!
/branches/qpid-3079/qpid/cpp/src/qpid/broker/Queue.h
<https://reviews.apache.org/r/860/#comment1725>
I agree it appears overly complex, but it addresses a couple of problems:
1) "lazy" allocation of tracking context. The caller does not know if the
dequeue is sync or async at the point of call. The factory callback allows the
caller to allocate context prior to needing to service the completion (and in
the current thread).
2) Less racy - it guarantees that the completion will be held off until the
factory method returns.
/branches/qpid-3079/qpid/cpp/src/qpid/broker/Queue.h
<https://reviews.apache.org/r/860/#comment1726>
Will do.
/branches/qpid-3079/qpid/cpp/src/qpid/broker/Queue.h
<https://reviews.apache.org/r/860/#comment1727>
See the above reply to Alan. I don't think the return code is used
anywhere - it was bool originally. But relying on the return code is racy.
/branches/qpid-3079/qpid/cpp/src/qpid/broker/Queue.cpp
<https://reviews.apache.org/r/860/#comment1732>
The P.M. _could_ be being dequeued simultaneously from multiple queues.
Won't that just move the map from the queue to the P.M.?
/branches/qpid-3079/qpid/cpp/src/qpid/broker/Queue.cpp
<https://reviews.apache.org/r/860/#comment1733>
Perhaps - I still think we'd need to track multiple outstanding callbacks
for a single message instance.
/branches/qpid-3079/qpid/cpp/src/qpid/broker/SemanticState.cpp
<https://reviews.apache.org/r/860/#comment1734>
Actually, it's very likely that all the messages that are in a
Message.accept sequence are on the same queue. I noticed that during debug - a
flush of a single message.accept command resulted in multiple flush calls to
the same queue. According to Kim, multiple flushes to the same queue are
ignored, so I could drop this, but that may not be true for store in general.
/branches/qpid-3079/qpid/cpp/src/qpid/broker/SemanticState.cpp
<https://reviews.apache.org/r/860/#comment1735>
There may be multiple pending dequeues for a given message, so we would
need some sort of container anyway.
/branches/qpid-3079/qpid/cpp/src/qpid/broker/SemanticState.cpp
<https://reviews.apache.org/r/860/#comment1736>
I'm not really happy with this approach, but I can't think of a way to
track N outstanding dequeue operations without using some dynamic container.
Each callback is actually a "functor" object that contains state needed to
map the dequeue back to the pending Message.Accept. This state impl is hidden
from the Message object - I'd hate to have to expose command-specific state
into the messages (but I'm probably not thinking abstractly enough at this
point :P )
/branches/qpid-3079/qpid/cpp/src/qpid/broker/SemanticState.cpp
<https://reviews.apache.org/r/860/#comment1737>
The behaviour of the unacked list should be the same as it is now. I'm
more worried about the completion of the Message.Accept cmd, which will happen
after the store has completed dequeue. That will be a visible change in
behaviour and I don't think clustering will like it.
/branches/qpid-3079/qpid/cpp/src/tests/AsyncCompletion.cpp
<https://reviews.apache.org/r/860/#comment1738>
bingo - added the sync, and all is well. Thanks.
- Kenneth
On 2011-06-08 20:31:05, Kenneth Giusti wrote:
bq.
bq. -----------------------------------------------------------
bq. This is an automatically generated e-mail. To reply, visit:
bq. https://reviews.apache.org/r/860/
bq. -----------------------------------------------------------
bq.
bq. (Updated 2011-06-08 20:31:05)
bq.
bq.
bq. Review request for qpid, Alan Conway, Gordon Sim, and Kim van der Riet.
bq.
bq.
bq. Summary
bq. -------
bq.
bq. Modifies the broker's handling of Message.Accept to hold off the
completion of the command until all messages related to the accept have
completed dequeue. This particularly applies to persistent messages, as the
store::dequeue() operation must complete before the message is considered fully
dequeued.
bq.
bq. Note this bugfix requires some changes to the broker's store module
interface: previously, the store only identified the message when a dequeue
was completed. This is not enough information - the queue from which is was
removed must also be identified (the message may be in the process of being
dequeued on several queues at once).
bq.
bq.
bq. This addresses bug qpid-3079.
bq. https://issues.apache.org/jira/browse/qpid-3079
bq.
bq.
bq. Diffs
bq. -----
bq.
bq. /branches/qpid-3079/qpid/cpp/src/qpid/broker/AsyncCompletion.h 1124895
bq. /branches/qpid-3079/qpid/cpp/src/qpid/broker/DeliveryRecord.h 1124895
bq. /branches/qpid-3079/qpid/cpp/src/qpid/broker/DeliveryRecord.cpp 1124895
bq. /branches/qpid-3079/qpid/cpp/src/qpid/broker/Message.h 1124895
bq. /branches/qpid-3079/qpid/cpp/src/qpid/broker/Message.cpp 1124895
bq. /branches/qpid-3079/qpid/cpp/src/qpid/broker/MessageStore.h 1124895
bq. /branches/qpid-3079/qpid/cpp/src/qpid/broker/MessageStoreModule.h
1124895
bq. /branches/qpid-3079/qpid/cpp/src/qpid/broker/MessageStoreModule.cpp
1124895
bq. /branches/qpid-3079/qpid/cpp/src/qpid/broker/NullMessageStore.h 1124895
bq. /branches/qpid-3079/qpid/cpp/src/qpid/broker/NullMessageStore.cpp
1124895
bq. /branches/qpid-3079/qpid/cpp/src/qpid/broker/PersistableMessage.h
1124895
bq. /branches/qpid-3079/qpid/cpp/src/qpid/broker/PersistableMessage.cpp
1124895
bq. /branches/qpid-3079/qpid/cpp/src/qpid/broker/PersistableQueue.h 1124895
bq. /branches/qpid-3079/qpid/cpp/src/qpid/broker/Queue.h 1124895
bq. /branches/qpid-3079/qpid/cpp/src/qpid/broker/Queue.cpp 1124895
bq. /branches/qpid-3079/qpid/cpp/src/qpid/broker/RecoverableQueue.h 1124895
bq. /branches/qpid-3079/qpid/cpp/src/qpid/broker/RecoveryManagerImpl.cpp
1124895
bq. /branches/qpid-3079/qpid/cpp/src/qpid/broker/SemanticState.cpp 1124895
bq. /branches/qpid-3079/qpid/cpp/src/qpid/broker/SessionContext.h 1124895
bq. /branches/qpid-3079/qpid/cpp/src/qpid/broker/SessionState.h 1124895
bq. /branches/qpid-3079/qpid/cpp/src/qpid/broker/SessionState.cpp 1124895
bq. /branches/qpid-3079/qpid/cpp/src/tests/AsyncCompletion.cpp 1124895
bq. /branches/qpid-3079/qpid/cpp/src/tests/QueueTest.cpp 1124895
bq. /branches/qpid-3079/qpid/cpp/src/tests/TestMessageStore.h 1124895
bq.
bq. Diff: https://reviews.apache.org/r/860/diff
bq.
bq.
bq. Testing
bq. -------
bq.
bq. broker unit tests, store unit tests (modified jboss store). Still needs
to be vetted on non-linux, and have latest trunk merged in.
bq.
bq.
bq. Thanks,
bq.
bq. Kenneth
bq.
bq.
> message.accept command should be completed on a per-dequeue basis
> -----------------------------------------------------------------
>
> Key: QPID-3079
> URL: https://issues.apache.org/jira/browse/QPID-3079
> Project: Qpid
> Issue Type: Bug
> Components: C++ Broker
> Affects Versions: 0.8, 0.9
> Reporter: Ken Giusti
> Assignee: Ken Giusti
> Fix For: 0.11
>
> Attachments: proposal.txt
>
>
> ** Overview
> Asynchronous completion means that command execution is initiated in one
> thread
> (a client connection thread) and completed in a different thread.
> When the async store is loaded, message.transfer commands are
> completed by a store thread that does the async write.
> ** Issues with asynchronous completion code as of revision r1029686
> *** Not really asynchronous
> IncompleteMessageList::process blocks the connection thread till all
> outstanding async commands (messages) for the session are complete.
> With the new cluster, this could deadlock since it is blocking a Poller
> thread.
> *** Race condition for message.transfer
>
> Sketch of the current code:
> // Called in connection thread
> PersistableMessage::enqueueAsync { ++counter; }
> // Called in store thread once message is written.
> PersistableMessage::enqueueComplete { if (--counter == 0) notifyCompleted(); }
> The intent is that notify be called once per message, after the
> message has been written to each queue it was routed to.
> However of a message is routed to N queues, it's possible for
> notifyCompleted to be called up to N times. The store thread could
> call notifyCompleted for the first queue before the connection thread
> has called enqueueAsync for the second queue, and so on.
> *** No asynchronous completion for message.accept
> We do not currently delay completion of message.accept until the
> message is deleted from the async store. This could cause duplicate
> delivery if the broker crashes after completing the message but
> before it is removed from store.
> There is code in PersistableMessage to maintain a counter for dequeues
> analogous to to the async enqueue code but this is incorrect.
> Completion of transfer is triggered when all enqueues for a message are
> complete.
> Completion of accept is triggered for *each* dequeue from a queue
> independently.
> Furthermore a single accept can reference many messages, so it can't be
> associated with a message.
> ** New requirements
> The new cluster design will need to participate in async completion, e.g.
> an accept cannot be comlpeted until the message is
> - removed from store (if present) AND
> - replicated to the cluster (if present) as dequeued
> The new cluster also needs to asynchronously complete binding commands
> (declare, bind, delete) when they are replicated to the cluster.
--
This message is automatically generated by JIRA.
For more information on JIRA, see: http://www.atlassian.com/software/jira
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:[email protected]
