Trim the server loop and batch sends

include/shackle_defaults.hrl

@@ -4,6 +4,7 @@
 -define(DEFAULT_INIT_OPTS, undefined).
 -define(DEFAULT_CONNECT_TIMEOUT, 500).
 -define(DEFAULT_MAX_RETRIES, 0).
+-define(MAX_CAST_BATCH, 64).
 -define(DEFAULT_POOL_SIZE, 16).
 -define(DEFAULT_POOL_STRATEGY, random).
 -define(DEFAULT_PROTOCOL, shackle_tcp).

include/shackle_internal.hrl

@@ -8,6 +8,7 @@
 -define(LOOKUP(Key, List), ?LOOKUP(Key, List, undefined)).
 -define(LOOKUP(Key, List, Default), shackle_utils:lookup(Key, List, Default)).
 -define(MSG_CONNECT, connect).
+-define(MSG_SWEEP, sweep).
 -define(SERVER, shackle_server).
 -define(SUPERVISOR, shackle_sup).
 

src/shackle_backlog.erl

@@ -3,19 +3,13 @@
 %% internal
 -export([
     check/3,
-    check/4,
     decrement/2,
-    decrement/3,
     delete/1,
-    delete/2,
-    new/1,
     new/2,
-    table_name/1
+    ref/1,
+    reset/2
 ]).
 
--define(DEFAULT_DECREMENT, -1).
--define(DEFAULT_INCREMENT, 1).
-
 %% types
 -type backlog_size() :: pos_integer() | infinity.
 
@@ -24,73 +18,61 @@
 ]).
 
 %% internal
--spec check(shackle:table(), shackle_server:id(), backlog_size()) ->
-    boolean().
-
-check(Table, ServerId, BacklogSize) ->
-    check(Table, ServerId, BacklogSize, ?DEFAULT_INCREMENT).
-
--spec check(shackle:table(), shackle_server:id(), backlog_size(), pos_integer()) ->
+-spec check(atomics:atomics_ref(), shackle_server:id(), backlog_size()) ->
     boolean().
 
-check(_Table, _ServerId, infinity, _Increment) ->
+check(_Ref, _ServerId, infinity) ->
     true;
-check(Table, ServerId, BacklogSize, Increment) ->
-    case increment(Table, ServerId, BacklogSize, Increment) of
-        [BacklogSize, BacklogSize] ->
-            false;
-        [_, Value] when Value =< BacklogSize ->
-            true
+check(Ref, {_PoolName, Index}, BacklogSize) ->
+    case atomics:add_get(Ref, Index, 1) of
+        Value when Value =< BacklogSize ->
+            true;
+        _Value ->
+            atomics:sub(Ref, Index, 1),
+            false
     end.
 
--spec decrement(shackle:table(), shackle_server:id()) ->
-    non_neg_integer().
-
-decrement(Table, ServerId) ->
-    decrement(Table, ServerId, ?DEFAULT_DECREMENT).
-
--spec decrement(shackle:table(), shackle_server:id(), neg_integer()) ->
-    non_neg_integer().
-
-decrement(Table, ServerId, Decrement) ->
-    ets:update_counter(Table, ServerId, {2, Decrement, 0, 0}).
+-spec decrement(atomics:atomics_ref(), shackle_server:id()) ->
+    integer().
+
+decrement(Ref, {_PoolName, Index}) ->
+    case atomics:sub_get(Ref, Index, 1) of
+        Value when Value < 0 ->
+            %% reset or infinity backlog: never go below empty
+            atomics:add(Ref, Index, 1),
+            0;
+        Value ->
+            Value
+    end.
 
 -spec delete(shackle_pool:name()) ->
     ok.
 
 delete(PoolName) ->
-    ets:delete(table_name(PoolName)),
+    persistent_term:erase({?MODULE, PoolName}),
     ok.
 
--spec delete(shackle_pool:name(), shackle_server:id()) ->
+-spec new(shackle_pool:name(), shackle_pool:pool_size()) ->
     ok.
 
-delete(PoolName, ServerId) ->
-    ets:delete(table_name(PoolName), ServerId),
+new(PoolName, PoolSize) ->
+    Ref = atomics:new(PoolSize, []),
+    persistent_term:put({?MODULE, PoolName}, Ref),
     ok.
 
--spec new(shackle_pool:name()) ->
-    ok.
-
-new(PoolName) ->
-    Table = ets:new(table_name(PoolName), shackle_utils:ets_options()),
-    ets:give_away(Table, whereis(shackle_ets_manager), undefined),
-    ok.
+-spec ref(shackle_pool:name()) ->
+    atomics:atomics_ref().
 
--spec new(shackle_pool:name(), shackle_server:id()) ->
-    ok.
+ref(PoolName) ->
+    persistent_term:get({?MODULE, PoolName}).
 
-new(PoolName, ServerId) ->
-    ets:insert(table_name(PoolName), {ServerId, 0}),
+-spec reset(shackle_pool:name(), shackle_server:id()) ->
     ok.
 
--spec table_name(shackle_pool:name()) ->
-    shackle:table().
-
-table_name(PoolName) ->
-    list_to_atom("shackle_backlog_" ++ atom_to_list(PoolName)).
-
-%% private
-increment(Table, ServerId, BacklogSize, Increment) ->
-    UpdateOps = [{2, 0}, {2, Increment, BacklogSize, BacklogSize}],
-    ets:update_counter(Table, ServerId, UpdateOps).
+reset(PoolName, {_PoolName, Index}) ->
+    case persistent_term:get({?MODULE, PoolName}, undefined) of
+        undefined ->
+            ok;
+        Ref ->
+            atomics:put(Ref, Index, 0)
+    end.

src/shackle_pool.erl

@@ -132,7 +132,6 @@ cleanup_ets(_Name, _OptionsRec) ->
 cleanup_foil(Name, #pool_options {pool_size = PoolSize}) ->
     foil:delete(?MODULE, Name),
     [foil:delete(?MODULE, {Name, N}) || N <- lists:seq(1, PoolSize)],
-    foil:delete(?MODULE, {Name, backlog}),
     foil:load(?MODULE).
 
 options(Name) ->
@@ -176,7 +175,7 @@ server(Name, #pool_options {
     ServerId = server_id(Name, PoolSize, PoolStrategy),
     case shackle_status:active(ServerId) of
         true ->
-            {ok, Backlog} = shackle_pool_foil:lookup({Name, backlog}),
+            Backlog = shackle_backlog:ref(Name),
             {ok, ServerName} = shackle_pool_foil:lookup(ServerId),
             case shackle_backlog:check(Backlog, ServerId, BacklogSize) of
                 true ->
@@ -199,7 +198,7 @@ server_id(Name, PoolSize, round_robin) ->
     {Name, ServerId}.
 
 setup(Name, #pool_options {pool_size = PoolSize} = OptionsRec) ->
-    shackle_backlog:new(Name),
+    shackle_backlog:new(Name, PoolSize),
     shackle_status:new(Name, PoolSize),
     setup_ets(Name, OptionsRec),
     setup_foil(Name, OptionsRec).
@@ -213,7 +212,6 @@ setup_foil(Name, #pool_options {pool_size = PoolSize} = OptionsRec) ->
     foil:insert(?MODULE, Name, OptionsRec),
     [foil:insert(?MODULE, {Name, N}, server_name(Name, N)) ||
         N <- lists:seq(1, PoolSize)],
-    foil:insert(?MODULE, {Name, backlog}, shackle_backlog:table_name(Name)),
     foil:load(?MODULE).
 
 server_name(Name, Index) ->