Packages

phoenix_pubsub

2.2.0
2.2.0 2.1.3 2.1.2 retired 2.1.1 2.1.0 2.0.0 1.1.2 1.1.1 1.1.0 1.0.2 1.0.1 1.0.0 1.0.0-rc.0 0.1.0 0.0.1

Distributed PubSub and Presence platform

HexDocs

Current section

15 Versions

Jump to
Readme
15 Versions
0 Dependencies
186 Dependants
Files
Activity
Hide whitespace

Compare versions

14 files changed
+503 additions
-200 deletions
  @@ -2,6 +2,20 @@
2 2
3 3 This new version of Phoenix.PubSub provides a simpler, more extensible, and more performant Phoenix.PubSub API. For users of Phoenix.PubSub, the API is the same, although frameworks and other adapters will have to migrate accordingly (which often means less code).
4 4
5 + ## 2.2.0 (2025-10-22)
6 +
7 + ### Enhancements
8 + - Allow the registry size to be set separate from pool size
9 + - Introduce `:broadcast_pool_size` option to allow safe pool size migration
10 +
11 + ### Bug fixes
12 + - Only restart shards if they terminate unexpectedly
13 +
14 + ## 2.1.4 (2024-09-27)
15 +
16 + ### Enhancements
17 + - Add `:permdown_on_shutdown` option
18 +
5 19 ## 2.1.3 (2023-06-14)
6 20
7 21 ### Bug fixes
  @@ -2,7 +2,7 @@
