defmodule Memcache.StateMachine do import Memcache.Command def single_command(command) do machine(:single_command, [command], fn :single_command, [result] -> {result, []} end) end # TODO: thundering herd mitigation def read_through(key, opts \\ [], func) do ttl = Keyword.get(opts, :ttl, 0) machine(:get, [get(key)], fn :get, [{:ok, value}] -> # Cache hit. {{:ok, value}, []} :get, [error: :not_found] -> # Cache miss. value = func.() {{:set, value}, [set(key, value, ttl)]} :get, [error: _error] -> # Some problem trying to get cached value. Memcache unavailable? Don't bother trying to set. value = func.() {{:ok, value}, []} {:set, value}, [:ok] -> {{:ok, value}, []} {:set, value}, [error: _error] -> # Failed to set value after a cache miss. Don't bother trying again. Just return the value we have. {{:ok, value}, []} end) end def read_through_term(key, opts \\ [], func) do ttl = Keyword.get(opts, :ttl, 0) machine(:get, [get(key)], fn :get, [{:ok, value}] -> # Cache hit. {{:ok, :erlang.binary_to_term(value)}, []} :get, [error: :not_found] -> # Cache miss. value = func.() {{:set, value}, [set(key, :erlang.term_to_binary(value), ttl)]} :get, [error: _error] -> # Some problem trying to get cached value. Memcache unavailable? Don't bother trying to set. value = func.() {{:ok, value}, []} {:set, value}, [:ok] -> {{:ok, value}, []} {:set, value}, [error: _error] -> # Failed to set value after a cache miss. Don't bother trying again. Just return the value we have. {{:ok, value}, []} end) end def read_modify_write(key, opts \\ [], func) do ttl = Keyword.get(opts, :ttl, 0) machine(:gets, [gets(key)], fn :gets, [{:ok, old_value, cas_unique}] -> new_value = func.({:ok, old_value}) {{:cas, new_value}, [cas(key, new_value, cas_unique, ttl)]} :gets, [error: :not_found] -> new_value = func.({:error, :not_found}) {{:add, new_value}, [add(key, new_value, ttl)]} {:cas, new_value}, [:ok] -> {{:ok, new_value}, []} {:cas, _new_value}, [error: :exists] -> # Race condition, the key changed after we read it. Try again. {:gets, [gets(key)]} {:cas, _new_value}, [error: :not_found] -> # Race condition, the key expired after we read it. Try again. new_value = func.({:error, :not_found}) {:add, [add(key, new_value, ttl)]} {:add, new_value}, [:ok] -> {{:ok, new_value}, []} {:add, _new_value}, [error: :not_stored] -> # Race condition, the key was created after we read it. Try again. {:gets, [gets(key)]} _, {:error, error} -> {{:error, error}, []} end) end def with_lock(key, opts \\ [], func) do ttl = Keyword.get(opts, :ttl, 0) machine(:add, [add(key, "locked", ttl)], fn :add, [:ok] -> value = func.() {{:delete, value}, [delete(key)]} :add, [error: :not_stored] -> # TODO: retry {{:error, :busy}, []} {:delete, value}, [:ok] -> {{:ok, value}, []} {:delete, value}, [error: _error] -> {{:ok, value}, []} end) end def machine(initial_state, commands, func) when is_list(commands) and is_function(func, 2) do {initial_state, commands, func} end def run(machines) do run(machines, Memcache.Router) end def run(machine, connection) when is_tuple(machine) do [result] = run([machine], connection) result end def run(machines, connection) when is_list(machines) do machines |> Enum.map(fn {_state, commands, _func} -> commands end) |> flattened(fn [_ | _] = flattened_commands -> Memcache.execute(flattened_commands, connection) [] -> throw(:all_machines_stopped) end) |> Enum.zip_with(machines, fn [_ | _] = result, {state, [_ | _], func} = _running_machine -> {new_state, new_batch} = func.(state, result) {new_state, new_batch, func} [], {_, [], _} = stopped_machine -> stopped_machine end) |> run(connection) catch :all_machines_stopped -> # All machines are stopped. Enum.map(machines, fn {state, [], _func} -> state end) end def flattened(lists, func) do unflatten(func.(flatten(lists)), lists) end def flatten(list) do flatten(list, []) end defp flatten([], bb) do bb end defp flatten([a | aa], bb) do flatten(a, flatten(aa, bb)) end defp flatten(a, bb) do [a | bb] end def unflatten(aa, bb) do {cc, []} = unflatten2(aa, bb) cc end defp unflatten2(aa, []) do {[], aa} end defp unflatten2(aa, [b | bb]) when is_list(b) do {c, aa} = unflatten2(aa, b) {cc, aa} = unflatten2(aa, bb) {[c | cc], aa} end defp unflatten2([a | aa], [b | bb]) when not is_list(b) do {cc, aa} = unflatten2(aa, bb) {[a | cc], aa} end end