Distribution

# Short names -- for development and local clusters
iex --sname alice@localhost
iex --sname bob@localhost

# Long names -- for production across machines
iex --name [email protected]
iex --name [email protected]

# Spawn a process on a remote node
pid = Node.spawn(:bob@localhost, fn ->
  receive do
    {:hello, sender} ->
      send(sender, {:hi_back, Node.self()})
  end
end)

# Send it a message from Alice
send(pid, {:hello, self()})

# Receive the reply
receive do
  {:hi_back, node} -> IO.puts("Got reply from #{node}")
end

# On bob@localhost -- start a named GenServer
{:ok, _pid} = GenServer.start_link(MyWorker, [], name: {:global, :my_worker})

# On alice@localhost -- call the remote GenServer
GenServer.call({:global, :my_worker}, :get_status)

# Execute a function on a remote node and get the result
:rpc.call(:bob@localhost, String, :upcase, ["hello"])
# "HELLO"

# Execute on a remote node without waiting for the result
:rpc.cast(:bob@localhost, IO, :puts, ["Hello from Alice!"])
# true

# Execute on multiple nodes simultaneously
:rpc.multicall([:"bob@localhost", :"charlie@localhost"], Enum, :sum, [[1, 2, 3]])
# {[6, 6], []}  -- {successful_results, failed_nodes}

# With a timeout (in milliseconds)
:rpc.call(:bob@localhost, MyModule, :slow_function, [arg], 5_000)

defmodule MyApp.ClusterSingleton do
  use GenServer

  def start_link(opts) do
    name = Keyword.get(opts, :name, __MODULE__)
    GenServer.start_link(__MODULE__, opts, name: {:global, name})
  end

  def get_state(name \\ __MODULE__) do
    GenServer.call({:global, name}, :get_state)
  end

  @impl true
  def init(opts) do
    {:ok, %{started_on: Node.self(), data: Keyword.get(opts, :data, %{})}}
  end

  @impl true
  def handle_call(:get_state, _from, state) do
    {:reply, state, state}
  end
end

defmodule MyApp.WorkerPool do
  use GenServer

  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts)
  end

  @impl true
  def init(opts) do
    group = Keyword.get(opts, :group, :workers)
    :pg.join(group, self())
    {:ok, %{group: group}}
  end

  @impl true
  def terminate(_reason, state) do
    :pg.leave(state.group, self())
    :ok
  end

  # Distribute work across all workers in the group
  def distribute_work(group \\ :workers, work_items) do
    members = :pg.get_members(group)

    work_items
    |> Enum.zip(Stream.cycle(members))
    |> Enum.each(fn {item, pid} ->
      GenServer.cast(pid, {:process, item})
    end)
  end

  @impl true
  def handle_cast({:process, item}, state) do
    # Process the work item
    result = do_work(item)
    IO.puts("#{inspect(Node.self())} processed: #{inspect(result)}")
    {:noreply, state}
  end

  defp do_work(item), do: item
end

# Using :pg for pub/sub across nodes
defmodule MyApp.PubSub do
  def subscribe(topic) do
    :pg.join(topic, self())
  end

  def unsubscribe(topic) do
    :pg.leave(topic, self())
  end

  def broadcast(topic, message) do
    for pid <- :pg.get_members(topic) do
      send(pid, {topic, message})
    end
    :ok
  end

  def local_broadcast(topic, message) do
    for pid <- :pg.get_local_members(topic) do
      send(pid, {topic, message})
    end
    :ok
  end
end

defmodule MyApp.DistributedComputation do
  def parallel_map(nodes, items, fun) do
    # Split work across nodes
    chunks = chunk_by_nodes(items, nodes)

    # Start tasks on each node
    tasks =
      Enum.zip(nodes, chunks)
      |> Enum.map(fn {node, chunk} ->
        Task.Supervisor.async({MyApp.TaskSupervisor, node}, fn ->
          Enum.map(chunk, fun)
        end)
      end)

    # Collect results
    tasks
    |> Task.await_many(30_000)
    |> List.flatten()
  end

  defp chunk_by_nodes(items, nodes) do
    chunk_size = div(length(items), length(nodes)) + 1
    Enum.chunk_every(items, chunk_size)
  end
end

# Usage -- distribute heavy computation across the cluster
nodes = [Node.self() | Node.list()]
results = MyApp.DistributedComputation.parallel_map(nodes, large_dataset, &expensive_transform/1)

# mix.exs
{:libcluster, "~> 3.3"}

# config/config.exs
config :libcluster,
  topologies: [
    k8s: [
      strategy: Cluster.Strategy.Kubernetes,
      config: [
        mode: :dns,
        kubernetes_node_basename: "my_app",
        kubernetes_selector: "app=my_app",
        polling_interval: 5_000
      ]
    ]
  ]

# For local development, use Gossip or Epmd strategies
config :libcluster,
  topologies: [
    local: [
      strategy: Cluster.Strategy.Gossip
    ]
  ]

defmodule MyApp.Application do
  use Application

  def start(_type, _args) do
    topologies = Application.get_env(:libcluster, :topologies, [])

    children = [
      {Cluster.Supervisor, [topologies, [name: MyApp.ClusterSupervisor]]},
      MyApp.Repo,
      MyAppWeb.Endpoint,
      {Task.Supervisor, name: MyApp.TaskSupervisor}
    ]

    Supervisor.start_link(children, strategy: :one_for_one)
  end
end

defmodule MyApp.ClusterMonitor do
  use GenServer

  require Logger

  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  @impl true
  def init(_opts) do
    :net_kernel.monitor_nodes(true)
    Logger.info("Cluster monitor started. Connected nodes: #{inspect(Node.list())}")
    {:ok, %{}}
  end

  @impl true
  def handle_info({:nodeup, node}, state) do
    Logger.info("Node joined the cluster: #{node}")
    # Rebalance work, sync state, etc.
    {:noreply, state}
  end

  @impl true
  def handle_info({:nodedown, node}, state) do
    Logger.warning("Node left the cluster: #{node}")
    # Redistribute work from the departed node
    {:noreply, state}
  end
end