this repo has no description
Convert timeouts to errors in server interfaces
+53
-30
+6
-6
lib/construct/sim_server.ex
+6
-6
lib/construct/sim_server.ex
···
182
182
183
183
See `:gen_server.receive_response/2` and `SimServer.send_request/2`.
184
184
"""
185
-
@spec receive_response(request_id, timeout) :: term
185
+
@spec receive_response(request_id, timeout) :: {:reply, term} | :timeout
186
186
def receive_response(request_id, timeout \\ 5000) do
187
187
case get_scheduler_pid() do
188
188
nil ->
···
191
191
sim_receive_response(scheduler_pid, request_id, timeout)
192
192
end
193
193
|> case do
194
-
# TODO: get rid of this and follow the gen_server API
195
-
{:reply, reply} -> reply
196
-
{:error, _error} -> exit(:timeout)
197
-
:timeout -> exit(:timeout)
194
+
{:reply, _reply} = reply -> reply
195
+
# Coalesce errors into timeouts for now
196
+
{:error, _err} -> :timeout
197
+
:timeout -> :timeout
198
198
end
199
199
end
200
200
···
217
217
ref
218
218
end
219
219
220
-
defp sim_receive_response(scheduler_pid, request_id, timeout) do
220
+
defp sim_receive_response(scheduler_pid, request_id, timeout) when is_reference(request_id) do
221
221
# TODO: catch timeout exit
222
222
try do
223
223
yield_receive(scheduler_pid, timeout) do
+1
-1
lib/servers/commit_buffer.ex
+1
-1
lib/servers/commit_buffer.ex
···
71
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
73
try do
74
-
SimServer.call(server, {:get_shards, keys_or_ranges})
74
+
SimServer.call(server, {:get_shards, keys_or_ranges}, 1000)
75
75
catch
76
76
:exit, {:timeout, _} -> {:error, :timeout}
77
77
end
+2
lib/servers/distributor.ex
+2
lib/servers/distributor.ex
+6
-1
lib/servers/manager.ex
+6
-1
lib/servers/manager.ex
···
389
389
|> Enum.map(fn pid ->
390
390
TLog.peek_many_send(pid, tags, start_version, end_version)
391
391
end)
392
-
|> Enum.map(&TLog.peek_many_receive/1)
392
+
|> Enum.map(fn req_id ->
393
+
case TLog.peek_many_receive(req_id) do
394
+
{:ok, batches} -> batches
395
+
# TODO: crash on timeout?
396
+
end
397
+
end)
393
398
|> merge_batches()
394
399
|> then(fn batches ->
395
400
[{0, []} | batches]
+15
-5
lib/servers/storage.ex
+15
-5
lib/servers/storage.ex
···
101
101
SimServer.send_request(server, {:read, read_version, keys})
102
102
end
103
103
104
-
@spec read_receive(SimServer.request_id) :: {:ok, [binary | nil]} | {:error, :read_version_too_old | :read_version_too_new}
104
+
@spec read_receive(SimServer.request_id) :: {:ok, [binary | nil]} | {:error, :read_version_too_old | :read_version_too_new | :timeout}
105
105
def read_receive(request_id) do
106
-
SimServer.receive_response(request_id)
106
+
case SimServer.receive_response(request_id, 2000) do
107
+
{:reply, reply} -> reply
108
+
:timeout -> {:error, :timeout}
109
+
end
107
110
end
108
111
109
112
@spec read_range(pid, non_neg_integer, binary, binary, Keyword.t) :: {:ok, RangeResult.t} | {:error, :read_version_too_old}
110
113
def read_range(server, read_version, start_key, end_key, opts \\ []) do
111
-
{timeout, opts} = Keyword.pop(opts, :timeout, 5000)
114
+
{timeout, opts} = Keyword.pop(opts, :timeout, 2000)
112
115
113
116
limit = Keyword.get(opts, :limit, :infinity)
114
117
reverse = Keyword.get(opts, :reverse, false)
115
-
SimServer.call(server, {:read_range, read_version, start_key, end_key, limit, reverse}, timeout)
118
+
try do
119
+
SimServer.call(server, {:read_range, read_version, start_key, end_key, limit, reverse}, timeout)
120
+
catch
121
+
:exit, {:timeout, _} -> {:error, :timeout}
122
+
end
116
123
end
117
124
118
125
@spec get_shard_stats(pid, binary, binary) :: {:ok, ShardStats.t}
···
127
134
128
135
@spec check_import_complete_receive(SimServer.request_id) :: {:ok, boolean} | {:error, :not_found}
129
136
def check_import_complete_receive(request_id) do
130
-
SimServer.receive_response(request_id)
137
+
case SimServer.receive_response(request_id) do
138
+
{:reply, reply} -> reply
139
+
:timeout -> {:error, :timeout}
140
+
end
131
141
end
132
142
133
143
def init(%{id: id, cluster: %Cluster{} = cluster}) do
+11
-5
lib/servers/tlog.ex
+11
-5
lib/servers/tlog.ex
···
65
65
SimServer.send_request(server, {:write_batch, batch})
66
66
end
67
67
68
-
@spec write_batch_receive(SimServer.request_id) :: :ok | {:error, :tlog_locked}
68
+
@spec write_batch_receive(SimServer.request_id) :: :ok | {:error, :tlog_locked | :timeout}
69
69
def write_batch_receive(request_id) do
70
-
SimServer.receive_response(request_id)
70
+
case SimServer.receive_response(request_id) do
71
+
{:reply, reply} -> reply
72
+
:timeout -> {:error, :timeout}
73
+
end
71
74
end
72
75
73
76
@doc """
···
98
101
SimServer.send_request(server, {:peek_many, tags, start_version, end_version})
99
102
end
100
103
101
-
@spec peek_many_receive(SimServer.request_id) :: [{non_neg_integer, [Utils.numbered_mutation]}]
104
+
@spec peek_many_receive(SimServer.request_id) :: {:ok, [{non_neg_integer, [Utils.numbered_mutation]}]} | {:error, :timeout}
102
105
def peek_many_receive(request_id) do
103
-
SimServer.receive_response(request_id)
106
+
case SimServer.receive_response(request_id) do
107
+
{:reply, reply} -> reply
108
+
:timeout -> {:error, :timeout}
109
+
end
104
110
end
105
111
106
112
def lock(server) do
···
134
140
TaggedQueue.peek(state.tagged_queue, tag, start_version, end_version)
135
141
end)
136
142
|> merge_batches()
137
-
{:reply, batches, state}
143
+
{:reply, {:ok, batches}, state}
138
144
end
139
145
140
146
def handle_call({:write_batch, %LogBatch{} = batch}, from, %State{} = state) do
+3
-3
lib/utils.ex
+3
-3
lib/utils.ex
···
302
302
end
303
303
304
304
@spec backoff(term, ({:cont | :halt, term} -> term), non_neg_integer, non_neg_integer) :: term
305
-
def backoff(acc \\ nil, fun, count \\ 10, attempt \\ 0)
305
+
def backoff(acc \\ nil, fun, count \\ 6, attempt \\ 0)
306
306
307
307
def backoff(_acc, _fun, count, attempt) when attempt == count do
308
308
{:error, :too_many_retries}
···
311
311
def backoff(acc, fun, count, attempt) when is_function(fun, 1) and is_integer(count) and is_integer(attempt) do
312
312
case fun.(acc) do
313
313
{:cont, acc} ->
314
-
delay = floor(Float.pow(1.6, attempt + 1))
314
+
delay = Integer.pow(2, attempt + 1)
315
315
SimServer.sleep(delay)
316
-
backoff(acc, fun, count, attempt)
316
+
backoff(acc, fun, count, attempt + 1)
317
317
{:halt, acc} ->
318
318
acc
319
319
end
+7
-9
lib/workloads/cycle.ex
+7
-9
lib/workloads/cycle.ex
···
60
60
def handle_info(:tick, %State{stopped: true} = state), do: {:noreply, state}
61
61
62
62
def handle_info(:tick, %State{} = state) do
63
-
state =
64
-
try do
65
-
work(state)
66
-
catch
67
-
:exit, {{:timeout, _mfa}, _stacktrace} ->
68
-
inc_stat(state, :timeout)
69
-
end
63
+
state = work(state)
70
64
SimServer.send_after(self(), :tick, state.tick_ms)
71
65
{:noreply, state}
72
66
end
67
+
68
+
def handle_info(_message, %State{} = state), do: {:noreply, state}
73
69
74
70
defp work(%State{} = state) do
75
71
state = inc_stat(state, :swaps)
···
90
86
do
91
87
case Hobbes.Workloads.Cycle.check_cycle_at_version(state.cluster, txn.commit_version) do
92
88
{:ok, _pairs} -> :noop
93
-
{:error, :read_version_too_old} -> :noop
89
+
{:error, _error} -> :noop
94
90
95
91
{:error, error, pairs} ->
96
92
raise """
···
139
135
clients
140
136
|> Enum.map(&SimServer.send_request(&1, :stop))
141
137
|> Enum.map(&SimServer.receive_response(&1, 30_000))
138
+
|> Enum.map(fn {:reply, reply} -> reply end)
142
139
143
140
cols = [:swaps, :success, :txn_too_old, :read_conflict, :database_locked, :rv_too_old, :too_many_retries, :timeout]
144
141
···
154
151
end)
155
152
|> table(
156
153
cols: ["i"] ++ Enum.map(cols, &to_string/1),
157
-
align: List.duplicate(:right, length(cols) + 1)
154
+
align: List.duplicate(:right, length(cols) + 1),
155
+
limit: 20
158
156
)
159
157
160
158
total_stats = Hobbes.Utils.sum_stats(client_stats)
+1
lib/workloads/read_write.ex
+1
lib/workloads/read_write.ex