Locked pgs

src/merle.erl

@@ -327,6 +327,9 @@ start_link(Host, Port) ->
 %% @private
 init([Host, Port]) ->
     log4erl:info("Socket initialized!"),
+
+    erlang:process_flag(trap_exit, true),
+
     gen_tcp:connect(Host, Port, ?TCP_OPTS_ACTIVE).
 
 handle_call({stats}, _From, Socket) ->
@@ -464,6 +467,11 @@ handle_info({tcp_closed, Socket}, Socket) ->
     {stop, {error, tcp_closed}, Socket};
 handle_info({tcp_error, Socket, Reason}, Socket) -> 
     {stop, {error, {tcp_error, Reason}}, Socket};
+
+handle_info({'EXIT', _, Reason}, Socket) ->
+    log4erl:warn("Exiting merle connection ~p", [Reason]),
+    {stop, normal, Socket};
+
 handle_info(_Info, State) -> {noreply, State}.
 
 %% @private
@@ -507,8 +515,8 @@ send_get_cmd(Socket, Cmd, Timeout) ->
 		[{_, Value}] -> {ok, Value};
 		[] -> {error, not_found};
 		{error, Error} -> 
-            log4erl:error("Encountered error from memcache; killing connection now: ~p", [Error]),
-            erlang:exit(Error),
+            log4erl:warn("Encountered error from memcache; killing connection now: ~p", [Error]),
+            erlang:exit(self(), Error),
             {error, Error}
     	end,
     inet:setopts(Socket, ?TCP_OPTS_ACTIVE),
@@ -550,7 +558,7 @@ do_recv_stats(Timeout) ->
             inet:setopts(Socket, ?TCP_OPTS_ACTIVE),  
             [{Field, Value} | do_recv_stats(Timeout)]
      after Timeout ->
-	timeout
+	     timeout
    end.
 %% @doc receive function for simple responses (not containing VALUEs)
 recv_simple_reply(Timeout) ->
@@ -559,12 +567,12 @@ recv_simple_reply(Timeout) ->
         	inet:setopts(Socket, ?TCP_OPTS_ACTIVE),
         	parse_simple_response_line(Data); 
         {error, closed} ->
-            log4erl:error("Encountered error while receiving simple reply from memcache; killing connection now."),
-            erlang:exit(connection_closed),
+            log4erl:warn("Encountered error while receiving simple reply from memcache; killing connection now."),
+            erlang:exit(self(), connection_closed),
   			connection_closed
     after Timeout -> 
-        log4erl:error("Encountered timeout while receiving simple reply from memcache; killing connection now."),
-        erlang:exit(timeout),
+        log4erl:warn("Encountered timeout while receiving simple reply from memcache; killing connection now."),
+        erlang:exit(self(), timeout),
         {error, timeout}
     end.
 parse_simple_response_line(<<"OK", _B/binary>>) -> ok;

src/merle_cluster.erl

@@ -1,6 +1,6 @@
 -module(merle_cluster).
 
--export([configure/2]).
+-export([configure/2, exec/4]).
 
 index_map(F, List) ->
     {Map, _} = lists:mapfoldl(fun(X, Iter) -> {F(X, Iter), Iter +1} end, 1, List),
@@ -26,14 +26,38 @@ configure(MemcachedHosts, ConnectionsPerHost) ->
     {M, B} = dynamic_compile:from_string(ModuleString),
     code:load_binary(M, "", B),
 
+    % start all merle watchers
     lists:foreach(
         fun([Host, Port]) ->
             lists:foreach(
                 fun(_) -> 
-                    supervisor:start_child(merle_sup, [[Host, Port]]) 
+                    supervisor:start_child(merle_watcher_sup, [[Host, Port]]) 
                 end,
                 lists:seq(1, ConnectionsPerHost)
             )
         end, 
         SortedMemcachedHosts
     ).
+
+
+exec(Key, Fun, FullDefault, Now) ->
+    S = merle_cluster_dynamic:get_server(Key),
+
+    case merle_pool:get_closest_pid(round_robin, S) of
+        in_use ->
+            FullDefault;
+
+        P ->
+            merle_watcher:monitor(P, self()),
+
+            MC = merle_watcher:merle_connection(P),
+
+            Value = Fun(MC, Key),
+
+            merle_watcher:demonitor(P),
+
+            merle_pool:checkin_pid(P, Now),
+
+            Value
+
+    end.