this repo has no description
Add latency to send instead of receive
+64
-35
+8
lib/construct/scheduler.ex
+8
lib/construct/scheduler.ex
···
235
235
end
236
236
237
237
def handle_cast({:queue_task, pid, delay, %Send{} = send}, %State{} = state) when is_pid(pid) and is_integer(delay) do
238
+
delay =
239
+
case send.dest == pid do
240
+
true -> delay
241
+
# Minimum 10ms of latency for all messages not sent to self()
242
+
# TODO: vary this based on whether processes are on the same node
243
+
false -> max(delay, 10)
244
+
end
245
+
238
246
case delay do
239
247
0 ->
240
248
log(state, {:send_now, state.clock, send.dest, send.message})
+1
-4
lib/construct/sim_server.ex
+1
-4
lib/construct/sim_server.ex
···
72
72
end
73
73
end
74
74
75
-
# We yield on receives because the simulation has to make clock progress somewhere
76
-
# (obviously we can't yield on send because send is asynchronous!)
77
-
Hobbes.Construct.Scheduler.yield(unquote(scheduler_pid), 10)
78
-
75
+
Hobbes.Construct.Scheduler.yield(unquote(scheduler_pid))
79
76
# We first check if the message queue already has a matching message
80
77
# If so, we don't want to yield_until_message because the Scheduler would
81
78
# have no way of knowing that a matching message is already in our queue
+25
lib/workloads.ex
+25
lib/workloads.ex
···
247
247
end
248
248
end)
249
249
end
250
+
251
+
def mean([]), do: 0
252
+
def mean(numbers), do: Enum.sum(numbers) / length(numbers)
253
+
254
+
def pretty_number(number) when is_integer(number) do
255
+
number
256
+
|> Integer.to_string()
257
+
|> commas()
258
+
end
259
+
260
+
def pretty_number(number) when is_float(number) do
261
+
string = :erlang.float_to_binary(number, decimals: 2)
262
+
[i, f] = String.split(string, ".")
263
+
commas(i) <> "." <> f
264
+
end
265
+
266
+
defp commas(number_string) when is_binary(number_string) do
267
+
number_string
268
+
|> String.to_charlist()
269
+
|> Enum.reverse()
270
+
|> Enum.chunk_every(3)
271
+
|> Enum.map(&Enum.reverse/1)
272
+
|> Enum.reverse()
273
+
|> Enum.join(",")
274
+
end
250
275
end
+27
-8
lib/workloads/cycle.ex
+27
-8
lib/workloads/cycle.ex
···
29
29
use Hobbes.Construct.SimServer
30
30
31
31
defmodule State do
32
-
@enforce_keys [:cluster, :count, :tick_ms, :stopped, :stats]
32
+
@enforce_keys [:cluster, :count, :tick_ms, :stopped, :stats, :commit_latencies]
33
33
defstruct @enforce_keys
34
34
end
35
35
···
46
46
tick_ms: tick_ms,
47
47
stopped: false,
48
48
stats: %{},
49
+
commit_latencies: [],
49
50
}}
50
51
end
51
52
52
53
def handle_call(:stop, _from, state) do
53
54
{
54
55
:reply,
55
-
state.stats,
56
+
Map.take(state, [:stats, :commit_latencies]),
56
57
%State{state | stopped: true},
57
58
}
58
59
end
···
70
71
defp work(%State{} = state) do
71
72
state = inc_stat(state, :swaps)
72
73
74
+
start_time = SimServer.current_time()
75
+
73
76
k1 = "key" <> pad(SimServer.deterministic_random(0..(state.count - 1)))
74
77
with {:ok, txn} <- Transaction.new(state.cluster),
75
78
{:ok, {v2, txn}} <- Transaction.read(txn, k1), k2 = "key" <> v2,
···
84
87
SimServer.sleep(SimServer.deterministic_random(100..300)),
85
88
{:ok, txn} <- Transaction.commit(txn)
86
89
do
90
+
duration = SimServer.current_time() - start_time
91
+
87
92
case Hobbes.Workloads.Cycle.check_cycle_at_version(state.cluster, txn.commit_version) do
88
93
{:ok, _pairs} -> :noop
89
94
{:error, _error} -> :noop
···
95
100
"""
96
101
end
97
102
103
+
state = update_in(state.commit_latencies, fn list -> [duration | list] end)
98
104
inc_stat(state, :success)
99
105
else
106
+
{:error, :read_version_too_old} -> inc_stat(state, :rv_too_old)
107
+
{:error, :read_version_too_new} -> inc_stat(state, :rv_too_new)
108
+
{:error, :database_locked} -> inc_stat(state, :db_lock)
109
+
{:error, :too_many_retries} -> inc_stat(state, :tm_retries)
110
+
{:error, :transaction_too_old} -> inc_stat(state, :txn_too_old)
111
+
{:error, :read_conflict} -> inc_stat(state, :read_cflt)
112
+
100
113
{:error, error} -> inc_stat(state, error)
101
114
end
102
115
end
···
131
144
132
145
SimServer.sleep(duration_ms)
133
146
134
-
client_stats =
147
+
client_results =
135
148
clients
136
149
|> Enum.map(&SimServer.send_request(&1, :stop))
137
150
|> Enum.map(&SimServer.receive_response(&1, 30_000))
138
151
|> Enum.map(fn {:reply, reply} -> reply end)
139
152
140
-
cols = [:swaps, :success, :txn_too_old, :read_conflict, :database_locked, :rv_too_old, :too_many_retries, :timeout]
153
+
client_stats = Enum.map(client_results, fn %{stats: stats, commit_latencies: latencies} ->
154
+
Map.put(stats, :avg_latency, mean(latencies) / 1_000_000)
155
+
end)
156
+
157
+
cols = [:swaps, :success, :txn_too_old, :read_cflt, :db_lock, :rv_too_old, :too_many_retries, :timeout, :avg_latency]
141
158
142
159
client_stats_table =
143
160
client_stats
···
146
163
rows =
147
164
cols
148
165
|> Enum.map(&Map.get(stats, &1, 0))
149
-
|> Enum.map(&to_string/1)
150
-
[to_string(i) | rows]
166
+
|> Enum.map(&pretty_number/1)
167
+
168
+
[pretty_number(i) | rows]
151
169
end)
152
170
|> table(
153
171
cols: ["i"] ++ Enum.map(cols, &to_string/1),
···
157
175
158
176
total_stats = Hobbes.Utils.sum_stats(client_stats)
159
177
178
+
total_cols = List.delete(cols, :avg_latency)
160
179
total_stats_table = table(
161
-
[Enum.map(cols, &Map.get(total_stats, &1, 0)) |> Enum.map(&to_string/1)],
162
-
cols: Enum.map(cols, &to_string/1)
180
+
[Enum.map(total_cols, &Map.get(total_stats, &1, 0)) |> Enum.map(&to_string/1)],
181
+
cols: Enum.map(total_cols, &to_string/1)
163
182
)
164
183
165
184
pairs = check_cycle(cluster, key_count)
-22
lib/workloads/read_write.ex
-22
lib/workloads/read_write.ex
···
291
291
defp pretty_stats(stats) when is_map(stats) do
292
292
Map.new(stats, fn {k, v} -> {k, pretty_number(v)} end)
293
293
end
294
-
295
-
defp pretty_number(number) when is_integer(number) do
296
-
number
297
-
|> Integer.to_string()
298
-
|> commas()
299
-
end
300
-
301
-
defp pretty_number(number) when is_float(number) do
302
-
string = :erlang.float_to_binary(number, decimals: 2)
303
-
[i, f] = String.split(string, ".")
304
-
commas(i) <> "." <> f
305
-
end
306
-
307
-
defp commas(number_string) when is_binary(number_string) do
308
-
number_string
309
-
|> String.to_charlist()
310
-
|> Enum.reverse()
311
-
|> Enum.chunk_every(3)
312
-
|> Enum.map(&Enum.reverse/1)
313
-
|> Enum.reverse()
314
-
|> Enum.join(",")
315
-
end
316
294
end
+2
lib/workloads/shard_move.ex
+2
lib/workloads/shard_move.ex