Merge pull request #112 from rabbitmq/cache-reader-context

kjnilsson · web-flow · commit d68042d4f42c · 2022-11-30T12:12:19.000Z
Cache reader context
diff --git a/WORKSPACE.bazel b/WORKSPACE.bazel
@@ -40,7 +40,7 @@ filegroup(
     visibility = ["//visibility:public"],
 )
 """,
-    sha256 = "22d1f92c04cc41e19b2c332c958f2d5c364a1c7ae78549041187e9e0a0080bf3",
+    # sha256 = "22d1f92c04cc41e19b2c332c958f2d5c364a1c7ae78549041187e9e0a0080bf3",
     strip_prefix = "tls-gen-main",
     urls = ["https://github.com/rabbitmq/tls-gen/archive/refs/heads/main.zip"],
 )
diff --git a/src/osiris.erl b/src/osiris.erl
@@ -196,7 +196,7 @@ init_reader(Pid, OffsetSpec, CounterSpec) ->
 init_reader(Pid, OffsetSpec, {_, _} = CounterSpec, Options)
     when is_pid(Pid) andalso node(Pid) =:= node() ->
     ?DEBUG("osiris: initialising reader. Spec: ~w", [OffsetSpec]),
-    {ok, Ctx0} = gen:call(Pid, '$gen_call', get_reader_context),
+    Ctx0 = osiris_util:get_reader_context(Pid),
     Ctx = Ctx0#{counter_spec => CounterSpec,
                 options => Options},
     osiris_log:init_offset_reader(OffsetSpec, Ctx).
@@ -274,7 +274,7 @@ configure_logger(Module) ->
                             first_chunk_id => integer()}.
 get_stats(Pid)
   when node(Pid) =:= node() ->
-    {ok, #{shared := Shared}} = gen:call(Pid, '$gen_call', get_reader_context),
+    #{shared := Shared} = osiris_util:get_reader_context(Pid),
     #{committed_chunk_id => osiris_log_shared:committed_chunk_id(Shared),
       first_chunk_id => osiris_log_shared:first_chunk_id(Shared),
       last_chunk_id => osiris_log_shared:last_chunk_id(Shared)};
diff --git a/src/osiris_ets.erl b/src/osiris_ets.erl
@@ -0,0 +1,77 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2022 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(osiris_ets).
+
+-behaviour(gen_server).
+
+%% API functions
+-export([start_link/0]).
+%% gen_server callbacks
+-export([init/1,
+         handle_call/3,
+         handle_cast/2,
+         handle_info/2,
+         terminate/2,
+         code_change/3]).
+
+-record(state, {}).
+
+%%%===================================================================
+%%% API functions
+%%%===================================================================
+
+%% @spec start_link() -> {ok, Pid} | ignore | {error, Error}
+start_link() ->
+    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+
+%%%===================================================================
+%%% gen_server callbacks
+%%%===================================================================
+
+% @spec init(Args) -> {ok, State} |
+%%                     {ok, State, Timeout} |
+%%                     ignore |
+%%                     {stop, Reason}
+init(_) ->
+    _ = ets:new(osiris_reader_context_cache, [set, named_table, public]),
+    {ok, #state{}}.
+
+%% @spec handle_call(Request, From, State) ->
+%%                                   {reply, Reply, State} |
+%%                                   {reply, Reply, State, Timeout} |
+%%                                   {noreply, State} |
+%%                                   {noreply, State, Timeout} |
+%%                                   {stop, Reason, Reply, State} |
+%%                                   {stop, Reason, State}
+handle_call(_Request, _From, State) ->
+    Reply = ok,
+    {reply, Reply, State}.
+
+%% @spec handle_cast(Msg, State) -> {noreply, State} |
+%%                                  {noreply, State, Timeout} |
+%%                                  {stop, Reason, State}
+handle_cast(_Msg, State) ->
+    {noreply, State}.
+
+%% @spec handle_info(Info, State) -> {noreply, State} |
+%%                                   {noreply, State, Timeout} |
+%%                                   {stop, Reason, State}
+handle_info(_Info, State) ->
+    {noreply, State}.
+
+%% @spec terminate(Reason, State) -> void()
+terminate(_Reason, _State) ->
+    ok.
+
+%% @spec code_change(OldVsn, State, Extra) -> {ok, NewState}
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
+
+%%%===================================================================
+%%% Internal functions
+%%%===================================================================
diff --git a/src/osiris_log.erl b/src/osiris_log.erl
@@ -53,6 +53,7 @@
 
 % maximum size of a segment in bytes
 -define(DEFAULT_MAX_SEGMENT_SIZE_B, 500 * 1000 * 1000).
+% -define(MIN_SEGMENT_SIZE_B, 4_000_000).
 % maximum number of chunks per segment
 -define(DEFAULT_MAX_SEGMENT_SIZE_C, 256_000).
 -define(INDEX_RECORD_SIZE_B, 29).
@@ -442,8 +443,8 @@ init(#{dir := Dir,
        epoch := Epoch} = Config,
      WriterType) ->
     %% scan directory for segments if in write mode
-    MaxSizeBytes =
-        maps:get(max_segment_size_bytes, Config, ?DEFAULT_MAX_SEGMENT_SIZE_B),
+    MaxSizeBytes = maps:get(max_segment_size_bytes, Config,
+                            ?DEFAULT_MAX_SEGMENT_SIZE_B),
     MaxSizeChunks = application:get_env(osiris, max_segment_size_chunks,
                                         ?DEFAULT_MAX_SEGMENT_SIZE_C),
     Retention = maps:get(retention, Config, []),
@@ -479,20 +480,21 @@ init(#{dir := Dir,
                counter_id = counter_id(Config),
                shared = Shared},
     ok = maybe_fix_corrupted_files(Config),
+    DefaultNextOffset = case Config of
+                            #{initial_offset := IO}
+                              when WriterType == acceptor ->
+                                IO;
+                            _ ->
+                                0
+                        end,
     case first_and_last_seginfos(Config) of
         none ->
-            NextOffset = case Config of
-                             #{initial_offset := IO}
-                               when WriterType == acceptor ->
-                                 IO;
-                             _ ->
-                                 0
-                         end,
-            osiris_log_shared:set_first_chunk_id(Shared, NextOffset - 1),
+            osiris_log_shared:set_first_chunk_id(Shared, DefaultNextOffset - 1),
             open_new_segment(#?MODULE{cfg = Cfg,
                                       mode =
                                           #write{type = WriterType,
-                                                 tail_info = {NextOffset, empty},
+                                                 tail_info = {DefaultNextOffset,
+                                                              empty},
                                                  current_epoch = Epoch}});
         {NumSegments,
          #seg_info{first = #chunk_info{id = FstChId,
@@ -555,11 +557,11 @@ init(#{dir := Dir,
             %% here too?
             {ok, _} = file:position(SegFd, eof),
             {ok, _} = file:position(IdxFd, eof),
-            osiris_log_shared:set_first_chunk_id(Shared, -1),
+            osiris_log_shared:set_first_chunk_id(Shared, DefaultNextOffset - 1),
             #?MODULE{cfg = Cfg,
                      mode =
                          #write{type = WriterType,
-                                tail_info = {0, empty},
+                                tail_info = {DefaultNextOffset, empty},
                                 current_epoch = Epoch},
                      current_file = filename:basename(Filename),
                      fd = SegFd,
@@ -847,16 +849,19 @@ truncate_to(_Name, _Range, [], IdxFiles) ->
     [];
 truncate_to(Name, RemoteRange, [{E, ChId} | NextEOs], IdxFiles) ->
     case find_segment_for_offset(ChId, IdxFiles) of
-        not_found ->
+        Result when Result == not_found orelse
+                    element(1, Result) == end_of_log ->
+            %% both not_found and end_of_log needs to be treated as not found
+            %% as they are...
             case build_seg_info(lists:last(IdxFiles)) of
                 {ok, #seg_info{last = #chunk_info{epoch = E,
                                                   id = LastChId,
                                                   num = Num}}}
-                when ChId > LastChId + Num ->
+                when ChId > LastChId ->
                     %% the last available local chunk id is smaller than the
-                    %% sources last chunk id but is in the same epoch
+                    %% source's last chunk id but is in the same epoch
                     %% check if there is any overlap
-                    LastOffsLocal = LastChId + Num,
+                    LastOffsLocal = LastChId + Num - 1,
                     FstOffsetRemote = case RemoteRange of
                                           empty -> 0;
                                           {F, _} -> F
@@ -878,8 +883,6 @@ truncate_to(Name, RemoteRange, [{E, ChId} | NextEOs], IdxFiles) ->
                     %% TODO: what to do if error is returned from
                     %% build_seg_info/1?
             end;
-        {end_of_log, _Info} ->
-            IdxFiles;
         {found, #seg_info{file = File, index = IdxFile}} ->
             ?DEBUG("osiris_log: ~s on node ~s truncating to chunk "
                    "id ~b in epoch ~b",
@@ -1502,18 +1505,21 @@ send_file(Sock,
                 true ->
                     %% this avoids any data sent in the Callback to be dispatched
                     %% in it's own TCP frame
-                    ok = setopts(Transport, Sock, [{nopush, true}]),
-                    _ = Callback(Header, ToSend),
-                    case sendfile(Transport, Fd, Sock, Pos, ToSend) of
+                    case setopts(Transport, Sock, [{nopush, true}]) of
                         ok ->
-                            ok = setopts(Transport, Sock, [{nopush, false}]),
-                            {ok, _} = file:position(Fd, NextFilePos),
-                            {ok, State};
-                        Err ->
-                            %% reset the position to the start of the current
-                            %% chunk so that subsequent reads won't error
-                            {ok, _} = file:position(Fd, Pos),
-                            Err
+                            _ = Callback(Header, ToSend),
+                            case sendfile(Transport, Fd, Sock, Pos, ToSend) of
+                                ok ->
+                                    ok = setopts(Transport, Sock, [{nopush, false}]),
+                                    {ok, _} = file:position(Fd, NextFilePos),
+                                    {ok, State};
+                                Err ->
+                                    %% reset the position to the start of the current
+                                    %% chunk so that subsequent reads won't error
+                                    {ok, _} = file:position(Fd, Pos),
+                                    Err
+                            end;
+                        Err -> Err
                     end;
                 false ->
                     {ok, _} = file:position(Fd, NextFilePos),
@@ -1997,10 +2003,10 @@ last_epoch_offsets([IdxFile]) ->
 last_epoch_offsets([FstIdxFile | _]  = IdxFiles) ->
     F = fun() ->
                 {ok, FstFd} = open(FstIdxFile, [read, raw, binary]),
-		%% on linux this disables read-ahead so should only
-		%% bring a single block into memory
-		%% having the first block of index files in page cache
-		%% should generally be a good thing
+                %% on linux this disables read-ahead so should only
+                %% bring a single block into memory
+                %% having the first block of index files in page cache
+                %% should generally be a good thing
                 _ = file:advise(FstFd, 0, 0, random),
                 {ok, <<FstO:64/unsigned,
                        _FstTimestamp:64/signed,
@@ -2186,9 +2192,9 @@ max_segment_size_reached(
     CurrentSizeChunks >= MaxSizeChunks.
 
 setopts(tcp, Sock, Opts) ->
-    ok = inet:setopts(Sock, Opts);
+    inet:setopts(Sock, Opts);
 setopts(ssl, Sock, Opts) ->
-    ok = ssl:setopts(Sock, Opts).
+    ssl:setopts(Sock, Opts).
 
 sendfile(_Transport, _Fd, _Sock, _Pos, 0) ->
     ok;
diff --git a/src/osiris_replica.erl b/src/osiris_replica.erl
@@ -197,7 +197,7 @@ handle_continue(#{name := Name0,
                     %% re-discover the committed offset
                     osiris_writer:ack(LeaderPid, {LastChId, LastTs})
             end,
-            ?INFO_(Name, "osiris replicas starting in epoch ~b, next offset ~b, tail info ~w",
+            ?INFO_(Name, "osiris replica starting in epoch ~b, next offset ~b, tail info ~w",
                    [Epoch, NextOffset, TailInfo]),
 
             %% HostName: append the HostName to the Ip(s) list: in some cases
@@ -246,8 +246,8 @@ handle_continue(#{name := Name0,
             RRPid = osiris_replica_reader:start(Node, ReplicaReaderConf),
             true = link(RRPid),
             GcInterval0 = application:get_env(osiris,
-                                             replica_forced_gc_default_interval,
-                                             4999),
+                                              replica_forced_gc_default_interval,
+                                              4999),
 
             GcInterval1 = case is_integer(GcInterval0) of
                               true ->
@@ -258,6 +258,11 @@ handle_continue(#{name := Name0,
                           end,
             counters:put(CntRef, ?C_COMMITTED_OFFSET, -1),
             counters:put(CntRef, ?C_EPOCH, Epoch),
+            Shared = osiris_log:get_shared(Log),
+            osiris_util:cache_reader_context(self(), Dir, Name, Shared, ExtRef,
+                                             fun(Inc) ->
+                                                     counters:add(CntRef, ?C_READERS, Inc)
+                                             end),
             EvtFmt = maps:get(event_formatter, Config, undefined),
             {noreply,
              #?MODULE{cfg =
@@ -570,6 +575,7 @@ terminate(_Reason, undefined) ->
 terminate(Reason, #?MODULE{cfg = #cfg{name = Name,
                                       socket = Sock}, log = Log}) ->
     ?DEBUG_(Name, "terminating with ~w ", [Reason]),
+    _ = ets:delete(osiris_reader_context_cache, self()),
     ok = osiris_log:close(Log),
     ok = gen_tcp:close(Sock),
     ok.
diff --git a/src/osiris_sup.erl b/src/osiris_sup.erl
@@ -22,6 +22,10 @@ init([]) ->
           intensity => 5,
           period => 5},
     %% todo put under own sup
+    Ets =
+        #{id => osiris_ets,
+          type => worker,
+          start => {osiris_ets, start_link, []}},
     Retention =
         #{id => osiris_retention,
           type => worker,
@@ -34,4 +38,4 @@ init([]) ->
         #{id => osiris_replica_reader_sup,
           type => supervisor,
           start => {osiris_replica_reader_sup, start_link, []}},
-    {ok, {SupFlags, [Retention, ServerSup, ReplicaReader]}}.
+    {ok, {SupFlags, [Ets, Retention, ServerSup, ReplicaReader]}}.
diff --git a/src/osiris_util.erl b/src/osiris_util.erl
@@ -18,7 +18,9 @@
          get_replication_configuration_from_tls_dist/1,
          get_replication_configuration_from_tls_dist/2,
          partition_parallel/3,
-         normalise_name/1
+         normalise_name/1,
+         get_reader_context/1,
+         cache_reader_context/6
         ]).
 
 %% For testing
@@ -267,3 +269,27 @@ normalise_name(Name) when is_binary(Name) ->
     Name;
 normalise_name(Name) when is_list(Name) ->
     list_to_binary(Name).
+
+get_reader_context(Pid)
+  when is_pid(Pid) andalso node(Pid) == node() ->
+    case ets:lookup(osiris_reader_context_cache, Pid) of
+        [] ->
+            {ok, Ctx0} = gen:call(Pid, '$gen_call', get_reader_context),
+            Ctx0;
+        [{_Pid, Dir, Name, Shared, Ref, ReadersCountersFun}] ->
+            #{dir => Dir,
+              name => Name,
+              shared => Shared,
+              reference => Ref,
+              readers_counter_fun => ReadersCountersFun}
+    end.
+
+cache_reader_context(Pid, Dir, Name, Shared, Ref, ReadersCounterFun)
+  when is_pid(Pid) andalso
+       ?IS_STRING(Dir) andalso
+       is_function(ReadersCounterFun) ->
+    true = ets:insert(osiris_reader_context_cache,
+                      {Pid, Dir, Name, Shared, Ref, ReadersCounterFun}),
+    ok.
+
+
diff --git a/src/osiris_writer.erl b/src/osiris_writer.erl
diff --git a/test/osiris_log_SUITE.erl b/test/osiris_log_SUITE.erl

Original file line number	Diff line number	Diff line change
`@@ -40,7 +40,7 @@ filegroup(`
`40`	`40`	`visibility = ["//visibility:public"],`
`41`	`41`	`)`
`42`	`42`	`""",`
`43`		`- sha256 = "22d1f92c04cc41e19b2c332c958f2d5c364a1c7ae78549041187e9e0a0080bf3",`
	`43`	`+ # sha256 = "22d1f92c04cc41e19b2c332c958f2d5c364a1c7ae78549041187e9e0a0080bf3",`
`44`	`44`	`strip_prefix = "tls-gen-main",`
`45`	`45`	`urls = ["https://github.com/rabbitmq/tls-gen/archive/refs/heads/main.zip"],`
`46`	`46`	`)`