fixing sync_transactions and tweaking timeouts

heroku · May 26, 2011 · 2acc100 · 2acc100
1 parent f448176
commit 2acc100
Show file tree

Hide file tree

Showing 4 changed files with 50 additions and 18 deletions.
diff --git a/src/logplex_channel.erl b/src/logplex_channel.erl
@@ -27,13 +27,29 @@
 -include_lib("logplex.hrl").
 
 create(ChannelName, AppId, Addon) when is_binary(ChannelName), is_integer(AppId), is_binary(Addon) ->
-    ChannelId = mnesia:dirty_update_counter(counters, channel, 1),
-    {atomic, _} = mnesia:transaction(
-        fun() ->
-            Channel = #channel{id=ChannelId, name=ChannelName, app_id=AppId, addon=Addon},
-            mnesia:write(channel, Channel, write)
-        end),
-    ChannelId.
+    case sync_incr_channel_id() of
+        {atomic, ChannelId} when is_integer(ChannelId) ->
+            {atomic, _} = mnesia:sync_transaction(
+                fun() ->
+                    Channel = #channel{id=ChannelId, name=ChannelName, app_id=AppId, addon=Addon},
+                    mnesia:write(channel, Channel, write)
+                end),
+            ChannelId;
+        {aborted, Reason} ->
+            {error, Reason}
+    end.
+
+sync_incr_channel_id() ->
+    mnesia:sync_transaction(fun() ->
+        case mnesia:wread({counters, channel}) of
+            [{counters, channel, ChannelId}] ->
+                mnesia:write(counters, {counters, channel, ChannelId+1}, write),
+                ChannelId+1;
+            [] ->
+                mnesia:write(counters, {counters, channel, 1}, write),
+                1
+        end
+    end).
 
 delete(ChannelId) when is_integer(ChannelId) ->
     case lookup(ChannelId) of

diff --git a/src/logplex_drain.erl b/src/logplex_drain.erl
@@ -36,16 +36,32 @@ create(ChannelId, Host, Port) when is_integer(ChannelId), is_binary(Host), (is_i
                     error_logger:error_msg("invalid drain: ~p:~p~n", [Host, Port]),
                     {error, invalid_drain};
                 Ip ->
-                    DrainId = mnesia:dirty_update_counter(counters, drain, 1),
-                    {atomic, _} = mnesia:transaction(
-                        fun() ->
-                            Drain = #drain{id=DrainId, channel_id=ChannelId, resolved_host=Ip, host=Host, port=Port},
-                            mnesia:write(drain, Drain, write)
-                        end),
-                    DrainId
+                    case sync_incr_drain_id() of
+                        {atomic, DrainId} when is_integer(DrainId) ->
+                            {atomic, _} = mnesia:transaction(
+                                fun() ->
+                                    Drain = #drain{id=DrainId, channel_id=ChannelId, resolved_host=Ip, host=Host, port=Port},
+                                    mnesia:write(drain, Drain, write)
+                                end),
+                            DrainId;
+                        {aborted, Reason} ->
+                            {error, Reason}
+                    end
             end
     end.
 
+sync_incr_drain_id() ->
+    mnesia:sync_transaction(fun() ->
+        case mnesia:wread({counters, drain}) of
+            [{counters, drain, DrainId}] ->
+                mnesia:write(counters, {counters, drain, DrainId+1}, write),
+                DrainId+1;
+            [] ->
+                mnesia:write(counters, {counters, drain, 1}, write),
+                1
+        end
+    end).
+
 delete(ChannelId, Host, Port) when is_integer(ChannelId), is_binary(Host) ->
     Port1 = if Port == "" -> undefined; true -> list_to_integer(Port) end,
     case ets:match_object(drain, #drain{id='_', channel_id=ChannelId, resolved_host='_', host=Host, port=Port1}) of

diff --git a/src/logplex_rate_limit.erl b/src/logplex_rate_limit.erl
@@ -31,7 +31,7 @@
 
 -include_lib("logplex.hrl").
 
--define(TIMEOUT, 10000).
+-define(LOCK_TIMEOUT, 1000).
 
 %% API functions
 start_link() ->
@@ -57,9 +57,9 @@ lock(ChannelId) when is_integer(ChannelId) ->
     end.
 
 set_lock(Node, ChannelId) when is_integer(ChannelId) ->
-    gen_server:call({?MODULE, Node}, {set_lock, ChannelId}, ?TIMEOUT).
+    gen_server:call({?MODULE, Node}, {set_lock, ChannelId}, ?LOCK_TIMEOUT).
 set_lock(ChannelId) when is_integer(ChannelId) ->
-    gen_server:call(?MODULE, {set_lock, ChannelId}, ?TIMEOUT).
+    gen_server:call(?MODULE, {set_lock, ChannelId}, ?LOCK_TIMEOUT).
 
 clear_all() ->
     gen_server:cast(?MODULE, clear_all).

diff --git a/src/logplex_token.erl b/src/logplex_token.erl
@@ -30,7 +30,7 @@ create(ChannelId, TokenName) when is_integer(ChannelId), is_binary(TokenName) ->
     case logplex_channel:lookup(ChannelId) of
         #channel{app_id=AppId, addon=Addon} ->
             TokenId = list_to_binary("t." ++ string:strip(os:cmd("uuidgen"), right, $\n)),
-            {atomic, _} = mnesia:transaction(
+            {atomic, _} = mnesia:sync_transaction(
                 fun() ->
                     Token = #token{id=TokenId, channel_id=ChannelId, name=TokenName, app_id=AppId, addon=Addon},
                     mnesia:write(token, Token, write)