futures

main.go

package main

import (
	"container/list"
	"fmt"
	"github.com/tjarratt/babble"
	"sync"
	"sync/atomic"
	"unsafe"
)

func main() {
	var f [20000]*FutureMessage
	ForkJoin(func(c *Concurrent) {
		for i := 0; i < len(f); i++ {
			f[i] = c.Get()
		}
	})
	for i := range f {
		fmt.Println(f[i].Response().Value)
	}
}

type Message struct {
	Value string
}

type BlockingStruct struct {
	babbler babble.Babbler
}

func NewExecutor() *BlockingStruct {
	return &BlockingStruct{babbler: babble.NewBabbler()}
}

func (blockingStruct *BlockingStruct) GetMessage() *Message {
	return &Message{Value: blockingStruct.babbler.Babble()}
}

var executor = NewExecutor()

type Concurrent struct {
	list     list.List
	wg       sync.WaitGroup
	executor *BlockingStruct
}

func (c *Concurrent) Get() *FutureMessage {
	fm := new(FutureMessage)

	future := func() {
		defer c.wg.Done()
		message := c.executor.GetMessage()
		atomic.StorePointer(&fm.pointer, unsafe.Pointer(message))
	}

	c.list.PushBack(future)

	return fm
}

type FutureMessage struct {
	pointer unsafe.Pointer
}

func (fm *FutureMessage) Response() *Message {
	return (*Message)(fm.pointer)
}

func ForkJoin(f func(*Concurrent)) {
	executor.ForkJoin(f)
}

func (blockingStruct *BlockingStruct) ForkJoin(f func(*Concurrent)) {
	c := new(Concurrent)
	c.executor = blockingStruct

	f(c)

	c.wg.Add(c.list.Len())

	for node := c.list.Front(); node != nil; node = node.Next() {
		go node.Value.(func())()
	}

	c.wg.Wait()
}