Add mnesia activity subscription message

A process that calls mnesia:subscribe(activity) will receive the message:
{mnesia_activity_event, ActivityID, complete}
when any activity that caused a change to a database has finished
committing its changes.  This allows a subscriber to collect messages
already available through the mnesia:subscribe({table, ...}) system
to group them as completed transactions.

4 files changed, 65 insertions, 17 deletions

diff --git a/lib/mnesia/doc/src/Mnesia_chap5.xmlsrc b/lib/mnesia/doc/src/Mnesia_chap5.xmlsrc
index 3ec0aa37f5..1c7e3662e1 100644
--- a/lib/mnesia/doc/src/Mnesia_chap5.xmlsrc
+++ b/lib/mnesia/doc/src/Mnesia_chap5.xmlsrc
@@ -4,7 +4,7 @@
 <chapter>
   <header>
     <copyright>
-      <year>1997</year><year>2009</year>
+      <year>1997</year><year>2010</year>
       <holder>Ericsson AB. All Rights Reserved.</holder>
     </copyright>
     <legalnotice>
@@ -13,12 +13,12 @@
       compliance with the License. You should have received a copy of the
       Erlang Public License along with this software. If not, it can be
       retrieved online at http://www.erlang.org/.
-    
+
       Software distributed under the License is distributed on an "AS IS"
       basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
       the License for the specific language governing rights and limitations
       under the License.
-    
+
     </legalnotice>
 
     <title>Miscellaneous Mnesia Features</title>
@@ -885,16 +885,16 @@ ok
       <item>Removes the subscription on events of type
       <c>Event-Category</c></item>
     </taglist>
-    <p><c>Event-Category</c> may either be the atom <c>system</c>, or
+    <p><c>Event-Category</c> may either be the atom <c>system</c>, the atom <c>activity</c>, or
       one of the tuples <c>{table, Tab, simple}</c>, <c>{table, Tab, detailed}</c>.  The old event-category <c>{table, Tab}</c> is the same
       event-category as <c>{table, Tab, simple}</c>.
       The subscribe functions activate a subscription
       of events. The events are delivered as messages to the process
       evaluating the <c>mnesia:subscribe/1</c> function. The syntax of
-      system events is <c>{mnesia_system_event, Event}</c> and
-      <c>{mnesia_table_event, Event}</c> for table events. What system
-      events and table events means is described below.
-      </p>
+      system events is <c>{mnesia_system_event, Event}</c>,
+      <c>{mnesia_activity_event, Event}</c> for activity events, and
+      <c>{mnesia_table_event, Event}</c> for table events. What the various
+      event types mean is described below.</p>
     <p>All system events are subscribed by Mnesia's
       gen_event handler. The default gen_event handler is
       <c>mnesia_event</c>. But it may be changed by using the application
@@ -1039,8 +1039,26 @@ ok
     </section>
 
     <section>
+      <title>Activity Events</title>
+      <p>Currently, there is only one type of activity event:</p>
+      <taglist>
+       <tag><c>{complete, ActivityID}</c></tag>
+       <item>
+         <p>This event occurs when a transaction that caused a modification to the database
+           has completed.  It is useful for determining when a set of table events
+           (see below) caused by a given activity have all been sent.  Once the this event
+           has been received, it is guaranteed that no further table events with the same
+           ActivityID will be received.  Note that this event may still be received even
+           if no table events with a corresponding ActivityID were received, depending on
+           the tables to which the receiving process is subscribed.</p>
+	 <p>Dirty operations always only contain one update and thus no activity event is sent.</p>
+       </item>
+     </taglist>
+    </section>
+
+    <section>
       <title>Table Events</title>
-      <p>Another category of events are table events, which are
+      <p>The final category of events are table events, which are
         events related to table updates.  There are two types of table
         events simple and detailed.  
         </p>
