Raw read speed.

rawread.go

//go:generate ffjson $GOFILE
package main

import (
	"bufio"
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"log"
	"os"
	"runtime"
	"sync"
	"time"
)

type ByteSize float64

type Focus struct {
	ID string `json:"finc.record_id"`
}

const (
	_           = iota // ignore first value by assigning to blank identifier
	KB ByteSize = 1 << (10 * iota)
	MB
	GB
	TB
	PB
	EB
	ZB
	YB
)

func (b ByteSize) String() string {
	switch {
	case b >= YB:
		return fmt.Sprintf("%.2fYB", b/YB)
	case b >= ZB:
		return fmt.Sprintf("%.2fZB", b/ZB)
	case b >= EB:
		return fmt.Sprintf("%.2fEB", b/EB)
	case b >= PB:
		return fmt.Sprintf("%.2fPB", b/PB)
	case b >= TB:
		return fmt.Sprintf("%.2fTB", b/TB)
	case b >= GB:
		return fmt.Sprintf("%.2fGB", b/GB)
	case b >= MB:
		return fmt.Sprintf("%.2fMB", b/MB)
	case b >= KB:
		return fmt.Sprintf("%.2fKB", b/KB)
	}
	return fmt.Sprintf("%.2fB", b)
}

func reverse(s string) string {
	runes := []rune(s)
	for i, j := 0, len(runes)-1; i < j; i, j = i+1, j-1 {
		runes[i], runes[j] = runes[j], runes[i]
	}
	return string(runes)
}

func decode(b []byte) {
	var dummy interface{}
	err := json.Unmarshal(b, &dummy)
	if err != nil {
		log.Println(b, err)
	}
}

func worker(in chan [][]byte, wg *sync.WaitGroup) {
	defer wg.Done()
	// var rx int
	for batch := range in {
		// rx++
		for _, line := range batch {
			// reverse(string(line))
			decode(line)
		}
		// log.Println(rx)
	}
}

func main() {
	bufsize := flag.Int("b", 4096, "read buffer size")
	flag.Parse()

	if flag.NArg() < 1 {
		log.Fatal("input file required")
	}

	start := time.Now()
	file, err := os.Open(flag.Arg(0))
	if err != nil {
		log.Fatal(err)
	}
	reader := bufio.NewReaderSize(file, *bufsize)
	var total int

	ch := make(chan [][]byte)
	var wg sync.WaitGroup
	for i := 0; i < runtime.NumCPU()*2; i++ {
		wg.Add(1)
		go worker(ch, &wg)
	}

	var size = 25000
	var batch [][]byte
	var i int

	for {
		b, err := reader.ReadBytes('\n')
		if err == io.EOF {
			break
		}
		if len(b) == 0 {
			continue
		}
		i++
		batch = append(batch, b)
		if i == size {
			bb := make([][]byte, size)
			copy(bb, batch)
			ch <- bb
			i = 0
			batch = batch[:0]
		}

		total += len(b)
		// go reverse(string(b))
		// ch <- string(b)
	}

	bb := make([][]byte, len(batch))
	copy(bb, batch)
	ch <- bb
	i = 0
	batch = batch[:0]

	close(ch)
	log.Println("done sending")
	wg.Wait()

	elapsed := time.Since(start)

	log.SetPrefix("rawread: ")
	log.SetFlags(0)
	log.Printf("%s in %s: %s/s", ByteSize(total), elapsed, ByteSize(float64(total)/elapsed.Seconds()))
}