Add base for thread safe go port.

Author: viferga

source/ports/go_port/source/go.mod

@@ -1,3 +1,3 @@
 module github.com/metacall/core/source/ports/go_port/source
 
-go 1.14
+go 1.17

source/ports/go_port/source/go_port.go

@@ -7,21 +7,25 @@ package metacall
 import "C"
 
 import (
-	"unsafe"
 	"errors"
+	"unsafe"
 )
 
 const PtrSizeInBytes = (32 << uintptr(^uintptr(0)>>63)) >> 3
 
 func Initialize() error {
 	// TODO: Remove this once go loader is implemented
-	if (int(C.metacall_initialize()) != 0) {
+	if int(C.metacall_initialize()) != 0 {
 		return errors.New("MetaCall failed to initialize")
 	}
 
 	return nil
 }
 
+func InitializeSafe() error {
+	// TODO
+}
+
 func LoadFromFile(tag string, scripts []string) error {
 	size := len(scripts)
 
@@ -32,7 +36,7 @@ func LoadFromFile(tag string, scripts []string) error {
 	defer C.free(unsafe.Pointer(cScripts))
 
 	// Convert cScripts to a Go Array so we can index it
-	goScripts := (*[1 << 30 - 1] * C.char)(cScripts)
+	goScripts := (*[1<<30 - 1]*C.char)(cScripts)
 
 	for index, script := range scripts {
 		goScripts[index] = C.CString(script)
@@ -62,7 +66,7 @@ func Call(function string, args ...interface{}) (interface{}, error) {
 	defer C.free(unsafe.Pointer(cArgs))
 
 	for index, arg := range args {
-		cArg := (*unsafe.Pointer)(unsafe.Pointer(uintptr(unsafe.Pointer(cArgs)) + uintptr(index) * PtrSizeInBytes))
+		cArg := (*unsafe.Pointer)(unsafe.Pointer(uintptr(unsafe.Pointer(cArgs)) + uintptr(index)*PtrSizeInBytes))
 
 		// Create int
 		if i, ok := arg.(int); ok {
@@ -89,9 +93,9 @@ func Call(function string, args ...interface{}) (interface{}, error) {
 		// TODO: Other types ...
 	}
 
-	defer (func () {
+	defer (func() {
 		for index, _ := range args {
-			cArg := (*unsafe.Pointer)(unsafe.Pointer(uintptr(unsafe.Pointer(cArgs)) + uintptr(index) * PtrSizeInBytes))
+			cArg := (*unsafe.Pointer)(unsafe.Pointer(uintptr(unsafe.Pointer(cArgs)) + uintptr(index)*PtrSizeInBytes))
 			C.metacall_value_destroy(*cArg)
 		}
 	})()
@@ -101,20 +105,24 @@ func Call(function string, args ...interface{}) (interface{}, error) {
 	if ret != nil {
 		defer C.metacall_value_destroy(ret)
 
-		switch (C.metacall_value_id(unsafe.Pointer(ret))) {
-			case C.METACALL_INT: {
+		switch C.metacall_value_id(unsafe.Pointer(ret)) {
+		case C.METACALL_INT:
+			{
 				return int(C.metacall_value_to_int(unsafe.Pointer(ret))), nil
 			}
 
-			case C.METACALL_FLOAT: {
+		case C.METACALL_FLOAT:
+			{
 				return float32(C.metacall_value_to_float(unsafe.Pointer(ret))), nil
 			}
 
-			case C.METACALL_DOUBLE: {
+		case C.METACALL_DOUBLE:
+			{
 				return float64(C.metacall_value_to_double(unsafe.Pointer(ret))), nil
 			}
 
-			case C.METACALL_STRING: {
+		case C.METACALL_STRING:
+			{
 				return C.GoString(C.metacall_value_to_string(unsafe.Pointer(ret))), nil
 			}
 

source/ports/go_port/source/queue.go

@@ -0,0 +1,158 @@
+package metacall
+
+import (
+	"sync/atomic"
+	"unsafe"
+)
+
+type node struct {
+	value interface{}
+	next  unsafe.Pointer
+}
+
+type queue struct {
+	head unsafe.Pointer
+	tail unsafe.Pointer
+}
+
+func load(p *unsafe.Pointer) *node {
+	return (*node)(atomic.LoadPointer(p))
+}
+
+func store(p *unsafe.Pointer, n *node) {
+	atomic.StorePointer(p, unsafe.Pointer(n))
+}
+
+func cas(p *unsafe.Pointer, old, new *node) bool {
+	return atomic.CompareAndSwapPointer(
+		p, unsafe.Pointer(old), unsafe.Pointer(new))
+}
+
+func newQueue() (q *queue) {
+	n := unsafe.Pointer(&node{})
+	q = &queue{head: n, tail: n}
+
+	return q
+}
+
+func (q *queue) enqueue(v interface{}) {
+	n := &node{value: v}
+	for {
+		last := load(&q.tail)
+		next := load(&last.next)
+		if last == load(&q.tail) {
+			if next == nil {
+				if cas(&last.next, next, n) {
+					cas(&q.tail, last, n)
+					return
+				}
+			} else {
+				cas(&q.tail, last, next)
+			}
+		}
+	}
+}
+
+func (q *queue) dequeue() (interface{}, bool) {
+	first := load(&q.head)
+	last := load(&q.tail)
+	next := load(&first.next)
+	if first == load(&q.head) {
+		if first == last {
+			if next == nil {
+				return nil, false
+			}
+			cas(&q.tail, last, next)
+		} else {
+			v := next.value
+			if cas(&q.head, first, next) {
+				return v, true
+			}
+		}
+	}
+
+	return nil, false
+}
+
+func (q *queue) iterator() <-chan interface{} {
+	c := make(chan interface{})
+	go func(q *queue, c chan<- interface{}) {
+		for {
+			v, ok := q.dequeue()
+			if !ok {
+				break
+			}
+
+			c <- v
+		}
+		close(c)
+	}(q, c)
+	return c
+}
+
+type Queue struct {
+	impl     *queue
+	watchers *queue
+	quit     chan int
+}
+
+func NewQueue() *Queue {
+	return &Queue{
+		impl:     newQueue(),
+		watchers: newQueue(),
+		quit:     make(chan int),
+	}
+}
+
+func (q *Queue) Enqueue(v interface{}) {
+	q.impl.enqueue(v)
+
+	for notify := range q.watchers.iterator() {
+		c := notify.(chan int)
+
+		go func() {
+			c <- 1
+		}()
+	}
+}
+
+func (q *Queue) Dequeue() (interface{}, bool) {
+	return q.impl.dequeue()
+}
+
+func (q *Queue) Iterator() <-chan interface{} {
+	c := make(chan interface{})
+	go func(q *Queue, c chan<- interface{}) {
+		for {
+			v, ok := q.Dequeue()
+
+			if !ok {
+				notify := make(chan int)
+				q.watchers.enqueue(notify)
+
+				select {
+				case <-notify:
+					continue
+				case <-q.quit:
+					go func() {
+						<-notify
+					}()
+					goto end
+				}
+
+			} else {
+				c <- v
+			}
+		}
+
+	end:
+		close(c)
+	}(q, c)
+	return c
+}
+
+func (q *Queue) Close() {
+	go func() {
+		q.quit <- 1
+	}()
+}