From 1935b414c70f350f6f416182449c5e46ad009fb2 Mon Sep 17 00:00:00 2001 From: Feng Lee Date: Sat, 10 Oct 2015 13:12:00 +0800 Subject: [PATCH 1/4] issue #282 - improve sm --- include/emqttd.hrl | 7 +- src/emqttd_sm.erl | 139 ++++++++++++++++----------------------- src/emqttd_sm_helper.erl | 37 ++++++++--- 3 files changed, 87 insertions(+), 96 deletions(-) diff --git a/include/emqttd.hrl b/include/emqttd.hrl index 53a5ba77d..5c2f61019 100644 --- a/include/emqttd.hrl +++ b/include/emqttd.hrl @@ -105,10 +105,9 @@ %% MQTT Session %%------------------------------------------------------------------------------ -record(mqtt_session, { - client_id, - sess_pid, - persistent, - on_node + client_id :: binary(), + sess_pid :: pid(), + persistent :: boolean() }). -type mqtt_session() :: #mqtt_session{}. diff --git a/src/emqttd_sm.erl b/src/emqttd_sm.erl index 99c99c83f..aeada1bc6 100644 --- a/src/emqttd_sm.erl +++ b/src/emqttd_sm.erl @@ -57,7 +57,6 @@ -define(SM_POOL, ?MODULE). -%% TODO... -define(SESSION_TIMEOUT, 60000). %%%============================================================================= @@ -158,7 +157,7 @@ prioritise_cast(_Msg, _Len, _State) -> 0. prioritise_info(_Msg, _Len, _State) -> - 1. + 2. %% persistent session handle_call({start_session, {false, ClientId, ClientPid}}, _From, State) -> @@ -170,6 +169,7 @@ handle_call({start_session, {false, ClientId, ClientPid}}, _From, State) -> {reply, resume_session(Session, ClientPid), State} end; +%% transient session handle_call({start_session, {true, ClientId, ClientPid}}, _From, State) -> case lookup_session(ClientId) of undefined -> @@ -216,12 +216,11 @@ create_session(CleanSess, ClientId, ClientPid) -> {ok, SessPid} -> Session = #mqtt_session{client_id = ClientId, sess_pid = SessPid, - persistent = not CleanSess, - on_node = node()}, + persistent = not CleanSess}, case insert_session(Session) of - {aborted, {conflict, Node}} -> - %% conflict with othe node? - lager:critical("Session ~s conflict with node ~p!", [ClientId, Node]), + {aborted, {conflict, ConflictPid}} -> + %% Conflict with othe node? + lager:critical("Session(~s): Conflict with ~p!", [ClientId, ConflictPid]), {error, conflict}; {atomic, ok} -> erlang:monitor(process, SessPid), @@ -232,89 +231,65 @@ create_session(CleanSess, ClientId, ClientPid) -> end. insert_session(Session = #mqtt_session{client_id = ClientId}) -> - mnesia:transaction(fun() -> - case mnesia:wread({session, ClientId}) of - [] -> - mnesia:write(session, Session, write); - [#mqtt_session{on_node = Node}] -> - mnesia:abort({conflict, Node}) - end - end). + mnesia:transaction( + fun() -> + case mnesia:wread({session, ClientId}) of + [] -> + mnesia:write(session, Session, write); + [#mqtt_session{sess_pid = SessPid}] -> + mnesia:abort({conflict, SessPid}) + end + end). -%% local node +%% Local node resume_session(#mqtt_session{client_id = ClientId, - sess_pid = SessPid, - on_node = Node}, ClientPid) - when Node =:= node() -> - case is_process_alive(SessPid) of - true -> - emqttd_session:resume(SessPid, ClientId, ClientPid), + sess_pid = SessPid}, ClientPid) + when node(SessPid) =:= node() -> + + emqttd_session:resume(SessPid, ClientId, ClientPid), + {ok, SessPid}; + +%% Remote node +resume_session(Session = #mqtt_session{client_id = ClientId, sess_pid = SessPid}, ClientPid) -> + Node = node(SessPid), + case rpc:call(Node, emqttd_session, resume, [SessPid, ClientId, ClientPid]) of + ok -> {ok, SessPid}; - false -> - lager:critical("Session ~s@~p died unexpectedly!", [ClientId, SessPid]), - {error, session_died} - end; - -%% remote node -resume_session(Session = #mqtt_session{client_id = ClientId, - sess_pid = SessPid, - on_node = Node}, ClientPid) -> - case emqttd:is_running(Node) of - true -> - case rpc:call(Node, emqttd_session, resume, [SessPid, ClientId, ClientPid]) of - ok -> - {ok, SessPid}; - {badrpc, Reason} -> - lager:critical("Resume session ~s on remote node ~p failed for ~p", - [ClientId, Node, Reason]), - {error, Reason} - end; - false -> - lager:critical("Session ~s died for node ~p down!", [ClientId, Node]), + {badrpc, nodedown} -> + lager:critical("Session(~s): Died for node ~s down!", [ClientId, Node]), remove_session(Session), - {error, session_node_down} + {error, session_nodedown}; + {badrpc, Reason} -> + lager:critical("Session(~s): Failed to resume from node ~s for ~p", + [ClientId, Node, Reason]), + {error, Reason} end. -%% local node -destroy_session(Session = #mqtt_session{client_id = ClientId, - sess_pid = SessPid, - on_node = Node}) when Node =:= node() -> - case is_process_alive(SessPid) of - true -> - emqttd_session:destroy(SessPid, ClientId); - false -> - lager:critical("Session ~s@~p died unexpectedly!", [ClientId, SessPid]) - end, - case remove_session(Session) of - {atomic, ok} -> ok; - {aborted, Error} -> {error, Error} - end; +%% Local node +destroy_session(Session = #mqtt_session{client_id = ClientId, sess_pid = SessPid}) + when node(SessPid) =:= node() -> + emqttd_session:destroy(SessPid, ClientId), + remove_session(Session); -%% remote node +%% Remote node destroy_session(Session = #mqtt_session{client_id = ClientId, - sess_pid = SessPid, - on_node = Node}) -> - case emqttd:is_running(Node) of - true -> - case rpc:call(Node, emqttd_session, destroy, [SessPid, ClientId]) of - ok -> - case remove_session(Session) of - {atomic, ok} -> ok; - {aborted, Error} -> {error, Error} - end; - {badrpc, Reason} -> - lager:critical("Destroy session ~s on remote node ~p failed for ~p", - [ClientId, Node, Reason]), - {error, list_to_atom("session_" ++ atom_to_list(Reason))} - end; - false -> - lager:error("Session ~s died for node ~p down!", [ClientId, Node]), - case remove_session(Session) of - {atomic, ok} -> ok; - {aborted, Error} -> {error, Error} - end - end. + sess_pid = SessPid}) -> + Node = node(SessPid), + case rpc:call(Node, emqttd_session, destroy, [SessPid, ClientId]) of + ok -> + remove_session(Session); + {badrpc, nodedown} -> + lager:error("Session(~s): Died for node ~s down!", [ClientId, Node]), + remove_session(Session); + {badrpc, Reason} -> + lager:error("Session(~s): Failed to destory ~p on remote node ~p for ~s", + [ClientId, SessPid, Node, Reason]), + {error, Reason} + end. remove_session(Session) -> - mnesia:transaction(fun() -> mnesia:delete_object(session, Session, write) end). + case mnesia:transaction(fun mnesia:delete_object/3, [session, Session, write]) of + {atomic, ok} -> ok; + {aborted, Error} -> {error, Error} + end. diff --git a/src/emqttd_sm_helper.erl b/src/emqttd_sm_helper.erl index b3a109ecd..0c32c6530 100644 --- a/src/emqttd_sm_helper.erl +++ b/src/emqttd_sm_helper.erl @@ -25,14 +25,14 @@ %%% @end %%%----------------------------------------------------------------------------- -%% TODO: Monitor mnesia node down... - -module(emqttd_sm_helper). -author("Feng Lee "). -include("emqttd.hrl"). +-include_lib("stdlib/include/ms_transform.hrl"). + %% API Function Exports -export([start_link/0]). @@ -42,7 +42,7 @@ -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). --record(state, {statsfun, ticker}). +-record(state, {stats_fun, tick_tref}). %%------------------------------------------------------------------------------ %% @doc Start a session helper @@ -53,9 +53,10 @@ start_link() -> gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). init([]) -> + %%mnesia:subscribe(system), + {ok, TRef} = timer:send_interval(1000, tick), StatsFun = emqttd_stats:statsfun('sessions/count', 'sessions/max'), - {ok, TRef} = timer:send_interval(1000, self(), tick), - {ok, #state{statsfun = StatsFun, ticker = TRef}}. + {ok, #state{stats_fun = StatsFun, tick_tref = TRef}}. handle_call(_Request, _From, State) -> {reply, ok, State}. @@ -64,20 +65,36 @@ handle_cast(Msg, State) -> lager:critical("Unexpected Msg: ~p", [Msg]), {noreply, State}. +handle_info({mnesia_system_event, {mnesia_down, Node}}, State) -> + lager:error("!!!Mnesia node down: ~s", [Node]), + Fun = fun() -> + ClientIds = + mnesia:select(session, [{#mqtt_session{client_id = '$1', sess_pid = '$2'}, + [{'==', {node, '$2'}, Node}], + ['$1']}]), + lists:foreach(fun(Id) -> mnesia:delete({session, Id}) end, ClientIds) + end, + mnesia:async_dirty(Fun), + {noreply, State}; + handle_info(tick, State) -> - {noreply, setstats(State)}; + {noreply, setstats(State), hibernate}; handle_info(Info, State) -> lager:critical("Unexpected Info: ~p", [Info]), {noreply, State}. -terminate(_Reason, _State = #state{ticker = TRef}) -> - timer:cancel(TRef). +terminate(_Reason, _State = #state{tick_tref = TRef}) -> + timer:cancel(TRef), + mnesia:unsubscribe(system). code_change(_OldVsn, State, _Extra) -> {ok, State}. -setstats(State = #state{statsfun = StatsFun}) -> +%%%============================================================================= +%%% Internal functions +%%%============================================================================= + +setstats(State = #state{stats_fun = StatsFun}) -> StatsFun(ets:info(mqtt_persistent_session, size)), State. - From fa84a2595ba53b661d10b56841f8e12fae6cd621 Mon Sep 17 00:00:00 2001 From: Feng Lee Date: Sat, 10 Oct 2015 13:30:50 +0800 Subject: [PATCH 2/4] critical -> error --- src/emqttd_broker.erl | 2 +- src/emqttd_sm.erl | 10 +++++----- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/src/emqttd_broker.erl b/src/emqttd_broker.erl index 5a21eb451..19dd6237a 100644 --- a/src/emqttd_broker.erl +++ b/src/emqttd_broker.erl @@ -257,7 +257,7 @@ handle_call({unhook, Hook, Name}, _From, State) -> {reply, Reply, State}; handle_call(Req, _From, State) -> - lager:critical("Unexpected request: ~p", [Req]), + lager:error("Unexpected request: ~p", [Req]), {reply, {error, badreq}, State}. handle_cast(_Msg, State) -> diff --git a/src/emqttd_sm.erl b/src/emqttd_sm.erl index aeada1bc6..e40bde20f 100644 --- a/src/emqttd_sm.erl +++ b/src/emqttd_sm.erl @@ -187,7 +187,7 @@ handle_call(_Request, _From, State) -> {reply, ok, State}. handle_cast(Msg, State) -> - lager:critical("Unexpected Msg: ~p", [Msg]), + lager:error("Unexpected Msg: ~p", [Msg]), {noreply, State}. handle_info({'DOWN', _MRef, process, DownPid, _Reason}, State) -> @@ -198,7 +198,7 @@ handle_info({'DOWN', _MRef, process, DownPid, _Reason}, State) -> {noreply, State}; handle_info(Info, State) -> - lager:critical("Unexpected Info: ~p", [Info]), + lager:error("Unexpected Info: ~p", [Info]), {noreply, State}. terminate(_Reason, #state{id = Id}) -> @@ -220,7 +220,7 @@ create_session(CleanSess, ClientId, ClientPid) -> case insert_session(Session) of {aborted, {conflict, ConflictPid}} -> %% Conflict with othe node? - lager:critical("Session(~s): Conflict with ~p!", [ClientId, ConflictPid]), + lager:error("Session(~s): Conflict with ~p!", [ClientId, ConflictPid]), {error, conflict}; {atomic, ok} -> erlang:monitor(process, SessPid), @@ -256,11 +256,11 @@ resume_session(Session = #mqtt_session{client_id = ClientId, sess_pid = SessPid} ok -> {ok, SessPid}; {badrpc, nodedown} -> - lager:critical("Session(~s): Died for node ~s down!", [ClientId, Node]), + lager:error("Session(~s): Died for node ~s down!", [ClientId, Node]), remove_session(Session), {error, session_nodedown}; {badrpc, Reason} -> - lager:critical("Session(~s): Failed to resume from node ~s for ~p", + lager:error("Session(~s): Failed to resume from node ~s for ~p", [ClientId, Node, Reason]), {error, Reason} end. From f57e6b43db9f752f35cc3d5983a14281efcd42a3 Mon Sep 17 00:00:00 2001 From: Feng Lee Date: Sat, 10 Oct 2015 13:36:06 +0800 Subject: [PATCH 3/4] critical -> error --- src/emqttd_client.erl | 8 ++++---- src/emqttd_cm.erl | 4 ++-- src/emqttd_pubsub.erl | 2 +- src/emqttd_retained.erl | 4 ++-- src/emqttd_session.erl | 6 +++--- src/emqttd_sm_helper.erl | 8 ++++---- src/emqttd_ws_client.erl | 2 +- 7 files changed, 17 insertions(+), 17 deletions(-) diff --git a/src/emqttd_client.erl b/src/emqttd_client.erl index 7eb4be8f4..5acb9e5d3 100644 --- a/src/emqttd_client.erl +++ b/src/emqttd_client.erl @@ -110,7 +110,7 @@ handle_call(kick, _From, State) -> {stop, {shutdown, kick}, ok, State}; handle_call(Req, _From, State = #state{peername = Peername}) -> - lager:critical("Client ~s: unexpected request - ~p", [emqttd_net:format(Peername), Req]), + lager:error("Client ~s: unexpected request - ~p", [emqttd_net:format(Peername), Req]), {reply, {error, unsupported_request}, State}. handle_cast({subscribe, TopicTable}, State) -> @@ -120,7 +120,7 @@ handle_cast({unsubscribe, Topics}, State) -> with_session(fun(SessPid) -> emqttd_session:unsubscribe(SessPid, Topics) end, State); handle_cast(Msg, State = #state{peername = Peername}) -> - lager:critical("Client ~s: unexpected msg - ~p",[emqttd_net:format(Peername), Msg]), + lager:error("Client ~s: unexpected msg - ~p",[emqttd_net:format(Peername), Msg]), {noreply, State}. handle_info(timeout, State) -> @@ -152,7 +152,7 @@ handle_info({inet_async, _Sock, _Ref, {error, Reason}}, State) -> network_error(Reason, State); handle_info({inet_reply, _Sock, {error, Reason}}, State = #state{peername = Peername}) -> - lager:critical("Client ~s: unexpected inet_reply '~p'", [emqttd_net:format(Peername), Reason]), + lager:error("Client ~s: unexpected inet_reply '~p'", [emqttd_net:format(Peername), Reason]), {noreply, State}; handle_info({keepalive, start, TimeoutSec}, State = #state{transport = Transport, socket = Socket, peername = Peername}) -> @@ -180,7 +180,7 @@ handle_info({keepalive, check}, State = #state{peername = Peername, keepalive = end; handle_info(Info, State = #state{peername = Peername}) -> - lager:critical("Client ~s: unexpected info ~p",[emqttd_net:format(Peername), Info]), + lager:error("Client ~s: unexpected info ~p",[emqttd_net:format(Peername), Info]), {noreply, State}. terminate(Reason, #state{peername = Peername, diff --git a/src/emqttd_cm.erl b/src/emqttd_cm.erl index d60ddc78d..db198c176 100644 --- a/src/emqttd_cm.erl +++ b/src/emqttd_cm.erl @@ -129,11 +129,11 @@ handle_cast({unregister, ClientId, Pid}, State) -> {noreply, setstats(State)}; handle_cast(Msg, State) -> - lager:critical("Unexpected Msg: ~p", [Msg]), + lager:error("Unexpected Msg: ~p", [Msg]), {noreply, State}. handle_info(Info, State) -> - lager:critical("Unexpected Msg: ~p", [Info]), + lager:error("Unexpected Msg: ~p", [Info]), {noreply, State}. terminate(_Reason, #state{id = Id}) -> diff --git a/src/emqttd_pubsub.erl b/src/emqttd_pubsub.erl index 8bd4534e0..4579be7c2 100644 --- a/src/emqttd_pubsub.erl +++ b/src/emqttd_pubsub.erl @@ -385,7 +385,7 @@ handle_info({'DOWN', _Mon, _Type, DownPid, _Info}, State = #state{submap = SubMa end; handle_info(Info, State) -> - lager:critical("Unexpected Info: ~p", [Info]), + lager:error("Unexpected Info: ~p", [Info]), {noreply, State}. terminate(_Reason, _State) -> diff --git a/src/emqttd_retained.erl b/src/emqttd_retained.erl index 87a1193a2..b82570800 100644 --- a/src/emqttd_retained.erl +++ b/src/emqttd_retained.erl @@ -162,7 +162,7 @@ handle_call(_Request, _From, State) -> {reply, ok, State}. handle_cast(Msg, State) -> - lager:critical("Unexpected Msg: ~p", [Msg]), + lager:error("Unexpected Msg: ~p", [Msg]), {noreply, State}. handle_info(stats, State = #state{stats_fun = StatsFun}) -> @@ -174,7 +174,7 @@ handle_info(expire, State = #state{expired_after = ExpiredAfter}) -> {noreply, State, hibernate}; handle_info(Info, State) -> - lager:critical("Unexpected Info: ~p", [Info]), + lager:error("Unexpected Info: ~p", [Info]), {noreply, State}. terminate(_Reason, _State = #state{stats_timer = TRef1, expire_timer = TRef2}) -> diff --git a/src/emqttd_session.erl b/src/emqttd_session.erl index a23f76f7d..4cb927ef4 100644 --- a/src/emqttd_session.erl +++ b/src/emqttd_session.erl @@ -293,7 +293,7 @@ handle_call({publish, Msg = #mqtt_message{qos = ?QOS_2, pktid = PktId}}, _From, end; handle_call(Req, _From, State) -> - lager:critical("Unexpected Request: ~p", [Req]), + lager:error("Unexpected Request: ~p", [Req]), {reply, ok, State}. handle_cast({subscribe, TopicTable0, Callback}, Session = #session{ @@ -469,7 +469,7 @@ handle_cast({pubcomp, PktId}, Session = #session{client_id = ClientId, awaiting_ end; handle_cast(Msg, State) -> - lager:critical("Unexpected Msg: ~p, State: ~p", [Msg, State]), + lager:error("Unexpected Msg: ~p, State: ~p", [Msg, State]), {noreply, State}. %% Queue messages when client is offline @@ -570,7 +570,7 @@ handle_info(session_expired, Session = #session{client_id = ClientId}) -> {stop, {shutdown, expired}, Session}; handle_info(Info, Session = #session{client_id = ClientId}) -> - lager:critical("Session(~s) unexpected info: ~p", [ClientId, Info]), + lager:error("Session(~s) unexpected info: ~p", [ClientId, Info]), {noreply, Session}. terminate(_Reason, #session{clean_sess = CleanSess, client_id = ClientId}) -> diff --git a/src/emqttd_sm_helper.erl b/src/emqttd_sm_helper.erl index 0c32c6530..4e953b8b2 100644 --- a/src/emqttd_sm_helper.erl +++ b/src/emqttd_sm_helper.erl @@ -53,7 +53,7 @@ start_link() -> gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). init([]) -> - %%mnesia:subscribe(system), + mnesia:subscribe(system), {ok, TRef} = timer:send_interval(1000, tick), StatsFun = emqttd_stats:statsfun('sessions/count', 'sessions/max'), {ok, #state{stats_fun = StatsFun, tick_tref = TRef}}. @@ -62,7 +62,7 @@ handle_call(_Request, _From, State) -> {reply, ok, State}. handle_cast(Msg, State) -> - lager:critical("Unexpected Msg: ~p", [Msg]), + lager:error("Unexpected Msg: ~p", [Msg]), {noreply, State}. handle_info({mnesia_system_event, {mnesia_down, Node}}, State) -> @@ -72,7 +72,7 @@ handle_info({mnesia_system_event, {mnesia_down, Node}}, State) -> mnesia:select(session, [{#mqtt_session{client_id = '$1', sess_pid = '$2'}, [{'==', {node, '$2'}, Node}], ['$1']}]), - lists:foreach(fun(Id) -> mnesia:delete({session, Id}) end, ClientIds) + lists:foreach(fun(ClientId) -> mnesia:delete({session, ClientId}) end, ClientIds) end, mnesia:async_dirty(Fun), {noreply, State}; @@ -81,7 +81,7 @@ handle_info(tick, State) -> {noreply, setstats(State), hibernate}; handle_info(Info, State) -> - lager:critical("Unexpected Info: ~p", [Info]), + lager:error("Unexpected Info: ~p", [Info]), {noreply, State}. terminate(_Reason, _State = #state{tick_tref = TRef}) -> diff --git a/src/emqttd_ws_client.erl b/src/emqttd_ws_client.erl index 827d7de18..fb83a4539 100644 --- a/src/emqttd_ws_client.erl +++ b/src/emqttd_ws_client.erl @@ -206,7 +206,7 @@ handle_info({'EXIT', WsPid, Reason}, State = #client_state{ws_pid = WsPid, proto stop({shutdown, websocket_closed}, State); handle_info(Info, State = #client_state{request = Req}) -> - lager:critical("Client(WebSocket) ~s: Unexpected Info - ~p", [Req:get(peer), Info]), + lager:error("Client(WebSocket) ~s: Unexpected Info - ~p", [Req:get(peer), Info]), noreply(State). terminate(Reason, #client_state{proto_state = ProtoState, keepalive = KeepAlive}) -> From b59a8664eb9050afc6f84fa36a9d40e61c224acb Mon Sep 17 00:00:00 2001 From: Feng Lee Date: Sat, 10 Oct 2015 14:32:33 +0800 Subject: [PATCH 4/4] fix issue #282 - emqttd_sm optimized --- src/emqttd_sm.erl | 10 ++++++++-- src/emqttd_sm_helper.erl | 7 +++++-- 2 files changed, 13 insertions(+), 4 deletions(-) diff --git a/src/emqttd_sm.erl b/src/emqttd_sm.erl index e40bde20f..49db5be9c 100644 --- a/src/emqttd_sm.erl +++ b/src/emqttd_sm.erl @@ -246,8 +246,14 @@ resume_session(#mqtt_session{client_id = ClientId, sess_pid = SessPid}, ClientPid) when node(SessPid) =:= node() -> - emqttd_session:resume(SessPid, ClientId, ClientPid), - {ok, SessPid}; + case is_process_alive(SessPid) of + true -> + emqttd_session:resume(SessPid, ClientId, ClientPid), + {ok, SessPid}; + false -> + lager:error("Session(~s): Cannot resume ~p, it seems already dead!", [ClientId, SessPid]), + {error, session_died} + end; %% Remote node resume_session(Session = #mqtt_session{client_id = ClientId, sess_pid = SessPid}, ClientPid) -> diff --git a/src/emqttd_sm_helper.erl b/src/emqttd_sm_helper.erl index 4e953b8b2..a3387b6a3 100644 --- a/src/emqttd_sm_helper.erl +++ b/src/emqttd_sm_helper.erl @@ -69,14 +69,17 @@ handle_info({mnesia_system_event, {mnesia_down, Node}}, State) -> lager:error("!!!Mnesia node down: ~s", [Node]), Fun = fun() -> ClientIds = - mnesia:select(session, [{#mqtt_session{client_id = '$1', sess_pid = '$2'}, + mnesia:select(session, [{#mqtt_session{client_id = '$1', sess_pid = '$2', _ = '_'}, [{'==', {node, '$2'}, Node}], ['$1']}]), - lists:foreach(fun(ClientId) -> mnesia:delete({session, ClientId}) end, ClientIds) + lists:foreach(fun(ClientId) -> mnesia:delete({session, ClientId}) end, ClientIds) end, mnesia:async_dirty(Fun), {noreply, State}; +handle_info({mnesia_system_event, {mnesia_up, _Node}}, State) -> + {noreply, State}; + handle_info(tick, State) -> {noreply, setstats(State), hibernate};