Send.timeout

Makefile

@@ -2,23 +2,25 @@ PROJECT = elysium
 
 DEPS = eper epocxy seestar vbisect
 dep_epocxy  = git https://github.com/duomark/epocxy 0.9.8
-dep_seestar = git https://github.com/iamaleksey/seestar master
+dep_seestar = git https://github.com/tigertext/seestar 0.1.1
 dep_vbisect = git https://github.com/jaynel/vbisect 0.1.2
 V = 0
 
 TEST_DEPS = proper test_commons
-dep_proper       = git https://github.com/th0114nd/proper master
+dep_proper       = git https://github.com/manopapad/proper master
 dep_test_commons = git https://github.com/tigertext/test_commons master
 
+PLATFORM_OPTS := `erlc -o ebin src/elysium_compile_utils.erl ; erl -noshell -pa ebin -s elysium_compile_utils platform_opts -s init stop`
 
-ERLC_OPTS := +debug_info +"{cover_enabled, true}"
+ERLC_OPTS := +debug_info +"{cover_enabled, true}" ${PLATFORM_OPTS}
 
 # Needed for testing
 TEST_ERLC_OPTS := -I include $(ERLC_OPTS)
 CT_OPTS := -cover test/elysium.coverspec
 # CT_SUITES := elysium_basic
 
 # DIALYZER_OPTS := -I include test/elysium -Werror_handling -Wrace_conditions -Wunmatched_returns
+DIALYZER_OPTS ?= -r deps/epocxy/src ${PLATFORM_OPTS} -Werror_handling -Wrace_conditions -Wunmatched_returns
 
 ## EDOC_DIRS := ["src", "examples"]
 ## EDOC_OPTS := {preprocess, true}, {source_path, ${EDOC_DIRS}}, nopackages, {subpackages, true}

src/elysium.app.src

