-772

lib/construct/scheduler.ex

reviewed

··· 1 1 - defmodule Hobbes.Construct.Scheduler do 2 2 - use GenServer 3 3 - 4 4 - alias Hobbes.Construct.SimLog 5 5 - alias Hobbes.Construct.Scheduler.{ProcStore, ProcQueue, ProcRegistry, FileStore} 6 6 - alias Hobbes.Construct.Scheduler.ProcStore.ProcState 7 7 - 8 8 - defmodule Spawn do 9 9 - @enforce_keys [:module, :function, :args, :link, :parent_resume] 10 10 - defstruct @enforce_keys 11 11 - end 12 12 - 13 13 - defmodule Resume do 14 14 - @enforce_keys [:ref, :pid, :value] 15 15 - defstruct @enforce_keys 16 16 - end 17 17 - 18 18 - defmodule Timeout do 19 19 - @enforce_keys [:pid, :ref] 20 20 - defstruct @enforce_keys 21 21 - end 22 22 - 23 23 - defmodule Send do 24 24 - @enforce_keys [:dest, :message] 25 25 - defstruct @enforce_keys 26 26 - end 27 27 - 28 28 - defmodule Exit do 29 29 - @enforce_keys [:from_pid, :to_pid, :reason] 30 30 - defstruct @enforce_keys 31 31 - end 32 32 - 33 33 - defmodule StartNode do 34 34 - @enforce_keys [:name] 35 35 - defstruct @enforce_keys 36 36 - end 37 37 - 38 38 - defmodule Node do 39 39 - @type t :: %__MODULE__{ 40 40 - pid: pid | nil, 41 41 - status: :running | :stopped, 42 42 - name: atom, 43 43 - app_module: module, 44 44 - args: term, 45 45 - } 46 46 - @enforce_keys [:pid, :status, :name, :app_module, :args] 47 47 - defstruct @enforce_keys 48 48 - end 49 49 - 50 50 - defmodule State do 51 51 - @type t :: %__MODULE__{ 52 52 - clock: non_neg_integer, 53 53 - current: pid | nil, 54 54 - 55 55 - nodes: [{atom, Node.t}], 56 56 - proc_store: ProcStore.t, 57 57 - proc_queue: ProcQueue.t, 58 58 - proc_registry: ProcRegistry.t, 59 59 - file_stores: %{atom => FileStore.t}, 60 60 - 61 61 - log_server_pid: pid, 62 62 - 63 63 - resumes_without_send: non_neg_integer, 64 64 - } 65 65 - 66 66 - defstruct [ 67 67 - clock: 0, 68 68 - 69 69 - nodes: [], 70 70 - current: nil, 71 71 - proc_store: nil, 72 72 - proc_queue: nil, 73 73 - proc_registry: nil, 74 74 - file_stores: %{}, 75 75 - 76 76 - log_server_pid: nil, 77 77 - resumes_without_send: 0, 78 78 - ] 79 79 - end 80 80 - 81 81 - @spec start_link(term) :: GenServer.on_start 82 82 - def start_link(seed) when is_integer(seed) do 83 83 - :rand.seed(:exsss, seed) 84 84 - scheduler_seed = random_seed() 85 85 - 86 86 - {:ok, scheduler_pid} = GenServer.start_link(__MODULE__, %{initial_pid: self(), seed: scheduler_seed}) 87 87 - Hobbes.Construct.SimServer.set_scheduler_pid(scheduler_pid) 88 88 - 89 89 - {:ok, scheduler_pid} 90 90 - end 91 91 - 92 92 - @spec yield(term) :: :ok 93 93 - def yield(scheduler, delay \\ 0) do 94 94 - resume = %Resume{ref: make_ref(), pid: self(), value: nil} 95 95 - GenServer.cast(scheduler, {:queue_task, self(), delay, resume}) 96 96 - GenServer.cast(scheduler, {:yield, self()}) 97 97 - 98 98 - nil = await_resume(resume) 99 99 - :ok 100 100 - end 101 101 - 102 102 - @spec yield_until_message(pid, function, non_neg_integer | :infinity) :: :ok 103 103 - def yield_until_message(scheduler, check_fun, timeout_ms) 104 104 - when is_function(check_fun) and ((is_integer(timeout_ms) and timeout_ms >= 0) or timeout_ms == :infinity) do 105 105 - resume = %Resume{ref: make_ref(), pid: self(), value: nil} 106 106 - 107 107 - GenServer.cast(scheduler, {:await_message, self(), check_fun, timeout_ms, resume}) 108 108 - GenServer.cast(scheduler, {:yield, self()}) 109 109 - 110 110 - nil = await_resume(resume) 111 111 - :ok 112 112 - end 113 113 - 114 114 - @spec spawn_and_yield(term, :link | :nolink, module, atom, [term]) :: pid 115 115 - def spawn_and_yield(scheduler, link, module, function, args) when link in [:link, :nolink] do 116 116 - resume = %Resume{ref: make_ref(), pid: self(), value: nil} 117 117 - spawn = %Spawn{module: module, function: function, args: args, parent_resume: resume, link: link == :link} 118 118 - 119 119 - GenServer.cast(scheduler, {:queue_task, self(), 0, spawn}) 120 120 - GenServer.cast(scheduler, {:yield, self()}) 121 121 - 122 122 - spawned_pid = await_resume(resume) 123 123 - spawned_pid 124 124 - end 125 125 - 126 126 - defp await_resume(%Resume{ref: ref} = _resume) do 127 127 - receive do 128 128 - {:resume, ^ref, value} -> value 129 129 - {:timeout, ^ref} -> exit(:timeout) 130 130 - end 131 131 - end 132 132 - 133 133 - @spec send(term, pid, term) :: :ok 134 134 - def send(scheduler, dest, message, delay \\ 0) do 135 135 - send = %Send{dest: dest, message: message} 136 136 - GenServer.cast(scheduler, {:queue_task, self(), delay, send}) 137 137 - :ok 138 138 - end 139 139 - 140 140 - @spec exit(term, pid, term) :: :ok 141 141 - def exit(scheduler, to_pid, reason) do 142 142 - exit = %Exit{from_pid: self(), to_pid: to_pid, reason: reason} 143 143 - GenServer.cast(scheduler, {:queue_task, self(), 0, exit}) 144 144 - :ok 145 145 - end 146 146 - 147 147 - @spec start_node(pid, atom, module, term) :: :ok 148 148 - def start_node(scheduler, name, app_module, args) when is_atom(name) and is_atom(app_module) do 149 149 - GenServer.call(scheduler, {:create_node, name, app_module, args}) 150 150 - :ok 151 151 - end 152 152 - 153 153 - @spec restart_node(pid, atom, non_neg_integer) :: :ok | {:error, :node_not_found | :node_stopped} 154 154 - def restart_node(scheduler, name, delay) when is_atom(name) do 155 155 - GenServer.call(scheduler, {:restart_node, name, delay}) 156 156 - end 157 157 - 158 158 - @spec list_nodes(pid) :: [atom] 159 159 - def list_nodes(scheduler) do 160 160 - GenServer.call(scheduler, :list_nodes) 161 161 - end 162 162 - 163 163 - def get_current_node(scheduler) do 164 164 - GenServer.call(scheduler, {:get_current_node, self()}) 165 165 - end 166 166 - 167 167 - @spec monitor(pid, pid) :: reference 168 168 - def monitor(scheduler, target_pid) do 169 169 - ref = make_ref() 170 170 - :ok = GenServer.call(scheduler, {:monitor, self(), target_pid, ref}) 171 171 - 172 172 - ref 173 173 - end 174 174 - 175 175 - @spec alias(pid) :: reference 176 176 - def alias(scheduler) do 177 177 - GenServer.call(scheduler, {:alias, self()}) 178 178 - end 179 179 - 180 180 - @spec unalias(pid, reference) :: boolean 181 181 - def unalias(scheduler, alias) when is_reference(alias) do 182 182 - GenServer.call(scheduler, {:unalias, self(), alias}) 183 183 - end 184 184 - 185 185 - @spec set_process_flag(pid, :trap_exit, boolean) :: :ok 186 186 - def set_process_flag(scheduler, :trap_exit = flag, value) when is_boolean(value) do 187 187 - {:ok, old_value} = GenServer.call(scheduler, {:set_process_flag, self(), flag, value}) 188 188 - old_value 189 189 - end 190 190 - 191 191 - def register_process(scheduler, pid, name) when is_pid(pid) and is_atom(name) do 192 192 - GenServer.call(scheduler, {:register_process, self(), pid, name}) 193 193 - end 194 194 - 195 195 - def whereis(scheduler, name) when is_atom(name) do 196 196 - GenServer.call(scheduler, {:whereis, self(), name}) 197 197 - end 198 198 - 199 199 - @spec get_file_store(pid) :: FileStore.t 200 200 - def get_file_store(scheduler) do 201 201 - GenServer.call(scheduler, {:get_file_store, self()}) 202 202 - end 203 203 - 204 204 - @spec current_time(term, pid) :: non_neg_integer 205 205 - def current_time(scheduler, for_pid \\ self()) when is_pid(for_pid) do 206 206 - GenServer.call(scheduler, {:current_time, for_pid}) 207 207 - end 208 208 - 209 209 - @spec configure_log_server(term, pid) :: :ok 210 210 - def configure_log_server(scheduler, log_server_pid) do 211 211 - GenServer.call(scheduler, {:configure_log_server, log_server_pid}) 212 212 - end 213 213 - 214 214 - @spec get_log(term) :: term 215 215 - def get_log(scheduler) do 216 216 - # We run the get_log through the scheduler to ensure 217 217 - # that all log messages from the scheduler process (which are cast) 218 218 - # have been processed first (messages are always received in order) 219 219 - GenServer.call(scheduler, :get_log) 220 220 - end 221 221 - 222 222 - def init(%{initial_pid: initial_pid, seed: seed}) do 223 223 - Process.flag(:trap_exit, true) 224 224 - :rand.seed(:exsss, seed) 225 225 - 226 226 - state = %State{ 227 227 - proc_store: ProcStore.new(), 228 228 - proc_queue: ProcQueue.new(), 229 229 - proc_registry: ProcRegistry.new(), 230 230 - # TODO: spawn per node instead 231 231 - file_stores: %{nonode: FileStore.new()}, 232 232 - } 233 233 - ProcStore.add_process(state.proc_store, initial_pid, :nonode) 234 234 - 235 235 - state = set_current_process(state, initial_pid) 236 236 - {:ok, state} 237 237 - end 238 238 - 239 239 - def handle_call({:alias, pid}, _from, %State{} = state) do 240 240 - alias = make_ref() 241 241 - ProcStore.add_alias(state.proc_store, pid, alias) 242 242 - {:reply, alias, state} 243 243 - end 244 244 - 245 245 - def handle_call({:unalias, pid, alias}, _from, %State{} = state) do 246 246 - case ProcStore.remove_alias(state.proc_store, pid, alias) do 247 247 - :ok -> {:reply, true, state} 248 248 - {:error, _err} -> {:reply, false, state} 249 249 - end 250 250 - end 251 251 - 252 252 - def handle_call({:set_process_flag, pid, flag, value}, _from, %State{} = state) do 253 253 - {:ok, old_value} = ProcStore.set_flag(state.proc_store, pid, flag, value) 254 254 - {:reply, {:ok, old_value}, state} 255 255 - end 256 256 - 257 257 - def handle_call({:register_process, from_pid, pid, name}, _from, %State{} = state) do 258 258 - # TODO: ensure name is valid? (not nil/true/false/:undefined) 259 259 - {:ok, %ProcState{node: from_node}} = ProcStore.fetch_state(state.proc_store, from_pid) 260 260 - {:ok, %ProcState{node: node}} = ProcStore.fetch_state(state.proc_store, pid) 261 261 - 262 262 - case node == from_node do 263 263 - true -> 264 264 - case ProcRegistry.register(state.proc_registry, pid, name, node) do 265 265 - :ok -> {:reply, :ok, state} 266 266 - {:error, _error} -> {:reply, :error, state} 267 267 - end 268 268 - 269 269 - false -> {:reply, :error, state} 270 270 - end 271 271 - end 272 272 - 273 273 - def handle_call({:whereis, from_pid, name}, _from, %State{} = state) do 274 274 - {:ok, %ProcState{node: node}} = ProcStore.fetch_state(state.proc_store, from_pid) 275 275 - 276 276 - case ProcRegistry.whereis(state.proc_registry, name, node) do 277 277 - {:ok, pid} -> {:reply, pid, state} 278 278 - {:error, :not_found} -> {:reply, nil, state} 279 279 - end 280 280 - end 281 281 - 282 282 - def handle_call({:get_file_store, pid}, _from, %State{} = state) when is_pid(pid) do 283 283 - {:ok, %ProcState{node: node}} = ProcStore.fetch_state(state.proc_store, pid) 284 284 - 285 285 - case Map.fetch(state.file_stores, node) do 286 286 - {:ok, file_store} -> 287 287 - {:reply, file_store, state} 288 288 - :error -> 289 289 - file_store = FileStore.new() 290 290 - state = %{state | file_stores: Map.put(state.file_stores, node, file_store)} 291 291 - {:reply, file_store, state} 292 292 - end 293 293 - end 294 294 - 295 295 - def handle_call({:current_time, for_pid}, _from, %State{} = state) when is_pid(for_pid) do 296 296 - # TODO: maybe we should store clock in microseconds? 297 297 - {:reply, state.clock * 1000, state} 298 298 - end 299 299 - 300 300 - def handle_call({:configure_log_server, log_server_pid}, _from, %State{} = state) do 301 301 - state = %{state | log_server_pid: log_server_pid} 302 302 - {:reply, :ok, state} 303 303 - end 304 304 - 305 305 - def handle_call(:get_log, _from, %State{} = state) do 306 306 - log = SimLog.get_log(state.log_server_pid) 307 307 - {:reply, log, state} 308 308 - end 309 309 - 310 310 - def handle_call({:monitor, pid, target_pid, ref}, _from, %State{} = state) when is_pid(pid) and is_pid(target_pid) and is_reference(ref) do 311 311 - case ProcStore.fetch_state(state.proc_store, target_pid) do 312 312 - {:ok, _proc} -> 313 313 - ProcStore.add_monitor(state.proc_store, pid, target_pid, ref) 314 314 - {:reply, :ok, state} 315 315 - 316 316 - :error -> 317 317 - send pid, {:DOWN, ref, :process, target_pid, :noproc} 318 318 - {:reply, :ok, state} 319 319 - end 320 320 - end 321 321 - 322 322 - def handle_call({:create_node, name, app_module, args}, _from, %State{} = state) when is_atom(name) and is_atom(app_module) do 323 323 - node = %Node{pid: nil, status: :stopped, name: name, app_module: app_module, args: args} 324 324 - state = %{state | nodes: [{node.name, node} | state.nodes]} 325 325 - 326 326 - queue_task(state, state.clock, %StartNode{name: node.name}) 327 327 - 328 328 - {:reply, :ok, state} 329 329 - end 330 330 - 331 331 - def handle_call({:restart_node, name, delay}, _from, %State{} = state) when is_atom(name) and is_integer(delay) do 332 332 - case List.keyfind(state.nodes, name, 0) do 333 333 - {^name, %Node{status: :running} = node} -> 334 334 - :ok = kill_node(state, node) 335 335 - 336 336 - node = %{node | pid: nil, status: :stopped} 337 337 - state = %{state | nodes: List.keyreplace(state.nodes, name, 0, {name, node})} 338 338 - 339 339 - start_node = %StartNode{name: node.name} 340 340 - state = queue_task(state, state.clock + delay, start_node) 341 341 - 342 342 - {:reply, :ok, state} 343 343 - 344 344 - {^name, %Node{status: :stopped}} -> {:reply, {:error, :node_stopped}, state} 345 345 - [] -> {:reply, {:error, :node_not_found}, state} 346 346 - end 347 347 - end 348 348 - 349 349 - def handle_call(:list_nodes, _from, %State{} = state) do 350 350 - nodes = 351 351 - state.nodes 352 352 - |> Enum.map(fn {_name, %Node{} = node} -> node.name end) 353 353 - |> Enum.reverse() 354 354 - {:reply, nodes, state} 355 355 - end 356 356 - 357 357 - def handle_call({:get_current_node, pid}, _from, %State{} = state) do 358 358 - {:ok, %ProcState{node: node}} = ProcStore.fetch_state(state.proc_store, pid) 359 359 - {:reply, node, state} 360 360 - end 361 361 - 362 362 - def handle_cast({:queue_task, pid, delay, %Send{} = send}, %State{} = state) when is_pid(pid) and is_integer(delay) do 363 363 - delay = 364 364 - case send.dest == pid do 365 365 - true -> delay 366 366 - # Minimum 10ms of latency for all messages not sent to self() 367 367 - # TODO: vary this based on whether processes are on the same node 368 368 - false -> max(delay, 10) 369 369 - end 370 370 - 371 371 - case delay do 372 372 - 0 -> 373 373 - log(state, {:send_now, state.clock, send.dest, send.message}) 374 374 - {:noreply, send_message(state, send)} 375 375 - 376 376 - delay when delay > 0 -> 377 377 - {:noreply, queue_task(state, state.clock + delay, send)} 378 378 - end 379 379 - end 380 380 - 381 381 - def handle_cast({:queue_task, pid, delay, task}, %State{} = state) when is_pid(pid) and is_integer(delay) do 382 382 - {:noreply, queue_task(state, state.clock + delay, task)} 383 383 - end 384 384 - 385 385 - def handle_cast({:await_message, pid, check_fun, timeout_ms, %Resume{} = resume}, %State{} = state) when is_pid(pid) and is_function(check_fun) do 386 386 - # Add new timeout if needed 387 387 - timeout_key = 388 388 - case timeout_ms do 389 389 - :infinity -> 390 390 - nil 391 391 - timeout_ms -> 392 392 - {:ok, key} = ProcQueue.enqueue(state.proc_queue, state.clock + timeout_ms, %Timeout{pid: pid, ref: resume.ref}) 393 393 - key 394 394 - end 395 395 - 396 396 - ProcStore.track_await(state.proc_store, pid, check_fun, resume, timeout_key) 397 397 - {:noreply, state} 398 398 - end 399 399 - 400 400 - def handle_cast({:yield, pid}, %State{} = state) when is_pid(pid) do 401 401 - case state.current do 402 402 - ^pid -> :noop 403 403 - end 404 404 - {:noreply, %{state | current: nil} |> perform_next_task()} 405 405 - end 406 406 - 407 407 - def handle_info({:EXIT, pid, reason} = message, %State{} = state) do 408 408 - case state.current do 409 409 - ^pid -> 410 410 - state = clean_up_dead_process(state, pid, reason) 411 411 - {:noreply, %{state | current: nil} |> perform_next_task()} 412 412 - 413 413 - _ -> 414 414 - raise """ 415 415 - Received invalid EXIT message: #{inspect(message)} 416 416 - 417 417 - Scheduler state: 418 418 - 419 419 - #{inspect(state, pretty: true)} 420 420 - """ 421 421 - end 422 422 - end 423 423 - 424 424 - defp queue_task(%State{} = state, time, %Resume{} = resume) when is_integer(time) do 425 425 - {:ok, key} = ProcQueue.enqueue(state.proc_queue, time, resume) 426 426 - ProcStore.track_queued(state.proc_store, resume.pid, key) 427 427 - 428 428 - state 429 429 - end 430 430 - 431 431 - defp queue_task(%State{} = state, time, task) when is_integer(time) do 432 432 - ProcQueue.enqueue(state.proc_queue, time, task) 433 433 - state 434 434 - end 435 435 - 436 436 - defp perform_next_task(%State{resumes_without_send: resumes_without_send} = state) when resumes_without_send >= 1000 do 437 437 - # This is a rather primitive form of deadlock detection but it works for now 438 438 - raise """ 439 439 - Resumed #{inspect(resumes_without_send)} times without sending a message! Possible deadlock? 440 440 - 441 441 - Scheduler state: 442 442 - 443 443 - #{inspect(state, pretty: true)} 444 444 - """ 445 445 - end 446 446 - 447 447 - defp perform_next_task(%State{} = state) do 448 448 - {time, task} = 449 449 - case ProcQueue.pop_next(state.proc_queue) do 450 450 - {:ok, {time, task}} -> 451 451 - {time, task} 452 452 - 453 453 - {:error, :empty} -> 454 454 - raise """ 455 455 - Attempted to call `perform_next_task/1` but the queue is empty! Possible deadlock? 456 456 - 457 457 - Scheduler state: 458 458 - 459 459 - #{inspect(state, pretty: true)} 460 460 - """ 461 461 - end 462 462 - 463 463 - if time < state.clock do 464 464 - raise """ 465 465 - Time ran backwards! 466 466 - 467 467 - Scheduler state: 468 468 - 469 469 - #{inspect(state, pretty: true)} 470 470 - """ 471 471 - end 472 472 - 473 473 - %{state | clock: time} 474 474 - |> perform(task) 475 475 - end 476 476 - 477 477 - defp perform(%State{} = state, %Spawn{} = spawn) do 478 478 - parent_pid = spawn.parent_resume.pid 479 479 - {:ok, %ProcState{node: node}} = ProcStore.fetch_state(state.proc_store, parent_pid) 480 480 - 481 481 - child_pid = do_spawn_apply(spawn.module, spawn.function, spawn.args) 482 482 - 483 483 - ProcStore.add_process(state.proc_store, child_pid, node) 484 484 - if spawn.link, do: ProcStore.add_link(state.proc_store, parent_pid, child_pid) 485 485 - 486 486 - log(state, {:spawn, child_pid, {spawn.module, spawn.function, spawn.args}}) 487 487 - 488 488 - parent_resume = %{%Resume{} = spawn.parent_resume | value: child_pid} 489 489 - 490 490 - state 491 491 - |> set_current_process(child_pid) 492 492 - |> queue_task(state.clock, parent_resume) 493 493 - end 494 494 - 495 495 - defp perform(%State{} = state, %StartNode{} = start_node) do 496 496 - {_name, %Node{} = node} = List.keyfind(state.nodes, start_node.name, 0) 497 497 - pid = do_spawn_apply(node.app_module, :start, [:temporary, node.args]) 498 498 - 499 499 - ProcStore.add_process(state.proc_store, pid, node.name) 500 500 - log(state, {:start_node, pid, {node.name, node.app_module, node.args}}) 501 501 - 502 502 - node = %{node | pid: pid, status: :running} 503 503 - state = %{state | nodes: List.keyreplace(state.nodes, node.name, 0, {node.name, node})} 504 504 - 505 505 - state 506 506 - |> set_current_process(pid) 507 507 - end 508 508 - 509 509 - defp perform(%State{} = state, %Send{} = send) do 510 510 - log(state, {:send, state.clock, send.dest, send.message}) 511 511 - 512 512 - state 513 513 - |> send_message(send) 514 514 - |> perform_next_task() 515 515 - end 516 516 - 517 517 - defp perform(%State{} = state, %Exit{} = exit) do 518 518 - log(state, {:exit, state.clock, exit.to_pid, exit.reason}) 519 519 - 520 520 - case ProcStore.fetch_state(state.proc_store, exit.to_pid) do 521 521 - {:ok, %ProcState{} = proc} -> 522 522 - case proc.trap_exit and exit.reason != :kill do 523 523 - true -> 524 524 - state 525 525 - |> send_message(%Send{dest: proc.pid, message: {:EXIT, exit.from_pid, exit.reason}}) 526 526 - |> perform_next_task() 527 527 - false -> 528 528 - state = set_current_process(state, proc.pid) 529 529 - Process.exit(proc.pid, exit.reason) 530 530 - 531 531 - state 532 532 - end 533 533 - 534 534 - # Process already died 535 535 - :error -> perform_next_task(state) 536 536 - end 537 537 - end 538 538 - 539 539 - defp perform(%State{} = state, %Resume{} = resume) do 540 540 - #log(state, {:resume, time, resume.pid, resume.value}) 541 541 - 542 542 - case ProcStore.fetch_state(state.proc_store, resume.pid) do 543 543 - {:ok, _proc} -> 544 544 - ProcStore.clear_queued(state.proc_store, resume.pid) 545 545 - 546 546 - state = set_current_process(state, resume.pid) 547 547 - send resume.pid, {:resume, resume.ref, resume.value} 548 548 - 549 549 - %{state | resumes_without_send: state.resumes_without_send + 1} 550 550 - 551 551 - :error -> 552 552 - # TODO: this should be unreachable because we remove queue entries proactively 553 553 - raise "Tried to resume dead process" 554 554 - end 555 555 - end 556 556 - 557 557 - defp perform(%State{} = state, %Timeout{pid: pid, ref: ref}) do 558 558 - ProcStore.clear_await(state.proc_store, pid) 559 559 - 560 560 - state = set_current_process(state, pid) 561 561 - send pid, {:timeout, ref} 562 562 - 563 563 - state 564 564 - end 565 565 - 566 566 - defp set_current_process(%State{} = state, pid) when is_pid(pid) do 567 567 - %{state | current: pid} 568 568 - end 569 569 - 570 570 - defp send_message(%State{} = state, %Send{} = send) do 571 571 - dest_pid = 572 572 - case send.dest do 573 573 - # TODO: for local names, send/2 actually raises if the name is not registered 574 574 - name when is_atom(name) -> 575 575 - # TODO: use node from sender 576 576 - case ProcRegistry.whereis(state.proc_registry, name, :nonode) do 577 577 - {:ok, pid} -> pid 578 578 - {:error, :not_found} -> nil 579 579 - end 580 580 - 581 581 - {name, node} when is_atom(name) and is_atom(node) -> 582 582 - case ProcRegistry.whereis(state.proc_registry, name, node) do 583 583 - {:ok, pid} -> pid 584 584 - {:error, :not_found} -> nil 585 585 - end 586 586 - 587 587 - alias when is_reference(alias) -> 588 588 - case ProcStore.resolve_alias(state.proc_store, alias) do 589 589 - {:ok, pid} -> pid 590 590 - :error -> nil 591 591 - end 592 592 - 593 593 - pid when is_pid(pid) -> pid 594 594 - end 595 595 - 596 596 - send_message_to_pid(state, dest_pid, send.message) 597 597 - end 598 598 - 599 599 - # Ignore messages sent to unregistered names 600 600 - # TODO: technically we should raise if name is unregistered and local 601 601 - # (but only for immediate send) 602 602 - defp send_message_to_pid(%State{} = state, nil, _message) do 603 603 - state 604 604 - end 605 605 - 606 606 - defp send_message_to_pid(%State{} = state, dest_pid, message) when is_pid(dest_pid) do 607 607 - send dest_pid, message 608 608 - state = %{state | resumes_without_send: 0} 609 609 - 610 610 - case ProcStore.fetch_state(state.proc_store, dest_pid) do 611 611 - {:ok, %ProcState{await: {check_fun, resume}, queue_key: timeout_key}} -> 612 612 - case check_fun.(message) do 613 613 - true -> 614 614 - if timeout_key, do: ProcQueue.remove(state.proc_queue, timeout_key) 615 615 - ProcStore.clear_await(state.proc_store, dest_pid) 616 616 - 617 617 - queue_task(state, state.clock, resume) 618 618 - 619 619 - false -> state 620 620 - end 621 621 - 622 622 - _ -> state 623 623 - end 624 624 - end 625 625 - 626 626 - defp clean_up_dead_process(%State{} = state, pid, reason) when is_pid(pid) do 627 627 - state = 628 628 - state 629 629 - |> dispatch_links(pid, reason) 630 630 - |> dispatch_monitors(pid, reason) 631 631 - 632 632 - :ok = remove_process(state, pid) 633 633 - state 634 634 - end 635 635 - 636 636 - # Don't send exits for reason :normal 637 637 - defp dispatch_links(%State{} = state, target_pid, :normal) do 638 638 - {:ok, %ProcState{} = target_proc} = ProcStore.fetch_state(state.proc_store, target_pid) 639 639 - 640 640 - # Clean up links for the exiting process 641 641 - state = Enum.reduce(target_proc.linked_to, state, fn linked_pid, state -> 642 642 - {:ok, %ProcState{} = linked_proc} = ProcStore.fetch_state(state.proc_store, linked_pid) 643 643 - 644 644 - state = 645 645 - case linked_proc.trap_exit do 646 646 - true -> send_message(state, %Send{dest: linked_proc.pid, message: {:EXIT, target_proc.pid, :normal}}) 647 647 - false -> state 648 648 - end 649 649 - 650 650 - ProcStore.remove_link(state.proc_store, target_pid, linked_pid) 651 651 - state 652 652 - end) 653 653 - 654 654 - state 655 655 - end 656 656 - 657 657 - defp dispatch_links(%State{} = state, target_pid, reason) when is_pid(target_pid) do 658 658 - {:ok, %ProcState{} = target_proc} = ProcStore.fetch_state(state.proc_store, target_pid) 659 659 - 660 660 - {state, pids_to_kill} = traverse_linked(state, target_proc, reason) 661 661 - # Cycles are prevented, but there could still be duplicates 662 662 - pids_to_kill = Enum.uniq(pids_to_kill) 663 663 - 664 664 - # Kill processes 665 665 - Enum.each(pids_to_kill, fn pid -> 666 666 - kill_process(pid, reason) 667 667 - end) 668 668 - 669 669 - # Dispatch monitors for all killed processes 670 670 - state = 671 671 - Enum.reduce(pids_to_kill, state, fn pid, state -> 672 672 - dispatch_monitors(state, pid, reason) 673 673 - end) 674 674 - 675 675 - # Once processes are all dead and monitors dispatched, 676 676 - # we can remove them from the queue and store 677 677 - # 678 678 - # Note that these processes still have links to each other, 679 679 - # but it does not matter because both sides of the links will be deleted 680 680 - # along with them 681 681 - Enum.each(pids_to_kill, fn pid -> 682 682 - remove_process(state, pid) 683 683 - end) 684 684 - 685 685 - state 686 686 - end 687 687 - 688 688 - defp traverse_linked(%State{} = state, %ProcState{} = target_proc, reason, parents \\ MapSet.new()) do 689 689 - target_proc.linked_to 690 690 - |> Enum.reject(&MapSet.member?(parents, &1)) 691 691 - |> Enum.reduce({state, []}, fn linked_pid, {state, acc} -> 692 692 - {:ok, %ProcState{} = linked_proc} = ProcStore.fetch_state(state.proc_store, linked_pid) 693 693 - 694 694 - case linked_proc.trap_exit do 695 695 - true -> 696 696 - # Since one side of this link will survive, we must clean up the link 697 697 - ProcStore.remove_link(state.proc_store, target_proc.pid, linked_pid) 698 698 - 699 699 - # Send trapped EXIT message 700 700 - message = {:EXIT, target_proc.pid, reason} 701 701 - state = send_message(state, %Send{dest: linked_pid, message: message}) 702 702 - 703 703 - {state, acc} 704 704 - 705 705 - false -> 706 706 - {state, to_kill} = traverse_linked(state, linked_proc, reason, MapSet.put(parents, target_proc.pid)) 707 707 - {state, acc ++ [linked_proc.pid | to_kill]} 708 708 - end 709 709 - end) 710 710 - end 711 711 - 712 712 - defp dispatch_monitors(%State{} = state, target_pid, reason) when is_pid(target_pid) do 713 713 - {:ok, %ProcState{} = target_proc} = ProcStore.fetch_state(state.proc_store, target_pid) 714 714 - 715 715 - target_proc.monitored_by 716 716 - |> Enum.reduce(state, fn {ref, to_pid}, state -> 717 717 - ProcStore.remove_monitor(state.proc_store, to_pid, ref) 718 718 - 719 719 - msg = {:DOWN, ref, :process, target_pid, reason} 720 720 - send_message(state, %Send{dest: to_pid, message: msg}) 721 721 - end) 722 722 - end 723 723 - 724 724 - defp log(%State{} = state, event) do 725 725 - SimLog.log(state.log_server_pid, event) 726 726 - end 727 727 - 728 728 - @spec do_spawn_apply(module, atom, list) :: pid 729 729 - defp do_spawn_apply(m, f, a) do 730 730 - scheduler_pid = self() 731 731 - seed = random_seed() 732 732 - spawn_link(fn -> 733 733 - Hobbes.Construct.SimServer.set_scheduler_pid(scheduler_pid) 734 734 - :rand.seed(:exsss, seed) 735 735 - apply(m, f, a) 736 736 - end) 737 737 - end 738 738 - 739 739 - defp kill_node(%State{} = state, %Node{} = node) do 740 740 - ProcStore.list_processes(state.proc_store) 741 741 - |> Enum.filter(&(&1.node == node.name)) 742 742 - |> Enum.each(fn %ProcState{pid: pid} -> 743 743 - # TODO: the order here is not deterministic, but it probably doesn't matter? 744 744 - :ok = kill_process(pid, :shutdown) 745 745 - :ok = remove_process(state, pid) 746 746 - end) 747 747 - :ok 748 748 - end 749 749 - 750 750 - # Note: remember to call remove_process() afterwards to clear the process from the ProcStore 751 751 - defp kill_process(pid, reason) do 752 752 - Process.exit(pid, reason) 753 753 - receive do 754 754 - {:EXIT, ^pid, ^reason} -> :noop 755 755 - after 756 756 - # Sanity to stop the scheduler hanging, this should never happen 757 757 - 1000 -> raise "Process #{inspect(pid)} failed to exit" 758 758 - end 759 759 - :ok 760 760 - end 761 761 - 762 762 - defp remove_process(%State{} = state, pid) when is_pid(pid) do 763 763 - {:ok, %ProcState{queue_key: queue_key}} = ProcStore.fetch_state(state.proc_store, pid) 764 764 - if queue_key, do: ProcQueue.remove(state.proc_queue, queue_key) 765 765 - 766 766 - ProcStore.remove_process(state.proc_store, pid) 767 767 - ProcRegistry.remove_process(state.proc_registry, pid) 768 768 - :ok 769 769 - end 770 770 - 771 771 - defp random_seed, do: Enum.random(1..1_000_000) 772 772 - end

