this repo has no description
Allocate extents and fix FreeList bug
+55
-2
+54
-1
lib/servers/tlog.ex
+54
-1
lib/servers/tlog.ex
···
9
9
alias Hobbes.Encoding.Keyset
10
10
11
11
alias Hobbes.XKS
12
-
alias Hobbes.XKS.Reader
12
+
alias Hobbes.XKS.{FreeList, Reader}
13
13
14
14
alias Hobbes.Utils
15
15
import Hobbes.Utils
···
23
23
known_committed_version: non_neg_integer,
24
24
buffer: %{non_neg_integer => {GenServer.from, %LogBatch{}}},
25
25
tagged_queue: TaggedQueue.t,
26
+
27
+
current_extent: %{i: non_neg_integer, extent: non_neg_integer, size: non_neg_integer},
26
28
}
27
29
28
30
@enforce_keys [
···
39
41
40
42
:xks,
41
43
:tagged_queue,
44
+
45
+
:current_extent,
42
46
]
43
47
defstruct @enforce_keys
44
48
end
···
47
51
48
52
@state_partition 0
49
53
@meta_partition 1
54
+
55
+
defmacrop extents_prefix, do: "extents/"
56
+
defmacrop extents_end, do: "extents0"
50
57
51
58
def start_link(arg), do: SimServer.start_link(__MODULE__, arg)
52
59
···
136
143
# Write initial meta pairs
137
144
load_meta_pairs(xks, meta_pairs)
138
145
146
+
# Write the first extent
147
+
write_new_extent(xks, 1, %{
148
+
i: 0,
149
+
extent: FreeList.reserve_extent(xks.free_list),
150
+
size: 0,
151
+
})
152
+
139
153
# Commit metadata
140
154
compact_and_commit(xks)
141
155
···
158
172
# Load persistent state from storage
159
173
state_fields = get_tlog_state(xks, load_read_version)
160
174
175
+
# Read the most recent extent's metadata from storage
176
+
[current_extent_pair] = Reader.scan(xks, @state_partition, load_read_version, extents_prefix(), extents_end(), true, 1)
177
+
current_extent = read_extent(current_extent_pair)
178
+
161
179
state = %State{
162
180
id: id,
163
181
cluster: cluster,
···
172
190
173
191
xks: xks,
174
192
tagged_queue: TaggedQueue.new(),
193
+
194
+
current_extent: current_extent,
175
195
}
176
196
# Sanity check persistent state
177
197
assert is_boolean(state.locked?)
···
423
443
424
444
defp append_batch(%State{} = _state, %LogBatch{} = _batch) do
425
445
# TODO
446
+
end
447
+
448
+
defp rotate_extent(%State{} = state) do
449
+
%State{
450
+
version: version,
451
+
xks: xks,
452
+
current_extent: current_extent,
453
+
} = state
454
+
455
+
new_extent = %{
456
+
i: current_extent.i + 1,
457
+
extent: FreeList.reserve_extent(xks),
458
+
size: 0,
459
+
}
460
+
461
+
write_new_extent(xks, version, new_extent)
462
+
%{state | current_extent: new_extent}
463
+
end
464
+
465
+
defp write_new_extent(%XKS{} = xks, version, %{i: i, extent: extent, size: size}) do
466
+
XKS.apply_batch(xks, @state_partition, version, [
467
+
{
468
+
:write,
469
+
extents_prefix() <> Keyset.pack([i]),
470
+
Keyset.pack([extent, size]),
471
+
},
472
+
])
473
+
end
474
+
475
+
defp read_extent({extents_prefix() <> key, value} = _pair) do
476
+
[i] = Keyset.unpack(key)
477
+
[extent, size] = Keyset.unpack(value)
478
+
%{i: i, extent: extent, size: size}
426
479
end
427
480
end