Channels Test (with comparison to Go)

channels_test.odin

package test

import "core:os"
import "core:fmt"
import "core:sync"
import "core:time"
import "core:thread"
import "core:text/scanner"
import "core:strings"

chan :: sync.Channel;
as_send :: sync.channel_as_send;
as_recv :: sync.channel_as_recv;

send :: sync.channel_send;
recv :: sync.channel_recv;
try_send :: sync.channel_try_send;
try_recv :: sync.channel_try_recv;

chan_make :: sync.channel_make;
chan_destroy :: sync.channel_destroy;
close :: sync.channel_close;
chan_iterator :: sync.channel_iterator;


/*
	Odin: chan(string)
	Go:   chan string

	Odin: chan(string, .Send)
	Go:   chan<- string

	Odin: chan(string, .Recv)
	Go:   <-chan string
*/


demo_channels :: proc() {
	messages := chan_make(string);
	defer chan_destroy(messages);

	thread.run_with_poly_data(messages, proc(messages: chan(string)) {
		send(messages, "ping");
	});

	msg := recv(messages);
	fmt.println(msg);
}

demo_channel_buffering :: proc() {
	messages := chan_make(string, 2);
	defer chan_destroy(messages);

	send(messages, "buffered");
	send(messages, "channel");

	fmt.println(recv(messages));
	fmt.println(recv(messages));
}


demo_channel_synchronization :: proc() {
	done := chan_make(bool, 1);

	thread.run_with_poly_data(done, proc(done: chan(bool)) {
		fmt.println("working...");
		time.sleep(time.Second);
		fmt.println("done");

		send(done, true);
	});

	_ = recv(done);
}


demo_channel_directions :: proc() {
	ping :: proc(pings: chan(string, .Send), msg: string) {
		send(pings, msg);
	}
	pong :: proc(pings: chan(string, .Recv), pongs: chan(string, .Send)) {
		msg := recv(pings);
		send(pongs, msg);
	}

	pings := chan_make(string, 1);
	pongs := chan_make(string, 1);
	ping(as_send(pings), "passed message");
	pong(as_recv(pings), as_send(pongs));
	fmt.println(recv(pongs));
}

demo_select :: proc() {
	c1 := chan_make(string);
	c2 := chan_make(string);
	defer chan_destroy(c1);
	defer chan_destroy(c2);

	thread.run_with_poly_data(c1, proc(c1: chan(string)) {
		time.sleep(1 * time.Second);
		send(c1, "one");
	});

	thread.run_with_poly_data(c2, proc(c2: chan(string)) {
		time.sleep(2 * time.Second);
		send(c2, "two");
	});

	for i := 0; i < 2; i += 1 {
		switch msg, index := sync.select_recv_msg(c1, c2); index {
		case 0: fmt.println("received", msg);
		case 1: fmt.println("received", msg);
		}
	}
}

demo_non_blocking_channel_operations :: proc() {
	messages := chan_make(string);
	signals := chan_make(string);
	defer chan_destroy(messages);
	defer chan_destroy(signals);

	if msg, ok := try_recv(messages); ok {
		fmt.println("received message", msg);
	} else {
		fmt.println("no message received");
	}

	msg := "hi";
	switch sync.select_try_send_msg(msg, messages) {
	case 0: fmt.println("sent message", msg);
	case:   fmt.println("no message sent");
	}

	switch msg, index := sync.select_try_recv_msg(messages, signals); index {
	case 0: fmt.println("received message", msg);
	case 1: fmt.println("received signal",  msg);
	case:   fmt.println("no activity");
	}
}

demo_closing_channels :: proc() {
	jobs := chan_make(int, 5);
	done := chan_make(bool);
	defer chan_destroy(jobs);
	defer chan_destroy(done);

	thread.run_with_poly_data2(jobs, done, proc(jobs: chan(int), done: chan(bool)) {
		for {
			j, more := try_recv(jobs);
			if more {
				fmt.println("received job", j);
			} else {
				fmt.println("received all jobs");
				send(done, true);
				return;
			}
		}
	});

	for j := 1; j <= 3; j += 1 {
		send(jobs, j);
		fmt.println("sent job", j);
	}
	close(jobs);
	fmt.println("sent all jobs");

	recv(done);
}

demo_iterator :: proc() {
	queue := chan_make(string, 2);
	defer chan_destroy(queue);

	send(queue, "one");
	send(queue, "two");
	close(queue);

	for elem in chan_iterator(queue) {
		fmt.println(elem);
	}
}

demo_worker_pools :: proc() {
	NUM_JOBS :: 5;
	jobs := chan_make(int, NUM_JOBS);
	results := chan_make(int, NUM_JOBS);
	defer chan_destroy(jobs);
	defer chan_destroy(results);

	for w := 1; w <= 3; w += 1 {
		thread.run_with_poly_data3(w, as_recv(jobs), as_send(results), proc(id: int, jobs: chan(int, .Recv), results: chan(int, .Send)) {
			for j in chan_iterator(jobs) {
				fmt.println("worker", id, "started job", j);
				time.sleep(time.Second);
				fmt.println("worker", id, "finished job", j);
				send(results, j * 2);
			}
		});
	}

	for j := 1; j <= NUM_JOBS; j += 1 {
		send(jobs, 1);
	}
	close(jobs);

	for a := 1; a <= NUM_JOBS; a += 1 {
		recv(results);
	}
}

demo_wait_groups :: proc() {
	wg := &sync.Wait_Group{};
	sync.wait_group_init(wg);
	defer sync.wait_group_destroy(wg);

	for i := 1; i <= 5; i += 1 {
		sync.wait_group_add(wg, 1);
		thread.run_with_poly_data2(i, wg, proc(id: int, wg: ^sync.Wait_Group) {
			defer sync.wait_group_done(wg);

			fmt.printf("Worker %d starting\n", id);

			time.sleep(time.Second);

			fmt.printf("Worker %d done\n", id);
		});
	}

	sync.wait_group_wait(wg);
}


main :: proc() {
	fmt.println("\n# demo_channels");
	demo_channels();
	fmt.println("\n# demo_channel_buffering");
	demo_channel_buffering();
	fmt.println("\n# demo_channel_synchronization");
	demo_channel_synchronization();
	fmt.println("\n# demo_channel_directions");
	demo_channel_directions();
	fmt.println("\n# demo_select");
	demo_select();
	fmt.println("\n# demo_non_blocking_channel_operations");
	demo_non_blocking_channel_operations();
	fmt.println("\n# demo_closing_channels");
	demo_closing_channels();
	fmt.println("\n# demo_iterator");
	demo_iterator();
	fmt.println("\n# demo_worker_pools");
	demo_worker_pools();
	fmt.println("\n# demo_wait_groups");
	demo_wait_groups();
}