refactored monitor code to understand how it works

Author: zookzook

lib/mongo.ex

@@ -573,7 +573,6 @@ defmodule Mongo do
     rp_opts = [read_preference: Keyword.get(opts, :read_preference, rp)]
     with {:ok, conn, slave_ok, _} <- select_server(topology_pid, :read, rp_opts),
          opts = Keyword.put(opts, :slave_ok, slave_ok),
-         opts = Keyword.put(opts, :lsid, session_id), füge die session_id hier rein
          do: exec_command(conn, cmd, opts)
   end
 
@@ -586,8 +585,6 @@ defmodule Mongo do
     with {:ok, _cmd, doc} <- DBConnection.execute(conn, action, [cmd], defaults(opts)),
          {:ok, doc} <- check_for_error(doc) do
       {:ok, doc}
-    else
-
     end
   end
 
@@ -911,11 +908,8 @@ defmodule Mongo do
           ] |> filter_nils()
 
 
-    ## {:ok, connection, slave_ok, mongos?}
-  {:ok, conn, slave_ok, mongos, session_id} = Topology.select_server(topology_pid, :write, opts)
-
     with {:ok, conn, _, _} <- select_server(topology_pid, :write, opts),
-         {:ok, doc}        <- exec_command(conn, cmd, session_id, opts) do
+         {:ok, doc}        <- exec_command(conn, cmd, opts) do
 
       case doc do
 

lib/mongo/monitor.ex

@@ -1,130 +1,174 @@
 # see https://github.com/mongodb/specifications/blob/master/source/server-discovery-and-monitoring/server-discovery-and-monitoring.rst#monitoring
 defmodule Mongo.Monitor do
-  @moduledoc false
+  @moduledoc """
+  Each server has a monitor process. The monitor process is created by the topology process.
+
+  If the network connection is working, then the monitor process reports this and the topology process starts the
+  connection pool. Per server we get 1 + pool size connections to each server.
+
+  After waiting for `heartbeat_frequency_ms` milliseconds, the monitor process calls `isMaster` command and
+  reports the result to the topology process.
+
+  The result of the `isMaster` command is mapped the `ServerDescription` structure and sent to the topology process, which
+  updates it internal data structure.
+  """
+
   use GenServer
-  use Bitwise
-  require Logger
+
+  # require Logger
+
+  alias Mongo.Topology
   alias Mongo.ServerDescription
-  alias Mongo.Events.ServerHeartbeatStartedEvent
-  alias Mongo.Events.{ServerHeartbeatStartedEvent, ServerHeartbeatFailedEvent,
-                      ServerHeartbeatSucceededEvent}
+  alias Mongo.Events.{ServerHeartbeatStartedEvent, ServerHeartbeatFailedEvent,ServerHeartbeatSucceededEvent}
 
   # this is not configurable because the specification says so
   # see https://github.com/mongodb/specifications/blob/master/source/server-discovery-and-monitoring/server-discovery-and-monitoring.rst#minheartbeatfrequencyms
   @min_heartbeat_frequency_ms 500
 
-  def start_link(args, gen_server_opts \\ []) do
-    GenServer.start_link(__MODULE__, args, gen_server_opts)
+  def start_link(args) do
+    GenServer.start_link(__MODULE__, args)
   end
 
   # We need to stop asynchronously because a Monitor can call the Topology
   # which may try to stop the same Monitor that called it. Ending in a timeout.
   # See issues #139 for some information.
+
+  @doc """
+  Stop the monitor process.
+
+  We need to stop asynchronously because a Monitor can call the Topology
+  which may try to stop the same Monitor that called it. Ending in a timeout.
+  """
   def stop(pid) do
     GenServer.cast(pid, :stop)
   end
 
+  @doc """
+  Try to update the server description.
+  """
   def force_check(pid) do
