query timeout

src/epgsql_pool.erl

@@ -46,16 +46,16 @@ query(Worker, Stmt, Params, Options) when is_pid(Worker) ->
                   undefined -> epgsql_pool_settings:get(query_timeout);
                   V -> V
               end,
-    error_logger:info_msg("Worker:~p Stmt:~p Params:~p", [Worker, Stmt, Params]), %% TEMP
-    %% TODO process timeout,
-    %% try-catch
-    %% send cancel
-    %% log error
-    %% reply to client with error
-    %% reconnect
-    %% return to pool
-    Res = gen_server:call(Worker, {query, Stmt, Params}, Timeout),
-    Res;
+    %% TStart = os:timestamp(),
+    %% Time = timer:now_diff(os:timestamp(), TStart),
+    try
+        gen_server:call(Worker, {equery, Stmt, Params}, Timeout)
+    catch
+        exit:{timeout, _} ->
+            gen_server:call(Worker, cancel),
+            error_logger:error_msg("query timeout ~p ~p", [Stmt, Params]),
+            {error, timeout}
+    end;
 
 query(PoolName, Stmt, Params, Options) ->
     case get_worker(PoolName) of
@@ -69,13 +69,13 @@ transaction(PoolName, Fun) ->
     case get_worker(PoolName) of
         {ok, Worker} ->
             try
-                gen_server:call(Worker, {query, "BEGIN", []}),
+                gen_server:call(Worker, {squery, "BEGIN"}),
                 Result = Fun(Worker),
-                gen_server:call(Worker, {query, "COMMIT", []}),
+                gen_server:call(Worker, {squery, "COMMIT"}),
                 Result
             catch
                 Err:Reason ->
-                    gen_server:call(Worker, {query, "ROLLBACK", []}),
+                    gen_server:call(Worker, {squery, "ROLLBACK"}),
                     erlang:raise(Err, Reason, erlang:get_stacktrace())
             after
                 pooler:return_member(PoolName, Worker, ok)

src/epgsql_pool_settings.erl

@@ -50,7 +50,7 @@ init([]) ->
     T = ets:new(?MODULE, [protected, named_table]),
     ets:insert(T, {{settings, connection_timeout}, 10000}),
     ets:insert(T, {{settings, query_timeout}, 10000}),
-    ets:insert(T, {{settings, pooler_get_worker_timeout}, 1000}),
+    ets:insert(T, {{settings, pooler_get_worker_timeout}, 10000}),
     ets:insert(T, {{settings, max_reconnect_timeout}, 5000}),
     ets:insert(T, {{settings, min_reconnect_timeout}, 100}),
     ets:insert(T, {{settings, keep_alive_timeout}, 60000}),

src/epgsql_pool_worker.erl

@@ -37,30 +37,32 @@ init(PoolName) ->
 
 
 -spec handle_call(gs_request(), gs_from(), gs_reply()) -> gs_call_reply().
-handle_call({query, _, _}, _From, #state{connection = undefined} = State) ->
+handle_call(_, _From, #state{connection = undefined} = State) ->
     {reply, {error, no_connection}, State};
 
-handle_call({query, _, _}, _From,
-            #state{connection = #epgsql_connection{sock = undefined}} = State) ->
+handle_call(_, _From, #state{connection = #epgsql_connection{sock = undefined}} = State) ->
     {reply, {error, reconnecting}, State};
 
-handle_call({query, Stmt, Params}, _From,
+handle_call({equery, Stmt, Params}, _From,
+            #state{connection = #epgsql_connection{sock = Sock}} = State) ->
+    Reply = case process_info(Sock, status) of
+                undefined -> {error, reconnecting};
+                {status, _} -> epgsql:equery(Sock, Stmt, Params)
+            end,
+    {reply, Reply, State};
+
+handle_call({squery, Stmt}, _From,
             #state{connection = #epgsql_connection{sock = Sock}} = State) ->
-    %% TStart = os:timestamp(),
-    %% timer:sleep(1000), %% TEMP
     Reply = case process_info(Sock, status) of
                 undefined -> {error, reconnecting};
-                {status, _} ->
-                    case Stmt of
-                        "BEGIN" -> epgsql:squery(Sock, Stmt); % skip Stmt parsing
-                        "COMMIT" -> epgsql:squery(Sock, Stmt);
-                        "ROLLBACK" -> epgsql:squery(Sock, Stmt);
-                        _ -> epgsql:equery(Sock, Stmt, Params)
-                    end
+                {status, _} -> epgsql:squery(Sock, Stmt)
             end,
-    %% Time = timer:now_diff(os:timestamp(), TStart),
     {reply, Reply, State};
 
+handle_call(cancel, _From, #state{connection = #epgsql_connection{sock = Sock}} = State) ->
+    epgsql:cancel(Sock),
+    {reply, ok, State};
+
 handle_call(Message, _From, State) ->
     error_logger:error_msg("unknown call ~p in ~p ~n", [Message, ?MODULE]),
     {reply, ok, State}.

test/epgsql_pool_SUITE.erl

@@ -165,5 +165,11 @@ reconnect_test(Config) ->
 
 
 timeout_test(_Config) ->
+    Res1 = epgsql_pool:query(my_pool, "SELECT pg_sleep(1)", [], [{timeout, 2000}]),
+    ct:pal("Res1:~p", [Res1]),
+    ?assertMatch({ok, _, _}, Res1),
 
+    Res2 = epgsql_pool:query(my_pool, "SELECT pg_sleep(1)", [], [{timeout, 500}]),
+    ct:pal("Res2:~p", [Res2]),
+    ?assertEqual({error, timeout}, Res2),
     ok.

test/epgsql_pool_settings_tests.erl

@@ -9,7 +9,7 @@ get_set_test() ->
 
     ?assertEqual(10000, epgsql_pool_settings:get(connection_timeout)),
     ?assertEqual(10000, epgsql_pool_settings:get(query_timeout)),
-    ?assertEqual(1000, epgsql_pool_settings:get(pooler_get_worker_timeout)),
+    ?assertEqual(10000, epgsql_pool_settings:get(pooler_get_worker_timeout)),
 
     epgsql_pool_settings:set(connection_timeout, 5000),
     ?assertEqual(5000, epgsql_pool_settings:get(connection_timeout)),