Add rabbit_queue_type behaviour #2052

kjnilsson · 2019-07-02T15:13:01Z

First cut queue type behaviour

Uses the implementing module as the value of the amqqueue record
type field. This will allow for easy dispatch to the queue type
implementation.

Provides a stateful API around queue interactions. Currently the classic queue type will need some special handling when implementing processes that interact with queues but this should simplify this as well as the development of new queue types.

lukebakken · 2019-07-02T16:19:09Z

Great idea 👍

And use the implementing module as the value of the amqqueue record `type` field. This will allow for easy dispatch to the queue type implementation.

First step to a greater abstraction!

To a more generic an extensible opaque queue type specific map.

dcorbacho · 2019-07-25T10:24:19Z

src/amqqueue.erl

-spec get_quorum_nodes(amqqueue()) -> [node()].
+-spec get_type_state(amqqueue()) -> map().
+get_type_state(#amqqueue{type_state = TState}) -> TState;
+get_type_state(_)                               -> [].


type_state is documented as a map. Why default to empty list here?

nice catch - that is what it returned when it was just quorum nodes - the value is separated by so much whitespace it's quite hard to read :) - will fix

dcorbacho · 2019-07-25T10:42:42Z

src/rabbit_channel.erl

-            %% the assumption here is that the queue state has been cleaned up and
-            %% this is a residual Ra notification
-            noreply_coalesce(State0)
+handle_info({ra_event, From, _} = Evt,


This is not very generic 🤔

no it isn't - we can consider at some point make all queues send messages with a particular format - we'd have to make a change in Ra to allow custom ra event formatting to do so though

It might be easier to convert the Ra-specific message to something more generic here as another clause, like:

handle_info({ra_event, From, Content}, ...) -> handle_info({generic_event, From, Content}, ...); handle_info({generic_event, From, _} = Evt, ...) -> ...

dcorbacho · 2019-07-25T11:18:26Z

src/rabbit_channel.erl

                             State = #ch{unconfirmed = UC,
-                                         queue_names = QNames})
+                                         queue_states = QStates})
  when ?IS_CLASSIC(QPid) ->


These macros are not generic either. Should we have a rabbit_queue_type:needs_handle_publishing_queue_down or something similar?

The same applies to all other ocurrences of this and the ?IS_QUORUM macro. Now that we have types they should disappear

I agree and to do that we'd have to move all the confirm accounting inside the queue type module. I change I found a bit scary at this point.

So having pondered this a bit more the current scope of the queue type abstraction isn't complete, it would need to handle confirms, downs etc as well to be considered so. That said I think at this point it achieves one goal: it makes it easier for us to deliver new queue types. The secondary goal of completing the queue type abstraction and making it easier for us to write new channel processes is one we can choose to defer at this point.

I agree that the abstraction does not need to be complete right away. What I would like us to keep in mind however, is the API we expose and the plugin ecosystem outside of our team: it would be nice that a plugin developer does not have to maintain and/or publish one version of his plugin per RabbitMQ branch.

All mirrors need to be present as queue names. This introduces context linking allowing additional queue refs to be linked to a single "master" queue ref contining the actual queue context.

Also update queue type smoke test to use a cluster by default.

dumbbell

Here is a first batch of comments.

I didn't finish to review the patch yet: I'm at a park and I've just been surrounded by two dozens kids, like they believe it's a good idea to sit on the same bench as me and shout and so on... Relocating to a more peaceful place!

dumbbell · 2019-07-29T12:36:55Z

include/amqqueue.hrl

        ((?is_amqqueue_v2(Q) andalso
-          ?amqqueue_v2_field_type(Q) =:= classic) orelse
+          ?amqqueue_v2_field_type(Q) =:= rabbit_classic_queue) orelse
         ?is_amqqueue_v1(Q))).


To follow up on Diana's comments, we can probably replace those functions & associated macros by something more generic. @lukebakken did change the initial generic functions/macros to those specific more readable versions. @lukebakken: I think you linked an article about that kind of API but I don't rember the content, what was the reason behind this already?

My goal was to put all "knowledge" of how to determine if an amqqueue is classic into this macro. This way, when the identification of a classic queue changes, we only have to change this macro and not all the places where it is used.

"Tell Don't Ask"

dumbbell · 2019-07-29T12:38:06Z

src/amqqueue.erl

 -spec get_leader(amqqueue_v2()) -> node().

-get_leader(#amqqueue{type = quorum, pid = {_, Leader}}) -> Leader.
+get_leader(#amqqueue{type = rabbit_quorum_queue, pid = {_, Leader}}) -> Leader.


Should the "PID" be an opaque type as well?

Again, that "leader" thing is very quorum queues/Ra specific.

dumbbell · 2019-07-29T12:40:04Z

src/rabbit_channel.erl

             %% messages pending consumer acknowledgement
             unacked_message_q,
-             %% a map of queue ref to queue name
-             queue_names,


Are we sure that the #ch{} record is never exchanged with remote nodes or used as a Mnesia table schema?

dumbbell · 2019-07-29T12:43:39Z

src/rabbit_channel.erl

-            %% the assumption here is that the queue state has been cleaned up and
-            %% this is a residual Ra notification
-            noreply_coalesce(State0)
+handle_info({ra_event, From, _} = Evt,


It might be easier to convert the Ra-specific message to something more generic here as another clause, like:

handle_info({ra_event, From, Content}, ...) -> handle_info({generic_event, From, Content}, ...); handle_info({generic_event, From, _} = Evt, ...) -> ...

dumbbell · 2019-07-29T12:53:02Z

src/rabbit_channel.erl

                             State = #ch{unconfirmed = UC,
-                                         queue_names = QNames})
+                                         queue_states = QStates})
  when ?IS_CLASSIC(QPid) ->


