this repo has no description
Pop queue once all virtual tags have been popped
+86
-15
+79
-15
lib/servers/tlog.ex
+79
-15
lib/servers/tlog.ex
···
18
18
id: non_neg_integer,
19
19
cluster: %Cluster{},
20
20
generation: non_neg_integer,
21
+
22
+
locked?: boolean,
21
23
version: non_neg_integer,
22
24
known_committed_version: non_neg_integer,
25
+
23
26
buffer: %{non_neg_integer => {GenServer.from, %LogBatch{}}},
27
+
28
+
tags: %{integer => [integer]},
29
+
virtual_tags: %{integer => integer},
30
+
vtag_popped_versions: %{integer => non_neg_integer},
24
31
25
32
current_extent: %{i: non_neg_integer, extent: non_neg_integer, size: non_neg_integer},
26
33
}
···
35
42
:known_committed_version,
36
43
37
44
:buffer,
38
-
:tag_popped_versions,
45
+
46
+
:tags,
47
+
:virtual_tags,
48
+
:vtag_popped_versions,
39
49
40
50
:xks,
41
51
:current_extent,
···
66
76
end
67
77
68
78
@doc """
69
-
Pops logs for `tag` up to `up_to_version` (inclusive).
79
+
Pops logs for a virtual tag up to `up_to_version` (inclusive).
70
80
"""
71
-
def pop(server, tag, up_to_version) do
72
-
SimServer.cast(server, {:pop, tag, up_to_version})
81
+
def pop(server, virtual_tag, up_to_version) do
82
+
SimServer.cast(server, {:pop, virtual_tag, up_to_version})
73
83
end
74
84
75
85
@spec write_batch_send(term, %LogBatch{}) :: SimServer.request_id
···
174
184
[current_extent_pair] = Reader.scan(xks, @state_partition, load_read_version, extents_prefix(), extents_end(), true, 1)
175
185
current_extent = read_extent(current_extent_pair)
176
186
187
+
tags =
188
+
Reader.scan(xks, @meta_partition, load_read_version, storage_teams_prefix(), storage_teams_end())
189
+
|> Map.new(&unpack_storage_team_pair/1)
190
+
191
+
virtual_tags =
192
+
Enum.reduce(tags, %{}, fn {tag, virtuals}, acc ->
193
+
Enum.reduce(virtuals, acc, fn v, acc -> Map.put(acc, v, tag) end)
194
+
end)
195
+
177
196
state = %State{
178
197
id: id,
179
198
cluster: cluster,
···
184
203
known_committed_version: state_fields.known_committed_version,
185
204
186
205
buffer: %{},
187
-
tag_popped_versions: %{},
206
+
207
+
tags: tags,
208
+
virtual_tags: virtual_tags,
209
+
vtag_popped_versions: %{},
188
210
189
211
xks: xks,
190
212
···
202
224
end
203
225
204
226
def handle_call({:peek_all, start_version, end_version}, _from, state) do
205
-
# tag_popped_versions is guaranteed to contain all tags which are present in the queue
206
-
tags = state.tag_popped_versions |> Map.keys() |> Enum.sort()
227
+
# vtag_popped_versions is guaranteed to contain all tags which are present in the queue
228
+
tags = state.vtag_popped_versions |> Map.keys() |> Enum.sort()
207
229
208
230
batches =
209
231
tags
···
260
282
{:noreply, state}
261
283
end
262
284
263
-
def handle_cast({:pop, tag, up_to_version}, state)
264
-
when is_integer(tag) and is_integer(up_to_version) and up_to_version >= 0 do
265
-
# TODO: virtual tags
266
-
state = put_in(state.tag_popped_versions[tag], up_to_version)
267
-
{:noreply, state}
285
+
def handle_cast({:pop, virtual_tag, up_to_version}, state) when is_integer(up_to_version) and up_to_version >= 0 do
286
+
%{virtual_tags: virtual_tags, vtag_popped_versions: vtag_popped_versions} = state
287
+
288
+
case Map.fetch(virtual_tags, virtual_tag) do
289
+
{:ok, tag} ->
290
+
state = %{state | vtag_popped_versions: Map.put(vtag_popped_versions, virtual_tag, up_to_version)}
291
+
state = pop_tag(state, tag)
292
+
293
+
{:noreply, state}
294
+
295
+
# If the virtual tag is not known, do nothing
296
+
# (this might happen if this tlog is lagging behind the storage server)
297
+
:error -> {:noreply, state}
298
+
end
268
299
end
269
300
270
301
def handle_cast(:lock, %State{} = state) do
···
453
484
},
454
485
])
455
486
456
-
maybe_write_meta(group, commit_version, xks)
487
+
state = maybe_write_meta(group, commit_version, state)
457
488
458
489
state = put_in(state.current_extent.size, size + group_size)
459
490
do_append_groups(groups_rest, commit_version, state)
460
491
end
461
492
end
462
493
463
-
defp maybe_write_meta({meta_tag(), mutations}, commit_version, %XKS{} = xks) do
494
+
defp maybe_write_meta({meta_tag(), mutations}, commit_version, %State{} = state) do
495
+
%State{xks: xks} = state
496
+
464
497
unwrapped = Enum.map(mutations, fn {_i, mut} -> mut end)
465
498
XKS.apply_batch(xks, @meta_partition, commit_version, unwrapped)
499
+
500
+
state
466
501
end
467
-
defp maybe_write_meta(_group, _commit_version, _xks), do: :noop
502
+
503
+
defp maybe_write_meta(_group, _commit_version, state), do: state
468
504
469
505
defp group_mutations(tagged_mutations) do
470
506
Enum.reduce(tagged_mutations, %{}, fn {tags, mut}, acc ->
···
474
510
end)
475
511
|> Enum.sort_by(&elem(&1, 0))
476
512
|> Enum.map(fn {k, v} -> {k, Enum.reverse(v)} end)
513
+
end
514
+
515
+
defp pop_tag(%State{} = state, tag) do
516
+
%State{
517
+
version: version,
518
+
tags: tags,
519
+
vtag_popped_versions: vtag_popped_versions,
520
+
xks: xks,
521
+
} = state
522
+
523
+
min_version =
524
+
Map.fetch!(tags, tag)
525
+
|> Enum.map(&Map.get(vtag_popped_versions, &1, 0))
526
+
|> Enum.min()
527
+
528
+
start_key = Keyset.pack([tag, 0])
529
+
# Note: min_version is inclusive because TLog.pop's up_to_version is inclusive
530
+
end_key = Keyset.pack([tag, min_version + 1])
531
+
532
+
mutations =
533
+
Reader.scan(xks, @queue_partition, version, start_key, end_key)
534
+
|> Enum.map(fn {k, _v} -> {:clear, k} end)
535
+
536
+
# Clearing writes at their own version could affect parallel readers,
537
+
# but they should never attempt to read these versions once popped
538
+
XKS.apply_batch(xks, @queue_partition, version, mutations)
539
+
540
+
state
477
541
end
478
542
479
543
defp rotate_extent(%State{} = state) do
+7
lib/utils.ex
+7
lib/utils.ex
···
115
115
}
116
116
end
117
117
118
+
@spec unpack_storage_team_pair({binary, binary}) :: {non_neg_integer, [non_neg_integer]}
119
+
def unpack_storage_team_pair({storage_teams_prefix() <> key, value}) do
120
+
[storage_team_id] = Keyset.unpack(key)
121
+
storage_server_ids = Keyset.unpack(value)
122
+
{storage_team_id, storage_server_ids}
123
+
end
124
+
118
125
@spec pack_key_storage_pair(binary, non_neg_integer, non_neg_integer | nil) :: {binary, binary}
119
126
def pack_key_storage_pair(shard_start_key, from_storage_team_id, to_storage_team_id)
120
127
when is_binary(shard_start_key) and is_integer(from_storage_team_id) and (is_integer(to_storage_team_id) or is_nil(to_storage_team_id)) do