-126

lib/construct/scheduler/file_store.ex

reviewed

··· 1 1 - defmodule Hobbes.Construct.Scheduler.FileStore do 2 2 - alias Hobbes.KV.FlatKV 3 3 - 4 4 - @type t :: FlatKV.t 5 5 - @type posix :: :enoent | :eexist | :enotdir | :eisdir 6 6 - 7 7 - @spec new :: t 8 8 - def new do 9 9 - kv = FlatKV.new(public: true) 10 10 - FlatKV.put(kv, "/", "directory") 11 11 - 12 12 - kv 13 13 - end 14 14 - 15 15 - @spec mkdir(t, binary) :: :ok | {:error, posix} 16 16 - def mkdir(kv, path) do 17 17 - path = normalize_directory(path) 18 18 - 19 19 - case all_components_dir?(kv, path) do 20 20 - true -> 21 21 - case exists?(kv, path) do 22 22 - false -> 23 23 - FlatKV.put(kv, path, "directory") 24 24 - :ok 25 25 - 26 26 - true -> {:error, :eexist} 27 27 - end 28 28 - 29 29 - false -> {:error, :enoent} 30 30 - end 31 31 - end 32 32 - 33 33 - @spec mkdir_p(t, binary) :: :ok | {:error, posix} 34 34 - def mkdir_p(kv, path) do 35 35 - path 36 36 - |> components() 37 37 - |> Enum.reduce_while(:ok, fn p, :ok -> 38 38 - case exists?(kv, p) do 39 39 - false -> 40 40 - :ok = mkdir(kv, p) 41 41 - {:cont, :ok} 42 42 - 43 43 - true -> 44 44 - case dir?(kv, p) do 45 45 - true -> {:cont, :ok} 46 46 - false -> {:halt, {:error, :enotdir}} 47 47 - end 48 48 - end 49 49 - end) 50 50 - end 51 51 - 52 52 - @spec write(t, binary, binary) :: :ok | {:error, posix} 53 53 - def write(kv, path, contents) do 54 54 - case all_components_dir?(kv, path) do 55 55 - true -> 56 56 - FlatKV.put(kv, path, "file:" <> contents) 57 57 - :ok 58 58 - false -> 59 59 - {:error, :enoent} 60 60 - end 61 61 - end 62 62 - 63 63 - @spec read(t, binary) :: {:ok, binary} | {:error, posix} 64 64 - def read(kv, path) do 65 65 - case FlatKV.get(kv, path) do 66 66 - "file:" <> contents -> 67 67 - {:ok, contents} 68 68 - 69 69 - "directory" -> 70 70 - {:error, :eisdir} 71 71 - 72 72 - nil -> 73 73 - {:error, :enoent} 74 74 - end 75 75 - end 76 76 - 77 77 - @spec ls(t, binary) :: {:ok, [binary]} 78 78 - def ls(kv, path) do 79 79 - # TODO: error handling (enoent, enotdir) 80 80 - path = normalize_directory(path) 81 81 - # TODO: would be more efficient to only scan keys 82 82 - %{pairs: pairs} = FlatKV.scan(kv, path <> "/", path <> "0") 83 83 - 84 84 - child_paths = 85 85 - pairs 86 86 - |> Enum.map(fn {^path <> "/" <> file_name, _contents} -> file_name end) 87 87 - |> Enum.reject(fn name -> String.contains?(name, "/") end) 88 88 - 89 89 - {:ok, child_paths} 90 90 - end 91 91 - 92 92 - @spec dir?(t, binary) :: boolean 93 93 - def dir?(kv, path) do 94 94 - path = normalize_directory(path) 95 95 - FlatKV.get(kv, path) == "directory" 96 96 - end 97 97 - 98 98 - @spec exists?(t, binary) :: boolean 99 99 - def exists?(kv, path) do 100 100 - FlatKV.get(kv, path) != nil 101 101 - end 102 102 - 103 103 - defp components(path) do 104 104 - split = Path.split(path) 105 105 - 106 106 - Enum.map(1..length(split), fn count -> 107 107 - Enum.take(split, count) |> Path.join() 108 108 - end) 109 109 - end 110 110 - 111 111 - defp all_components_dir?(kv, path) do 112 112 - path 113 113 - |> components() 114 114 - |> case do 115 115 - [] -> [] 116 116 - list -> List.delete_at(list, -1) 117 117 - end 118 118 - |> Enum.all?(&dir?(kv, &1)) 119 119 - end 120 120 - 121 121 - defp normalize_directory("/"), do: "/" 122 122 - defp normalize_directory(path), do: String.trim_trailing(path, "/") 123 123 - 124 124 - @doc false 125 125 - def dump(kv), do: FlatKV.dump(kv) 126 126 - end

