basecompl_par2.go

/// --------------------------------------------------------------------
// Dummy processing DNA code (by taking base complement) 
// Threaded version
// Author: Samuel Lampa, samuel.lampa@gmail.com
// Date: 2013-06-21
// --------------------------------------------------------------------

package main

import (
	"fmt"
	"bufio"
	"os"
	"log"
	"runtime"
)

const(
    BUFSIZE = 512
)

func main() {
    // Set the number of Operating System-threads to use
    numThreads := 3 // runtime.NumCPU() - 1
    fmt.Println("Starting ", numThreads, " threads ...")
    runtime.GOMAXPROCS(numThreads)

    // Read DNA file
    inFileName := "Homo_sapiens.GRCh37.67.dna_rm.chromosome.Y.fa"
    fileReadChan := fileReaderGen(inFileName)

    // Setting up a sequence of dummy processing, where
    // each operation is done in a separate goroutine
    // and thus possibly will be multiplexed upon an OS 
    // thread), and communication is done through channels
    // between each pair of goroutines.
    complChan1 := baseComplementGen(fileReadChan)
    complChan2 := baseComplementGen(complChan1)
    complChan3 := baseComplementGen(complChan2)
    complChan4 := baseComplementGen(complChan3)

    // Read the last channel in the "pipeline"
    for line := range complChan4 {
        if len(line) > 0 {
            fmt.Println(string(line))
        }
    }
}

// create a goroutine that will read the dna file
// line by line. it returns a channel which can be
// read line by line, to "draw" the goroutine's execution.
func fileReaderGen(filename string) chan []byte {
    fileReadChan := make(chan []byte, BUFSIZE)
    go func() {
        file, err := os.Open(filename)
        if err != nil {
            log.Fatal(err)
        } else {
            scan := bufio.NewScanner(file)
            for scan.Scan() {
                fileReadChan <- scan.Bytes()
            }
            close(fileReadChan)
            fmt.Println("Closed file reader channel")
        }
    }()
	return fileReadChan
}

// Initiates a goroutine that will take the base complement
// of each line in the input channel specified, and returns 
// an output channel that can be read line by line.
func baseComplementGen(inChan chan []byte) chan []byte {
    returnChan := make(chan []byte, BUFSIZE)
    go func() {
        for line := range inChan {
            for pos := range line {
                if line[pos] == 'A' {
                    line[pos] = 'T'
                } else if line[pos] == 'T' {
                    line[pos] = 'A'
                } else if line[pos] == 'C' {
                    line[pos] = 'G'
                } else if line[pos] == 'G' {
                    line[pos] = 'C'
                }
            }
            returnChan <- append([]byte(nil), line...)
        }
        close(returnChan)
        fmt.Println("Closed base complement generator channel")
    }()
    return returnChan
}

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

---

real    0m3.819s
user    0m8.800s
sys     0m2.296s

---

real    0m3.872s
user    0m8.720s
sys     0m2.324s

---

real    0m3.922s
user    0m8.752s
sys     0m2.328s