-    GenServer.call(pid, :check, :infinity)
-  end
-
-  ## GenServer callbacks
-
-  @doc false
-  def init([server_description, topology_pid, heartbeat_frequency_ms, connection_opts]) do
-    opts = # monitors don't authenticate and use the "admin" database
-      connection_opts
-      |> Keyword.put(:database, "admin")
-      |> Keyword.put(:skip_auth, true)
-      |> Keyword.put(:after_connect, {__MODULE__, :connected, [self(), topology_pid]})
-      |> Keyword.put(:backoff_min, heartbeat_frequency_ms)
-      |> Keyword.put(:backoff_max, heartbeat_frequency_ms)
-      |> Keyword.put(:backoff_type, :rand)
-      |> Keyword.put(:connection_type, :monitor)
-      |> Keyword.put(:topology_pid, topology_pid)
-      |> Keyword.put(:pool_size, 1)
-      |> Keyword.put(:idle_interval, 5_000)
-
-    {:ok, pid} = DBConnection.start_link(Mongo.MongoDBConnection, opts)
-    :ok = GenServer.cast(self(), :check)
-    {:ok, %{
-      connection_pid: pid,
-      topology_pid: topology_pid,
-      server_description: server_description,
-      heartbeat_frequency_ms: heartbeat_frequency_ms,
-      opts: opts
-    }}
-  end
-
-  @doc false
+    GenServer.cast(pid, :update)
+  end
+
+
+  @doc """
+  Initialize the monitor process
+  """
+  def init([address, topology_pid, heartbeat_frequency_ms, connection_opts]) do
+
+    # monitors don't authenticate and use the "admin" database
+    opts = connection_opts
+           |> Keyword.put(:database, "admin")
+           |> Keyword.put(:skip_auth, true)
+           |> Keyword.put(:after_connect, {__MODULE__, :connected, [self(), topology_pid]})
+           |> Keyword.put(:backoff_min, heartbeat_frequency_ms)
+           |> Keyword.put(:backoff_max, heartbeat_frequency_ms)
+           |> Keyword.put(:backoff_type, :rand)
+           |> Keyword.put(:connection_type, :monitor)
+           |> Keyword.put(:topology_pid, topology_pid)
+           |> Keyword.put(:pool_size, 1)
+           |> Keyword.put(:idle_interval, 5_000)
+
+    with {:ok, pid} <- DBConnection.start_link(Mongo.MongoDBConnection, opts) do
+
+      ## we start after one second
+      Process.send_after(self(), :update, 1_000)
+
+      {:ok, %{
+        connection_pid: pid,                            ## our connection pid to the mongodb server
+        topology_pid: topology_pid,                     ## the topology_pid to which we report
+        address: address,                               ## the address of the server, needed to make updates
+        round_trip_time: nil,                           ## current round_trip_time, needed to make average value
+        heartbeat_frequency_ms: heartbeat_frequency_ms, ## current heartbeat_frequency_ms
+        opts: opts ## options
+      }}
+
+    end
+
+  end
+
+  @doc """
+  In this case we stop the DBConnection.
+  """
   def terminate(reason, state) do
     GenServer.stop(state.connection_pid, reason)
   end
 
-  @doc false
+  @doc """
+  Report the connection event, so the topology process can now create the connection pool.
+  """
   def connected(_connection, me, topology_pid) do
-    GenServer.cast(topology_pid, {:connected, me})
+    Topology.monitor_connected(topology_pid, me)
   end
 
-  @doc false
-  def handle_cast(:check, state) do
-    check(state)
+  @doc """
+  Time to update the topology. Calling `isMaster` and updating the server description
+  """
+  def handle_cast(:update, state) do
+    new_state = update_server_description(state)
+    # we return with heartbeat_frequency_ms, so after heartbeat_frequency_ms handle_info(:timeout...) gets called.
+    {:noreply, new_state, new_state.heartbeat_frequency_ms}
   end
+
   def handle_cast(:stop, state) do
     exit(:normal)
     {:noreply, state}
   end
 
-  @doc false
-  def handle_call(:check, _from, state) do
-    {_, state, diff} = check(state)
-    {:reply, diff, state}
-  end
-
-  @doc false
+  @doc """
+  The `:timeout` call is the periodic call defined by the heartbeat frequency
+  The ':update' call updates the server description for the topology process
+  """
   def handle_info(:timeout, state) do