2 2
3 3 > Distributed PubSub and Presence platform for the Phoenix Framework
4 4
5 - [![Build Status](https://api.travis-ci.org/phoenixframework/phoenix_pubsub.svg)](https://travis-ci.org/phoenixframework/phoenix_pubsub)
5 + [![Build Status](https://github.com/phoenixframework/phoenix_pubsub/actions/workflows/ci.yml/badge.svg)](https://github.com/phoenixframework/phoenix_pubsub/actions/workflows/ci.yml)
6 6
7 7 ## Usage
  @@ -1,10 +1,13 @@
1 - {<<"app">>,<<"phoenix_pubsub">>}.
2 - {<<"build_tools">>,[<<"mix">>]}.
1 + {<<"links">>,
2 + [{<<"github">>,<<"https://github.com/phoenixframework/phoenix_pubsub">>}]}.
3 + {<<"name">>,<<"phoenix_pubsub">>}.
4 + {<<"version">>,<<"2.2.0">>}.
3 5 {<<"description">>,<<"Distributed PubSub and Presence platform">>}.
4 6 {<<"elixir">>,<<"~> 1.6">>}.
5 7 {<<"files">>,
6 8 [<<"lib">>,<<"lib/phoenix">>,<<"lib/phoenix/tracker">>,
7 9 <<"lib/phoenix/tracker/clock.ex">>,<<"lib/phoenix/tracker/replica.ex">>,
10 + <<"lib/phoenix/tracker/shutdown_handler.ex">>,
8 11 <<"lib/phoenix/tracker/delta_generation.ex">>,
9 12 <<"lib/phoenix/tracker/state.ex">>,<<"lib/phoenix/tracker/shard.ex">>,
10 13 <<"lib/phoenix/pubsub">>,<<"lib/phoenix/pubsub/adapter.ex">>,
  @@ -13,9 +16,7 @@
13 16 <<"lib/phoenix/tracker.ex">>,<<"lib/phoenix/pubsub.ex">>,<<"test/shared">>,
14 17 <<"test/shared/pubsub_test.exs">>,<<"CHANGELOG.md">>,<<"LICENSE.md">>,
15 18 <<"mix.exs">>,<<"README.md">>]}.
19 + {<<"app">>,<<"phoenix_pubsub">>}.
16 20 {<<"licenses">>,[<<"MIT">>]}.
17 - {<<"links">>,
18 - [{<<"github">>,<<"https://github.com/phoenixframework/phoenix_pubsub">>}]}.
19 - {<<"name">>,<<"phoenix_pubsub">>}.
20 21 {<<"requirements">>,[]}.
21 - {<<"version">>,<<"2.1.3">>}.
22 + {<<"build_tools">>,[<<"mix">>]}.
  @@ -34,7 +34,7 @@ defmodule Phoenix.PubSub do
34 34 It supports a `:pool_size` option to be given alongside
35 35 the name, defaults to `1`. Note the `:pool_size` must
36 36 be the same throughout the cluster, therefore don't
37 - configure the pool size based on `System.schedulers_online/1`,
37 + configure the pool size based on `System.schedulers_online/0`,
38 38 especially if you are using machines with different specs.
39 39
40 40 * `Phoenix.PubSub.Redis` - uses Redis to exchange data between
  @@ -59,6 +59,83 @@ defmodule Phoenix.PubSub do
59 59 custom `value` to provide "fastlaning", allowing messages broadcast
60 60 to thousands or even millions of users to be encoded once and written
61 61 directly to sockets instead of being encoded per channel.
62 +
63 + ## Safe pool size migration (when using `Phoenix.PubSub.PG2` adapter)
64 +
65 + When you need to change the pool size in a running cluster,
66 + you can use the `broadcast_pool_size` option to ensure no
67 + messages are lost during deployment. This is particularly
68 + important when increasing the pool size.
69 +
70 + Here's how to safely increase the pool size from 1 to 2:
71 +
72 + 1. Initial state - Current configuration with `pool_size: 1`:
73 + ```
74 + {Phoenix.PubSub, name: :my_pubsub, pool_size: 1}
75 + ```
76 +
77 + ```mermaid
78 + graph TD
79 + subgraph "Initial State"
80 + subgraph "Node 1"
81 + A1[Shard 1<br/>Broadcast & Receive]
82 + end
83 + subgraph "Node 2"
84 + B1[Shard 1<br/>Broadcast & Receive]
85 + end
86 + A1 <--> B1
87 + end
88 + ```
89 +
90 + 2. First deployment - Set the new pool size but keep broadcasting on the old size:
91 + ```
92 + {Phoenix.PubSub, name: :my_pubsub, pool_size: 2, broadcast_pool_size: 1}
93 + ```
94 +
95 + ```mermaid
96 + graph TD
97 + subgraph "First Deployment"
98 + subgraph "Node 1"
99 + A1[Shard 1<br/>Broadcast & Receive]
100 + A2[Shard 2<br/>Broadcast & Receive]
101 + end
102 + subgraph "Node 2"
103 + B1[Shard 1<br/>Broadcast & Receive]
104 + B2[Shard 2<br/>Receive Only]
105 + end
106 + A1 <--> B1
107 + A2 --> B2
108 + end
109 + ```
110 +
111 + 3. Final deployment - All nodes running with new pool size:
112 + ```
113 + {Phoenix.PubSub, name: :my_pubsub, pool_size: 2}
114 + ```
115 +
116 + ```mermaid
117 + graph TD
118 + subgraph "Final State"
119 + subgraph "Node 1"
120 + A1[Shard 1<br/>Broadcast & Receive]
121 + A2[Shard 2<br/>Broadcast & Receive]
122 + end
123 + subgraph "Node 2"
124 + B1[Shard 1<br/>Broadcast & Receive]
125 + B2[Shard 2<br/>Broadcast & Receive]
126 + end
127 + A1 <--> B1
128 + A2 <--> B2
129 + end
130 + ```
131 +
132 + This two-step process ensures that:
133 + - All nodes can receive messages from both old and new pool sizes
134 + - No messages are lost during the transition
135 + - The cluster remains fully functional throughout the deployment
136 +
137 + To decrease the pool size, follow the same process in reverse order.
138 +
62 139 """
63 140
64 141 @type node_name :: atom | binary
  @@ -87,6 +164,13 @@ defmodule Phoenix.PubSub do
87 164 * `:adapter` - the adapter to use (defaults to `Phoenix.PubSub.PG2`)
88 165 * `:pool_size` - number of pubsub partitions to launch
89 166 (defaults to one partition for every 4 cores)
167 + * `:registry_size` - number of `Registry` partitions to launch
168 + (defaults to `:pool_size`). This controls the number of Registry partitions
169 + used for storing subscriptions and can be tuned independently from `:pool_size`
170 + for better performance characteristics.
171 + * `:broadcast_pool_size` - number of pubsub partitions used for broadcasting messages
172 + (defaults to `:pool_size`). This option is used during pool size migrations to ensure
173 + no messages are lost. See the "Safe Pool Size Migration" section in the module documentation.
90 174
91 175 """
92 176 @spec child_spec(keyword) :: Supervisor.child_spec()
  @@ -162,7 +246,7 @@ defmodule Phoenix.PubSub do
162 246
163 247 The default dispatcher will broadcast the message to all subscribers except for the
164 248 process that initiated the broadcast.
165 -
249 +
166 250 A custom dispatcher may also be given as a fifth, optional argument.
167 251 See the "Custom dispatching" section in the module documentation.
168 252 """
  @@ -202,7 +286,7 @@ defmodule Phoenix.PubSub do
202 286
203 287 The default dispatcher will broadcast the message to all subscribers except for the
204 288 process that initiated the broadcast.
205 -
289 +
206 290 A custom dispatcher may also be given as a fifth, optional argument.
207 291 See the "Custom dispatching" section in the module documentation.
208 292 """
  @@ -226,7 +310,7 @@ defmodule Phoenix.PubSub do
226 310 A custom dispatcher may also be given as a fifth, optional argument.
227 311 See the "Custom dispatching" section in the module documentation.
228 312 """
229 - @spec direct_broadcast(t, topic, message, dispatcher) :: :ok | {:error, term}
313 + @spec direct_broadcast(node_name, t, topic, message, dispatcher) :: :ok | {:error, term}
230 314 def direct_broadcast(node_name, pubsub, topic, message, dispatcher \\ __MODULE__)
231 315 when is_atom(pubsub) and is_binary(topic) and is_atom(dispatcher) do
232 316 {:ok, {adapter, name}} = Registry.meta(pubsub, :pubsub)
  @@ -61,29 +61,41 @@ defmodule Phoenix.PubSub.PG2 do
61 61 def start_link(opts) do
62 62 name = Keyword.fetch!(opts, :name)
63 63 pool_size = Keyword.get(opts, :pool_size, 1)
64 - adapter_name = Keyword.fetch!(opts, :adapter_name)
65 - Supervisor.start_link(__MODULE__, {name, adapter_name, pool_size}, name: :"#{adapter_name}_supervisor")
64 + broadcast_pool_size = Keyword.get(opts, :broadcast_pool_size, pool_size)
65 +
66 + if pool_size < broadcast_pool_size do
67 + {:error, "the :pool_size option must be greater than or equal to the :broadcast_pool_size option"}
68 + else
69 + adapter_name = Keyword.fetch!(opts, :adapter_name)
70 + Supervisor.start_link(__MODULE__, {name, adapter_name, pool_size, broadcast_pool_size}, name: :"#{adapter_name}_supervisor")
71 + end
66 72 end
67 73
68 74 @impl true
69 - def init({name, adapter_name, pool_size}) do
75 + def init({name, adapter_name, pool_size, broadcast_pool_size}) do
76 +
77 + listener_groups = groups(adapter_name, pool_size)
78 + broadcast_groups = groups(adapter_name, broadcast_pool_size)
79 +
80 + :persistent_term.put(adapter_name, List.to_tuple(broadcast_groups))
81 +
82 + children =
83 + for group <- listener_groups do
84 + Supervisor.child_spec({Phoenix.PubSub.PG2Worker, {name, group}}, id: group)
85 + end
86 +
87 + Supervisor.init(children, strategy: :one_for_one)
88 + end
89 +
90 + defp groups(adapter_name, pool_size) do
70 91 [_ | groups] =
71 92 for number <- 1..pool_size do
72 93 :"#{adapter_name}_#{number}"
73 94 end
74 95
75 - # Use `adapter_name` for the first in the pool for backwards compatability
96 + # Use `adapter_name` for the first in the pool for backwards compatibility
76 97 # with v2.0 when the pool_size is 1.
77 - groups = [adapter_name | groups]
78 -
79 - :persistent_term.put(adapter_name, List.to_tuple(groups))
80 -
81 - children =
82 - for group <- groups do
83 - Supervisor.child_spec({Phoenix.PubSub.PG2Worker, {name, group}}, id: group)
84 - end
85 -
86 - Supervisor.init(children, strategy: :one_for_one)
98 + [adapter_name | groups]
87 99 end
88 100 end
Loading more files…