ft: 新增 atomics 版本

c_src/eNpc.cmd

@@ -0,0 +1,4 @@
+@echo off
+setlocal
+set rebarscript=%~f0
+escript.exe "%rebarscript:.cmd=%" %*

c_src/termInt/rebar.config

@@ -0,0 +1,7 @@
+{port_specs, [
+   {"../../priv/termInt.so", ["*.c"]}
+]}.
+
+
+
+

c_src/termInt/termInt.c

@@ -0,0 +1,12 @@
+#include "erl_nif.h"
+#include <stdio.h>
+
+static ERL_NIF_TERM termType(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[]) {
+    ErlNifUInt64 TermInt = (ErlNifUInt64)argv[0];
+    return enif_make_uint64(env, TermInt);
+}
+static ErlNifFunc nif_funcs[] = {
+    {"termInt", 1, termInt}
+};
+
+ERL_NIF_INIT(termInt, nif_funcs, NULL, NULL, NULL, NULL);

include/eGLock.hrl

@@ -1,9 +1,19 @@
 %% 锁key
 -define(EtsGLockKey, '$EtsGLockKey').
+
+%% 锁key
+-define(EtsGLockPid, '$EtsGLockPid').
+
 %% 默认超时时间单位:Ms
 -define(LockTimeOut, 5000).
 
 %% 超时重试时间单位:Ms
 -define(ReTryTime, 10).
 
--define(eGLockMgr, eGLockMgr).
+-define(eELockMgr, eELockMgr).
+-define(eALockMgr, eALockMgr).
+
+%% 数组数量
+-define(eALockSize, 1048576).
+%% 数组数量
+-define(eALockRef, eALockRef).

rebar.config

@@ -3,6 +3,12 @@
 {deps, []}.
 
 {shell, [
-  % {config, "config/sys.config"},
+    % {config, "config/sys.config"},
     {apps, [eGLock]}
 ]}.
+
+{pre_hooks,
+    [{"", compile, "escript c_src/eNpc compile"}]}.
+
+{post_hooks,
+    [{"", clean, "escript c_src/eNpc clean"}]}.

src/eALock.erl