-45

lib/construct/scheduler/proc_queue.ex

reviewed

··· 1 1 - defmodule Hobbes.Construct.Scheduler.ProcQueue do 2 2 - @type t :: :ets.table 3 3 - @type key :: {non_neg_integer, non_neg_integer} 4 4 - 5 5 - def new do 6 6 - :ets.new(:proc_queue, [:private, :ordered_set]) 7 7 - end 8 8 - 9 9 - @spec enqueue(t, non_neg_integer, term) :: {:ok, key} 10 10 - def enqueue(table, time, task) when is_integer(time) do 11 11 - # Tasks are enqueued with key {time, i} to ensure FIFO order for a given time 12 12 - i = 13 13 - case :ets.prev(table, {time, :infinity}) do 14 14 - {^time, i} -> i + 1 15 15 - _ -> 0 16 16 - end 17 17 - 18 18 - key = {time, i} 19 19 - :ets.insert(table, {key, task}) 20 20 - {:ok, key} 21 21 - end 22 22 - 23 23 - @spec remove(t, key) :: :ok 24 24 - def remove(table, key) do 25 25 - case :ets.member(table, key) do 26 26 - true -> 27 27 - :ets.delete(table, key) 28 28 - :ok 29 29 - false -> 30 30 - raise "Tried to remove key that does not exist: #{inspect(key)}" 31 31 - end 32 32 - end 33 33 - 34 34 - @spec pop_next(t) :: {:ok, {non_neg_integer, term}} | {:error, :empty} 35 35 - def pop_next(table) do 36 36 - case :ets.first_lookup(table) do 37 37 - {_key, [{{time, _i} = key, task}]} -> 38 38 - :ets.delete(table, key) 39 39 - {:ok, {time, task}} 40 40 - 41 41 - :"$end_of_table" -> 42 42 - {:error, :empty} 43 43 - end 44 44 - end 45 45 - end

-58

lib/construct/scheduler/proc_registry.ex

reviewed

··· 1 1 - defmodule Hobbes.Construct.Scheduler.ProcRegistry do 2 2 - @type t :: :ets.table 3 3 - 4 4 - @spec new :: t 5 5 - def new do 6 6 - :ets.new(:proc_registry, [:set, :private]) 7 7 - end 8 8 - 9 9 - @spec register(t, pid, atom, atom) :: :ok | {:error, :name_already_registered | :pid_already_registered} 10 10 - def register(table, pid, name, node) when is_pid(pid) and is_atom(name) and is_atom(node) do 11 11 - case whereis(table, name, node) do 12 12 - {:error, :not_found} -> 13 13 - case :ets.lookup(table, pid) do 14 14 - [] -> 15 15 - :ets.insert(table, {{name, node}, pid}) 16 16 - :ets.insert(table, {pid, {name, node}}) 17 17 - :ok 18 18 - 19 19 - [{^pid, _value}] -> {:error, :pid_already_registered} 20 20 - end 21 21 - 22 22 - {:ok, _pid} -> {:error, :name_already_registered} 23 23 - end 24 24 - end 25 25 - 26 26 - @spec whereis(t, atom, atom) :: {:ok, pid} | {:error, :not_found} 27 27 - def whereis(table, name, node) when is_atom(name) and is_atom(node) do 28 28 - case :ets.lookup(table, {name, node}) do 29 29 - [{{^name, ^node}, pid}] -> {:ok, pid} 30 30 - [] -> {:error, :not_found} 31 31 - end 32 32 - end 33 33 - 34 34 - @spec unregister(t, atom, atom) :: :ok | :error 35 35 - def unregister(table, name, node) when is_atom(name) and is_atom(node) do 36 36 - case whereis(table, name, node) do 37 37 - {:ok, pid} -> 38 38 - :ets.delete(table, {name, node}) 39 39 - :ets.delete(table, pid) 40 40 - :ok 41 41 - 42 42 - {:error, :not_found} -> 43 43 - :error 44 44 - end 45 45 - end 46 46 - 47 47 - @spec remove_process(t, pid) :: :ok | :error 48 48 - def remove_process(table, pid) when is_pid(pid) do 49 49 - case :ets.lookup(table, pid) do 50 50 - [{pid, {name, node}}] -> 51 51 - :ets.delete(table, {name, node}) 52 52 - :ets.delete(table, pid) 53 53 - :ok 54 54 - 55 55 - [] -> :error 56 56 - end 57 57 - end 58 58 - end

-220

lib/construct/scheduler/proc_store.ex

reviewed

··· 1 1 - defmodule Hobbes.Construct.Scheduler.ProcStore do 2 2 - alias Hobbes.Construct.Scheduler.{ProcStore, ProcQueue} 3 3 - 4 4 - @type check_fun :: (-> boolean) 5 5 - 6 6 - defmodule ProcState do 7 7 - @type t :: %__MODULE__{ 8 8 - pid: pid, 9 9 - node: atom, 10 10 - monitors: %{reference => pid}, 11 11 - monitored_by: [{reference, pid}], 12 12 - linked_to: [pid], 13 13 - trap_exit: boolean, 14 14 - queue_key: ProcQueue.key | nil, 15 15 - await: {ProcStore.check_fun, term} | nil, 16 16 - } 17 17 - @enforce_keys [:pid, :node] 18 18 - defstruct [ 19 19 - monitors: %{}, 20 20 - monitored_by: [], 21 21 - linked_to: [], 22 22 - trap_exit: false, 23 23 - queue_key: nil, 24 24 - await: nil, 25 25 - ] ++ @enforce_keys 26 26 - end 27 27 - 28 28 - @type t :: %__MODULE__{ 29 29 - proc_table: :ets.table, 30 30 - alias_table: :ets.table, 31 31 - } 32 32 - 33 33 - @enforce_keys [:proc_table, :alias_table] 34 34 - defstruct @enforce_keys 35 35 - 36 36 - def new do 37 37 - %ProcStore{ 38 38 - proc_table: :ets.new(:proc_table, [:set, :private]), 39 39 - alias_table: :ets.new(:alias_table, [:set, :private]), 40 40 - } 41 41 - end 42 42 - 43 43 - @spec add_process(t, pid, atom) :: :ok 44 44 - def add_process(%ProcStore{} = ps, pid, node) when is_pid(pid) and is_atom(node) do 45 45 - state = %ProcState{pid: pid, node: node} 46 46 - 47 47 - :ets.insert(ps.proc_table, {pid, state}) 48 48 - :ok 49 49 - end 50 50 - 51 51 - @spec remove_process(t, pid) :: :ok 52 52 - def remove_process(%ProcStore{} = ps, pid) when is_pid(pid) do 53 53 - case fetch_state(ps, pid) do 54 54 - {:ok, _proc} -> 55 55 - :ets.delete(ps.proc_table, pid) 56 56 - :ok 57 57 - :error -> 58 58 - raise "Process #{inspect(pid)} cannot be removed because it does not exist" 59 59 - end 60 60 - end 61 61 - 62 62 - @spec fetch_state(t, pid) :: {:ok, ProcState.t} | :error 63 63 - def fetch_state(%ProcStore{} = ps, pid) when is_pid(pid) do 64 64 - case :ets.lookup(ps.proc_table, pid) do 65 65 - [{^pid, %ProcState{} = state}] -> 66 66 - {:ok, state} 67 67 - _ -> 68 68 - :error 69 69 - end 70 70 - end 71 71 - 72 72 - @spec track_queued(t, pid, ProcQueue.key) :: :ok 73 73 - def track_queued(%ProcStore{} = ps, pid, {_, _} = queue_key) when is_pid(pid) do 74 74 - {:ok, %ProcState{} = state} = fetch_state(ps, pid) 75 75 - if state.queue_key != nil, do: raise "Process is already in queue: #{inspect(state)}" 76 76 - 77 77 - state = %{state | queue_key: queue_key} 78 78 - :ets.insert(ps.proc_table, {pid, state}) 79 79 - :ok 80 80 - end 81 81 - 82 82 - @spec clear_queued(t, pid) :: :ok 83 83 - def clear_queued(%ProcStore{} = ps, pid) do 84 84 - {:ok, %ProcState{} = state} = fetch_state(ps, pid) 85 85 - 86 86 - :ets.insert(ps.proc_table, {pid, %{state | queue_key: nil}}) 87 87 - :ok 88 88 - end 89 89 - 90 90 - @spec track_await(t, pid, check_fun, term, ProcQueue.key | nil) :: :ok 91 91 - def track_await(%ProcStore{} = ps, pid, check_fun, resume, timeout_queue_key) 92 92 - when is_pid(pid) and is_function(check_fun) and is_struct(resume) and (is_nil(timeout_queue_key) or is_tuple(timeout_queue_key)) do 93 93 - {:ok, %ProcState{} = state} = fetch_state(ps, pid) 94 94 - if state.queue_key != nil, do: raise "Process is in queue: #{inspect(state)}" 95 95 - 96 96 - state = %{state | await: {check_fun, resume}, queue_key: timeout_queue_key} 97 97 - :ets.insert(ps.proc_table, {pid, state}) 98 98 - :ok 99 99 - end 100 100 - 101 101 - @spec clear_await(t, pid) :: :ok 102 102 - def clear_await(%ProcStore{} = ps, pid) do 103 103 - {:ok, %ProcState{} = state} = fetch_state(ps, pid) 104 104 - if !state.await, do: raise "Process is not awaiting: #{inspect(state)}" 105 105 - 106 106 - state = %{state | await: nil, queue_key: nil} 107 107 - :ets.insert(ps.proc_table, {pid, state}) 108 108 - :ok 109 109 - end 110 110 - 111 111 - @spec add_link(t, pid, pid) :: :ok | :error 112 112 - def add_link(%ProcStore{} = ps, pid1, pid2) do 113 113 - {:ok, %ProcState{} = state1} = fetch_state(ps, pid1) 114 114 - {:ok, %ProcState{} = state2} = fetch_state(ps, pid2) 115 115 - 116 116 - case pid2 in state1.linked_to do 117 117 - false -> 118 118 - state1 = Map.update!(state1, :linked_to, &[pid2 | &1]) 119 119 - state2 = Map.update!(state2, :linked_to, &[pid1 | &1]) 120 120 - 121 121 - :ets.insert(ps.proc_table, {pid1, state1}) 122 122 - :ets.insert(ps.proc_table, {pid2, state2}) 123 123 - :ok 124 124 - 125 125 - true -> :error 126 126 - end 127 127 - end 128 128 - 129 129 - @spec remove_link(t, pid, pid) :: :ok | :error 130 130 - def remove_link(%ProcStore{} = ps, pid1, pid2) do 131 131 - {:ok, %ProcState{} = state1} = fetch_state(ps, pid1) 132 132 - {:ok, %ProcState{} = state2} = fetch_state(ps, pid2) 133 133 - 134 134 - case pid2 in state1.linked_to do 135 135 - true -> 136 136 - state1 = Map.update!(state1, :linked_to, &List.delete(&1, pid2)) 137 137 - state2 = Map.update!(state2, :linked_to, &List.delete(&1, pid1)) 138 138 - 139 139 - :ets.insert(ps.proc_table, {pid1, state1}) 140 140 - :ets.insert(ps.proc_table, {pid2, state2}) 141 141 - :ok 142 142 - 143 143 - false -> :error 144 144 - end 145 145 - end 146 146 - 147 147 - @spec add_monitor(t, pid, pid, reference) :: :ok 148 148 - def add_monitor(%ProcStore{} = ps, pid, target_pid, ref) when is_pid(pid) and is_pid(target_pid) and is_reference(ref) do 149 149 - {:ok, %ProcState{} = proc_state} = fetch_state(ps, pid) 150 150 - {:ok, %ProcState{} = target_state} = fetch_state(ps, target_pid) 151 151 - 152 152 - monitors = Map.put(proc_state.monitors, ref, target_pid) 153 153 - monitored_by = [{ref, pid} | target_state.monitored_by] 154 154 - 155 155 - :ets.insert(ps.proc_table, {pid, %{proc_state | monitors: monitors}}) 156 156 - :ets.insert(ps.proc_table, {target_pid, %{target_state | monitored_by: monitored_by}}) 157 157 - :ok 158 158 - end 159 159 - 160 160 - @spec remove_monitor(t, pid, reference) :: :ok 161 161 - def remove_monitor(%ProcStore{} = ps, pid, ref) do 162 162 - {:ok, %ProcState{} = proc_state} = fetch_state(ps, pid) 163 163 - 164 164 - {target_pid, monitors} = Map.pop!(proc_state.monitors, ref) 165 165 - 166 166 - {:ok, %ProcState{} = target_state} = fetch_state(ps, target_pid) 167 167 - monitored_by = 168 168 - target_state.monitored_by 169 169 - |> Enum.filter(fn 170 170 - {^ref, _pid} -> false 171 171 - _ -> true 172 172 - end) 173 173 - 174 174 - :ets.insert(ps.proc_table, {pid, %{proc_state | monitors: monitors}}) 175 175 - :ets.insert(ps.proc_table, {target_pid, %{target_state | monitored_by: monitored_by}}) 176 176 - :ok 177 177 - end 178 178 - 179 179 - @spec add_alias(t, pid, reference) :: :ok 180 180 - def add_alias(%ProcStore{} = ps, pid, alias) when is_pid(pid) and is_reference(alias) do 181 181 - :ets.insert(ps.alias_table, {alias, pid}) 182 182 - :ok 183 183 - end 184 184 - 185 185 - @spec remove_alias(t, pid, reference) :: :ok | {:error, :wrong_process | :not_found} 186 186 - def remove_alias(%ProcStore{} = ps, pid, alias) when is_pid(pid) and is_reference(alias) do 187 187 - case resolve_alias(ps, alias) do 188 188 - {:ok, ^pid} -> 189 189 - :ets.delete(ps.alias_table, alias) 190 190 - :ok 191 191 - 192 192 - {:ok, _other} -> {:error, :wrong_process} 193 193 - :error -> {:error, :not_found} 194 194 - end 195 195 - end 196 196 - 197 197 - @spec resolve_alias(t, reference) :: {:ok, pid} | :error 198 198 - def resolve_alias(%ProcStore{} = ps, alias) when is_reference(alias) do 199 199 - case :ets.lookup(ps.alias_table, alias) do 200 200 - [{^alias, pid}] when is_pid(pid) -> 201 201 - {:ok, pid} 202 202 - [] -> 203 203 - :error 204 204 - end 205 205 - end 206 206 - 207 207 - @spec set_flag(t, pid, :trap_exit, boolean) :: {:ok, boolean} 208 208 - def set_flag(%ProcStore{} = ps, pid, :trap_exit, value) when is_boolean(value) do 209 209 - {:ok, %ProcState{} = state} = fetch_state(ps, pid) 210 210 - 211 211 - :ets.insert(ps.proc_table, {pid, %{state | trap_exit: value}}) 212 212 - {:ok, state.trap_exit} 213 213 - end 214 214 - 215 215 - @spec list_processes(t) :: [ProcState.t] 216 216 - def list_processes(%ProcStore{} = ps) do 217 217 - :ets.tab2list(ps.proc_table) 218 218 - |> Enum.map(fn {_pid, %ProcState{} = proc} -> proc end) 219 219 - end 220 220 - end

