// fanOutShardsHandler fetches a list of shards from the job and launches
// named ProcessShard tasks, one per shard.
3900
func (ctl *Controller) fanOutShardsHandler(c context.Context, payload proto.Message) error {
	msg := payload.(*tasks.FanOutShards)

	// Make sure the job is still present. If it is aborted, we still need to
	// launch the shards, so they notice they are being aborted. We could try
	// to abort all shards right here and now, but it basically means implementing
	// an alternative shard abort flow. Seems simpler just to let the regular flow
	// to proceed.
	job, err := getJobInState(c, JobID(msg.JobId), State_RUNNING, State_ABORTING)
	if err != nil || job == nil {
		return errors.Annotate(err, "in FanOutShards").Err()
	}

	// Grab the list of shards created in SplitAndLaunch. It must exist at this
	// point, since the job is in Running state.
	shardIDs, err := job.fetchShardIDs(c)
	if err != nil {
		return errors.Annotate(err, "in FanOutShards").Err()
	}

	// Enqueue a bunch of named ProcessShard tasks (one per shard) to actually
	// launch shard processing. This is idempotent operation, so if FanOutShards
	// crashes midway and later retried, nothing bad happens.
	tsks := make([]*tq.Task, len(shardIDs))
	for idx, sid := range shardIDs {
		tsks[idx] = makeProcessShardTask(job.ID, sid, 0, true)
	}
	return ctl.tq().AddTask(c, tsks...)
}
1071869
func SassHandler(h SassFunc) libs.SassCallback {
	return func(v interface{}, usv libs.UnionSassValue, rsv *libs.UnionSassValue) error {
		if *rsv == nil {
			*rsv = libs.MakeNil()
		}

		libCtx, ok := v.(*compctx)
		if !ok {
			return errors.New("libsass Context not found")
		}

		ctx := NewCompilerContext(libCtx.compiler)

		// Cast to exported Go types
		req := SassValue{value: usv}
		res, err := h(ctx, req)

		if err != nil {
			// Returns the error to libsass Compiler
			*rsv = libs.MakeError(err.Error())
			// Returning an error does nothing as libsass is in charge of
			// reporting error to user
			return err
		}

		*rsv = res.Val()
		return err
	}
}
1320936
func (e *executor) executeNotShard(ctx context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeNotShard")
	defer span.Finish()

	if len(c.Children) == 0 {
		return nil, errors.New("Not() requires an input row")
	} else if len(c.Children) > 1 {
		return nil, errors.New("Not() only accepts a single row input")
	}

	// Make sure the index supports existence tracking.
	idx := e.Holder.Index(index)
	if idx == nil {
		return nil, ErrIndexNotFound
	} else if idx.existenceField() == nil {
		return nil, errors.Errorf("index does not support existence tracking: %s", index)
	}

	var existenceRow *Row
	existenceFrag := e.Holder.fragment(index, existenceFieldName, viewStandard, shard)
	if existenceFrag == nil {
		existenceRow = NewRow()
	} else {
		existenceRow = existenceFrag.row(0)
	}

	row, err := e.executeBitmapCallShard(ctx, index, c.Children[0], shard)
	if err != nil {
		return nil, err
	}

	return existenceRow.Difference(row), nil
}
666710
func (c *RouteAllocationController) AllocateRouterShard(route *routeapi.Route) (*routeapi.RouterShard, error) {

	klog.V(4).Infof("Allocating router shard for Route: %s [alias=%s]",
		route.Name, route.Spec.Host)

	shard, err := c.Plugin.Allocate(route)

	if err != nil {
		klog.Errorf("unable to allocate router shard: %v", err)
		return shard, err
	}

	klog.V(4).Infof("Route %s allocated to shard %s [suffix=%s]",
		route.Name, shard.ShardName, shard.DNSSuffix)

	return shard, err
}
4153
func (e *executor) executeSumCountShard(ctx context.Context, index string, c *pql.Call, shard uint64) (ValCount, error) {
	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeSumCountShard")
	defer span.Finish()

	var filter *Row
	if len(c.Children) == 1 {
		row, err := e.executeBitmapCallShard(ctx, index, c.Children[0], shard)
		if err != nil {
			return ValCount{}, errors.Wrap(err, "executing bitmap call")
		}
		filter = row
	}

	fieldName, _ := c.Args["field"].(string)

	field := e.Holder.Field(index, fieldName)
	if field == nil {
		return ValCount{}, nil
	}

	bsig := field.bsiGroup(fieldName)
	if bsig == nil {
		return ValCount{}, nil
	}

	fragment := e.Holder.fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard)
	if fragment == nil {
		return ValCount{}, nil
	}

	vsum, vcount, err := fragment.sum(filter, bsig.BitDepth())
	if err != nil {
		return ValCount{}, errors.Wrap(err, "computing sum")
	}
	return ValCount{
		Val:   int64(vsum) + (int64(vcount) * bsig.Min),
		Count: int64(vcount),
	}, nil
}
748338
func (d *BoundDistributor) HandleTaskQueueTask(r *http.Request) ([]*distributor.Notification, error) {
	panic("not implemented")
}
322864
func (s *GetRecordsInput) SetShardIterator(v string) *GetRecordsInput {
	s.ShardIterator = &v
	return s
}
1065521
func (c *Ring) ForEachShard(fn func(client *Client) error) error {
	shards := c.shards.List()
	var wg sync.WaitGroup
	errCh := make(chan error, 1)
	for _, shard := range shards {
		if shard.IsDown() {
			continue
		}

		wg.Add(1)
		go func(shard *ringShard) {
			defer wg.Done()
			err := fn(shard.Client)
			if err != nil {
				select {
				case errCh <- err:
				default:
				}
			}
		}(shard)
	}
	wg.Wait()

	select {
	case err := <-errCh:
		return err
	default:
		return nil
	}
}
411840
func (self *Matrix) TryFlushSuccess() {
	if cache.Task.Mode != status.SERVER && cache.Task.SuccessInherit {
		self.history.FlushSuccess(cache.Task.OutType)
	}
}
872008
func (h *Handler) handleDeleteRemoteAvailableShard(w http.ResponseWriter, r *http.Request) {
	if !validHeaderAcceptJSON(r.Header) {
		http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
		return
	}

	indexName := mux.Vars(r)["index"]
	fieldName := mux.Vars(r)["field"]
	shardID, _ := strconv.ParseUint(mux.Vars(r)["shardID"], 10, 64)

	resp := successResponse{h: h}
	err := h.api.DeleteAvailableShard(r.Context(), indexName, fieldName, shardID)
	resp.write(w, err)
}
957958
func (ts *Server) updateShard(ctx context.Context, si *ShardInfo) error {
	span, ctx := trace.NewSpan(ctx, "TopoServer.UpdateShard")
	span.Annotate("keyspace", si.keyspace)
	span.Annotate("shard", si.shardName)
	defer span.Finish()

	data, err := proto.Marshal(si.Shard)
	if err != nil {
		return err
	}
	shardPath := path.Join(KeyspacesPath, si.keyspace, ShardsPath, si.shardName, ShardFile)
	newVersion, err := ts.globalCell.Update(ctx, shardPath, data, si.version)
	if err != nil {
		return err
	}
	si.version = newVersion

	event.Dispatch(&events.ShardChange{
		KeyspaceName: si.Keyspace(),
		ShardName:    si.ShardName(),
		Shard:        si.Shard,
		Status:       "updated",
	})
	return nil
}
329134
func (j *Job) mainHashArgs() []interface{} {
	hashArgs := []interface{}{j.Key(),
		"data", string(j.data),
		"type", j.typ.name,
		"time", j.time,
		"freq", j.freq,
		"priority", j.priority,
		"retries", j.retries,
		"status", j.status,
		"started", j.started,
		"finished", j.finished,
		"poolId", j.poolId,
	}
	if j.err != nil {
		hashArgs = append(hashArgs, "error", j.err.Error())
	}
	return hashArgs
}
965730
func processShard(c context.Context, cfg *Config, timestamp time.Time, shard uint64, loop bool) error {

	logging.Fields{
		"shard": shard,
	}.Infof(c, "Processing tumble shard.")

	q := processShardQuery(c, cfg, shard)
	if q == nil {
		logging.Warningf(c, "dead shard, quitting")
		return nil
	}

	// Calculate our end itme. If we're not looping or we have a <= 0 duration,
	// we will perform a single loop.
	var endTime time.Time
	if cfg.ProcessLoopDuration > 0 {
		endTime = clock.Now(c).Add(time.Duration(cfg.ProcessLoopDuration))
		logging.Debugf(c, "Process loop is configured to exit after [%s] at %s",
			cfg.ProcessLoopDuration.String(), endTime)
	}

	// Lock around the shard that we are trying to modify.
	//
	// Since memcache is namespaced, we don't need to include the namespace in our
	// lock name.
	task := makeProcessTask(timestamp, endTime, shard, loop)
	lockKey := fmt.Sprintf("%s.%d.lock", baseName, shard)
	clientID := fmt.Sprintf("%d_%d_%s", timestamp.Unix(), shard, info.RequestID(c))

	err := memlock.TryWithLock(c, lockKey, clientID, func(c context.Context) error {
		return task.process(c, cfg, q)
	})
	if err == memlock.ErrFailedToLock {
		logging.Infof(c, "Couldn't obtain lock (giving up): %s", err)
		return nil
	}
	return err
}
1198454
func (psh *PubSubHub) WebSocketHandler(w http.ResponseWriter, r *http.Request) {
	wsHandler := websocket.Handler(func(ws *websocket.Conn) {
		// Set the max payload size for this connection.
		ws.MaxPayloadBytes = psh.wsHub.requestLimit

		// Register websocket client.
		ch := psh.wsHub.NewClientHubSpoke()

		// The receive loop will be sitting on websocket.JSON.Receive, while the
		// send loop will be waiting for signals from the WebSocketHub. One must
		// close the other depending on whether the connection was closed/lost,
		// or the WebSocketHub quit or forcibly unregistered the client. The
		// receive loop unregisters the client (thus closing the update signal
		// channel) when the connection is closed and it returns. The send loop
		// closes the websocket.Conn on return, which will interrupt the receive
		// loop from its waiting to receive data on the connection.

		conn := &connection{
			client: ch,
			ws:     ws,
		}

		// Start listening for websocket messages from client, returning when
		// the connection is closed.
		conn.Add(1)
		go psh.receiveLoop(conn)

		// Send loop (ping, new tx, block, etc. update loop)
		conn.Add(1)
		go psh.sendLoop(conn)

		// Hang out until the send and receive loops have quit.
		conn.Wait()
	})

	// Use a websocket.Server to avoid checking Origin.
	wsServer := websocket.Server{
		Handler: wsHandler,
	}
	wsServer.ServeHTTP(w, r)
}
1368669
func (r *Runner) queueHandler(ctx context.Context) {
consumeQueue:
	for {
		select {
		case task := <-r.taskQueue:
			r.handleTask(task)
		case <-ctx.Done():
			break consumeQueue
		}
	}

	// consume remainders
	for len(r.taskQueue) > 0 {
		r.handleTask(<-r.taskQueue)
	}
}
13619
func (worker *copIteratorWorker) handleTaskOnce(bo *Backoffer, task *copTask, ch chan<- *copResponse) ([]*copTask, error) {
	failpoint.Inject("handleTaskOnceError", func(val failpoint.Value) {
		if val.(bool) {
			failpoint.Return(nil, errors.New("mock handleTaskOnce error"))
		}
	})

	sender := NewRegionRequestSender(worker.store.regionCache, worker.store.client)
	req := &tikvrpc.Request{
		Type: task.cmdType,
		Cop: &coprocessor.Request{
			Tp:     worker.req.Tp,
			Data:   worker.req.Data,
			Ranges: task.ranges.toPBRanges(),
		},
		Context: kvrpcpb.Context{
			IsolationLevel: pbIsolationLevel(worker.req.IsolationLevel),
			Priority:       kvPriorityToCommandPri(worker.req.Priority),
			NotFillCache:   worker.req.NotFillCache,
			HandleTime:     true,
			ScanDetail:     true,
		},
	}
	startTime := time.Now()
	resp, rpcCtx, err := sender.SendReqCtx(bo, req, task.region, ReadTimeoutMedium)
	if err != nil {
		return nil, errors.Trace(err)
	}
	// Set task.storeAddr field so its task.String() method have the store address information.
	task.storeAddr = sender.storeAddr
	costTime := time.Since(startTime)
	if costTime > minLogCopTaskTime {
		worker.logTimeCopTask(costTime, task, bo, resp)
	}
	metrics.TiKVCoprocessorHistogram.Observe(costTime.Seconds())

	if task.cmdType == tikvrpc.CmdCopStream {
		return worker.handleCopStreamResult(bo, rpcCtx, resp.CopStream, task, ch)
	}

	// Handles the response for non-streaming copTask.
	return worker.handleCopResponse(bo, rpcCtx, &copResponse{pbResp: resp.Cop}, task, ch, nil)
}
1324846
func Fanout(in interface{}, out ...interface{}) {
	cases := newSendCases(out)
	inChan := reflect.ValueOf(in)

	for {
		data, more := inChan.Recv()
		if !more {
			return
		}
		for i := range cases {
			cases[i].Send = data
		}
		reflect.Select(cases)
	}
}