GitHub PullRequestReviewCommentEvent at 2014-11-13 18:27:12
Repository: elysium
Actor: elvisinaka
Payload: {"action":"created","comment":{"url":"https://api.github.com/repos/inaka/elysium/pulls/comments/20311312","id":20311312,"diff_hunk":"@@ -0,0 +1,537 @@\n+%%%------------------------------------------------------------------------------\n+%%% @copyright (c) 2014, DuoMark International, Inc.\n+%%% @author Jay Nelson <jay@duomark.com> [http://duomark.com/]\n+%%% @reference 2014 Development sponsored by TigerText, Inc. [http://tigertext.com/]\n+%%% @reference The license is based on the template for Modified BSD from\n+%%%   <a href=\"http://opensource.org/licenses/BSD-3-Clause\">OSI</a>\n+%%% @doc\n+%%%   Elysium_bs_parallel is a buffering strategy which enforces serial\n+%%%   initiation of Cassandra session queries, although the duration of\n+%%%   a query will vary so they are not guaranteed to execute in serial\n+%%%   order. This approach allows spikes in the traffic which exceed the\n+%%%   number of availabel elysium sessions. The buffer is maintained as\n+%%%   a FIFO ets_buffer, so it has overhead when unloading expired pending\n+%%%   requests when compared to a LIFO buffer for the same task. All\n+%%%   requests attempt to fetch an idle session before being added to\n+%%%   the end of the pending queue.\n+%%%\n+%%% @since 0.1.5\n+%%% @end\n+%%%------------------------------------------------------------------------------\n+-module(elysium_bs_parallel).\n+-author('jay@duomark.com').\n+\n+%% External API\n+-export([\n+         insert_audit_counts/1,\n+         create_connection/3,\n+         checkin_connection/4,\n+         checkout_connection/1,\n+         handle_pending_request/6,\n+         pend_request/2,\n+         status/1\n+        ]).\n+\n+%% FSM states\n+-type fun_request()   :: fun((pid(), [any()], seestar:consistency()) -> [any()]).\n+-type query_request() :: {bare_fun, config_type(), fun_request(), [any()], seestar:consistency()}\n+                       | {mod_fun, config_type(), module(), atom(), [any()], seestar:consistency()}.\n+\n+-define(SERVER, ?MODULE).\n+\n+-include(\"elysium_types.hrl\").\n+\n+-type timestamp() :: binary().\n+-record(audit_serial, {\n+          session_id_key    :: {module(), pid()},\n+\n+          init_checkin      :: timestamp(),\n+          last_checkin      :: timestamp(),\n+          num_checkins  = 0 :: non_neg_integer(),\n+\n+          init_checkout     :: timestamp(),\n+          last_checkout     :: timestamp(),\n+          num_checkouts = 0 :: non_neg_integer(),\n+\n+          init_pending      :: timestamp(),\n+          last_pending      :: timestamp(),\n+          num_pendings  = 0 :: non_neg_integer()\n+         }).\n+-type audit_serial() :: #audit_serial{}.\n+\n+-record(audit_serial_counts, {\n+          count_type_key            :: {module(), counts},\n+          pending_dead          = 0 :: non_neg_integer(),\n+          pending_ets_errors    = 0 :: non_neg_integer(),\n+          pending_missing_data  = 0 :: non_neg_integer(),\n+          pending_timeouts      = 0 :: non_neg_integer(),\n+          session_dead          = 0 :: non_neg_integer(),\n+          session_decay         = 0 :: non_neg_integer(),\n+          session_ets_errors    = 0 :: non_neg_integer(),\n+          session_missing_data  = 0 :: non_neg_integer(),\n+          session_timeouts      = 0 :: non_neg_integer(),\n+          session_wrong         = 0 :: non_neg_integer(),\n+          worker_errors         = 0 :: non_neg_integer(),\n+          worker_timeouts       = 0 :: non_neg_integer()\n+         }).\n+-type audit_serial_counts() :: #audit_serial_counts{}.\n+\n+-export_type([audit_serial/0, audit_serial_counts/0]).\n+\n+-define(COUNTS_KEY, {?MODULE, counts}).\n+\n+\n+%%%-----------------------------------------------------------------------\n+%%% External API\n+%%%-----------------------------------------------------------------------\n+\n+-spec insert_audit_counts(atom()) -> true.\n+%% @doc Insert audit_serial_counts record to hold error increments.\n+insert_audit_counts(Audit_Name) ->\n+    Count_Rec = #audit_serial_counts{count_type_key=?COUNTS_KEY},\n+    true = ets:insert_new(Audit_Name, Count_Rec).\n+\n+-type resource_counts() :: {idle_connections | pending_requests,\n+                            {atom(), non_neg_integer(), non_neg_integer()}}.\n+-spec status(config_type()) -> {status, [resource_counts()]}.\n+%% @doc Get the current queue size of the pending queue.\n+status(Config) ->\n+    {status, [idle_connections(Config),\n+              pending_requests(Config)]}.\n+\n+-spec checkout_connection(config_type()) -> {{Ip::string(), Port::pos_integer()}, pid()} | none_available.\n+%% @doc\n+%%   Allocate a seestar_session to the caller by popping an entry\n+%%   from the front of the connection queue. This function either\n+%%   returns a live pid(), or none_available to indicate that all\n+%%   connections to Cassandra are currently checked out.\n+%%\n+%%   If there are internal delays on the ets_buffer FIFO queue,\n+%%   this function will retry. If the session handed back is no\n+%%   longer live, it is tossed and a new session is fetched. In\n+%%   both cases, up to Config_Module:checkout_max_retry attempts\n+%%   are tried before returning none_available. If the queue is\n+%%   actually empty, no retries are performed.\n+%%\n+%%   The configuration parameter is not validated because this\n+%%   function should be a hotspot and we don't want it to slow\n+%%   down or become a concurrency bottleneck.\n+%% @end\n+checkout_connection(Config) ->\n+    Session_Queue = elysium_config:session_queue_name (Config),\n+    Max_Retries   = elysium_config:checkout_max_retry (Config),\n+    fetch_pid_from_queue(Config, Session_Queue, Max_Retries, -1).\n+\n+-spec create_connection(config_type(), {Ip::string(), Port::pos_integer()}, Session_Id::pid())\n+                       -> {boolean() | pending, {session_queue_name(), Idle_Count, Max_Count}}\n+                              when Idle_Count :: max_sessions() | ets_buffer:buffer_error(),\n+                                   Max_Count  :: max_sessions() | ets_buffer:buffer_error().\n+%% @doc Create an audit ets_entry, then do checkin_connection.\n+create_connection(Config, {_Ip, _Port} = Node, Session_Id)\n+  when is_pid(Session_Id) ->\n+    Audit_Name = elysium_config:audit_ets_name(Config),\n+    Audit_Key  = {?MODULE, Session_Id},\n+    true = ets:insert_new(Audit_Name, #audit_serial{session_id_key=Audit_Key, init_checkin=timestamp()}),\n+    checkin_connection(Config, Node, Session_Id, true).\n+                               \n+-spec checkin_connection(config_type(), {Ip::string(), Port::pos_integer()},\n+                         Session_Id::pid(), Is_New_Connection::boolean())\n+                        -> {boolean() | pending, {session_queue_name(), Idle_Count, Max_Count}}\n+                               when Idle_Count :: max_sessions() | ets_buffer:buffer_error(),\n+                                    Max_Count  :: max_sessions() | ets_buffer:buffer_error().\n+%% @doc\n+%%   Checkin a seestar_session, IF there are no pending requests.\n+%%   A checkin will continue looping on the pending queue with\n+%%   the chance for decay on each attempt. If it decays, any\n+%%   newly spawned replacement is expected to check the pending\n+%%   queue for outstanding requests. Brand new connections are\n+%%   not checked for decay before first use.\n+%%\n+%%   This function can loop forever if there are pending requests,\n+%%   so it performs an asynchronous send_event.\n+%% @end\n+checkin_connection(Config, {_Ip, _Port} = Node, Session_Id, Is_New_Connection)\n+  when is_pid(Session_Id) ->\n+    Pending_Queue = elysium_config:requests_queue_name(Config),\n+    case is_process_alive(Session_Id) andalso ets_buffer:num_entries_dedicated(Pending_Queue) > 0 of\n+        false -> checkin_immediate (Config, Node, Session_Id, Pending_Queue, Is_New_Connection);\n+        true  -> checkin_pending   (Config, Node, Session_Id, Pending_Queue, Is_New_Connection)\n+    end.\n+\n+-spec pend_request(config_type(), query_request()) -> any() | pend_request_error().\n+%%   Block the caller while the request is serially queued. When\n+%%   a session is avialable to run this pending request, the\n+%%   blocking recieve loop will unblock and a spawned process\n+%%   will execute the request, so that the caller can still\n+%%   timeout if the request takes too long.\n+%% @end\n+pend_request(Config, Query_Request) ->\n+    Sid_Reply_Ref = make_ref(),\n+    Start_Time    = os:timestamp(),\n+    Pending_Queue = elysium_config:requests_queue_name   (Config),\n+    Reply_Timeout = elysium_config:request_reply_timeout (Config),\n+    wait_for_session(Config, Pending_Queue, Sid_Reply_Ref, Start_Time, Query_Request, Reply_Timeout).\n+\n+\n+%%%-----------------------------------------------------------------------\n+%%% Internal support functions\n+%%%-----------------------------------------------------------------------\n+\n+idle_connections(Config) ->\n+    Session_Queue = elysium_config:session_queue_name (Config),\n+    Max_Sessions  = elysium_config:session_max_count  (Config),\n+    Buffer_Count  = ets_buffer:num_entries_dedicated (Session_Queue),\n+    {idle_connections, report_available_resources(Session_Queue, Buffer_Count, Max_Sessions)}.\n+    \n+pending_requests(Config) ->\n+    Pending_Queue = elysium_config:requests_queue_name   (Config),\n+    Reply_Timeout = elysium_config:request_reply_timeout (Config),\n+    Pending_Count = ets_buffer:num_entries_dedicated (Pending_Queue),\n+    {pending_requests, report_available_resources(Pending_Queue, Pending_Count, Reply_Timeout)}.\n+\n+report_available_resources(Queue_Name, {missing_ets_buffer, Queue_Name}, Max) ->\n+    {Queue_Name, {missing_ets_buffer, 0, Max}};\n+report_available_resources(Queue_Name, Num_Sessions, Max) ->\n+    {Queue_Name, Num_Sessions, Max}.\n+\n+%% Internal loop function to retry getting from the queue.\n+fetch_pid_from_queue(_Config, _Session_Queue, Max_Retries, Times_Tried)\n+  when Times_Tried >= Max_Retries ->\n+    none_available;\n+fetch_pid_from_queue( Config,  Session_Queue, Max_Retries, Times_Tried) ->\n+    case ets_buffer:read_dedicated(Session_Queue) of\n+\n+        %% Race condition with checkin, try again...\n+        %% (When this happens, a connection is left behind in the queue and will never get reused!)\n+        {missing_ets_data, Session_Queue, Read_Loc} ->\n+            _ = audit_count(Config, session_missing_data),\n+            lager:error(\"Missing ETS data reading ~p at location ~p~n\", [Session_Queue, Read_Loc]),\n+            fetch_pid_from_queue(Config, Session_Queue, Max_Retries, Times_Tried+1);\n+\n+        %% Give up if there are no connections available...\n+        [] -> none_available;\n+\n+        %% Return only a live pid, otherwise get the next one.\n+        [{_Node, Session_Id} = Session_Data] when is_pid(Session_Id) ->\n+            case is_process_alive(Session_Id) of\n+                %% NOTE: we toss only MAX_CHECKOUT_RETRY dead pids\n+                false -> _ = audit_count(Config, session_dead),\n+                         fetch_pid_from_queue(Config, Session_Queue, Max_Retries, Times_Tried+1);\n+                true  -> _ = audit_data_checkout(Config, Session_Id),\n+                         Session_Data\n+            end;\n+\n+        %% Somehow the connection buffer died, or something even worse!\n+        Error ->\n+            audit_count(Config, session_ets_errors),\n+            lager:error(\"Connection buffer error: ~9999p~n\", [Error]),\n+            Error\n+    end.\n+\n+wait_for_session(Config, Pending_Queue, Sid_Reply_Ref, Start_Time, Query_Request, Reply_Timeout) ->\n+    Session_Queue = elysium_config:session_queue_name(Config),\n+    case fetch_pid_from_queue(Config, Session_Queue, 1, 0) of\n+\n+        %% A free connection showed up since we first checked...\n+        {Node, Session_Id} when is_pid(Session_Id) ->\n+            case is_process_alive(Session_Id) of\n+                true  -> handle_pending_request(Config, 0, Reply_Timeout, Node, Session_Id, Query_Request);\n+                false -> _ = audit_count(Config, session_dead),\n+                         wait_for_session(Config, Pending_Queue, Sid_Reply_Ref,\n+                                          Start_Time, Query_Request, Reply_Timeout)\n+            end;\n+\n+        %% None are still available, queue the request and wait for one to free up.\n+        _None_Available ->\n+            _Pending_Count = ets_buffer:write_dedicated(Pending_Queue, {{self(), Sid_Reply_Ref}, Start_Time}),\n+            wait_for_session_loop(Config, Pending_Queue, Sid_Reply_Ref, Start_Time, Query_Request, Reply_Timeout)\n+    end.\n+\n+wait_for_session_loop(Config, Pending_Queue, Sid_Reply_Ref, Start_Time, Query_Request, Reply_Timeout) ->\n+    receive\n+        %% A live elysium session channel is now available to make the request...\n+        {sid, Sid_Reply_Ref, Node, Session_Id, Pending_Queue, Is_New_Connection} ->\n+            Elapsed_Time = timer:now_diff(os:timestamp(), Start_Time),\n+            case {Elapsed_Time >= Reply_Timeout * 1000, is_process_alive(Session_Id)} of\n+\n+                %% Alas, we timed out waiting...\n+                {true,  true}  -> _ = audit_count(Config, session_timeouts),\n+                                  _ = checkin_connection(Config, Node, Session_Id, Is_New_Connection),\n+                                  {wait_for_session_timeout, Reply_Timeout};\n+                {true,  false} -> _ = audit_count(Config, session_dead),\n+                                  _ = audit_count(Config, session_timeouts),\n+                                  {wait_for_session_timeout, Reply_Timeout};\n+\n+                %% Dead session, loop waiting for another (hopefully live) connection to free up...\n+                {false, false} -> _ = audit_count(Config, session_dead),\n+                                  New_Timeout = Reply_Timeout - (Elapsed_Time div 1000),\n+                                  wait_for_session(Config, Pending_Queue, Sid_Reply_Ref,\n+                                                   Start_Time, Query_Request, New_Timeout);\n+\n+                %% Get some results while we still have time!\n+                {false, true}  -> handle_pending_request(Config, Elapsed_Time, Reply_Timeout,\n+                                                         Node, Session_Id, Query_Request)\n+            end;\n+\n+        %% Previous timed out request sent a Session_Id late, check it in and wait for our expected one.\n+        {sid, _, Node, Session_Id, Pending_Queue, Is_New_Connection} ->\n+            _ = audit_count(Config, session_wrong),\n+            _ = case is_process_alive(Session_Id) of\n+                    false -> audit_count(Config, session_dead);\n+                    true  -> checkin_immediate(Config, Node, Session_Id, Pending_Queue, Is_New_Connection)\n+                end,\n+            Elapsed_Time = timer:now_diff(os:timestamp(), Start_Time),\n+            case Elapsed_Time >= Reply_Timeout * 1000 of\n+                true -> _ = audit_count(Config, session_timeouts),\n+                        {wait_for_session_timeout, Reply_Timeout};\n+                false -> New_Timeout = Reply_Timeout - (Elapsed_Time div 1000),\n+                         wait_for_session_loop(Config, Pending_Queue, Sid_Reply_Ref, Start_Time,\n+                                               Query_Request, New_Timeout)\n+            end\n+\n+        %% Any other messages are intended for the blocked caller, leave them in the message queue.\n+\n+    after Reply_Timeout ->\n+            %% Handle race condition messaging vs timeout waiting for message.\n+            erlang:yield(),\n+            _ = receive\n+                    {sid, Sid_Reply_Ref, Node, Session_Id, Pending_Queue, Is_New_Connection} ->\n+                        case is_process_alive(Session_Id) of\n+                            false -> audit_count(Config, session_dead);\n+                            true  -> checkin_connection(Config, Node, Session_Id, Is_New_Connection)\n+                        end\n+                after 0 -> no_msgs\n+                end,\n+            _ = audit_count(Config, session_timeouts),\n+            {wait_for_session_timeout, Reply_Timeout}\n+    end.\n+\n+%% Use the Session_Id to run the query if we aren't out of time\n+handle_pending_request(Config, Elapsed_Time, Reply_Timeout, Node, Session_Id, Query_Request) ->\n+    %% Self cannot be executed inside the fun(), it needs to be set in the current context.\n+    Self = self(),\n+    Worker_Reply_Ref = make_ref(),\n+    %% Avoiding export of exec_pending_request/5\n+    Worker_Fun = fun() -> exec_pending_request(Worker_Reply_Ref, Self, Node, Session_Id, Query_Request) end,\n+    {Worker_Pid, Worker_Monitor_Ref} = spawn_opt(Worker_Fun, [monitor]),   % May want to add other options\n+    Timeout_Remaining = Reply_Timeout - (Elapsed_Time div 1000),\n+    try   receive_worker_reply(Config, Worker_Reply_Ref, Timeout_Remaining, Worker_Pid, Worker_Monitor_Ref)\n+    after erlang:demonitor(Worker_Monitor_Ref, [flush])\n+    end.\n+\n+%% Worker_Pid is passed to allow tracing\n+receive_worker_reply(Config, Worker_Reply_Ref, Timeout_Remaining, Worker_Pid, Worker_Monitor_Ref) ->\n+    receive\n+        {wrr, Worker_Reply_Ref, Reply} -> Reply;\n+        {'DOWN', Worker_Monitor_Ref, process, Worker_Pid, Reason} ->\n+            _ = audit_count(Config, worker_errors),\n+            {worker_reply_error, Reason}\n+    after Timeout_Remaining ->\n+            _ = audit_count(Config, worker_timeouts),\n+            {worker_reply_timeout, Timeout_Remaining}\n+    end.\n+\n+-spec checkin_immediate(config_type(), {Ip::string(), Port::pos_integer()},\n+                        Session_Id::pid(), Pending_Queue::requests_queue_name(), Is_New_Connection::boolean())\n+                       -> {boolean(), {session_queue_name(), Idle_Count, Max_Count}}\n+                              when Idle_Count :: max_sessions() | ets_buffer:buffer_error(),\n+                                   Max_Count  :: max_sessions() | ets_buffer:buffer_error().\n+%% @doc\n+%%   Checkin a seestar_session by putting it at the end of the\n+%%   available connection queue. Returns whether the checkin was\n+%%   successful (it fails if the process is dead when checkin is\n+%%   attempted), and how many connections are available after the\n+%%   checkin.\n+%%\n+%%   Sessions have a fixed probability of failure on checkin.\n+%%   The decay probability is a number of chances of dying per\n+%%   1 Million checkin attempts. If the session is killed, it\n+%%   will be replaced by the supervisor automatically spawning\n+%%   a new worker and placing it at the end of the queue.\n+%%\n+%%   The configuration parameter is not validated because this\n+%%   function should be a hotspot and we don't want it to slow\n+%%   down or become a concurrency bottleneck.\n+%% @end\n+checkin_immediate(Config, Node, Session_Id, Pending_Queue, true) ->\n+    Session_Queue = elysium_config:session_queue_name (Config),\n+    Max_Sessions  = elysium_config:session_max_count  (Config),\n+    case is_process_alive(Session_Id) of\n+        false -> _ = audit_count(Config, session_dead),\n+                 fail_checkin(Session_Queue, Max_Sessions, {Node, Session_Id}, Config);\n+        true  -> succ_checkin(Session_Queue, Max_Sessions, {Node, Session_Id}, Config, Pending_Queue, true)\n+    end;\n+checkin_immediate(Config, Node, Session_Id, Pending_Queue, false) ->\n+    Session_Queue = elysium_config:session_queue_name (Config),\n+    Max_Sessions  = elysium_config:session_max_count  (Config),\n+    case is_process_alive(Session_Id) of\n+        false -> _ = audit_count(Config, session_dead),\n+                 fail_checkin(Session_Queue, Max_Sessions, {Node, Session_Id}, Config);\n+        true  -> case decay_causes_death(Config, Session_Id) of\n+                     false -> succ_checkin(Session_Queue, Max_Sessions, {Node, Session_Id},\n+                                           Config, Pending_Queue, false);\n+                     true  -> _ = decay_session(Config, Session_Id),\n+                              fail_checkin(Session_Queue, Max_Sessions, {Node, Session_Id}, Config)\n+                  end\n+    end.\n+\n+%% Session_Data is passed to allow tracing\n+fail_checkin(Session_Queue, Max_Sessions, {_Node, Session_Id}, Config) ->\n+    _ = audit_data_delete(Config, Session_Id),\n+    Available = ets_buffer:num_entries_dedicated(Session_Queue),\n+    {false, report_available_resources(Session_Queue, Available, Max_Sessions)}.\n+\n+succ_checkin(Session_Queue, Max_Sessions, {Node, Session_Id} = Session_Data,\n+             Config, Pending_Queue, Is_New_Connection) ->\n+    case ets_buffer:num_entries_dedicated(Pending_Queue) > 0 of\n+        true  -> checkin_pending(Config, Node, Session_Id, Pending_Queue, Is_New_Connection);\n+        false -> Available  = checkin_session(Session_Queue, Session_Data),\n+                 _ = audit_data_checkin(Config, Session_Id),\n+                 {true, report_available_resources(Session_Queue, Available, Max_Sessions)}\n+    end.\n+\n+checkin_session(Session_Queue, Session_Data) ->\n+    elysium_session_enqueuer:checkin_session(Session_Queue, Session_Data).\n+\n+delay_checkin(Config) ->\n+    Session_Queue = elysium_config:session_queue_name  (Config),\n+    Max_Sessions  = elysium_config:session_max_count   (Config),\n+    Available     = ets_buffer:num_entries_dedicated (Session_Queue),\n+    {pending, report_available_resources(Session_Queue, Available, Max_Sessions)}.\n+\n+decay_causes_death(Config, _Session_Id) ->\n+    case elysium_config:decay_probability(Config) of\n+        Never_Decays when is_integer(Never_Decays), Never_Decays =:= 0 ->\n+            false;\n+        Probability  when is_integer(Probability),  Probability   >  0, Probability =< 1000000000 ->\n+            R = elysium_random:random_int_up_to(1000000000),\n+            R =< Probability\n+    end.\n+\n+decay_session(Config, Session_Id) ->\n+    Supervisor_Pid = elysium_queue:get_connection_supervisor(),\n+    _ = case elysium_connection_sup:stop_child(Supervisor_Pid, Session_Id) of\n+            {error, not_found} -> dont_replace_child;\n+            ok -> elysium_connection_sup:start_child(Supervisor_Pid, [Config])\n+        end,\n+    _ = audit_count(Config, session_decay),\n+    audit_data_delete(Config, Session_Id).\n+\n+checkin_pending(Config, Node, Sid, Pending_Queue, Is_New_Connection) ->\n+    case ets_buffer:read_dedicated(Pending_Queue) of\n+\n+        %% Race condition with pend_request, try again...\n+        %% (When this happens, a pending request is left behind in the queue and will timeout)\n+        {missing_ets_data, Pending_Queue, Read_Loc} ->\n+            _ = audit_count(Config, pending_missing_data),\n+            lager:error(\"Missing ETS data reading ~p at location ~p~n\", [Pending_Queue, Read_Loc]),\n+            checkin_pending(Config, Node, Sid, Pending_Queue, Is_New_Connection);\n+\n+        %% There are no pending requests, return the session...\n+        [] -> checkin_immediate(Config, Node, Sid, Pending_Queue, Is_New_Connection);\n+\n+        %% Got a pending request, let's run it...\n+        [{{Waiting_Pid, Sid_Reply_Ref}, When_Originally_Queued}] ->\n+\n+            Reply_Timeout = elysium_config:request_reply_timeout(Config),\n+            case timer:now_diff(os:timestamp(), When_Originally_Queued) of\n+\n+                %% Too much time has passed, skip this request and try another...\n+                Expired when Expired > Reply_Timeout * 1000 ->\n+                    _ = audit_count(Config, pending_timeouts),\n+                    checkin_pending(Config, Node, Sid, Pending_Queue, Is_New_Connection);\n+\n+                %% There's still time to reply, run the request if the session is still alive.\n+                _Remaining_Time ->\n+                    case is_process_alive(Waiting_Pid) of\n+                        false -> _ = audit_count(Config, pending_dead),\n+                                 checkin_pending(Config, Node, Sid, Pending_Queue, Is_New_Connection);\n+                        true  -> Waiting_Pid ! {sid, Sid_Reply_Ref, Node, Sid, Pending_Queue, Is_New_Connection},\n+                                 _ = audit_data_pending(Config, Sid),\n+                                 delay_checkin(Config)\n+                    end\n+            end;\n+\n+        %% Somehow the pending buffer died, or something even worse!\n+        Error ->\n+            _ = audit_count(Config, pending_ets_errors),\n+            lager:error(\"Pending requests buffer error: ~9999p~n\", [Error]),\n+            Error\n+    end.\n+\n+%% Watch Out! This function swaps from the Config on a checkin request to the\n+%% Config on the original pending query. If you somehow mix connection queues\n+%% by passing different Configs, the clusters which queries run on may get\n+%% mixed up resulting in queries/updates/deletes talking to the wrong clusters.\n+exec_pending_request(Reply_Ref, Reply_Pid, Node, Sid, {bare_fun, Config, Query_Fun, Args, Consistency}) ->\n+    try   Reply = Query_Fun(Sid, Args, Consistency),\n+          Reply_Pid ! {wrr, Reply_Ref, Reply}\n+    catch A:B -> lager:error(\"Query execution caught ~p:~p for ~p ~p ~9999p~n\",\n+                             [A,B, Reply_Pid, Args, erlang:get_stacktrace()])\n+    after _ = checkin_connection(Config, Node, Sid, false)\n+    end;\n+exec_pending_request(Reply_Ref, Reply_Pid, Node, Sid, {mod_fun,  Config, Mod,  Fun, Args, Consistency}) ->\n+    try   Reply = Mod:Fun(Sid, Args, Consistency),\n+          Reply_Pid ! {wrr, Reply_Ref, Reply}\n+    catch A:B -> lager:error(\"Query execution caught ~p:~p for ~p ~p ~9999p~n\",\n+                             [A,B, Reply_Pid, Args, erlang:get_stacktrace()])\n+    after _ = checkin_connection(Config, Node, Sid, false)\n+    end.\n+\n+%% Internal auditing functions\n+audit_count(Config, Type) ->\n+    Audit_Key   = {?MODULE, counts},\n+    Audit_Name  = elysium_config:audit_ets_name(Config),\n+    Counter_Pos = case Type of\n+                      pending_dead          -> #audit_serial_counts.pending_dead;\n+                      pending_ets_errors    -> #audit_serial_counts.pending_ets_errors;\n+                      pending_missing_data  -> #audit_serial_counts.pending_missing_data;\n+                      pending_timeouts      -> #audit_serial_counts.pending_timeouts;\n+                      session_dead          -> #audit_serial_counts.session_dead;\n+                      session_decay         -> #audit_serial_counts.session_decay;\n+                      session_ets_errors    -> #audit_serial_counts.session_ets_errors;","path":"src/elysium_bs_parallel.erl","position":492,"original_position":492,"commit_id":"37cca477d0beb04f75c5062b03155b11ac33fbdd","original_commit_id":"37cca477d0beb04f75c5062b03155b11ac33fbdd","user":{"login":"elvisinaka","id":8136322,"avatar_url":"https://avatars.githubusercontent.com/u/8136322?v=3","gravatar_id":"","url":"https://api.github.com/users/elvisinaka","html_url":"https://github.com/elvisinaka","followers_url":"https://api.github.com/users/elvisinaka/followers","following_url":"https://api.github.com/users/elvisinaka/following{/other_user}","gists_url":"https://api.github.com/users/elvisinaka/gists{/gist_id}","starred_url":"https://api.github.com/users/elvisinaka/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/elvisinaka/subscriptions","organizations_url":"https://api.github.com/users/elvisinaka/orgs","repos_url":"https://api.github.com/users/elvisinaka/repos","events_url":"https://api.github.com/users/elvisinaka/events{/privacy}","received_events_url":"https://api.github.com/users/elvisinaka/received_events","type":"User","site_admin":false},"body":"Line 492 is too long: \"                      session_ets_errors    -> #audit_serial_counts.session_ets_errors;\".","created_at":"2014-11-13T18:27:12Z","updated_at":"2014-11-13T18:27:12Z","html_url":"https://github.com/inaka/elysium/pull/1#discussion_r20311312","pull_request_url":"https://api.github.com/repos/inaka/elysium/pulls/1","_links":{"self":{"href":"https://api.github.com/repos/inaka/elysium/pulls/comments/20311312"},"html":{"href":"https://github.com/inaka/elysium/pull/1#discussion_r20311312"},"pull_request":{"href":"https://api.github.com/repos/inaka/elysium/pulls/1"}}},"pull_request":{"url":"https://api.github.com/repos/inaka/elysium/pulls/1","id":24389247,"html_url":"https://github.com/inaka/elysium/pull/1","diff_url":"https://github.com/inaka/elysium/pull/1.diff","patch_url":"https://github.com/inaka/elysium/pull/1.patch","issue_url":"https://api.github.com/repos/inaka/elysium/issues/1","number":1,"state":"open","locked":false,"title":"Up to date","user":{"login":"HernanRivasAcosta","id":5691671,"avatar_url":"https://avatars.githubusercontent.com/u/5691671?v=3","gravatar_id":"","url":"https://api.github.com/users/HernanRivasAcosta","html_url":"https://github.com/HernanRivasAcosta","followers_url":"https://api.github.com/users/HernanRivasAcosta/followers","following_url":"https://api.github.com/users/HernanRivasAcosta/following{/other_user}","gists_url":"https://api.github.com/users/HernanRivasAcosta/gists{/gist_id}","starred_url":"https://api.github.com/users/HernanRivasAcosta/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/HernanRivasAcosta/subscriptions","organizations_url":"https://api.github.com/users/HernanRivasAcosta/orgs","repos_url":"https://api.github.com/users/HernanRivasAcosta/repos","events_url":"https://api.github.com/users/HernanRivasAcosta/events{/privacy}","received_events_url":"https://api.github.com/users/HernanRivasAcosta/received_events","type":"User","site_admin":false},"body":"","created_at":"2014-11-13T18:26:57Z","updated_at":"2014-11-13T18:27:12Z","closed_at":null,"merged_at":null,"merge_commit_sha":"a72986d87e4cefc1039607628657465b2ad79a70","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/inaka/elysium/pulls/1/commits","review_comments_url":"https://api.github.com/repos/inaka/elysium/pulls/1/comments","review_comment_url":"https://api.github.com/repos/inaka/elysium/pulls/comments/{number}","comments_url":"https://api.github.com/repos/inaka/elysium/issues/1/comments","statuses_url":"https://api.github.com/repos/inaka/elysium/statuses/37cca477d0beb04f75c5062b03155b11ac33fbdd","head":{"label":"tigertext:master","ref":"master","sha":"37cca477d0beb04f75c5062b03155b11ac33fbdd","user":{"login":"tigertext","id":173280,"avatar_url":"https://avatars.githubusercontent.com/u/173280?v=3","gravatar_id":"","url":"https://api.github.com/users/tigertext","html_url":"https://github.com/tigertext","followers_url":"https://api.github.com/users/tigertext/followers","following_url":"https://api.github.com/users/tigertext/following{/other_user}","gists_url":"https://api.github.com/users/tigertext/gists{/gist_id}","starred_url":"https://api.github.com/users/tigertext/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tigertext/subscriptions","organizations_url":"https://api.github.com/users/tigertext/orgs","repos_url":"https://api.github.com/users/tigertext/repos","events_url":"https://api.github.com/users/tigertext/events{/privacy}","received_events_url":"https://api.github.com/users/tigertext/received_events","type":"Organization","site_admin":false},"repo":{"id":24917017,"name":"elysium","full_name":"tigertext/elysium","owner":{"login":"tigertext","id":173280,"avatar_url":"https://avatars.githubusercontent.com/u/173280?v=3","gravatar_id":"","url":"https://api.github.com/users/tigertext","html_url":"https://github.com/tigertext","followers_url":"https://api.github.com/users/tigertext/followers","following_url":"https://api.github.com/users/tigertext/following{/other_user}","gists_url":"https://api.github.com/users/tigertext/gists{/gist_id}","starred_url":"https://api.github.com/users/tigertext/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tigertext/subscriptions","organizations_url":"https://api.github.com/users/tigertext/orgs","repos_url":"https://api.github.com/users/tigertext/repos","events_url":"https://api.github.com/users/tigertext/events{/privacy}","received_events_url":"https://api.github.com/users/tigertext/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/tigertext/elysium","description":"The blessed and happy erlang afterlife for Cassandra","fork":true,"url":"https://api.github.com/repos/tigertext/elysium","forks_url":"https://api.github.com/repos/tigertext/elysium/forks","keys_url":"https://api.github.com/repos/tigertext/elysium/keys{/key_id}","collaborators_url":"https://api.github.com/repos/tigertext/elysium/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/tigertext/elysium/teams","hooks_url":"https://api.github.com/repos/tigertext/elysium/hooks","issue_events_url":"https://api.github.com/repos/tigertext/elysium/issues/events{/number}","events_url":"https://api.github.com/repos/tigertext/elysium/events","assignees_url":"https://api.github.com/repos/tigertext/elysium/assignees{/user}","branches_url":"https://api.github.com/repos/tigertext/elysium/branches{/branch}","tags_url":"https://api.github.com/repos/tigertext/elysium/tags","blobs_url":"https://api.github.com/repos/tigertext/elysium/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/tigertext/elysium/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/tigertext/elysium/git/refs{/sha}","trees_url":"https://api.github.com/repos/tigertext/elysium/git/trees{/sha}","statuses_url":"https://api.github.com/repos/tigertext/elysium/statuses/{sha}","languages_url":"https://api.github.com/repos/tigertext/elysium/languages","stargazers_url":"https://api.github.com/repos/tigertext/elysium/stargazers","contributors_url":"https://api.github.com/repos/tigertext/elysium/contributors","subscribers_url":"https://api.github.com/repos/tigertext/elysium/subscribers","subscription_url":"https://api.github.com/repos/tigertext/elysium/subscription","commits_url":"https://api.github.com/repos/tigertext/elysium/commits{/sha}","git_commits_url":"https://api.github.com/repos/tigertext/elysium/git/commits{/sha}","comments_url":"https://api.github.com/repos/tigertext/elysium/comments{/number}","issue_comment_url":"https://api.github.com/repos/tigertext/elysium/issues/comments/{number}","contents_url":"https://api.github.com/repos/tigertext/elysium/contents/{+path}","compare_url":"https://api.github.com/repos/tigertext/elysium/compare/{base}...{head}","merges_url":"https://api.github.com/repos/tigertext/elysium/merges","archive_url":"https://api.github.com/repos/tigertext/elysium/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/tigertext/elysium/downloads","issues_url":"https://api.github.com/repos/tigertext/elysium/issues{/number}","pulls_url":"https://api.github.com/repos/tigertext/elysium/pulls{/number}","milestones_url":"https://api.github.com/repos/tigertext/elysium/milestones{/number}","notifications_url":"https://api.github.com/repos/tigertext/elysium/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/tigertext/elysium/labels{/name}","releases_url":"https://api.github.com/repos/tigertext/elysium/releases{/id}","created_at":"2014-10-08T00:02:02Z","updated_at":"2014-10-15T21:47:30Z","pushed_at":"2014-11-13T06:07:18Z","git_url":"git://github.com/tigertext/elysium.git","ssh_url":"git@github.com:tigertext/elysium.git","clone_url":"https://github.com/tigertext/elysium.git","svn_url":"https://github.com/tigertext/elysium","homepage":null,"size":675,"stargazers_count":0,"watchers_count":0,"language":"Erlang","has_issues":false,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":2,"mirror_url":null,"open_issues_count":0,"forks":2,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"inaka:master","ref":"master","sha":"cdd4be536f0a749ee4b48d6e5d9036ab33c39e28","user":{"login":"inaka","id":867053,"avatar_url":"https://avatars.githubusercontent.com/u/867053?v=3","gravatar_id":"","url":"https://api.github.com/users/inaka","html_url":"https://github.com/inaka","followers_url":"https://api.github.com/users/inaka/followers","following_url":"https://api.github.com/users/inaka/following{/other_user}","gists_url":"https://api.github.com/users/inaka/gists{/gist_id}","starred_url":"https://api.github.com/users/inaka/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/inaka/subscriptions","organizations_url":"https://api.github.com/users/inaka/orgs","repos_url":"https://api.github.com/users/inaka/repos","events_url":"https://api.github.com/users/inaka/events{/privacy}","received_events_url":"https://api.github.com/users/inaka/received_events","type":"Organization","site_admin":false},"repo":{"id":25830237,"name":"elysium","full_name":"inaka/elysium","owner":{"login":"inaka","id":867053,"avatar_url":"https://avatars.githubusercontent.com/u/867053?v=3","gravatar_id":"","url":"https://api.github.com/users/inaka","html_url":"https://github.com/inaka","followers_url":"https://api.github.com/users/inaka/followers","following_url":"https://api.github.com/users/inaka/following{/other_user}","gists_url":"https://api.github.com/users/inaka/gists{/gist_id}","starred_url":"https://api.github.com/users/inaka/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/inaka/subscriptions","organizations_url":"https://api.github.com/users/inaka/orgs","repos_url":"https://api.github.com/users/inaka/repos","events_url":"https://api.github.com/users/inaka/events{/privacy}","received_events_url":"https://api.github.com/users/inaka/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/inaka/elysium","description":"The blessed and happy erlang afterlife for Cassandra","fork":true,"url":"https://api.github.com/repos/inaka/elysium","forks_url":"https://api.github.com/repos/inaka/elysium/forks","keys_url":"https://api.github.com/repos/inaka/elysium/keys{/key_id}","collaborators_url":"https://api.github.com/repos/inaka/elysium/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/inaka/elysium/teams","hooks_url":"https://api.github.com/repos/inaka/elysium/hooks","issue_events_url":"https://api.github.com/repos/inaka/elysium/issues/events{/number}","events_url":"https://api.github.com/repos/inaka/elysium/events","assignees_url":"https://api.github.com/repos/inaka/elysium/assignees{/user}","branches_url":"https://api.github.com/repos/inaka/elysium/branches{/branch}","tags_url":"https://api.github.com/repos/inaka/elysium/tags","blobs_url":"https://api.github.com/repos/inaka/elysium/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/inaka/elysium/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/inaka/elysium/git/refs{/sha}","trees_url":"https://api.github.com/repos/inaka/elysium/git/trees{/sha}","statuses_url":"https://api.github.com/repos/inaka/elysium/statuses/{sha}","languages_url":"https://api.github.com/repos/inaka/elysium/languages","stargazers_url":"https://api.github.com/repos/inaka/elysium/stargazers","contributors_url":"https://api.github.com/repos/inaka/elysium/contributors","subscribers_url":"https://api.github.com/repos/inaka/elysium/subscribers","subscription_url":"https://api.github.com/repos/inaka/elysium/subscription","commits_url":"https://api.github.com/repos/inaka/elysium/commits{/sha}","git_commits_url":"https://api.github.com/repos/inaka/elysium/git/commits{/sha}","comments_url":"https://api.github.com/repos/inaka/elysium/comments{/number}","issue_comment_url":"https://api.github.com/repos/inaka/elysium/issues/comments/{number}","contents_url":"https://api.github.com/repos/inaka/elysium/contents/{+path}","compare_url":"https://api.github.com/repos/inaka/elysium/compare/{base}...{head}","merges_url":"https://api.github.com/repos/inaka/elysium/merges","archive_url":"https://api.github.com/repos/inaka/elysium/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/inaka/elysium/downloads","issues_url":"https://api.github.com/repos/inaka/elysium/issues{/number}","pulls_url":"https://api.github.com/repos/inaka/elysium/pulls{/number}","milestones_url":"https://api.github.com/repos/inaka/elysium/milestones{/number}","notifications_url":"https://api.github.com/repos/inaka/elysium/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/inaka/elysium/labels{/name}","releases_url":"https://api.github.com/repos/inaka/elysium/releases{/id}","created_at":"2014-10-27T17:16:14Z","updated_at":"2014-10-15T21:47:30Z","pushed_at":"2014-10-27T17:16:47Z","git_url":"git://github.com/inaka/elysium.git","ssh_url":"git@github.com:inaka/elysium.git","clone_url":"https://github.com/inaka/elysium.git","svn_url":"https://github.com/inaka/elysium","homepage":null,"size":169,"stargazers_count":0,"watchers_count":0,"language":null,"has_issues":false,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":1,"forks":0,"open_issues":1,"watchers":0,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/inaka/elysium/pulls/1"},"html":{"href":"https://github.com/inaka/elysium/pull/1"},"issue":{"href":"https://api.github.com/repos/inaka/elysium/issues/1"},"comments":{"href":"https://api.github.com/repos/inaka/elysium/issues/1/comments"},"review_comments":{"href":"https://api.github.com/repos/inaka/elysium/pulls/1/comments"},"review_comment":{"href":"https://api.github.com/repos/inaka/elysium/pulls/comments/{number}"},"commits":{"href":"https://api.github.com/repos/inaka/elysium/pulls/1/commits"},"statuses":{"href":"https://api.github.com/repos/inaka/elysium/statuses/37cca477d0beb04f75c5062b03155b11ac33fbdd"}}}}