-    check(state)
+    new_state = update_server_description(state)
+    {:noreply, new_state, new_state.heartbeat_frequency_ms}
+  end
+  def handle_info(:update, state) do
+    new_state = update_server_description(state)
+    # we return with heartbeat_frequency_ms, so after heartbeat_frequency_ms handle_info(:timeout...) gets called.
+    {:noreply, new_state, new_state.heartbeat_frequency_ms}
   end
 
-  ## Private functions
-
-  defp check(state) do
-    diff = :os.system_time(:milli_seconds) - state.server_description.last_update_time
-    case diff < @min_heartbeat_frequency_ms  do
-       true  -> {:noreply, state, diff}
-       false ->
-         server_description = get_server_description(state, 0)
-         GenServer.cast(state.topology_pid, {:server_description, server_description})
-         {:noreply, %{state | server_description: server_description}, state.heartbeat_frequency_ms}
-    end
+  ##
+  # Get a new server description from the server and send it to the Topology process.
+  #
+  defp update_server_description(%{topology_pid: topology_pid} = state) do
+    %{:round_trip_time => round_trip_time} = server_description = get_server_description(state, 0)
+    Topology.update_server_description(topology_pid, server_description)
+    %{state | round_trip_time: round_trip_time}
   end
 
   # see https://github.com/mongodb/specifications/blob/master/source/server-discovery-and-monitoring/server-discovery-and-monitoring.rst#network-error-when-calling-ismaster
   ##
   # Calls isMaster and parses the result to update the server description
+  # In case of one network error, the function tries one more time to call isMaster command
   #
-  defp get_server_description(%{connection_pid: conn_pid, server_description: last_server_description, opts: opts} = state, counter) do
+  defp get_server_description(%{connection_pid: conn_pid, address: address, round_trip_time: last_rtt, opts: opts} = state, counter) do
 
     Mongo.Events.notify(%ServerHeartbeatStartedEvent{ connection_pid: conn_pid})
 
-    {result, finish_time, rtt} = call_is_master(conn_pid, opts)
+    {result, finish_time, rtt} = call_is_master_command(conn_pid, opts)
     case result do
       {:ok, is_master_reply} ->
         notify_success(rtt, is_master_reply, conn_pid)
-        ServerDescription.from_is_master(last_server_description, rtt, finish_time, is_master_reply)
+        ServerDescription.from_is_master(address, average_rtt(last_rtt, rtt), finish_time, is_master_reply)
 
       {:error, error} when counter == 1 ->
         notify_error(rtt, error, conn_pid)
-        ServerDescription.from_is_master_error(last_server_description, error)
-      {:error, error} ->  get_server_description(state, counter + 1)
+        ServerDescription.from_is_master_error(address, error)
+      {:error, _error} -> get_server_description(state, counter + 1)
     end
   end
 
+  defp average_rtt(nil, rtt) do
+    round(rtt)
+  end
   defp average_rtt(last_rtt, rtt) do
     round(0.2 * rtt + 0.8 * last_rtt)
   end
 
-  defp call_is_master(conn_pid, opts) do
+  defp call_is_master_command(conn_pid, opts) do
+
     start_time  = System.monotonic_time
     result      = Mongo.exec_command(conn_pid, [isMaster: 1], opts)
     finish_time = System.monotonic_time
@@ -150,30 +194,4 @@ defmodule Mongo.Monitor do
     })
   end
 
