pregel blog

goroutine.go

package main

import (
        "fmt"
        "math"
)

type message float64
type Vertex func(output chan<- message, input ...<-chan message)

func main() {
        A := make(chan message, 0)
        B := make(chan message, 0)
        C := make(chan message, 0)
        D := make(chan message, 0)
        E := make(chan message, 0)
        output := make(chan message, 0)

        add := func(output chan<- message, input ...<-chan message) {
                a := <-input[0]
                b := <-input[1]
                output <- message(a + b)
        }
        mul := func(output chan<- message, input ...<-chan message) {
                a := <-input[0]
                b := <-input[1]
                output <- message(a * b)
        }
        sigma := func(output chan<- message, input ...<-chan message) {
                a := <-input[0]
                output <- message(1 / (1 + math.Exp(-float64(a))))
        }

        go mul(C, A, B)
        go add(E, C, D)
        go sigma(output, E)

        A <- message(1.0)
        B <- message(1.0)
        D <- message(1.0)

        fmt.Println(<-output)
}

main.go

package main

import (
        "fmt"
        "math"
)

type message float64
type Vertex func(output chan<- message, input ...<-chan message)

func main() {
        A := make(chan message, 1)
        B := make(chan message, 1)
        C := make(chan message, 1)
        D := make(chan message, 1)
        E := make(chan message, 1)
        output := make(chan message, 1)

        add := func(output chan<- message, input ...<-chan message) {
                a := <-input[0]
                b := <-input[1]
                output <- message(a + b)
        }
        mul := func(output chan<- message, input ...<-chan message) {
                a := <-input[0]
                b := <-input[1]
                output <- message(a * b)
        }
        sigma := func(output chan<- message, input ...<-chan message) {
                a := <-input[0]
                output <- message(1 / (1 + math.Exp(-float64(a))))
        }

        A <- message(1.0)
        B <- message(1.0)
        D <- message(1.0)

        mul(C, A, B)
        add(E, C, D)
        sigma(output, E)

        fmt.Println(<-output)
}

pregel_test.go

package main

import (
        "math"
        "testing"
        "time"
)

type message float64
type Vertex func(output chan<- message, input ...<-chan message)

var (
        add = func(output chan<- message, input ...<-chan message) {
                a := <-input[0]
                b := <-input[1]
                time.Sleep(0 * time.Microsecond)
                output <- message(a + b)
        }
        mul = func(output chan<- message, input ...<-chan message) {
                a := <-input[0]
                b := <-input[1]
                time.Sleep(0 * time.Microsecond)
                output <- message(a * b)
        }
        sigma = func(output chan<- message, input ...<-chan message) {
                a := <-input[0]
                output <- message(1 / (1 + math.Exp(-float64(a))))
        }
)

func BenchmarkTest_concurrent(b *testing.B) {
        A := make(chan message, 0)
        B := make(chan message, 0)
        C := make(chan message, 0)
        D := make(chan message, 0)
        E := make(chan message, 0)
        F := make(chan message, 0)
        G := make(chan message, 0)
        output := make(chan message, 0)

        for i := 0; i < b.N; i++ {
                go mul(C, A, B)
                go mul(D, F, G)
                go add(E, C, D)
                go sigma(output, E)

                A <- message(1.0)
                B <- message(1.0)
                F <- message(1.0)
                G <- message(1.0)
                <-output
        }
}
func BenchmarkTest_sequential(b *testing.B) {
        A := make(chan message, 1)
        B := make(chan message, 1)
        C := make(chan message, 1)
        D := make(chan message, 1)
        E := make(chan message, 1)
        F := make(chan message, 1)
        G := make(chan message, 1)
        output := make(chan message, 1)

        for i := 0; i < b.N; i++ {
                A <- message(1.0)
                B <- message(1.0)
                F <- message(1.0)
                G <- message(1.0)
                mul(C, A, B)
                mul(D, F, G)
                add(E, C, D)
                sigma(output, E)

                <-output
        }
}