b0ddcb4e43763f9494489c9a05635f8b83b75cbb-PR-224

Committed 08 Apr 2026 12:14PM UTC coverage: 98.51% (-0.02%) from 98.532%

Build # b0ddcb4e43763f9494489c9a05635f8b83b75cbb-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

7 of 9 new or added lines in 2 files covered. (77.78%)

1 existing line in 1 file now uncovered.

595 of 604 relevant lines covered (98.51%)

25.97 hits per line

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

90.16

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

satoren / y_ex / b0ddcb4e43763f9494489c9a05635f8b83b75cbb-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