Add commit queue for low priority requests

Summary:
The follow stack of diffs will implement priority requests (as per [this](https://docs.google.com/document/d/181je6C5Iwmi5Ckl87UPLOug2V0eLVp8jIM2DoDlWbpo/edit?tab=t.0#heading=h.9l8mmncnncc8) doc).

 ---

This diff separates the pending commit queue into two separate queues: one for each high priority requests, and one for low priority requests.

In the current paradigm, we treat all requests as high priority, and this diff continues doing that.

Reviewed By: jaher

Differential Revision: D82855032

fbshipit-source-id: da9b19129d6c703d7a0a1d6d5b37d179eb176c68

Author: patelarjav

include/wa_raft.hrl

@@ -142,9 +142,12 @@
 %% Maximum bytes of pending applies for any single RAFT partition
 -define(RAFT_MAX_PENDING_APPLY_BYTES, raft_max_pending_apply_bytes).
 -define(RAFT_MAX_PENDING_APPLY_BYTES(App), ?RAFT_APP_CONFIG(App, ?RAFT_MAX_PENDING_APPLY_BYTES, 32_000_000)).
-%% Maximum number of pending commits for any single RAFT partition
--define(RAFT_MAX_PENDING_COMMITS, raft_max_pending_commits).
--define(RAFT_MAX_PENDING_COMMITS(App), ?RAFT_APP_CONFIG(App, ?RAFT_MAX_PENDING_COMMITS, 1500)).
+%% Maximum number of pending high priority commits for any single RAFT partition
+-define(RAFT_MAX_PENDING_HIGH_PRIORITY_COMMITS, raft_max_pending_high_priority_commits).
+-define(RAFT_MAX_PENDING_HIGH_PRIORITY_COMMITS(App), ?RAFT_APP_CONFIG(App, ?RAFT_MAX_PENDING_HIGH_PRIORITY_COMMITS, 1500)).
+%% Maximum number of pending low priority commits for any single RAFT partition
+-define(RAFT_MAX_PENDING_LOW_PRIORITY_COMMITS, raft_max_pending_low_priority_commits).
+-define(RAFT_MAX_PENDING_LOW_PRIORITY_COMMITS(App), ?RAFT_APP_CONFIG(App, ?RAFT_MAX_PENDING_LOW_PRIORITY_COMMITS, 150)).
 %% Maximum number of pending reads for any single RAFT partition
 -define(RAFT_MAX_PENDING_READS, raft_max_pending_reads).
 -define(RAFT_MAX_PENDING_READS(App), ?RAFT_APP_CONFIG(App, ?RAFT_MAX_PENDING_READS, 5000)).

src/wa_raft_acceptor.erl

@@ -251,11 +251,11 @@ terminate(Reason, #state{name = Name}) ->
 
 %% Enqueue a commit.
 -spec commit_impl(From :: gen_server:from(), Request :: op(), Priority :: priority(), State :: #state{}) -> continue | commit_error().
-commit_impl(From, {Key, _} = Op, _Priority, #state{name = Name, server = Server, queues = Queues}) ->
+commit_impl(From, {Key, _} = Op, Priority, #state{name = Name, server = Server, queues = Queues}) ->
     StartT = os:timestamp(),
     try
         ?RAFT_LOG_DEBUG("Acceptor[~0p] starts to handle commit of ~0P from ~0p.", [Name, Op, 30, From]),
-        case wa_raft_queue:commit_started(Queues) of
+        case wa_raft_queue:commit_started(Queues, Priority) of
             commit_queue_full ->
                 ?RAFT_LOG_WARNING(
                     "Acceptor[~0p] is rejecting commit request from ~0p because the commit queue is full.",

src/wa_raft_queue.erl

@@ -13,10 +13,10 @@
 -export([
     queues/1,
     queues/2,
-    commit_queue_size/1,
     commit_queue_size/2,
-    commit_queue_full/1,
+    commit_queue_size/3,
     commit_queue_full/2,
+    commit_queue_full/3,
     apply_queue_size/1,
     apply_queue_size/2,
     apply_queue_byte_size/1,
@@ -35,7 +35,7 @@
 
 %% PENDING COMMIT QUEUE API
 -export([
-    commit_started/1,
+    commit_started/2,
     commit_cancelled/3,
     commit_completed/3
 ]).
@@ -85,15 +85,17 @@
 -define(RAFT_QUEUE_TABLE_OPTIONS, [named_table, public, {read_concurrency, true}, {write_concurrency, true}]).
 
 %% Total number of counters for RAFT partition specfic counters
--define(RAFT_NUMBER_OF_QUEUE_SIZE_COUNTERS, 4).
+-define(RAFT_NUMBER_OF_QUEUE_SIZE_COUNTERS, 5).
 %% Index into counter reference for counter tracking apply queue size
 -define(RAFT_APPLY_QUEUE_SIZE_COUNTER, 1).
 %% Index into counter reference for counter tracking apply total byte size
 -define(RAFT_APPLY_QUEUE_BYTE_SIZE_COUNTER, 2).
-%% Index into counter reference for counter tracking commit queue size
--define(RAFT_COMMIT_QUEUE_SIZE_COUNTER, 3).
+%% Index into counter reference for counter tracking high priority commit queue size
+-define(RAFT_HIGH_PRIORITY_COMMIT_QUEUE_SIZE_COUNTER, 3).
+%% Index into counter reference for counter tracking low priority commit queue size
+-define(RAFT_LOW_PRIORITY_COMMIT_QUEUE_SIZE_COUNTER, 4).
 %% Index into counter reference for counter tracking read queue size
--define(RAFT_READ_QUEUE_SIZE_COUNTER, 4).
+-define(RAFT_READ_QUEUE_SIZE_COUNTER, 5).
 
 %%-------------------------------------------------------------------
 %% INTERNAL TYPES
@@ -129,33 +131,37 @@ queues(Table, Partition) ->
         Options -> queues(Options)
     end.
 
--spec commit_queue_size(Queues :: queues()) -> non_neg_integer().
-commit_queue_size(#queues{counters = Counters}) ->
-    atomics:get(Counters, ?RAFT_COMMIT_QUEUE_SIZE_COUNTER).
+-spec commit_queue_size(Queues :: queues(), Priority :: wa_raft_acceptor:priority()) -> non_neg_integer().
+commit_queue_size(#queues{counters = Counters}, high) ->
+    atomics:get(Counters, ?RAFT_HIGH_PRIORITY_COMMIT_QUEUE_SIZE_COUNTER);
+commit_queue_size(#queues{counters = Counters}, low) ->
+    atomics:get(Counters, ?RAFT_LOW_PRIORITY_COMMIT_QUEUE_SIZE_COUNTER).
 
--spec commit_queue_size(wa_raft:table(), wa_raft:partition()) -> non_neg_integer().
-commit_queue_size(Table, Partition) ->
+-spec commit_queue_size(Table :: wa_raft:table(), Partition :: wa_raft:partition(), Priority :: wa_raft_acceptor:priority()) -> non_neg_integer().
+commit_queue_size(Table, Partition, Priority) ->
     case queues(Table, Partition) of
         undefined -> 0;
-        Queue     -> commit_queue_size(Queue)
+        Queue     -> commit_queue_size(Queue, Priority)
     end.
 
--spec commit_queue_full(Queues :: queues()) -> boolean().
-commit_queue_full(#queues{application = App, counters = Counters}) ->
-    atomics:get(Counters, ?RAFT_COMMIT_QUEUE_SIZE_COUNTER) >= ?RAFT_MAX_PENDING_COMMITS(App).
+-spec commit_queue_full(Queues :: queues(), Priority :: wa_raft_acceptor:priority()) -> boolean().
+commit_queue_full(#queues{application = App, counters = Counters}, high) ->
+    atomics:get(Counters, ?RAFT_HIGH_PRIORITY_COMMIT_QUEUE_SIZE_COUNTER) >= ?RAFT_MAX_PENDING_HIGH_PRIORITY_COMMITS(App);
+commit_queue_full(#queues{application = App, counters = Counters}, low) ->
+    atomics:get(Counters, ?RAFT_LOW_PRIORITY_COMMIT_QUEUE_SIZE_COUNTER) >= ?RAFT_MAX_PENDING_LOW_PRIORITY_COMMITS(App).
 
--spec commit_queue_full(wa_raft:table(), wa_raft:partition()) -> boolean().
-commit_queue_full(Table, Partition) ->
+-spec commit_queue_full(Table :: wa_raft:table(), Partition :: wa_raft:partition(), Priority :: wa_raft_acceptor:priority()) -> boolean().
+commit_queue_full(Table, Partition, Priority) ->
     case queues(Table, Partition) of
         undefined -> false;
-        Queues    -> commit_queue_full(Queues)
+        Queues    -> commit_queue_full(Queues, Priority)
     end.
 
 -spec apply_queue_size(Queues :: queues()) -> non_neg_integer().
 apply_queue_size(#queues{counters = Counters}) ->
     atomics:get(Counters, ?RAFT_APPLY_QUEUE_SIZE_COUNTER).
 
--spec apply_queue_size(wa_raft:table(), wa_raft:partition()) -> non_neg_integer().
+-spec apply_queue_size(Table :: wa_raft:table(), Partition :: wa_raft:partition()) -> non_neg_integer().
 apply_queue_size(Table, Partition) ->
     case queues(Table, Partition) of
         undefined -> 0;
@@ -219,17 +225,17 @@ registered_name(Table, Partition) ->
 %% PENDING COMMIT QUEUE API
 %%-------------------------------------------------------------------
 
--spec commit_started(Queues :: queues()) -> ok | apply_queue_full | commit_queue_full.
-commit_started(#queues{counters = Counters} = Queues) ->
-    case commit_queue_full(Queues) of
+-spec commit_started(Queues :: queues(), Priority :: wa_raft_acceptor:priority()) -> ok | apply_queue_full | commit_queue_full.
+commit_started(#queues{counters = Counters} = Queues, Priority) ->
+    case commit_queue_full(Queues, Priority) of
         true ->
             commit_queue_full;
         false ->
             case apply_queue_full(Queues) of
                 true ->
                     apply_queue_full;
                 false ->
-                    PendingCommits = atomics:add_get(Counters, ?RAFT_COMMIT_QUEUE_SIZE_COUNTER, 1),
+                    PendingCommits = atomics:add_get(Counters, ?RAFT_HIGH_PRIORITY_COMMIT_QUEUE_SIZE_COUNTER, 1),
                     ?RAFT_GATHER('raft.acceptor.commit.request.pending', PendingCommits),
                     ok
             end
@@ -238,13 +244,13 @@ commit_started(#queues{counters = Counters} = Queues) ->
 
 -spec commit_cancelled(Queues :: queues(), From :: gen_server:from(), Reason :: wa_raft_acceptor:commit_error() | undefined) -> ok.
 commit_cancelled(#queues{counters = Counters}, From, Reason) ->
-    atomics:sub(Counters, ?RAFT_COMMIT_QUEUE_SIZE_COUNTER, 1),
+    atomics:sub(Counters, ?RAFT_HIGH_PRIORITY_COMMIT_QUEUE_SIZE_COUNTER, 1),
     Reason =/= undefined andalso gen_server:reply(From, Reason),
     ok.
 
 -spec commit_completed(Queues :: queues(), From :: gen_server:from(), Reply :: term()) -> ok.
 commit_completed(#queues{counters = Counters}, From, Reply) ->
-    atomics:sub(Counters, ?RAFT_COMMIT_QUEUE_SIZE_COUNTER, 1),
+    atomics:sub(Counters, ?RAFT_HIGH_PRIORITY_COMMIT_QUEUE_SIZE_COUNTER, 1),
     gen_server:reply(From, Reply),
     ok.