Erlang 中的互斥测试

Question

我目前有一个使用 erlang 编写的简单银行账户，我也重写了同一个银行账户，以期相互排斥，其想法是不能进行两次存款，其中设置/获取余额可以中断，以便最终值错误，例如 bal A = 10 bal B = 20：

WRONG
get_bal.A 0 → get_bal.B 0 → set_bal.A 10 → set_bal.B 20 == 20
RIGHT
get_bal.A 0 → set_bal.A 10 → get_bal.B 10 → set_bal.B 30 == 30

我的代码初始代码如下：

-module(bank).
-export([account/1, start/0, stop/0, deposit/1, get_bal/0, set_bal/1]).

account(Balance) ->
receive
    {set, NewBalance} ->
        account(NewBalance);
    {get, From} ->
        From ! {balance, Balance},
        account(Balance);
    stop -> ok
end.

start() ->
    Account_PID = spawn(bank, account, [0]),
    register(account_process, Account_PID).

stop() ->
    account_process ! stop,
    unregister(account_process).

set_bal(B) ->
    account_process ! {set, B}.

get_bal() ->
    account_process ! {get, self()},
    receive
    {balance, B} -> B
end.

deposit(Amount) ->
    OldBalance = get_bal(),
    NewBalance = OldBalance + Amount,
    set_bal(NewBalance).

这个想法是设置一个测试，这样如果最终余额可能错误，我可以收到错误消息，如果按计划进行，我可以通过。 我重写的代码也如下：

account(Balance) ->
receive
    {deposit, Amount, From} ->
        NewBalance = Balance + Amount,
        From ! {deposit, Amount, NewBalance},
        account(NewBalance);
    {withdraw, Amount, From} when Amount > Balance ->
        From ! {error, {insufficient_funds, Amount, Balance}},
        account(Balance);
    {withdraw, Amount, From} ->
        NewBalance = Balance - Amount,
        From ! {withdrawal, Amount, NewBalance},
        account(NewBalance);    
    {get, From} ->
        From ! {balance, Balance},
        account(Balance);
    stop -> ok
end.

deposit(Amount) when Amount > 0 ->
account_process ! {deposit, Amount, self()},
receive
    {deposit, Amount, NewBalance} ->
        {ok, NewBalance}
end.

withdraw(Amount) when Amount > 0 ->
account_process ! {withdraw, Amount, self()},
receive
    {withdrawal, Amount, NewBalance} ->
        {ok, NewBalance};
    Error ->
        Error
end.

感谢您的阅读，我们将不胜感激。

Answer 1

对此进行测试的一种方法是让两个或多个银行客户端进程将它们的消息交错发送给银行。每个客户端进程都可以使用要发送到银行的消息列表进行初始化，然后每个客户端进程都将处于循环中，等待来自某个控制器进程的消息，告诉它将下一条消息发送到银行。控制器将充当门，依次告诉每个客户端发送其下一条消息，因此总体效果是客户端消息将混合在一起。

如果您有两个这样的客户被这样控制，如果客户 A 想要存入 10 而客户 B 想要存入 20，那么如果他们使用您的原始银行代码执行您的问题中显示的顺序，会发生以下情况：

1. A 执行get_bal，得到 0
2. B 做get_bal，得到 0
3. A 做set_bal(0+10)，账户现在持有 10 个
4. B 做set_bal(0+20)，账户现在持有 20 个

显然这是不正确的，因为结果帐户余额应该是 30。

对正确的银行应用相同的客户序列会产生正确的金额：

1. A 做deposit(10)，账户现在持有 10 个
2. B 做deposit(20)，账户现在持有30

Answer 2

正如@Stratus 所说，您编写第二种方法的方式保证了存款方法中没有竞争条件的风险，因为帐户过程本身使操作在单个事务中获取余额+更新余额。

如果您想说服自己并比较这两种方法，您可以生成多个并行更新同一帐户的进程，并将所有存款完成后的实际余额与预期余额进行比较。下面的代码做存款测试：

-module(bank).
-export([account/1, start/0, stop/0, deposit1/1, deposit2/1, get_bal/0, set_bal/1, withdraw/1]).

