this repo has no description
Fix cycle recovery timeouts
+35
-25
+35
-25
lib/workloads/cycle.ex
+35
-25
lib/workloads/cycle.ex
···
21
21
22
22
alias Hobbes.Transaction
23
23
import Hobbes.Workloads
24
+
import Hobbes.Utils
24
25
25
26
@behaviour Hobbes.Workloads.Workload
26
27
···
183
184
cols: Enum.map(total_cols, &to_string/1)
184
185
)
185
186
186
-
pairs = check_cycle(cluster_name, key_count)
187
+
{:ok, pairs} =
188
+
retry_acc(nil, fn nil ->
189
+
try do
190
+
pairs = check_cycle(cluster_name, key_count)
191
+
{:halt, {:ok, pairs}}
192
+
rescue
193
+
e in [Hobbes.BeginError, Hobbes.CommitError, Hobbes.ReadError] ->
194
+
{:cont, nil, {:error, e.error}}
195
+
end
196
+
end, 10)
187
197
188
198
pretty_pairs =
189
199
pairs
···
215
225
end
216
226
217
227
defp check_cycle(cluster_name, count) do
218
-
{:ok, txn} = Transaction.new(cluster_name)
228
+
Hobbes.transaction(cluster_name, fn ->
229
+
all_keys = Enum.map(0..(count - 1), fn i -> "key" <> Client.pad(i) end)
230
+
key_values = Hobbes.read(all_keys)
219
231
220
-
all_keys = Enum.map(0..(count - 1), fn i -> "key" <> Client.pad(i) end)
221
-
{:ok, {key_values, txn}} = Transaction.read(txn, all_keys)
232
+
key0 = "key" <> Client.pad(0)
222
233
223
-
key0 = "key" <> Client.pad(0)
234
+
{_visited, pairs, prev} =
235
+
Enum.reduce(0..(count - 1), {MapSet.new(), [], key0}, fn _i, {visited, pairs, prev} ->
236
+
value = key_values[prev]
224
237
225
-
{_txn, _visited, pairs, prev} =
226
-
Enum.reduce(0..(count - 1), {txn, MapSet.new(), [], key0}, fn _i, {txn, visited, pairs, prev} ->
227
-
value = key_values[prev]
238
+
unless value, do: raise "Expected value for key #{inspect(prev)}, got: #{inspect(value)}"
239
+
if MapSet.member?(visited, prev), do: raise "Already visited #{inspect(prev)}!"
228
240
229
-
unless value, do: raise "Expected value for key #{inspect(prev)}, got: #{inspect(value)}"
230
-
if MapSet.member?(visited, prev), do: raise "Already visited #{inspect(prev)}!"
241
+
{MapSet.put(visited, prev), [{prev, value} | pairs], "key" <> value}
242
+
end)
231
243
232
-
{txn, MapSet.put(visited, prev), [{prev, value} | pairs], "key" <> value}
233
-
end)
244
+
pairs = Enum.reverse(pairs)
234
245
235
-
pairs = Enum.reverse(pairs)
246
+
# TODO: better error handling
247
+
case prev do
248
+
^key0 ->
249
+
pairs
250
+
_ ->
251
+
raise """
252
+
Cycle broken! Expected "key0" at end, got: #{inspect(prev)}
236
253
237
-
# TODO: better error handling
238
-
case prev do
239
-
^key0 ->
240
-
pairs
241
-
_ ->
242
-
raise """
243
-
Cycle broken! Expected "key0" at end, got: #{inspect(prev)}
244
-
245
-
Pairs:
246
-
#{inspect(pairs, pretty: true, limit: :infinity)}
247
-
"""
248
-
end
254
+
Pairs:
255
+
#{inspect(pairs, pretty: true, limit: :infinity)}
256
+
"""
257
+
end
258
+
end)
249
259
end
250
260
251
261
def check_cycle_at_version(cluster_name, version) when is_integer(version) do