From 98f13fecd896e790e4272a7f578074de6633e487 Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Tue, 30 Jun 2015 16:24:20 +0000 Subject: [PATCH] adjusted loading routines to match intel littleEndian --- commands/pilosa-cruncher/cruncher.go | 7 +++++++ config/config.go | 2 +- core/batch.go | 2 ++ core/load.go | 10 ++++++++-- index/bitmap.go | 20 ++++++++++++++++++++ index/storage_cass.go | 12 ++++++------ util/constants.go | 10 ++++++---- util/statd.go | 3 ++- util/storage.go | 2 +- 9 files changed, 53 insertions(+), 15 deletions(-) diff --git a/commands/pilosa-cruncher/cruncher.go b/commands/pilosa-cruncher/cruncher.go index 70769042e..f71adf9c5 100644 --- a/commands/pilosa-cruncher/cruncher.go +++ b/commands/pilosa-cruncher/cruncher.go @@ -5,8 +5,11 @@ import ( log "github.com/cihub/seelog" "github.com/mitchellh/panicwrap" "os" + "pilosa/config" "pilosa/core" "pilosa/cruncher" + "pilosa/index" + "pilosa/util" "runtime/pprof" ) @@ -39,6 +42,10 @@ func main() { pprof.StartCPUProfile(f) defer pprof.StopCPUProfile() } + config.SetupConfig() + util.SetupUtil() + index.SetupCassandra() + cruncher := cruncher.NewCruncher() cruncher.Run() log.Warn("STOP") diff --git a/config/config.go b/config/config.go index d0ba69e0d..b0b7d8946 100644 --- a/config/config.go +++ b/config/config.go @@ -19,7 +19,7 @@ type Config struct { var config *Config -func init() { +func SetupConfig() { config = NewConfig("") } diff --git a/core/batch.go b/core/batch.go index 5030f3f18..31bb71e55 100644 --- a/core/batch.go +++ b/core/batch.go @@ -2,6 +2,7 @@ package core import ( "encoding/gob" + log "github.com/cihub/seelog" "pilosa/db" "pilosa/util" ) @@ -32,6 +33,7 @@ func init() { } func (self *Service) Batch(database_name, frame, compressed_bitmap string, bitmap_id uint64, slice int, filter uint64) error { + log.Trace("Batch:", "db:", database_name, " frame:", frame, " slice:", slice, " cb:", compressed_bitmap, " bid:", bitmap_id, "f:", filter) //determine the fragment_id from database/frame/slice database := self.Cluster.GetOrCreateDatabase(database_name) oslice := database.GetOrCreateSlice(slice) diff --git a/core/load.go b/core/load.go index 8014a517e..811259a2e 100644 --- a/core/load.go +++ b/core/load.go @@ -7,6 +7,7 @@ import ( "compress/gzip" "encoding/base64" "encoding/binary" + "errors" log "github.com/cihub/seelog" "pilosa/index" ) @@ -19,6 +20,11 @@ func copy_raw(src [32]uint64) index.BlockArray { return index.BlockArray{o} } func sendBitmap(service *Service, bitmap index.IBitmap, db string, frame string, bitmap_id, filter uint64, slice int, finish chan error) { + if slice < 0 { + log.Warn("Bad split", db, frame, slice, bitmap_id) + finish <- errors.New("BadSplit") + return + } compressed_bitmap := bitmap.ToCompressString() results := service.Batch(db, frame, compressed_bitmap, bitmap_id, slice, filter) finish <- results @@ -36,7 +42,7 @@ func FromApiString(service *Service, db string, frame string, api_string string, return "Bad" } var numChunks uint64 - err = binary.Read(reader, binary.BigEndian, &numChunks) + err = binary.Read(reader, binary.LittleEndian, &numChunks) if err != nil { log.Warn(err) return "Bad" @@ -52,7 +58,7 @@ func FromApiString(service *Service, db string, frame string, api_string string, Key uint64 Block [32]uint64 } - binary.Read(reader, binary.BigEndian, &raw) + binary.Read(reader, binary.LittleEndian, &raw) slice := raw.Key >> 5 if slice != last_slice { if first { diff --git a/index/bitmap.go b/index/bitmap.go index 2fd864946..31d7ed148 100644 --- a/index/bitmap.go +++ b/index/bitmap.go @@ -6,6 +6,7 @@ import ( "bytes" "compress/gzip" "encoding/base64" + "encoding/binary" "encoding/gob" log "github.com/cihub/seelog" "io/ioutil" @@ -413,6 +414,7 @@ type IBitmap interface { ToBytes() []byte FromBytes([]byte) ToCompressString() string + ToRawCompressString() (string, int) FromCompressString(string) } @@ -457,6 +459,24 @@ func (b *Bitmap) Get(a *Chunk) *Chunk { } return nil } +func (b *Bitmap) ToRawCompressString() (string, int) { + var bt bytes.Buffer + buf := gzip.NewWriter(&bt) + binary.Write(buf, binary.LittleEndian, uint64(b.nodes.Len())) + max_slice := 0 + for i := b.nodes.Min(); !i.Limit(); i = i.Next() { + obj := i.Item().(*Chunk) + max_slice = int(obj.Key) + binary.Write(buf, binary.LittleEndian, obj.Key) + for _, v := range obj.Value.Block { + binary.Write(buf, binary.LittleEndian, v) + } + } + buf.Flush() + //buf.Close() + max_slice = max_slice / 32 + return base64.StdEncoding.EncodeToString(bt.Bytes()), max_slice +} func (b *Bitmap) ToBytes() []byte { var ( diff --git a/index/storage_cass.go b/index/storage_cass.go index ddf23f102..272fa82de 100644 --- a/index/storage_cass.go +++ b/index/storage_cass.go @@ -27,7 +27,7 @@ type CassandraStorage struct { var cluster *gocql.ClusterConfig var session *gocql.Session -func init() { +func SetupCassandra() { var err error hosts := config.GetStringArrayDefault("cassandra_hosts", []string{"localhost"}) keyspace := config.GetStringDefault("cassandra_keyspace", "pilosa") @@ -44,12 +44,12 @@ func init() { func BuildSchema() { /* - "CREATE KEYSPACE IF NOT EXISTS pilosa WITH strategy_class = SimpleStrategy AND strategy_options:replication_factor = 1" - create keyspace if not exists pilosa with replication = {'class': 'SimpleStrategy', 'replication_factor' : 1} and durable_writes = true; - CREATE KEYSPACE pilosa WITH replication = {'class': 'NetworkTopologyStrategy', 'pilpang': '2'} AND durable_writes = true; + "CREATE KEYSPACE IF NOT EXISTS pilosa WITH strategy_class = SimpleStrategy AND strategy_options:replication_factor = 1" + create keyspace if not exists pilosa with replication = {'class': 'SimpleStrategy', 'replication_factor' : 1} and durable_writes = true; + CREATE KEYSPACE pilosa WITH replication = {'class': 'NetworkTopologyStrategy', 'pilpang': '2'} AND durable_writes = true; - CREATE TABLE IF NOT EXISTS bitmap (bitmap_id bigint, db varchar, frame varchar, slice int, filter int, chunkkey bigint, blockindex int, block bigint, PRIMARY KEY ((bitmap_id, db, frame, slice), chunkkey, blockindex) ) - " + CREATE TABLE IF NOT EXISTS bitmap (bitmap_id bigint, db varchar, frame varchar, slice int, filter int, chunkkey bigint, blockindex int, block bigint, PRIMARY KEY ((bitmap_id, db, frame, slice), chunkkey, blockindex) ) + " */ } diff --git a/util/constants.go b/util/constants.go index 512476bea..b9ba72020 100644 --- a/util/constants.go +++ b/util/constants.go @@ -30,10 +30,12 @@ func ByteToInt64(data []byte) int64 { return value } func Uint64ToInt64(before uint64) int64 { - buf := Uint64ToByte(before) - return ByteToInt64(buf) + return int64(before) + //buf := Uint64ToByte(before) + //return ByteToInt64(buf) } func Int64ToUint64(before int64) uint64 { - buf := Int64ToByte(before) - return ByteToUint64(buf) + return uint64(before) + //buf := Int64ToByte(before) + //return ByteToUint64(buf) } diff --git a/util/statd.go b/util/statd.go index 3771ca03e..9a836d176 100644 --- a/util/statd.go +++ b/util/statd.go @@ -20,7 +20,8 @@ var ( end chan bool ) -func init() { +func SetupUtil() { + setup_storage() timer = make(chan args, 32768) count = make(chan string, 32768) end = make(chan bool) diff --git a/util/storage.go b/util/storage.go index 0164a3a35..21cfac17e 100644 --- a/util/storage.go +++ b/util/storage.go @@ -9,7 +9,7 @@ import ( "github.com/kr/s3/s3util" ) -func init() { +func setup_storage() { access_key := config.GetString("AWS_ACCESS_KEY_ID") secret := config.GetString("AWS_SECRET_ACCESS_KEY")