+215

lib/workloads/model.ex

reviewed

··· 1 1 + defmodule Hobbes.Workloads.Model do 2 2 + alias Hobbes.Transaction 3 3 + alias Hobbes.Transaction.TxnState 4 4 + alias Hobbes.Structs.Cluster 5 5 + alias Hobbes.Construct.SimServer 6 6 + 7 7 + @behaviour Hobbes.Workloads.Workload 8 8 + 9 9 + defmodule HistoryTxn do 10 10 + @enforce_keys [ 11 11 + :status, 12 12 + 13 13 + :read_version, 14 14 + :commit_version, 15 15 + :batch_index, 16 16 + 17 17 + :read_results, 18 18 + :mutations, 19 19 + ] 20 20 + defstruct @enforce_keys 21 21 + 22 22 + def new(status, %TxnState{} = txn, read_results, mutations) 23 23 + when is_atom(status) and is_list(read_results) and is_list(mutations) do 24 24 + %HistoryTxn{ 25 25 + status: status, 26 26 + 27 27 + read_version: txn.read_version, 28 28 + commit_version: txn.commit_version, 29 29 + # TODO 30 30 + batch_index: 0, 31 31 + 32 32 + read_results: read_results, 33 33 + mutations: mutations, 34 34 + } 35 35 + end 36 36 + end 37 37 + 38 38 + defmodule Client do 39 39 + use Hobbes.Construct.SimServer 40 40 + 41 41 + import Hobbes.Utils 42 42 + 43 43 + defmodule State do 44 44 + @enforce_keys [ 45 45 + :cluster, 46 46 + :tick_ms, 47 47 + 48 48 + :history, 49 49 + ] 50 50 + defstruct @enforce_keys 51 51 + end 52 52 + 53 53 + def start_link(opts), do: SimServer.start_link(__MODULE__, opts) 54 54 + 55 55 + def init(%{cluster: %Cluster{} = cluster, tick_ms: tick_ms}) do 56 56 + state = %State{ 57 57 + cluster: cluster, 58 58 + tick_ms: tick_ms, 59 59 + history: [], 60 60 + } 61 61 + SimServer.send self(), :tick 62 62 + {:ok, state} 63 63 + end 64 64 + 65 65 + def handle_call(:stop, _from, %State{} = state) do 66 66 + {:reply, %{history: state.history}, state} 67 67 + end 68 68 + 69 69 + def handle_info(:tick, %State{} = state) do 70 70 + state = tick(state) 71 71 + SimServer.send_after self(), :tick, state.tick_ms 72 72 + {:noreply, state} 73 73 + end 74 74 + 75 75 + defp tick(%State{} = state) do 76 76 + reads = random_reads(state) 77 77 + mutations = random_mutations(state) 78 78 + 79 79 + with {:ok, %TxnState{} = txn} <- Transaction.new(state.cluster), 80 80 + {:ok, {%TxnState{} = txn, read_results}} <- do_reads(txn, reads) 81 81 + do 82 82 + txn = add_mutations(txn, mutations) 83 83 + 84 84 + case Transaction.commit(txn) do 85 85 + {:ok, %TxnState{} = txn} -> 86 86 + append_history(state, HistoryTxn.new(:committed, txn, read_results, mutations)) 87 87 + 88 88 + {:error, :read_conflict} -> 89 89 + append_history(state, HistoryTxn.new(:read_conflict, txn, read_results, mutations)) 90 90 + # TODO: more errors 91 91 + end 92 92 + else 93 93 + # TODO: new() errors, read() errors 94 94 + error -> raise "Error: #{inspect(error)}" 95 95 + end 96 96 + end 97 97 + 98 98 + defp append_history(%State{} = state, %HistoryTxn{} = ht) do 99 99 + %{state | history: [ht | state.history]} 100 100 + end 101 101 + 102 102 + @spec add_mutations(TxnState.t, list) :: TxnState.t 103 103 + defp add_mutations(%TxnState{} = txn, mutations) do 104 104 + Enum.reduce(mutations, txn, fn 105 105 + {:write, key, value}, txn -> Transaction.write(txn, key, value) 106 106 + end) 107 107 + end 108 108 + 109 109 + @spec do_reads(TxnState.t, [{binary, binary} | binary]) :: {:ok, {TxnState.t, list}} | {:error, {TxnState.t, list}} 110 110 + defp do_reads(%TxnState{} = txn, reads) when is_list(reads) do 111 111 + Enum.reduce_while(reads, {txn, []}, fn 112 112 + {:read_range, {sk, ek}} = read, {txn, acc} -> 113 113 + case Transaction.read_range(txn, sk, ek) do 114 114 + {:ok, {pairs, txn}} -> 115 115 + acc = [{read, pairs} | acc] 116 116 + {:cont, {txn, acc}} 117 117 + {:error, _err} -> 118 118 + {:halt, {:error, {txn, acc}}} 119 119 + end 120 120 + 121 121 + {:read, keys} = read, {txn, acc} -> 122 122 + case Transaction.read(txn, keys) do 123 123 + {:ok, {result, txn}} -> 124 124 + acc = [{read, result} | acc] 125 125 + {:cont, {txn, acc}} 126 126 + {:error, _err} -> 127 127 + {:halt, {:error, {txn, acc}}} 128 128 + end 129 129 + end) 130 130 + |> case do 131 131 + {:error, {_txn, _acc}} = error -> 132 132 + error 133 133 + {%TxnState{} = txn, results} when is_list(results) -> 134 134 + {:ok, {txn, results}} 135 135 + end 136 136 + end 137 137 + 138 138 + defp random_mutations(%State{} = state) do 139 139 + count = Enum.random(1..10) 140 140 + Enum.map(1..count, fn _i -> 141 141 + case Enum.random(1..4) do 142 142 + # TODO: range clears 143 143 + _ -> 144 144 + {:write, random_key(state), random_key(state)} 145 145 + end 146 146 + end) 147 147 + end 148 148 + 149 149 + defp random_reads(%State{} = state) do 150 150 + count = Enum.random([1, 1, 1, 2, 2, 2, 3, 3, 4, 10]) 151 151 + Enum.map(1..count, fn _i -> 152 152 + case Enum.random(1..4) do 153 153 + 1 -> 154 154 + {:read_range, random_range(state)} 155 155 + _ -> 156 156 + key_count = Enum.random([1, 1, 1, 2, 2, 2, 3, 3, 4, 10]) 157 157 + {:read, Enum.map(1..key_count, fn _i -> random_key(state) end)} 158 158 + end 159 159 + end) 160 160 + end 161 161 + 162 162 + defp random_range(%State{} = state) do 163 163 + k1 = random_key(state) 164 164 + k2 = random_key(state) 165 165 + cond do 166 166 + k1 < k2 -> {k1, k2} 167 167 + k1 > k2 -> {k2, k1} 168 168 + k1 == k2 -> {k1, next_key(k1)} 169 169 + end 170 170 + end 171 171 + 172 172 + defp random_key(%State{} = _state) do 173 173 + case Enum.random(1..10) do 174 174 + 1 -> "" 175 175 + 2 -> "\xFF" 176 176 + _ -> Enum.random(1..100) |> Integer.to_string() |> String.pad_leading(3, "0") 177 177 + end 178 178 + end 179 179 + end 180 180 + 181 181 + def run(%{cluster: %Cluster{} = cluster}, opts) do 182 182 + client_count = Keyword.get(opts, :clients, 10) 183 183 + duration_ms = Keyword.get(opts, :duration_ms, 10_000) 184 184 + client_tick_ms = Keyword.get(opts, :client_tick_ms, 100) 185 185 + 186 186 + clients = 187 187 + Enum.map(1..client_count, fn _i -> 188 188 + {:ok, pid} = Client.start_link(%{cluster: cluster, tick_ms: client_tick_ms}) 189 189 + pid 190 190 + end) 191 191 + 192 192 + SimServer.sleep(duration_ms) 193 193 + 194 194 + results = 195 195 + clients 196 196 + |> Enum.map(&SimServer.send_request(&1, :stop)) 197 197 + |> Enum.map(&SimServer.receive_response/1) 198 198 + |> Enum.map(fn {:reply, reply} -> reply end) 199 199 + 200 200 + history = 201 201 + results 202 202 + |> Enum.map(fn %{history: history} -> history end) 203 203 + |> Enum.concat() 204 204 + |> Enum.sort_by(fn %HistoryTxn{} = ht -> 205 205 + {ht.commit_version, ht.batch_index} 206 206 + end) 207 207 + 208 208 + { 209 209 + :ok, 210 210 + """ 211 211 + Checked #{length(history)} transactions from #{client_count} clients. 212 212 + """ 213 213 + } 214 214 + end 215 215 + end

+14

test/hobbes_test.exs

reviewed

··· 42 42 end 43 43 end 44 44 45 45 + defmodule ModelWorkloadTest do 46 46 + use ExUnit.Case, async: true 47 47 + @tag :model 48 48 + test "Model", %{test: test} do 49 49 + Workloads.run([ 50 50 + {Workloads.Model, [ 51 51 + clients: 10, 52 52 + duration_ms: 10_000, 53 53 + client_tick_ms: 100, 54 54 + ]}, 55 55 + ], HobbesTest.SimOpts.sim_opts(name: test, preset: :cycle)) 56 56 + end 57 57 + end 58 58 + 45 59 defmodule CycleWorkloadTest do 46 60 use ExUnit.Case, async: true 47 61 @tag :cycle