diff --git a/.circleci/config.yml b/.circleci/config.yml index efacf47f1..6484d4458 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -20,48 +20,48 @@ commands: add-github-auth: steps: - run: git config --global url."https://moleculacorp:${GITHUB_PERSONAL_ACCESS_TOKEN}@github.com".insteadOf "https://github.com" + restore-mod-cache: + steps: + - restore_cache: + key: mod-cache-{{ checksum "go.sum" }} + save-mod-cache: + steps: + - save_cache: + key: mod-cache-{{ checksum "go.sum" }} + paths: + - /go/pkg/mod/ + checkout-plus: + steps: + - add-github-auth + - checkout + - restore-mod-cache jobs: setup: executor: name: golang steps: - - add-github-auth - - checkout - - restore_cache: - keys: - - mod-cache-{{ checksum "go.sum" }} - - run: "go mod download" - - save_cache: - key: mod-cache-{{ checksum "go.sum" }} - paths: - - /go/pkg/mod/ - - persist_to_workspace: - root: . - paths: "*" + - checkout-plus + - run: go mod download + - save-mod-cache check-license-headers: executor: name: golang steps: - - attach_workspace: - at: . + - checkout-plus - run: make check-license-headers linter: executor: name: golang steps: - - attach_workspace: - at: . - - add-github-auth + - checkout-plus - run: curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sudo sh -s -- -b /usr/local/bin v1.23.8 - run: make golangci-lint test-build-arm: executor: name: golang steps: - - attach_workspace: - at: . - - add-github-auth + - checkout-plus - run: make build GOOS=linux GOARCH=arm GOARM=5 - run: make build GOOS=linux GOARCH=arm GOARM=6 - run: make build GOOS=linux GOARCH=arm GOARM=7 @@ -91,9 +91,7 @@ jobs: version: << parameters.golang_version >> resource_class: << parameters.resource_class >> steps: - - attach_workspace: - at: . - - add-github-auth + - checkout-plus - run: sudo apt-get install lsof - run: command: make << parameters.test_make_target >> SHARD_WIDTH=<< parameters.shard_width >> GOARCH=<< parameters.goarch >> @@ -102,18 +100,14 @@ jobs: executor: name: golang steps: - - attach_workspace: - at: . - - add-github-auth + - checkout-plus - setup_remote_docker - run: make clustertests-build prerelease: executor: name: golang steps: - - attach_workspace: - at: . - - add-github-auth + - checkout-plus - run: make prerelease - store_artifacts: path: build @@ -124,6 +118,7 @@ jobs: executor: name: golang steps: + - checkout-plus - attach_workspace: at: . - run: make release @@ -136,6 +131,7 @@ jobs: docker: - image: circleci/python:2.7-jessie steps: + - checkout-plus - attach_workspace: at: . - run: sudo pip install awscli @@ -144,10 +140,7 @@ jobs: executor: name: golang steps: - - checkout - - attach_workspace: - at: . - - add-github-auth + - checkout-plus - setup_remote_docker - run: make docker - run: docker login -u $DOCKER_USER -p $DOCKER_PASS @@ -156,10 +149,7 @@ jobs: executor: name: golang steps: - - checkout - - attach_workspace: - at: . - - add-github-auth + - checkout-plus - setup_remote_docker - run: make docker - run: docker login -u $DOCKER_USER -p $DOCKER_PASS @@ -170,6 +160,7 @@ workflows: build: jobs: - setup: + context: molecula filters: tags: only: /^v.*/ diff --git a/api.go b/api.go index 5f5b3807f..8b6f3249d 100644 --- a/api.go +++ b/api.go @@ -795,11 +795,30 @@ func (api *API) ClusterMessage(ctx context.Context, reqBody io.Reader) error { // Forward the message. if err := api.server.receiveMessage(msg); err != nil { - return errors.Wrap(err, "receiving message") + return MessageProcessingError{err} } return nil } +// MessageProcessingError is an error indicating that a cluster message could not be processed. +type MessageProcessingError struct { + Err error +} + +func (err MessageProcessingError) Error() string { + return "processing message: " + err.Err.Error() +} + +// Cause allows the error to be unwrapped. +func (err MessageProcessingError) Cause() error { + return err.Err +} + +// Unwrap allows the error to be unwrapped. +func (err MessageProcessingError) Unwrap() error { + return err.Err +} + // Schema returns information about each index in Pilosa including which fields // they contain. func (api *API) Schema(ctx context.Context) []*IndexInfo { diff --git a/cluster.go b/cluster.go index eacad25f7..669d2bbce 100644 --- a/cluster.go +++ b/cluster.go @@ -2036,7 +2036,9 @@ func (c *cluster) nodeJoin(node *Node) error { if c.haveTopologyAgreement() { return c.unprotectedSetStateAndBroadcast(ClusterStateNormal) } - return nil + // This lets the remote node to proceed with opening its holder, + // instead of waiting in DOWN state because cluster is in STARTING state. + return c.sendTo(node, c.unprotectedStatus()) } else if err != nil { return errors.Wrap(err, "checking if holder has data") } diff --git a/executor.go b/executor.go index f2dd49fcb..c54ad6d8d 100644 --- a/executor.go +++ b/executor.go @@ -3125,11 +3125,19 @@ func (e *executor) executeSetRow(ctx context.Context, indexName string, c *pql.C // Merge returned results at coordinating node. reduceFn := func(ctx context.Context, prev, v interface{}) interface{} { - val := v.(bool) - if prev == nil { + val, ok := v.(bool) + if !ok { + return errors.Errorf("executeSetRow.reduceFn: val is non-bool (%+v)", v) + } + if val { return val } - return val || prev.(bool) + + pval, ok := prev.(bool) + if !ok { + return errors.Errorf("executeSetRow.reduceFn: prev is non-bool (%+v)", prev) + } + return pval } result, err := e.mapReduce(ctx, indexName, shards, c, opt, mapFn, reduceFn) diff --git a/http/handler.go b/http/handler.go index b215433e7..67be9e297 100644 --- a/http/handler.go +++ b/http/handler.go @@ -1783,8 +1783,12 @@ func (h *Handler) handlePostClusterMessage(w http.ResponseWriter, r *http.Reques } err := h.api.ClusterMessage(r.Context(), r.Body) if err != nil { - // TODO this was the previous behavior, but perhaps not everything is a bad request - http.Error(w, err.Error(), http.StatusBadRequest) + switch err := err.(type) { + case pilosa.MessageProcessingError: + http.Error(w, err.Error(), http.StatusInternalServerError) + default: + http.Error(w, err.Error(), http.StatusBadRequest) + } return } diff --git a/roaring/container_stash.go b/roaring/container_stash.go index 2c7c573db..64a6d217e 100644 --- a/roaring/container_stash.go +++ b/roaring/container_stash.go @@ -16,8 +16,6 @@ package roaring import ( "fmt" - "reflect" - "runtime" "unsafe" ) @@ -123,9 +121,9 @@ func NewContainerBitmapN(bitmap []uint64, n int32) *Container { } // pad to required length if len(bitmap) < bitmapN { - bm2 := make([]uint64, bitmapN) - copy(bm2, bitmap) - bitmap = bm2 + var bm [bitmapN]uint64 + copy(bm[:], bitmap) + bitmap = bm[:] } c := &Container{typeID: containerBitmap, n: n} c.setBitmap(bitmap) @@ -135,7 +133,7 @@ func NewContainerBitmapN(bitmap []uint64, n int32) *Container { // NewContainerArray returns an array container using the provided set of // values. It's okay if the slice is nil; that's a length of zero. func NewContainerArray(set []uint16) *Container { - c := &Container{typeID: containerArray, n: int32(len(set))} + c := &Container{typeID: containerArray} c.setArray(set) return c } @@ -144,15 +142,17 @@ func NewContainerArray(set []uint16) *Container { // values. It's okay if the slice is nil; that's a length of zero. It copies // the provided slice to new storage. func NewContainerArrayCopy(set []uint16) *Container { - c := &Container{typeID: containerArray, n: int32(len(set))} + c := &Container{typeID: containerArray} c.setArrayMaybeCopy(set, true) return c } // NewContainerArrayN returns an array container using the specified // set of values, but overriding n. +// This is deprecated. It never worked in the first place. +// The provided value of n is ignored and instead derived from the set length. func NewContainerArrayN(set []uint16, n int32) *Container { - c := &Container{typeID: containerArray, n: n} + c := &Container{typeID: containerArray} c.setArray(set) return c } @@ -287,7 +287,7 @@ func (c *Container) Thaw() *Container { func (c *Container) unmapOrClone() *Container { if c.flags&flagFrozen != 0 { - // Caqn't modify this container, therefore, we have to make a + // Can't modify this container, therefore, we have to make a // copy. return c.Clone() } @@ -296,45 +296,44 @@ func (c *Container) unmapOrClone() *Container { switch c.typeID { case containerArray: // mapped flag is wrong here - if c.pointer == (*uint16)(unsafe.Pointer(&c.data)) { + if c.pointer == &c.data[0] { return c } // maybe it fits in storage - if c.len <= stashedArraySize { - copy(c.data[:stashedArraySize], c.array()) - c.pointer, c.cap = (*uint16)(unsafe.Pointer(&c.data)), stashedArraySize + if c.len <= int32(len(c.data)) { + copy(c.data[:], c.array()) + c.pointer, c.cap = &c.data[0], stashedArraySize return c } - array := c.array() - tmp := make([]uint16, c.len) - copy(tmp, array) - h := (*reflect.SliceHeader)(unsafe.Pointer(&tmp)) - c.pointer, c.cap = (*uint16)(unsafe.Pointer(h.Data)), int32(h.Cap) - runtime.KeepAlive(&tmp) + arr := make([]uint16, c.len) + copy(arr, c.array()) + if cap(arr) > 1<<16 { + arr = arr[: len(arr) : 1<<16] + } + c.pointer, c.cap = &arr[0], int32(cap(arr)) case containerRun: // mapped flag is wrong here - if c.pointer == (*uint16)(unsafe.Pointer(&c.data)) { + if c.pointer == &c.data[0] { return c } oldRuns := c.runs() // maybe it fits in storage if c.len <= stashedRunSize { - c.pointer, c.cap = (*uint16)(unsafe.Pointer(&c.data)), stashedRunSize + c.pointer, c.cap = &c.data[0], stashedRunSize copy(c.runs(), oldRuns) return c } - tmp := make([]interval16, c.len) - copy(tmp, oldRuns) - h := (*reflect.SliceHeader)(unsafe.Pointer(&tmp)) - c.pointer, c.cap = (*uint16)(unsafe.Pointer(h.Data)), int32(h.Cap) - runtime.KeepAlive(&tmp) + runs := make([]interval16, c.len) + copy(runs, oldRuns) + if cap(runs) > 1<<15 { + runs = runs[: len(runs) : 1<<15] + } + c.pointer, c.cap = &runs[0].start, int32(cap(runs)) case containerBitmap: - bitmap := c.bitmap() - tmp := make([]uint64, bitmapN) - copy(tmp, bitmap) - h := (*reflect.SliceHeader)(unsafe.Pointer(&tmp)) - c.pointer, c.len, c.cap = (*uint16)(unsafe.Pointer(h.Data)), bitmapN, bitmapN - runtime.KeepAlive(&tmp) + oldBitmap := c.bitmap() + var bitmap [1024]uint64 + copy(bitmap[:], oldBitmap) + c.pointer, c.len, c.cap = (*uint16)(unsafe.Pointer(&bitmap)), bitmapN, bitmapN default: panic(fmt.Sprintf("can't thaw invalid container, type %d", c.typeID)) } @@ -351,7 +350,7 @@ func (c *Container) array() []uint16 { panic("attempt to read non-array's array") } } - return *(*[]uint16)(unsafe.Pointer(&reflect.SliceHeader{Data: uintptr(unsafe.Pointer(c.pointer)), Len: int(c.len), Cap: int(c.cap)})) + return (*[1 << 16]uint16)(unsafe.Pointer(c.pointer))[:c.len:c.cap] } // setArrayMaybeCopy stores a set of uint16s as data. c must not be frozen. @@ -366,36 +365,32 @@ func (c *Container) setArrayMaybeCopy(array []uint16, doCopy bool) { panic("attempt to write non-array's array") } } - // no array: start with our default 5-value array - if array == nil { - c.pointer, c.len, c.cap = (*uint16)(unsafe.Pointer(&c.data[0])), 0, stashedArraySize - c.n = c.len - return - } - h := (*reflect.SliceHeader)(unsafe.Pointer(&array)) - if h.Data == uintptr(unsafe.Pointer(c.pointer)) { - // nothing to do but update length - c.len = int32(h.Len) - c.n = c.len - return + if len(array) > 1<<16 { + panic("impossibly large array") } // array we can fit in data store: if len(array) <= stashedArraySize { copy(c.data[:stashedArraySize], array) - c.pointer, c.len, c.cap = (*uint16)(unsafe.Pointer(&c.data[0])), int32(len(array)), stashedArraySize + c.pointer, c.len, c.cap = &c.data[0], int32(len(array)), stashedArraySize c.n = c.len c.flags &^= flagMapped // this is no longer using a hypothetical mmapped input array return } + if &array[0] == c.pointer { + // nothing to do but update length + c.len = int32(len(array)) + c.n = c.len + return + } // copy the array if doCopy { - a2 := make([]uint16, len(array)) - copy(a2, array) - h = (*reflect.SliceHeader)(unsafe.Pointer(&a2)) + array = append([]uint16(nil), array...) } - c.pointer, c.len, c.cap = (*uint16)(unsafe.Pointer(h.Data)), int32(h.Len), int32(h.Cap) + if cap(array) > 1<<16 { + array = array[: len(array) : 1<<16] + } + c.pointer, c.len, c.cap = &array[0], int32(len(array)), int32(cap(array)) c.n = c.len - runtime.KeepAlive(&array) } // setArrayMaybeCopy stores a set of uint16s as data. c must not be frozen. @@ -413,7 +408,7 @@ func (c *Container) bitmap() []uint64 { panic("attempt to read non-bitmap's bitmap") } } - return *(*[]uint64)(unsafe.Pointer(&reflect.SliceHeader{Data: uintptr(unsafe.Pointer(c.pointer)), Len: int(c.len), Cap: int(c.cap)})) + return (*[1024]uint64)(unsafe.Pointer(c.pointer))[:] } // AsBitmap yields a 65k-bit bitmap, storing it in the target if a target @@ -448,16 +443,16 @@ func (c *Container) AsBitmap(target []uint64) (out []uint64) { return out } // in theory this shouldn't happen? - return out + panic("unreachable") } // fillerBitmap is a bitmap full of filler. var fillerBitmap = func() (a [1024]uint64) { - for i := range a { - a[i] = ^uint64(0) - } - return a - }() + for i := range a { + a[i] = ^uint64(0) + } + return a +}() func splatRun(into *[1024]uint64, from interval16) { // Handle the case where the start and end fall within the same word. @@ -498,9 +493,10 @@ func (c *Container) setBitmap(bitmap []uint64) { panic("attempt to write non-bitmap's bitmap") } } - h := (*reflect.SliceHeader)(unsafe.Pointer(&bitmap)) - c.pointer, c.len, c.cap = (*uint16)(unsafe.Pointer(h.Data)), int32(h.Len), int32(h.Cap) - runtime.KeepAlive(&bitmap) + if len(bitmap) != 1024 { + panic("illegal bitmap length") + } + c.pointer, c.len, c.cap = (*uint16)(unsafe.Pointer(&bitmap[0])), bitmapN, bitmapN } // runs yields the data viewed as a slice of intervals. @@ -513,7 +509,7 @@ func (c *Container) runs() []interval16 { panic("attempt to read non-run's runs") } } - return *(*[]interval16)(unsafe.Pointer(&reflect.SliceHeader{Data: uintptr(unsafe.Pointer(c.pointer)), Len: int(c.len), Cap: int(c.cap)})) + return (*[1 << 15]interval16)(unsafe.Pointer(c.pointer))[:c.len:c.cap] } // setRuns stores a set of intervals as data. c must not be frozen. @@ -532,33 +528,29 @@ func (c *Container) setRunsMaybeCopy(runs []interval16, doCopy bool) { panic("attempt to write non-run's runs") } } - // no array: start with our default 2-value array - if runs == nil { - c.pointer, c.len, c.cap = (*uint16)(unsafe.Pointer(&c.data[0])), 0, stashedRunSize - return + if len(runs) > 1<<15 { + panic("impossibly large run set") } - h := (*reflect.SliceHeader)(unsafe.Pointer(&runs)) - if h.Data == uintptr(unsafe.Pointer(c.pointer)) { - // nothing to do but update length - c.len = int32(h.Len) - return - } - // array we can fit in data store: if len(runs) <= stashedRunSize { - newRuns := *(*[]interval16)(unsafe.Pointer(&reflect.SliceHeader{Data: uintptr(unsafe.Pointer(&c.data[0])), Len: stashedRunSize, Cap: stashedRunSize})) + newRuns := (*[stashedRunSize]interval16)(unsafe.Pointer(&c.data))[:len(runs)] copy(newRuns, runs) - c.pointer, c.len, c.cap = (*uint16)(unsafe.Pointer(&c.data[0])), int32(len(runs)), stashedRunSize + c.pointer, c.len, c.cap = &c.data[0], int32(len(newRuns)), int32(cap(newRuns)) c.flags &^= flagMapped // this is no longer using a hypothetical mmapped input array return } - if doCopy { - r2 := make([]interval16, len(runs)) - copy(r2, runs) - h = (*reflect.SliceHeader)(unsafe.Pointer(&r2)) + if &runs[0].start == c.pointer { + // nothing to do but update length + c.len = int32(len(runs)) + return } - c.pointer, c.len, c.cap = (*uint16)(unsafe.Pointer(h.Data)), int32(h.Len), int32(h.Cap) - runtime.KeepAlive(&runs) + if doCopy { + runs = append([]interval16(nil), runs...) + } + if cap(runs) > 1<<15 { + runs = runs[: len(runs) : 1<<15] + } + c.pointer, c.len, c.cap = &runs[0].start, int32(len(runs)), int32(cap(runs)) } // UpdateOrMake updates the container, yielding a new container if necessary. @@ -585,9 +577,9 @@ func (c *Container) UpdateOrMake(typ byte, n int32, mapped bool) *Container { // we don't know that any existing slice is usable, so let's ditch it switch c.typeID { case containerArray: - c.pointer, c.len, c.cap = (*uint16)(unsafe.Pointer(&c.data[0])), int32(0), stashedArraySize + c.pointer, c.len, c.cap = &c.data[0], 0, stashedArraySize case containerRun: - c.pointer, c.len, c.cap = (*uint16)(unsafe.Pointer(&c.data[0])), 0, stashedRunSize + c.pointer, c.len, c.cap = &c.data[0], 0, stashedRunSize default: c.pointer, c.len, c.cap = nil, 0, 0 } @@ -608,9 +600,9 @@ func (c *Container) Update(typ byte, n int32, mapped bool) { // we don't know that any existing slice is usable, so let's ditch it switch c.typeID { case containerArray: - c.pointer, c.len, c.cap = (*uint16)(unsafe.Pointer(&c.data[0])), int32(0), stashedArraySize + c.pointer, c.len, c.cap = nil, 0, 0 case containerRun: - c.pointer, c.len, c.cap = (*uint16)(unsafe.Pointer(&c.data[0])), 0, stashedRunSize + c.pointer, c.len, c.cap = nil, 0, 0 default: c.pointer, c.len, c.cap = nil, 0, 0 } diff --git a/roaring/roaring_internal_test.go b/roaring/roaring_internal_test.go index 4bb014b73..085b6e1d1 100644 --- a/roaring/roaring_internal_test.go +++ b/roaring/roaring_internal_test.go @@ -249,20 +249,20 @@ func TestBitmapCountRange(t *testing.T) { tests := []struct { start int32 end int32 - bitmap []uint64 + bitmap [bitmapN]uint64 exp int32 }{ - {start: 0, end: 1, bitmap: []uint64{1}, exp: 1}, - {start: 2, end: 7, bitmap: []uint64{0xFFFFFFFFFFFFFF18}, exp: 2}, - {start: 67, end: 68, bitmap: []uint64{0, 0x8}, exp: 1}, - {start: 1, end: 68, bitmap: []uint64{0x3, 0x8, 0xF}, exp: 2}, - {start: 1, end: 258, bitmap: []uint64{0xF, 0x8, 0xA, 0x4, 0xFFFFFFFFFFFFFFFF}, exp: 9}, - {start: 66, end: 71, bitmap: []uint64{0xF, 0xFFFFFFFFFFFFFF18}, exp: 2}, - {start: 63, end: 64, bitmap: []uint64{0x8000000000000000}, exp: 1}, + {start: 0, end: 1, bitmap: [bitmapN]uint64{1}, exp: 1}, + {start: 2, end: 7, bitmap: [bitmapN]uint64{0xFFFFFFFFFFFFFF18}, exp: 2}, + {start: 67, end: 68, bitmap: [bitmapN]uint64{0, 0x8}, exp: 1}, + {start: 1, end: 68, bitmap: [bitmapN]uint64{0x3, 0x8, 0xF}, exp: 2}, + {start: 1, end: 258, bitmap: [bitmapN]uint64{0xF, 0x8, 0xA, 0x4, 0xFFFFFFFFFFFFFFFF}, exp: 9}, + {start: 66, end: 71, bitmap: [bitmapN]uint64{0xF, 0xFFFFFFFFFFFFFF18}, exp: 2}, + {start: 63, end: 64, bitmap: [bitmapN]uint64{0x8000000000000000}, exp: 1}, } for i, test := range tests { - c.setBitmap(test.bitmap) + c.setBitmap(test.bitmap[:]) if ret := c.bitmapCountRange(test.start, test.end); ret != test.exp { t.Fatalf("test #%v count of %v from %v to %v should be %v but got %v", i, test.bitmap, test.start, test.end, test.exp, ret) } @@ -294,39 +294,39 @@ func TestIntersectionCountArrayBitmap2(t *testing.T) { a, b := NewContainerArray(nil), NewContainerBitmap(0, nil) tests := []struct { array []uint16 - bitmap []uint64 + bitmap [bitmapN]uint64 exp int32 }{ { array: []uint16{0}, - bitmap: []uint64{1}, + bitmap: [bitmapN]uint64{1}, exp: 1, }, { array: []uint16{0, 1}, - bitmap: []uint64{3}, + bitmap: [bitmapN]uint64{3}, exp: 2, }, { array: []uint16{64, 128, 129, 2000}, - bitmap: []uint64{932421, 2}, + bitmap: [bitmapN]uint64{932421, 2}, exp: 0, }, { array: []uint16{0, 65, 130, 195}, - bitmap: []uint64{255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255}, + bitmap: [bitmapN]uint64{255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255}, exp: 4, }, { array: []uint16{63, 120, 543, 639, 12000}, - bitmap: []uint64{0x8000000000000000, 0, 0, 0, 0, 0, 0, 0, 0, 0x8000000000000000}, + bitmap: [bitmapN]uint64{0x8000000000000000, 0, 0, 0, 0, 0, 0, 0, 0, 0x8000000000000000}, exp: 2, }, } for i, test := range tests { a.setArray(test.array) - b.setBitmap(test.bitmap) + b.setBitmap(test.bitmap[:]) ret := intersectionCountArrayBitmap(a, b) if ret != test.exp { t.Fatalf("test #%v intersectCountArrayBitmap fail received: %v exp: %v", i, ret, test.exp) @@ -3751,15 +3751,17 @@ func TestContainerCombinations(t *testing.T) { for _, x := range containerTypes { for _, y := range containerTypes { desc := fmt.Sprintf("%s(%s/%s, %s/%s)", getFunctionName(testOp.f), containerTypeNames[x], testOp.x, containerTypeNames[y], testOp.y) - ret := runContainerFunc(testOp.f, cts[x][testOp.x], cts[y][testOp.y]) - exp := testOp.exp + t.Run(desc, func(t *testing.T) { + ret := runContainerFunc(testOp.f, cts[x][testOp.x], cts[y][testOp.y]) + exp := testOp.exp - // Convert to all container types and check result. - for _, ct := range containerTypes { - if err := ret.BitwiseCompare(cts[ct][exp]); err != nil { - t.Errorf("test %s: %v", desc, err) + // Convert to all container types and check result. + for _, ct := range containerTypes { + if err := ret.BitwiseCompare(cts[ct][exp]); err != nil { + t.Error(err) + } } - } + }) } } }