Merge pull request #20 from duomark/master

Introduce elysium_buffering_strategy behaviour

Author: georgeye

Makefile

@@ -13,6 +13,7 @@ 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}" ${PLATFORM_OPTS}
+COMPILE_FIRST := elysium_buffering_audit elysium_buffering_strategy
 
 # Needed for testing
 TEST_ERLC_OPTS := -I include $(ERLC_OPTS)

include/elysium_audit_types.hrl

@@ -0,0 +1,19 @@
+-type audit_counts_ets_name()   :: atom().
+-type audit_connection_key()    :: {buffering_strategy_module(), connection_id()}.
+-type audit_std_counts_key()    :: {buffering_strategy_module(), counts}.
+-type audit_custom_counts_key() :: {buffering_strategy_module(), custom_counts}.
+-type audit_counts_key()        :: audit_std_counts_key() | audit_custom_counts_key().
+
+-type audit_timestamp()         :: binary().
+-type audit_count()             :: non_neg_integer().
+
+-type audit_custom_counts()     :: any().
+-type audit_custom_event()      :: atom().
+-type audit_count_event()       :: pending_dead
+                                 | pending_timeouts
+                                 | session_dead
+                                 | session_decay
+                                 | session_timeouts
+                                 | session_wrong
+                                 | worker_errors
+                                 | worker_timeouts.

include/elysium_types.hrl

@@ -1,22 +1,68 @@
+%% Two ways of specifying configuration parameters, designed for high concurrency.
 -type config_type() :: {config_mod, module()}
                      | {vbisect,    vbisect:bindict()}.
 
--type lb_queue_name()       :: ets_buffer:buffer_name().
--type audit_ets_name()      :: ets_buffer:buffer_name().
--type session_queue_name()  :: ets_buffer:buffer_name().
--type requests_queue_name() :: ets_buffer:buffer_name().
--type host_list()           :: [{Ip_Addr::string(), Port::pos_integer()}].
--type timeout_in_ms()       :: pos_integer().      %% in milliseconds
--type max_sessions()        :: pos_integer().
--type max_retries()         :: non_neg_integer().
--type decay_prob()          :: non_neg_integer().  %% number of chances in 1M of death
+%% Buffering module names are used for telemetry.
+-type buffering_strategy_module() :: module().
 
+%% There are three queues used by elsyium.
+-type queue_name()             :: ets_buffer:buffer_name() | atom().
+-type lb_queue_name()          :: queue_name().    % Cassandra node load balancer queue
+-type connection_queue_name()  :: queue_name().    % Cassandra open connections queue
+-type requests_queue_name()    :: queue_name().    % Pending requests queue
+
+-type audit_ets_name()         :: atom().          % Audit ets table for telemetry data
+
+-type cassandra_node()   :: {Ip_Or_Hostname::inet:hostname(), Port::inet:port_number()}.   % Cassandra node
+-type host_list()        :: [cassandra_node()].    % Host list used by load balancer
+
+-type timeout_in_ms()    :: pos_integer().         % In milliseconds
+-type max_retries()      :: non_neg_integer().
+-type decay_prob()       :: non_neg_integer().     % Chances in 1B of connection death
+
+%% Currently Cassandra connections are seestar_sessions.
+-type connection_id() :: pid().                    % Live process holding socket to Cassandra.
+-type fun_request()   :: fun((connection_id(), [any()], seestar:consistency()) -> [any()]).
+-type query_request() :: {bare_fun, config_type(), fun_request(),    [any()], seestar:consistency()}
+                       | {mod_fun,  config_type(), module(), atom(), [any()], seestar:consistency()}.
+
+-type cassandra_connection()  :: {cassandra_node(), connection_id()}.
+-type connection_baton()      :: {{connection_id(), reference}, erlang:timestamp()}.
+
+-type pending_request_pid()   :: pid().
+-type pending_request_baton() :: {{pending_request_pid(), reference()}, erlang:timestamp()}.
+
+%% Connection count errors reported by ets_buffer, new buffering strategies may need new error types.
+-type connection_count_error() :: ets_buffer:buffer_error().
+-type max_connections()        :: pos_integer() | connection_count_error().
+-type pending_count()          :: pos_integer() | connection_count_error().
+
+-type idle_status()     :: {idle_connections, {connection_queue_name(),
+                                               Idle::max_connections(), Max::max_connections()}}.
+-type pending_status()  :: {pending_requests, {requests_queue_name(),
+                                               pending_count(), timeout_in_ms()}}.
+-type status_reply()    :: {status, {idle_status(), pending_status()}}.
+
+-type pending_checkin() :: {boolean() | pending, {connection_queue_name(),
+                                                  max_connections(), max_connections()}}.
+
+%% Errors when a pending request does not get a chance to return a query result.
 -type wait_for_session_error() :: {wait_for_session_timeout, pos_integer()}
                                 | {wait_for_session_error,   any()}.
 
+%% Errors when a query request fails after submitting to Cassandra.
 -type worker_reply_error()     :: {worker_reply_timeout,     pos_integer()}
                                 | {worker_reply_error,       any()}.
 
+%% Errors getting a connection to Cassandra.
+-type connection_error()       :: {error, no_db_connections}.
+
+%% All errors that a pending request may encounter.
 -type pend_request_error()     :: ets_buffer:buffer_error()
                                 | wait_for_session_error()
-                                | worker_reply_error().
+                                | worker_reply_error()
+                                | connection_error().
+
+%% Query_reply is a successful return; query_result contains all possible query returns.
+-type query_reply()  :: any().
+-type query_result() :: pend_request_error() | query_reply().

src/elysium.app.src

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