-74

lib/construct/sim_file.ex

reviewed

··· 1 1 - defmodule Hobbes.Construct.SimFile do 2 2 - alias Hobbes.Construct.Scheduler 3 3 - alias Hobbes.Construct.Scheduler.FileStore 4 4 - 5 5 - import Hobbes.Construct.SimServer, only: [get_scheduler_pid: 0] 6 6 - 7 7 - @spec mkdir(binary) :: :ok | {:error, File.posix} 8 8 - def mkdir(path) when is_binary(path) do 9 9 - case get_scheduler_pid() do 10 10 - spid when is_pid(spid) -> 11 11 - fs = Scheduler.get_file_store(spid) 12 12 - FileStore.mkdir(fs, path) 13 13 - end 14 14 - end 15 15 - 16 16 - @spec mkdir_p(binary) :: :ok | {:error, File.posix} 17 17 - def mkdir_p(path) when is_binary(path) do 18 18 - case get_scheduler_pid() do 19 19 - spid when is_pid(spid) -> 20 20 - fs = Scheduler.get_file_store(spid) 21 21 - FileStore.mkdir_p(fs, path) 22 22 - end 23 23 - end 24 24 - 25 25 - @spec write(binary, binary) :: :ok | {:error, File.posix} 26 26 - def write(path, contents) when is_binary(path) and is_binary(contents) do 27 27 - case get_scheduler_pid() do 28 28 - spid when is_pid(spid) -> 29 29 - fs = Scheduler.get_file_store(spid) 30 30 - FileStore.write(fs, path, contents) 31 31 - end 32 32 - end 33 33 - 34 34 - @spec read(binary) :: {:ok, binary} | {:error, File.posix} 35 35 - def read(path) when is_binary(path) do 36 36 - case get_scheduler_pid() do 37 37 - spid when is_pid(spid) -> 38 38 - fs = Scheduler.get_file_store(spid) 39 39 - FileStore.read(fs, path) 40 40 - end 41 41 - end 42 42 - 43 43 - @spec ls(binary) :: {:ok, [binary]} 44 44 - def ls(path) when is_binary(path) do 45 45 - case get_scheduler_pid() do 46 46 - spid when is_pid(spid) -> 47 47 - fs = Scheduler.get_file_store(spid) 48 48 - FileStore.ls(fs, path) 49 49 - end 50 50 - end 51 51 - 52 52 - @spec exists?(binary) :: boolean 53 53 - def exists?(path) when is_binary(path) do 54 54 - case get_scheduler_pid() do 55 55 - spid when is_pid(spid) -> 56 56 - fs = Scheduler.get_file_store(spid) 57 57 - FileStore.exists?(fs, path) 58 58 - end 59 59 - end 60 60 - 61 61 - @doc false 62 62 - def dump_paths do 63 63 - case get_scheduler_pid() do 64 64 - spid when is_pid(spid) -> 65 65 - fs = Scheduler.get_file_store(spid) 66 66 - 67 67 - FileStore.dump(fs) 68 68 - |> Enum.map(fn {path, _contents} -> path end) 69 69 - 70 70 - nil -> 71 71 - raise "SimFile.dump_paths/0 is a debug function which can only be called in simulation" 72 72 - end 73 73 - end 74 74 - end

-106

lib/construct/sim_internal.ex

reviewed

··· 1 1 - defmodule Hobbes.Construct.SimInternal do 2 2 - @moduledoc """ 3 3 - Internal functions for running a `SimServer` (registration, receive loop, etc). 4 4 - 5 5 - Similar to erlang's `:gen` but for `SimServer`. 6 6 - """ 7 7 - 8 8 - require Logger 9 9 - alias Hobbes.Construct.{SimServer, Scheduler} 10 10 - import Hobbes.Construct.SimServer, only: [yield_receive: 2, yield_receive: 3, fetch_scheduler_pid!: 0] 11 11 - 12 12 - @type name :: {:global, term} 13 13 - 14 14 - @spec start(module, term, Keyword.t) :: {:ok, pid} 15 15 - def start(module, init_arg, opts) do 16 16 - scheduler_pid = fetch_scheduler_pid!() 17 17 - pid = Scheduler.spawn_and_yield(scheduler_pid, :link, __MODULE__, :server_init, [module, init_arg, opts, self()]) 18 18 - 19 19 - # We await ack from the spawned SimServer just like :proc_lib.start() 20 20 - # Timeout is for sanity, an init should never take so long 21 21 - yield_receive(scheduler_pid, 60_000) do 22 22 - {:ack, ^pid} -> 23 23 - if name = opts[:name], do: SimServer.register(pid, name) 24 24 - {:ok, pid} 25 25 - 26 26 - # TODO: error handling? 27 27 - end 28 28 - end 29 29 - 30 30 - # These functions run within the SimServer process 31 31 - 32 32 - def server_init(module, arg, _options, parent) 33 33 - when is_atom(module) and is_pid(parent) do 34 34 - {:ok, initial_state} = module.init(arg) 35 35 - 36 36 - SimServer.send parent, {:ack, self()} 37 37 - sim_loop(module, initial_state) 38 38 - end 39 39 - 40 40 - defp sim_loop(module, state) do 41 41 - state = yield_receive(fetch_scheduler_pid!()) do 42 42 - message -> dispatch(message, module, state) 43 43 - end 44 44 - 45 45 - sim_loop(module, state) 46 46 - end 47 47 - 48 48 - defp dispatch({:"$sim_call", {_pid, [:alias | alias_ref] = tag} = from, request}, module, state) do 49 49 - try do 50 50 - module.handle_call(request, from, state) 51 51 - catch 52 52 - :exit, reason -> 53 53 - log_exit(reason, __STACKTRACE__, state) 54 54 - exit(reason) 55 55 - end 56 56 - |> case do 57 57 - {:reply, response, state} -> 58 58 - SimServer.send alias_ref, {tag, response} 59 59 - state 60 60 - {:noreply, state} -> 61 61 - state 62 62 - # TODO: error message for invalid reply 63 63 - end 64 64 - end 65 65 - 66 66 - defp dispatch({:"$sim_cast", request}, module, state) do 67 67 - try do 68 68 - module.handle_cast(request, state) 69 69 - catch 70 70 - :exit, reason -> 71 71 - log_exit(reason, __STACKTRACE__, state) 72 72 - exit(reason) 73 73 - end 74 74 - |> case do 75 75 - {:noreply, state} -> state 76 76 - # TODO: error message for invalid reply 77 77 - end 78 78 - end 79 79 - 80 80 - defp dispatch(message, module, state) do 81 81 - try do 82 82 - module.handle_info(message, state) 83 83 - catch 84 84 - :exit, reason -> 85 85 - log_exit(reason, __STACKTRACE__, state) 86 86 - exit(reason) 87 87 - end 88 88 - |> case do 89 89 - {:noreply, state} -> state 90 90 - # TODO: error message for invalid reply 91 91 - end 92 92 - end 93 93 - 94 94 - defp log_exit(:normal, _stacktrace, _state), do: :noop 95 95 - defp log_exit(:shutdown, _stacktrace, _state), do: :noop 96 96 - 97 97 - defp log_exit(reason, stacktrace, state) do 98 98 - require Logger 99 99 - Logger.error """ 100 100 - SimServer #{inspect(self())} terminating 101 101 - ** (stop) #{Exception.format_exit(reason)} 102 102 - #{Exception.format_stacktrace(stacktrace)}\ 103 103 - State: #{inspect(state)}\ 104 104 - """ 105 105 - end 106 106 - end

-124

lib/construct/sim_log.ex

reviewed

