Merge pull request #139 from lpgauth/hot-path-opts

Trim server-loop overhead and add a socket NIF protocol

Author: lpgauth

src/shackle.erl

@@ -1,9 +1,6 @@
 -module(shackle).
 -include("shackle_internal.hrl").
 
--compile(inline).
--compile({inline_size, 512}).
-
 %% public
 -export([
     call/2,
@@ -22,10 +19,10 @@
 -type external_request_id() :: term().
 -type inet_address() :: inet:ip_address() | inet:hostname().
 -type inet_port() :: inet:port_number().
--type protocol() :: shackle_ssl| shackle_tcp | shackle_udp.
+-type protocol() :: shackle_socket | shackle_ssl | shackle_tcp | shackle_udp.
 -type request_id() :: {shackle_server:name(), reference()}.
 -type response() :: {external_request_id(), term()}.
--type socket() :: inet:socket() | ssl:sslsocket().
+-type socket() :: inet:socket() | socket:socket() | ssl:sslsocket().
 -type socket_option() :: gen_tcp:connect_option() | gen_udp:option() | ssl:tls_client_option().
 -type socket_options() :: [socket_option()].
 -type table() :: atom().
@@ -103,7 +100,7 @@ cast(PoolName, Request, Pid, Timeout) ->
 
 receive_response(RequestId) ->
     receive
-        {#cast {request_id = RequestId}, Reply} ->
+        {shackle_reply, RequestId, Reply} ->
             Reply
     end.
 
@@ -112,7 +109,7 @@ receive_response(RequestId) ->
 
 receive_response(RequestId, Timeout) ->
     receive
-        {#cast {request_id = RequestId}, Reply} ->
+        {shackle_reply, RequestId, Reply} ->
             Reply
     after Timeout ->
         {error, timeout}

src/shackle_app.erl

@@ -31,6 +31,8 @@ stop() ->
     {ok, pid()}.
 
 start(_StartType, _StartArgs) ->
+    persistent_term:put({shackle, telemetry},
+        application:get_env(?APP, telemetry, true)),
     shackle_sup:start_link().
 
 -spec stop(term()) ->

src/shackle_backlog.erl

@@ -1,8 +1,5 @@
 -module(shackle_backlog).
 
--compile(inline).
--compile({inline_size, 512}).
-
 %% internal
 -export([
     check/3,

src/shackle_pool.erl

@@ -120,7 +120,6 @@ terminate() ->
 %% private
 cleanup(Name, OptionsRec) ->
     shackle_backlog:delete(Name),
-    shackle_queue:delete(Name),
     shackle_status:delete(Name),
     cleanup_ets(Name, OptionsRec),
     cleanup_foil(Name, OptionsRec).
@@ -201,7 +200,6 @@ server_id(Name, PoolSize, round_robin) ->
 
 setup(Name, #pool_options {pool_size = PoolSize} = OptionsRec) ->
     shackle_backlog:new(Name),
-    shackle_queue:new(Name),
     shackle_status:new(Name, PoolSize),
     setup_ets(Name, OptionsRec),
     setup_foil(Name, OptionsRec).

src/shackle_queue.erl

@@ -1,9 +1,6 @@
 -module(shackle_server).
 -include("shackle_internal.hrl").
 
--compile(inline).
--compile({inline_size, 512}).
-
 -export([
     start_link/2
 ]).
@@ -26,7 +23,8 @@
     pool_name        :: shackle_pool:name(),
     port             :: shackle:inet_port(),
     protocol         :: shackle:protocol(),
-    queue            :: shackle:table(),
+    queue = #{}      :: #{shackle:external_request_id() =>
+                              {shackle:cast(), reference()}},
     reconnect_state  :: undefined | reconnect_state(),
     socket           :: undefined | shackle:socket(),
     socket_options   :: shackle:socket_options(),
@@ -86,7 +84,6 @@ init(Name, Parent, Opts) ->
         pool_name = PoolName,
         port = Port,
         protocol = Protocol,
-        queue = shackle_queue:table_name(PoolName),
         reconnect_state = ReconnectState,
         socket_options = SocketOptions
     }, undefined}}.
@@ -104,7 +101,6 @@ handle_msg({Request, #cast {
         timeout = Timeout
     } = Cast}, {#state {
         client = Client,
-        id = Id,
         pool_name = PoolName,
         protocol = Protocol,
         queue = Queue,
@@ -118,14 +114,15 @@ handle_msg({Request, #cast {
                     shackle_telemetry:send(Client, iolist_size(Data)),
                     case ExtRequestId of
                         undefined ->
-                            reply(ok, Cast, State);
+                            reply(ok, Cast, State),
+                            {ok, {State, ClientState2}};
                         _ ->
                             Msg = {timeout, ExtRequestId},
                             TimerRef = erlang:send_after(Timeout, self(), Msg),
-                            shackle_queue:add(Queue, Id, ExtRequestId, Cast,
-                                TimerRef)
-                    end,
-                    {ok, {State, ClientState2}};
+                            Queue2 = maps:put(ExtRequestId, {Cast, TimerRef},
+                                Queue),
+                            {ok, {State#state {queue = Queue2}, ClientState2}}
+                    end;
                 {error, Reason} ->
                     ?WARN(PoolName, "send error: ~p", [Reason]),
                     Protocol:close(Socket),
@@ -139,6 +136,24 @@ handle_msg({Request, #cast {
             reply({error, client_crash}, Cast, State),
             {ok, {State, ClientState}}
     end;
+handle_msg({'$socket', Socket, select, _Handle}, {#state {
+        socket = Socket
+    } = State, ClientState}) ->
+
+    case shackle_socket:recv(Socket) of
+        {ok, Data} ->
+            handle_msg_data(Socket, Data, State, ClientState);
+        wait ->
+            {ok, {State, ClientState}};
+        {error, closed} ->
+            handle_msg_close(Socket, State, ClientState);
+        {error, Reason} ->
+            handle_msg_error(Socket, Reason, State, ClientState)
+    end;
+handle_msg({'$socket', _Socket, select, _Handle}, {State, ClientState}) ->
+    {ok, {State, ClientState}};
+handle_msg({'$socket', Socket, abort, _Info}, {State, ClientState}) ->
+    handle_msg_close(Socket, State, ClientState);
 handle_msg({ssl, Socket, Data}, {State, ClientState}) ->
     handle_msg_data(Socket, Data, State, ClientState);
 handle_msg({ssl_closed, Socket}, {State, ClientState}) ->
@@ -187,7 +202,6 @@ handle_msg(?MSG_CONNECT, {#state {
     end;
 handle_msg({timeout, ExtRequestId}, {#state {
         client = Client,
-        id = Id,
         pool_name = PoolName,
         protocol = Protocol,
         queue = Queue,
@@ -199,8 +213,8 @@ handle_msg({timeout, ExtRequestId}, {#state {
             try Client:handle_timeout(ExtRequestId, ClientState) of
                 {ok, Reply, ClientState2} ->
                     shackle_telemetry:handle_timeout(Client),
-                    process_responses([Reply], State),
-                    {ok, {State, ClientState2}};
+                    State2 = process_responses([Reply], State),
+                    {ok, {State2, ClientState2}};
                 {error, Reason, ClientState2} ->
                     ?WARN(PoolName, "handle_timeout error: ~p", [Reason]),
                     Protocol:close(Socket),
@@ -213,14 +227,14 @@ handle_msg({timeout, ExtRequestId}, {#state {
                     close(State, ClientState)
             end;
         false ->
-            case shackle_queue:remove(Queue, Id, ExtRequestId) of
-                {ok, Cast, _TimerRef} ->
+            case maps:take(ExtRequestId, Queue) of
+                {{Cast, _TimerRef}, Queue2} ->
                     shackle_telemetry:timeout(Client),
-                    reply({error, timeout}, Cast, State);
-                {error, not_found} ->
-                    ok
-            end,
-            {ok, {State, ClientState}}
+                    reply({error, timeout}, Cast, State),
+                    {ok, {State#state {queue = Queue2}, ClientState}};
+                error ->
+                    {ok, {State, ClientState}}
+            end
     end;
 handle_msg(Msg, {#state {
         pool_name = PoolName
@@ -303,8 +317,8 @@ client_setup(Client, PoolName, Protocol, Socket, ClientState) ->
 
 close(#state {id = Id} = State, ClientState) ->
     shackle_status:disable(Id),
-    reply_all({error, socket_closed}, State),
-    reconnect(State, ClientState).
+    State2 = reply_all({error, socket_closed}, State),
+    reconnect(State2, ClientState).
 
 connect(Protocol, Address, Port, SocketOptions, PoolName) ->
     case inet:getaddrs(Address, inet) of
@@ -342,8 +356,8 @@ handle_msg_data(Socket, Data, #state {
     shackle_telemetry:recv(Client, size(Data)),
     try Client:handle_data(Data, ClientState) of
         {ok, Replies, ClientState2} ->
-            process_responses(Replies, State),
-            {ok, {State, ClientState2}};
+            State2 = process_responses(Replies, State),
+            {ok, {State2, ClientState2}};
         {error, Reason, ClientState2} ->
             ?WARN(PoolName, "handle_data error: ~p", [Reason]),
             Protocol:close(Socket),
@@ -370,27 +384,26 @@ handle_msg_error(Socket, Reason, #state {
 handle_msg_error(_Socket, _Reason, State, ClientState) ->
     {ok, {State, ClientState}}.
 
-process_responses([], _State) ->
-    ok;
+process_responses([], State) ->
+    State;
 process_responses([{ExtRequestId, Reply} | T], #state {
         client = Client,
-        id = Id,
         queue = Queue
     } = State) ->
 
     shackle_telemetry:replies(Client),
-    case shackle_queue:remove(Queue, Id, ExtRequestId) of
-        {ok, #cast {timestamp = Timestamp} = Cast, TimerRef} ->
+    case maps:take(ExtRequestId, Queue) of
+        {{#cast {timestamp = Timestamp} = Cast, TimerRef}, Queue2} ->
             shackle_telemetry:found(Client),
             Diff = erlang:monotonic_time(microsecond) - Timestamp,
             shackle_telemetry:reply(Client, Diff),
             erlang:cancel_timer(TimerRef),
-            reply(Reply, Cast, State);
-        {error, not_found} ->
+            reply(Reply, Cast, State),
+            process_responses(T, State#state {queue = Queue2});
+        error ->
             shackle_telemetry:not_found(Client),
-            ok
-    end,
-    process_responses(T, State).
+            process_responses(T, State)
+    end.
 
 reconnect(State, undefined) ->
     reconnect_timer(State, undefined);
@@ -459,22 +472,18 @@ reply(_Reply, #cast {pid = undefined}, #state {
 
     shackle_backlog:decrement(Backlog, Id),
     ok;
-reply(Reply, #cast {pid = Pid} = Cast, #state {
+reply(Reply, #cast {pid = Pid, request_id = RequestId}, #state {
         backlog = Backlog,
         id = Id
     }) ->
 
     shackle_backlog:decrement(Backlog, Id),
-    Pid ! {Cast, Reply},
+    Pid ! {shackle_reply, RequestId, Reply},
     ok.
 
-reply_all(Reply, #state {
-        id = Id,
-        queue = Queue
-    } = State) ->
-
-    Requests = shackle_queue:clear(Queue, Id),
-    reply_all(Reply, Requests, State).
+reply_all(Reply, #state {queue = Queue} = State) ->
+    reply_all(Reply, maps:values(Queue), State),
+    State#state {queue = #{}}.
 
 reply_all(_Reply, [], _State) ->
     ok;