%test

-export ([test/3,user/3]).

account(Balance) ->
receive
    {set, NewBalance} ->
        account(NewBalance);
    {get, From} ->
        From ! {balance, Balance},
        account(Balance);
    {deposit, Amount, From} ->
        NewBalance = Balance + Amount,
        From ! {deposit, Amount, NewBalance},
        account(NewBalance);
    {withdraw, Amount, From} when Amount > Balance ->
        From ! {error, {insufficient_funds, Amount, Balance}},
        account(Balance);
    {withdraw, Amount, From} ->
        NewBalance = Balance - Amount,
        From ! {withdrawal, Amount, NewBalance},
        account(NewBalance);    
    stop -> ok
end.





start() ->
    Account_PID = spawn(bank, account, [0]),
    register(account_process, Account_PID).

stop() ->
    account_process ! stop,
    unregister(account_process).

set_bal(B) ->
    account_process ! {set, B}.

get_bal() ->
    account_process ! {get, self()},
    receive
        {balance, B} -> B
    end.

deposit1(Amount) ->
    OldBalance = get_bal(),
    NewBalance = OldBalance + Amount,
    set_bal(NewBalance).

deposit2(Amount) when Amount > 0 ->
    account_process ! {deposit, Amount, self()},
    receive
        {deposit, Amount, NewBalance} ->
            {ok, NewBalance}
    end.

withdraw(Amount) when Amount > 0 ->
    account_process ! {withdraw, Amount, self()},
    receive
        {withdrawal, Amount, NewBalance} ->
            {ok, NewBalance};
        Error ->
            Error
    end.


test(Nbuser, Nbdeposit, Method) ->
    start(),
    done = spawn_users(Nbuser,Nbdeposit,Method,self()),
    receive_loop(Nbuser),
    Res = (get_bal() == Nbdeposit*Nbuser),
    stop(),
    Res.

spawn_users(0,_Nbdeposit,_Method,_Pid) -> done;
spawn_users(Nbuser,Nbdeposit,Method,Pid) ->
    spawn(?MODULE,user,[Nbdeposit,Method,Pid]),
    spawn_users(Nbuser-1,Nbdeposit,Method,Pid).

receive_loop(0) -> done;
receive_loop(N) ->
    receive
        end_deposit -> receive_loop(N-1)
    end.

user(0,_,Pid) ->
    get_bal(), % to be sure that with method deposit1, the last set_bal is processed
    Pid ! end_deposit;
user(N,Method,Pid) ->
    ?MODULE:Method(1),
    user(N-1,Method,Pid).

您可以验证，如果有 2 个用户进行 1 次存款，则方法 1 会出现错误，而使用方法 2，即使 1000 名用户进行 1000 次存款，您也不会出错。

2> bank:test(1,100,deposit1).
true
3> bank:test(2,1,deposit1).  
false
4> bank:test(1,100,deposit2).
true
5> bank:test(2,1,deposit2).  
true
6> bank:test(1000,1000,deposit2).
true

备注

结果将取决于您正在使用的机器。我是用smp的四核，所以错误的方法立即失败，我猜它可能需要更多的用户或单核上的存款。

Answer 3

在 Erlang 中，互斥不是问题。 进程是参与者，它们之间不共享内存。

看看这个问题：Is it easy to write traditional concurrency problems in Erlang?

至于代码，我可能会做这样的事情（“银行”表示为 gen_server）。这并不是您问题的真正解决方案，而是使用 OTP 实现相同目标的不同方式：

-module(bank).

-behaviour(gen_server).

%% API
-export([start_link/0, new_account/1, withdraw/2, deposit/2, get_bal/1]).

%% gen_server callbacks
-export([init/1,
     handle_call/3,
     handle_cast/2,
     handle_info/2,
     terminate/2,
     code_change/3]).

-record(state, {accounts = [] :: list()}).

%%%===================================================================
%%% API
%%%===================================================================

start_link() ->
    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).

new_account(Name) ->
    gen_server:call(?MODULE, {new_account, Name}).