-  defp get_server_description_old(%{connection_pid: conn_pid, server_description: last_server_description, opts: opts}) do
-    Mongo.Events.notify(%ServerHeartbeatStartedEvent{ connection_pid: conn_pid})
-
-    {result, finish_time, rtt} = call_is_master(conn_pid, opts)
-    case result do
-      {:ok, is_master_reply} ->
-        notify_success(rtt, is_master_reply, conn_pid)
-        ServerDescription.from_is_master(last_server_description, rtt, finish_time, is_master_reply)
-
-      {:error, error} ->
-        if last_server_description.type in [:unknown, :possible_primary] do
-          notify_error(rtt, error, conn_pid)
-          ServerDescription.from_is_master_error(last_server_description, error)
-        else
-          {result, finish_time, rtt} = call_is_master(conn_pid, opts)
-          case result do
-            {:ok, is_master_reply} ->
-              notify_success(rtt, is_master_reply, conn_pid)
-              ServerDescription.from_is_master(last_server_description, rtt, finish_time, is_master_reply)
-            {:error, error} ->
-              notify_error(rtt, error, conn_pid)
-              ServerDescription.from_is_master_error(last_server_description, error)
-          end
-        end
-    end
-  end
 end

lib/mongo/server_description.ex

@@ -48,39 +48,34 @@ defmodule Mongo.ServerDescription do
     }, map)
   end
 
-  def from_is_master_error(last_server_description, error) do
+  def from_is_master_error(address, error) do
     defaults(%{
-      address: last_server_description.address,
+      address: address,
       error: error
     })
   end
 
   # see https://github.com/mongodb/specifications/blob/master/source/server-discovery-and-monitoring/server-discovery-and-monitoring.rst#parsing-an-ismaster-response
-  def from_is_master(last_description, rtt, finish_time, is_master_reply) do
-    last_rtt = last_description.round_trip_time || rtt
-
-    defaults(%{
-      address: last_description.address,
-      round_trip_time: round(0.2 * rtt + 0.8 * last_rtt),
+  def from_is_master(address, rtt, finish_time, is_master_reply) do
+    %{
+      address: address,
+      round_trip_time: rtt,
       type: determine_server_type(is_master_reply),
-      last_write_date: get_in(is_master_reply,
-                              ["lastWrite", "lastWriteDate"]),
+      last_write_date: get_in(is_master_reply,["lastWrite", "lastWriteDate"]),
       op_time: get_in(is_master_reply, ["lastWrite", "opTime"]),
       last_update_time: finish_time,
       min_wire_version: is_master_reply["minWireVersion"] || 0,
       max_wire_version: is_master_reply["maxWireVersion"] || 0,
       me: is_master_reply["me"],
       hosts: (is_master_reply["hosts"] || []) |> Enum.map(&String.downcase/1),
-      passives: (is_master_reply["passives"] || [])
-                |> Enum.map(&String.downcase/1),
-      arbiters: (is_master_reply["arbiters"] || [])
-                |> Enum.map(&String.downcase/1),
+      passives: (is_master_reply["passives"] || []) |> Enum.map(&String.downcase/1),
+      arbiters: (is_master_reply["arbiters"] || []) |> Enum.map(&String.downcase/1),
       tag_set: is_master_reply["tags"] || %{},
       set_name: is_master_reply["setName"],
       set_version: is_master_reply["setVersion"],
       election_id: is_master_reply["electionId"],
       primary: is_master_reply["primary"]
-    })
+    }
   end
 
   # see https://github.com/mongodb/specifications/blob/master/source/server-discovery-and-monitoring/server-discovery-and-monitoring.rst#type
@@ -89,14 +84,10 @@ defmodule Mongo.ServerDescription do
   defp determine_server_type(%{"isreplicaset" => true}), do: :rs_ghost
   defp determine_server_type(%{"setName" => set_name} = is_master_reply) when set_name != nil do
     case is_master_reply do
-      %{"ismaster" => true} ->
-        :rs_primary
-      %{"secondary" => true} ->
-        :rs_secondary
-      %{"arbiterOnly" => true} ->
-        :rs_arbiter
-      _ ->
-        :rs_other
+      %{"ismaster" => true}    ->  :rs_primary
+      %{"secondary" => true}   -> :rs_secondary
+      %{"arbiterOnly" => true} -> :rs_arbiter
+      _                        -> :rs_other
     end
   end
   defp determine_server_type(_), do: :standalone