Added request Close and optimized availble shard with new view capabilities

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 {

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
 	}

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)

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() {