yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor(dsrepl): make shard allocator more robust and consistent 

Co-Authored-By: Thales Macedo Garitezi <thalesmg@gmail.com>
This commit is contained in:
Andrew Mayorov 2024-03-19 15:11:35 +01:00
parent 0e18bd6e80
commit 404e919494
No known key found for this signature in database
GPG Key ID: 2837C62ACFBFED5D
1 changed files with 6 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
apps/emqx_durable_storage/src/emqx_ds_replication_shard_allocator.erl
View File

@ -54,7 +54,7 @@ init({DB, Opts}) ->
_ = logger:set_process_metadata(#{db => DB, domain => [ds, db, shard_allocator]}), _ = logger:set_process_metadata(#{db => DB, domain => [ds, db, shard_allocator]}),
State = #{db => DB, opts => Opts, status => allocating}, State = #{db => DB, opts => Opts, status => allocating},
case allocate_shards(State) of case allocate_shards(State) of
NState = #{} -> {ok, NState} ->
{ok, NState}; {ok, NState};
{error, Data} -> {error, Data} ->
_ = logger:notice( _ = logger:notice(
@ -74,7 +74,7 @@ handle_cast(_Cast, State) ->
handle_info(timeout, State) -> handle_info(timeout, State) ->
case allocate_shards(State) of case allocate_shards(State) of
NState = #{} -> {ok, NState} ->
{noreply, NState}; {noreply, NState};
{error, Data} -> {error, Data} ->
_ = logger:notice( _ = logger:notice(
@ -84,7 +84,9 @@ handle_info(timeout, State) ->
} }
), ),
{noreply, State, ?ALLOCATE_RETRY_TIMEOUT} {noreply, State, ?ALLOCATE_RETRY_TIMEOUT}
end. end;
handle_info(_Info, State) ->
{noreply, State}.
terminate(_Reason, #{db := DB, shards := Shards}) -> terminate(_Reason, #{db := DB, shards := Shards}) ->
erase_db_meta(DB), erase_db_meta(DB),
@ -102,7 +104,7 @@ allocate_shards(State = #{db := DB, opts := Opts}) ->
ok = start_egresses(DB, Shards), ok = start_egresses(DB, Shards),
ok = save_db_meta(DB, Shards), ok = save_db_meta(DB, Shards),
ok = save_shards_meta(DB, Shards), ok = save_shards_meta(DB, Shards),
State#{shards => Shards, status := ready}; {ok, State#{shards => Shards, status := ready}};
{error, Reason} -> {error, Reason} ->
{error, Reason} {error, Reason}
end. end.