Flow-based version of the base-complement example.

basecompl_flow.go

package main

import (
	"bufio"
	"fmt"
	"github.com/trustmaster/goflow"
	"log"
	"os"
	"runtime"
)

const (
	BUFSIZE = 16
    NUMTHREADS = 4
)

// -------------------------------------------------------------
// Basecomlementer Component
// -------------------------------------------------------------
type BaseComplementer struct {
	flow.Component               // Embedding "superclass"
	Dna            <-chan []byte // Input port
	DnaBaseCompl   chan<- []byte // Output port
}

func (bc *BaseComplementer) OnDna(dna []byte) {
	for pos := range dna {
		if dna[pos] == 'A' {
			dna[pos] = 'T'
		} else if dna[pos] == 'T' {
			dna[pos] = 'A'
		} else if dna[pos] == 'C' {
			dna[pos] = 'G'
		} else if dna[pos] == 'G' {
			dna[pos] = 'C'
		}
	}
	bc.DnaBaseCompl <- dna
}

// -------------------------------------------------------------
// Printer
// -------------------------------------------------------------
type Printer struct {
	flow.Component
	Line <-chan []byte // Input
}

// Prints a line when it gets it
func (p *Printer) OnLine(line []byte) {
	fmt.Println(string(line))
}

// ---------------------------------------------------------
// BaseComplementer network
// ---------------------------------------------------------
type BaseComplementerApp struct {
	flow.Graph
}

func NewBaseCompelenterApp() *BaseComplementerApp {
	network := new(BaseComplementerApp)
	network.InitGraphState()

	// Add components
	network.Add(new(BaseComplementer), "basecompl1")
	network.Add(new(BaseComplementer), "basecompl2")
	network.Add(new(BaseComplementer), "basecompl3")
	network.Add(new(BaseComplementer), "basecompl4")
	network.Add(new(Printer), "printer")

	// Connect components
	network.Connect("basecompl1", "DnaBaseCompl", "basecompl2", "Dna", make(chan []byte, BUFSIZE))
	network.Connect("basecompl2", "DnaBaseCompl", "basecompl3", "Dna", make(chan []byte, BUFSIZE))
	network.Connect("basecompl3", "DnaBaseCompl", "basecompl4", "Dna", make(chan []byte, BUFSIZE))
	network.Connect("basecompl4", "DnaBaseCompl", "printer", "Line", make(chan []byte, BUFSIZE))
	network.MapInPort("In", "basecompl1", "Dna")

	return network
}

// ---------------------------------------------------------
// Main method
// ---------------------------------------------------------
var finish chan bool

// Use this handler to let main() know when the network terminates
func (a *BaseComplementerApp) Finish() {
	finish <- true
}

func main() {
	// Set the number of Operating System-threads to use
	fmt.Println("Starting ", NUMTHREADS, " threads ...")
	runtime.GOMAXPROCS(NUMTHREADS)

	// Termination signal channel
	finish = make(chan bool)
	// Create network
	net := NewBaseCompelenterApp()

	// Create the "In" channel
	in := make(chan []byte, BUFSIZE)
	net.SetInPort("In", in)

	// Run net
	flow.RunNet(net)

	file, err := os.Open("Homo_sapiens.GRCh37.67.dna_rm.chromosome.Y.fa")
	if err != nil {
		log.Fatal(err)
	} else {
		scan := bufio.NewScanner(file)
		for scan.Scan() {
			line := scan.Bytes()
			in <- line
		}
	}

	// Close the input to shut the network down
	close(in)
	// ... wait until done ...
	<-finish
}

[samuel blow]$ for i in $(seq 1 1 3); do echo "---"; time ./basecompl_flow > out.tmp; done

---

real    0m6.893s
user    0m22.936s
sys     0m4.072s

---

real    0m6.748s
user    0m22.020s
sys     0m4.056s

---

real    0m6.849s
user    0m22.576s
sys     0m4.084s