d7e7f9b43cc2aa584a6fa1810dd59222d9a5a9e4-PR-224

Committed 08 Apr 2026 03:07PM UTC coverage: 98.675% (+0.1%) from 98.532%

Build # d7e7f9b43cc2aa584a6fa1810dd59222d9a5a9e4-PR-224

Build Type

Pull #224

github

Committed by

satoren

Commit Message

performance optimization for doc server

| Benchmark | Before | After | Improvement |
|-----------|--------|-------|-------------|
| sync_step1 (100 cells) | 9.99 μs | 4.21 μs | **2.37x** |
| sync_step1 (10,000 cells) | 223.54 μs | 52.46 μs | **4.26x** ⭐ |
| sync_step1 (empty) | 8.61 μs | 3.08 μs | **2.79x** |
| query_awareness | 5.56 μs | 2.83 μs | **1.96x** |
| **Average** | — | — | **2.75x** |

Pull Request Pull Request #224: performance optimization for doc server

Coverage Stats

8 of 9 new or added lines in 2 files covered. (88.89%)

1 existing line in 1 file now uncovered.

596 of 604 relevant lines covered (98.68%)

26.25 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.8

/lib/server/doc_server_worker.ex

defmodule Yex.DocServer.Worker do
  @moduledoc false
  use GenServer, restart: :temporary
  require Logger

  alias Yex.DocServer.State

  alias Yex.{Doc, Awareness}

  # MSG_QUERY_AWARENESS (<<3>>) — bypass message_decode NIF
  @query_awareness_call :__yex_query_awareness
  @sync_step1_raw_call :__yex_sync_step1_raw

  @query_awareness_message <<3>>
  def process_message_v1(server, @query_awareness_message, _origin) do
    GenServer.call(server, @query_awareness_call)
  end

  # MSG_SYNC (0) + MSG_SYNC_STEP_1 (0) — bypass message_decode NIF, pass sv_payload directly
  def process_message_v1(server, <<0, 0, sv_payload::binary>>, _origin) do
    GenServer.call(server, {@sync_step1_raw_call, sv_payload})
  end

  def process_message_v1(server, message, origin) do
    fallback_process_message_v1(server, message, origin)
  end

  defp fallback_process_message_v1(server, message, origin) do
    case Yex.Sync.message_decode(message) do
      {:ok, message} ->
        message_v1(server, message, origin)

      error ->
        error
    end
  end

  defp message_v1(server, {:sync, {:sync_step1, encoded_state_vector}}, origin) do
    GenServer.call(server, {__MODULE__, :document_sync_step1, encoded_state_vector, origin})
  end

  defp message_v1(server, {:sync, {message_type, encoded_diff}}, origin)
       when message_type in [:sync_step2, :sync_update] do
    GenServer.cast(server, {__MODULE__, :document_update, encoded_diff, origin})
  end

  defp message_v1(server, {:awareness, awareness}, origin) do
    GenServer.cast(server, {__MODULE__, :awareness_update, awareness, origin})
  end

  defp message_v1(server, :query_awareness, _origin) do
    GenServer.call(server, @query_awareness_call)
  end

  defp message_v1(_server, _message, _origin) do
    {:error, :unknown_message}
  end

  ## Callbacks

  @impl true
  def init(arg) do
    module = Keyword.fetch!(arg, :module)
    option = Keyword.get(arg, :doc_option, nil)
    assigns = Keyword.get(arg, :assigns, %{})
    doc = if option, do: Doc.with_options(option), else: Doc.new()

    if function_exported?(module, :handle_update_v1, 4) do
      Doc.monitor_update_v1(doc, metadata: __MODULE__)
    end

    awareness = setup_awareness(doc, module)

    module.init(arg, %State{
      assigns: assigns,
      doc: doc,
      awareness: awareness,
      module: module
    })
  end

  defp setup_awareness(doc, module) do
    if function_exported?(module, :handle_awareness_change, 4) or
         function_exported?(module, :handle_awareness_update, 4) do
      case Awareness.new(doc) do
        {:ok, awareness} ->
          monitor_awareness_events(awareness, module)

          awareness
      end
    end
  end

  defp monitor_awareness_events(awareness, module) do
    if function_exported?(module, :handle_awareness_change, 4) do
      Awareness.monitor_change(awareness, metadata: __MODULE__)
    end

    if function_exported?(module, :handle_awareness_update, 4) do
      Awareness.monitor_update(awareness, metadata: __MODULE__)
    end
  end

  @impl true
  def handle_call(
        {@sync_step1_raw_call, sv_payload},
        _from,
        %{doc: doc, awareness: awareness} = state
      ) do
    {:reply, Yex.Nif.encode_sync_step1_response_v1(doc, nil, sv_payload, awareness), state}
  end

  @impl true
  def handle_call(
        @query_awareness_call,
        _from,
        %{awareness: nil} = state
      ) do
    {:reply, {:ok, []}, state}
  end

  @impl true
  def handle_call(
        @query_awareness_call,
        _from,
        %{awareness: awareness} = state
      ) do
    {:reply, Yex.Nif.encode_awareness_reply_v1(awareness), state}
  end

  @impl true
  def handle_call(
        {Yex.Doc, :run, fun},
        _from,
        state
      ) do
    {:reply, fun.(), state}
  end

  @impl true
  def handle_call(request, from, %{module: module} = state) do
    module.handle_call(request, from, state)
  end

  @impl true
  def handle_cast(
        {__MODULE__, :document_update, update, origin},
        %{doc: doc} = state
      ) do
    Yex.Doc.transaction(doc, origin, fn ->
      case Yex.apply_update(doc, update) do
        :ok ->
          :ok

        {:error, reason} ->
          Logger.log(:warning, inspect(reason))
          :ok
      end
    end)

    # Process update messages immediately
    handle_update_v1_immediately(state)
  end

  @impl true
  def handle_cast(
        {__MODULE__, :awareness_update, _message, _origin},
        %{awareness: nil} = state
      ) do
    #    Logger.warning("Received an awareness message, but ignored it because it is not enabled in this module. ")
    {:noreply, state}
  end

  @impl true
  def handle_cast(
        {__MODULE__, :awareness_update, message, origin},
        %{awareness: awareness} = state
      ) do
    Awareness.apply_update(awareness, message, origin)

    # Process update messages immediately
    handle_awareness_event_immediately(state)
  end

  @impl true
  def handle_cast(request, %{module: module} = state) do
    module.handle_cast(request, state)
  end

  @impl true
  def handle_info({:update_v1, update, origin, __MODULE__}, %{module: module, doc: doc} = state) do
    module.handle_update_v1(doc, update, origin, state)
  end

  @impl true
  def handle_info(
        {:awareness_change, change, origin, __MODULE__},
        %{module: module, awareness: awareness} = state
      ) do
    module.handle_awareness_change(awareness, change, origin, state)
  end

  @impl true
  def handle_info(
        {:awareness_update, change, origin, __MODULE__},
        %{module: module, awareness: awareness} = state
      ) do
    module.handle_awareness_update(awareness, change, origin, state)
  end

  @impl true
  def handle_info(msg, %{module: module} = state) do
    module.handle_info(msg, state)
  end

  @impl true
  def terminate(reason, %{module: module} = state) do
    if function_exported?(module, :terminate, 2) do
      module.terminate(reason, state)
    else
      :ok
    end
  end

  defp handle_update_v1_immediately(%{doc: doc, module: module} = state) do
    receive do
      {:update_v1, update, origin, __MODULE__} ->
        case module.handle_update_v1(doc, update, origin, state) do
          {:noreply, state} ->
            handle_update_v1_immediately(state)

          result ->
            result
        end
    after
      0 ->
        {:noreply, state}
    end
  end

  defp handle_awareness_event_immediately(%{awareness: awareness, module: module} = state) do
    receive do
      {:awareness_change, change, origin, __MODULE__} ->
        case module.handle_awareness_change(awareness, change, origin, state) do
          {:noreply, state} ->
            handle_awareness_event_immediately(state)

          result ->
            result
        end

      {:awareness_update, change, origin, __MODULE__} ->
        case module.handle_awareness_update(awareness, change, origin, state) do
          {:noreply, state} ->
            handle_awareness_event_immediately(state)

          result ->
            result
        end
    after
      0 ->
        {:noreply, state}
    end
  end
