benchmark set for packing msgpack (fluentd packed forward)

README.md

Benchmark set of packing logs as fluentd packed forward data

usage

$ make

requirements

• Perl
• Go

example result

$ echo `curl -s 169.254.169.254/latest/meta-data/instance-type`
c3.2xlarge
$ cat /etc/system-release
Amazon Linux AMI release 2014.03
$ uname -a
Linux fluentd-test 3.10.48-55.140.amzn1.x86_64 #1 SMP Wed Jul 9 23:32:19 UTC 2014 x86_64 x86_64 x86_64 GNU/Linux
$ perl -v | grep version
This is perl 5, version 16, subversion 3 (v5.16.3) built for x86_64-linux-thread-multi
$ go version
go version go1.3 linux/amd64
$ make
testing perl
1.88user 0.04system 0:01.93elapsed 99%CPU (0avgtext+0avgdata 11136maxresident)k
0inputs+0outputs (0major+748minor)pagefaults 0swaps
testing go
3.98user 0.26system 0:03.94elapsed 107%CPU (0avgtext+0avgdata 21968maxresident)k
0inputs+0outputs (0major+1460minor)pagefaults 0swaps

cpanfile

requires "Data::MessagePack";

Makefile

bench: packedforward test.log extlib
	@echo "testing perl"
	@cat test.log | PERL5LIB=extlib time perl ./packedforward.pl > /dev/null
	@echo "testing go"
	@cat test.log | time ./packedforward > /dev/null

check: packedforward test.log extlib
	@head -1234 test.log | TIME=1407856567 PERL5LIB=extlib perl ./packedforward.pl | md5sum
	@head -1234 test.log | TIME=1407856567 ./packedforward | md5sum

extlib:
	cpanm --installdeps -l extlib .

test.log:
	perl -E 'say "x" x 80 while 1' | head -1000000 > test.log

packedforward: packedforward.go
	go get github.com/ugorji/go/codec
	go build -o packedforward

clean:
	rm -rf test.log packedforward extlib/

packedforward.go

package main

import (
	"bufio"
	"fmt"
	"encoding/binary"
	"os"
	"strconv"
	"time"
	"bytes"
)

const (
	mpInt64     byte = 0xd3
	mpStr            = 0xa0
	mpStr8           = 0xd9
	mpStr16          = 0xda
	mpStr32          = 0xdb
	mp2ElmArray      = 0x92
	mp1ElmMap        = 0x81
	mpBytes8         = 0xc4
	mpBytes16        = 0xc5
	mpBytes32        = 0xc6
)

func writeMpStringHead(buf *bytes.Buffer, l int) {
	switch {
	case l < 32:
		buf.WriteByte(mpStr | byte(l))
	case l < 256:
		buf.WriteByte(mpStr8)
		binary.Write(buf, binary.BigEndian, uint8(l))
	case l < 65536:
		buf.WriteByte(mpStr16)
		binary.Write(buf, binary.BigEndian, uint16(l))
	default:
		buf.WriteByte(mpStr32)
		binary.Write(buf, binary.BigEndian, uint32(l))
	}
}

func writeMpBytesHead(buf *bytes.Buffer, l int) {
	switch {
	case l < 256:
		buf.WriteByte(mpBytes8)
		binary.Write(buf, binary.BigEndian, uint8(l))
	case l < 65536:
		buf.WriteByte(mpBytes16)
		binary.Write(buf, binary.BigEndian, uint16(l))
	default:
		buf.WriteByte(mpBytes32)
		binary.Write(buf, binary.BigEndian, uint32(l))
	}
}

func toMsgpackRecord(ts int64, key string, value string) []byte {
	buf := new(bytes.Buffer)
	// 2 elments array [ts, {key: value}]
	buf.WriteByte(mp2ElmArray)
	// ts
	buf.WriteByte(mpInt64)
	binary.Write(buf, binary.BigEndian, ts)
	// 1 element map {key: value}
	buf.WriteByte(mp1ElmMap)
	// key
	writeMpStringHead(buf, len(key))
	buf.WriteString(key)
	// value
	writeMpStringHead(buf, len(value))
	buf.WriteString(value)
	return buf.Bytes()
}

func toMsgpackRecordSet(tag string, bin *[]byte) []byte {
	buf := new(bytes.Buffer)
	// 2 elments array [ts, bin]
	buf.WriteByte(mp2ElmArray)
	// tag
	writeMpStringHead(buf, len(tag))
	buf.WriteString(tag)
	// buf
	writeMpBytesHead(buf, len(*bin))
	buf.Write(*bin)
	return buf.Bytes()
}

func main() {
	scanner := bufio.NewScanner(os.Stdin)
	count := 0
	tag := "dummy"
	key := "message"
	buffer := make([]byte, 0, 1024*1024)
	var now int64
	if t := os.Getenv("TIME"); t == "" {
		now = time.Now().Unix()
	} else {
		_now, _ := strconv.Atoi(t)
		now = int64(_now)
	}
	for scanner.Scan() {
		count++
		data := toMsgpackRecord(now, key, scanner.Text())
		buffer = append(buffer, data...)
		if count%100 == 0 {
			data := toMsgpackRecordSet(tag, &buffer)
			fmt.Print(string(data))
			buffer = []byte{}
		}
	}
	if len(buffer) != 0 {
		data := toMsgpackRecordSet(tag, &buffer)
		fmt.Print(string(data))
	}
}

packedforward.pl

#!/usr/bin/env perl
use 5.12.1;
use Data::MessagePack;
my $mp     = Data::MessagePack->new;
my $count  = 0;
my $tag    = "dummy";
my $key    = "message";
my $buffer = "";
my $now    = int($ENV{TIME} || time);
while (my $message = <>) {
    $count++;
    chomp $message;
    $buffer .= $mp->pack([$now, { $key => $message }]);
    if ($count % 100 == 0) {
        print $mp->pack([$tag, $buffer]);
        $buffer = "";
    }
}
if ($buffer ne "") {
    print $mp->pack([$tag, $buffer]);
}