Tuple space implementation in ~100 of GDScript

tuple_space.gd

extends Node
signal send_ready
signal read_ready

enum ReadMode { Consuming, Preserving }
enum BlockingMode { NonBlocking, Blocking }

class RWComutex:
	signal unlocked

	var read_blocked := false
	var readers := 0

	func lock() -> void:
		while read_blocked or readers > 0:
			read_blocked = true
			await unlocked
		read_blocked = true

	func unlock() -> void:
		read_blocked = false
		unlocked.emit()

	func read_lock() -> void:
		while read_blocked: await unlocked
		readers += 1

	func read_unlock() -> void:
		readers -= 1
		if readers == 0: unlocked.emit()


class Mailbox:
	signal send(Dictionary)

	var filter : Callable
	var mode : ReadMode

	func _init(filter : Callable, mode : ReadMode) -> void:
		self.filter = filter
		self.mode = mode


var tuples            : Array[Dictionary] = []
var waiting_mailboxes : Array[Mailbox]    = []

var waiting_mutex := RWComutex.new()
var space_mutex   := RWComutex.new()

func send(tuple : Dictionary) -> void:
	await waiting_mutex.lock()
	for i in range(len(waiting_mailboxes)):
		var mailbox := waiting_mailboxes[i]
		if mailbox.filter.call(tuple):
			mailbox.send.emit(tuple)
			waiting_mailboxes.remove_at(i)
			if mailbox.mode == ReadMode.Consuming:
				waiting_mutex.unlock()
				return
	waiting_mutex.unlock()

	await space_mutex.lock()
	tuples.push_back(tuple)
	space_mutex.unlock()


func take   (filter : Callable) -> Dictionary: return await find_tuple(filter,  ReadMode.Consuming,    BlockingMode.Blocking)
func read   (filter : Callable) -> Dictionary: return await find_tuple(filter, ReadMode.Preserving,    BlockingMode.Blocking)
func snatch (filter : Callable) -> Dictionary: return await find_tuple(filter,  ReadMode.Consuming, BlockingMode.NonBlocking)
func peek   (filter : Callable) -> Dictionary: return await find_tuple(filter, ReadMode.Preserving, BlockingMode.NonBlocking)


func find_tuple(filter : Callable, mode : ReadMode, blocking : BlockingMode) -> Dictionary:
	var tuple

	match mode:
		ReadMode.Preserving: await space_mutex.read_lock()
		ReadMode.Consuming:  await space_mutex.lock()

	for i in range(len(tuples)):
		tuple = tuples[i].duplicate(true)
		if filter.call(tuple):
			if mode == ReadMode.Consuming: tuples.remove_at(i)
			break

	match mode:
		ReadMode.Preserving: space_mutex.read_unlock()
		ReadMode.Consuming: space_mutex.unlock()

	if tuple: return tuple

	match blocking:
		BlockingMode.Blocking:
			var mailbox = Mailbox.new(filter, mode)
			await waiting_mutex.lock()
			waiting_mailboxes.push_back(mailbox)
			waiting_mutex.unlock()
			return await mailbox.send

		_ :return {}