erlang学习:gen_server书上案例22.6练习题5

news2024/11/11 6:09:33

检查是否有懒惰的工人，也就是接受工作但不按时完成的进程。把任务请求函数修改为
返回{JobNumber, JobTime, F}，其中JobTime是工人必须完成任务的秒数。如果工人在JobTime

1时还未完成任务，服务器就应当向其发送一个hurry_up（快点儿）消息，而在JobTime + 1
时应该用调用exit(Pid, youre_fired)（你被解雇了）来杀掉这个工人进程。

在上次添加工作的方法中记录了JobTime字段,如果执行超时的话就按照hurry_up催促逻辑进行催促输出

-module(gen_server_test).

-export([start_link/0, add_job/3, work_wanted/0, job_done/1,hurry_up/1,statistics/0]).
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
  terminate/2, code_change/3]).
-export([test_job_centre/0]).
-define(SERVER, ?MODULE).

%%创建一个记录，记录中包含了一个工作队列，与下一个工作任务编号
-record(state,
{
  jobs = queue:new(),
  next_job_number = 1,
  workers = gb_sets:new(),
  completed_jobs = gb_sets:new()
}).

start_link() ->
  gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
add_job(Fun,WorkerPid,JobTime) ->
  gen_server:call(?SERVER, {add_job, Fun,WorkerPid,JobTime}).
work_wanted() ->
  gen_server:call(?SERVER, work_wanted).
job_done(JobNumber) ->
  gen_server:call(?SERVER, {job_done, JobNumber}).
hurry_up(WorkerPid) ->  %
  gen_server:call(?SERVER, {hurry_up, WorkerPid}).
statistics() ->
  gen_server:call(?SERVER, statistics).
%%---------------------------------------------------------------------------
init([]) ->
  {ok, #state{}}.

handle_call({add_job, Fun,WorkerPid,JobTime}, _From, State) ->
  %%向队列中添加元素,队列的任务为执行Fun中的操作
  NewJobs = queue:in({State#state.next_job_number,JobTime,Fun}, State#state.jobs),
  %%执行任务函数
  Fun(),
  NewWorkers = gb_sets:add(WorkerPid, State#state.workers),
  % 建立与工人的连接
  link(WorkerPid),
  {reply, State#state.next_job_number,
    State#state{jobs = NewJobs,
      next_job_number = State#state.next_job_number + 1,
      workers = NewWorkers}};
handle_call(work_wanted, _From, State) ->
  IsEmpty = queue:is_empty(State#state.jobs),
  case IsEmpty of
    true ->
      {reply, no, State};
    false ->
      {{value, {JobNumber, JobTime,Fun}}, NewJobs} = queue:out(State#state.jobs),
      spawn_link(fun() -> track_job(JobTime, self(),State) end),
      {reply, {JobNumber, JobTime, Fun}, State#state{jobs = NewJobs}}
  end;
handle_call({job_done, JobNumber}, _From, State) ->
  NewCompletedJobs = gb_sets:add(JobNumber, State#state.completed_jobs),
  {reply, ok, State#state{completed_jobs = NewCompletedJobs}};
handle_call({hurry_up, WorkerPid}, _From, State) ->
  % 这里可以发送"hurry_up"消息给WorkerPid
  io:format("Worker ~p is being asked to hurry up~n", [WorkerPid]),
  {reply, ok, State};
handle_call(statistics, _From, State) ->
  {reply, #{waiting => queue:len(State#state.jobs),
    in_progress => gb_sets:size(State#state.workers),
    completed => gb_sets:size(State#state.completed_jobs)}, State};
handle_call(Request, _From, State) ->
  {reply, {error, {unknown_request, Request}}, State}.

handle_cast(_Msg, State) ->
  {noreply, State}.
handle_info({exit, WorkerPid, _Reason}, State) ->
  %% 如果工人进程崩溃，则从workers集合中移除，并将任务重新添加到队列
  NewWorkers = gb_sets:delete(WorkerPid, State#state.workers),
  %% 查找并重新添加任务
  {ok, Fun} = find_task_for_worker(WorkerPid, State#state.jobs),
  NewJobs = queue:in({State#state.next_job_number - 1, Fun}, State#state.jobs),
  {noreply, State#state{workers = NewWorkers, jobs = NewJobs}};
handle_info(_Info, State) ->
  {noreply, State}.

terminate(_Reason, _State) ->
  ok.

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

track_job(JobTime, WorkerPid,State) ->
  timer:sleep(JobTime - 1),
  case gb_sets:is_member(WorkerPid, State#state.workers) of
    true ->
      gen_server:call(?SERVER, {hurry_up, WorkerPid});
    false ->
      ok  % 工人已经完成了任务或被解雇
  end,
  timer:sleep(2),
  case gb_sets:is_member(WorkerPid, State#state.workers) of
    true ->
      exit(WorkerPid, youre_fired);
    false ->
      ok  % 工人已经完成了任务或被解雇
  end.
find_task_for_worker(WorkerPid, Jobs) ->
  case gb_sets:is_member(WorkerPid, gb_sets:from_list(Jobs)) of
    true ->
      %% 找到对应的任务
      {value, {_, Fun}} = queue:out(Jobs),
      {ok, Fun};
    false ->
      {error, not_found}
  end.
test_job_centre() ->
  %%创建两个任务
  Job1 = fun
           () ->
             io:format("Doing job 1~n")
         end,
  Job2 = fun
           () ->
             io:format("Doing job 2~n")
         end,
  %%将两个任务放入任务队列中
  JobNumber1 = add_job(Job1,self(),5),
  io:format("Job number 1 is ~p~n", [JobNumber1]),
  JobNumber2 = add_job(Job2,self(),10),
  io:format("Job number 2 is ~p~n", [JobNumber2]),
  %%工人领取任务队列任务
  {JobNumber1, JobTime1,Job1} = work_wanted(),
  io:format("Doing job ~p with time limit ~p~n", [{JobNumber1, Job1}, JobTime1]),
  {JobNumber2, JobTime2, Job2} = work_wanted(),
  io:format("Doing job ~p with time limit ~p~n", [{JobNumber2, Job2}, JobTime2]),

  %% 测试hurry_up方法
  hurry_up(self()),

  %% 模拟超时情况
  timer:sleep(6),  % 等待超过Job1的时间限制
  receive
    {hurry_up, _} ->
      io:format("Received hurry up message~n")
  after 1000 ->  % 超时时间
    io:format("No hurry up message received~n")
  end,
  job_done(JobNumber1),
  job_done(JobNumber2),
  case work_wanted() of
    no ->
      io:format("No more jobs~n");
    {JobNumber, Job} ->
      io:format("Doing job ~p~n", [{JobNumber,Job}])
  end,
  exit(normal),

  ok.