this repo has no description
Add multiple ShardMove workload clients
+41
-7
+2
-1
ROADMAP.md
+2
-1
ROADMAP.md
···
73
73
- [X] Optimization: avoid waking a process unless it has received messages
74
74
- [X] Use a single clock source in Scheduler (above optimization slowed per-process clock progression)
75
75
- [X] Optimization: only wake a process when it receives the message it's waiting for
76
+
- [ ] Add timeouts to SimServer calls
76
77
77
78
78
79
### Testing
···
84
85
- [X] Make cycle checking more aggressive by checking every commit
85
86
- [ ] Add event tracing/logging system for debugging purposes
86
87
- [X] Add ReadWrite workload to test read/write correctness
87
-
- [ ] Make ShardMove workload more aggressive (multiple moves)
88
+
- [X] Make ShardMove workload more aggressive (multiple moves)
+1
-1
lib/servers/storage.ex
+1
-1
lib/servers/storage.ex
···
430
430
# TODO: assert invariants
431
431
432
432
too_new_backoff(fn ->
433
-
Storage.read_range(from_pid, read_version, si.current_key, si.end_key, limit: 10)
433
+
Storage.read_range(from_pid, read_version, si.current_key, si.end_key, limit: 4)
434
434
end)
435
435
|> case do
436
436
{:ok, %RangeResult{pairs: pairs, more: more}} ->
+31
-5
lib/workloads/shard_move.ex
+31
-5
lib/workloads/shard_move.ex
···
87
87
88
88
{:error, :same_servers} ->
89
89
state
90
+
91
+
{:error, :already_moving} ->
92
+
state
90
93
end
91
94
end
92
95
···
104
107
]
105
108
106
109
def run(%{cluster: %Cluster{} = cluster}, opts) do
107
-
delay_ms = Keyword.get(opts, :delay_ms, 3_000)
110
+
delay_ms = Keyword.get(opts, :delay_ms, 1_000)
111
+
num_clients = Keyword.get(opts, :clients, 1)
108
112
duration_ms = Keyword.get(opts, :duration_ms, 10_000)
109
113
110
114
SimServer.sleep(delay_ms)
111
-
{:ok, client} = Client.start_link(cluster, [])
115
+
clients = Enum.map(1..num_clients, fn _i ->
116
+
{:ok, client} = Client.start_link(cluster, [])
117
+
client
118
+
end)
112
119
113
120
SimServer.sleep(duration_ms)
114
-
%{moves: moves} = SimServer.call(client, :stop)
121
+
122
+
client_stats =
123
+
clients
124
+
|> Enum.map(&SimServer.send_request(&1, :stop))
125
+
|> Enum.map(&SimServer.receive_response/1)
126
+
127
+
client_stats_message =
128
+
client_stats
129
+
|> Enum.with_index()
130
+
|> Enum.map(fn {%{moves: moves}, i} ->
131
+
"#{i}: " <> inspect(moves, pretty: true)
132
+
end)
133
+
|> Enum.join("\n")
134
+
135
+
total_moves =
136
+
client_stats
137
+
|> Enum.map(fn %{moves: moves} -> length(moves) end)
138
+
|> Enum.sum()
115
139
116
140
{:ok, "Moved shard"}
117
141
{
118
142
:ok,
119
143
"""
120
-
Shard moves:
144
+
Executed #{inspect(total_moves)} shard moves.
145
+
146
+
Shard moves (per client):
121
147
122
-
#{inspect(moves, pretty: true)}
148
+
#{client_stats_message}
123
149
""",
124
150
}
125
151
end
+7
test/hobbes_test.exs
+7
test/hobbes_test.exs
···
85
85
]},
86
86
{Workloads.ShardMove, [
87
87
delay_ms: 3_000,
88
+
clients: 10,
88
89
duration_ms: 20_000,
89
90
]},
90
91
], HobbesTest.SimOpts.sim_opts(name: test, cluster_opts: [
92
+
num_storage: 12,
93
+
91
94
initial_shards: [
92
95
"",
96
+
"key10",
93
97
"key20",
98
+
"key30",
94
99
"key40",
100
+
"key50",
95
101
"key60",
102
+
"key70",
96
103
],
97
104
]))
98
105
end