subprojects: Prepare for release

Switch to a fixed-slot NDP layout to eliminate the slow O(k²) "shrink
until it fits" algorithm. The new packer runs in O(k) time and stops
adding datagrams when the NTB max size would be exceeded.

This greatly reduces CPU time spent building NTBs, dropping total layout
time from ~1.1s to negligible levels for a 256 MiB transfer on a modern
desktop.

meson.build

@@ -561,9 +561,6 @@ if have_fruity_backend
     'dns=disabled',
     'arp=disabled',
     'ethernet=enabled',
-    'tcp_mss=4036',
-    'tcp_snd_buf=65535',
-    'tcp_wnd=65535',
   ]
   if optimize_for_prod
     lwip_options += 'lwip_debug=disabled'

src/fruity/ncm.vala

@@ -31,9 +31,18 @@ namespace Frida.Fruity {
 		private uint16 ntb_out_max_datagrams;
 		private size_t max_in_transfers;
 		private size_t max_out_transfers;
+		private size_t max_out_bytes_in_flight;
+		private uint16 ndp_reserved_slots;
+		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 bool writing = false;
+		private size_t out_in_flight_count = 0;
+		private size_t out_in_flight_bytes = 0;
+		private Gee.Map<Future<uint>, uint16> out_in_flight_sizes = new Gee.HashMap<Future<uint>, uint16> ();
+		private bool out_topup_scheduled = false;
 		private uint16 next_outgoing_sequence = 1;
 
 		private InetAddress? _remote_ipv6_address;
@@ -115,7 +124,7 @@ namespace Frida.Fruity {
 			ndp_out_divisor = ntb_params.read_uint16 (20);
 			ndp_out_payload_remainder = ntb_params.read_uint16 (22);
 			ndp_out_alignment = ntb_params.read_uint16 (24);
-			ntb_out_max_datagrams = uint16.min (ntb_params.read_uint16 (26), 16);
+			ntb_out_max_datagrams = ntb_params.read_uint16 (26);
 
 			if (ntb_in_max_size != device_ntb_in_max_size) {
 				var ntb_size_buf = new BufferBuilder (LITTLE_ENDIAN)
@@ -137,17 +146,24 @@ namespace Frida.Fruity {
 				max_out_transfers = (5 * MAX_TRANSFER_MEMORY) / ntb_out_max_size;
 			} else if (speed == LibUSB.Speed.HIGH) {
 				max_in_transfers = MAX_TRANSFER_MEMORY / ntb_in_max_size;
-				max_out_transfers = MAX_TRANSFER_MEMORY / ntb_out_max_size;
+				max_out_transfers = (3 * MAX_TRANSFER_MEMORY) / ntb_out_max_size;
+				max_out_transfers = size_t.max (max_out_transfers, 6);
+				max_out_transfers = size_t.min (max_out_transfers, 16);
 			} else {
 				max_in_transfers = 4;
 				max_out_transfers = 4;
 			}
 
+			max_out_bytes_in_flight = 4 * MAX_TRANSFER_MEMORY;
+
+			ndp_reserved_slots = (uint16) uint16.min (21, ntb_out_max_datagrams);
+			ndp_fixed_header_size = (uint16) align (8 + (ndp_reserved_slots * 4) + 4, ndp_out_alignment);
+
 			Usb.check (handle.set_interface_alt_setting (config.data_iface, config.data_altsetting),
 				"Failed to set USB interface alt setting");
 
 			_netstack = new VirtualNetworkStack (new Bytes (mac_address), null, 1500);
-			_netstack.outgoing_datagram.connect (on_netif_outgoing_datagram);
+			_netstack.outgoing_datagrams.connect (on_netif_outgoing_datagrams);
 
 			process_incoming_datagrams.begin ();
 
@@ -234,63 +250,96 @@ namespace Frida.Fruity {
 							notify_property ("remote-ipv6-address");
 					}
 
-					_netstack.handle_incoming_datagram (datagram);
+					on_incoming_datagram (datagram);
 				}
 
 				ndp_index = next_ndp_index;
 			} while (ndp_index != 0);
 		}
 