··· 1 1 - defmodule Hobbes.Construct.SimLog do 2 2 - use GenServer 3 3 - 4 4 - defmodule Known do 5 5 - @enforce_keys [:pids, :refs] 6 6 - defstruct @enforce_keys 7 7 - end 8 8 - 9 9 - defmodule State do 10 10 - @type t :: %__MODULE__{ 11 11 - num_events: non_neg_integer, 12 12 - rolling_hash: term, 13 13 - known: %Known{}, 14 14 - log: list, 15 15 - } 16 16 - @enforce_keys [:known, :log, :rolling_hash, :num_events] 17 17 - defstruct @enforce_keys 18 18 - end 19 19 - 20 20 - def start_link(args) do 21 21 - GenServer.start_link(__MODULE__, args) 22 22 - end 23 23 - 24 24 - @spec get_log(term) :: list 25 25 - def get_log(server) do 26 26 - GenServer.call(server, :get_log) 27 27 - end 28 28 - 29 29 - @spec log(term, term) :: :ok 30 30 - def log(server, event) do 31 31 - GenServer.cast(server, {:log, event}) 32 32 - end 33 33 - 34 34 - def init(_) do 35 35 - {:ok, %State{ 36 36 - num_events: 0, 37 37 - rolling_hash: nil, 38 38 - known: %Known{pids: %{}, refs: %{}}, 39 39 - log: [], 40 40 - }} 41 41 - end 42 42 - 43 43 - def handle_call(:get_log, _from, %State{} = state) do 44 44 - response = %{ 45 45 - num_events: state.num_events, 46 46 - rolling_hash: state.rolling_hash, 47 47 - known: state.known, 48 48 - log: Enum.reverse(state.log), 49 49 - } 50 50 - {:reply, response, state} 51 51 - end 52 52 - 53 53 - def handle_cast({:log, event}, %State{} = state) do 54 54 - {known, event} = homogenize(state.known, event) 55 55 - 56 56 - state = %{state | 57 57 - num_events: state.num_events + 1, 58 58 - rolling_hash: :erlang.phash2({state.rolling_hash, event}), 59 59 - known: known, 60 60 - # TODO: configurable 61 61 - #log: [event | state.log], 62 62 - } 63 63 - 64 64 - {:noreply, state} 65 65 - end 66 66 - 67 67 - defmodule Homogenized do 68 68 - @enforce_keys [:type, :i] 69 69 - defstruct @enforce_keys 70 70 - end 71 71 - 72 72 - defimpl Inspect, for: Homogenized do 73 73 - def inspect(%Homogenized{type: :pid, i: i}, _opts), do: "#sPID<#{i}>" 74 74 - def inspect(%Homogenized{type: :ref, i: i}, _opts), do: "#sRef<#{i}>" 75 75 - end 76 76 - 77 77 - defp homogenize(%Known{} = known, value) when is_pid(value) do 78 78 - case Map.get(known.pids, value) do 79 79 - nil -> 80 80 - hmg = %Homogenized{type: :pid, i: map_size(known.pids)} 81 81 - known = %Known{known | pids: Map.put(known.pids, value, hmg)} 82 82 - {known, Map.fetch!(known.pids, value)} 83 83 - %Homogenized{} = hmg -> 84 84 - {known, hmg} 85 85 - end 86 86 - end 87 87 - 88 88 - defp homogenize(%Known{} = known, value) when is_reference(value) do 89 89 - case Map.get(known.refs, value) do 90 90 - nil -> 91 91 - hmg = %Homogenized{type: :ref, i: map_size(known.refs)} 92 92 - known = %Known{known | refs: Map.put(known.refs, value, hmg)} 93 93 - {known, Map.fetch!(known.refs, value)} 94 94 - %Homogenized{} = hmg -> 95 95 - {known, hmg} 96 96 - end 97 97 - end 98 98 - 99 99 - defp homogenize(%Known{} = known, [:alias | ref]) when is_reference(ref) do 100 100 - {known, h_ref} = homogenize(known, ref) 101 101 - {known, [:alias | h_ref]} 102 102 - end 103 103 - 104 104 - defp homogenize(%Known{} = known, list) when is_list(list) do 105 105 - {known, h_list} = Enum.reduce(list, {known, []}, fn element, {known, acc_list} -> 106 106 - {k, h_el} = homogenize(known, element) 107 107 - {k, [h_el | acc_list]} 108 108 - end) 109 109 - {known, Enum.reverse(h_list)} 110 110 - end 111 111 - 112 112 - defp homogenize(%Known{} = known, tuple) when is_tuple(tuple) do 113 113 - {known, h_list} = homogenize(known, Tuple.to_list(tuple)) 114 114 - {known, List.to_tuple(h_list)} 115 115 - end 116 116 - 117 117 - # This one also handles structs 118 118 - defp homogenize(%Known{} = known, map) when is_map(map) do 119 119 - {known, h_list} = homogenize(known, Map.to_list(map)) 120 120 - {known, Map.new(h_list)} 121 121 - end 122 122 - 123 123 - defp homogenize(%Known{} = known, value), do: {known, value} 124 124 - end

-458

lib/construct/sim_server.ex

reviewed

··· 1 1 - defmodule Hobbes.Construct.SimServer do 2 2 - alias Hobbes.Construct.{SimInternal, Scheduler, SimLog} 3 3 - 4 4 - # TODO: support more GenServer init/1 return types here 5 5 - @callback init(init_arg :: term) :: {:ok, term} 6 6 - 7 7 - defmacro __using__(_opts) do 8 8 - quote do 9 9 - alias Hobbes.Construct.SimServer 10 10 - use GenServer 11 11 - end 12 12 - end 13 13 - 14 14 - @doc """ 15 15 - Yields until a matching message is received. 16 16 - 17 17 - ## Examples 18 18 - 19 19 - yield_receive(get_scheduler_pid()) do 20 20 - :good_message -> :ok 21 21 - :bad_message -> :error 22 22 - end 23 23 - 24 24 - """ 25 25 - defmacro yield_receive(scheduler_pid, timeout \\ :infinity, [do: do_block]) do 26 26 - # Here, we use `do_block` to build a function which 27 27 - # checks if a value matches the intended receive block 28 28 - # 29 29 - # e.g. 30 30 - # 31 31 - # yield_receive do 32 32 - # {:ok, :hello} -> :ok 33 33 - # {:error, error} -> error 34 34 - # end 35 35 - # 36 36 - # becomes... 37 37 - # 38 38 - # fn 39 39 - # {:ok, :hello} -> true 40 40 - # {:error, error} -> true 41 41 - # _ -> false 42 42 - # end 43 43 - check_block = 44 44 - Enum.map(do_block, fn {:->, _meta, [clause, _result]} -> 45 45 - {:->, [], [clause, true]} 46 46 - end) 47 47 - final_clause = 48 48 - quote do 49 49 - _ -> false 50 50 - end 51 51 - check_block = check_block ++ final_clause 52 52 - check_func_ast = {:fn, [], check_block} 53 53 - # Walk the ast and mark as generated to prevent warnings about unused variables 54 54 - # and redundant clauses 55 55 - # (also remove line numbers just because) 56 56 - check_func_ast = Macro.prewalk(check_func_ast, fn 57 57 - {a, meta, b} -> {a, meta |> Keyword.drop([:line, :column]) |> Keyword.put(:generated, true), b} 58 58 - other -> other 59 59 - end) 60 60 - 61 61 - quote do 62 62 - check_func = unquote(check_func_ast) 63 63 - 64 64 - # We use a ref to communicate whether a value was actually received 65 65 - # (since it is impossible for receive to return our unique ref) 66 66 - ref = make_ref() 67 67 - receive_func = fn -> 68 68 - receive do 69 69 - unquote(do_block) 70 70 - after 71 71 - 0 -> ref 72 72 - end 73 73 - end 74 74 - 75 75 - Hobbes.Construct.Scheduler.yield(unquote(scheduler_pid)) 76 76 - # We first check if the message queue already has a matching message 77 77 - # If so, we don't want to yield_until_message because the Scheduler would 78 78 - # have no way of knowing that a matching message is already in our queue 79 79 - case receive_func.() do 80 80 - ^ref -> 81 81 - # If there is no matching message, we yield until one is sent to us 82 82 - Hobbes.Construct.Scheduler.yield_until_message(unquote(scheduler_pid), check_func, unquote(timeout)) 83 83 - 84 84 - # Now that a message has arrived, we extract it with the receive 85 85 - # The case is a sanity check 86 86 - case receive_func.() do 87 87 - ^ref -> raise "Resumed for message but no message received!" 88 88 - value -> value 89 89 - end 90 90 - 91 91 - # A matching message was already in our queue, so we simply return it 92 92 - value -> 93 93 - value 94 94 - end 95 95 - end 96 96 - end 97 97 - 98 98 - @type request_id :: :gen_server.request_id | reference 99 99 - 100 100 - @scheduler_key :scheduler_key 101 101 - def set_scheduler_pid(pid), do: nil = Process.put(@scheduler_key, pid) 102 102 - def get_scheduler_pid, do: Process.get(@scheduler_key) 103 103 - def fetch_scheduler_pid!, do: get_scheduler_pid() || raise "No scheduler pid!" 104 104 - 105 105 - @spec simulated? :: boolean 106 106 - def simulated?, do: get_scheduler_pid() != nil 107 107 - 108 108 - @spec start_scheduler() :: {:ok, pid} 109 109 - def start_scheduler(seed \\ 100) do 110 110 - {:ok, scheduler_pid} = Scheduler.start_link(seed) 111 111 - 112 112 - {:ok, log_pid} = SimLog.start_link(nil) 113 113 - Scheduler.configure_log_server(scheduler_pid, log_pid) 114 114 - 115 115 - {:ok, scheduler_pid} 116 116 - end 117 117 - 118 118 - @doc """ 119 119 - Starts a `SimServer`. 120 120 - 121 121 - ## Examples 122 122 - 123 123 - iex> SimServer.start(__MODULE__, nil, name: {:global, :my_server}) 124 124 - {:ok, #PID{0.0.0}} 125 125 - 126 126 - """ 127 127 - @spec start_link(module, term, Keyword.t) :: GenServer.on_start 128 128 - def start_link(module, init_arg, options \\ []) do 129 129 - case get_scheduler_pid() do 130 130 - nil -> 131 131 - GenServer.start_link(module, init_arg, options) 132 132 - scheduler_pid when is_pid(scheduler_pid) -> 133 133 - SimInternal.start(module, init_arg, options) 134 134 - end 135 135 - end 136 136 - 137 137 - @doc """ 138 138 - Performs a `SimServer` call. 139 139 - 140 140 - ## Examples 141 141 - 142 142 - iex> SimServer.call({:global, :my_server}, :foo) 143 143 - :bar 144 144 - 145 145 - """ 146 146 - @spec call(term, term, timeout) :: term 147 147 - def call(server, request, timeout \\ 5000) do 148 148 - case get_scheduler_pid() do 149 149 - nil -> 150 150 - GenServer.call(server, request, timeout) 151 151 - scheduler_pid when is_pid(scheduler_pid) -> 152 152 - sim_call(scheduler_pid, server, request, timeout) 153 153 - end 154 154 - end 155 155 - 156 156 - @doc """ 157 157 - Sends a request to a SimServer. 158 158 - 159 159 - See `:gen_server.send_request/2` and `SimServer.receive_response/2`. 160 160 - 161 161 - ## Examples 162 162 - 163 163 - iex> req_id = SimServer.send_request(pid, :foo) 164 164 - iex> SimServer.receive_response(req_id) 165 165 - 166 166 - """ 167 167 - @spec send_request(term, term) :: request_id 168 168 - def send_request(server, request) do 169 169 - case get_scheduler_pid() do 170 170 - nil -> 171 171 - :gen_server.send_request(server, request) 172 172 - scheduler_pid when is_pid(scheduler_pid) -> 173 173 - sim_send_request(scheduler_pid, server, request) 174 174 - end 175 175 - end 176 176 - 177 177 - @doc """ 178 178 - Receives a response from a SimServer. 179 179 - 180 180 - See `:gen_server.receive_response/2` and `SimServer.send_request/2`. 181 181 - """ 182 182 - @spec receive_response(request_id, timeout) :: {:reply, term} | :timeout 183 183 - def receive_response(request_id, timeout \\ 5000) do 184 184 - case get_scheduler_pid() do 185 185 - nil -> 186 186 - :gen_server.receive_response(request_id, timeout) 187 187 - scheduler_pid when is_pid(scheduler_pid) -> 188 188 - sim_receive_response(scheduler_pid, request_id, timeout) 189 189 - end 190 190 - |> case do 191 191 - {:reply, _reply} = reply -> reply 192 192 - # Coalesce errors into timeouts for now 193 193 - {:error, _err} -> :timeout 194 194 - :timeout -> :timeout 195 195 - end 196 196 - end 197 197 - 198 198 - defp sim_call(scheduler_pid, server, request, timeout) do 199 199 - alias_ref = sim_send_request(scheduler_pid, server, request) 200 200 - 201 201 - try do 202 202 - yield_receive(scheduler_pid, timeout) do 203 203 - {[:alias | ^alias_ref], reply} -> reply 204 204 - end 205 205 - catch 206 206 - :exit, reason -> 207 207 - true = Scheduler.unalias(scheduler_pid, alias_ref) 208 208 - receive do 209 209 - {[:alias | ^alias_ref], reply} -> reply 210 210 - after 211 211 - 0 -> exit({reason, {__MODULE__, :call, [server, request, timeout]}}) 212 212 - end 213 213 - end 214 214 - end 215 215 - 216 216 - @dialyzer {:no_improper_lists, sim_send_request: 3} 217 217 - defp sim_send_request(scheduler_pid, server, request) do 218 218 - alias_ref = Scheduler.alias(scheduler_pid) 219 219 - Scheduler.send(scheduler_pid, server, {:"$sim_call", {self(), [:alias | alias_ref]}, request}) 220 220 - 221 221 - alias_ref 222 222 - end 223 223 - 224 224 - defp sim_receive_response(scheduler_pid, request_id, timeout) when is_reference(request_id) do 225 225 - alias_ref = request_id 226 226 - # TODO: catch timeout exit 227 227 - try do 228 228 - yield_receive(scheduler_pid, timeout) do 229 229 - {[:alias | ^alias_ref], reply} -> {:reply, reply} 230 230 - end 231 231 - catch 232 232 - :exit, :timeout -> 233 233 - true = Scheduler.unalias(scheduler_pid, alias_ref) 234 234 - receive do 235 235 - {[:alias | ^alias_ref], reply} -> {:reply, reply} 236 236 - after 237 237 - 0 -> :timeout 238 238 - end 239 239 - end 240 240 - end 241 241 - 242 242 - @doc """ 243 243 - Performs a `SimServer` cast. 244 244 - 245 245 - ## Examples 246 246 - 247 247 - iex> SimServer.cast({:global, :my_server}, :foo} 248 248 - :ok 249 249 - 250 250 - """ 251 251 - @spec cast(term, term) :: :ok 252 252 - def cast(server, request) do 253 253 - case get_scheduler_pid() do 254 254 - nil -> 255 255 - GenServer.cast(server, request) 256 256 - scheduler_pid when is_pid(scheduler_pid) -> 257 257 - sim_cast(scheduler_pid, server, request) 258 258 - end 259 259 - end 260 260 - 261 261 - defp sim_cast(scheduler_pid, server, request) do 262 262 - Scheduler.send(scheduler_pid, server, {:"$sim_cast", request}) 263 263 - :ok 264 264 - end 265 265 - 266 266 - @spec reply(GenServer.from, term) :: :ok 267 267 - def reply(client, response) do 268 268 - case get_scheduler_pid() do 269 269 - nil -> 270 270 - GenServer.reply(client, response) 271 271 - 272 272 - scheduler_pid when is_pid(scheduler_pid) -> 273 273 - {_client_pid, [:alias | alias_ref] = tag} = client 274 274 - Scheduler.send(scheduler_pid, alias_ref, {tag, response}) 275 275 - :ok 276 276 - end 277 277 - end 278 278 - 279 279 - @spec spawn(function) :: pid 280 280 - def spawn(fun) when is_function(fun) do 281 281 - case get_scheduler_pid() do 282 282 - scheduler_pid when is_pid(scheduler_pid) -> 283 283 - Scheduler.spawn_and_yield(scheduler_pid, :nolink, Kernel, :apply, [fun, []]) 284 284 - nil -> 285 285 - Kernel.spawn(fun) 286 286 - end 287 287 - end 288 288 - 289 289 - @spec spawn_link(function) :: pid 290 290 - def spawn_link(fun) when is_function(fun) do 291 291 - case get_scheduler_pid() do 292 292 - scheduler_pid when is_pid(scheduler_pid) -> 293 293 - Scheduler.spawn_and_yield(scheduler_pid, :link, Kernel, :apply, [fun, []]) 294 294 - nil -> 295 295 - Kernel.spawn_link(fun) 296 296 - end 297 297 - end 298 298 - 299 299 - @spec sleep(non_neg_integer) :: term 300 300 - def sleep(time) do 301 301 - case get_scheduler_pid() do 302 302 - scheduler_pid when is_pid(scheduler_pid) -> 303 303 - Scheduler.yield(scheduler_pid, time) 304 304 - nil -> 305 305 - :timer.sleep(time) 306 306 - end 307 307 - end 308 308 - 309 309 - @spec send(pid, term) :: term 310 310 - def send(dest, message) do 311 311 - case get_scheduler_pid() do 312 312 - scheduler_pid when is_pid(scheduler_pid) -> 313 313 - # TODO: time=nil 314 314 - Scheduler.send(scheduler_pid, dest, message, 0) 315 315 - nil -> 316 316 - Kernel.send(dest, message) 317 317 - end 318 318 - end 319 319 - 320 320 - @spec send_after(pid, term, non_neg_integer) :: term 321 321 - def send_after(dest, message, time) do 322 322 - case get_scheduler_pid() do 323 323 - scheduler_pid when is_pid(scheduler_pid) -> 324 324 - Scheduler.send(scheduler_pid, dest, message, time) 325 325 - nil -> 326 326 - Process.send_after(dest, message, time) 327 327 - end 328 328 - end 329 329 - 330 330 - @spec exit(pid, term) :: term 331 331 - def exit(pid, reason) do 332 332 - case get_scheduler_pid() do 333 333 - scheduler_pid when is_pid(scheduler_pid) -> 334 334 - Scheduler.exit(scheduler_pid, pid, reason) 335 335 - nil -> 336 336 - Process.exit(pid, reason) 337 337 - end 338 338 - end 339 339 - 340 340 - @spec node :: atom 341 341 - def node do 342 342 - case get_scheduler_pid() do 343 343 - scheduler_pid when is_pid(scheduler_pid) -> 344 344 - Scheduler.get_current_node(scheduler_pid) 345 345 - nil -> 346 346 - # TODO 347 347 - raise "Not supported" 348 348 - end 349 349 - end 350 350 - 351 351 - @spec list_nodes :: [atom] 352 352 - def list_nodes do 353 353 - if not simulated?(), do: raise "list_nodes/0 can only be called in simulation" 354 354 - scheduler_pid = fetch_scheduler_pid!() 355 355 - 356 356 - Scheduler.list_nodes(scheduler_pid) 357 357 - end 358 358 - 359 359 - @spec start_node(atom, module, term) :: :ok 360 360 - def start_node(name, app_module, args) when is_atom(name) and is_atom(app_module) do 361 361 - if not simulated?(), do: raise "start_node/3 can only be called in simulation" 362 362 - scheduler_pid = fetch_scheduler_pid!() 363 363 - 364 364 - Scheduler.start_node(scheduler_pid, name, app_module, args) 365 365 - end 366 366 - 367 367 - @spec restart_node(atom, non_neg_integer) :: :ok | {:error, :node_not_found | :node_stopped} 368 368 - def restart_node(name, delay_ms \\ 0) when is_atom(name) and is_integer(delay_ms) do 369 369 - if not simulated?(), do: raise "restart_node/1 can only be called in simulation" 370 370 - scheduler_pid = fetch_scheduler_pid!() 371 371 - 372 372 - Scheduler.restart_node(scheduler_pid, name, delay_ms) 373 373 - end 374 374 - 375 375 - @spec monitor(pid) :: reference 376 376 - def monitor(monitor_pid) do 377 377 - case get_scheduler_pid() do 378 378 - scheduler_pid when is_pid(scheduler_pid) -> 379 379 - Scheduler.monitor(scheduler_pid, monitor_pid) 380 380 - nil -> 381 381 - Process.monitor(monitor_pid) 382 382 - end 383 383 - end 384 384 - 385 385 - @spec flag(:trap_exit, boolean) :: boolean 386 386 - def flag(flag, value) when flag in [:trap_exit] and is_boolean(value) do 387 387 - case get_scheduler_pid() do 388 388 - scheduler_pid when is_pid(scheduler_pid) -> 389 389 - Scheduler.set_process_flag(scheduler_pid, flag, value) 390 390 - nil -> 391 391 - Process.flag(flag, value) 392 392 - end 393 393 - end 394 394 - 395 395 - @spec register(pid, atom) :: true 396 396 - def register(pid, name) when is_pid(pid) and is_atom(name) do 397 397 - case get_scheduler_pid() do 398 398 - scheduler_pid when is_pid(scheduler_pid) -> 399 399 - case Scheduler.register_process(scheduler_pid, pid, name) do 400 400 - :ok -> true 401 401 - :error -> raise ArgumentError 402 402 - end 403 403 - nil -> 404 404 - Process.register(pid, name) 405 405 - end 406 406 - end 407 407 - 408 408 - @spec whereis(atom) :: pid | nil 409 409 - def whereis(name) do 410 410 - case get_scheduler_pid() do 411 411 - scheduler_pid when is_pid(scheduler_pid) -> 412 412 - Scheduler.whereis(scheduler_pid, name) 413 413 - nil -> 414 414 - Process.whereis(name) 415 415 - end 416 416 - end 417 417 - 418 418 - @doc """ 419 419 - Returns a monotonic microsecond timestamp. 420 420 - 421 421 - Outside of simulation this is UNIX time. 422 422 - 423 423 - Inside the simulation the timestamp starts at `0`. 424 424 - 425 425 - ## Examples 426 426 - 427 427 - iex> current_time() 428 428 - 1000 429 429 - 430 430 - iex> current_time() 431 431 - 1100 432 432 - 433 433 - """ 434 434 - @spec current_time :: non_neg_integer 435 435 - def current_time do 436 436 - case get_scheduler_pid() do 437 437 - scheduler_pid when is_pid(scheduler_pid) -> 438 438 - Scheduler.current_time(scheduler_pid) 439 439 - nil -> 440 440 - System.monotonic_time(:microsecond) 441 441 - end 442 442 - end 443 443 - 444 444 - @spec deterministic_random(Enum.t) :: Enum.element 445 445 - def deterministic_random(enumerable), do: Enum.random(enumerable) 446 446 - 447 447 - @spec simulate_work(integer | [integer]) :: :ok 448 448 - def simulate_work(delay_or_delays) do 449 449 - case get_scheduler_pid() do 450 450 - nil -> :noop 451 451 - pid -> do_work_delay(delay_or_delays, pid) 452 452 - end 453 453 - :ok 454 454 - end 455 455 - 456 456 - defp do_work_delay([_ | _] = delays, pid), do: Scheduler.yield(pid, Enum.random(delays)) 457 457 - defp do_work_delay(delay, pid) when is_integer(delay), do: Scheduler.yield(pid, delay) 458 458 - end

-60

lib/construct/sim_utils.ex

reviewed

··· 1 1 - defmodule Hobbes.Construct.SimUtils do 2 2 - require Logger 3 3 - 4 4 - @doc """ 5 5 - Builds a printable message for the given log. 6 6 - 7 7 - Options: 8 8 - - `:full` - returns a full log message including PIDs and events. 9 9 - """ 10 10 - @spec build_log_message(%{log: list, known: map}, keyword) :: String.t 11 11 - def build_log_message(%{log: log, known: known} = log_state, opts \\ []) do 12 12 - pid_text = 13 13 - known.pids 14 14 - |> Enum.sort_by(fn {_k, v} -> v end) 15 15 - |> Enum.map(fn {k, v} -> "#{inspect(k)} -> #{inspect(v)}" end) 16 16 - |> Enum.join("\n") 17 17 - 18 18 - log_text = 19 19 - case (log_length = length(log)) > 100 do 20 20 - true -> 21 21 - first = Enum.take(log, 50) 22 22 - last = Enum.take(log, -50) 23 23 - format_log(first) <> "\n\n... #{inspect(log_length - 100)} more ...\n\n" <> format_log(last) 24 24 - 25 25 - false -> 26 26 - format_log(log) 27 27 - end 28 28 - 29 29 - hash = log_state.rolling_hash 30 30 - # For readability (difficult to remember integers) 31 31 - # sha256 results in better distribution, the bytes from phash2 are too visually similar 32 32 - hash_alpha = :crypto.hash(:sha256, <<hash::integer-size(32)>>) |> Base.encode32(padding: false) |> String.slice(0, 8) 33 33 - 34 34 - full_text = if opts[:full] do 35 35 - """ 36 36 - Scheduler Log: 37 37 - 38 38 - PIDs: 39 39 - #{pid_text} 40 40 - 41 41 - Log: 42 42 - #{log_text} 43 43 - 44 44 - """ 45 45 - else 46 46 - "" 47 47 - end 48 48 - 49 49 - full_text <> """ 50 50 - Logged #{log_state.num_events} events with hash #{inspect(hash)} 51 51 - SHA256: #{hash_alpha}\ 52 52 - """ 53 53 - end 54 54 - 55 55 - defp format_log(events) when is_list(events) do 56 56 - events 57 57 - |> Enum.map(&inspect(&1, [])) 58 58 - |> Enum.join("\n") 59 59 - end 60 60 - end

-368

lib/hybrid_kv.ex

reviewed

··· 1 1 - defmodule Hobbes.HybridKV do 2 2 - alias Hobbes.{HybridKV, MemKV, RangeForest, Utils} 3 3 - alias Hobbes.RangeForest.RangeTree 4 4 - alias Hobbes.KV.{FlatKV, FlatStorageKV, MutationLog} 5 5 - alias Hobbes.Structs.RangeResult 6 6 - 7 7 - @type t :: %__MODULE__{ 8 8 - mem_kv: term, 9 9 - storage_module: module, 10 10 - storage_kv: FlatKV.t | FlatStorageKV.t, 11 11 - deleted_forest: term, 12 12 - mutation_log: MutationLog.t, 13 13 - flushed_version: non_neg_integer, 14 14 - } 15 15 - 16 16 - @enforce_keys [:mem_kv, :storage_module, :storage_kv, :deleted_forest, :mutation_log, :flushed_version] 17 17 - defstruct @enforce_keys 18 18 - 19 19 - def new(opts \\ []) do 20 20 - {storage_module, storage_kv} = 21 21 - case Keyword.get(opts, :path) do 22 22 - nil -> {FlatKV, FlatKV.new()} 23 23 - path when is_binary(path) -> {FlatStorageKV, FlatStorageKV.new(path)} 24 24 - end 25 25 - 26 26 - new(storage_module, storage_kv) 27 27 - end 28 28 - 29 29 - def new(storage_module, storage_kv) when is_atom(storage_module) do 30 30 - %HybridKV{ 31 31 - mem_kv: MemKV.new(), 32 32 - storage_module: storage_module, 33 33 - storage_kv: storage_kv, 34 34 - deleted_forest: RangeForest.new(), 35 35 - mutation_log: MutationLog.new(), 36 36 - flushed_version: 0, 37 37 - } 38 38 - end 39 39 - 40 40 - @spec put(%HybridKV{}, non_neg_integer, binary, binary) :: %HybridKV{} 41 41 - def put(%HybridKV{} = kv, version, key, value) 42 42 - when is_integer(version) and version >= 0 and is_binary(key) and is_binary(value) do 43 43 - :ok = MemKV.put(kv.mem_kv, version, key, value) 44 44 - 45 45 - case RangeForest.split_at(kv.deleted_forest, version, key) do 46 46 - {:updated, deleted_forest} -> %HybridKV{kv | deleted_forest: deleted_forest} 47 47 - :noop -> kv 48 48 - end 49 49 - end 50 50 - 51 51 - @spec delete(%HybridKV{}, non_neg_integer, binary) :: :ok 52 52 - def delete(%HybridKV{} = kv, version, key) when is_integer(version) and version >= 0 and is_binary(key) do 53 53 - :ok = MemKV.delete(kv.mem_kv, version, key) 54 54 - end 55 55 - 56 56 - @spec delete_range(%HybridKV{}, non_neg_integer, binary, binary) :: %HybridKV{} 57 57 - def delete_range(%HybridKV{} = kv, version, start_key, end_key) do 58 58 - %HybridKV{kv | deleted_forest: RangeForest.add_range(kv.deleted_forest, version, start_key, end_key)} 59 59 - end 60 60 - 61 61 - @spec get(%HybridKV{}, non_neg_integer, binary) :: binary | nil 62 62 - def get(%HybridKV{} = kv, version, key) 63 63 - when is_integer(version) and version >= 0 and is_binary(key) do 64 64 - kv.deleted_forest 65 65 - |> RangeForest.tree_at(version) 66 66 - |> RangeTree.intersect_key(kv.flushed_version + 1, key) 67 67 - |> case do 68 68 - nil -> 69 69 - # There are no overlapping range deletes 70 70 - case MemKV.get(kv.mem_kv, version, 0, key) do 71 71 - nil -> kv.storage_module.get(kv.storage_kv, key) 72 72 - :deleted -> nil 73 73 - value -> value 74 74 - end 75 75 - 76 76 - {_sk, _ek, range_deleted_at} -> 77 77 - # There is an overlapping range delete, so we return nil if 78 78 - # a newer key is not found 79 79 - case MemKV.get(kv.mem_kv, version, range_deleted_at + 1, key) do 80 80 - nil -> nil 81 81 - :deleted -> nil 82 82 - value -> value 83 83 - end 84 84 - end 85 85 - end 86 86 - 87 87 - @spec scan(%HybridKV{}, non_neg_integer, binary, binary, keyword) :: RangeResult.t 88 88 - def scan(%HybridKV{} = kv, version, start_key, end_key, opts \\ []) do 89 89 - limit = Keyword.get(opts, :limit, :infinity) 90 90 - reverse = Keyword.get(opts, :reverse, false) 91 91 - 92 92 - deleted_ranges = 93 93 - kv.deleted_forest 94 94 - |> RangeForest.tree_at(version) 95 95 - |> RangeTree.intersect_range(kv.flushed_version, start_key, end_key) 96 96 - 97 97 - read_limit = case limit do 98 98 - :infinity -> :infinity 99 99 - limit -> limit + 1 100 100 - end 101 101 - 102 102 - {pairs, count} = 103 103 - case reverse do 104 104 - false -> do_scan(:forward, kv, deleted_ranges, version, start_key, end_key, read_limit, 0) 105 105 - true -> do_scan(:backward, kv, Enum.reverse(deleted_ranges), version, start_key, end_key, read_limit, 0) 106 106 - end 107 107 - 108 108 - # We over-read by 1 and then use the extra read 109 109 - # to check if there are more 110 110 - cond do 111 111 - count < read_limit -> 112 112 - %RangeResult{pairs: pairs, count: count, more: false} 113 113 - count == read_limit -> 114 114 - %RangeResult{pairs: Enum.take(pairs, limit), count: limit, more: true} 115 115 - end 116 116 - end 117 117 - 118 118 - # This exists purely as a sanity check and should be unreachable except for bugs 119 119 - defp do_scan(_direction, _kv, _deleted_ranges, _version, _start_key, _end_key, _limit, 1000), do: raise "Scan caught in loop!" 120 120 - 121 121 - # Abandon hope all ye who enter here 122 122 - defp do_scan(:forward, %HybridKV{} = kv, deleted_ranges, version, start_key, end_key, limit, scan_count) do 123 123 - {merged_pairs, scanned_end_key, deleted_ranges} = 124 124 - case deleted_ranges do 125 125 - [{sk, ek, del_v} | rest] when sk <= start_key -> 126 126 - # We are "inside" a range clear, so we scan mem only to the end of min(ek, end_key) 127 127 - mem_result = MemKV.scan(kv.mem_kv, version, del_v + 1, start_key, min(ek, end_key), limit: limit) 128 128 - {_start, mem_end_key} = mem_result.range 129 129 - # Merge just to clear :deleted 130 130 - pairs = merge([], mem_result.pairs, false) 131 131 - 132 132 - # Note: it would be wrong to return "rest" here if we are still inside 133 133 - # the deleted range (due to hitting the limit), however if a mem scan 134 134 - # hits the limit then we are done scanning because there is nothing 135 135 - # for the tombstones to clear out, so we don't care 136 136 - {pairs, mem_end_key, rest} 137 137 - 138 138 - _ -> 139 139 - # If there is a range delete ahead, we want to stop this hybrid scan before that 140 140 - # delete's start_key so that the next iteration will scan it mem only (the previous clause) 141 141 - stop_key = case deleted_ranges do 142 142 - [{sk, _ek, _del_v} | _] -> sk 143 143 - [] -> end_key 144 144 - end 145 145 - 146 146 - storage_result = kv.storage_module.scan(kv.storage_kv, start_key, stop_key, limit: limit) 147 147 - {_start, storage_end_key} = storage_result.range 148 148 - 149 149 - # Read mem only up to the end of the range scanned by storage 150 150 - mem_result = MemKV.scan(kv.mem_kv, version, 0, start_key, storage_end_key, limit: limit) 151 151 - {_start, mem_end_key} = mem_result.range 152 152 - 153 153 - # Both KVs were scanned up to this key 154 154 - # Anything past this key was only scanned by storage and must be discarded 155 155 - scanned_end_key = min(storage_end_key, mem_end_key) 156 156 - 157 157 - merged_pairs = 158 158 - merge(storage_result.pairs, mem_result.pairs, false) 159 159 - # TODO: more efficient to do this in merge/3 160 160 - |> Enum.take_while(fn {k, _v} -> k < scanned_end_key end) 161 161 - 162 162 - {merged_pairs, scanned_end_key, deleted_ranges} 163 163 - end 164 164 - 165 165 - # TODO: compute count in merge/3 166 166 - count = length(merged_pairs) 167 167 - 168 168 - cond do 169 169 - count > limit -> 170 170 - # We got more pairs than we asked for, which can happen if the 171 171 - # storage and mem keys are disjoint (which is fairly likely in practice) 172 172 - {Enum.take(merged_pairs, limit), limit} 173 173 - 174 174 - count == limit -> 175 175 - # We got exactly what we wanted! 176 176 - {merged_pairs, count} 177 177 - 178 178 - count < limit -> 179 179 - case scanned_end_key == end_key do 180 180 - true -> 181 181 - # We got <limit pairs but we really did scan the full range 182 182 - {merged_pairs, count} 183 183 - false -> 184 184 - # We got <limit pairs and we have not yet scanned the full range, 185 185 - # so we need to keep scanning 186 186 - # 187 187 - # This could be because we hit the limit and then the :deleted 188 188 - # tombstones cleared out enough pairs to bring us back under, 189 189 - # or because we ran into a range delete and had to skip over it 190 190 - {next_pairs, next_count} = do_scan(:forward, kv, deleted_ranges, version, scanned_end_key, end_key, subtract_limit(limit, count), scan_count + 1) 191 191 - {merged_pairs ++ next_pairs, count + next_count} 192 192 - end 193 193 - end 194 194 - end 195 195 - 196 196 - # See :forward for comments, the :backward version is the same except key logic 197 197 - # is inverted (deals with start_key instead of end_key) 198 198 - defp do_scan(:backward, %HybridKV{} = kv, deleted_ranges, version, start_key, end_key, limit, scan_count) do 199 199 - {merged_pairs, scanned_start_key, deleted_ranges} = 200 200 - case deleted_ranges do 201 201 - [{sk, ek, del_v} | rest] when ek >= end_key -> 202 202 - mem_result = MemKV.scan(kv.mem_kv, version, del_v + 1, max(sk, start_key), end_key, limit: limit, reverse: true) 203 203 - {mem_start_key, _end_key} = mem_result.range 204 204 - pairs = merge([], mem_result.pairs, true) 205 205 - {pairs, mem_start_key, rest} 206 206 - 207 207 - _ -> 208 208 - stop_key = case deleted_ranges do 209 209 - [{_sk, ek, _del_v} | _] -> ek 210 210 - [] -> start_key 211 211 - end 212 212 - 213 213 - storage_result = kv.storage_module.scan(kv.storage_kv, stop_key, end_key, limit: limit, reverse: true) 214 214 - {storage_start_key, _end_key} = storage_result.range 215 215 - 216 216 - mem_result = MemKV.scan(kv.mem_kv, version, 0, storage_start_key, end_key, limit: limit, reverse: true) 217 217 - {mem_start_key, _end_key} = mem_result.range 218 218 - 219 219 - scanned_start_key = max(storage_start_key, mem_start_key) 220 220 - 221 221 - merged_pairs = 222 222 - merge(storage_result.pairs, mem_result.pairs, true) 223 223 - |> Enum.take_while(fn {k, _v} -> k >= scanned_start_key end) 224 224 - 225 225 - {merged_pairs, scanned_start_key, deleted_ranges} 226 226 - end 227 227 - 228 228 - count = length(merged_pairs) 229 229 - 230 230 - cond do 231 231 - count > limit -> 232 232 - {Enum.take(merged_pairs, limit), limit} 233 233 - 234 234 - count == limit -> 235 235 - {merged_pairs, count} 236 236 - 237 237 - count < limit -> 238 238 - case scanned_start_key == start_key do 239 239 - true -> 240 240 - {merged_pairs, count} 241 241 - false -> 242 242 - {next_pairs, next_count} = do_scan(:backward, kv, deleted_ranges, version, start_key, scanned_start_key, subtract_limit(limit, count), scan_count + 1) 243 243 - {merged_pairs ++ next_pairs, count + next_count} 244 244 - end 245 245 - end 246 246 - end 247 247 - 248 248 - defp subtract_limit(:infinity, _n), do: :infinity 249 249 - defp subtract_limit(limit, n), do: limit - n 250 250 - 251 251 - @spec merge([{binary, binary}], [{binary, binary}], boolean) :: [{binary, binary}] 252 252 - defp merge(list1, list2, reverse), do: do_merge(reverse, list1, list2, []) |> Enum.reverse() 253 253 - 254 254 - defp do_merge(_reverse, [], [], acc), do: acc 255 255 - defp do_merge(reverse, [p1 | rest1], [], acc), do: do_merge(reverse, rest1, [], acc_pair(p1, acc)) 256 256 - defp do_merge(reverse, [], [p2 | rest2], acc), do: do_merge(reverse, [], rest2, acc_pair(p2, acc)) 257 257 - 258 258 - defp do_merge(false, [{k1, _} = p1 | rest1] = list1, [{k2, _} = p2 | rest2] = list2, acc) do 259 259 - cond do 260 260 - k1 < k2 -> 261 261 - do_merge(false, rest1, list2, acc_pair(p1, acc)) 262 262 - k1 > k2 -> 263 263 - do_merge(false, list1, rest2, acc_pair(p2, acc)) 264 264 - k1 == k2 -> 265 265 - do_merge(false, rest1, rest2, acc_pair(p2, acc)) 266 266 - end 267 267 - end 268 268 - 269 269 - defp do_merge(true, [{k1, _} = p1 | rest1] = list1, [{k2, _} = p2 | rest2] = list2, acc) do 270 270 - cond do 271 271 - k1 > k2 -> 272 272 - do_merge(true, rest1, list2, acc_pair(p1, acc)) 273 273 - k1 < k2 -> 274 274 - do_merge(true, list1, rest2, acc_pair(p2, acc)) 275 275 - k1 == k2 -> 276 276 - do_merge(true, rest1, rest2, acc_pair(p2, acc)) 277 277 - end 278 278 - end 279 279 - 280 280 - defp acc_pair({_key, :deleted}, acc), do: acc 281 281 - defp acc_pair({_key, _value} = pair, acc), do: [pair | acc] 282 282 - 283 283 - @spec apply_batch(t, non_neg_integer, [Utils.mutation]) :: :ok 284 284 - def apply_batch(kv, version, mutations) when is_list(mutations) do 285 285 - MutationLog.insert(kv.mutation_log, version, mutations) 286 286 - 287 287 - Enum.reduce(mutations, kv, fn 288 288 - {:write, k, v}, kv -> 289 289 - put(kv, version, k, v) 290 290 - {:clear, k}, kv -> 291 291 - delete(kv, version, k) 292 292 - kv 293 293 - {:clear_range, sk, ek}, kv -> 294 294 - delete_range(kv, version, sk, ek) 295 295 - end) 296 296 - end 297 297 - 298 298 - @doc """ 299 299 - Flushes mutations with a version <= `version` to unversioned storage. 300 300 - """ 301 301 - @spec flush(%HybridKV{}, non_neg_integer) :: %HybridKV{} 302 302 - def flush(%HybridKV{storage_module: s_mod, storage_kv: s_kv} = kv, version) when is_integer(version) and version >= 0 do 303 303 - batches = MutationLog.pop_up_to(kv.mutation_log, version) 304 304 - 305 305 - mem_kv = kv.mem_kv 306 306 - Enum.each(batches, fn {ver, mutations} -> 307 307 - Enum.each(mutations, fn 308 308 - {:write, k, v} -> 309 309 - MemKV.remove_key_at_version(mem_kv, ver, k) 310 310 - s_mod.put(s_kv, k, v) 311 311 - 312 312 - {:clear, k} -> 313 313 - MemKV.remove_key_at_version(mem_kv, ver, k) 314 314 - s_mod.delete(s_kv, k) 315 315 - 316 316 - {:clear_range, sk, ek} -> 317 317 - MemKV.remove_range_at_version(mem_kv, ver, sk, ek) 318 318 - s_mod.delete_range(s_kv, sk, ek) 319 319 - end) 320 320 - end) 321 321 - 322 322 - {_ranges, deleted_forest} = RangeForest.flush(kv.deleted_forest, version, kv.flushed_version + 1) 323 323 - 324 324 - %{kv | deleted_forest: deleted_forest, flushed_version: version} 325 325 - end 326 326 - 327 327 - @spec put_storage(t, binary, binary) :: :ok 328 328 - def put_storage(%HybridKV{} = kv, key, value) when is_binary(key) and is_binary(value) do 329 329 - kv.storage_module.put(kv.storage_kv, key, value) 330 330 - end 331 331 - 332 332 - @spec commit(t) :: :ok 333 333 - def commit(%HybridKV{} = kv) do 334 334 - kv.storage_module.commit(kv.storage_kv) 335 335 - end 336 336 - 337 337 - @spec load_storage(%HybridKV{}, [{binary, binary}]) :: :ok 338 338 - def load_storage(%HybridKV{storage_module: s_mod, storage_kv: s_kv}, pairs) when is_list(pairs) do 339 339 - Enum.each(pairs, fn {k, v} when is_binary(k) and is_binary(v) -> 340 340 - s_mod.put(s_kv, k, v) 341 341 - end) 342 342 - end 343 343 - 344 344 - @spec delete_range_storage(%HybridKV{}, binary, binary) :: :ok 345 345 - def delete_range_storage(%HybridKV{storage_module: s_mod, storage_kv: s_kv}, start_key, end_key) do 346 346 - s_mod.delete_range(s_kv, start_key, end_key) 347 347 - end 348 348 - 349 349 - @doc """ 350 350 - Clears a range from both memory and storage **at all versions**. 351 351 - """ 352 352 - @spec nuke_range(%HybridKV{}, binary, binary) :: :ok 353 353 - def nuke_range(%HybridKV{} = kv, start_key, end_key) do 354 354 - MemKV.nuke_range(kv.mem_kv, start_key, end_key) 355 355 - delete_range_storage(kv, start_key, end_key) 356 356 - :ok 357 357 - end 358 358 - 359 359 - @doc false 360 360 - def dump(%HybridKV{} = kv) do 361 361 - %{ 362 362 - mem_kv: MemKV.dump(kv.mem_kv), 363 363 - storage_kv: kv.storage_module.dump(kv.storage_kv), 364 364 - deleted_forest: RangeForest.dump(kv.deleted_forest), 365 365 - mutation_log: MutationLog.dump(kv.mutation_log), 366 366 - } 367 367 - end 368 368 - end

-138

lib/kv/byte_sample.ex

reviewed

··· 1 1 - defmodule Hobbes.KV.ByteSample do 2 2 - alias Hobbes.Utils 3 3 - import Hobbes.Utils 4 4 - 5 5 - # Byte sample should be 1/250th the size of k/v data 6 6 - @byte_sample_factor 250 7 7 - # TODO: 250 is to rare for tests, buggify 8 8 - #@byte_sample_factor 2 9 9 - 10 10 - # Approximate overhead per sample (other than key size) 11 11 - # (currently just 8 bytes to store the size value as a string) 12 12 - @byte_sample_overhead_bytes 8 13 13 - 14 14 - @type t :: :ets.table 15 15 - 16 16 - @spec new :: t 17 17 - def new do 18 18 - :ets.new(__MODULE__, [:ordered_set, :private]) 19 19 - end 20 20 - 21 21 - @spec load(t, [{binary, binary}]) :: :ok 22 22 - def load(table, pairs) when is_list(pairs) do 23 23 - Enum.each(pairs, fn {k, v} when is_binary(k) and is_binary(v) -> 24 24 - bytes = decode_float(v) 25 25 - :ets.insert(table, {k, bytes}) 26 26 - end) 27 27 - :ok 28 28 - end 29 29 - 30 30 - @spec delete_range(t, binary, binary) :: :ok 31 31 - def delete_range(table, start_key, end_key) when is_binary(start_key) and is_binary(end_key) do 32 32 - :ets.delete(table, start_key) 33 33 - do_scan_delete(table, end_key, start_key) 34 34 - end 35 35 - 36 36 - defp do_scan_delete(table, end_key, prev_key) do 37 37 - case :ets.next(table, prev_key) do 38 38 - key when is_binary(key) and key < end_key -> 39 39 - :ets.delete(table, key) 40 40 - do_scan_delete(table, end_key, key) 41 41 - 42 42 - _ -> :ok 43 43 - end 44 44 - end 45 45 - 46 46 - @spec scan(t, binary, binary) :: [{binary, float}] 47 47 - def scan(table, start_key, end_key) do 48 48 - acc = 49 49 - case :ets.lookup(table, start_key) do 50 50 - [{^start_key, _size}] = result -> result 51 51 - [] -> [] 52 52 - end 53 53 - 54 54 - do_scan(table, end_key, start_key, acc) 55 55 - |> Enum.reverse() 56 56 - end 57 57 - 58 58 - defp do_scan(table, end_key, prev_key, acc) do 59 59 - case :ets.next_lookup(table, prev_key) do 60 60 - {_key, [{key, _size} = pair]} -> 61 61 - case key < end_key do 62 62 - true -> do_scan(table, end_key, key, [pair | acc]) 63 63 - false -> acc 64 64 - end 65 65 - 66 66 - :"$end_of_table" -> acc 67 67 - end 68 68 - end 69 69 - 70 70 - @spec apply_batch(t, [Utils.mutation]) :: [Utils.mutation] 71 71 - def apply_batch(table, mutations) when is_list(mutations) do 72 72 - mutations 73 73 - |> Enum.reduce([], fn 74 74 - {:write, k, v}, acc -> 75 75 - key_size = byte_size(k) 76 76 - pair_size = key_size + byte_size(v) 77 77 - probability = byte_sample_probability(key_size, pair_size) 78 78 - 79 79 - case (:erlang.phash2(k, 1000) / 1000) < probability do 80 80 - true -> 81 81 - # Correct for sampling probability (see comments in byte_sample_probability/2) 82 82 - sampled_size = pair_size / min(probability, 1) 83 83 - 84 84 - :ets.insert(table, {k, sampled_size}) 85 85 - mut = {:write, special_byte_sample_prefix() <> k, encode_float(sampled_size)} 86 86 - [mut | acc] 87 87 - 88 88 - false -> acc 89 89 - end 90 90 - 91 91 - {:clear, k}, acc -> 92 92 - case :ets.member(table, k) do 93 93 - true -> 94 94 - :ets.delete(table, k) 95 95 - mut = {:clear, special_byte_sample_prefix() <> k} 96 96 - [mut | acc] 97 97 - 98 98 - false -> acc 99 99 - end 100 100 - 101 101 - {:clear_range, sk, ek}, acc -> 102 102 - delete_range(table, sk, ek) 103 103 - # TODO: if the range was empty in the byte sample we don't need this mutation 104 104 - mut = {:clear_range, special_byte_sample_prefix() <> sk, special_byte_sample_prefix() <> ek} 105 105 - [mut | acc] 106 106 - end) 107 107 - |> Enum.reverse() 108 108 - end 109 109 - 110 110 - defp byte_sample_probability(key_size, pair_size) do 111 111 - # Probability that a key/value pair of this size belongs in the byte sample 112 112 - # This is a function of the size rather than a percentage of keys so that 113 113 - # we can maintain the byte sample as a fraction of *total KV size* 114 114 - # 115 115 - # Intuitively: the byte sample only stores keys, so if values are larger than 116 116 - # the overhead we can afford to store more samples while staying under the limit 117 117 - # Therefore, if the value is large, probability should increase 118 118 - # 119 119 - # We then correct out the probability factor by dividing size to get sampled_size 120 120 - # at the end, so that the sample is not biased by the larger pairs 121 121 - # 122 122 - # This algorithm is borrowed directly from FDB (storageserver isKeyValueInSample) 123 123 - (pair_size / (key_size + @byte_sample_overhead_bytes)) / @byte_sample_factor 124 124 - end 125 125 - 126 126 - defp encode_float(float) when is_number(float) do 127 127 - Integer.to_string(round(float * 1000)) 128 128 - end 129 129 - 130 130 - defp decode_float(string) when is_binary(string) do 131 131 - String.to_integer(string) / 1000 132 132 - end 133 133 - 134 134 - @doc false 135 135 - def dump(table) do 136 136 - :ets.tab2list(table) 137 137 - end 138 138 - end

-58

lib/kv/flat_storage_kv.ex

reviewed

··· 1 1 - defmodule Hobbes.KV.FlatStorageKV do 2 2 - alias Hobbes.Construct.SimFile 3 3 - 4 4 - alias Hobbes.KV.{FlatKV, FlatStorageKV} 5 5 - 6 6 - @behaviour Hobbes.KV.Storage 7 7 - 8 8 - @type t :: %__MODULE__{ 9 9 - path: String.t, 10 10 - kv: FlatKV.t, 11 11 - } 12 12 - @enforce_keys [:path, :kv] 13 13 - defstruct @enforce_keys 14 14 - 15 15 - @spec new(String.t) :: t 16 16 - def new(path) do 17 17 - kv = %FlatStorageKV{path: path, kv: FlatKV.new()} 18 18 - 19 19 - #case SimFile.read(path) do 20 20 - # {:ok, contents} -> 21 21 - # pairs = decode(contents) 22 22 - # FlatKV.load(kv.kv, pairs) 23 23 - 24 24 - # {:error, :enoent} -> :noop 25 25 - #end 26 26 - 27 27 - kv 28 28 - end 29 29 - 30 30 - def commit(%FlatStorageKV{kv: kv, path: _path}) do 31 31 - _contents = FlatKV.dump(kv) |> encode() 32 32 - #:ok = SimFile.write(path, contents) 33 33 - :ok 34 34 - end 35 35 - 36 36 - def put(%{kv: kv}, key, value), do: FlatKV.put(kv, key, value) 37 37 - def delete(%{kv: kv}, key), do: FlatKV.delete(kv, key) 38 38 - def delete_range(%{kv: kv}, start_key, end_key), do: FlatKV.delete_range(kv, start_key, end_key) 39 39 - 40 40 - def get(%{kv: kv}, key), do: FlatKV.get(kv, key) 41 41 - def scan(%{kv: kv}, start_key, end_key, opts \\ []), do: FlatKV.scan(kv, start_key, end_key, opts) 42 42 - 43 43 - defp encode(pairs) when is_list(pairs), do: :erlang.term_to_binary(pairs, [:deterministic]) 44 44 - defp decode(contents) when is_binary(contents), do: :erlang.binary_to_term(contents, [:safe]) 45 45 - 46 46 - @doc false 47 47 - def dump(%{kv: kv}) do 48 48 - FlatKV.dump(kv) 49 49 - end 50 50 - 51 51 - @doc false 52 52 - def dump_file(%{path: path}) do 53 53 - case SimFile.read(path) do 54 54 - {:ok, contents} -> decode(contents) 55 55 - {:error, _} -> nil 56 56 - end 57 57 - end 58 58 - end

-53

lib/kv/mutation_log.ex

reviewed

··· 1 1 - defmodule Hobbes.KV.MutationLog do 2 2 - alias Hobbes.Utils 3 3 - 4 4 - @type t :: :ets.table 5 5 - 6 6 - @spec new :: t 7 7 - def new do 8 8 - :ets.new(__MODULE__, [:ordered_set, :private]) 9 9 - end 10 10 - 11 11 - @spec insert(t, non_neg_integer, [Utils.mutation]) :: :ok 12 12 - def insert(table, version, mutations) when is_integer(version) and is_list(mutations) do 13 13 - :ets.insert(table, {version, mutations}) 14 14 - :ok 15 15 - end 16 16 - 17 17 - @spec append(t, non_neg_integer, [Utils.mutation]) :: :ok 18 18 - def append(table, version, mutations) when is_integer(version) and is_list(mutations) do 19 19 - case :ets.lookup(table, version) do 20 20 - [{^version, existing}] -> :ets.insert(table, {version, existing ++ mutations}) 21 21 - [] -> :ets.insert(table, {version, mutations}) 22 22 - end 23 23 - :ok 24 24 - end 25 25 - 26 26 - @spec pop_up_to(t, non_neg_integer) :: [{non_neg_integer, [Utils.mutation]}] 27 27 - def pop_up_to(table, end_version) when is_integer(end_version) do 28 28 - scan_pop(table, end_version, -1, []) 29 29 - |> Enum.reverse() 30 30 - end 31 31 - 32 32 - defp scan_pop(table, end_version, prev_version, acc) do 33 33 - case :ets.next(table, prev_version) do 34 34 - ver when is_integer(ver) -> 35 35 - case ver <= end_version do 36 36 - true -> 37 37 - [{^ver, mutations}] = :ets.lookup(table, ver) 38 38 - :ets.delete(table, ver) 39 39 - 40 40 - scan_pop(table, end_version, ver, [{ver, mutations} | acc]) 41 41 - 42 42 - false -> acc 43 43 - end 44 44 - 45 45 - :"$end_of_table" -> acc 46 46 - end 47 47 - end 48 48 - 49 49 - @doc false 50 50 - def dump(table) do 51 51 - :ets.tab2list(table) 52 52 - end 53 53 - end

-106

lib/storage_queue.ex

reviewed

··· 1 1 - defmodule Hobbes.StorageQueue do 2 2 - alias Hobbes.{StorageQueue, Utils} 3 3 - alias Hobbes.KV.{FlatKV, FlatStorageKV} 4 4 - 5 5 - import Hobbes.Utils 6 6 - 7 7 - @type t :: %__MODULE__{ 8 8 - storage_module: module, 9 9 - storage_kv: FlatKV.t | FlatStorageKV.t, 10 10 - } 11 11 - 12 12 - @enforce_keys [:storage_module, :storage_kv] 13 13 - defstruct @enforce_keys 14 14 - 15 15 - @spec new(module, term) :: t 16 16 - def new(storage_module, storage_kv) when is_atom(storage_module) do 17 17 - %StorageQueue{ 18 18 - storage_module: storage_module, 19 19 - storage_kv: storage_kv, 20 20 - } 21 21 - end 22 22 - 23 23 - @locked_key special_prefix() <> "/locked?" 24 24 - @version_key special_prefix() <> "/version" 25 25 - @kcv_key special_prefix() <> "/known_committed_version" 26 26 - 27 27 - @spec put_state(t, map) :: :ok 28 28 - def put_state(%StorageQueue{storage_module: s_mod, storage_kv: s_kv} = _sq, fields) when is_map(fields) do 29 29 - %{ 30 30 - locked?: locked?, 31 31 - version: version, 32 32 - known_committed_version: known_committed_version, 33 33 - } = fields 34 34 - 35 35 - s_mod.put(s_kv, @locked_key, encode_boolean(locked?)) 36 36 - s_mod.put(s_kv, @version_key, Integer.to_string(version)) 37 37 - s_mod.put(s_kv, @kcv_key, Integer.to_string(known_committed_version)) 38 38 - :ok 39 39 - end 40 40 - 41 41 - def get_state(%StorageQueue{storage_module: s_mod, storage_kv: s_kv} = _sq) do 42 42 - %{ 43 43 - locked?: s_mod.get(s_kv, @locked_key) |> decode_boolean(), 44 44 - version: s_mod.get(s_kv, @version_key) |> String.to_integer(), 45 45 - known_committed_version: s_mod.get(s_kv, @kcv_key) |> String.to_integer(), 46 46 - } 47 47 - end 48 48 - 49 49 - @spec peek_batches(t) :: [{non_neg_integer, [Utils.tagged_mutation]}] 50 50 - def peek_batches(%StorageQueue{} = sq) do 51 51 - start_key = encode_version(0) 52 52 - # TODO: fix after switching to binary encoding 53 53 - end_key = String.duplicate("9", 20) 54 54 - 55 55 - sq.storage_module.scan(sq.storage_kv, start_key, end_key) 56 56 - |> Map.fetch!(:pairs) 57 57 - |> Enum.map(fn {k, v} -> 58 58 - {decode_version(k), decode_mutations(v)} 59 59 - end) 60 60 - end 61 61 - 62 62 - @spec append_batch(t, non_neg_integer, [Utils.tagged_mutation]) :: :ok 63 63 - def append_batch(%StorageQueue{} = sq, version, tagged_mutations) when is_integer(version) and is_list(tagged_mutations) do 64 64 - sq.storage_module.put(sq.storage_kv, encode_version(version), encode_mutations(tagged_mutations)) 65 65 - end 66 66 - 67 67 - @spec pop_batches(t, non_neg_integer) :: :ok 68 68 - def pop_batches(%StorageQueue{} = sq, up_to_version) when is_integer(up_to_version) do 69 69 - start_key = encode_version(0) 70 70 - # Note: we need next_key/1 because up_to_version is inclusive 71 71 - end_key = encode_version(up_to_version) |> next_key() 72 72 - 73 73 - sq.storage_module.delete_range(sq.storage_kv, start_key, end_key) 74 74 - :ok 75 75 - end 76 76 - 77 77 - @spec commit(t) :: :ok 78 78 - def commit(%StorageQueue{storage_module: s_mod, storage_kv: kv}) do 79 79 - s_mod.commit(kv) 80 80 - end 81 81 - 82 82 - # TODO: binary encoding 83 83 - defp encode_version(version) when is_integer(version), do: Integer.to_string(version) |> String.pad_leading(20, "0") 84 84 - defp decode_version(string), do: String.to_integer(string) 85 85 - 86 86 - defp encode_mutations(mutations) when is_list(mutations) do 87 87 - :erlang.term_to_binary(mutations, [:deterministic]) 88 88 - end 89 89 - 90 90 - defp decode_mutations(binary) when is_binary(binary) do 91 91 - :erlang.binary_to_term(binary, [:safe]) 92 92 - |> case do 93 93 - mutations when is_list(mutations) -> mutations 94 94 - end 95 95 - end 96 96 - 97 97 - defp encode_boolean(boolean) when is_boolean(boolean), do: Atom.to_string(boolean) 98 98 - 99 99 - defp decode_boolean("true"), do: true 100 100 - defp decode_boolean("false"), do: false 101 101 - 102 102 - @doc false 103 103 - def dump(%StorageQueue{} = sq) do 104 104 - sq.storage_module.dump(sq.storage_kv) 105 105 - end 106 106 - end

-99

test/construct/scheduler/file_store_test.exs

reviewed

··· 1 1 - defmodule Hobbes.Construct.Scheduler.FileStoreTest do 2 2 - use ExUnit.Case, async: true 3 3 - 4 4 - alias Hobbes.Construct.Scheduler.FileStore 5 5 - 6 6 - @moduletag :file_store 7 7 - 8 8 - setup do 9 9 - %{fs: FileStore.new()} 10 10 - end 11 11 - 12 12 - describe "mkdir/2" do 13 13 - test "returns error if file in path", %{fs: fs} do 14 14 - :ok = FileStore.mkdir(fs, "/foo") 15 15 - :ok = FileStore.write(fs, "/foo/bar", "hello") 16 16 - 17 17 - assert {:error, :eexist} = FileStore.mkdir(fs, "/foo/bar") 18 18 - assert {:error, :enoent} = FileStore.mkdir(fs, "/foo/bar/baz") 19 19 - end 20 20 - 21 21 - test "returns error if directory exists", %{fs: fs} do 22 22 - :ok = FileStore.mkdir(fs, "/foo") 23 23 - assert {:error, :eexist} = FileStore.mkdir(fs, "/foo") 24 24 - end 25 25 - end 26 26 - 27 27 - describe "mkdir_p/2" do 28 28 - test "returns error if file in path", %{fs: fs} do 29 29 - assert :ok = FileStore.mkdir(fs, "/foo") 30 30 - assert :ok = FileStore.write(fs, "/foo/file", "hello world") 31 31 - 32 32 - assert {:error, :enotdir} = FileStore.mkdir_p(fs, "/foo/file/bar") 33 33 - end 34 34 - 35 35 - test "makes directories", %{fs: fs} do 36 36 - assert :ok = FileStore.mkdir_p(fs, "/foo/bar/baz") 37 37 - assert :ok = FileStore.mkdir_p(fs, "/foo/bar/baz/qux") 38 38 - 39 39 - assert FileStore.dir?(fs, "/") 40 40 - assert FileStore.dir?(fs, "/foo") 41 41 - assert FileStore.dir?(fs, "/foo/bar") 42 42 - assert FileStore.dir?(fs, "/foo/bar/baz") 43 43 - assert FileStore.dir?(fs, "/foo/bar/baz/qux") 44 44 - assert FileStore.dir?(fs, "/foo/bar/baz/qux/") 45 45 - end 46 46 - end 47 47 - 48 48 - describe "write/3" do 49 49 - test "returns error if no directory", %{fs: fs} do 50 50 - assert {:error, :enoent} = FileStore.write(fs, "/foo/bar.txt", "hello") 51 51 - end 52 52 - 53 53 - test "returns error if file in path", %{fs: fs} do 54 54 - :ok = FileStore.write(fs, "/foo", "hello") 55 55 - assert {:error, :enoent} = FileStore.write(fs, "/foo/bar.txt", "hello") 56 56 - end 57 57 - 58 58 - test "writes file", %{fs: fs} do 59 59 - :ok = FileStore.mkdir_p(fs, "/foo/bar") 60 60 - 61 61 - assert :ok = FileStore.write(fs, "/foo/bar/baz.txt", "hello world!") 62 62 - 63 63 - assert {:ok, "hello world!"} = FileStore.read(fs, "/foo/bar/baz.txt") 64 64 - end 65 65 - end 66 66 - 67 67 - describe "read/2" do 68 68 - test "returns error if file does not exist", %{fs: fs} do 69 69 - assert {:error, :enoent} = FileStore.read(fs, "/foo/bar.txt") 70 70 - end 71 71 - 72 72 - test "returns error if file is a directory", %{fs: fs} do 73 73 - :ok = FileStore.mkdir_p(fs, "/foo/bar") 74 74 - assert {:error, :eisdir} = FileStore.read(fs, "/foo/bar") 75 75 - end 76 76 - 77 77 - test "reads file", %{fs: fs} do 78 78 - :ok = FileStore.mkdir(fs, "/foo") 79 79 - :ok = FileStore.write(fs, "/foo/bar.txt", "hello world!") 80 80 - 81 81 - assert {:ok, "hello world!"} = FileStore.read(fs, "/foo/bar.txt") 82 82 - end 83 83 - end 84 84 - 85 85 - describe "ls/2" do 86 86 - test "returns paths", %{fs: fs} do 87 87 - FileStore.mkdir_p(fs, "/foo/bar/a.txt") 88 88 - FileStore.mkdir_p(fs, "/foo/bar/b.txt") 89 89 - FileStore.mkdir_p(fs, "/foo/bar/c.txt") 90 90 - FileStore.mkdir_p(fs, "/foo/bar/baz/d.txt") 91 91 - 92 92 - assert {:ok, ["a.txt", "b.txt", "baz", "c.txt"]} = FileStore.ls(fs, "/foo/bar/") 93 93 - assert {:ok, ["a.txt", "b.txt", "baz", "c.txt"]} = FileStore.ls(fs, "/foo/bar") 94 94 - 95 95 - assert {:ok, ["bar"]} = FileStore.ls(fs, "/foo") 96 96 - assert {:ok, ["d.txt"]} = FileStore.ls(fs, "/foo/bar/baz") 97 97 - end 98 98 - end 99 99 - end

-68

test/construct_test.exs

reviewed

··· 1 1 - defmodule Hobbes.ConstructTest do 2 2 - use ExUnit.Case, async: false 3 3 - 4 4 - alias Hobbes.Construct.{SimServer, SimFile} 5 5 - 6 6 - @moduletag :construct 7 7 - 8 8 - describe "process exits" do 9 9 - test "handles child process exiting normally" do 10 10 - SimServer.start_scheduler() 11 11 - 12 12 - SimServer.spawn(fn -> 13 13 - 1 + 2 14 14 - end) 15 15 - 16 16 - assert true 17 17 - end 18 18 - 19 19 - # These tests use SimServer.sleep(0) to hand off to each other, 20 20 - # which ensures the monitors/links connect *before* the error is raised 21 21 - # 22 22 - # Note: the :timer.sleep(1) at the end is needed to prevent error logs from being 23 23 - # printed even though :capture_log is true (test process otherwise seems to die too early) 24 24 - @tag :capture_log 25 25 - test "handles unlinked child process raising error" do 26 26 - SimServer.start_scheduler() 27 27 - 28 28 - pid = SimServer.spawn(fn -> 29 29 - SimServer.sleep(0) 30 30 - raise "Some error" 31 31 - end) 32 32 - 33 33 - mref = Process.monitor(pid) 34 34 - SimServer.sleep(0) 35 35 - 36 36 - assert_receive {:DOWN, ^mref, :process, ^pid, {%RuntimeError{message: "Some error"}, _}} 37 37 - :timer.sleep(1) 38 38 - end 39 39 - 40 40 - @tag :capture_log 41 41 - test "handles linked process raising error" do 42 42 - SimServer.start_scheduler() 43 43 - 44 44 - SimServer.flag(:trap_exit, true) 45 45 - 46 46 - pid = SimServer.spawn_link(fn -> 47 47 - SimServer.sleep(0) 48 48 - raise "Some error" 49 49 - end) 50 50 - SimServer.sleep(0) 51 51 - 52 52 - assert_receive {:EXIT, ^pid, {%RuntimeError{message: "Some error"}, _}} 53 53 - Process.sleep(1) 54 54 - end 55 55 - end 56 56 - 57 57 - describe "SimFile" do 58 58 - test "reads and writes files" do 59 59 - SimServer.start_scheduler() 60 60 - 61 61 - assert :ok = SimFile.mkdir_p("/foo/bar/baz") 62 62 - assert :ok = SimFile.write("/foo/bar/baz/qux.txt", "hello world!") 63 63 - 64 64 - assert {:ok, "hello world!"} = SimFile.read("/foo/bar/baz/qux.txt") 65 65 - assert {:error, :enoent} = SimFile.read("/foo/qux.txt") 66 66 - end 67 67 - end 68 68 - end

-197

test/hybrid_kv_test.exs

reviewed

··· 1 1 - defmodule Hobbes.HybridKVTest do 2 2 - use ExUnit.Case, async: true 3 3 - 4 4 - alias Hobbes.HybridKV 5 5 - alias Hobbes.KV.TestKV 6 6 - 7 7 - @moduletag :hybrid_kv 8 8 - 9 9 - setup do 10 10 - %{kv: HybridKV.new()} 11 11 - end 12 12 - 13 13 - defmodule Verifier do 14 14 - @moduledoc """ 15 15 - This module is used to verify the correctness of HybridKV by checking 16 16 - all of its operations against a much simpler implementation (TestKV). 17 17 - 18 18 - We generate random (deterministic by seed) operations (get, put, etc) 19 19 - and run them through both KVs, and then check that the results match. 20 20 - """ 21 21 - 22 22 - alias Verifier 23 23 - 24 24 - @enforce_keys [:hybrid_kv, :test_kv, :version, :durable_version, :seed] 25 25 - defstruct @enforce_keys 26 26 - 27 27 - def new(seed) do 28 28 - :rand.seed(:exsss, seed) 29 29 - 30 30 - %Verifier{ 31 31 - hybrid_kv: HybridKV.new(), 32 32 - test_kv: TestKV.new(), 33 33 - version: 1, 34 34 - durable_version: 0, 35 35 - seed: seed, 36 36 - } 37 37 - end 38 38 - 39 39 - def run(%Verifier{} = verifier, op_count) do 40 40 - Enum.reduce(1..op_count, verifier, fn i, verifier -> 41 41 - #if i == 0, do: dump(verifier) 42 42 - 43 43 - try do 44 44 - perform(random_op(), verifier) 45 45 - rescue 46 46 - e in [ExUnit.AssertionError] -> 47 47 - e = Map.update!(e, :message, &(&1 <> " (at op #{i}, seed=#{inspect(verifier.seed)})")) 48 48 - reraise e, __STACKTRACE__ 49 49 - e -> 50 50 - require Logger 51 51 - Logger.error("Error #{inspect(e)} at op=#{i}, seed=#{inspect(verifier.seed)}") 52 52 - reraise e, __STACKTRACE__ 53 53 - end 54 54 - end) 55 55 - end 56 56 - 57 57 - @doc false 58 58 - def dump(%Verifier{} = verifier) do 59 59 - dbg [ 60 60 - hybrid_kv: HybridKV.dump(verifier.hybrid_kv), 61 61 - test_kv: TestKV.dump(verifier.test_kv), 62 62 - durable_version: verifier.durable_version, 63 63 - ], limit: :infinity 64 64 - end 65 65 - 66 66 - @ops [:apply_batch, :get, :scan] 67 67 - defp random_op do 68 68 - case Enum.random(1..100) do 69 69 - 1 -> :flush 70 70 - _ -> Enum.random(@ops) 71 71 - end 72 72 - end 73 73 - 74 74 - defp perform(:get, %Verifier{hybrid_kv: hkv, test_kv: tkv} = verifier) do 75 75 - read_version = rand_read_version(verifier) 76 76 - 77 77 - key = rand_hash() 78 78 - 79 79 - hkv_result = HybridKV.get(hkv, read_version, key) 80 80 - tkv_result = TestKV.get(tkv, read_version, key) 81 81 - 82 82 - assert hkv_result == tkv_result 83 83 - 84 84 - verifier 85 85 - end 86 86 - 87 87 - defp perform(:scan, %Verifier{hybrid_kv: hkv, test_kv: tkv} = verifier) do 88 88 - read_version = rand_read_version(verifier) 89 89 - 90 90 - start_key = rand_range_key() 91 91 - end_key = rand_range_key() 92 92 - {start_key, end_key, reverse} = 93 93 - cond do 94 94 - start_key < end_key -> {start_key, end_key, false} 95 95 - start_key > end_key -> {end_key, start_key, true} 96 96 - start_key == end_key -> {start_key, start_key <> "\x00", false} 97 97 - end 98 98 - 99 99 - limit = rand_limit() 100 100 - 101 101 - hkv_result = HybridKV.scan(hkv, read_version, start_key, end_key, limit: limit, reverse: reverse) 102 102 - tkv_result = TestKV.scan(tkv, read_version, start_key, end_key, limit: limit, reverse: reverse) 103 103 - 104 104 - assert hkv_result == tkv_result 105 105 - verifier 106 106 - end 107 107 - 108 108 - defp perform(:apply_batch, %Verifier{hybrid_kv: hkv, test_kv: tkv} = verifier) do 109 109 - verifier = %{verifier | version: verifier.version + Enum.random(1..1000)} 110 110 - 111 111 - mutation_count = Enum.random(1..10) 112 112 - mutations = Enum.reduce(1..mutation_count, [], fn _i, acc -> 113 113 - [rand_mutation() | acc] 114 114 - end) 115 115 - 116 116 - version = verifier.version 117 117 - hkv = HybridKV.apply_batch(hkv, version, mutations) 118 118 - tkv = TestKV.apply_batch(tkv, version, mutations) 119 119 - 120 120 - %{verifier | hybrid_kv: hkv, test_kv: tkv} 121 121 - end 122 122 - 123 123 - defp perform(:flush, %Verifier{hybrid_kv: hkv} = verifier) do 124 124 - flush_version = rand(verifier.durable_version, verifier.version) 125 125 - hkv = HybridKV.flush(hkv, flush_version) 126 126 - 127 127 - %Verifier{verifier | hybrid_kv: hkv, durable_version: flush_version} 128 128 - end 129 129 - 130 130 - defp rand_mutation do 131 131 - case Enum.random(1..10) do 132 132 - 1 -> 133 133 - start_key = rand_hash() 134 134 - end_key = rand_hash() 135 135 - {start_key, end_key} = 136 136 - cond do 137 137 - start_key < end_key -> {start_key, end_key} 138 138 - start_key > end_key -> {end_key, start_key} 139 139 - start_key == end_key -> {start_key, start_key <> "\x00"} 140 140 - end 141 141 - {:clear_range, start_key, end_key} 142 142 - 143 143 - i when i in [2, 3] -> 144 144 - {:clear, rand_hash()} 145 145 - 146 146 - _ -> 147 147 - {:write, rand_hash(), rand_hash()} 148 148 - end 149 149 - end 150 150 - 151 151 - defp rand(s \\ 0, e), do: Enum.random(s..e) 152 152 - 153 153 - defp rand_read_version(%Verifier{} = verifier) do 154 154 - rand(verifier.durable_version, verifier.version) 155 155 - end 156 156 - 157 157 - defp rand_limit do 158 158 - case Enum.random(1..3) do 159 159 - 1 -> Enum.random(2..9) 160 160 - 2 -> Enum.random([1, 10, 100, 1000]) 161 161 - 3 -> :infinity 162 162 - end 163 163 - end 164 164 - 165 165 - defp rand_range_key do 166 166 - case Enum.random(1..10) do 167 167 - 1 -> "" 168 168 - 2 -> "\xFF" 169 169 - _ -> rand_hash() 170 170 - end 171 171 - end 172 172 - 173 173 - defp rand_hash(bound \\ 99) do 174 174 - :crypto.hash(:sha256, Integer.to_string(rand(bound))) 175 175 - |> Base.encode16(padding: false) 176 176 - |> String.slice(0, 8) 177 177 - end 178 178 - end 179 179 - 180 180 - describe "verify HybridKV" do 181 181 - @tag :hkv_verify 182 182 - test "operations" do 183 183 - Verifier.new({100, 101, 102}) 184 184 - |> Verifier.run(1000) 185 185 - end 186 186 - 187 187 - @tag :hkv_verify_multi 188 188 - @tag :disable 189 189 - test "operations multi (slow)" do 190 190 - # 300 verification tests (takes about 1.3s) 191 191 - Enum.map(1..300, fn s -> 192 192 - Verifier.new({100 + s, 101, 102}) 193 193 - |> Verifier.run(1000) 194 194 - end) 195 195 - end 196 196 - end 197 197 - end

-22

test/kv/byte_sample_test.exs

reviewed

··· 1 1 - defmodule Hobbes.KV.ByteSampleTest do 2 2 - use ExUnit.Case, async: true 3 3 - 4 4 - alias Hobbes.KV.ByteSample 5 5 - 6 6 - @moduletag :byte_sample 7 7 - 8 8 - setup do 9 9 - %{bs: ByteSample.new()} 10 10 - end 11 11 - 12 12 - describe "ByteSample" do 13 13 - test "samples", %{bs: bs} do 14 14 - mutations = Enum.map(1..4000, fn i -> {:write, "k#{String.pad_leading(to_string(i), 4, "0")}", "v#{i}"} end) 15 15 - ByteSample.apply_batch(bs, mutations) 16 16 - 17 17 - pairs = ByteSample.scan(bs, "k1000", "k2000") 18 18 - # Anything else will flake if we change the parameters 19 19 - assert is_list(pairs) 20 20 - end 21 21 - end 22 22 - end

-48

test/kv/mutation_log_test.exs

reviewed

··· 1 1 - defmodule Hobbes.KV.MutationLogTest do 2 2 - use ExUnit.Case, async: true 3 3 - 4 4 - alias Hobbes.KV.MutationLog 5 5 - 6 6 - @moduletag :mutation_log 7 7 - 8 8 - setup do 9 9 - %{ml: MutationLog.new()} 10 10 - end 11 11 - 12 12 - describe "mutation log" do 13 13 - test "logs mutations", %{ml: ml} do 14 14 - MutationLog.insert(ml, 0, [{:write, "k0", "v0"}]) 15 15 - MutationLog.insert(ml, 1, [{:write, "k1", "v1"}]) 16 16 - MutationLog.insert(ml, 3, [{:write, "k3", "v3"}]) 17 17 - MutationLog.insert(ml, 4, [{:write, "k4", "v4"}]) 18 18 - 19 19 - assert MutationLog.pop_up_to(ml, 1) == [ 20 20 - {0, [{:write, "k0", "v0"}]}, 21 21 - {1, [{:write, "k1", "v1"}],}, 22 22 - ] 23 23 - assert MutationLog.pop_up_to(ml, 1) == [] 24 24 - assert MutationLog.pop_up_to(ml, 2) == [] 25 25 - 26 26 - assert MutationLog.pop_up_to(ml, 4) == [ 27 27 - {3, [{:write, "k3", "v3"}]}, 28 28 - {4, [{:write, "k4", "v4"}],}, 29 29 - ] 30 30 - 31 31 - assert MutationLog.dump(ml) == [] 32 32 - end 33 33 - 34 34 - test "appends", %{ml: ml} do 35 35 - MutationLog.append(ml, 0, [{:write, "k", "1"}]) 36 36 - MutationLog.append(ml, 0, [{:write, "k", "2"}]) 37 37 - MutationLog.append(ml, 0, [{:write, "k", "3"}]) 38 38 - 39 39 - MutationLog.insert(ml, 1, [{:write, "k", "1"}]) 40 40 - MutationLog.append(ml, 1, [{:write, "k", "2"}]) 41 41 - 42 42 - assert MutationLog.dump(ml) == [ 43 43 - {0, [{:write, "k", "1"}, {:write, "k", "2"}, {:write, "k", "3"}]}, 44 44 - {1, [{:write, "k", "1"}, {:write, "k", "2"}]}, 45 45 - ] 46 46 - end 47 47 - end 48 48 - end