diff --git a/lib/mnesia/src/mnesia_monitor.erl b/lib/mnesia/src/mnesia_monitor.erl
index 5df5df4969..5bd93d6b9b 100644
--- a/lib/mnesia/src/mnesia_monitor.erl
+++ b/lib/mnesia/src/mnesia_monitor.erl
@@ -256,6 +256,7 @@ init([Parent]) ->
     ?ets_new_table(mnesia_gvar, [set, public, named_table]), 
     ?ets_new_table(mnesia_stats, [set, public, named_table]), 
     set(subscribers, []),
+    set(activity_subscribers, []),
     mnesia_lib:verbose("~p starting: ~p~n", [?MODULE, self()]),
     Version = mnesia:system_info(version),
     set(version, Version),
diff --git a/lib/mnesia/src/mnesia_subscr.erl b/lib/mnesia/src/mnesia_subscr.erl
index afd1704dec..93d4a86f7f 100644
--- a/lib/mnesia/src/mnesia_subscr.erl
+++ b/lib/mnesia/src/mnesia_subscr.erl
@@ -1,19 +1,19 @@
 %%
 %% %CopyrightBegin%
-%% 
-%% Copyright Ericsson AB 1997-2009. All Rights Reserved.
-%% 
+%%
+%% Copyright Ericsson AB 1997-2010. All Rights Reserved.
+%%
 %% The contents of this file are subject to the Erlang Public License,
 %% Version 1.1, (the "License"); you may not use this file except in
 %% compliance with the License. You should have received a copy of the
 %% Erlang Public License along with this software. If not, it can be
 %% retrieved online at http://www.erlang.org/.
-%% 
+%%
 %% Software distributed under the License is distributed on an "AS IS"
 %% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
 %% the License for the specific language governing rights and limitations
 %% under the License.
-%% 
+%%
 %% %CopyrightEnd%
 %%
 
@@ -30,7 +30,8 @@
 	 subscribers/0,
 	 report_table_event/4,
 	 report_table_event/5, 
-	 report_table_event/6
+	 report_table_event/6,
+	 report_activity/1
 	]).
 
 %% gen_server callbacks
@@ -91,6 +92,8 @@ set_debug_level(Level, OldEnv) ->
 
 subscribe(ClientPid, system) ->
     change_subscr(activate, ClientPid, system);
+subscribe(ClientPid, activity) ->
+    change_subscr(activate, ClientPid, activity);
 subscribe(ClientPid, {table, Tab}) ->
     change_subscr(activate, ClientPid, {table, Tab, simple});
 subscribe(ClientPid, {table, Tab, simple}) ->
@@ -102,6 +105,8 @@ subscribe(_ClientPid, What) ->
 
 unsubscribe(ClientPid, system) ->
     change_subscr(deactivate, ClientPid, system);
+unsubscribe(ClientPid, activity) ->
+    change_subscr(deactivate, ClientPid, activity);
 unsubscribe(ClientPid, {table, Tab}) ->
     change_subscr(deactivate, ClientPid, {table, Tab, simple});
 unsubscribe(ClientPid, {table, Tab, simple}) ->
@@ -120,6 +125,15 @@ change_subscr(Kind, ClientPid, What) ->
 subscribers() ->
     [whereis(mnesia_event) | mnesia_lib:val(subscribers)].
 
+report_activity({dirty, _pid}) -> 
+    ok;
+report_activity(Tid) ->
+    case ?catch_val(activity_subscribers) of
+	{'EXIT', _} -> ok;
+	Subscribers ->
+	    deliver(Subscribers, {mnesia_activity_event, {complete, Tid}})
+	end.
+
 report_table_event(Tab, Tid, Obj, Op) ->   
     case ?catch_val({Tab, commit_work}) of
 	{'EXIT', _} -> ok;
@@ -300,6 +314,9 @@ code_change(_OldVsn, State, _Extra) ->
 do_change({activate, ClientPid, system}, SubscrTab) when is_pid(ClientPid) ->
     Var = subscribers,
     activate(ClientPid, system, Var, subscribers(), SubscrTab);
