8000 fruity: Fix dbus starvation and backpressure by mrmacete · Pull Request #1211 · frida/frida-core · GitHub
[go: up one dir, main page]
Skip to content

fruity: Fix dbus starvation and backpressure #1211

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
oleavr merged 2 commits into from
Feb 18, 2026
Merged

fruity: Fix dbus starvation and backpressure #1211

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e7253f0
network-stack: Handle incoming datagrams async
mrmacete Feb 6, 2026
c95ccbe
websocket: Pause libsoup input to avoid starvation
mrmacete Feb 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8000
4 changes: 4 additions & 0 deletions lib/base/socket.vala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -940,6 +940,9 @@ namespace Frida {
Memory.copy (buffer, recv_queue.data, n);
recv_queue.remove_range (0, (uint) n);

if (recv_queue.len == 0)
websocket.resume_input ();

update_pending_io ();
} else {
if (state == OPEN)
Expand Down Expand Up @@ -1006,6 +1009,7 @@ namespace Frida {
private void on_message (int type, Bytes message) {
with_state_lock (() => {
recv_queue.append (message.get_data ());
websocket.pause_input ();
update_pending_io ();
});
}
Expand Down
48 changes: 19 additions & 29 deletions src/fruity/ncm.vala
View file
Open in desktop
8000
Original file line number Diff line n 8000 umber Diff line change
Expand Up @@ -36,8 +36,6 @@ namespace Frida.Fruity {
private uint16 ndp_fixed_header_size;

private VirtualNetworkStack? _netstack;
private Gee.Queue<Bytes> pending_input = new Gee.ArrayQueue<Bytes> ();
private bool input_flush_scheduled = false;
private Gee.Queue<Bytes> pending_output = new Gee.ArrayQueue<Bytes> ();
private size_t out_in_flight_count = 0;
private size_t out_in_flight_bytes = 0;
Expand Down Expand Up @@ -184,8 +182,20 @@ namespace Frida.Fruity {
}

try {
var frame = yield pending.poll ().future.wait_async (io_cancellable);
handle_ncm_frame (frame);
var datagrams = new Gee.ArrayList<Bytes> ();
while (true) {
var request = pending.peek ();

if (request == null)
break;
if (!datagrams.is_empty && !request.future.ready)
break;

var frame = yield pending.poll ().future.wait_async (io_cancellable);
datagrams.add_all (handle_ncm_frame (frame));
}

yield _netstack.handle_incoming_datagrams (datagrams);
} catch (GLib.Error e) {
return;
}
Expand All @@ -208,7 +218,7 @@ namespace Frida.Fruity {
}
}

private void handle_ncm_frame (Bytes frame) throws GLib.Error {
private Gee.List<Bytes> handle_ncm_frame (Bytes frame) throws GLib.Error {
var input = new DataInputStream (new MemoryInputStream.from_bytes (frame));
input.byte_order = LITTLE_ENDIAN;

Expand All @@ -222,6 +232,8 @@ namespace Frida.Fruity {
input.skip (6);
var ndp_index = input.read_uint16 ();

var datagrams = new Gee.ArrayList<Bytes> ();

do {
input.seek (ndp_index, SET);
input.read_all (raw_signature[:4], out bytes_read);
Expand Down Expand Up @@ -250,35 +262,13 @@ namespace Frida.Fruity {
notify_property ("remote-ipv6-address");
}

on_incoming_datagram (datagram);
datagrams.add (datagram);
}

ndp_index = next_ndp_index;
} while (ndp_index != 0);
}

private void on_incoming_datagram (Bytes datagram) {
pending_input.offer (datagram);

if (!input_flush_scheduled) {
var source = new IdleSource ();
source.set_callback (() => {
input_flush_scheduled = false;
flush_pending_input ();
return Source.REMOVE;
});
source.attach (MainContext.get_thread_default ());
input_flush_scheduled = true;
}
}

private void flush_pending_input () {
var datagrams = pending_input;
pending_input = new Gee.ArrayQueue<Bytes> ();
try {
_netstack.handle_incoming_datagrams (datagrams);
} catch (Error e) {
}
return datagrams;
}

private void on_netif_outgoing_datagrams (Gee.Collection<Bytes> datagrams) {
Expand Down
27 changes: 21 additions & 6 deletions src/fruity/network-stack.vala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -283,14 +283,14 @@ namespace Frida.Fruity {
return new Ipv6UdpSocket (this);
}

public void handle_incoming_datagrams (Gee.Collection<Bytes> datagrams) throws Error {
public async void handle_incoming_datagrams (Gee.Collection<Bytes> datagrams) throws Error {
LWIP.ErrorCode err = OK;

check_started ();

check (perform_on_lwip_thread (() => {
schedule_on_lwip_thread (() => {
begin_output ();

LWIP.ErrorCode err = OK;

foreach (var d in datagrams) {
var pbuf = LWIP.PacketBuffer.alloc (RAW, (uint16) d.get_size (), POOL);
pbuf.take (d.get_data ());
Expand All @@ -304,8 +304,14 @@ namespace Frida.Fruity {

end_output ();

schedule_on_frida_thread (handle_incoming_datagrams.callback);

return err;
}));
});

yield;

check (err);
}

private static LWIP.ErrorCode on_netif_link_output (LWIP.NetworkInterface handle, LWIP.PacketBuffer pbuf) {
Expand Down Expand Up @@ -359,6 +365,15 @@ namespace Frida.Fruity {
}

internal LWIP.ErrorCode perform_on_lwip_thread (owned WorkFunc work) {
var req = submit_work ((owned) work);
return req.join ();
}

internal void schedule_on_lwip_thread (owned WorkFunc work) {
submit_work ((owned) work);
}

private Request submit_work (owned WorkFunc work) {
var req = new Request ((owned) work);

if (Thread.self<bool> () == lwip_thread) {
Expand All @@ -369,7 +384,7 @@ namespace Frida.Fruity {
LWIP.Runtime.schedule (perform_next_request);
}

return req.join ();
return req;
}

private void perform_next_request () {
Expand Down
92 changes: 51 additions & 41 deletions src/fruity/xpc.vala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2044,8 +2044,6 @@ namespace Frida.Fruity {
private BufferedInputStream input;
private OutputStream output;

private Gee.Queue<Bytes> pending_input = new Gee.ArrayQueue<Bytes> ();
private bool input_flush_scheduled = false;
private Gee.Queue<Bytes> pending_output = new Gee.ArrayQueue<Bytes> ();
private bool writing = false;

Expand Down Expand Up @@ -2147,8 +2145,39 @@ namespace Frida.Fruity {
private async void process_incoming_messages () {
try {
while (true) {
var datagram = yield read_datagram (io_cancellable);
on_incoming_datagram (datagram);
var pending_input = new Gee.ArrayQueue<Bytes> ();
while (true) {
var size = peek_datagram_size ();
if (size == -1)
break;

var datagram = new uint8[size];
input.read (datagram, io_cancellable);

pending_input.offer (new Bytes.take ((owned) datagram));
}

if (pending_input.size != 0) {
try {
yield _tunnel_netstack.handle_incoming_datagrams (pending_input);
} catch (Error e) {
}

var source = new IdleSource ();
source.set_callback (process_incoming_messages.callback);
source.attach (MainContext.get_thread_default ());
yield;
}

ssize_t n;
try {
n = yield input.fill_async (-1, Priority.DEFAULT, io_cancellable);
} catch (GLib.Error e) {
throw new Error.TRANSPORT ("Connection closed");
}

if (n == 0)
throw new Error.TRANSPORT ("Connection closed");
}
} catch (GLib.Error e) {
}
Expand All @@ -2173,30 +2202,6 @@ namespace Frida.Fruity {
closed ();
}

private void on_incoming_datagram (Bytes datagram) {
pending_input.offer (datagram);

if (!input_flush_scheduled) {
var source = new IdleSource ();
source.set_callback (() => {
input_flush_scheduled = false;
flush_pending_input ();
return Source.REMOVE;
});
source.attach (MainContext.get_thread_default ());
input_flush_scheduled = true;
}
}

private void flush_pending_input () {
var datagrams = pending_input;
pending_input = new Gee.ArrayQueue<Bytes> ();
try {
_tunnel_netstack.handle_incoming_datagrams (datagrams);
} catch (Error e) {
}
}

private void post (Bytes datagram) {
var batch = new Gee.ArrayList<Bytes> ();
batch.add (datagram);
Expand Down Expand Up @@ -2269,24 +2274,21 @@ namespace Frida.Fruity {
return new JsonObjectReader (json);
}

private async Bytes read_datagram (Cancellable? cancellable) throws Error, IOError {
size_t header_size = 40;
private ssize_t peek_datagram_size () {
ssize_t header_size = 40;
if (input.get_available () < header_size)
yield fill_until_n_bytes_available (header_size, cancellable);
return -1;

uint16 payload_size = 0;
unowned uint8[] size_buf = ((uint8[]) &payload_size)[:2];
input.peek (size_buf, 4);
payload_size = uint16.from_big_endian (payload_size);

size_t full_size = header_size + payload_size;
ssize_t full_size = header_size + payload_size;
if (input.get_available () < full_size)
yield fill_until_n_bytes_available (full_size, cancellable);
return -1;

var datagram = new uint8[full_size];
input.read (datagram, cancellable);

return new Bytes.take ((owned) datagram);
return full_size;
}

private async void fill_until_n_bytes_available (size_t minimum, Cancellable? cancellable) throws Error, IOError {
Expand Down Expand Up @@ -2367,6 +2369,7 @@ namespace Frida.Fruity {
private Gee.Map<int64?, Stream> streams = new Gee.HashMap<int64?, Stream> (Numeric.int64_hash, Numeric.int64_equal);
private Gee.Queue<Bytes> pending_input = new Gee.ArrayQueue<Bytes> ();
private bool input_flush_scheduled = false;
private bool input_flush_happening = false;
private Gee.Queue<Bytes> pending_output = new Gee.ArrayQueue<Bytes> ();

private DatagramBasedSource? rx_source;
Expand Down Expand Up @@ -2842,7 +2845,7 @@ namespace Frida.Fruity {
var source = new IdleSource ();
source.set_callback (() => {
input_flush_scheduled = false;
flush_pending_input ();
flush_pending_input.begin ();
return Source.REMOVE;
});
source.attach (MainContext.get_thread_default ());
Expand All @@ -2852,13 +2855,20 @@ namespace Frida.Fruity {
return 0;
}

private void flush_pending_input () {
private async void flush_pending_input () {
if (input_flush_happening)
return;
input_flush_happening = true;

var datagrams = pending_input;
pending_input = new Gee.ArrayQueue<Bytes> ();

try {
_tunnel_netstack.handle_incoming_datagrams (datagrams);
} catch (Error e) {
yield _tunnel_netstack.handle_incoming_datagrams (datagrams);
} catch (GLib.Error e) {
}

input_flush_happening = false;
}

private static void on_rand (uint8[] dest, NGTcp2.RNGContext rand_ctx) {
Expand Down
Loading
0