@@ -0,0 +1,142 @@
+-module(eALock).
+
+-include("eGLock.hrl").
+-define(CASE(Cond, Then, That), case Cond of true -> Then; _ -> That end).
+
+-export([
+	lockApply/2
+	, lockApply/3
+]).
+
+-spec lockApply(KeyOrKeys :: term() | [term()], MFAOrFun :: {M :: atom(), F :: atom(), Args :: list()} | {Fun :: function(), Args :: list()}) -> term().
+lockApply(KeyOrKeys, MFAOrFun) ->
+	lockApply(KeyOrKeys, MFAOrFun, ?LockTimeOut).
+
+-spec lockApply(KeyOrKeys :: term() | [term()], MFAOrFun :: {M :: atom(), F :: atom(), Args :: list()} | {Fun :: function(), Args :: list()}, TimeOut :: integer() | infinity) -> term().
+lockApply(KeyOrKeys, MFAOrFun, TimeOut) ->
+	GLockMgrPid = persistent_term:get(?eALockMgr),
+	ALockRef = persistent_term:get(?eALockRef),
+	CurPid = self(),
+	PidInt = termInt:termInt(CurPid),
+	case is_list(KeyOrKeys) of
+		true ->
+			KeyIxs = getKexIxs(KeyOrKeys, []),
+			lockApplys(KeyIxs, KeyOrKeys, ALockRef, CurPid, PidInt, GLockMgrPid, MFAOrFun, TimeOut);
+		_ ->
+			lockApply(erlang:phash2(KeyOrKeys, ?eALockSize) + 1, KeyOrKeys, ALockRef, CurPid, PidInt, GLockMgrPid, MFAOrFun, TimeOut)
+	end.
+
+lockApply(KeyIx, Key, ALockRef, CurPid, PidInt, GLockMgrPid, MFAOrFun, TimeOut) ->
+	link(GLockMgrPid),
+	ets:insert(?EtsGLockPid, {CurPid, KeyIx}),
+	case atomics:compare_exchange(ALockRef, KeyIx, 0, PidInt) of
+		ok ->
+			try doApply(MFAOrFun)
+			catch C:R:S ->
+				{error, {lock_apply_error, {C, R, S}}}
+			after
+				atomics:exchange(ALockRef, KeyIx, 0),
+				ets:delete(?EtsGLockPid, CurPid),
+				unlink(GLockMgrPid),
+				ok
+			end;
+		_ ->
+			loopTry(KeyIx, Key, ALockRef, CurPid, PidInt, GLockMgrPid, MFAOrFun, TimeOut)
+	end.
+
+loopTry(KeyIx, Key, ALockRef, CurPid, PidInt, GLockMgrPid, MFAOrFun, TimeOut) ->
+	receive
+	after ?ReTryTime ->
+		LTimeOut = ?CASE(TimeOut == infinity, TimeOut, TimeOut - ?ReTryTime),
+		case LTimeOut >= 0 of
+			true ->
+				case atomics:compare_exchange(ALockRef, KeyIx, 0, PidInt) of
+					ok ->
+						try doApply(MFAOrFun)
+						catch C:R:S ->
+							{error, {lock_apply_error, {C, R, S}}}
+						after
+							atomics:exchange(ALockRef, KeyIx, 0),
+							ets:delete(?EtsGLockPid, CurPid),
+							unlink(GLockMgrPid),
+							ok
+						end;
+					_ ->
+						loopTry(KeyIx, Key, ALockRef, CurPid, PidInt, GLockMgrPid, MFAOrFun, LTimeOut)
+				end;
+			_ ->
+				ets:delete(?EtsGLockPid, CurPid),
+				unlink(GLockMgrPid),
+				{error, {lock_timeout, Key}}
+		end
+	end.
+
+lockApplys(KeyIxs, Keys, ALockRef, CurPid, PidInt, GLockMgrPid, MFAOrFun, TimeOut) ->
+	CurPid = self(),
+	link(GLockMgrPid),
+	ets:insert(?EtsGLockPid, {CurPid, KeyIxs}),
+	case tryLockAll(KeyIxs, ALockRef, PidInt, []) of
+		ok ->
+			try doApply(MFAOrFun)
+			catch C:R:S ->
+				{error, {lock_apply_error, {C, R, S}}}
+			after
+				[atomics:exchange(ALockRef, KeyIx, 0) || KeyIx <- KeyIxs],
+				ets:delete(?EtsGLockPid, CurPid),
+				unlink(GLockMgrPid),
+				ok
+			end;
+		_ ->
+			loopTrys(KeyIxs, Keys, ALockRef, CurPid, PidInt, GLockMgrPid, MFAOrFun, TimeOut)
+	end.
+
+loopTrys(KeyIxs, Keys, ALockRef, CurPid, PidInt, GLockMgrPid, MFAOrFun, TimeOut) ->
+	receive
+	after ?ReTryTime ->
+		LTimeOut = ?CASE(TimeOut == infinity, TimeOut, TimeOut - ?ReTryTime),
+		case LTimeOut >= 0 of
+			true ->
+				case tryLockAll(KeyIxs, PidInt, ALockRef, []) of
+					ok ->
+						try doApply(MFAOrFun)
+						catch C:R:S ->
+							{error, {lock_apply_error, {C, R, S}}}
+						after
+							[atomics:exchange(ALockRef, KeyIx, 0) || KeyIx <- KeyIxs],
+							ets:delete(?EtsGLockPid, CurPid),
+							unlink(GLockMgrPid),
+							ok
+						end;
+					_ ->
+						loopTrys(KeyIxs, Keys, ALockRef, CurPid, PidInt, GLockMgrPid, MFAOrFun, TimeOut)
+				end;
+			_ ->
+				ets:delete(?EtsGLockPid, CurPid),
+				unlink(GLockMgrPid),
+				{error, {lock_timeout, Keys}}
+		end
+	end.
+
+getKexIxs([], IxAcc) -> IxAcc;
+getKexIxs([Key | Keys], IxAcc) ->
+	KeyIx = erlang:phash2(Key, ?eALockSize) + 1,
+	getKexIxs(Keys, ?CASE(lists:member(KeyIx, IxAcc), IxAcc, [KeyIx | IxAcc])).
+
+tryLockAll([], _ALockRef, _PidInt, _LockAcc) ->
+	ok;
+tryLockAll([KeyIx | KeyIxs], ALockRef, PidInt, LockAcc) ->
+	case atomics:compare_exchange(ALockRef, KeyIx, 0, PidInt) of
+		ok ->
+			tryLockAll(KeyIxs, ALockRef, PidInt, [KeyIx | LockAcc]);
+		_ ->
+			[atomics:exchange(ALockRef, OneLock, 0) || OneLock <- LockAcc],
+			false
+	end.
+
+doApply({M, F, A}) ->
+	apply(M, F, A);
+doApply({Fun, Args}) ->
+	apply(Fun, Args).
+
+	
+

src/eALockMgr.erl

