nbio: add package

5 files changed, 1006 insertions, 0 deletions

diff --git a/tests/core/nbio/fs.odin b/tests/core/nbio/fs.odin
new file mode 100644
index 000000000..6e079f96e
--- /dev/null
+++ b/tests/core/nbio/fs.odin
@@ -0,0 +1,100 @@
+package tests_nbio
+
+import    "core:nbio"
+import    "core:testing"
+import    "core:time"
+import os "core:os/os2"
+
+@(test)
+close_invalid_handle :: proc(t: ^testing.T) {
+	if event_loop_guard(t) {
+		testing.set_fail_timeout(t, time.Minute)
+
+		nbio.close(max(nbio.Handle))
+
+		ev(t, nbio.run(), nil)
+	}
+}
+
+@(test)
+write_read_close :: proc(t: ^testing.T) {
+	if event_loop_guard(t) {
+		testing.set_fail_timeout(t, time.Minute)
+
+		@static content := [20]byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20}
+		@static result: [20]byte
+
+		FILENAME :: "test_write_read_close"
+
+		nbio.open_poly(FILENAME, t, on_open, mode={.Read, .Write, .Create, .Trunc})
+
+		on_open :: proc(op: ^nbio.Operation, t: ^testing.T) {
+			ev(t, op.open.err, nil)
+
+			nbio.write_poly(op.open.handle, 0, content[:], t, on_write)
+		}
+
+		on_write :: proc(op: ^nbio.Operation, t: ^testing.T) {
+			ev(t, op.write.err, nil)
+			ev(t, op.write.written, len(content))
+
+			nbio.read_poly(op.write.handle, 0, result[:], t, on_read, all=true)
+		}
+
+		on_read :: proc(op: ^nbio.Operation, t: ^testing.T) {
+			ev(t, op.read.err, nil)
+			ev(t, op.read.read, len(result))
+			ev(t, result, content)
+
+			nbio.close_poly(op.read.handle, t, on_close)
+		}
+
+		on_close :: proc(op: ^nbio.Operation, t: ^testing.T) {
+			ev(t, op.close.err, nil)
+			os.remove(FILENAME)
+		}
+
+		ev(t, nbio.run(), nil)
+	}
+}
+
+@(test)
+read_empty_file :: proc(t: ^testing.T) {
+	if event_loop_guard(t) {
+		testing.set_fail_timeout(t, time.Minute)
+
+		FILENAME :: "test_read_empty_file"
+
+		handle, err := nbio.open_sync(FILENAME, mode={.Read, .Write, .Create, .Trunc})
+		ev(t, err, nil)
+
+		buf: [128]byte
+		nbio.read_poly(handle, 0, buf[:], t, proc(op: ^nbio.Operation, t: ^testing.T) {
+			ev(t, op.read.err, nbio.FS_Error.EOF)
+			ev(t, op.read.read, 0)
+
+			nbio.close_poly(op.read.handle, t, proc(op: ^nbio.Operation, t: ^testing.T) {
+				ev(t, op.close.err, nil)
+				os.remove(FILENAME)
+			})
+		})
+
+		ev(t, nbio.run(), nil)
+	}
+}
+
+@(test)
+read_entire_file :: proc(t: ^testing.T) {
+	if event_loop_guard(t) {
+		testing.set_fail_timeout(t, time.Minute)
+
+		nbio.read_entire_file(#file, t, on_read)
+
+		on_read :: proc(t: rawptr, data: []byte, err: nbio.Read_Entire_File_Error) {
+			t := (^testing.T)(t)
+			ev(t, err.value, nil)
+			ev(t, string(data), #load(#file, string))
+			delete(data)
+		}
+	}
+}
diff --git a/tests/core/nbio/nbio.odin b/tests/core/nbio/nbio.odin
new file mode 100644
index 000000000..2f454f55b
--- /dev/null
+++ b/tests/core/nbio/nbio.odin
@@ -0,0 +1,258 @@
+package tests_nbio
+
+import    "core:log"
+import    "core:nbio"
+import    "core:testing"
+import    "core:thread"
+import    "core:time"
+import os "core:os/os2"
+
+ev :: testing.expect_value
+e  :: testing.expect
+
+@(deferred_in=event_loop_guard_exit)
+event_loop_guard :: proc(t: ^testing.T) -> bool {
+	err := nbio.acquire_thread_event_loop()
+	if err == .Unsupported || !nbio.FULLY_SUPPORTED {
+		log.warn("nbio unsupported, skipping")
+		return false
+	}
+
+	ev(t, err, nil)
+	return true
+}
+
+event_loop_guard_exit :: proc(t: ^testing.T) {
+	ev(t, nbio.run(), nil) // Could have some things to clean up from a `defer` in the test.
+	nbio.release_thread_event_loop()
+}
+
+// Tests that all poly variants are correctly passing through arguments, and that
+// all procs eventually get their callback called.
+//
+// This is important because the poly procs are only checked when they are called,
+// So this will also catch any typos in their implementations.
+@(test)
+all_poly_work :: proc(tt: ^testing.T) {
+	if event_loop_guard(tt) {
+		testing.set_fail_timeout(tt, time.Minute)
+
+		@static t: ^testing.T
+		t = tt
+
+		@static n: int
+		n = 0
+		NUM_TESTS :: 39
+
+		UDP_SOCKET :: max(nbio.UDP_Socket)
+		TCP_SOCKET :: max(nbio.TCP_Socket)
+
+		tmp, terr := os.create_temp_file("", "tests_nbio_poly*", {.Non_Blocking})
+		ev(t, terr, nil)
+		defer os.close(tmp)
+
+		HANDLE, aerr := nbio.associate_handle(os.fd(tmp))
+		ev(t, aerr, nil)
+
+		_buf: [1]byte
+		buf := _buf[:]
+
+		one :: proc(op: ^nbio.Operation, one: int) {
+			n += 1
+			ev(t, one, 1)
+		}
+
+		two :: proc(op: ^nbio.Operation, one: int, two: int) {
+			n += 1
+			ev(t, one, 1)
+			ev(t, two, 2)
+		}
+
+		three :: proc(op: ^nbio.Operation, one: int, two: int, three: int) {
+			n += 1
+			ev(t, one, 1)
+			ev(t, two, 2)
+			ev(t, three, 3)
+		}
+
+		nbio.accept_poly(TCP_SOCKET, 1, one)
+		nbio.accept_poly2(TCP_SOCKET, 1, 2, two)
+		nbio.accept_poly3(TCP_SOCKET, 1, 2, 3, three)
+
+		nbio.close_poly(max(nbio.Handle), 1, one)
+		nbio.close_poly2(max(nbio.Handle), 1, 2, two)
+		nbio.close_poly3(max(nbio.Handle), 1, 2, 3, three)
+
+		nbio.dial_poly({nbio.IP4_Address{127, 0, 0, 1}, 0}, 1, one)
+		nbio.dial_poly2({nbio.IP4_Address{127, 0, 0, 1}, 0}, 1, 2, two)
+		nbio.dial_poly3({nbio.IP4_Address{127, 0, 0, 1}, 0}, 1, 2, 3, three)
+
+		nbio.recv_poly(TCP_SOCKET, {buf}, 1, one)
+		nbio.recv_poly2(TCP_SOCKET, {buf}, 1, 2, two)
+		nbio.recv_poly3(TCP_SOCKET, {buf}, 1, 2, 3, three)
+
+		nbio.send_poly(TCP_SOCKET, {buf}, 1, one)
+		nbio.send_poly2(TCP_SOCKET, {buf}, 1, 2, two)
+		nbio.send_poly3(TCP_SOCKET, {buf}, 1, 2, 3, three)
+
+		nbio.sendfile_poly(TCP_SOCKET, HANDLE, 1, one)
+		nbio.sendfile_poly2(TCP_SOCKET, HANDLE, 1, 2, two)
+		nbio.sendfile_poly3(TCP_SOCKET, HANDLE, 1, 2, 3, three)
+
+		nbio.read_poly(HANDLE, 0, buf, 1, one)
+		nbio.read_poly2(HANDLE, 0, buf, 1, 2, two)
+		nbio.read_poly3(HANDLE, 0, buf, 1, 2, 3, three)
+
+		nbio.write_poly(HANDLE, 0, buf, 1, one)
+		nbio.write_poly2(HANDLE, 0, buf, 1, 2, two)
+		nbio.write_poly3(HANDLE, 0, buf, 1, 2, 3, three)
+
+		nbio.next_tick_poly(1, one)
+		nbio.next_tick_poly2(1, 2, two)
+		nbio.next_tick_poly3(1, 2, 3, three)
+
+		nbio.timeout_poly(1, 1, one)
+		nbio.timeout_poly2(1, 1, 2, two)
+		nbio.timeout_poly3(1, 1, 2, 3, three)
+
+		nbio.poll_poly(TCP_SOCKET, .Receive, 1, one)
+		nbio.poll_poly2(TCP_SOCKET, .Receive, 1, 2, two)
+		nbio.poll_poly3(TCP_SOCKET, .Receive, 1, 2, 3, three)
+
+		nbio.open_poly("", 1, one)
+		nbio.open_poly2("", 1, 2, two)
+		nbio.open_poly3("", 1, 2, 3, three)
+
+		nbio.stat_poly(HANDLE, 1, one)
+		nbio.stat_poly2(HANDLE, 1, 2, two)
+		nbio.stat_poly3(HANDLE, 1, 2, 3, three)
+
+		ev(t, n, 0) // Test that no callbacks are ran before the loop is ticked.
+		ev(t, nbio.run(), nil)
+		ev(t, n, NUM_TESTS) // Test that all callbacks have ran.
+	}
+}
+
+@(test)
+two_ops_at_the_same_time :: proc(t: ^testing.T) {
+	if event_loop_guard(t) {
+		testing.set_fail_timeout(t, time.Minute)
+
+		server, err := nbio.create_udp_socket(.IP4)
+		ev(t, err, nil)
+		defer nbio.close(server)
+
+		berr := nbio.bind(server, {nbio.IP4_Loopback, 0})
+		ev(t, berr, nil)
+		ep, eperr := nbio.bound_endpoint(server)
+		ev(t, eperr, nil)
+
+		// Server.
+		{
+			nbio.poll_poly(server, .Receive, t, on_poll)
+
+			on_poll :: proc(op: ^nbio.Operation, t: ^testing.T) {
+				ev(t, op.poll.result, nbio.Poll_Result.Ready)
+			}
+
+			buf: [128]byte
+			nbio.recv_poly(server, {buf[:]}, t, on_recv)
+
+			on_recv :: proc(op: ^nbio.Operation, t: ^testing.T) {
+				ev(t, op.recv.err, nil)
+			}
+		}
+
+		// Client.
+		{
+			sock, cerr := nbio.create_udp_socket(.IP4)
+			ev(t, cerr, nil)
+
+			// Make sure the server would block.
+			nbio.timeout_poly3(time.Millisecond*10, t, sock, ep.port, on_timeout)
+
+			on_timeout :: proc(op: ^nbio.Operation, t: ^testing.T, sock: nbio.UDP_Socket, port: int) {
+				nbio.send_poly(sock, {transmute([]byte)string("Hiya")}, t, on_send, {nbio.IP4_Loopback, port})
+			}
+
+			on_send :: proc(op: ^nbio.Operation, t: ^testing.T) {
+				ev(t, op.send.err, nil)
+				ev(t, op.send.sent, 4)
+
+				// Do another send after a bit, some backends don't trigger both ops when one was enough to
+				// use up the socket.
+				nbio.timeout_poly3(time.Millisecond*10, t, op.send.socket.(nbio.UDP_Socket), op.send.endpoint.port, on_timeout2)
+			}
+
+			on_timeout2 :: proc(op: ^nbio.Operation, t: ^testing.T, sock: nbio.UDP_Socket, port: int) {
+				nbio.send_poly(sock, {transmute([]byte)string("Hiya")}, t, on_send2, {nbio.IP4_Loopback, port})
+			}
+
+			on_send2 :: proc(op: ^nbio.Operation, t: ^testing.T) {
+				ev(t, op.send.err, nil)
+				ev(t, op.send.sent, 4)
+
+				nbio.close(op.send.socket.(nbio.UDP_Socket))
+			}
+		}
+
+		ev(t, nbio.run(), nil)
+	}
+}
+
+@(test)
+timeout :: proc(t: ^testing.T) {
+	if event_loop_guard(t) {
+		testing.set_fail_timeout(t, time.Minute)
+
+		start := time.now()
+
+		nbio.timeout_poly2(time.Millisecond*20, t, start, on_timeout)
+
+		on_timeout :: proc(op: ^nbio.Operation, t: ^testing.T, start: time.Time) {
+			since := time.since(start)
+			log.infof("timeout ran after: %v", since)
+			testing.expect(t, since >= time.Millisecond*19) // A ms grace, for some reason it is sometimes ran after 19.8ms.
+			if since < 20 {
+				log.warnf("timeout ran after: %v", since)
+			}
+		}
+
+		ev(t, nbio.run(), nil)
+	}
+}
+
+@(test)
+wake_up :: proc(t: ^testing.T) {
+	testing.set_fail_timeout(t, time.Minute)
+	if event_loop_guard(t) {
+		for _ in 0..<2 {
+			sock, _ := open_next_available_local_port(t)
+
+			// Add an accept, with nobody dialling this should block the event loop forever.
+			accept := nbio.accept(sock, proc(op: ^nbio.Operation) {
+				log.error("shouldn't be called")
+			})
+
+			// Make sure the accept is in progress.
+			ev(t, nbio.tick(timeout=0), nil)
+
+			hit: bool
+			thr := thread.create_and_start_with_poly_data2(nbio.current_thread_event_loop(), &hit, proc(l: ^nbio.Event_Loop, hit: ^bool) {
+				hit^ = true
+				nbio.wake_up(l)
+			}, context)
+			defer thread.destroy(thr)
+
+			// Should block forever until the thread calling wake_up will make it return.
+			ev(t, nbio.tick(), nil)
+			e(t, hit)
+
+			nbio.remove(accept)
+			nbio.close(sock)
+
+			ev(t, nbio.run(), nil)
+			ev(t, nbio.tick(timeout=0), nil)
+		}
+	}
+}
diff --git a/tests/core/nbio/net.odin b/tests/core/nbio/net.odin
new file mode 100644
index 000000000..688ee0b45
--- /dev/null
+++ b/tests/core/nbio/net.odin
@@ -0,0 +1,400 @@
+package tests_nbio
+
+import "core:mem"
+import "core:nbio"
+import "core:net"
+import "core:testing"
+import "core:time"
+import "core:log"
+
+open_next_available_local_port :: proc(t: ^testing.T, addr: net.Address = net.IP4_Loopback, loc := #caller_location) -> (sock: net.TCP_Socket, ep: net.Endpoint) {
+	err: net.Network_Error
+	sock, err = nbio.listen_tcp({addr, 0})
+	if err != nil {
+		log.errorf("listen_tcp: %v", err, location=loc)
+		return
+	}
+
+	ep, err = net.bound_endpoint(sock)
+	if err != nil {
+		log.errorf("bound_endpoint: %v", err, location=loc)
+	}
+
+	return
+}
+
+@(test)
+client_and_server_send_recv :: proc(t: ^testing.T) {
+	if event_loop_guard(t) {
+		testing.set_fail_timeout(t, time.Minute)
+
+		server, ep := open_next_available_local_port(t)
+
+		CONTENT :: [20]byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20}
+
+		State :: struct {
+			server:        net.TCP_Socket,
+			server_client: net.TCP_Socket,
+			client:        net.TCP_Socket,
+			recv_buf:      [20]byte,
+			send_buf:      [20]byte,
+		}
+
+		state := State{
+			server   = server,
+			send_buf = CONTENT,
+		}
+
+		close_ok :: proc(op: ^nbio.Operation, t: ^testing.T) {
+			ev(t, op.close.err, nil)
+		}
+
+		// Server
+		{
+			nbio.accept_poly2(server, t, &state, on_accept)
+
+			on_accept :: proc(op: ^nbio.Operation, t: ^testing.T, state: ^State) {
+				ev(t, op.accept.err, nil)
+
+				state.server_client = op.accept.client
+
+				log.debugf("accepted connection from: %v", op.accept.client_endpoint)
+
+				nbio.recv_poly2(state.server_client, {state.recv_buf[:]}, t, state, on_recv)
+			}
+
+			on_recv :: proc(op: ^nbio.Operation, t: ^testing.T, state: ^State) {
+				ev(t, op.recv.err, nil)
+				ev(t, op.recv.received, 20)
+				ev(t, state.recv_buf, CONTENT)
+
+				nbio.close_poly(state.server_client, t, close_ok)
+				nbio.close_poly(state.server, t, close_ok)
+			}
+
+			ev(t, nbio.tick(0), nil)
+		}
+
+		// Client
+		{
+			nbio.dial_poly2(ep, t, &state, on_dial)
+
+			on_dial :: proc(op: ^nbio.Operation, t: ^testing.T, state: ^State) {
+				ev(t, op.dial.err, nil)
+
+				state.client = op.dial.socket
+
+				nbio.send_poly2(state.client, {state.send_buf[:]}, t, state, on_send)
+			}
+
+			on_send :: proc(op: ^nbio.Operation, t: ^testing.T, state: ^State) {
+				ev(t, op.send.err, nil)
+				ev(t, op.send.sent, 20)
+
+				nbio.close_poly(state.client, t, close_ok)
+			}
+		}
+
+		ev(t, nbio.run(), nil)
+	}
+}
+
+@(test)
+close_and_remove_accept :: proc(t: ^testing.T) {
+	if event_loop_guard(t) {
+		testing.set_fail_timeout(t, time.Minute)
+
+		server, _ := open_next_available_local_port(t)
+
+		accept := nbio.accept_poly(server, t, proc(_: ^nbio.Operation, t: ^testing.T) {
+			testing.fail_now(t)
+		})
+
+		ev(t, nbio.tick(0), nil)
+
+		nbio.close_poly(server, t, proc(op: ^nbio.Operation, t: ^testing.T) {
+			ev(t, op.close.err, nil)
+		})
+
+		nbio.remove(accept)
+		ev(t, nbio.run(), nil)
+	}
+}
+
+// Tests that when a client calls `close` on it's socket, `recv` returns with `0, nil` (connection closed).
+@(test)
+close_errors_recv :: proc(t: ^testing.T) {
+	if event_loop_guard(t) {
+		testing.set_fail_timeout(t, time.Minute)
+
+		server, ep := open_next_available_local_port(t)
+
+		// Server
+		{
+			nbio.accept_poly(server, t, on_accept)
+
+			on_accept :: proc(op: ^nbio.Operation, t: ^testing.T) {
+				ev(t, op.accept.err, nil)
+
+				bytes := make([]byte, 128, context.temp_allocator)
+				nbio.recv_poly(op.accept.client, {bytes}, t, on_recv)
+			}
+
+			on_recv :: proc(op: ^nbio.Operation, t: ^testing.T) {
+				ev(t, op.recv.received, 0)
+				ev(t, op.recv.err, nil)
+			}
+
+			ev(t, nbio.tick(0), nil)
+		}
+
+		// Client
+		{
+			nbio.dial_poly(ep, t, on_dial)
+
+			on_dial :: proc(op: ^nbio.Operation, t: ^testing.T) {
+				ev(t, op.dial.err, nil)
+				nbio.close_poly(op.dial.socket, t, on_close)
+			}
+
+			on_close :: proc(op: ^nbio.Operation, t: ^testing.T) {
+				ev(t, op.close.err, nil)
+			}
+		}
+
+		ev(t, nbio.run(), nil)
+	}
+}
+
+@(test)
+ipv6 :: proc(t: ^testing.T) {
+	if event_loop_guard(t) {
+		testing.set_fail_timeout(t, time.Minute)
+
+		server, ep := open_next_available_local_port(t, net.IP6_Loopback)
+
+		nbio.accept_poly(server, t, on_accept)
+		on_accept :: proc(op: ^nbio.Operation, t: ^testing.T) {
+			ev(t, op.accept.err, nil)
+			addr, is_ipv6 := op.accept.client_endpoint.address.(net.IP6_Address)
+			e(t, is_ipv6)
+			ev(t, addr, net.IP6_Loopback)
+			e(t, op.accept.client_endpoint.port != 0)
+			nbio.close(op.accept.client)
+			nbio.close(op.accept.socket)
+		}
+
+		nbio.dial_poly(ep, t, on_dial)
+		on_dial :: proc(op: ^nbio.Operation, t: ^testing.T) {
+			ev(t, op.dial.err, nil)
+			nbio.close(op.dial.socket)
+		}
+
+		ev(t, nbio.run(), nil)
+	}
+}
+
+@(test)
+accept_timeout :: proc(t: ^testing.T) {
+	if event_loop_guard(t) {
+		testing.set_fail_timeout(t, time.Minute)
+
+		sock, _ := open_next_available_local_port(t)
+
+		hit: bool
+		nbio.accept_poly2(sock, t, &hit, on_accept, timeout=time.Millisecond)
+
+		on_accept :: proc(op: ^nbio.Operation, t: ^testing.T, hit: ^bool) {
+			hit^ = true
+			ev(t, op.accept.err, net.Accept_Error.Timeout)
+			nbio.close(op.accept.socket)
+		}
+
+		ev(t, nbio.run(), nil)
+
+		e(t, hit)
+	}
+}
+
+@(test)
+poll_timeout :: proc(t: ^testing.T) {
+	if event_loop_guard(t) {
+		testing.set_fail_timeout(t, time.Minute)
+
+		sock, err := nbio.create_udp_socket(.IP4)
+		ev(t, err, nil)
+		berr := nbio.bind(sock, {nbio.IP4_Loopback, 0})
+		ev(t, berr, nil)
+
+		nbio.poll_poly(sock, .Receive, t, on_poll, time.Millisecond)
+		on_poll :: proc(op: ^nbio.Operation, t: ^testing.T) {
+			ev(t, op.poll.result, nbio.Poll_Result.Timeout)
+		}
+
+		ev(t, nbio.run(), nil)
+	}
+}
+
+/*
+This test walks through the scenario where a user wants to `poll` in order to check if some other package (in this case `core:net`),
+would be able to do an operation without blocking.
+
+It also tests whether a poll can be issues when it is already in a ready state.
+And it tests big send/recv buffers being handled properly.
+*/
+@(test)
+poll :: proc(t: ^testing.T) {
+	if event_loop_guard(t) {
+//		testing.set_fail_timeout(t, time.Minute)
+
+		can_recv: bool
+
+		sock, ep := open_next_available_local_port(t)
+
+		// Server
+		{
+			nbio.accept_poly2(sock, t, &can_recv, on_accept)
+
+			on_accept :: proc(op: ^nbio.Operation, t: ^testing.T, can_recv: ^bool) {
+				ev(t, op.accept.err, nil)
+
+				check_recv :: proc(op: ^nbio.Operation, t: ^testing.T, can_recv: ^bool, client: net.TCP_Socket) {
+					// Not ready to unblock the client yet, requeue for after 10ms.
+					if !can_recv^ {
+						nbio.timeout_poly3(time.Millisecond * 10, t, can_recv, client, check_recv)
+						return
+					}
+
+					free_all(context.temp_allocator)
+
+					// Connection was closed by client, close server.
+					if op.type == .Recv && op.recv.received == 0 && op.recv.err == nil {
+						nbio.close(client)
+						return
+					}
+
+					if op.type == .Recv {
+						log.debugf("received %M this time", op.recv.received)
+					}
+
+					// Receive some data to unblock the client, which should complete the poll it does, allowing it to send data again.
+					buf, mem_err := make([]byte, mem.Gigabyte, context.temp_allocator)
+					ev(t, mem_err, nil)
+					nbio.recv_poly3(client, {buf}, t, can_recv, client, check_recv)
+				}
+				nbio.timeout_poly3(time.Millisecond * 10, t, can_recv, op.accept.client, check_recv)
+			}
+
+			ev(t, nbio.tick(0), nil)
+		}
+
+		// Client
+		{
+			nbio.dial_poly2(ep, t, &can_recv, on_dial)
+
+			on_dial :: proc(op: ^nbio.Operation, t: ^testing.T, can_recv: ^bool) {
+				ev(t, op.dial.err, nil)
+
+				// Do a poll even though we know it's ready, so we can test that all implementations can handle that.
+				nbio.poll_poly2(op.dial.socket, .Send, t, can_recv, on_poll1)
+			}
+
+			on_poll1 :: proc(op: ^nbio.Operation, t: ^testing.T, can_recv: ^bool) {
+				ev(t, op.poll.result, nil)
+
+				// Send 4 GB of data, which in my experience causes a Would_Block error because we filled up the internal buffer.
+				buf, mem_err := make([]byte, mem.Gigabyte*4, context.temp_allocator)
+				ev(t, mem_err, nil)
+
+				// Use `core:net` as example external code that doesn't care about the event loop.
+				net.set_blocking(op.poll.socket, false)
+				n, send_err := net.send(op.poll.socket, buf)
+				ev(t, send_err, net.TCP_Send_Error.Would_Block)
+
+				log.debugf("blocking after %M", n)
+
+				// Tell the server it can start issueing recv calls, so it unblocks us.
+				can_recv^ = true
+
+				// Now poll again, when the server reads enough data it should complete, telling us we can send without blocking again.
+				nbio.poll_poly(op.poll.socket, .Send, t, on_poll2)
+			}
+
+			on_poll2 :: proc(op: ^nbio.Operation, t: ^testing.T) {
+				ev(t, op.poll.result, nil)
+
+				buf: [128]byte
+				bytes_written, send_err := net.send(op.poll.socket, buf[:])
+				ev(t, bytes_written, 128)
+				ev(t, send_err, nil)
+
+				nbio.close(op.poll.socket.(net.TCP_Socket))
+			}
+		}
+
+		ev(t, nbio.run(), nil)
+		nbio.close(sock)
+		ev(t, nbio.run(), nil)
+	}
+}
+
+@(test)
+sendfile :: proc(t: ^testing.T) {
+	if event_loop_guard(t) {
+		testing.set_fail_timeout(t, time.Minute)
+
+		CONTENT :: #load(#file)
+
+		sock, ep := open_next_available_local_port(t)
+
+		// Server
+		{
+			nbio.accept_poly(sock, t, on_accept)
+
+			on_accept :: proc(op: ^nbio.Operation, t: ^testing.T) {
+				ev(t, op.accept.err, nil)
+				e(t, op.accept.client != 0)
+
+				log.debugf("connection from: %v", op.accept.client_endpoint)
+				nbio.open_poly3(#file, t, op.accept.socket, op.accept.client, on_open)
+			}
+
+			on_open :: proc(op: ^nbio.Operation, t: ^testing.T, server, client: net.TCP_Socket) {
+				ev(t, op.open.err, nil)
+
+				nbio.sendfile_poly2(client, op.open.handle, t, server, on_sendfile)
+			}
+
+			on_sendfile :: proc(op: ^nbio.Operation, t: ^testing.T, server: net.TCP_Socket) {
+				ev(t, op.sendfile.err, nil)
+				ev(t, op.sendfile.sent, len(CONTENT))
+
+				nbio.close(op.sendfile.file)
+				nbio.close(op.sendfile.socket)
+				nbio.close(server)
+			}
+		}
+
+		// Client
+		{
+			nbio.dial_poly(ep, t, on_dial)
+
+			on_dial :: proc(op: ^nbio.Operation, t: ^testing.T) {
+				ev(t, op.dial.err, nil)
+
+				buf := make([]byte, len(CONTENT), context.temp_allocator)
+				nbio.recv_poly(op.dial.socket, {buf}, t, on_recv, all=true)
+			}
+
+			on_recv :: proc(op: ^nbio.Operation, t: ^testing.T) {
+				ev(t, op.recv.err, nil)
+				ev(t, op.recv.received, len(CONTENT))
+				ev(t, string(op.recv.bufs[0]), string(CONTENT))
+
+				nbio.close(op.recv.socket.(net.TCP_Socket))
+			}
+		}
+
+		ev(t, nbio.run(), nil)
+	}
+}
diff --git a/tests/core/nbio/remove.odin b/tests/core/nbio/remove.odin
new file mode 100644
index 000000000..063c2cf58
--- /dev/null
+++ b/tests/core/nbio/remove.odin
@@ -0,0 +1,247 @@
+package tests_nbio
+
+import "core:nbio"
+import "core:net"
+import "core:testing"
+import "core:time"
+import "core:log"
+
+// Removals are pretty complex.
+
+@(test)
+immediate_remove_of_sendfile :: proc(t: ^testing.T) {
+	if event_loop_guard(t) {
+		testing.set_fail_timeout(t, time.Minute)
+
+		sock, ep := open_next_available_local_port(t)
+
+		// Server
+		{
+			nbio.accept_poly(sock, t, on_accept)
+
+			on_accept :: proc(op: ^nbio.Operation, t: ^testing.T) {
+				ev(t, op.accept.err, nil)
+				e(t, op.accept.client != 0)
+
+				log.debugf("connection from: %v", op.accept.client_endpoint)
+				nbio.open_poly3(#file, t, op.accept.socket, op.accept.client, on_open)
+			}
+
+			on_open :: proc(op: ^nbio.Operation, t: ^testing.T, server, client: net.TCP_Socket) {
+				ev(t, op.open.err, nil)
+				e(t, op.open.handle != 0)
+
+				sendfile_op := nbio.sendfile_poly2(client, op.open.handle, t, server, on_sendfile)
+
+				// oh no changed my mind.
+				nbio.remove(sendfile_op)
+
+				nbio.close(op.open.handle)
+				nbio.close(client)
+				nbio.close(server)
+			}
+
+			on_sendfile :: proc(op: ^nbio.Operation, t: ^testing.T, server: net.TCP_Socket) {
+				log.error("on_sendfile shouldn't be called")
+			}
+		}
+
+		// Client
+		{
+			nbio.dial_poly(ep, t, on_dial)
+
+			on_dial :: proc(op: ^nbio.Operation, t: ^testing.T) {
+				ev(t, op.dial.err, nil)
+
+				buf := make([]byte, 128, context.temp_allocator)
+				nbio.recv_poly(op.dial.socket, {buf}, t, on_recv)
+			}
+
+			on_recv :: proc(op: ^nbio.Operation, t: ^testing.T) {
+				ev(t, op.recv.err, nil) 
+
+				nbio.close(op.recv.socket.(net.TCP_Socket))
+			}
+		}
+
+		ev(t, nbio.run(), nil)
+	}
+}
+
+@(test)
+immediate_remove_of_sendfile_without_stat :: proc(t: ^testing.T) {
+	if event_loop_guard(t) {
+		testing.set_fail_timeout(t, time.Minute)
+
+		sock, ep := open_next_available_local_port(t)
+
+		// Server
+		{
+			nbio.accept_poly(sock, t, on_accept)
+
+			on_accept :: proc(op: ^nbio.Operation, t: ^testing.T) {
+				ev(t, op.accept.err, nil)
+				e(t, op.accept.client != 0)
+
+				log.debugf("connection from: %v", op.accept.client_endpoint)
+				nbio.open_poly3(#file, t, op.accept.socket, op.accept.client, on_open)
+			}
+
+			on_open :: proc(op: ^nbio.Operation, t: ^testing.T, server, client: net.TCP_Socket) {
+				ev(t, op.open.err, nil)
+				e(t, op.open.handle != 0)
+
+				nbio.stat_poly3(op.open.handle, t, server, client, on_stat)
+			}
+
+			on_stat :: proc(op: ^nbio.Operation, t: ^testing.T, server, client: net.TCP_Socket) {
+				ev(t, op.stat.err, nil)
+
+				sendfile_op := nbio.sendfile_poly2(client, op.stat.handle, t, server, on_sendfile, nbytes=int(op.stat.size))
+
+				// oh no changed my mind.
+				nbio.remove(sendfile_op)
+
+				nbio.timeout_poly3(time.Millisecond * 10, op.stat.handle, client, server, proc(op: ^nbio.Operation, p1: nbio.Handle, p2, p3: net.TCP_Socket){
+					nbio.close(p1)
+					nbio.close(p2)
+					nbio.close(p3)
+				})
+			}
+
+			on_sendfile :: proc(op: ^nbio.Operation, t: ^testing.T, server: net.TCP_Socket) {
+				log.error("on_sendfile shouldn't be called")
+			}
+		}
+
+		// Client
+		{
+			nbio.dial_poly(ep, t, on_dial)
+
+			on_dial :: proc(op: ^nbio.Operation, t: ^testing.T) {
+				ev(t, op.dial.err, nil)
+
+				buf := make([]byte, 128, context.temp_allocator)
+				nbio.recv_poly(op.dial.socket, {buf}, t, on_recv)
+			}
+
+			on_recv :: proc(op: ^nbio.Operation, t: ^testing.T) {
+				ev(t, op.recv.err, nil) 
+
+				nbio.close(op.recv.socket.(net.TCP_Socket))
+			}
+		}
+
+		ev(t, nbio.run(), nil)
+	}
+}
+
+// Open should free the temporary memory allocated for the path when removed.
+// Can't really test that though, so should be checked manually that the internal callback is called but not the external.
+@(test)
+remove_open :: proc(t: ^testing.T) {
+	if event_loop_guard(t) {
+		testing.set_fail_timeout(t, time.Minute)
+
+		open := nbio.open(#file, on_open)
+		nbio.remove(open)
+
+		on_open :: proc(op: ^nbio.Operation) {
+			log.error("on_open shouldn't be called")
+		}
+
+		ev(t, nbio.run(), nil)
+	}
+}
+
+// Dial should close the socket when removed.
+// Can't really test that though, so should be checked manually that the internal callback is called but not the external.
+@(test)
+remove_dial :: proc(t: ^testing.T) {
+	if event_loop_guard(t) {
+		testing.set_fail_timeout(t, time.Minute)
+
+		sock, ep := open_next_available_local_port(t)
+		defer nbio.close(sock)
+
+		dial := nbio.dial(ep, on_dial)
+		nbio.remove(dial)
+
+		on_dial :: proc(op: ^nbio.Operation) {
+			log.error("on_dial shouldn't be called")
+		}
+
+		ev(t, nbio.run(), nil)
+	}
+}
+
+@(test)
+remove_next_tick :: proc(t: ^testing.T) {
+	if event_loop_guard(t) {
+		testing.set_fail_timeout(t, time.Minute)
+
+		nt := nbio.next_tick_poly(t, proc(op: ^nbio.Operation, t: ^testing.T) {
+			log.error("shouldn't be called")
+		})
+		nbio.remove(nt)
+
+		ev(t, nbio.run(), nil)
+	}
+}
+
+@(test)
+remove_timeout :: proc(t: ^testing.T) {
+	if event_loop_guard(t) {
+		testing.set_fail_timeout(t, time.Minute)
+
+		hit: bool
+		timeout := nbio.timeout_poly(time.Second, &hit, proc(_: ^nbio.Operation, hit: ^bool) {
+			hit^ = true
+		})
+
+		nbio.remove(timeout)
+
+		ev(t, nbio.run(), nil)
+
+		e(t, !hit)
+	}
+}
+
+@(test)
+remove_multiple_poll :: proc(t: ^testing.T) {
+	if event_loop_guard(t) {
+		testing.set_fail_timeout(t, time.Minute)
+
+		sock, ep := open_next_available_local_port(t)
+		defer nbio.close(sock)
+
+		hit: bool
+
+		first := nbio.poll(sock, .Receive, on_poll)
+		nbio.poll_poly2(sock, .Receive, t, &hit, on_poll2)
+
+		on_poll :: proc(op: ^nbio.Operation) {
+			log.error("shouldn't be called")
+		}
+
+		on_poll2 :: proc(op: ^nbio.Operation, t: ^testing.T, hit: ^bool) {
+			ev(t, op.poll.result, nbio.Poll_Result.Ready)
+			hit^ = true
+		}
+
+		ev(t, nbio.tick(0), nil)
+
+		nbio.remove(first)
+
+		ev(t, nbio.tick(0), nil)
+
+		nbio.dial_poly(ep, t, on_dial)
+
+		on_dial :: proc(op: ^nbio.Operation, t: ^testing.T) {
+			ev(t, op.dial.err, nil)
+		}
+
+		ev(t, nbio.run(), nil)
+		e(t, hit)
+	}
+}
diff --git a/tests/core/normal.odin b/tests/core/normal.odin
index 42da389d2..d0889bf89 100644
--- a/tests/core/normal.odin
+++ b/tests/core/normal.odin
@@ -32,6 +32,7 @@ download_assets :: proc "contextless" () {
 @(require) import "math/noise"
 @(require) import "math/rand"
 @(require) import "mem"
+@(require) import "nbio"
 @(require) import "net"
 @(require) import "odin"
 @(require) import "os"