aggregation_test.go

package main

import (
	"fmt"
	"sync"
	"testing"
)

/*
Simple API for aggregating large sets of data that are cached in application memory.

Each shard corresponds to one day's worth of data.
*/
type DayShard [][1*24]int
var Database []DayShard
var DatabaseCreated = false

type HourBucket struct {
	Day int
	Hour int	
	Value int
}

func createDatabase(days int, objects int) []DayShard {
	shards := make([]DayShard, days)

	for index, _ := range shards {
		shards[index] = make([][1*24]int, objects)

		// create individual time series for 
		for i, _ := range shards[index] {
			for j := 0; j < 1*24; j++ {
				shards[index][i][j] = int(5e5)
			}
		}
	}

	return shards
}

func fetchTimeSeriesShard(day int, objects int, output chan HourBucket) {
	// each shard consists of 24 1-hour buckets
	results := make([]int, 24)
	shard := Database[day]

	// can split this based upon the number of objects in question
	for i:=0; i<objects; i++ {
		// we know that each individual objects' time series has 7*36 entries in the day
		for hour, value := range shard[i] {
			results[hour] += value
		}
	}


	hourBucket := HourBucket{Day: day}
	for i := 0; i < 24; i++ {
		hourBucket.Hour = i
		hourBucket.Value = results[i]
		output <- hourBucket
	}
}

func fetchTimeSeries(days int, objects int) * map[string]int {
	resultChannel := make(chan HourBucket)
	results := make(map[string]int)

	// for now, shard by day :)
	var wg sync.WaitGroup
	wg.Add(days)
	defer wg.Wait()

	for shardIndex:=0; shardIndex < days; shardIndex++ {
		go func(index int) {
			defer wg.Done()
			fetchTimeSeriesShard(index, objects, resultChannel)
		}(shardIndex)
	}

	// listen for changes and update the `results` map
	go func() {
		for {
			hourBucket, ok := <- resultChannel
			if !ok {
				return
			}

			results[fmt.Sprintf("%d::%d", hourBucket.Day, hourBucket.Hour)] = hourBucket.Value
		}
	}()

	return &results
}

func benchmarkPerfN(b *testing.B, days int, objects int) {
	b.StopTimer()
	if !DatabaseCreated {
		Database = createDatabase(365, int(1e5))
		DatabaseCreated = true
	}
	b.StartTimer()

	for i := 0; i < b.N; i++ {
		createDatabase(days, objects)
	}
}

//func BenchmarkPerf1e3Day(b *testing.B) { benchmarkPerfN(b, 1, int(1e3)) }
//func BenchmarkPerf1e4Day(b *testing.B) { benchmarkPerfN(b, 1, int(1e6)) }
//func BenchmarkPerf1e5Day(b *testing.B) { benchmarkPerfN(b, 1, int(1e5)) }

func BenchmarkPerf1e3(b *testing.B) { benchmarkPerfN(b, 365, int(1e3)) }
func BenchmarkPerf1e4(b *testing.B) { benchmarkPerfN(b, 365, int(1e4)) }
func BenchmarkPerf1e5(b *testing.B) { benchmarkPerfN(b, 365, int(1e5)) }

func main() {
	// NOT IN USE
}