Add dispcount:transaction/2 for running funs in watcher context

src/dispcount.erl

@@ -2,12 +2,15 @@
 -behaviour(application).
 -export([start/2,stop/1]).
 -export([start_dispatch/3, stop_dispatch/1,
-         dispatcher_info/1, checkout/1, checkout/2, checkin/3]).
+         dispatcher_info/1, checkout/1, checkout/2, transaction/2, checkin/3]).
 
 -callback init(Args::[any()]) -> {ok, CallbackState::any()}.
 -callback checkout(From::pid(), CallbackState::any()) -> {ok, Resource::any(), NewCallbackState::any()} |
                                     {error, Reason::any(), NewCallbackState::any()} |
                                     {stop, Reason::any(), NewCallbackState::any()}.
+-callback transaction(From::pid(), Fun::function(), CallbackState::any()) -> ok |
+                                    {ok, NewCallbackState::any()} |
+                                    {stop, Reason::any(), NewCallbackState::any()}.
 -callback checkin(Resource::any(), CallbackState::any()) -> {ok, NewCallbackState::any()} |
                                    {ignore, CallbackState::any()} |
                                    {stop, Reason::any(), NewCallbackState::any()}.
@@ -49,6 +52,10 @@ checkout(Info) ->
 checkout(Info, Timeout) ->
     dispcount_watcher:checkout(Info, Timeout).
 
+-spec transaction(term(), function()) -> ok.
+transaction(Info, Fun) ->
+    dispcount_watcher:transaction(Info, Fun).
+
 -spec checkin(term(), term(), term()) -> ok.
 checkin(Info, CheckinRef, Resource) ->
     dispcount_watcher:checkin(Info, CheckinRef, Resource).

src/dispcount_watcher.erl

@@ -8,7 +8,7 @@
                 id :: pos_integer(),
                 ref :: reference() | undefined}).
 
--export([start_link/3, checkout/1, checkout/2, checkin/3]).
+-export([start_link/3, checkout/1, checkout/2, transaction/2, checkin/3]).
 -export([init/1, handle_call/3, handle_cast/2,
          handle_info/2, code_change/3, terminate/2]).
 
@@ -42,6 +42,20 @@ checkout(ToPid,#config{dispatch_name=Name, num_watchers=Num, watcher_type=Type,
             {error, busy}
     end.
 
+transaction(#config{dispatch_name=Name, num_watchers=Num, watcher_type=Type, dispatch_table=DTid, dispatch_mechanism=DType, worker_table=WTid}, Fun) ->
+    case {Type, is_free(Type, DTid, Id = dispatch_id(Type, DType, DTid, Num))} of
+        {ets, true} ->
+            [{_,Pid}] = ets:lookup(WTid, Id),
+            gen_server:cast(Pid, {txn,self(), Fun});
+        {atomics, true} ->
+            [{_,Pid}] = ets:lookup(WTid, Id),
+            gen_server:cast(Pid, {txn,self(), Fun});
+        {named, true} ->
+            gen_server:cast(get_name(Name, Id), {txn,self(),Fun});
+        {_, false} ->
+            {error, busy}
+    end.
+
 -spec checkin(#config{}, Ref::term(), Resource::term()) -> ok.
 checkin(#config{}, {Pid,Ref}, Resource) ->
     %% we cheated, using a Pid for the CheckRef. Dirty optimisation!
@@ -84,6 +98,23 @@ handle_call({get, _Pid}, _From, State) -> % busy
 handle_call(_Call, _From, State) ->
     {noreply, State}.
 
+
+handle_cast({txn, Pid, Fun}, S=#state{callback=M, callback_state=CS, ref=undefined, config=Conf, id=Id}) ->
+    #config{watcher_type=Type, dispatch_table=DTid} = Conf,
+    try M:transaction(Pid, Fun, CS) of
+        ok ->
+            set_free(Type, DTid, Id),
+            {noreply, S};
+        {ok, NewCS} ->
+            set_free(Type, DTid, Id),
+            {noreply, S#state{callback_state=NewCS}};
+        {stop, Reason, NewCS} ->
+            M:terminate(Reason, NewCS),
+            {stop, Reason, S}
+    catch
+        Type:Reason ->
+            {stop, {Type,Reason}, S}
+    end;
 handle_cast({put, Ref, Res},
             S=#state{callback=M, callback_state=CS, config=Conf, id=Id, ref=Ref}) ->
     try M:checkin(Res, CS) of