deposit(Account, Amount) when Amount > 0 ->
    gen_server:call(?MODULE, {deposit, Account, Amount}).

withdraw(Account, Amount) when Amount > 0 ->
    gen_server:call(?MODULE, {withdraw, Account, Amount}).

get_bal(Account) ->
    gen_server:call(?MODULE, {get_bal, Account}).

%%%===================================================================
%%% gen_server callbacks
%%%===================================================================

init([]) ->
    {ok, #state{}}.

handle_call({new_account, Name}, _From, State) ->
    Accounts = State#state.accounts,
    case find_account(Name, Accounts) of
        none ->
            {reply, {account_created, Name}, State#state{accounts=[{Name, 0}|Accounts]}};
        _ ->
            {reply, already_exists, State}
        end;

handle_call({get_bal, Account}, _From, State) ->
    Accounts = State#state.accounts,
    {_Name, Balance} = find_account(Account, Accounts),
    {reply, Balance, State};

handle_call({deposit, Account, Amount}, _From, State) ->
    Accounts = State#state.accounts,
    {Name, Balance} = find_account(Account, Accounts),
    NewBalance = Balance + Amount,
    NewAccounts = lists:keyreplace(Name, 1, Accounts, {Name, NewBalance}),
    {reply, {deposit, Amount, NewBalance}, State#state{accounts=NewAccounts}};

handle_call({withdraw, Account, Amount}, _From, State) ->
    Accounts = State#state.accounts,
    {Name, Balance} = find_account(Account, Accounts),
     case Amount of
        Amount when Amount > Balance ->
            {reply, {insufficient_funds, Amount, Balance}, State};
        _ ->
            NewBalance = Balance - Amount,
            NewAccounts = lists:keyreplace(Name, 1, Accounts, {Name, NewBalance}),
            {reply, {withdrawal, Amount, NewBalance}, State#state{accounts=NewAccounts}}
    end;

handle_call(_Request, _From, State) ->
    Reply = not_implemented,
    {reply, Reply, State}.

handle_cast(_Msg, State) ->
    {noreply, State}.

handle_info(_Info, State) ->
    {noreply, State}.

terminate(_Reason, _State) ->
    ok.

code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%%%===================================================================
%%% Internal functions
%%%===================================================================
find_account(Account, Accounts) ->
    proplists:lookup(Account, Accounts).

Answer 4

答案类似于“提高 Erlang 的瓶颈性能”。在瓶颈的情况下，目标不是改进它（使其性能更高），而是完全消除它（这很少是不可能的）。

在排除的情况下，目标不是证明您“在过程 Y 期间锁定了数据 X 并回滚到 Z”，而是以完全不需要锁定的方式编写程序.我确信在某些情况下这可能是不可避免的，但我从未在 Erlang 中遇到过（至少我不记得了）。进程不共享内存。这就是为什么 Steve Vinoski 对您之前的（几乎相同的）问题 (Applying a mutex into an erlang example) 的回答展示了您应该如何组合操作，而不是在外部流程 API 中将它们的步骤分开。

如果您公开了一个过程add(Account, Value)，它恰好执行loop(Current + Value) 而没有发生任何其他事情，那么您肯定是在自找麻烦。但这暴露了一个极低级别的 API，不是吗？解决这个问题的正确方法是按照 Vinoski 的建议，只公开一个更高级别的 API，该 API 结合了值更改的操作和报告更改的效果。 不可能来自另一个进程的相同表单的另一个挂起操作可以在您现在尝试读取它时更改该值，从而导致一个或另一个 API 调用被绊倒，因为 API调用是排队的消息，而不是发生在不同线程之间的 C 风格函数调用，它们可能以任意顺序更改内存中同一位置的底层值，而无需锁定它们。

进程邮箱是你的互斥体。如果您按照预期的方式使用 Erlang 进程，则根本不存在此类错误。你不能搞砸它。每个动作都是完全原子的，按照消息接收的顺序排队，完全阻塞/锁定数据，并且底层数据在任何情况下都无法从外部访问。

每个进程在其存在期间已经对其所有数据拥有独占锁。