+8

lib/transaction.ex

··· 103 103 case acc || get_shards.() do 104 104 {:ok, shards} = shard_result -> 105 105 case do_multi_read(txn, keys, shards) do 106 106 + # Successful read, return results immediately 106 107 {:ok, _map} = read_result -> {:halt, read_result} 108 108 + # Shard information is outdated, retry get_shards 107 109 {:error, :wrong_server} = error -> {:cont, get_shards.(), error} 110 110 + # These are retryable, retry with the same shards 108 111 {:error, :timeout} = error -> {:cont, shard_result, error} 109 112 {:error, :read_version_too_new} = error -> {:cont, shard_result, error} 113 113 + # These are not retryable, halt immediately 110 114 {:error, :read_version_too_old} = error -> {:halt, error} 111 115 {:error, :wrong_generation} = error -> {:halt, error} 112 116 end ··· 189 193 case acc || get_ranges.() do 190 194 {:ok, ranges} = acc -> 191 195 case do_read_split_range(txn, ranges) do 196 196 + # Successful read, return results immediately 192 197 {:ok, _pairs} = result -> {:halt, result} 198 198 + # Shard information is outdated, retry get_ranges 193 199 {:error, :wrong_server} = error -> {:cont, nil, error} 200 200 + # These are retryable, retry with the same shards 194 201 {:error, :timeout} = error -> {:cont, acc, error} 195 202 {:error, :read_version_too_new} = error -> {:cont, acc, error} 203 203 + # These are not retryable, halt immediately 196 204 {:error, :read_version_too_old} = error -> {:halt, error} 197 205 {:error, :wrong_generation} = error -> {:halt, error} 198 206 end

+9

lib/utils.ex

··· 376 376 |> Enum.sort() 377 377 end 378 378 379 379 + @doc """ 380 380 + Retries a function `fun` with an accumulator. 381 381 + 382 382 + The function should return: 383 383 + - `{:cont, acc, error}` - where `error` is the return value if we're out of attempts 384 384 + - `{:halt, result}` - where `result` should be returned immediately 385 385 + 386 386 + The `acc` will be passed along to each invocation as long as `{:cont, ...}` is returned. 387 387 + """ 379 388 @spec retry_acc(term, (term -> {:cont, term, term} | {:halt, term}), pos_integer) :: term 380 389 def retry_acc(acc, fun, count \\ 6) when is_function(fun, 1) and is_integer(count) and count > 0 do 381 390 do_retry_acc(acc, fun, count, 1)

+8 -6

lib/workloads.ex

··· 149 149 cluster_opts = Keyword.put(cluster_opts, :distributed, Sim.simulated?()) 150 150 {:ok, coordinator_pids} = Hobbes.Sandbox.start_cluster(cluster_opts) 151 151 152 152 - {:ok, %Cluster{} = cluster} = retry_acc(nil, fn _acc -> 153 153 - case Hobbes.get_cluster(coordinator_pids) do 154 154 - {:ok, _cluster} = result -> {:halt, result} 155 155 - {:error, _err} = error -> {:cont, nil, error} 156 156 - end 157 157 - end, 20) 152 152 + # Keep retrying get_cluster until coordinators are connected 153 153 + {:ok, %Cluster{} = cluster} = 154 154 + retry_acc(nil, fn nil -> 155 155 + case Hobbes.get_cluster(coordinator_pids) do 156 156 + {:ok, _cluster} = result -> {:halt, result} 157 157 + {:error, _err} = error -> {:cont, nil, error} 158 158 + end 159 159 + end, 20) 158 160 159 161 context = %{cluster: cluster} 160 162