@@ -0,0 +1,58 @@
+-module(eALockMgr).
+
+-behaviour(gen_server).
+
+-include("eGLock.hrl").
+
+-export([start_link/0]).
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
+	code_change/3]).
+
+-define(SERVER, ?MODULE).
+
+-record(state, {}).
+
+%%%===================================================================
+%%% Spawning and gen_server implementation
+%%%===================================================================
+
+start_link() ->
+	gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
+
+init([]) ->
+	process_flag(trap_exit, true),
+	ATLockRef = atomics:new(?eALockSize, [{signed, false}]),
+	persistent_term:put(?eALockRef, ATLockRef),
+	persistent_term:put(?eALockMgr, self()),
+	ets:new(?EtsGLockPid, [named_table, set, public, {write_concurrency, auto}]),
+	{ok, #state{}}.
+
+handle_call(_Request, _From, State) ->
+	{reply, ok, State}.
+
+handle_cast(_Request, State) ->
+	{noreply, State}.
+
+handle_info({'EXIT', Pid, _Reason}, State) ->
+	case ets:take(?EtsGLockPid, Pid) of
+		[] -> ignore;
+		[KeyIxOrKeyIxs] ->
+			ALockRef = persistent_term:get(?eALockRef),
+			PidInt = termInt:termInt(PidInt),
+			case is_integer(KeyIxOrKeyIxs) of
+				true ->
+					atomics:compare_exchange(ALockRef, KeyIxOrKeyIxs, PidInt, 0);
+				_ ->
+					[atomics:compare_exchange(ALockRef, KeyIx, PidInt, 0) || KeyIx <- KeyIxOrKeyIxs],
+					ok
+			end
+	end,
+	{noreply, State};
+handle_info(_Info, State) ->
+	{noreply, State}.
+
+terminate(_Reason, _State) ->
+	ok.
+
+code_change(_OldVsn, State, _Extra) ->
+	{ok, State}.

src/eGLock.erl → src/eELock.erl

@@ -1,6 +1,7 @@
--module(eGLock).
+-module(eELock).
 
 -include("eGLock.hrl").
+-define(CASE(Cond, Then, That), case Cond of true -> Then; _ -> That end).
 
 -export([
 	lockApply/2
@@ -13,7 +14,7 @@ lockApply(KeyOrKeys, MFAOrFun) ->
 
 -spec lockApply(KeyOrKeys :: term() | [term()], MFAOrFun :: {M :: atom(), F :: atom(), Args :: list()} | {Fun :: function(), Args :: list()}, TimeOut :: integer() | infinity) -> term().
 lockApply(KeyOrKeys, MFAOrFun, TimeOut) ->
-	GLockMgrPid = persistent_term:get(?eGLockMgr),
+	GLockMgrPid = persistent_term:get(?eELockMgr),
 	link(GLockMgrPid),
 	Pid = self(),
 	case is_list(KeyOrKeys) of
@@ -25,7 +26,6 @@ lockApply(KeyOrKeys, MFAOrFun, TimeOut) ->
 			lockApply({KeyOrKeys, Pid}, KeyOrKeys, GLockMgrPid, MFAOrFun, TimeOut)
 	end.
 
--define(CASE(Cond, Then, That), case Cond of true -> Then; _ -> That end).
 lockApply(KeyPid, Key, GLockMgrPid, MFAOrFun, TimeOut) ->
 	case ets:insert_new(?EtsGLockKey, KeyPid) of
 		true ->

src/eGLockMgr.erl → src/eELockMgr.erl

@@ -1,4 +1,4 @@
--module(eGLockMgr).
+-module(eELockMgr).
 
 -behaviour(gen_server).
 
@@ -17,12 +17,12 @@
 %%%===================================================================
 
 start_link() ->
-	gen_server:start_link({local, ?eGLockMgr}, ?MODULE, [], []).
+	gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
 
 init([]) ->
 	process_flag(trap_exit, true),
+	persistent_term:put(?eELockMgr, self()),
 	ets:new(?EtsGLockKey, [named_table, set, public, {write_concurrency, auto}, {read_concurrency, true}]),
-	persistent_term:put(?eGLockMgr, self()),
 	{ok, #state{}}.
 
 handle_call(_Request, _From, State) ->

src/eGLock_sup.erl

@@ -15,12 +15,20 @@ init([]) ->
 	SupFlags = #{strategy => one_for_all, intensity => 100, period => 3600},
 	ChildSpecs = [
 		#{
-			id => ?eGLockMgr,
-			start => {eGLockMgr, start_link, []},
+			id => ?eELockMgr,
+			start => {eELockMgr, start_link, []},
 			restart => permanent,
 			shutdown => 3000,
 			type => worker,
-			modules => [eGLockMgr]
+			modules => [eELockMgr]
+		},
+		#{
+			id => eALockMgr,
+			start => {eALockMgr, start_link, []},
+			restart => permanent,
+			shutdown => 3000,
+			type => worker,
+			modules => [eALockMgr]
 		}
 	],
 	{ok, {SupFlags, ChildSpecs}}.

src/tremInt.erl

@@ -0,0 +1,23 @@
+-module(tremInt).
+
+-export([termInt/1]).
+
+-on_load(init/0).
+
+init() ->
+	SoName =
+		case code:priv_dir(?MODULE) of
+			{error, _} ->
+				case code:which(?MODULE) of
+					Filename when is_list(Filename) ->
+						filename:join([filename:dirname(Filename), "../priv", "termInt"]);
+					_ ->
+						filename:join("../priv", "termInt")
+				end;
+			Dir ->
+				filename:join(Dir, "termInt")
+		end,
+	erlang:load_nif(SoName, 0).
+
+termInt(_Term) ->
+	erlang:error({"NIF not implemented in nif_test at line", ?LINE}).