this repo has no description
Refactor commit pipeline to prepare for clears
+199
-109
+1
ROADMAP.md
+1
ROADMAP.md
···
67
67
- [X] Handle deleted ranges in HybridKV.flush
68
68
- [X] Handle deleted ranges in HybridKV.scan
69
69
- [ ] Implement range clears in the database
70
+
- [ ] Use ets to store Resolver mutation log and pop unneeded entries
70
71
- [ ] Recovery
71
72
- [ ] Monitor all transaction processes in Manager to detect failures
72
73
- [ ] Lock TLogs and choose recovery version
+7
-5
lib/meta_store.ex
+7
-5
lib/meta_store.ex
···
35
35
alias Hobbes.{MemKV, MetaStore}
36
36
alias Hobbes.Structs.{Cluster, RangeResult}
37
37
38
+
alias Hobbes.Utils
38
39
import Hobbes.Utils
39
40
40
41
@enforce_keys [:kv]
···
187
188
Returns a list of diffs of the form `{key, {previous_value, new_value}}`
188
189
for use in implementing side-effects.
189
190
"""
190
-
@spec apply_meta_mutations(%MetaStore{}, non_neg_integer, [{binary, binary}]) :: [{binary, {binary, binary}}]
191
+
@spec apply_meta_mutations(%MetaStore{}, non_neg_integer, [Utils.mutation]) :: [{binary, {binary, binary}}]
191
192
def apply_meta_mutations(%MetaStore{} = store, commit_version, meta_mutations) when is_integer(commit_version) and is_list(meta_mutations) do
192
193
# TODO: should the return values be deduplicated?
193
194
# A key could be written multiple times in a batch...
194
-
Enum.map(meta_mutations, fn {@meta_prefix <> _ = k, v} when is_binary(k) and is_binary(v) ->
195
-
existing = get(store.kv, commit_version, k)
196
-
put(store.kv, commit_version, k, v)
197
-
{k, {existing, v}}
195
+
Enum.map(meta_mutations, fn
196
+
{:write, @meta_prefix <> _ = k, v} when is_binary(k) and is_binary(v) ->
197
+
existing = get(store.kv, commit_version, k)
198
+
put(store.kv, commit_version, k, v)
199
+
{k, {existing, v}}
198
200
end)
199
201
end
200
202
+41
-39
lib/servers/commit_buffer.ex
+41
-39
lib/servers/commit_buffer.ex
···
114
114
end
115
115
116
116
def handle_call({:commit, %CommitTxn{} = txn}, from, state) do
117
-
# TODO: error handling?
118
-
validate_transaction!(txn)
119
-
120
117
txn = %CommitTxn{txn | from: from}
121
118
state = %State{state | buffer: [txn | state.buffer], buffer_size: state.buffer_size + 1}
122
119
state = maybe_commit_batch(state)
···
136
133
# in order to keep meta mutations flowing from resolvers
137
134
@spec commit_batch(%State{}) :: %State{}
138
135
defp commit_batch(%State{} = state) do
139
-
transactions =
140
-
state.buffer
141
-
|> Enum.reverse()
142
-
|> Enum.with_index()
143
-
|> Enum.map(fn {%CommitTxn{} = txn, i} ->
144
-
%CommitTxn{txn |
145
-
id: i,
146
-
meta: txn.mutations |> Enum.any?(fn {k, _v} -> MetaStore.is_meta?(k) end),
147
-
}
148
-
end)
149
-
150
136
[%Server{pid: seq_pid}] = get_servers(state.cluster, Hobbes.Servers.Sequencer)
151
137
{commit_version, prev_commit_version} = Sequencer.get_commit_version(seq_pid)
152
138
139
+
transactions_reversed = state.buffer
140
+
141
+
# Buffer is reversed, so these are in the correct order
142
+
resolver_txns =
143
+
Enum.reduce(transactions_reversed, [], fn %CommitTxn{} = txn, acc ->
144
+
[{
145
+
txn.read_version,
146
+
txn.read_conflicts,
147
+
txn.write_conflicts,
148
+
extract_meta(txn.mutations),
149
+
} | acc]
150
+
end)
151
+
153
152
batch = %CommitBatch{
154
153
commit_buffer_id: state.id,
155
154
commit_version: commit_version,
156
155
prev_commit_version: prev_commit_version,
157
-
transactions: transactions,
156
+
transactions: resolver_txns,
158
157
}
159
158
160
159
[%Server{pid: resolver_pid}] = get_servers(state.cluster, Hobbes.Servers.Resolver)
161
-
{allowed_txn_ids, meta_log} = Resolver.resolve_batch(resolver_pid, batch)
160
+
{txn_results_reversed, meta_log} = Resolver.resolve_batch(resolver_pid, batch)
162
161
162
+
# Apply meta mutations received, including our own from this batch
163
163
Enum.each(meta_log, fn {commit_version, mutations} ->
164
164
MetaStore.apply_meta_mutations(state.meta_store, commit_version, mutations)
165
165
end)
166
166
167
167
{allowed_transactions, rejected_transactions} =
168
-
batch.transactions
169
-
|> Enum.split_with(fn %CommitTxn{} = txn ->
170
-
txn.id in allowed_txn_ids
168
+
# Both are reversed, so they come out of this the right way around
169
+
Enum.zip(transactions_reversed, txn_results_reversed)
170
+
|> Enum.reduce({[], []}, fn
171
+
{txn, true}, {a, r} -> {[txn | a], r}
172
+
{txn, false}, {a, r} -> {a, [txn | r]}
171
173
end)
172
174
173
175
# If the database is locked, filter out all non-meta transactions
174
176
# TODO: use a lock-aware flag instead like FDB
175
177
{allowed_transactions, locked_transactions} =
176
178
case MetaStore.locked?(state.meta_store, commit_version) do
177
-
true -> Enum.split_with(allowed_transactions, fn %CommitTxn{} = txn -> txn.meta end)
179
+
true -> Enum.split_with(allowed_transactions, fn txn -> has_meta?(txn.mutations) end)
178
180
false -> {allowed_transactions, []}
179
181
end
180
182
···
184
186
allowed_transactions
185
187
|> Enum.map(fn %CommitTxn{} = txn -> txn.mutations end)
186
188
|> Enum.concat()
187
-
|> Enum.map(fn {k, _v} = mut ->
188
-
tags = MetaStore.get_key_server_mutation_tags(meta_store, commit_version, k)
189
-
{tags, mut}
189
+
|> Enum.map(fn
190
+
{:write, k, _v} = mut ->
191
+
tags = MetaStore.get_key_server_mutation_tags(meta_store, commit_version, k)
192
+
{tags, mut}
193
+
# TODO: clears
194
+
# TODO: range clears (will have to be split)
190
195
end)
191
196
192
197
# TODO: get latest generation only
···
201
206
# Note: prepending reverses the mutations, hence the name
202
207
# (they are reversed again when the batch is created below)
203
208
log_mutations_reversed =
204
-
Enum.reduce(tagged_mutations, log_mutations_reversed, fn {tags, {k, _v}} = tm, acc ->
209
+
Enum.reduce(tagged_mutations, log_mutations_reversed, fn {tags, _mut} = tm, acc ->
205
210
tlogs =
206
-
# Note: could also use (-1 in tags) to check if meta - which is faster?
207
-
case MetaStore.is_meta?(k) do
208
-
# Send meta mutations (already tagged with -1 above) to all tlogs
211
+
case -1 in tags do
212
+
# Send meta mutations (tagged with -1) to all tlogs
209
213
true -> all_tlogs
210
214
false -> tlogs_for_tags(num_tlogs, @tlog_replication_factor, tags)
211
215
end
···
255
259
%State{state | last_commit_version: commit_version, buffer: [], buffer_size: 0}
256
260
end
257
261
258
-
defp validate_transaction!(%CommitTxn{} = txn) do
259
-
unless is_integer(txn.read_version) and txn.read_version >= 0 do
260
-
raise "Invalid read_version #{inspect(txn.read_version)} found in transaction #{inspect(txn)}"
261
-
end
262
-
263
-
Enum.each(txn.reads, fn
264
-
k when is_binary(k) -> :noop
265
-
{s, e} when is_binary(s) and is_binary(e) -> :noop
266
-
other -> raise "Invalid read #{inspect(other)} found in transaction #{inspect(txn)}"
262
+
defp has_meta?(mutations) do
263
+
Enum.any?(mutations, fn
264
+
{:write, "\xFF" <> _, _v} -> true
265
+
{:write, _k, _v} -> false
267
266
end)
267
+
end
268
268
269
-
Enum.each(txn.mutations, fn
270
-
{k, v} when is_binary(k) and is_binary(v) -> :noop
271
-
other -> raise "Invalid mutation #{inspect(other)} found in transaction #{inspect(txn)}"
269
+
defp extract_meta(mutations) do
270
+
Enum.filter(mutations, fn
271
+
# TODO: clears, range clears
272
+
{:write, "\xFF" <> _, _v} -> true
273
+
{:write, _k, _v} -> false
272
274
end)
273
275
end
274
276
end
+53
-38
lib/servers/resolver.ex
+53
-38
lib/servers/resolver.ex
···
1
1
defmodule Hobbes.Servers.Resolver do
2
2
use Hobbes.Construct.SimServer
3
3
4
-
alias Hobbes.Structs.{Cluster, CommitBatch, CommitTxn}
4
+
alias Hobbes.Structs.{Cluster, CommitBatch}
5
5
alias Hobbes.VersionMap
6
+
7
+
alias Hobbes.Utils
6
8
7
9
defmodule State do
8
10
@enforce_keys [
···
23
25
24
26
def start_link, do: SimServer.start_link(__MODULE__, nil)
25
27
26
-
@spec resolve_batch(pid, %CommitBatch{}) :: {:allow, [non_neg_integer]}
28
+
@doc """
29
+
Resolves a batch of transactions.
30
+
31
+
Returns a tuple of `{results, meta_log}`.
32
+
33
+
Results is a **reversed** list of booleans where each entry is
34
+
true if the transaction was allowed and false if not.
35
+
36
+
Meta log is a list of batches of mutations applied by other
37
+
CommitBuffers since the last request from this CommitBuffer.
38
+
"""
39
+
@spec resolve_batch(pid, %CommitBatch{}) :: {[boolean], [{non_neg_integer, [Utils.mutation]}]}
27
40
def resolve_batch(server, %CommitBatch{} = batch) do
28
41
SimServer.call(server, {:resolve_batch, batch})
29
42
end
···
62
75
{{from, %CommitBatch{} = batch}, buffer} ->
63
76
state = %State{state | buffer: buffer}
64
77
65
-
do_resolve_batch(state, from, batch)
78
+
do_resolve_batch(batch, from, state)
66
79
|> maybe_resolve_next()
67
80
{nil, _buffer} ->
68
81
state
69
82
end
70
83
end
71
84
72
-
defp do_resolve_batch(%State{} = state, from, %CommitBatch{} = batch) do
73
-
{state, allowed_ids, allowed_meta} =
74
-
batch.transactions
75
-
|> Enum.reduce({state, [], []}, fn %CommitTxn{} = txn, {%State{} = state, allowed_ids, allowed_meta} ->
76
-
case allow_transaction?(state, txn.read_version, txn.reads) do
85
+
defp do_resolve_batch(%CommitBatch{} = batch, from, %State{} = state) do
86
+
vm = state.version_map
87
+
commit_version = batch.commit_version
88
+
89
+
{results, allowed_meta_mutations} =
90
+
Enum.reduce(batch.transactions, {[], []}, fn txn, acc ->
91
+
{results, allowed_meta_mutations} = acc
92
+
{read_version, read_conflicts, write_conflicts, meta_mutations} = txn
93
+
94
+
case allow_read_conflicts?(vm, read_version, read_conflicts) do
77
95
true ->
78
-
nil
96
+
update_write_conflicts(vm, commit_version, write_conflicts)
97
+
79
98
{
80
-
update_last_commits(state, batch.commit_version, txn.mutations),
81
-
[txn.id | allowed_ids],
82
-
case txn.meta do
83
-
true -> [txn | allowed_meta]
84
-
false -> allowed_meta
85
-
end
99
+
[true | results],
100
+
[meta_mutations | allowed_meta_mutations],
86
101
}
87
102
88
103
false ->
89
-
{state, allowed_ids, allowed_meta}
104
+
{
105
+
[false | results],
106
+
allowed_meta_mutations,
107
+
}
90
108
end
91
109
end)
92
110
93
-
batch_meta_mutations =
94
-
allowed_meta
95
-
|> Enum.map(fn %CommitTxn{} = txn -> txn.mutations end)
96
-
|> Enum.concat()
111
+
# TODO: we will need to keep track of the index of each allowed transaction once
112
+
# we have multiple Resolvers so that CommitBuffers can discern which transactions
113
+
# passed on all Resolvers
114
+
meta_mutations_log = [{commit_version, Enum.concat(allowed_meta_mutations)} | state.meta_mutations_log]
97
115
98
-
meta_mutations_log = [{batch.commit_version, batch_meta_mutations} | state.meta_mutations_log]
116
+
# TODO: store the log in ets, maybe reuse TaggedQueue?
99
117
buffer_last_version = Map.get(state.commit_buffer_last_versions, batch.commit_buffer_id, 0)
100
-
101
118
reply_meta_mutations =
102
-
meta_mutations_log
103
-
|> Enum.reduce_while([], fn {commit_ver, _mutations} = meta_log_entry, acc ->
104
-
case commit_ver >= buffer_last_version do
105
-
true -> {:cont, [meta_log_entry | acc]}
119
+
Enum.reduce_while(meta_mutations_log, [], fn {ver, _mutations} = entry, acc ->
120
+
case ver > buffer_last_version do
121
+
true -> {:cont, [entry | acc]}
106
122
false -> {:halt, acc}
107
123
end
108
124
end)
109
125
110
-
SimServer.reply(from, {allowed_ids, reply_meta_mutations})
126
+
# Note: we intentionally reply with results reversed because it's
127
+
# better to reverse them on the CommitBuffer as it's less of a bottleneck
128
+
SimServer.reply(from, {results, reply_meta_mutations})
111
129
112
130
%State{state |
113
-
version: batch.commit_version,
114
-
commit_buffer_last_versions: Map.put(state.commit_buffer_last_versions, batch.commit_buffer_id, batch.commit_version),
131
+
version: commit_version,
115
132
meta_mutations_log: meta_mutations_log,
133
+
commit_buffer_last_versions: Map.put(state.commit_buffer_last_versions, batch.commit_buffer_id, commit_version),
116
134
}
117
135
end
118
136
119
-
defp allow_transaction?(%State{version_map: vm}, read_version, reads)
120
-
when is_integer(read_version) and is_list(reads) do
121
-
Enum.all?(reads, fn read ->
122
-
not VersionMap.written_after?(vm, read_version, read)
137
+
defp allow_read_conflicts?(version_map, read_version, read_conflicts) do
138
+
Enum.all?(read_conflicts, fn key_or_range ->
139
+
not VersionMap.written_after?(version_map, read_version, key_or_range)
123
140
end)
124
141
end
125
142
126
-
defp update_last_commits(%State{} = state, commit_version, mutations) when is_integer(commit_version) and is_list(mutations) do
127
-
keys = Enum.map(mutations, fn {k, _v} -> k end)
128
-
VersionMap.add_writes(state.version_map, commit_version, keys)
129
-
130
-
state
143
+
defp update_write_conflicts(version_map, commit_version, write_conflicts) do
144
+
# TODO: handle ranges
145
+
VersionMap.add_writes(version_map, commit_version, write_conflicts)
131
146
end
132
147
end
+14
-9
lib/servers/storage.ex
+14
-9
lib/servers/storage.ex
···
6
6
alias Hobbes.Structs.{Cluster, Server, RangeResult, ShardStats}
7
7
alias Hobbes.Servers.{TLog, Storage}
8
8
9
+
alias Hobbes.Utils
9
10
import Hobbes.Utils
10
11
11
12
defmodule ShardImport do
···
293
294
%State{state | data_version: largest_version}
294
295
end
295
296
296
-
@spec apply_batch(%State{}, non_neg_integer, [{binary, binary}], [{binary, binary}]) :: %State{}
297
+
@spec apply_batch(%State{}, non_neg_integer, [Utils.mutation], [Utils.mutation]) :: %State{}
297
298
defp apply_batch(%State{} = state, commit_version, meta_mutations, data_mutations)
298
299
when is_integer(commit_version) and is_list(meta_mutations) and is_list(data_mutations) do
299
300
# Note: meta mutations are always applied first in a batch
···
304
305
end
305
306
306
307
defp apply_data_mutations(%State{kv: kv} = state, version, data_mutations) when is_list(data_mutations) do
307
-
Enum.each(data_mutations, fn {k, v} ->
308
-
HybridKV.put(kv, version, k, v)
308
+
Enum.each(data_mutations, fn
309
+
{:write, k, v} ->
310
+
HybridKV.put(kv, version, k, v)
309
311
310
-
case byte_sample_pair(k, v) do
311
-
{sk, sv} -> HybridKV.put(kv, version, sk, sv)
312
-
:not_in_sample -> :noop
313
-
end
312
+
case byte_sample_pair(k, v) do
313
+
{sk, sv} -> HybridKV.put(kv, version, sk, sv)
314
+
:not_in_sample -> :noop
315
+
end
316
+
# TODO: clear
317
+
# TODO: range clear
314
318
end)
315
319
316
320
state
···
350
354
end
351
355
end
352
356
353
-
@spec apply_meta_mutations(%State{}, non_neg_integer, [{binary, binary}]) :: %State{}
357
+
@spec apply_meta_mutations(%State{}, non_neg_integer, [Utils.mutation]) :: %State{}
354
358
defp apply_meta_mutations(%State{} = state, _commit_version, []), do: state
355
359
356
360
defp apply_meta_mutations(%State{} = state, commit_version, meta_mutations)
···
472
476
|> case do
473
477
{:ok, %RangeResult{pairs: pairs, more: more}} ->
474
478
# Write imported KV pairs at `read_version`
475
-
apply_data_mutations(state, read_version, pairs)
479
+
# TODO: use a separate function for this to avoid the map
480
+
apply_data_mutations(state, read_version, Enum.map(pairs, fn {k, v} -> {:write, k, v} end))
476
481
477
482
case more do
478
483
true ->
+3
-1
lib/servers/tlog.ex
+3
-1
lib/servers/tlog.ex
···
4
4
alias Hobbes.TaggedQueue
5
5
alias Hobbes.Structs.{Cluster, LogBatch}
6
6
7
+
alias Hobbes.Utils
8
+
7
9
@meta_tag -1
8
10
9
11
defmodule State do
···
32
34
Returns `{{start_version, end_version}, {meta_mutations, data_mutations}}`.
33
35
"""
34
36
@spec peek(pid, term, non_neg_integer | nil) ::
35
-
{{non_neg_integer, non_neg_integer}, {[{non_neg_integer, [{binary, binary}]}], [{non_neg_integer, [{binary, binary}]}]}}
37
+
{{non_neg_integer, non_neg_integer}, {[{non_neg_integer, [Utils.mutation]}], [{non_neg_integer, [Utils.mutation]}]}}
36
38
def peek(server, tag, start_version) do
37
39
SimServer.call(server, {:peek, tag, start_version})
38
40
end
+28
-1
lib/structs.ex
+28
-1
lib/structs.ex
···
4
4
multiple types of Server.
5
5
"""
6
6
7
+
alias Hobbes.Utils
8
+
7
9
defmodule Cluster do
8
10
@type t :: %__MODULE__{
9
11
config: ClusterConfig.t,
···
40
42
defstruct @enforce_keys
41
43
end
42
44
45
+
# TODO: rename to ResolveBatch
43
46
defmodule CommitBatch do
47
+
@type t :: %__MODULE__{
48
+
commit_buffer_id: integer,
49
+
commit_version: non_neg_integer,
50
+
prev_commit_version: non_neg_integer,
51
+
transactions: [
52
+
# {read_version, read_conflicts, write_conflicts, meta_mutations}
53
+
{
54
+
non_neg_integer,
55
+
[binary | {binary, binary}],
56
+
[binary | {binary, binary}],
57
+
[Utils.mutation],
58
+
}
59
+
]
60
+
}
44
61
@enforce_keys [:commit_buffer_id, :commit_version, :prev_commit_version, :transactions]
45
62
defstruct @enforce_keys
46
63
end
47
64
48
65
defmodule CommitTxn do
49
-
@enforce_keys [:read_version, :reads, :mutations]
66
+
@type t :: %__MODULE__{
67
+
id: term,
68
+
from: term,
69
+
meta: boolean,
70
+
71
+
read_version: non_neg_integer,
72
+
read_conflicts: [binary | {binary, binary}],
73
+
write_conflicts: [binary | {binary, binary}],
74
+
mutations: [Utils.mutation],
75
+
}
76
+
@enforce_keys [:read_version, :read_conflicts, :write_conflicts, :mutations]
50
77
defstruct [:id, :from, :meta] ++ @enforce_keys
51
78
end
52
79
+47
-13
lib/transaction.ex
+47
-13
lib/transaction.ex
···
4
4
5
5
alias Hobbes.Construct.SimServer
6
6
7
+
alias Hobbes.Utils
7
8
import Hobbes.Utils
8
9
9
10
defmodule TxnState do
10
11
@type t :: %__MODULE__{
11
12
cluster: %Cluster{},
12
13
read_version: integer,
13
-
reads: [binary | {binary, binary}],
14
-
writes: %{String.t => String.t},
14
+
15
+
read_conflicts: [binary | {binary, binary}],
16
+
write_conflicts: [binary | {binary, binary}],
17
+
mutations: [Utils.mutation],
18
+
15
19
commit_version: nil | non_neg_integer,
16
20
}
17
21
@enforce_keys [:cluster, :read_version]
18
-
defstruct @enforce_keys ++ [reads: [], writes: %{}, commit_version: nil]
22
+
defstruct @enforce_keys ++ [read_conflicts: [], write_conflicts: [], mutations: [], commit_version: nil]
19
23
end
20
24
21
25
@spec new(%Cluster{}) :: TxnState.t
···
34
38
35
39
case too_new_backoff(fn -> Storage.read(storage_pid, txn.read_version, key) end) do
36
40
{:ok, value} ->
37
-
{:ok, {value, %TxnState{txn | reads: [key | txn.reads]}}}
41
+
txn = add_read_conflict(txn, key)
42
+
{:ok, {value, txn}}
38
43
39
44
{:error, error} ->
40
45
{:error, error}
···
49
54
# If a read touches a large number of servers this could become a problem
50
55
case too_new_backoff(fn -> do_multi_read(txn, keys, shards) end) do
51
56
{:ok, results} ->
52
-
{:ok, {results, %TxnState{txn | reads: keys ++ txn.reads}}}
57
+
txn = Enum.reduce(keys, txn, fn k, txn -> add_read_conflict(txn, k) end)
58
+
{:ok, {results, txn}}
53
59
{:error, _} = error ->
54
60
error
55
61
end
···
120
126
end)
121
127
|> Enum.concat()
122
128
123
-
{:ok, {results, %TxnState{txn | reads: [{start_key, end_key} | txn.reads]}}}
129
+
txn = add_read_conflict(txn, {start_key, end_key})
130
+
{:ok, {results, txn}}
124
131
end
125
132
126
133
@spec write(TxnState.t, binary, binary) :: TxnState.t
127
134
def write(%TxnState{} = txn, key, value) when is_binary(key) and is_binary(value) do
128
-
%TxnState{txn | writes: Map.put(txn.writes, key, value)}
135
+
txn
136
+
|> add_mutation({:write, key, value})
137
+
|> add_write_confict(key)
129
138
end
130
139
131
140
@spec write(TxnState.t, [{binary, binary}]) :: TxnState.t
132
141
def write(%TxnState{} = txn, pairs) when is_list(pairs) do
133
-
Enum.reduce(pairs, txn, fn {k, v}, acc when is_binary(k) and is_binary(v) ->
134
-
write(acc, k, v)
135
-
end)
142
+
Enum.reduce(pairs, txn, fn {k, v}, txn -> write(txn, k, v) end)
143
+
end
144
+
145
+
@spec add_mutation(TxnState.t, Utils.mutation) :: TxnState.t
146
+
defp add_mutation(%TxnState{} = txn, mutation) do
147
+
%TxnState{txn | mutations: [mutation | txn.mutations]}
148
+
end
149
+
150
+
@spec add_read_conflict(TxnState.t, binary | {binary, binary}) :: TxnState.t
151
+
def add_read_conflict(txn, key_or_range)
152
+
153
+
def add_read_conflict(%TxnState{} = txn, key) when is_binary(key) do
154
+
%TxnState{txn | read_conflicts: [key | txn.read_conflicts]}
155
+
end
156
+
157
+
def add_read_conflict(%TxnState{} = txn, {start_key, end_key} = range) when is_binary(start_key) and is_binary(end_key) do
158
+
%TxnState{txn | read_conflicts: [range | txn.read_conflicts]}
159
+
end
160
+
161
+
@spec add_write_confict(TxnState.t, binary | {binary, binary}) :: TxnState.t
162
+
def add_write_confict(txn, key_or_range)
163
+
164
+
def add_write_confict(%TxnState{} = txn, key) when is_binary(key) do
165
+
%TxnState{txn | write_conflicts: [key | txn.write_conflicts]}
166
+
end
167
+
168
+
def add_write_confict(%TxnState{} = txn, {start_key, end_key} = range) when is_binary(start_key) and is_binary(end_key) do
169
+
%TxnState{txn | write_conflicts: [range | txn.write_conflicts]}
136
170
end
137
171
138
172
@spec commit(TxnState.t) :: {:ok, %TxnState{}} | {:error, term}
139
173
def commit(%TxnState{} = txn) do
140
174
commit_txn = %CommitTxn{
141
175
read_version: txn.read_version,
142
-
reads: txn.reads,
143
-
# TODO: store mutations as list in TxnState
144
-
mutations: txn.writes |> Map.to_list() |> Enum.sort_by(&elem(&1, 0))
176
+
read_conflicts: Enum.reverse(txn.read_conflicts),
177
+
write_conflicts: Enum.reverse(txn.write_conflicts),
178
+
mutations: Enum.reverse(txn.mutations),
145
179
}
146
180
147
181
commit_buf_pid = random_commit_buffer(txn.cluster)
+2
lib/utils.ex
+2
lib/utils.ex
···
6
6
alias Hobbes.Structs.{Cluster, Server}
7
7
alias Hobbes.Construct.SimServer
8
8
9
+
@type mutation :: {:write, binary, binary} | {:clear, binary} | {:clear_range, binary, binary}
10
+
9
11
@spec get_servers(%Cluster{}, module) :: [%Server{}]
10
12
def get_servers(%Cluster{} = cluster, type) when is_atom(type) do
11
13
cluster.servers
+3
-3
test/meta_store_test.exs
+3
-3
test/meta_store_test.exs
···
10
10
setup do
11
11
store = MetaStore.new()
12
12
MetaStore.apply_meta_mutations(store, 1, [
13
-
{meta("key_servers/"), "1,2,3"},
14
-
{meta("key_servers/hello"), "4,5,6"},
15
-
{meta("key_servers/world"), "7,8,9"},
13
+
{:write, meta("key_servers/"), "1,2,3"},
14
+
{:write, meta("key_servers/hello"), "4,5,6"},
15
+
{:write, meta("key_servers/world"), "7,8,9"},
16
16
])
17
17
18
18
%{store: store}