elixir官方教程Mix与OTP(五) ETS

#ETS

1. ETS作缓存
2. 竞态条件

每当咱们须要查找一个桶,咱们就要发送一个信息给注册表.这时咱们的注册表会被多个进程并发访问,就会遇到瓶颈.

本章咱们会学习ETS(Erlang条件存储)以及如何使用它做为缓存机制.

警告!不要贸然地使用ETS作缓存!查看日志并分析你的应用表现,肯定那一部分是瓶颈,这样你就能够知道是否应该使用缓存,以及缓存什么.本章仅仅是一个如何使用ETS作缓存的例子.

#ETS作缓存

ETS容许咱们存储任何Elixir条件到一个内存中的表格.操做ETS表格须要经过Erlang的:ets模块:

iex> table = :ets.new(:buckets_registry, [:set, :protected])
8207
iex> :ets.insert(table, {"foo", self})
true
iex> :ets.lookup(table, "foo")
[{"foo", #PID<0.41.0>}]

建立一个ETS表格时,须要两个参数:表格名以及一些选项.经过这些选项,咱们传送了表格类型和它的访问规则.咱们已经选择了:set类型,它意味着键不能够被复制.咱们也将表格访问设置成了:protected,意味着只有建立了这个表格的进程能够写入它,但全部进程均可以从表中读取.这些都是默认值,因此咱们将在以后跳过它们.

ETS表格能够被命名,容许咱们经过名称来访问:

iex> :ets.new(:buckets_registry, [:named_table])
:buckets_registry
iex> :ets.insert(:buckets_registry, {"foo", self})
true
iex> :ets.lookup(:buckets_registry, "foo")
[{"foo", #PID<0.41.0>}]

让咱们修改KV.Registry来使用ETS表格.因为咱们的注册表须要一个名字做为参数,咱们能够用相同的名字命名ETS表格.ETS名与进程名存储在不一样的位置,因此它们不会冲突.

打开lib/kv/registry.ex,让咱们来改变它的实现.咱们已经为源代码的修改添加了注释:

defmodule KV.Registry do
  use GenServer

  ## Client API

  @doc """
  Starts the registry with the given `name`.
  """
  def start_link(name) do
    # 1. 传送名字给GenServer的init
    GenServer.start_link(__MODULE__, name, name: name)
  end

  @doc """
  Looks up the bucket pid for `name` stored in `server`.

  Returns `{:ok, pid}` if the bucket exists, `:error` otherwise.
  """
  def lookup(server, name) when is_atom(server) do
    # 2. 查找将直接在ETS中运行,不须要访问服务器
    case :ets.lookup(server, name) do
      [{^name, pid}] -> {:ok, pid}
      [] -> :error
    end
  end

  @doc """
  Ensures there is a bucket associated to the given `name` in `server`.
  """
  def create(server, name) do
    GenServer.cast(server, {:create, name})
  end

  @doc """
  Stops the registry.
  """
  def stop(server) do
    GenServer.stop(server)
  end

  ## Server callbacks

  def init(table) do
    # 3. 咱们已经用ETS表格取代了名称映射
    names = :ets.new(table, [:named_table, read_concurrency: true])
    refs  = %{}
    {:ok, {names, refs}}
  end

  # 4. 以前用于查找的handle_call回调已经删除

  def handle_cast({:create, name}, {names, refs}) do
    # 5. 对ETS表格进行读写,而非对映射
    case lookup(names, name) do
      {:ok, _pid} ->
        {:noreply, {names, refs}}
      :error ->
        {:ok, pid} = KV.Bucket.Supervisor.start_bucket
        ref = Process.monitor(pid)
        refs = Map.put(refs, ref, name)
        :ets.insert(names, {name, pid})
        {:noreply, {names, refs}}
    end
  end

  def handle_info({:DOWN, ref, :process, _pid, _reason}, {names, refs}) do
    {name, refs} = Map.pop(refs, ref)
    names = Map.delete(names, name)
    {:noreply, {names, refs}}
  end

  def handle_info(_msg, state) do
    {:noreply, state}
  end
end

注意到在修改以前,KV.Reigstry.lookup/2会将请求发送给服务器,但如今会直接从ETS表格中读取,改表格会被全部进程分享.这就是咱们实现的缓存机制背后的主要原理.

为了使缓存机制运行,ETS表格须要有让对其的访问:protected(默认),这样全部客户端均可以从中读取,而只有KV.Registry进程能写入.咱们也在建立表格时设置了read_concurrency: true,为普通的并发读取操做的脚本作了表格优化.

上诉修改破坏了咱们的测试,由于以前为全部操做使用的是注册表进程的pid,而如今注册表查找要求的是ETS表格名.然而,因为注册表进程和ETS表格有着相同的名字,就很好解决这个问题.将test/kv/registry_test.exs中的设置函数修改成:

setup context do
  {:ok, _} = KV.Registry.start_link(context.test)
  {:ok, registry: context.test}
end

咱们修改了setup,仍然有一些测试失败了.你可能会注意到每次测试的结果不一致.例如,"生成桶"的测试:

test "spawns buckets", %{registry: registry} do
  assert KV.Registry.lookup(registry, "shopping") == :error

  KV.Registry.create(registry, "shopping")
  assert {:ok, bucket} = KV.Registry.lookup(registry, "shopping")

  KV.Bucket.put(bucket, "milk", 1)
  assert KV.Bucket.get(bucket, "milk") == 1
end

可能会在这里失败:

{:ok, bucket} = KV.Registry.lookup(registry, "shopping")

为何这条会失败,咱们明明已经在上一条代码中建立了桶?

缘由就是,出于教学目的,咱们犯了两个错误:

\1. 咱们贸然作了优化(经过添加这个缓存层) \2. 咱们使用了cast/2(应该使用call/2)

#竞态条件?

使用Elixir开发并不能使你的代码魔兽竞态条件影响.然而,Elixir中默认的不共用任何事物的简单概念使得咱们更容易发现产生竞态条件的缘由.

在发出操做和从ETS表格观察到改变之间会有延迟,致使了咱们测试的失败.下面是咱们但愿看到的:

\1. 咱们调用KV.Rgistry.create(registry, "shopping") \2. 注册表建立了桶并更新了缓存表格 \3. 咱们使用KV.Registry.lookup(registry, "shopping")从表格中获取信息 \4. 返回{:ok, bucket}

然而,因为KV.Registry.create/2是一个投掷操做,这条命令将会在咱们真正写入表格以前返回!换句话或,实际发生的是:

\1. 咱们调用KV.Rgistry.create(registry, "shopping") \2. 咱们使用KV.Registry.lookup(ets, "shopping")从表格中获取信息 \3. 返回:error \4. 注册表建立了桶并更新了缓存表格

为了修正这个错误,咱们须要使得KV.Registry.create/2变为异步的,经过使用call/2代替cast/2.这就保证了客户端只会在对表格的改动发生事后继续.让咱们修改函数,以及它的回调:

def create(server, name) do
  GenServer.call(server, {:create, name})
end

def handle_call({:create, name}, _from, {names, refs}) do
  case lookup(names, name) do
    {:ok, pid} ->
      {:reply, pid, {names, refs}}
    :error ->
      {:ok, pid} = KV.Bucket.Supervisor.start_bucket
      ref = Process.monitor(pid)
      refs = Map.put(refs, ref, name)
      :ets.insert(names, {name, pid})
      {:reply, pid, {names, refs}}
  end
end

咱们简单地将回调从handle_cast/2改成了handle_call/3,并回复被建立了的桶的pid.一般来讲,Elixir开发者更喜欢使用call/2而不是cast/2,由于它也提供了背压(你会被挡住直到获得回复).在没必要要时使用cast/2也能够被看作是贸然的优化.

让咱们再次运行测试.这一次,咱们会传送--trace选项:

$ mix test --trace

--trace选项在你的测试死锁或遇到竞态条件时颇有用,由于它会异步运行全部测试(async: true无效)并展现每一个测试的详细信息.这一次咱们应该会获得一两个断断续续的失败:

1) test removes buckets on exit (KV.RegistryTest)
   test/kv/registry_test.exs:19
   Assertion with == failed
   code: KV.Registry.lookup(registry, "shopping") == :error
   lhs:  {:ok, #PID<0.109.0>}
   rhs:  :error
   stacktrace:
     test/kv/registry_test.exs:23

根据错误信息,咱们指望桶再也不存在,但它仍在那儿!这个问题与咱们刚才解决的正相反:以前在命令建立桶与更新表格之间存在延迟,如今是在桶进程死亡与它在表中的记录被删除之间存在延迟.

不走运的是这一次咱们不能简单地将handle_info/2这个负责清洁ETS表格的操做,改变为一个异步操做.相反咱们须要找到一个方法来保证注册表已经处理了:DOWN通知的发送,当桶崩溃时.

简单的方法是发送一个异步请求给注册表:由于信息会被按顺序处理,若是注册表回复了一个在Agent.stop调用以后的发送的请求,就意味着:DOWN消息已经被处理了.让咱们建立一个"bogus"桶,它是一个异步请求,在每一个测试中排在Agent.stop以后:

test "removes buckets on exit", %{registry: registry} do
  KV.Registry.create(registry, "shopping")
  {:ok, bucket} = KV.Registry.lookup(registry, "shopping")
  Agent.stop(bucket)

  # Do a call to ensure the registry processed the DOWN message
  _ = KV.Registry.create(registry, "bogus")
  assert KV.Registry.lookup(registry, "shopping") == :error
end

test "removes bucket on crash", %{registry: registry} do
  KV.Registry.create(registry, "shopping")
  {:ok, bucket} = KV.Registry.lookup(registry, "shopping")

  # Kill the bucket and wait for the notification
  Process.exit(bucket, :shutdown)

  # Wait until the bucket is dead
  ref = Process.monitor(bucket)
  assert_receive {:DOWN, ^ref, _, _, _}

  # Do a call to ensure the registry processed the DOWN message
  _ = KV.Registry.create(registry, "bogus")
  assert KV.Registry.lookup(registry, "shopping") == :error
end

咱们的测试如今能够(一直)经过了!

该总结一下咱们的优化章节了.咱们使用了ETS做为缓存机制,一人写万人读.更重要的是,一但数据能够被同步读取,就要小心竞态条件.

下一章咱们将讨论外部和内部的依赖,以及Mix如何帮助咱们管理巨型代码库.