@@ -1,7 +1,7 @@
 {application, elysium,
  [
   {description, "Elysium for Cassandra"},
-  {vsn, "0.1.6"},
+  {vsn, "0.1.7"},
   {id, "elysium"},
   {registered, [elysium_sup, elysium_connection_sup]},
   {applications, [kernel, stdlib]},

src/elysium_bs_serial.erl

@@ -158,9 +158,9 @@ fetch_pid_from_queue( Queue_Name, Max_Retries, Times_Tried) ->
         Error -> Error
     end.
 
-%% TODO: This receive loop needs to handle status queries and kill requests so it can be monitored.
 wait_for_session(Config, Pending_Request_Count, Sid_Reply_Ref, Start_Time, Query_Request, Reply_Timeout) ->
     receive
+        %% An elysium session channel is now available to make the request...
         {sid, Sid_Reply_Ref, Node, Session_Id, Pending_Queue} ->
             Elapsed_Time = timer:now_diff(os:timestamp(), Start_Time),
             case is_process_alive(Session_Id) of
@@ -171,9 +171,8 @@ wait_for_session(Config, Pending_Request_Count, Sid_Reply_Ref, Start_Time, Query
                 %% Handle, but there may be no time left to run the query...
                 true  -> handle_pending_request(Config, Elapsed_Time, Reply_Timeout,
                                                 Node, Session_Id, Query_Request)
-            end;
-        %% There is only one message we are expecting...
-        Other           -> {wait_for_session_error,   Other}
+            end
+        %% Any other messages are intended for the blocked caller, leave them in the message queue.
     after Reply_Timeout -> {wait_for_session_timeout, Reply_Timeout}
     end.
 
@@ -239,7 +238,7 @@ checkin_immediate(Config, Node, Session_Id) ->
         false -> fail_checkin(Queue_Name, Max_Sessions);
         true  -> case decay_causes_death(Config, Session_Id) of
                      false -> succ_checkin(Queue_Name, Max_Sessions, {Node, Session_Id});
-                     true  -> exit(Session_Id, kill),
+                     true  -> decay_session(Config, Session_Id),
                               fail_checkin(Queue_Name, Max_Sessions)
                   end
     end.
@@ -267,6 +266,12 @@ decay_causes_death(Config, _Session_Id) ->
             R =< Probability
     end.
 
+decay_session(Config, Session_Id) ->
+    Supervisor_Pid = elysium_queue:get_connection_supervisor(),
+    try   _ = elysium_connection_sup:stop_child  (Supervisor_Pid, Session_Id)
+    after _ = elysium_connection_sup:start_child (Supervisor_Pid, [Config])
+    end.
+
 checkin_pending(Config, Node, Sid, Pending_Queue) ->
     case ets_buffer:read_dedicated(Pending_Queue) of
         [] -> checkin_immediate(Config, Node, Sid);
@@ -291,12 +296,14 @@ checkin_pending(Config, Node, Sid, Pending_Queue) ->
 exec_pending_request(Reply_Ref, Reply_Pid, Node, Sid, {bare_fun, Config, Query_Fun, Args, Consistency}) ->
     try   Reply = Query_Fun(Sid, Args, Consistency),
           Reply_Pid ! {wrr, Reply_Ref, Reply}
-    catch A:B -> error_logger:error_msg("Query execution caught ~p:~p for ~p ~p", [A,B, Reply_Pid, Args])
+    catch A:B -> error_logger:error_msg("Query execution caught ~p:~p for ~p ~p ~9999p",
+                                        [A,B, Reply_Pid, Args, erlang:get_stacktrace()])
     after _ = checkin_connection(Config, Node, Sid)
     end;
 exec_pending_request(Reply_Ref, Reply_Pid, Node, Sid, {mod_fun,  Config, Mod,  Fun, Args, Consistency}) ->
     try   Reply = Mod:Fun(Sid, Args, Consistency),
           Reply_Pid ! {wrr, Reply_Ref, Reply}
-    catch A:B -> error_logger:error_msg("Query execution caught ~p:~p for ~p ~p", [A,B, Reply_Pid, Args])
+    catch A:B -> error_logger:error_msg("Query execution caught ~p:~p for ~p ~p ~9999p",
+                                        [A,B, Reply_Pid, Args, erlang:get_stacktrace()])
     after _ = checkin_connection(Config, Node, Sid)
     end.

src/elysium_buffer_sup.erl

@@ -68,21 +68,21 @@ init({Config}) ->
     %% Setup a load balancing FIFO Queue for all the Cassandra nodes to contact.
     Lb_Queue_Name = elysium_config:load_balancer_queue(Config),
     Lb_Queue_Name = ets_buffer:create_dedicated(Lb_Queue_Name, fifo),
-    lager:error("Creating Cassandra round-robin ets_buffer '~p' with the following nodes:~n", [Lb_Queue_Name]),
+    lager:info("Creating Cassandra round-robin ets_buffer '~p' with the following nodes:~n", [Lb_Queue_Name]),
     _ = [begin
              _ = ets_buffer:write_dedicated(Lb_Queue_Name, Node),
-             lager:error("   ~p~n", [Node])
+             lager:info("   ~p~n", [Node])
          end || {_Ip, _Port} = Node <- elysium_config:round_robin_hosts(Config),
                 is_list(_Ip), is_integer(_Port), _Port > 0],
 
     %% Create a FIFO Queue for the live sessions that are connected to Cassandra.
     Session_Queue_Name = elysium_config:session_queue_name(Config),
-    lager:error("Creating Cassandra session ets_buffer '~p'~n", [Session_Queue_Name]),
+    lager:info("Creating Cassandra session ets_buffer '~p'~n", [Session_Queue_Name]),
     Session_Queue_Name = ets_buffer:create_dedicated(Session_Queue_Name, fifo),
 
     %% Create a FIFO Queue for pending query requests.
     Pending_Queue_Name = elysium_config:requests_queue_name(Config),
-    lager:error("Creating Cassandra pending_requests ets_buffer '~p'~n", [Pending_Queue_Name]),
+    lager:info("Creating Cassandra pending_requests ets_buffer '~p'~n", [Pending_Queue_Name]),
     Pending_Queue_Name = ets_buffer:create_dedicated(Pending_Queue_Name, fifo),
 
     {ok, {{one_for_one, 1, 10}, []}}.