-		private void on_netif_outgoing_datagram (Bytes datagram) {
-			pending_output.offer (datagram);
+		private void on_incoming_datagram (Bytes datagram) {
+			pending_input.offer (datagram);
 
-			if (!writing) {
-				writing = true;
+			if (!input_flush_scheduled) {
 				var source = new IdleSource ();
 				source.set_callback (() => {
-					process_pending_output.begin ();
-					return false;
+					input_flush_scheduled = false;
+					flush_pending_input ();
+					return Source.REMOVE;
 				});
 				source.attach (MainContext.get_thread_default ());
+				input_flush_scheduled = true;
 			}
 		}
 
-		private async void process_pending_output () {
+		private void flush_pending_input () {
+			var datagrams = pending_input;
+			pending_input = new Gee.ArrayQueue<Bytes> ();
 			try {
-				while (!pending_output.is_empty) {
-					var pending = new Gee.ArrayList<Promise<uint>> ();
-
-					do {
-						var request = transfer_next_output_batch ();
-						pending.add (request);
-					} while (pending.size < max_out_transfers && !pending_output.is_empty);
-
-					foreach (var request in pending)
-						yield request.future.wait_async (io_cancellable);
-				}
-			} catch (GLib.Error e) {
-			} finally {
-				writing = false;
+				_netstack.handle_incoming_datagrams (datagrams);
+			} catch (Error e) {
 			}
 		}
 
-		private Promise<uint> transfer_next_output_batch () {
-			size_t num_datagrams = ntb_out_max_datagrams;
-			TransferLayout layout;
-			while ((layout = TransferLayout.compute (pending_output, num_datagrams, ndp_out_alignment,
-					ndp_out_divisor, ndp_out_payload_remainder)).size > ntb_out_max_size) {
-				num_datagrams--;
+		private void on_netif_outgoing_datagrams (Gee.Collection<Bytes> datagrams) {
+			foreach (var d in datagrams)
+				pending_output.offer (d);
+
+			schedule_top_up ();
+		}
+
+		private void schedule_top_up () {
+			if (out_topup_scheduled)
+				return;
+			out_topup_scheduled = true;
+
+			var source = new IdleSource ();
+			source.set_callback (() => {
+				out_topup_scheduled = false;
+				top_up_output_window ();
+				return Source.REMOVE;
+			});
+			source.attach (MainContext.get_thread_default ());
+		}
+
+		private void top_up_output_window () {
+			while (!pending_output.is_empty && out_in_flight_count < max_out_transfers) {
+				size_t max_datagrams = size_t.min (pending_output.size, (size_t) ndp_reserved_slots);
+				var layout = TransferLayout.compute (pending_output, max_datagrams, ntb_out_max_size, ndp_fixed_header_size,
+					ndp_out_alignment, ndp_out_divisor, ndp_out_payload_remainder);
+				if ((out_in_flight_bytes + layout.size) > max_out_bytes_in_flight)
+					break;
+
+				var batch = new Gee.ArrayList<Bytes> ();
+				for (var i = 0; i != layout.offsets.length; i++)
+					batch.add (pending_output.poll ());
+				var transfer = build_output_transfer (batch, layout, next_outgoing_sequence++);
+
+				var request = new Promise<uint> ();
+				do_transfer_output_batch.begin (transfer, request);
+
+				out_in_flight_count++;
+				out_in_flight_bytes += layout.size;
+
+				out_in_flight_sizes.set (request.future, layout.size);
+
+				request.future.then (fut => {
+					on_output_completed (fut);
+				});
 			}
+		}
 
-			var batch = new Gee.ArrayList<Bytes> ();
-			for (var i = 0; i != layout.offsets.size; i++)
-				batch.add (pending_output.poll ());
+		private void on_output_completed (Future<uint> fut) {
+			uint16 size;
+			out_in_flight_sizes.unset (fut, out size);
 
-			var transfer = build_output_transfer (batch, layout, next_outgoing_sequence++);
+			out_in_flight_count--;
+			out_in_flight_bytes -= size;
 
-			var request = new Promise<uint> ();
-			do_transfer_output_batch.begin (transfer, request);
-			return request;
+			if (fut.error != null)
+				return;
+
+			top_up_output_window ();
 		}
 
 		private async void do_transfer_output_batch (Bytes transfer, Promise<uint> request) {
@@ -306,53 +355,45 @@ namespace Frida.Fruity {
 			public uint16 size;
 			public uint16 ndp_header_offset;
 			public uint16 ndp_header_size;
-			public Gee.List<uint16> offsets;
+			public uint16[] offsets;
 
-			public static TransferLayout compute (Gee.Collection<Bytes> datagrams, size_t max_datagrams, size_t ndp_alignment,
+			public static TransferLayout compute (Gee.Collection<Bytes> datagrams, size_t max_datagrams,
+					uint32 max_transfer_size, uint16 ndp_fixed_header_size, size_t ndp_alignment,
 					size_t datagram_modulus, size_t datagram_remainder) {
-				size_t ndp_header_base_size = 4 + 2 + 2;
-				size_t ndp_entry_size = 2 + 2;
 				size_t ethernet_header_size = 14;
 
 				size_t ndp_header_offset = align (TRANSFER_HEADER_SIZE, ndp_alignment, 0);
-				size_t num_datagram_slots = size_t.min (datagrams.size, max_datagrams);
-				size_t ndp_header_size = ndp_header_base_size + ((num_datagram_slots + 1) * ndp_entry_size);
 
-				size_t current_transfer_size = ndp_header_offset + ndp_header_size;
-				var offsets = new Gee.ArrayList<uint16> ();
+				size_t current_transfer_size = ndp_header_offset + ndp_fixed_header_size;
+				var offsets = new uint16[max_datagrams];
 
-				uint i = 0;
+				int i = 0;
 				foreach (var datagram in datagrams) {
 					var size = (uint16) datagram.get_size ();
 
 					size_t start_offset =
 						align (current_transfer_size + ethernet_header_size, datagram_modulus, datagram_remainder);
 					size_t end_offset = start_offset + size;
-					if (end_offset > uint16.MAX)
+					if (end_offset > uint16.MAX || end_offset > max_transfer_size)
 						break;
 
 					current_transfer_size = end_offset;
-					offsets.add ((uint16) start_offset);
+					offsets[i] = (uint16) start_offset;
 
 					i++;
 					if (i == max_datagrams)
 						break;
 				}
 
+				offsets.resize (i);
+
 				return new TransferLayout () {
 					size = (uint16) current_transfer_size,
 					ndp_header_offset = (uint16) ndp_header_offset,
-					ndp_header_size = (uint16) ndp_header_size,
-					offsets = offsets,
+					ndp_header_size = ndp_fixed_header_size,
+					offsets = (owned) offsets,
 				};
 			}
-
-			private static size_t align (size_t val, size_t modulus, size_t remainder) {
-				var delta = val % modulus;
-				if (delta != remainder)
-					return val + modulus - delta + remainder;
-				return val;
-			}
 		}
 
 		private static Bytes build_output_transfer (Gee.List<Bytes> datagrams, TransferLayout layout, uint16 sequence_number) {
@@ -409,6 +450,13 @@ namespace Frida.Fruity {
 
 			return new InetAddress.from_bytes (datagram[22:22 + 16].get_data (), IPV6);
 		}
+
+		private static size_t align (size_t val, size_t modulus, size_t remainder = 0) {
+			var delta = val % modulus;
+			if (delta != remainder)
+				return val + modulus - delta + remainder;
+			return val;
+		}
 	}
 
 	internal sealed class UsbNcmConfig {

src/fruity/network-stack.vala

@@ -149,7 +149,7 @@ namespace Frida.Fruity {
 	}
 
 	public sealed class VirtualNetworkStack : Object, NetworkStack {
-		public signal void outgoing_datagram (Bytes datagram);
+		public signal void outgoing_datagrams (Gee.Collection<Bytes> datagrams);
 
 		public Bytes? ethernet_address {
 			get;
@@ -182,6 +182,8 @@ namespace Frida.Fruity {
 
 		private State state = STARTED;
 
+		private Gee.Queue<Bytes>? pending_outgoing;
+
 		private Gee.Queue<Request> requests = new Gee.ArrayQueue<Request> ();
 		private unowned Thread<bool>? lwip_thread;
 
@@ -281,16 +283,26 @@ namespace Frida.Fruity {
 			return new Ipv6UdpSocket (this);
 		}
 
-		public void handle_incoming_datagram (Bytes datagram) throws Error {
+		public void handle_incoming_datagrams (Gee.Collection<Bytes> datagrams) throws Error {
 			check_started ();
 
 			check (perform_on_lwip_thread (() => {
-				var pbuf = LWIP.PacketBuffer.alloc (RAW, (uint16) datagram.get_size (), POOL);
-				pbuf.take (datagram.get_data ());
+				begin_output ();
 
-				var err = handle.input (pbuf, ref handle);
-				if (err == OK)
-					*((void **) &pbuf) = null;
+				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 ());
+
+					err = handle.input (pbuf, ref handle);
+					if (err == OK)
+						*((void **) &pbuf) = null;
+					else
+						break;
+				}
+
+				end_output ();
 
 				return err;
 			}));
@@ -309,14 +321,39 @@ namespace Frida.Fruity {
 			return OK;
 		}
 
+		internal void begin_output () {
+			pending_outgoing = new Gee.ArrayQueue<Bytes> ();
+		}
+
+		internal void end_output () {
+			var datagrams = pending_outgoing;
+			pending_outgoing = null;
+			if (datagrams.is_empty)
+				return;
+
+			schedule_on_frida_thread (() => {
+				if (state == STARTED)
+					outgoing_datagrams (datagrams);
+				return Source.REMOVE;
+			});
+		}
+
 		private void emit_datagram (LWIP.PacketBuffer pbuf) {
 			var buffer = new uint8[pbuf.tot_len];
 			unowned uint8[] packet = pbuf.get_contiguous (buffer, pbuf.tot_len);
 			var datagram = new Bytes (packet[:pbuf.tot_len]);
 
+			if (pending_outgoing != null) {
+				pending_outgoing.offer (datagram);
+				return;
+			}
+
 			schedule_on_frida_thread (() => {
-				if (state == STARTED)
-					outgoing_datagram (datagram);
+				if (state == STARTED) {
+					var datagrams = new Gee.ArrayQueue<Bytes> ();
+					datagrams.offer (datagram);
+					outgoing_datagrams (datagrams);
+				}
 				return Source.REMOVE;
 			});
 		}
@@ -713,7 +750,10 @@ namespace Frida.Fruity {
 						}
 					}
 
+					netstack.begin_output ();
 					var output_err = pcb.output ();
+					netstack.end_output ();
+
 					if (output_err == RTE)
 						return output_err;
 

src/fruity/xpc.vala

@@ -2044,7 +2044,9 @@ namespace Frida.Fruity {
 		private BufferedInputStream input;
 		private OutputStream output;
 
-		private ByteArray pending_output = new ByteArray ();
+		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;
 
 		private Cancellable io_cancellable = new Cancellable ();
@@ -2125,7 +2127,7 @@ namespace Frida.Fruity {
 			tunnel_params = TunnelParameters.from_json (yield read_message (cancellable));
 
 			_tunnel_netstack = new VirtualNetworkStack (null, tunnel_params.address, tunnel_params.mtu);
-			_tunnel_netstack.outgoing_datagram.connect (post);
+			_tunnel_netstack.outgoing_datagrams.connect (post_batch);
 
 			process_incoming_messages.begin ();
 
@@ -2146,8 +2148,7 @@ namespace Frida.Fruity {
 			try {
 				while (true) {
 					var datagram = yield read_datagram (io_cancellable);
-
-					_tunnel_netstack.handle_incoming_datagram (datagram);
+					on_incoming_datagram (datagram);
 				}
 			} catch (GLib.Error e) {
 			}
@@ -2172,28 +2173,61 @@ namespace Frida.Fruity {
 			closed ();
 		}
 
-		private void post (Bytes bytes) {
-			pending_output.append (bytes.get_data ());
+		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);
+			post_batch (batch);
+		}
+
+		private void post_batch (Gee.Collection<Bytes> datagrams) {
+			foreach (var d in datagrams)
+				pending_output.offer (d);
 
 			if (!writing) {
 				writing = true;
-
-				var source = new IdleSource ();
-				source.set_callback (() => {
-					process_pending_output.begin ();
-					return false;
-				});
-				source.attach (MainContext.get_thread_default ());
+				process_pending_output.begin ();
 			}
 		}
 
 		private async void process_pending_output () {
-			while (pending_output.len > 0) {
-				uint8[] batch = pending_output.steal ();
+			while (!pending_output.is_empty) {
+				size_t size = 0;
+				foreach (var d in pending_output)
+					size += d.get_size ();
+
+				var batch = new ByteArray.sized ((uint) size);
+				foreach (var d in pending_output)
+					batch.append (d.get_data ());
+
+				pending_output = new Gee.ArrayQueue<Bytes> ();
 
 				size_t bytes_written;
 				try {
-					yield output.write_all_async (batch, Priority.DEFAULT, io_cancellable, out bytes_written);
+					yield output.write_all_async (batch.data, Priority.DEFAULT, io_cancellable, out bytes_written);
 				} catch (GLib.Error e) {
 					break;
 				}
@@ -2331,7 +2365,9 @@ namespace Frida.Fruity {
 		private VirtualNetworkStack? _tunnel_netstack;
 
 		private Gee.Map<int64?, Stream> streams = new Gee.HashMap<int64?, Stream> (Numeric.int64_hash, Numeric.int64_equal);
-		private Gee.Queue<Bytes> tx_datagrams = new Gee.ArrayQueue<Bytes> ();
+		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 DatagramBasedSource? rx_source;
 		private uint8[] rx_buf = new uint8[MAX_UDP_PAYLOAD_SIZE];
@@ -2511,7 +2547,7 @@ namespace Frida.Fruity {
 			tunnel_params = TunnelParameters.from_json (new JsonObjectReader (json));
 
 			_tunnel_netstack = new VirtualNetworkStack (null, tunnel_params.address, tunnel_params.mtu);
-			_tunnel_netstack.outgoing_datagram.connect (send_datagram);
+			_tunnel_netstack.outgoing_datagrams.connect (send_datagrams);
 
 			establish_request.resolve (true);
 		}
@@ -2602,7 +2638,14 @@ namespace Frida.Fruity {
 		}
 
 		private void send_datagram (Bytes datagram) {
-			tx_datagrams.offer (datagram);
+			var batch = new Gee.ArrayList<Bytes> ();
+			batch.add (datagram);
+			send_datagrams (batch);
+		}
+
+		private void send_datagrams (Gee.Collection<Bytes> datagrams) {
+			foreach (var d in datagrams)
+				pending_output.offer (d);
 			process_pending_writes ();
 		}
 
@@ -2659,13 +2702,13 @@ namespace Frida.Fruity {
 					n = connection.write_connection_close (null, &pi, tx_buf, error, ts);
 					state = CLOSE_WRITTEN;
 				} else {
-					Bytes? datagram = tx_datagrams.peek ();
+					Bytes? datagram = pending_output.peek ();
 					if (datagram != null) {
 						int accepted = -1;
 						n = connection.write_datagram (null, null, tx_buf, &accepted, NGTcp2.WriteDatagramFlags.MORE, 0,
 							datagram.get_data (), ts);
 						if (accepted > 0)
-							tx_datagrams.poll ();
+							pending_output.poll ();
 					} else {
 						Stream? stream = null;
 						unowned uint8[]? data = null;
@@ -2793,14 +2836,31 @@ namespace Frida.Fruity {
 		}
 
 		private int on_recv_datagram (uint32 flags, uint8[] data) {
-			try {
-				_tunnel_netstack.handle_incoming_datagram (new Bytes (data));
-			} catch (Error e) {
+			pending_input.offer (new Bytes (data));
+
+			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;
 			}
 
 			return 0;
 		}
 
+		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 static void on_rand (uint8[] dest, NGTcp2.RNGContext rand_ctx) {
 			OpenSSL.Rng.generate (dest);
 		}

subprojects/frida-gum.wrap

@@ -1,6 +1,6 @@
 [wrap-git]
 url = https://github.com/frida/frida-gum.git
-revision = 17.3.0
+revision = 17.3.2
 depth = 1
 
 [provide]

subprojects/lwip.wrap

@@ -1,6 +1,6 @@
 [wrap-git]
 url = https://github.com/frida/lwip.git
-revision = c5e2835a213436eaf0c40d0a95a35e7cf0dad2d3
+revision = 4fe7223c2e80dc328266fb2483e789ee5fad7c79
 depth = 1
 
 [provide]

subprojects/openssl.wrap

@@ -1,6 +1,6 @@
 [wrap-git]
 url = https://github.com/frida/openssl.git
-revision = 7b86cb6a0c5cb9d79dca012c98a0a30a58eef5b5
+revision = 5029c4562d23547d8e29768e1b0de573c52bf3fc
 depth = 1
 
 [provide]

tests/test-host-session.vala

@@ -67,6 +67,11 @@ namespace Frida.HostSessionTest {
 			h.run ();
 		});
 
+		GLib.Test.add_func ("/HostSession/Fruity/Manual/speedtest", () => {
+			var h = new Harness ((h) => Fruity.Manual.speedtest.begin (h as Harness));
+			h.run ();
+		});
+
 		GLib.Test.add_func ("/HostSession/Fruity/Manual/Xpc/list", () => {
 			var h = new Harness ((h) => Fruity.Manual.Xpc.list.begin (h as Harness));
 			h.run ();
@@ -3813,6 +3818,77 @@ namespace Frida.HostSessionTest {
 				h.done ();
 			}
 
+			private static async void speedtest (Harness h) {
+				/*
+				 * Configured like this:
+				 *
+				 * ./configure \
+				 *   --disable-local-backend \
+				 *   --disable-droidy-backend \
+				 *   --disable-socket-backend \
+				 *   --disable-barebone-backend \
+				 *   --disable-compiler-backend \
+				 *   --with-compat=disabled \
+				 *   --disable-gadget \
+				 *   --disable-server \
+				 *   --disable-portal \
+				 *   --disable-inject \
+				 *   -- \
+				 *   --force-fallback-for=lwip
+				 */
+				if (!GLib.Test.slow ()) {
+					stdout.printf ("<skipping, run in slow mode with iOS device connected> ");
+					h.done ();
+					return;
+				}
+
+				h.disable_timeout ();
+
+				var device_manager = new DeviceManager ();
+
+				try {
+					var device = yield device_manager.get_device_by_id (DEVICE_ID);
+
+					{
+						var stream = yield device.open_channel ("tcp:5050");
+
+						var jumbo_message = new uint8[64 * 1024 * 1024];
+						yield stream.get_output_stream ().write_all_async (jumbo_message, Priority.DEFAULT, null,
+							null);
+
+						yield stream.close_async ();
+					}
+
+					{
+						var stream = yield device.open_channel ("tcp:5051");
+
+						var buf = new uint8[4 * 1024 * 1024];
+						while (true) {
+							var n = yield stream.get_input_stream ().read_async (buf);
+							if (n == 0)
+								break;
+						}
+
+						yield stream.close_async ();
+					}
+
+					var source = new TimeoutSource (500);
+					source.set_callback (speedtest.callback);
+					source.attach (MainContext.get_thread_default ());
+					yield;
+				} catch (GLib.Error e) {
+					printerr ("\nFAIL: %s\n\n", e.message);
+				}
+
+				try {
+					yield device_manager.close ();
+				} catch (IOError e) {
+					assert_not_reached ();
+				}
+
+				h.done ();
+			}
+
 			namespace Xpc {
 
 				private static async void list (Harness h) {