From fdf28232f1007af441b9c1854bd512115c0390d0 Mon Sep 17 00:00:00 2001
From: Andrew Thompson <andrew@hijacked.us>
Date: Sat, 21 Mar 2020 10:35:07 -0700
Subject: [PATCH] Add dispcount:transaction/2 for running funs in watcher
 context

To avoid unnecessary calls and waiting for responses allow for a fun to
be passed as a transaction that can be run by the watcher. As soon as
the fun completes, the watcher is marked as free again without
intervention by the calling process. This should improve throughput for
use cases where the caller having direct access to the resource is
not necessary.
---
 src/dispcount.erl         |  9 ++++++++-
 src/dispcount_watcher.erl | 33 ++++++++++++++++++++++++++++++++-
 2 files changed, 40 insertions(+), 2 deletions(-)

diff --git a/src/dispcount.erl b/src/dispcount.erl
index f3af719..fca3938 100644
--- a/src/dispcount.erl
+++ b/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).
diff --git a/src/dispcount_watcher.erl b/src/dispcount_watcher.erl
index 2139f41..7d782d0 100644
--- a/src/dispcount_watcher.erl
+++ b/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