I agree that the abstraction does not need to be complete right away. What I would like us to keep in mind however, is the API we expose and the plugin ecosystem outside of our team: it would be nice that a plugin developer does not have to maintain and/or publish one version of his plugin per RabbitMQ branch.

dumbbell · 2019-07-29T12:54:45Z

src/rabbit_channel.erl

                  {recover, {NoAck, ConsumerPrefetch, Exclusive, Args}} ->
-                      case catch basic_consume( %% [0]
+                      rabbit_log:info("recovering consumer  ~p", [CTag]),
+                      case catch basic_consume(


The %% [0] comment callout is removed: is the associated comment irrelevant now? In which case it should be removed as well.

dumbbell · 2019-07-29T12:57:46Z

src/rabbit_channel.erl

-                                                   self(), Acc0)
+                    fun({QRef, CTag}, MsgIds, Acc0) ->
+                            rabbit_queue_type:reject(QRef, CTag, Requeue,
+                                                     MsgIds, Acc0)


Picking a random line for a more general comment: I find the rabbit_queue_type name a bit confusing. All functions such as reject() act on a queue, not a queue type; e.g. the line above wants to reject a message in a queue, not a queue type (which makes little sense).

Did you think about the simpler rabbit_queue name already?

dumbbell

Ok, second half done. I mostly focused on the API, not on the actual changes to channels and other RabbitMQ internals, as I don't really master them.

dumbbell · 2019-07-29T13:32:07Z

src/rabbit_queue_type.erl

+                       queue_state()) ->
+    {ok, queue_state(), actions()} | {error, term()} | eol.
+
+-callback deliver([{amqqueue:amqqueue(), queue_state()}],


After a bit more thoughts, I find this callback (and all others working on a list of queues) a bit more weird in fact compared to last week:

As I said, from an API PoV, this looks inconsistent: some callbacks take a list of queues, others a single queue, and the name doesn't reflect this change of behavior.

Should the queues in that list share the same queue type? What if a queue type implementation finds a queue from a different queue type in the middle of that list?

It seems to me that this feature should be outside of the new behavior and it should call an actual deliver() callback which takes a queue as argument. Again, my comment applies to other callbacks such as recover().

dumbbell · 2019-07-29T13:36:53Z

src/rabbit_queue_type.erl

+    queue_state().
+
+-callback dequeue(NoAck :: boolean(), LimiterPid :: pid(),
+                  rabbit_types:ctag(), queue_state()) ->


Why is the ctag third here?

kjnilsson · 2019-08-01T10:49:05Z

More work is needed to make this PR fit for purpose. The confirm logic needs to be moved inside the queue type abstraction and DOWN events need to be passed down to each queue type so that they can mutate their state and run logic as appropriate.

#2062 has been created to ensure the breaking changes to the amqqueue record are included in 3.8 allowing us to work on this further and include it in a future release.

kjnilsson changed the title ~~Add marker rabbit_queue_type behaviour~~ Add rabbit_queue_type behaviour Jul 23, 2019

kjnilsson force-pushed the marker-queue-type branch from a2376e7 to cc8393e Compare July 23, 2019 13:50

kjnilsson marked this pull request as ready for review July 23, 2019 13:51

kjnilsson added 9 commits July 23, 2019 14:53

Add marker rabbit_queue_type behaviour

7fa38c0

And use the implementing module as the value of the amqqueue record `type` field. This will allow for easy dispatch to the queue type implementation.

Move queue declare into rabbit_queue_type

e5dff9f

First step to a greater abstraction!

Move queue delete into queue type implementation

8aa9d9f

First cut stateful queue type

1e8500b

Queue type: dequeue/basic_get

7a7bf15

Move info inside queue type abstraction

7d34763

Move policy change into queue type interface

f0aaa63

Add purge to queue type

4a91b5d

Add recovery to the queue type interface

5277cbb

kjnilsson force-pushed the marker-queue-type branch from cc8393e to 5277cbb Compare July 23, 2019 13:55

kjnilsson requested review from dcorbacho, dumbbell and michaelklishin July 23, 2019 13:55

kjnilsson and others added 2 commits July 24, 2019 15:28

Rename amqqueue quorum_nodes field

3603048

To a more generic an extensible opaque queue type specific map.

Fix tests and handle classic API response

2e3aaf4

dcorbacho reviewed Jul 25, 2019

View reviewed changes

kjnilsson added 6 commits July 25, 2019 12:31

Fix HA queue confirm bug

277840b

All mirrors need to be present as queue names. This introduces context linking allowing additional queue refs to be linked to a single "master" queue ref contining the actual queue context.

Fix issue with events of deleted queues

4f6ebd8

Also update queue type smoke test to use a cluster by default.

be defensive around state_info

e489655

fix basic.get reject bug

550b980

correct default value of amqqueue getter

5b6bcec

wip

899beb3

dumbbell reviewed Jul 29, 2019

View reviewed changes

kjnilsson mentioned this pull request Aug 1, 2019

Add marker rabbit_queue_type behaviour #2062

Merged

wip

1465bb6

kjnilsson closed this Sep 4, 2019

Add rabbit_queue_type behaviour #2052

Add rabbit_queue_type behaviour #2052

Uh oh!

Conversation

kjnilsson commented Jul 2, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

lukebakken commented Jul 2, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kjnilsson Jul 26, 2019 • edited by dumbbell Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dumbbell left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dumbbell left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kjnilsson commented Aug 1, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

kjnilsson commented Jul 2, 2019 •

edited

Loading

kjnilsson Jul 26, 2019 •

edited by dumbbell

Loading