diff --git a/api.go b/api.go index 6de319dbc..35366ac4c 100644 --- a/api.go +++ b/api.go @@ -414,7 +414,6 @@ func (api *API) ImportRoaring(ctx context.Context, indexName, fieldName string, if field == nil { return newNotFoundError(ErrFieldNotFound) } - errCh := make(chan error, len(nodes)) for _, node := range nodes { diff --git a/field.go b/field.go index 6f805256f..12bc1ce4e 100644 --- a/field.go +++ b/field.go @@ -1186,7 +1186,6 @@ func (f *Field) newView(path, name string) *view { view.rowAttrStore = f.rowAttrStore view.stats = f.Stats view.broadcaster = f.broadcaster - view.remoteShardPresent = f.containsShard if f.snapshotQueue != nil { view.snapshotQueue = f.snapshotQueue } @@ -1914,7 +1913,6 @@ func (f *Field) importRoaring(ctx context.Context, data []byte, shard uint64, vi if err != nil { return errors.Wrap(err, "creating fragment") } - if err := frag.importRoaring(ctx, data, clear); err != nil { return err } @@ -1939,7 +1937,6 @@ func (f *Field) importRoaringOverwrite(ctx context.Context, data []byte, shard u if err != nil { return errors.Wrap(err, "creating fragment") } - if err := frag.importRoaringOverwrite(ctx, data, block); err != nil { return err } diff --git a/http/handler.go b/http/handler.go index 8d05a5596..e19f2a777 100644 --- a/http/handler.go +++ b/http/handler.go @@ -278,6 +278,7 @@ func (h *Handler) collectStats(next http.Handler) http.Handler { h.logger.Printf("%s %s %v %s", r.Method, r.URL.String(), dur, queryString) statsTags = append(statsTags, "slow_query") + r.Body.Close() } pathParts := strings.Split(r.URL.Path, "/") @@ -1540,6 +1541,7 @@ func (h *Handler) handlePostClusterResizeRemoveNode(w http.ResponseWriter, r *ht http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable) return } + defer r.Body.Close() // Decode request. var req removeNodeRequest err := json.NewDecoder(r.Body).Decode(&req) @@ -1647,7 +1649,7 @@ func (h *Handler) handlePostTranslateData(w http.ResponseWriter, r *http.Request http.Error(w, err.Error(), http.StatusInternalServerError) return } - + defer r.Body.Close() // Stream all translation data. rd, err := h.api.GetTranslateEntryReader(r.Context(), offsets) if errors.Cause(err) == pilosa.ErrNotImplemented { @@ -1751,6 +1753,7 @@ func (h *Handler) handlePostImportColumnAttrs(w http.ResponseWriter, r *http.Req http.Error(w, "Not acceptable", http.StatusNotAcceptable) return } + defer r.Body.Close() opts := []pilosa.ImportOption{} @@ -1787,6 +1790,7 @@ func (h *Handler) handlePostImportColumnAttrs(w http.ResponseWriter, r *http.Req // handlPostRoaringImport func (h *Handler) handlePostImportRoaring(w http.ResponseWriter, r *http.Request) { + defer r.Body.Close() // Verify that request is only communicating over protobufs. if r.Header.Get("Content-Type") != "application/x-protobuf" { http.Error(w, "Unsupported media type", http.StatusUnsupportedMediaType) @@ -1795,7 +1799,6 @@ func (h *Handler) handlePostImportRoaring(w http.ResponseWriter, r *http.Request http.Error(w, "Not acceptable", http.StatusNotAcceptable) return } - indexName := mux.Vars(r)["index"] fieldName := mux.Vars(r)["field"] @@ -1833,7 +1836,6 @@ func (h *Handler) handlePostImportRoaring(w http.ResponseWriter, r *http.Request http.Error(w, "shard should be an unsigned integer", http.StatusBadRequest) return } - resp := &pilosa.ImportResponse{} // TODO give meaningful stats for import err = h.api.ImportRoaring(ctx, indexName, fieldName, shard, remote, req) @@ -1870,6 +1872,7 @@ func (h *Handler) handlePostTranslateKeys(w http.ResponseWriter, r *http.Request http.Error(w, "Not acceptable", http.StatusNotAcceptable) return } + defer r.Body.Close() buf, err := h.api.TranslateKeys(r.Context(), r.Body) if err != nil { @@ -1893,7 +1896,7 @@ func (h *Handler) handlePostTranslateIDs(w http.ResponseWriter, r *http.Request) http.Error(w, "Not acceptable", http.StatusNotAcceptable) return } - + defer r.Body.Close() buf, err := h.api.TranslateIDs(r.Context(), r.Body) if err != nil { http.Error(w, fmt.Sprintf("translate ids: %v", err), http.StatusInternalServerError) diff --git a/view.go b/view.go index 2377781ec..a851b8653 100644 --- a/view.go +++ b/view.go @@ -56,12 +56,11 @@ type view struct { // Fragments by shard. fragments map[uint64]*fragment - broadcaster broadcaster - stats stats.StatsClient - rowAttrStore AttrStore - logger logger.Logger - snapshotQueue snapshotQueue - remoteShardPresent func(uint64) bool + broadcaster broadcaster + stats stats.StatsClient + rowAttrStore AttrStore + logger logger.Logger + snapshotQueue snapshotQueue knownShards *roaring.Bitmap knownShardsCopied uint32 @@ -81,11 +80,10 @@ func newView(path, index, field, name string, fieldOptions FieldOptions) *view { fragments: make(map[uint64]*fragment), - broadcaster: NopBroadcaster, - stats: stats.NopStatsClient, - logger: logger.NopLogger, - remoteShardPresent: func(uint64) bool { return false }, - knownShards: roaring.NewSliceBitmap(), + broadcaster: NopBroadcaster, + stats: stats.NopStatsClient, + logger: logger.NopLogger, + knownShards: roaring.NewSliceBitmap(), } } @@ -322,15 +320,17 @@ func (v *view) CreateFragmentIfNotExists(shard uint64) (*fragment, error) { frag.RowAttrStore = v.rowAttrStore v.fragments[shard] = frag - v.addKnownShard(shard) v.notifyIfNewShard(shard) + v.addKnownShard(shard) return frag, nil } func (v *view) notifyIfNewShard(shard uint64) { - if v.remoteShardPresent(shard) { //checks the fields remoteShards bitmap to see if broadcast needed + + if v.knownShards.Contains(shard) { //checks the fields remoteShards bitmap to see if broadcast needed return } + broadcastChan := make(chan struct{}) go func() {