this repo has no description
Refactor Transaction and Cycle to handle errors and timeouts
+83
-80
+1
lib/construct/sim_server.ex
+1
lib/construct/sim_server.ex
+14
-3
lib/servers/commit_buffer.ex
+14
-3
lib/servers/commit_buffer.ex
···
68
68
]
69
69
70
70
"""
71
-
@spec get_shards_multi(pid, [binary | {binary, binary}]) :: {:ok, [[shard]]}
71
+
@spec get_shards_multi(pid, [binary | {binary, binary}]) :: {:ok, [[shard]]} | {:error, :timeout}
72
72
def get_shards_multi(server, keys_or_ranges) when is_list(keys_or_ranges) do
73
-
SimServer.call(server, {:get_shards, keys_or_ranges})
73
+
try do
74
+
SimServer.call(server, {:get_shards, keys_or_ranges})
75
+
catch
76
+
:exit, {:timeout, _} -> {:error, :timeout}
77
+
end
74
78
end
75
79
76
-
def commit(server, %CommitTxn{} = txn), do: SimServer.call(server, {:commit, txn})
80
+
@spec commit(pid, CommitTxn.t) :: {:ok, map} | {:error, :transaction_too_old | :read_conflict | :database_locked | :timeout}
81
+
def commit(server, %CommitTxn{} = txn) do
82
+
try do
83
+
SimServer.call(server, {:commit, txn})
84
+
catch
85
+
:exit, {:timeout, _} -> {:error, :timeout}
86
+
end
87
+
end
77
88
78
89
def init(%{id: id, cluster: %Cluster{} = cluster, meta_pairs: meta_pairs}) do
79
90
state = %State{
+38
-24
lib/transaction.ex
+38
-24
lib/transaction.ex
···
64
64
def read(%TxnState{} = txn, keys) when is_list(keys) do
65
65
get_shards = fn ->
66
66
buf = random_commit_buffer(txn.cluster)
67
-
{:ok, shards} = CommitBuffer.get_shards_multi(buf, keys)
68
-
shards
67
+
CommitBuffer.get_shards_multi(buf, keys)
69
68
end
70
69
71
70
# Note: this resends *all* requests if a single one fails due to too_old/too_new
72
71
# If a read touches a large number of servers this could become a problem
73
-
get_shards.()
74
-
|> backoff(fn shards ->
75
-
case do_multi_read(txn, keys, shards) do
76
-
{:ok, _map} = result -> {:halt, result}
77
-
{:error, :wrong_server} -> {:cont, get_shards.()}
78
-
{:error, :read_version_too_new} -> {:cont, shards}
79
-
{:error, :read_version_too_old} = err -> {:halt, err}
72
+
backoff(nil, fn acc ->
73
+
case acc || get_shards.() do
74
+
{:ok, shards} = shard_result ->
75
+
case do_multi_read(txn, keys, shards) do
76
+
{:ok, _map} = read_result -> {:halt, read_result}
77
+
{:error, :wrong_server} -> {:cont, get_shards.()}
78
+
{:error, :timeout} -> {:cont, shard_result}
79
+
{:error, :read_version_too_new} -> {:cont, shard_result}
80
+
{:error, :read_version_too_old} = err -> {:halt, err}
81
+
end
82
+
83
+
# Retry get_shards
84
+
{:error, :timeout} -> {:cont, nil}
80
85
end
81
86
end)
82
87
|> case do
···
141
146
when is_binary(start_key) and is_binary(end_key) and start_key >= "" and end_key <= "\xFF\xFF" do
142
147
get_ranges = fn ->
143
148
buf = random_commit_buffer(txn.cluster)
144
-
{:ok, [shards]} = CommitBuffer.get_shards_multi(buf, [{start_key, end_key}])
145
-
intersect_ranges({start_key, end_key}, shards)
149
+
case CommitBuffer.get_shards_multi(buf, [{start_key, end_key}]) do
150
+
{:ok, [shards]} -> {:ok, intersect_ranges({start_key, end_key}, shards)}
151
+
{:error, _err} = error -> error
152
+
end
146
153
end
147
154
148
-
get_ranges.()
149
-
|> backoff(fn ranges ->
150
-
case do_read_split_range(txn, ranges) do
151
-
{:ok, _pairs} = result -> {:halt, result}
152
-
{:error, :wrong_server} -> {:cont, get_ranges.()}
153
-
{:error, :read_version_too_new} -> {:cont, ranges}
154
-
{:error, :read_version_too_old} = err -> {:halt, err}
155
+
backoff(nil, fn acc ->
156
+
case acc || get_ranges.() do
157
+
{:ok, ranges} = acc ->
158
+
case do_read_split_range(txn, ranges) do
159
+
{:ok, _pairs} = result -> {:halt, result}
160
+
{:error, :wrong_server} -> {:cont, nil}
161
+
{:error, :timeout} -> {:cont, acc}
162
+
{:error, :read_version_too_new} -> {:cont, acc}
163
+
{:error, :read_version_too_old} = error -> {:halt, error}
164
+
end
165
+
166
+
# Retry get_ranges
167
+
{:error, :timeout} -> {:cont, nil}
155
168
end
156
169
end)
157
170
|> case do
···
239
252
mutations: Enum.reverse(txn.mutations),
240
253
}
241
254
242
-
commit_buf_pid = random_commit_buffer(txn.cluster)
243
-
result = CommitBuffer.commit(commit_buf_pid, commit_txn)
244
-
case result do
255
+
buf = random_commit_buffer(txn.cluster)
256
+
257
+
CommitBuffer.commit(buf, commit_txn)
258
+
|> case do
245
259
{:ok, %{commit_version: commit_version}} ->
246
-
{:ok, %TxnState{txn | commit_version: commit_version}}
247
-
{:error, error} ->
248
-
{:error, error}
260
+
{:ok, %{txn | commit_version: commit_version}}
261
+
262
+
{:error, _error} = error -> error
249
263
end
250
264
end
251
265
+30
-53
lib/workloads/cycle.ex
+30
-53
lib/workloads/cycle.ex
···
73
73
74
74
defp work(%State{} = state) do
75
75
state = inc_stat(state, :swaps)
76
-
{:ok, txn} = Transaction.new(state.cluster)
77
76
78
77
k1 = "key" <> pad(SimServer.deterministic_random(0..(state.count - 1)))
78
+
with {:ok, txn} <- Transaction.new(state.cluster),
79
+
{:ok, {v2, txn}} <- Transaction.read(txn, k1), k2 = "key" <> v2,
80
+
{:ok, {v3, txn}} <- Transaction.read(txn, k2), k3 = "key" <> v3,
81
+
{:ok, {v4, txn}} <- Transaction.read(txn, k3),
79
82
80
-
with {:ok, {v2, txn}} <- Transaction.read(txn, k1),
81
-
{:ok, {v3, txn}} <- Transaction.read(txn, "key" <> v2),
82
-
{:ok, {v4, txn}} <- Transaction.read(txn, "key" <> v3)
83
+
# Swaps (1 -> 2 -> 3 -> 4) to (1 -> 3 -> 2 -> 4)
84
+
# clear(k1) is meant to catch bugs in clear implementation or mutation ordering
85
+
txn = txn |> Transaction.clear(k1) |> Transaction.write([{k1, v3}, {k3, v2}, {k2, v4}]),
86
+
87
+
# Random delay between reads/commit for additional concurrency
88
+
SimServer.sleep(SimServer.deterministic_random(100..300)),
89
+
{:ok, txn} <- Transaction.commit(txn)
83
90
do
84
-
k2 = "key" <> v2
85
-
k3 = "key" <> v3
86
-
# Random delay between reads/writes for additional concurrency
87
-
SimServer.sleep SimServer.deterministic_random(100..300)
91
+
case Hobbes.Workloads.Cycle.check_cycle_at_version(state.cluster, txn.commit_version) do
92
+
{:ok, _pairs} -> :noop
93
+
{:error, :read_version_too_old} -> :noop
88
94
89
-
# Swaps
90
-
# 1 -> 2 -> 3 -> 4
91
-
# to
92
-
# 1 -> 3 -> 2 -> 4
93
-
commit_result =
94
-
txn
95
-
|> Transaction.clear(k1)
96
-
|> Transaction.write(k1, v3)
97
-
|> Transaction.write(k3, v2)
98
-
|> Transaction.write(k2, v4)
99
-
|> Transaction.commit()
95
+
{:error, error, pairs} ->
96
+
raise """
97
+
Cycle check error at version #{txn.commit_version}: #{error}
98
+
Pairs: #{inspect(pairs, pretty: true, limit: :infinity)}\
99
+
"""
100
+
end
100
101
101
-
case commit_result do
102
-
{:ok, txn} ->
103
-
#IO.inspect {txn.read_version, txn.commit_version, txn.writes}, label: "Committed"
104
-
105
-
case Hobbes.Workloads.Cycle.check_cycle_at_version(state.cluster, txn.commit_version) do
106
-
{:ok, _pairs} ->
107
-
:noop
108
-
{:error, :read_version_too_old} ->
109
-
# TODO: maybe raise if this happens too often?
110
-
# or track stats? it's currently *very* rare
111
-
:noop
112
-
{:error, error, pairs} ->
113
-
raise """
114
-
Cycle check error at version #{txn.commit_version}: #{error}
115
-
Pairs: #{inspect(pairs, pretty: true, limit: :infinity)}\
116
-
"""
117
-
end
118
-
inc_stat(state, :success)
119
-
120
-
{:error, :transaction_too_old} -> inc_stat(state, :txn_too_old)
121
-
{:error, :read_conflict} -> inc_stat(state, :read_conflict)
122
-
{:error, :database_locked} -> inc_stat(state, :database_locked)
123
-
end
102
+
inc_stat(state, :success)
124
103
else
125
-
{:error, :read_version_too_old} -> inc_stat(state, :rv_too_old)
126
-
{:error, :too_many_retries} -> inc_stat(state, :too_many_retries)
104
+
{:error, error} -> inc_stat(state, error)
127
105
end
128
106
end
129
107
···
162
140
|> Enum.map(&SimServer.send_request(&1, :stop))
163
141
|> Enum.map(&SimServer.receive_response(&1, 30_000))
164
142
165
-
cols = [:swaps, :success, :txn_too_old, :read_conflict, :database_locked, :rv_too_old, :too_many_retries]
143
+
cols = [:swaps, :success, :txn_too_old, :read_conflict, :database_locked, :rv_too_old, :too_many_retries, :timeout]
166
144
167
145
client_stats_table =
168
146
client_stats
···
252
230
end
253
231
254
232
def check_cycle_at_version(%Cluster{} = cluster, version) when is_integer(version) do
255
-
{:ok, txn} = Transaction.new(cluster)
256
-
txn = Map.put(txn, :read_version, version)
257
-
258
-
case Transaction.read_range(txn, "key", "key\xFF") do
259
-
{:ok, {pairs, _txn}} ->
260
-
check_pairs(pairs)
261
-
{:error, :read_version_too_old} ->
262
-
{:error, :read_version_too_old}
233
+
with {:ok, txn} <- Transaction.new(cluster),
234
+
txn = Map.put(txn, :read_version, version),
235
+
{:ok, {pairs, _txn}} <- Transaction.read_range(txn, "key", "key\xFF")
236
+
do
237
+
check_pairs(pairs)
238
+
else
239
+
{:error, _err} = error -> error
263
240
end
264
241
end
265
242