nbio/workers

Source

main.odin ¶

Wrap linesTab width 1234567882 linesSource

1/*
2Example shows how you can execute CPU intensive (or blocking IO),
3without blocking the event loop thread.
4
5It creates a thread pool, starts a TCP server, which creates a task on the thread pool
6for each connection (the task currently just sleeps, but could do anything).
7
8Once the work on the worker thread is done, it uses `nbio.next_tick`,
9passing it the main event loop. This will queue that callback onto the main event loop.
10
11When the main thread runs it's next tick, it will see that queued work and execute it's callback.
12That callback finally sends the result to the client, and once it is sent, cleans up.
13
14You can run this with the netcat utility as the client like so: `nc 127.0.0.1 1234`.
15*/
16package main
17
18import "core:nbio"
19import "core:thread"
20import "core:time"
21
22Connection :: struct {
23	loop:   ^nbio.Event_Loop,
24	socket: nbio.TCP_Socket,
25	// What is sent has to outlive `do_work`, as the send completes later on the main thread.
26	bufs:   [1][]byte,
27}
28
29main :: proc() {
30	// accept, put work on worker, worker adds back on main thread
31
32	// Set up a thread pool with 2 workers.
33	workers: thread.Pool
34	thread.pool_init(&workers, context.allocator, 2)
35	thread.pool_start(&workers)
36
37	ep, ok := nbio.parse_endpoint("127.0.0.1:1234")
38	assert(ok)
39
40	err := nbio.acquire_thread_event_loop()
41	defer nbio.release_thread_event_loop()
42	assert(err == nil)
43
44	server, listen_err := nbio.listen_tcp(ep)
45	assert(listen_err == nil)
46	nbio.accept_poly(server, &workers, on_accept)
47
48	err = nbio.run()
49	assert(err == nil)
50
51	on_accept :: proc(op: ^nbio.Operation, workers: ^thread.Pool) {
52		assert(op.accept.err == nil)
53
54		// Accept next connection.
55		nbio.accept_poly(op.accept.socket, workers, on_accept)
56
57		// Add work to worker.
58		thread.pool_add_task(workers, context.allocator, do_work, new_clone(Connection{
59			loop   = op.l,
60			socket = op.accept.client,
61		}))
62	}
63
64	do_work :: proc(t: thread.Task) {
65		connection := (^Connection)(t.data)
66
67		// Imagine CPU intensive work.
68		time.sleep(time.Second * 5)
69
70		// Work has been done, we can now tell the client about it.
71		// NOTE: that we pass the event loop of the main IO thread here so it queues it on that.
72		connection.bufs = {transmute([]byte)string("Hellope!\n")}
73		nbio.send_poly(connection.socket, connection.bufs[:], connection, on_sent, l=connection.loop)
74	}
75
76	on_sent :: proc(op: ^nbio.Operation, connection: ^Connection) {
77		assert(op.send.err == nil)
78		// Client got our message, clean up.
79		nbio.close(connection.socket)
80		free(connection)
81	}
82}