+do_change({activate, ClientPid, activity}, SubscrTab) when is_pid(ClientPid) ->
+    Var = activity_subscribers,
+    activate(ClientPid, activity, Var, mnesia_lib:val(Var), SubscrTab);
 do_change({activate, ClientPid, {table, Tab, How}}, SubscrTab) when is_pid(ClientPid) ->
     case ?catch_val({Tab, where_to_read}) of
 	Node when Node == node() ->
@@ -313,6 +330,9 @@ do_change({activate, ClientPid, {table, Tab, How}}, SubscrTab) when is_pid(Clien
 do_change({deactivate, ClientPid, system}, SubscrTab) ->
     Var = subscribers,
     deactivate(ClientPid, system, Var, SubscrTab);
+do_change({deactivate, ClientPid, activity}, SubscrTab) ->
+    Var = activity_subscribers,
+    deactivate(ClientPid, activity, Var, SubscrTab);
 do_change({deactivate, ClientPid, {table, Tab, How}}, SubscrTab) ->
     Var = {Tab, commit_work},
     deactivate(ClientPid, {table, Tab, How}, Var, SubscrTab);
@@ -345,7 +365,7 @@ do_change(_, _) ->
 
 activate(ClientPid, What, Var, OldSubscribers, SubscrTab) ->
     Old = 
-	if Var == subscribers -> 
+	if Var == subscribers orelse Var == activity_subscribers ->
 		OldSubscribers;
 	   true -> 
 		case lists:keysearch(subscribers, 1, OldSubscribers) of
@@ -379,6 +399,9 @@ activate(ClientPid, What, Var, OldSubscribers, SubscrTab) ->
 add_subscr(subscribers, _What, Pid) ->
     mnesia_lib:add(subscribers, Pid),
     {ok, node()};
+add_subscr(activity_subscribers, _What, Pid) ->
+    mnesia_lib:add(activity_subscribers, Pid),
+    {ok, node()};
 add_subscr({Tab, commit_work}, What, Pid) ->
     Commit = mnesia_lib:val({Tab, commit_work}),
     case lists:keysearch(subscribers, 1, Commit) of
@@ -427,6 +450,8 @@ deactivate(ClientPid, What, Var, SubscrTab) ->
 
 del_subscr(subscribers, _What, Pid) ->
     mnesia_lib:del(subscribers, Pid);
+del_subscr(activity_subscribers, _What, Pid) ->
+    mnesia_lib:del(activity_subscribers, Pid);
 del_subscr({Tab, commit_work}, What, Pid) ->
     Commit = mnesia_lib:val({Tab, commit_work}),
     case lists:keysearch(subscribers, 1, Commit) of
@@ -473,6 +498,8 @@ do_handle_exit([{ClientPid, What} | Tail]) ->
     case What of
 	system ->
 	    del_subscr(subscribers, What, ClientPid);
+	activity ->
+	    del_subscr(activity_subscribers, What, ClientPid);
 	{_, Tab, _Level} ->
 	    del_subscr({Tab, commit_work}, What, ClientPid)    
     end,
diff --git a/lib/mnesia/src/mnesia_tm.erl b/lib/mnesia/src/mnesia_tm.erl
index d42109c3da..f3ffac5493 100644
--- a/lib/mnesia/src/mnesia_tm.erl
+++ b/lib/mnesia/src/mnesia_tm.erl
@@ -1733,7 +1733,9 @@ do_commit(Tid, C, DumperMode) ->
     R  = do_snmp(Tid, C#commit.snmp),
     R2 = do_update(Tid, ram_copies, C#commit.ram_copies, R),
     R3 = do_update(Tid, disc_copies, C#commit.disc_copies, R2),
-    do_update(Tid, disc_only_copies, C#commit.disc_only_copies, R3).
+    R4 = do_update(Tid, disc_only_copies, C#commit.disc_only_copies, R3),
+    mnesia_subscr:report_activity(Tid),
+    R4.
 
 %% Update the items
 do_update(Tid, Storage, [Op | Ops], OldRes) ->