Dierk/StatelessKanbanFlowGOLmassive.groovy

## StatelessKanbanFlowGOLmassive.groovy
import groovyx.gpars.dataflow.KanbanFlow
import groovyx.gpars.dataflow.KanbanLink
import groovyx.gpars.dataflow.KanbanTray
import groovyx.gpars.dataflow.ProcessingNode
import static groovyx.gpars.dataflow.ProcessingNode.node

/*
A massively parallel game of life with KanbanFlow.
Every cell signals to all neighbors when it has a new value.
Every cell waits for signals from all its neighbors.
Every cell signals to itself to proceed with the next generation.
Every cell publishes to an observer that funnels it to a single printer.
All coordination is implicit as every node depends on all input signals.

Noteworthy:
- Cell nodes are stateless. All cell information is passed as products in the trays. There is no "board".
- Observers are stateless, except being bound to the shared printer at construction time.
- The printer is stateless. The map of generations is passed between invocations.
- The whole system has almost no shared state.
  (Except each immutable "signal" sitting in neighbor-, self-, and publish-trays. Even this could be easily avoided)
- There is no heartbeat. Future generations are partly calculated ahead of reporting time.
- The printer only reports fully calculated, consistent generations.
  (and how many cell values of future generations are already known)
- Some "hot" areas are calculated faster than other ones - sometimes many generations ahead.
  However, an external observer only sees a homogeneous speed all over the board.
- The system is lock-free. There are neither read locks nor write locks.
- There are no livelocks, no deadlocks, no inconsistent states, no contentions, no rollback.

The game of life is not so difficult since the calculation of a cell does not require other
cells to change in the same generation based on the calculated cell value.
But it is a good example to prepare us for the next step when we will move "ants" from one
cell to the next and thus need "transactional" behavior (remove ant from source cell, put in destination cell)
and resolve possible conflicts like occupied destination cells or two ants trying to go to the same destination cell.

@author Dierk Koenig
*/

ProcessingNode makeCellNode() {
    def ProcessingNode thisNode = new ProcessingNode()
    thisNode.body = {
        publish,
        inSelf, outSelf,
        in1, out1, in2, out2, in3, out3, in4, out4, // in and out meaning depends on wiring sequence
        in6, out6, in7, out7, in8, out8, in9, out9
        ->
            def my = inSelf.take()

            def all = [in1, out1, in2, out2, in3, out3, in4, out4, in6, out6, in7, out7, in8, out8, in9, out9]
            def ins  = all.findAll{ it.link.consumerSpec == thisNode }
            def outs = all.findAll{ it.link.producerSpec == thisNode }

            def signals = ins*.take()
            assert signals*.generation.every { it == my.generation } // sanity check: all inputs are for the same generation

            def nextValue = my.value
            int aliveNeighbors = signals*.value.sum()  // the business rules of GOL
            switch(aliveNeighbors){
                case 0..1 : nextValue = 0; break
                case 3    : nextValue = 1; break
                case 4..8 : nextValue = 0; break
            }

            def product = [x: my.x, y: my.y, generation: my.generation + 1, value: nextValue].asImmutable()
            publish << product
            outs.each { it << product }
            outSelf << product
    }
    return thisNode
}

ProcessingNode makeObserver(KanbanLink funnel) {
    return node { signal ->
        funnel.downstream << new KanbanTray(link: funnel, product: signal.take())
    }
}

ProcessingNode makePrinter() {
    return node { inMap, outMap, inSignal, outSignal ->
        def signal =  inSignal.take()
        def genMap = inMap.take()
        def genNumber = signal.generation
        def signalsInGeneration = genMap[genNumber]
        signalsInGeneration << signal
        if ( signalsInGeneration.size() == size * size ) {
            genMap.remove genNumber
            def board = makeBoard()
            for (cell in signalsInGeneration) {
                board[cell.y][cell.x] = cell.value ? '*' : ' '
            }
            println "\ngeneration ${genNumber} :\n" +
                board.join("\n") +
                "\nfuture generations: " +
                genMap.collect{ gen, signals -> "$gen:${signals.size()}" }
        }
        outMap << genMap
        ~outSignal // outSignal will be triggered from the outside
    }
}


