more simple timestamp to days from 1970-01-01 conversion

Author: lomik

carbon/collector.go

@@ -10,7 +10,6 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/lomik/carbon-clickhouse/helper/RowBinary"
-	"github.com/lomik/carbon-clickhouse/helper/days1970"
 	"github.com/lomik/stop"
 	"github.com/lomik/zapwriter"
 )
@@ -178,14 +177,13 @@ func (c *Collector) readData(exit chan struct{}) []*Point {
 }
 
 func (c *Collector) local(exit chan struct{}) {
-	days := &days1970.Days{}
-
 	for {
 		points := c.readData(exit)
 		if points == nil || len(points) == 0 {
 			// exit closed
 			return
 		}
+		now := uint32(time.Now().Unix())
 
 		b := RowBinary.GetWriteBuffer()
 
@@ -206,8 +204,7 @@ func (c *Collector) local(exit chan struct{}) {
 				[]byte(p.Metric),
 				p.Value,
 				p.Timestamp,
-				days.Timestamp(p.Timestamp),
-				uint32(time.Now().Unix()),
+				now,
 			)
 		}
 

helper/RowBinary/date.go

@@ -0,0 +1,49 @@
+package RowBinary
+
+import "time"
+
+var daysTimestampStart []int64
+
+func init() {
+	daysTimestampStart = make([]int64, 0)
+	end := time.Now().UTC().Add(10 * 365 * 24 * time.Hour)
+
+	t := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC)
+	for t.Before(end) {
+		ts := time.Date(t.Year(), t.Month(), t.Day(), 0, 0, 0, 0, time.Local).Unix()
+		daysTimestampStart = append(daysTimestampStart, ts)
+		t = t.Add(24 * time.Hour)
+	}
+}
+
+func TimestampToDays(timestamp uint32) uint16 {
+	if int64(timestamp) < daysTimestampStart[0] {
+		return 0
+	}
+
+	i := int(timestamp / 86400)
+	ts := int64(timestamp)
+
+	if i < 10 || i > len(daysTimestampStart)-10 {
+		// fallback to slow method
+		return SlowTimestampToDays(timestamp)
+	}
+
+FindLoop:
+	for {
+		if ts < daysTimestampStart[i] {
+			i--
+			continue FindLoop
+		}
+		if ts >= daysTimestampStart[i+1] {
+			i++
+			continue FindLoop
+		}
+		return uint16(i)
+	}
+}
+
+func SlowTimestampToDays(timestamp uint32) uint16 {
+	t := time.Unix(int64(timestamp), 0)
+	return uint16(time.Date(t.Year(), t.Month(), t.Day(), 0, 0, 0, 0, time.UTC).Unix() / 86400)
+}

helper/RowBinary/date_test.go

@@ -0,0 +1,41 @@
+package RowBinary
+
+import (
+	"testing"
+	"time"
+)
+
+func TestTimestampToDays(t *testing.T) {
+	ts := uint32(0)
+	end := uint32(time.Now().Unix()) + 473040000 // +15 years
+	for ts < end {
+		d1 := SlowTimestampToDays(ts)
+		d2 := TimestampToDays(ts)
+		if d1 != d2 {
+			t.FailNow()
+		}
+
+		ts += 780 // step 13 minutes
+	}
+}
+func BenchmarkTimestampToDays(b *testing.B) {
+	timestamp := uint32(time.Now().Unix())
+	x := SlowTimestampToDays(timestamp)
+
+	for i := 0; i < b.N; i++ {
+		if TimestampToDays(timestamp) != x {
+			b.FailNow()
+		}
+	}
+}
+
+func BenchmarkSlowTimestampToDays(b *testing.B) {
+	timestamp := uint32(time.Now().Unix())
+	x := SlowTimestampToDays(timestamp)
+
+	for i := 0; i < b.N; i++ {
+		if SlowTimestampToDays(timestamp) != x {
+			b.FailNow()
+		}
+	}
+}

