Merge pull request #9 from rabbitmq/writer_dedupe

writer deduplication

Author: acogoluegnes

src/osiris.erl

@@ -9,9 +9,10 @@
 
 -include("osiris.hrl").
 -export([
-         write/3,
+         write/4,
          write_tracking/3,
          read_tracking/2,
+         fetch_writer_seq/2,
          init_reader/2,
          register_offset_listener/2,
          register_offset_listener/3,
@@ -50,14 +51,21 @@
 -type retention_spec() :: {max_bytes, non_neg_integer()} |
                           {max_age, milliseconds()}.
 
+-type writer_id() :: binary().
+-type data() :: iodata() |
+                {batch, non_neg_integer(), 0, iodata()}.
+
 -export_type([
               state/0,
               config/0,
               offset/0,
               epoch/0,
               tail_info/0,
               offset_spec/0,
-              retention_spec/0
+              retention_spec/0,
+              milliseconds/0,
+              writer_id/0,
+              data/0
               ]).
 
 -spec start_cluster(config()) ->
@@ -97,8 +105,12 @@ start_writer(Config) ->
 start_replica(Replica, Config) ->
     osiris_replica:start(Replica, Config).
 
-write(Pid, Corr, Data) ->
-    osiris_writer:write(Pid, self(), Corr, Data).
+-spec write(Pid :: pid(),
+            WriterId :: binary() | undefined,
+            CorrOrSeq :: non_neg_integer() | term(),
+            Data :: data()) -> ok.
+write(Pid, WriterId, Corr, Data) ->
+    osiris_writer:write(Pid, self(), WriterId, Corr, Data).
 
 -spec write_tracking(pid(), binary(), offset()) -> ok.
 write_tracking(Pid, TrackingId, Offset) ->
@@ -108,6 +120,18 @@ write_tracking(Pid, TrackingId, Offset) ->
 read_tracking(Pid, TrackingId) ->
     osiris_writer:read_tracking(Pid, TrackingId).
 
+-spec fetch_writer_seq(pid(), binary()) ->
+    non_neg_integer() | undefined.
+fetch_writer_seq(Pid, WriterId) when is_pid(Pid) andalso is_binary(WriterId) ->
+    osiris_writer:query_writers(Pid,
+                                fun(W) ->
+                                        case maps:get(WriterId, W, undefined) of
+                                            undefined -> undefined;
+                                            {_, _, Seq} ->
+                                                Seq
+                                        end
+                                end).
+
 %% @doc Initialise a new offset reader
 %% @param Pid the pid of a writer or replica process
 %% @param OffsetSpec specifies where in the log to attach the reader

src/osiris.hrl

@@ -29,12 +29,13 @@
 -define(MAGIC, 5).
 %% chunk format version
 -define(VERSION, 0).
--define(HEADER_SIZE_B, 40).
+-define(HEADER_SIZE_B, 44).
 -define(FILE_OPTS_WRITE, [raw, binary, write, read]).
 
 
 %% chunk types
 -define(CHNK_USER, 0).
 -define(CHNK_TRK_DELTA, 1).
 -define(CHNK_TRK_SNAPSHOT, 2).
+-define(CHNK_WRT_SNAPSHOT, 3).
 

src/osiris_bench.erl

@@ -83,7 +83,7 @@ do_publish0(Conf, 0) ->
     end;
 do_publish0(#{leader := Leader} = Conf, InFlight) ->
     Ref = make_ref(),
-    ok = osiris:write(Leader, Ref, <<"datadata">>),
+    ok = osiris:write(Leader, undefined, Ref, <<"datadata">>),
     do_publish0(Conf, InFlight - 1).
 
 start_metrics_gatherer(Node) ->