int getSize() { 10 }

List<List<Integer>> makeBoard() { (0..9).collect { [0] * 10 } } // endless torus world

int wrap(int i) { (i + size) % size } // indexes wrap around in the torus world

startBoard = makeBoard()
[[1,0],[2,1],[0,2],[1,2],[2,2]].each { startBoard[it[0]][it[1]] = 1 } // glider

void seed(KanbanLink link, int x, int y) {
    def tray = new KanbanTray(link: link, product: [x:x, y:y, generation: 0, value: startBoard[x][y]])
    link.downstream << tray
}

new KanbanFlow().with {
    cycleAllowed = true

    List<List<ProcessingNode>> cells = (0..<size).collect{ x -> (0..<size).collect { y -> makeCellNode() } }

    def field = [
        [-1, -1], [ 0, -1], [ 1, -1],
        [-1,  0],           [ 1,  0],
        [-1,  1], [ 0,  1], [ 1,  1]
    ]

    ProcessingNode printer = makePrinter()
    KanbanLink mapLink = link printer to printer
    mapLink.downstream << new KanbanTray(link:mapLink, product: [:].withDefault { [ ] } )
    KanbanLink signalLink = link printer to printer

    def currentLink

    for (int x in 0..<size) {
        for (int y in 0..<size) {
            def center = cells[x][y]
            currentLink = link center to makeObserver(signalLink)
            seed currentLink, x, y
            currentLink = link center to center
            seed currentLink, x, y
        }
    }

    for (int x in 0..<size) {
        for (int y in 0..<size) {
            def center = cells[x][y]
            field.each {
                def senderX = wrap(x + it[0])
                def senderY = wrap(y + it[1])
                def neighborCell = cells[senderX][senderY]
                currentLink = link neighborCell to center
                seed currentLink, senderX, senderY
            }
        }
    }

    start()
    sleep 1000
    stop()
}
	import groovyx.gpars.dataflow.KanbanFlow
	import groovyx.gpars.dataflow.KanbanLink
	import groovyx.gpars.dataflow.KanbanTray
	import groovyx.gpars.dataflow.ProcessingNode
	import static groovyx.gpars.dataflow.ProcessingNode.node

	/*
	A massively parallel game of life with KanbanFlow.
	Every cell signals to all neighbors when it has a new value.
	Every cell waits for signals from all its neighbors.
	Every cell signals to itself to proceed with the next generation.
	Every cell publishes to an observer that funnels it to a single printer.
	All coordination is implicit as every node depends on all input signals.

	Noteworthy:
	- Cell nodes are stateless. All cell information is passed as products in the trays. There is no "board".
	- Observers are stateless, except being bound to the shared printer at construction time.
	- The printer is stateless. The map of generations is passed between invocations.
	- The whole system has almost no shared state.
	(Except each immutable "signal" sitting in neighbor-, self-, and publish-trays. Even this could be easily avoided)
	- There is no heartbeat. Future generations are partly calculated ahead of reporting time.
	- The printer only reports fully calculated, consistent generations.
	(and how many cell values of future generations are already known)
	- Some "hot" areas are calculated faster than other ones - sometimes many generations ahead.
	However, an external observer only sees a homogeneous speed all over the board.
	- The system is lock-free. There are neither read locks nor write locks.
	- There are no livelocks, no deadlocks, no inconsistent states, no contentions, no rollback.

	The game of life is not so difficult since the calculation of a cell does not require other
	cells to change in the same generation based on the calculated cell value.
	But it is a good example to prepare us for the next step when we will move "ants" from one
	cell to the next and thus need "transactional" behavior (remove ant from source cell, put in destination cell)
	and resolve possible conflicts like occupied destination cells or two ants trying to go to the same destination cell.

	@author Dierk Koenig
	*/

	ProcessingNode makeCellNode() {
	def ProcessingNode thisNode = new ProcessingNode()
	thisNode.body = {
	publish,
	inSelf, outSelf,
	in1, out1, in2, out2, in3, out3, in4, out4, // in and out meaning depends on wiring sequence
	in6, out6, in7, out7, in8, out8, in9, out9
	->
	def my = inSelf.take()

	def all = [in1, out1, in2, out2, in3, out3, in4, out4, in6, out6, in7, out7, in8, out8, in9, out9]
	def ins = all.findAll{ it.link.consumerSpec == thisNode }
	def outs = all.findAll{ it.link.producerSpec == thisNode }

	def signals = ins*.take()
	assert signals*.generation.every { it == my.generation } // sanity check: all inputs are for the same generation

	def nextValue = my.value
	int aliveNeighbors = signals*.value.sum() // the business rules of GOL
	switch(aliveNeighbors){
	case 0..1 : nextValue = 0; break
	case 3 : nextValue = 1; break
	case 4..8 : nextValue = 0; break
	}

	def product = [x: my.x, y: my.y, generation: my.generation + 1, value: nextValue].asImmutable()
	publish << product
	outs.each { it << product }
	outSelf << product
	}
	return thisNode
	}

	ProcessingNode makeObserver(KanbanLink funnel) {
	return node { signal ->
	funnel.downstream << new KanbanTray(link: funnel, product: signal.take())
	}
	}

	ProcessingNode makePrinter() {
	return node { inMap, outMap, inSignal, outSignal ->
	def signal = inSignal.take()
	def genMap = inMap.take()
	def genNumber = signal.generation
	def signalsInGeneration = genMap[genNumber]
	signalsInGeneration << signal
	if ( signalsInGeneration.size() == size * size ) {
	genMap.remove genNumber
	def board = makeBoard()
	for (cell in signalsInGeneration) {
	board[cell.y][cell.x] = cell.value ? '*' : ' '
	}
	println "\ngeneration ${genNumber} :\n" +
	board.join("\n") +
	"\nfuture generations: " +
	genMap.collect{ gen, signals -> "$gen:${signals.size()}" }
	}
	outMap << genMap
	~outSignal // outSignal will be triggered from the outside
	}
	}


	int getSize() { 10 }

	List<List<Integer>> makeBoard() { (0..9).collect { [0] * 10 } } // endless torus world

	int wrap(int i) { (i + size) % size } // indexes wrap around in the torus world

	startBoard = makeBoard()
	[[1,0],[2,1],[0,2],[1,2],[2,2]].each { startBoard[it[0]][it[1]] = 1 } // glider

	void seed(KanbanLink link, int x, int y) {
	def tray = new KanbanTray(link: link, product: [x:x, y:y, generation: 0, value: startBoard[x][y]])
	link.downstream << tray
	}

	new KanbanFlow().with {
	cycleAllowed = true

	List<List<ProcessingNode>> cells = (0..<size).collect{ x -> (0..<size).collect { y -> makeCellNode() } }

	def field = [
	[-1, -1], [ 0, -1], [ 1, -1],
	[-1, 0], [ 1, 0],
	[-1, 1], [ 0, 1], [ 1, 1]
	]

	ProcessingNode printer = makePrinter()
	KanbanLink mapLink = link printer to printer
	mapLink.downstream << new KanbanTray(link:mapLink, product: [:].withDefault { [ ] } )
	KanbanLink signalLink = link printer to printer

	def currentLink

	for (int x in 0..<size) {
	for (int y in 0..<size) {
	def center = cells[x][y]
	currentLink = link center to makeObserver(signalLink)
	seed currentLink, x, y
	currentLink = link center to center
	seed currentLink, x, y
	}
	}

	for (int x in 0..<size) {
	for (int y in 0..<size) {
	def center = cells[x][y]
	field.each {
	def senderX = wrap(x + it[0])
	def senderY = wrap(y + it[1])
	def neighborCell = cells[senderX][senderY]
	currentLink = link neighborCell to center
	seed currentLink, senderX, senderY
	}
	}
	}

	start()
	sleep 1000
	stop()
	}