helper/RowBinary/reader.go

@@ -10,19 +10,15 @@ import (
 	"math"
 	"os"
 	"time"
-
-	"github.com/lomik/carbon-clickhouse/helper/days1970"
 )
 
 // Read all good records from unfinished RowBinary file.
 type Reader struct {
-	now       uint32
 	fd        *os.File
 	reader    *bufio.Reader
 	offset    int
 	size      int
 	eof       bool
-	days      days1970.Days
 	line      [524288]byte
 	isReverse bool
 }
@@ -97,7 +93,7 @@ func (r *Reader) readRecord() ([]byte, error) {
 	}
 	r.size += 18
 
-	if r.Days() != r.days.TimestampWithNow(r.Timestamp(), r.now) {
+	if r.Days() != TimestampToDays(r.Timestamp()) {
 		return nil, errors.New("date and timestamp mismatch")
 	}
 
@@ -158,7 +154,6 @@ func NewReader(filename string) (*Reader, error) {
 	return &Reader{
 		fd:     fd,
 		reader: bufio.NewReader(fd),
-		now:    uint32(time.Now().Unix()),
 	}, nil
 }
 

helper/RowBinary/write_buffer.go

@@ -127,11 +127,11 @@ func (wb *WriteBuffer) Write(p []byte) {
 	wb.Used += copy(wb.Body[wb.Used:], p)
 }
 
-func (wb *WriteBuffer) WriteGraphitePoint(name []byte, value float64, timestamp uint32, days uint16, version uint32) {
+func (wb *WriteBuffer) WriteGraphitePoint(name []byte, value float64, timestamp uint32, version uint32) {
 	wb.WriteBytes(name)
 	wb.WriteFloat64(value)
 	wb.WriteUint32(timestamp)
-	wb.WriteUint16(days)
+	wb.WriteUint16(TimestampToDays(timestamp))
 	wb.WriteUint32(version)
 }
 

helper/days1970/days.go

@@ -1,7 +1,6 @@
 package receiver
 
 import (
-	"encoding/binary"
 	"errors"
 	"net"
 	"sync"
@@ -15,7 +14,6 @@ import (
 	"github.com/golang/protobuf/ptypes/empty"
 	pb "github.com/lomik/carbon-clickhouse/grpc"
 	"github.com/lomik/carbon-clickhouse/helper/RowBinary"
-	"github.com/lomik/carbon-clickhouse/helper/days1970"
 	"github.com/lomik/carbon-clickhouse/helper/tags"
 	"github.com/lomik/stop"
 	"go.uber.org/zap"
@@ -130,11 +128,6 @@ func (g *GRPC) doStore(ctx context.Context, in *pb.Payload, confirmRequired bool
 	// save to buffers
 	now := uint32(time.Now().Unix())
 
-	version := make([]byte, 4)
-	binary.LittleEndian.PutUint32(version, now)
-
-	var days days1970.Days
-
 	var wg *sync.WaitGroup
 	var errorChan chan error
 
@@ -165,11 +158,12 @@ func (g *GRPC) doStore(ctx context.Context, in *pb.Payload, confirmRequired bool
 				wb = RowBinary.GetWriterBufferWithConfirm(wg, errorChan)
 			}
 
-			wb.WriteBytes([]byte(m.Metric))
-			wb.WriteFloat64(m.Points[j].Value)
-			wb.WriteUint32(m.Points[j].Timestamp)
-			wb.WriteUint16(days.TimestampWithNow(m.Points[j].Timestamp, now))
-			wb.Write(version)
+			wb.WriteGraphitePoint(
+				[]byte(m.Metric),
+				m.Points[j].Value,
+				m.Points[j].Timestamp,
+				now,
+			)
 		}
 	}
 

helper/days1970/days_test.go

@@ -5,25 +5,23 @@ import (
 	"time"
 
 	"github.com/lomik/carbon-clickhouse/helper/RowBinary"
-	"github.com/lomik/carbon-clickhouse/helper/days1970"
 	"github.com/lomik/carbon-clickhouse/helper/tags"
 	pickle "github.com/lomik/graphite-pickle"
 )
 
 func PickleParser(exit chan struct{}, in chan []byte, out chan *RowBinary.WriteBuffer, metricsReceived *uint32, errors *uint32) {
-	days := &days1970.Days{}
 
 	for {
 		select {
 		case <-exit:
 			return
 		case b := <-in:
-			PickeParseBytes(exit, b, uint32(time.Now().Unix()), out, days, metricsReceived, errors)
+			PickeParseBytes(exit, b, uint32(time.Now().Unix()), out, metricsReceived, errors)
 		}
 	}
 }
 
-func PickeParseBytes(exit chan struct{}, b []byte, now uint32, out chan *RowBinary.WriteBuffer, days *days1970.Days, metricsReceived *uint32, errors *uint32) {
+func PickeParseBytes(exit chan struct{}, b []byte, now uint32, out chan *RowBinary.WriteBuffer, metricsReceived *uint32, errors *uint32) {
 	metricCount := uint32(0)
 	wb := RowBinary.GetWriteBuffer()
 
@@ -69,7 +67,6 @@ func PickeParseBytes(exit chan struct{}, b []byte, now uint32, out chan *RowBina
 			[]byte(name),
 			value,
 			uint32(timestamp),
-			days.TimestampWithNow(uint32(timestamp), now),
 			now,
 		)
 

receiver/grpc.go

@@ -2,15 +2,13 @@ package receiver
 
 import (
 	"bytes"
-	"encoding/binary"
 	"fmt"
 	"math"
 	"strconv"
 	"sync/atomic"
 	"unsafe"
 
 	"github.com/lomik/carbon-clickhouse/helper/RowBinary"
-	"github.com/lomik/carbon-clickhouse/helper/days1970"
 	"github.com/lomik/carbon-clickhouse/helper/tags"
 )
 
@@ -90,14 +88,11 @@ func PlainParseLine(p []byte) ([]byte, float64, uint32, error) {
 	return RemoveDoubleDot(p[:i1]), value, uint32(tsf), nil
 }
 
-func PlainParseBuffer(exit chan struct{}, b *Buffer, out chan *RowBinary.WriteBuffer, days *days1970.Days, metricsReceived *uint32, errors *uint32) {
+func PlainParseBuffer(exit chan struct{}, b *Buffer, out chan *RowBinary.WriteBuffer, metricsReceived *uint32, errors *uint32) {
 	offset := 0
 	metricCount := uint32(0)
 	errorCount := uint32(0)
 
-	version := make([]byte, 4)
-	binary.LittleEndian.PutUint32(version, b.Time)
-
 	wb := RowBinary.GetWriteBuffer()
 
 MainLoop:
@@ -125,11 +120,7 @@ MainLoop:
 		}
 
 		// write result to buffer for clickhouse
-		wb.WriteBytes(name)
-		wb.WriteFloat64(value)
-		wb.WriteUint32(timestamp)
-		wb.WriteUint16(days.TimestampWithNow(timestamp, b.Time))
-		wb.Write(version)
+		wb.WriteGraphitePoint(name, value, timestamp, b.Time)
 		metricCount++
 	}
 
@@ -154,14 +145,12 @@ MainLoop:
 }
 
 func PlainParser(exit chan struct{}, in chan *Buffer, out chan *RowBinary.WriteBuffer, metricsReceived *uint32, errors *uint32) {
-	days := &days1970.Days{}
-
 	for {
 		select {
 		case <-exit:
 			return
 		case b := <-in:
-			PlainParseBuffer(exit, b, out, days, metricsReceived, errors)
+			PlainParseBuffer(exit, b, out, metricsReceived, errors)
 			b.Release()
 		}
 	}