end

1	defmodule Yex.DocServer.Worker do
2	@moduledoc false
3	use GenServer, restart: :temporary
4	require Logger
5
6	alias Yex.DocServer.State
7
8	alias Yex.{Doc, Awareness}
9
10	# MSG_QUERY_AWARENESS (<<3>>) — bypass message_decode NIF
11	@query_awareness_call :__yex_query_awareness
12	@sync_step1_raw_call :__yex_sync_step1_raw
13
14	@query_awareness_message <<3>>
15	def process_message_v1(server, @query_awareness_message, _origin) do
16	GenServer.call(server, @query_awareness_call)	4✔
17	end
18
19	# MSG_SYNC (0) + MSG_SYNC_STEP_1 (0) — bypass message_decode NIF, pass sv_payload directly
20	def process_message_v1(server, <<0, 0, sv_payload::binary>>, _origin) do
21	GenServer.call(server, {@sync_step1_raw_call, sv_payload})	11✔
22	end
23
24	def process_message_v1(server, message, origin) do
25	fallback_process_message_v1(server, message, origin)	21✔
26	end
27
28	defp fallback_process_message_v1(server, message, origin) do
29	case Yex.Sync.message_decode(message) do	21✔
30	{:ok, message} ->
31	message_v1(server, message, origin)	20✔
32
33	error ->
34	error	1✔
35	end
36	end
37
38	defp message_v1(server, {:sync, {:sync_step1, encoded_state_vector}}, origin) do
UNCOV 39	GenServer.call(server, {__MODULE__, :document_sync_step1, encoded_state_vector, origin})	×
40	end
41
42	defp message_v1(server, {:sync, {message_type, encoded_diff}}, origin)
43	when message_type in [:sync_step2, :sync_update] do
44	GenServer.cast(server, {__MODULE__, :document_update, encoded_diff, origin})	11✔
45	end
46
47	defp message_v1(server, {:awareness, awareness}, origin) do
48	GenServer.cast(server, {__MODULE__, :awareness_update, awareness, origin})	8✔
49	end
50
51	defp message_v1(server, :query_awareness, _origin) do
NEW 52	GenServer.call(server, @query_awareness_call)	×
53	end
54
55	defp message_v1(_server, _message, _origin) do	1✔
56	{:error, :unknown_message}
57	end
58
59	## Callbacks
60
61	@impl true
62	def init(arg) do
63	module = Keyword.fetch!(arg, :module)	37✔
64	option = Keyword.get(arg, :doc_option, nil)	37✔
65	assigns = Keyword.get(arg, :assigns, %{})	37✔
66	doc = if option, do: Doc.with_options(option), else: Doc.new()	37✔
67
68	if function_exported?(module, :handle_update_v1, 4) do	37✔
69	Doc.monitor_update_v1(doc, metadata: __MODULE__)	25✔
70	end
71
72	awareness = setup_awareness(doc, module)	37✔
73
74	module.init(arg, %State{	37✔
75	assigns: assigns,
76	doc: doc,
77	awareness: awareness,
78	module: module
79	})
80	end
81
82	defp setup_awareness(doc, module) do
83	if function_exported?(module, :handle_awareness_change, 4) or	37✔
84	function_exported?(module, :handle_awareness_update, 4) do	24✔
85	case Awareness.new(doc) do	23✔
86	{:ok, awareness} ->
87	monitor_awareness_events(awareness, module)	23✔
88
89	awareness	23✔
90	end
91	end
92	end
93
94	defp monitor_awareness_events(awareness, module) do
95	if function_exported?(module, :handle_awareness_change, 4) do	23✔
96	Awareness.monitor_change(awareness, metadata: __MODULE__)	13✔
97	end
98
99	if function_exported?(module, :handle_awareness_update, 4) do	23✔
100	Awareness.monitor_update(awareness, metadata: __MODULE__)	10✔
101	end
102	end
103
104	@impl true
105	def handle_call(
106	{@sync_step1_raw_call, sv_payload},
107	_from,
108	%{doc: doc, awareness: awareness} = state
109	) do
110	{:reply, Yex.Nif.encode_sync_step1_response_v1(doc, nil, sv_payload, awareness), state}	11✔
111	end
112
113	@impl true
114	def handle_call(
115	@query_awareness_call,
116	_from,
117	%{awareness: nil} = state
118	) do
119	{:reply, {:ok, []}, state}	1✔
120	end
121
122	@impl true
123	def handle_call(
124	@query_awareness_call,
125	_from,
126	%{awareness: awareness} = state
127	) do
128	{:reply, Yex.Nif.encode_awareness_reply_v1(awareness), state}	3✔
129	end
130
131	@impl true
132	def handle_call(
133	{Yex.Doc, :run, fun},
134	_from,
135	state
136	) do
137	{:reply, fun.(), state}	40✔
138	end
139
140	@impl true
141	def handle_call(request, from, %{module: module} = state) do
142	module.handle_call(request, from, state)	29✔
143	end
144
145	@impl true
146	def handle_cast(
147	{__MODULE__, :document_update, update, origin},
148	%{doc: doc} = state
149	) do
150	Yex.Doc.transaction(doc, origin, fn ->	10✔
151	case Yex.apply_update(doc, update) do	10✔
152	:ok ->	9✔
153	:ok
154
155	{:error, reason} ->
156	Logger.log(:warning, inspect(reason))	1✔
157	:ok
158	end
159	end)
160
161	# Process update messages immediately
162	handle_update_v1_immediately(state)	10✔
163	end
164
165	@impl true
166	def handle_cast(	1✔
167	{__MODULE__, :awareness_update, _message, _origin},
168	%{awareness: nil} = state
169	) do
170	# Logger.warning("Received an awareness message, but ignored it because it is not enabled in this module. ")
171	{:noreply, state}
172	end
173
174	@impl true
175	def handle_cast(
176	{__MODULE__, :awareness_update, message, origin},
177	%{awareness: awareness} = state
178	) do
179	Awareness.apply_update(awareness, message, origin)	6✔
180
181	# Process update messages immediately
182	handle_awareness_event_immediately(state)	6✔
183	end
184
185	@impl true
186	def handle_cast(request, %{module: module} = state) do
187	module.handle_cast(request, state)	1✔
188	end
189
190	@impl true
191	def handle_info({:update_v1, update, origin, __MODULE__}, %{module: module, doc: doc} = state) do
192	module.handle_update_v1(doc, update, origin, state)	19✔
193	end
194
195	@impl true
196	def handle_info(
197	{:awareness_change, change, origin, __MODULE__},
198	%{module: module, awareness: awareness} = state
199	) do
200	module.handle_awareness_change(awareness, change, origin, state)	13✔
201	end
202
203	@impl true
204	def handle_info(
205	{:awareness_update, change, origin, __MODULE__},
206	%{module: module, awareness: awareness} = state
207	) do
208	module.handle_awareness_update(awareness, change, origin, state)	4✔
209	end
210
211	@impl true
212	def handle_info(msg, %{module: module} = state) do
213	module.handle_info(msg, state)	9✔
214	end
215
216	@impl true
217	def terminate(reason, %{module: module} = state) do
218	if function_exported?(module, :terminate, 2) do	7✔
219	module.terminate(reason, state)	6✔
220	else
221	:ok
222	end
223	end
224
225	defp handle_update_v1_immediately(%{doc: doc, module: module} = state) do
226	receive do	17✔
227	{:update_v1, update, origin, __MODULE__} ->
228	case module.handle_update_v1(doc, update, origin, state) do	7✔
229	{:noreply, state} ->
230	handle_update_v1_immediately(state)	7✔
231
232	result ->
233	result	×
234	end
235	after
236	0 ->	10✔
237	{:noreply, state}
238	end
239	end
240
241	defp handle_awareness_event_immediately(%{awareness: awareness, module: module} = state) do
242	receive do	12✔
243	{:awareness_change, change, origin, __MODULE__} ->
244	case module.handle_awareness_change(awareness, change, origin, state) do	1✔
245	{:noreply, state} ->
246	handle_awareness_event_immediately(state)	1✔
247
248	result ->
249	result	×
250	end
251
252	{:awareness_update, change, origin, __MODULE__} ->
253	case module.handle_awareness_update(awareness, change, origin, state) do	5✔
254	{:noreply, state} ->
255	handle_awareness_event_immediately(state)	5✔
256
257	result ->
258	result	×
259	end
260	after
261	0 ->	6✔
262	{:noreply, state}
263	end
264	end
265	end

satoren / y_ex / d7e7f9b43cc2aa584a6fa1810dd59222d9a5a9e4-PR-224

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous