// processData pumps the data out of the provided QueryResultReader.
// It returns any error the source encounters.
19343
func (scw *LegacySplitCloneWorker) processData(ctx context.Context, dbNames []string, td *tabletmanagerdatapb.TableDefinition, tableIndex int, rr ResultReader, rowSplitter *RowSplitter, insertChannels []chan string, destinationPackCount int) error {
	// Store the baseCmd per destination shard because each tablet may have a
	// different dbName.
	baseCmds := make([]string, len(dbNames))
	for i, dbName := range dbNames {
		baseCmds[i] = "INSERT INTO " + sqlescape.EscapeID(dbName) + "." + sqlescape.EscapeID(td.Name) + " (" + strings.Join(escapeAll(td.Columns), ", ") + ") VALUES "
	}
	sr := rowSplitter.StartSplit()
	packCount := 0

	fields := rr.Fields()
	for {
		r, err := rr.Next()
		if err != nil {
			// we are done, see if there was an error
			if err != io.EOF {
				return err
			}

			// send the remainder if any (ignoring
			// the return value, we don't care
			// here if we're aborted)
			if packCount > 0 {
				rowSplitter.Send(fields, sr, baseCmds, insertChannels, ctx.Done())
			}
			return nil
		}

		// Split the rows by keyspace ID, and insert each chunk into each destination
		if err := rowSplitter.Split(sr, r.Rows); err != nil {
			return vterrors.Wrapf(err, "RowSplitter failed for table %v", td.Name)
		}
		scw.tableStatusList.addCopiedRows(tableIndex, len(r.Rows))

		// see if we reach the destination pack count
		packCount++
		if packCount < destinationPackCount {
			continue
		}

		// send the rows to be inserted
		if aborted := rowSplitter.Send(fields, sr, baseCmds, insertChannels, ctx.Done()); aborted {
			return nil
		}

		// and reset our row buffer
		sr = rowSplitter.StartSplit()
		packCount = 0
	}
}
485458
func ResultHandler(w http.ResponseWriter, r *http.Request) {
	var (
		status int
		err    error
	)
	defer func() {
		if nil != err {
			http.Error(w, err.Error(), status)
		}
	}()
	val := r.Context().Value(ctxDBKey)
	if val == nil {
		httpError(w, r, http.StatusInternalServerError,
			fmt.Sprintf("Could not find database handler in request context"))
		return
	}

	db := val.(*pg.DB)

	idStr := r.FormValue("id")

	id, err := strconv.ParseInt(idStr, 10, 64)
	if err != nil {
		httpError(w, r, http.StatusBadRequest,
			fmt.Sprintf("Could not parse provided scan id %q: %v", idStr, err))
		return
	}

	scan, err := db.GetScanByID(id)
	if err != nil {
		httpError(w, r, http.StatusInternalServerError,
			fmt.Sprintf("Failed to retrieve scan id %d from database: %v", id, err))
		return
	}

	if scan.ID == -1 {
		httpError(w, r, http.StatusBadRequest,
			fmt.Sprintf("Failed to find scan id %d in database", id))
		return
	}

	// display the analysis results in alphabetical order of worker name
	sort.Sort(scan.AnalysisResults)

	jsScan, err := json.Marshal(scan)
	if err != nil {
		httpError(w, r, http.StatusInternalServerError,
			fmt.Sprintf("Failed to convert scan data into JSON: %v", err))
		return
	}
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusOK)
	fmt.Fprint(w, string(jsScan))
}
1251774
func (client *QueryClient) StreamExecute(query string, bindvars map[string]*querypb.BindVariable) (*sqltypes.Result, error) {
	return client.StreamExecuteWithOptions(query, bindvars, &querypb.ExecuteOptions{IncludedFields: querypb.ExecuteOptions_ALL})
}
1169586
func (_class SRClass) QueryDataSource(sessionID SessionRef, sr SRRef, dataSource string) (_retval float64, _err error) {
	_method := "SR.query_data_source"
	_sessionIDArg, _err := convertSessionRefToXen(fmt.Sprintf("%s(%s)", _method, "session_id"), sessionID)
	if _err != nil {
		return
	}
	_srArg, _err := convertSRRefToXen(fmt.Sprintf("%s(%s)", _method, "sr"), sr)
	if _err != nil {
		return
	}
	_dataSourceArg, _err := convertStringToXen(fmt.Sprintf("%s(%s)", _method, "data_source"), dataSource)
	if _err != nil {
		return
	}
	_result, _err := _class.client.APICall(_method, _sessionIDArg, _srArg, _dataSourceArg)
	if _err != nil {
		return
	}
	_retval, _err = convertFloatToGo(_method + " -> ", _result.Value)
	return
}
472066
func process(workload Workload, result *Result) (err error) {

	c := make(chan int, len(workload.Requests))
	result.Responses = make([]Response, len(workload.Requests))

	for index, _ := range workload.Requests {

		if workload.UseHeaders {
			replaceHeaderValues(&workload.Requests[index].Header, &workload.header)
		}

		if workload.StrictOrder {
			syncRequest(&workload.Requests[index], &result.Responses[index])
		} else {
			go asyncRequest(&workload.Requests[index], &result.Responses[index], c)
		}
	}

	// if async, wait for our responses or timeout
	if !workload.StrictOrder {

		// create a timeout so we don't wait forever
		var timeout <-chan time.Time

		if workload.Timeout > 0 {
			timeout = time.After(time.Duration(workload.Timeout))
		} else {
			timeout = time.After(DefaultTimeout)
		}

		for index := 0; index < len(workload.Requests); index++ {
			select {
			case <-c:
			case <-timeout:
				log.Warn("[process] Timed out waiting for all go routines to complete")
				break
			}
		}
	}
	return
}
286954
func (c *Connection) processRequest(ch *Channel, req *vppRequest) error {
	// check whether we are connected to VPP
	if atomic.LoadUint32(&c.vppConnected) == 0 {
		err := ErrNotConnected
		log.Errorf("processing request failed: %v", err)
		return err
	}

	// retrieve message ID
	msgID, err := c.GetMessageID(req.msg)
	if err != nil {
		log.WithFields(logger.Fields{
			"msg_name": req.msg.GetMessageName(),
			"msg_crc":  req.msg.GetCrcString(),
			"seq_num":  req.seqNum,
			"error":    err,
		}).Errorf("failed to retrieve message ID")
		return fmt.Errorf("unable to retrieve message ID: %v", err)
	}

	// encode the message into binary
	data, err := c.codec.EncodeMsg(req.msg, msgID)
	if err != nil {
		log.WithFields(logger.Fields{
			"channel":  ch.id,
			"msg_id":   msgID,
			"msg_name": req.msg.GetMessageName(),
			"seq_num":  req.seqNum,
			"error":    err,
		}).Errorf("failed to encode message: %#v", req.msg)
		return fmt.Errorf("unable to encode the message: %v", err)
	}

	context := packRequestContext(ch.id, req.multi, req.seqNum)

	if log.Level == logger.DebugLevel { // for performance reasons - logrus does some processing even if debugs are disabled
		log.WithFields(logger.Fields{
			"channel":  ch.id,
			"context":  context,
			"is_multi": req.multi,
			"msg_id":   msgID,
			"msg_size": len(data),
			"seq_num":  req.seqNum,
			"msg_crc":  req.msg.GetCrcString(),
		}).Debugf("--> govpp send: %s: %+v", req.msg.GetMessageName(), req.msg)
	}

	// send the request to VPP
	err = c.vppClient.SendMsg(context, data)
	if err != nil {
		err = fmt.Errorf("unable to send the message: %v", err)
		log.WithFields(logger.Fields{
			"context": context,
			"msg_id":  msgID,
			"seq_num": req.seqNum,
		}).Error(err)
		return err
	}

	if req.multi {
		// send a control ping to determine end of the multipart response
		pingData, _ := c.codec.EncodeMsg(msgControlPing, c.pingReqID)

		log.WithFields(logger.Fields{
			"channel":  ch.id,
			"context":  context,
			"msg_id":   c.pingReqID,
			"msg_size": len(pingData),
			"seq_num":  req.seqNum,
		}).Debug("  -> sending control ping")

		if err := c.vppClient.SendMsg(context, pingData); err != nil {
			log.WithFields(logger.Fields{
				"context": context,
				"msg_id":  msgID,
				"seq_num": req.seqNum,
			}).Warnf("unable to send control ping: %v", err)
		}
	}

	return nil
}
273827
func (c *Client) processStream(baseURL *url.URL, path string, authHandler AuthHandler, callb func(interface{}) error, data interface{}) error {
	var requestUrl string

	if q := strings.Index(path, "?"); q > 0 {
		requestUrl = baseURL.Scheme + "://" + baseURL.Host + path[:q] + "?" + path[q+1:]
	} else {
		requestUrl = baseURL.Scheme + "://" + baseURL.Host + path
	}

	req, err := http.NewRequest("GET", requestUrl, nil)
	if err != nil {
		return err
	}

	err = maybeAddAuth(req, authHandler)
	if err != nil {
		return err
	}

	res, err := doHTTPRequest(req)
	if err != nil {
		return err
	}

	defer res.Body.Close()
	if res.StatusCode != 200 {
		bod, _ := ioutil.ReadAll(io.LimitReader(res.Body, 512))
		return fmt.Errorf("HTTP error %v getting %q: %s",
			res.Status, requestUrl, bod)
	}

	reader := bufio.NewReader(res.Body)
	for {
		bs, err := reader.ReadBytes('\n')
		if err != nil {
			return err
		}
		if len(bs) == 1 && bs[0] == '\n' {
			continue
		}

		err = json.Unmarshal(bs, data)
		if err != nil {
			return err
		}
		err = callb(data)
		if err != nil {
			return err
		}
	}
	return nil

}
82857
func Process(c http.ResponseWriter, req *http.Request) {
	mainServer.Process(c, req)
}
477387
func (s *StreamFramer) readData() []byte {
	// Compute the amount to read from the buffer
	size := s.data.Len()
	if size > s.frameSize {
		size = s.frameSize
	}
	if size == 0 {
		return nil
	}
	d := s.data.Next(size)
	return d
}
631152
func WithProcessCacheData(c context.Context, data *ProcessCacheData) context.Context {
	if data.caches == nil {
		panic("use NewProcessCacheData to allocate ProcessCacheData")
	}
	return context.WithValue(c, &processCacheKey, data)
}
1214884
func (cc *clientConn) handleLoadData(ctx context.Context, loadDataInfo *executor.LoadDataInfo) error {
	// If the server handles the load data request, the client has to set the ClientLocalFiles capability.
	if cc.capability&mysql.ClientLocalFiles == 0 {
		return errNotAllowedCommand
	}
	if loadDataInfo == nil {
		return errors.New("load data info is empty")
	}

	err := cc.writeReq(loadDataInfo.Path)
	if err != nil {
		return err
	}

	var shouldBreak bool
	var prevData, curData []byte
	// TODO: Make the loadDataRowCnt settable.
	loadDataInfo.SetMaxRowsInBatch(defaultLoadDataBatchCnt)
	err = loadDataInfo.Ctx.NewTxn(ctx)
	if err != nil {
		return err
	}
	for {
		curData, err = cc.readPacket()
		if err != nil {
			if terror.ErrorNotEqual(err, io.EOF) {
				logutil.Logger(ctx).Error("read packet failed", zap.Error(err))
				break
			}
		}
		if len(curData) == 0 {
			shouldBreak = true
			if len(prevData) == 0 {
				break
			}
		}
		prevData, err = insertDataWithCommit(ctx, prevData, curData, loadDataInfo)
		if err != nil {
			break
		}
		if shouldBreak {
			break
		}
	}
	loadDataInfo.SetMessage()

	if err != nil {
		loadDataInfo.Ctx.StmtRollback()
	} else {
		err = loadDataInfo.Ctx.StmtCommit()
	}

	var txn kv.Transaction
	var err1 error
	txn, err1 = loadDataInfo.Ctx.Txn(true)
	if err1 == nil {
		if txn != nil && txn.Valid() {
			if err != nil {
				if err1 := txn.Rollback(); err1 != nil {
					logutil.Logger(ctx).Error("load data rollback failed", zap.Error(err1))
				}
				return err
			}
			return cc.ctx.CommitTxn(sessionctx.SetCommitCtx(ctx, loadDataInfo.Ctx))
		}
	}
	// Should never reach here.
	panic(err1)
}
705280
func (l *Linode) Process() ([]*Response, error) {
	// Quickly handle the case where we have zero or one actions to perform.
	count := len(l.actions)
	if count == 0 {
		return nil, NewError(errors.New("linode: request must contain at least one action"))
	}
	defer l.Reset()
	if count == 1 {
		resp, err := l.Request(l.actions[0].parameters.Get(apiAction), l.actions[0].parameters, l.actions[0].result)
		if resp == nil {
			return nil, err
		}
		return []*Response{resp}, err
	}

	// Prepare the parameters.
	params := make([]Parameters, 0, len(l.actions))
	for _, action := range l.actions {
		params = append(params, action.parameters)
	}
	jsonParams, err := json.Marshal(params)
	if err != nil {
		return nil, NewError(err)
	}

	// Initialize the request.
	req, err := http.NewRequest("GET", l.apiEndpoint, nil)
	if err != nil {
		return nil, NewError(err)
	}
	query := Parameters{
		apiKey:          l.apiKey,
		apiAction:       "batch",
		apiRequestArray: string(jsonParams),
	}
	req.URL.RawQuery = query.Encode()

	// Do the request.
	resp, err := client.Do(req)
	if err != nil {
		return nil, NewError(err)
	}
	defer resp.Body.Close()

	// Decode the response.
	var response []*Response
	decoder := json.NewDecoder(resp.Body)
	if err = decoder.Decode(&response); err != nil {
		return nil, NewError(err)
	}

	// Decode the raw data into action-specific data.
	// FIXME: This whole block needs error handling, and I think since this
	// is handling multiple actions, it should probably handle errors on a
	// per-action basis.
	for index, action := range response {
		if l.actions[index].result == nil {
			continue
		}
		if err = json.Unmarshal(action.RawData, &l.actions[index].result); err != nil {
			continue
		}
		response[index].Data = l.actions[index].result
	}

	return response, nil
}
898748
func (s *Server) processRequestStream(ctx context.Context, streamLog *logrus.Entry, stream Stream,
	reqCh <-chan *envoy_api_v2.DiscoveryRequest, defaultTypeURL string) error {
	// The request state for every type URL.
	typeStates := make([]perTypeStreamState, len(s.watchers))
	defer func() {
		for _, state := range typeStates {
			if state.pendingWatchCancel != nil {
				state.pendingWatchCancel()
			}
		}
	}()

	// A map of a resource type's URL to the corresponding index in typeStates
	// for the resource type.
	typeIndexes := make(map[string]int, len(typeStates))

	// The set of channels to select from. Since the set of channels is
	// dynamic, we use reflection for selection.
	// The indexes in selectCases from 0 to len(typeStates)-1 match the indexes
	// in typeStates.
	selectCases := make([]reflect.SelectCase, len(typeStates)+2)

	// The last select case index is always the request channel.
	reqChIndex := len(selectCases) - 1
	selectCases[reqChIndex] = reflect.SelectCase{
		Dir:  reflect.SelectRecv,
		Chan: reflect.ValueOf(reqCh),
	}

	// The next-to-last select case is the context's Done channel.
	doneChIndex := reqChIndex - 1
	selectCases[doneChIndex] = reflect.SelectCase{
		Dir:  reflect.SelectRecv,
		Chan: reflect.ValueOf(ctx.Done()),
	}

	// Initially there are no pending watches, so just select a dead channel
	// that will never be selected.
	quietCh := make(chan *VersionedResources)
	defer close(quietCh)
	quietChValue := reflect.ValueOf(quietCh)

	i := 0
	for typeURL := range s.watchers {
		typeStates[i] = perTypeStreamState{
			typeURL: typeURL,
		}

		selectCases[i] = reflect.SelectCase{
			Dir:  reflect.SelectRecv,
			Chan: quietChValue,
		}

		typeIndexes[typeURL] = i

		i++
	}

	streamLog.Info("starting xDS stream processing")

	for {
		// Process either a new request from the xDS stream or a response
		// from the resource watcher.
		chosen, recv, recvOK := reflect.Select(selectCases)

		switch chosen {
		case doneChIndex: // Context got canceled.
			streamLog.WithError(ctx.Err()).Error("xDS stream context canceled")
			return ctx.Err()

		case reqChIndex: // Request received from the stream.
			if !recvOK {
				streamLog.Info("xDS stream closed")
				return nil
			}

			req := recv.Interface().(*envoy_api_v2.DiscoveryRequest)

			requestLog := streamLog.WithFields(getXDSRequestFields(req))

			// Ensure that the version info is a string that was sent by this
			// server or the empty string (the first request in a stream should
			// always have an empty version info).
			var versionInfo uint64
			if req.GetVersionInfo() != "" {
				var err error
				versionInfo, err = strconv.ParseUint(req.VersionInfo, 10, 64)
				if err != nil {
					requestLog.Errorf("invalid version info in xDS request, not a uint64")
					return ErrInvalidVersionInfo
				}
			}
			var nonce uint64
			if req.GetResponseNonce() != "" {
				var err error
				nonce, err = strconv.ParseUint(req.ResponseNonce, 10, 64)
				if err != nil {
					requestLog.Error("invalid response nonce info in xDS request, not a uint64")
					return ErrInvalidResponseNonce
				}
			}
			var detail string
			status := req.GetErrorDetail()
			if status != nil {
				detail = status.Message
			}

			typeURL := req.GetTypeUrl()
			if defaultTypeURL == AnyTypeURL && typeURL == "" {
				requestLog.Error("no type URL given in ADS request")
				return ErrNoADSTypeURL
			}

			index, exists := typeIndexes[typeURL]

			if !exists {
				requestLog.Error("unknown type URL in xDS request")
				return ErrUnknownTypeURL
			}

			state := &typeStates[index]

			// Response nonce is always the same as the response version.
			// Request version indicates the last acked version. If the
			// response nonce in the request is different (smaller) than
			// the version, all versions upto that version are acked, but
			// the versions from that to and including the nonce are nacked.
			if versionInfo <= nonce {
				ackObserver := s.ackObservers[typeURL]
				// ACK versions up to the received versionInfo
				if ackObserver != nil {
					requestLog.Debug("notifying observers of ACKs")
					ackObserver.HandleResourceVersionAck(versionInfo, nonce, req.GetNode(), state.resourceNames, typeURL, detail)
					if versionInfo < nonce {
						// versions after VersionInfo, upto and including ResponseNonce are NACKed
						requestLog.Warningf("NACK received for versions after %s and up to %s; waiting for a version update before sending again", req.VersionInfo, req.ResponseNonce)
						// Watcher will behave as if the sent version was acked.
						// Otherwise we will just be sending the same failing
						// version over and over filling logs.
						versionInfo = state.version
					}

				} else {
					requestLog.Debug("ACK received but no observers are waiting for ACKs")
				}
				if state.pendingWatchCancel != nil {
					// A pending watch exists for this type URL. Cancel it to
					// start a new watch.
					requestLog.Debug("canceling pending watch")
					state.pendingWatchCancel()
				}

				respCh := make(chan *VersionedResources, 1)
				selectCases[index].Chan = reflect.ValueOf(respCh)

				ctx, cancel := context.WithCancel(ctx)
				state.pendingWatchCancel = cancel

				requestLog.Debugf("starting watch on %d resources", len(req.GetResourceNames()))
				go s.watchers[typeURL].WatchResources(ctx, typeURL, versionInfo,
					req.GetNode(), req.GetResourceNames(), respCh)
			} else {
				requestLog.Debug("received invalid nonce in xDS request; ignoring request")
			}
		default: // Pending watch response.
			state := &typeStates[chosen]
			state.pendingWatchCancel()
			state.pendingWatchCancel = nil

			if !recvOK {
				streamLog.WithField(logfields.XDSTypeURL, state.typeURL).
					Error("xDS resource watch failed; terminating")
				return ErrResourceWatch
			}

			// Disabling reading from the channel after reading any from it,
			// since the watcher will close it anyway.
			selectCases[chosen].Chan = quietChValue

			resp := recv.Interface().(*VersionedResources)

			responseLog := streamLog.WithFields(logrus.Fields{
				logfields.XDSVersionInfo: resp.Version,
				logfields.XDSCanary:      resp.Canary,
				logfields.XDSTypeURL:     state.typeURL,
				logfields.XDSNonce:       resp.Version,
			})

			resources := make([]*any.Any, len(resp.Resources))

			// Marshall the resources into protobuf's Any type.
			for i, res := range resp.Resources {
				data, err := proto.Marshal(res)
				if err != nil {
					responseLog.WithError(err).Errorf("error marshalling xDS response (%d resources)", len(resp.Resources))
					return err
				}
				resources[i] = &any.Any{
					TypeUrl: state.typeURL,
					Value:   data,
				}
			}

			responseLog.Debugf("sending xDS response with %d resources", len(resp.Resources))

			versionStr := strconv.FormatUint(resp.Version, 10)
			out := &envoy_api_v2.DiscoveryResponse{
				VersionInfo: versionStr,
				Resources:   resources,
				Canary:      resp.Canary,
				TypeUrl:     state.typeURL,
				Nonce:       versionStr,
			}
			err := stream.Send(out)
			if err != nil {
				return err
			}

			state.version = resp.Version
			state.resourceNames = resp.ResourceNames
		}
	}
}
1253624
func (p *JSONFormat) TransformData(data interface{}) (string, error) {
	var buf bytes.Buffer
	enc := codec.NewEncoder(&buf, jsonHandlePretty)
	err := enc.Encode(data)
	if err != nil {
		return "", err
	}

	return buf.String(), nil
}
1187486
func Process(out io.Writer, in io.Reader, opts ...Option) error {
	e := embedder{Fetcher: fetcher{}}
	for _, opt := range opts {
		opt.f(&e)
	}
	return process(out, in, e.runCommand)
}
848041
func (h *QueryHandler) handlePostQuery(w http.ResponseWriter, r *http.Request) {
	ctx := r.Context()
	var req query.Request
	req.WithCompilerMappings(h.CompilerMappings)
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		EncodeError(ctx, err, w)
		return
	}

	results, err := h.QueryService.Query(ctx, &req)
	if err != nil {
		EncodeError(ctx, err, w)
		return
	}
	// Always cancel the results to free resources.
	// If all results were consumed cancelling does nothing.
	defer results.Release()

	// Setup headers
	stats, hasStats := results.(flux.Statisticser)
	if hasStats {
		w.Header().Set("Trailer", queryStatisticsTrailer)
	}

	// NOTE: We do not write out the headers here.
	// It is possible that if the encoding step fails
	// that we can write an error header so long as
	// the encoder did not write anything.
	// As such we rely on the http.ResponseWriter behavior
	// to write an StatusOK header with the first write.

	switch r.Header.Get("Accept") {
	case "text/csv":
		fallthrough
	default:
		h.csvDialect.SetHeaders(w)
		encoder := h.csvDialect.Encoder()
		n, err := encoder.Encode(w, results)
		if err != nil {
			if n == 0 {
				// If the encoder did not write anything, we can write an error header.
				EncodeError(ctx, err, w)
			} else {
				h.Logger.Info("Failed to encode client response",
					zap.Error(err),
				)
			}
		}
	}

	if hasStats {
		data, err := json.Marshal(stats.Statistics())
		if err != nil {
			h.Logger.Info("Failed to encode statistics", zap.Error(err))
			return
		}
		// Write statisitcs trailer
		w.Header().Set(queryStatisticsTrailer, string(data))
	}
}
804321
func (g *Generator) ProcessRequest(req *CertificateRequest) (csr, key []byte, err error) {

	log.Info("generate received request")
	err = g.Validator(req)
	if err != nil {
		log.Warningf("invalid request: %v", err)
		return nil, nil, err
	}

	csr, key, err = ParseRequest(req)
	if err != nil {
		return nil, nil, err
	}
	return
}