// bufferDatapoints is responsible for batching incoming datapoints into a buffer
355488
func (w *datapointWorker) bufferFunc(msg *dpMsg) (stop bool) {
	var lastTokenSeen = msg.token
	w.processMsg(msg)
outer:
	for len(w.buffer) < w.batchSize {
		select {
		case msg = <-w.input:
			if msg.token != lastTokenSeen {
				// if the token changes, then emit what ever is in the buffer before proceeding
				w.emit(lastTokenSeen)
				lastTokenSeen = msg.token
			}
			w.processMsg(msg)
		default:
			break outer // emit what ever is in the buffer if there are no more datapoints to read
		}
	}
	// emit the data in the buffer
	w.emit(msg.token)
	return
}
204180
func (t *TagReplace) AddDatapoints(ctx context.Context, points []*datapoint.Datapoint) error {
	return t.next.AddDatapoints(ctx, points)
}
227393
func (r *Rx) PostBuffers(buffers []RxBuffer) bool {
	for i := range buffers {
		b := r.tx.Push(sizeOfPostedBuffer)
		if b == nil {
			r.tx.Abort()
			return false
		}

		pb := &buffers[i]
		binary.LittleEndian.PutUint64(b[postedOffset:], pb.Offset)
		binary.LittleEndian.PutUint32(b[postedSize:], pb.Size)
		binary.LittleEndian.PutUint32(b[postedRemainingInGroup:], 0)
		binary.LittleEndian.PutUint64(b[postedUserData:], pb.UserData)
		binary.LittleEndian.PutUint64(b[postedID:], pb.ID)
	}

	r.tx.Flush()

	return true
}
906059
func (backend *openTSDBBackend) getDataPoints() []dataPoint {
	dataCollector := &dataCollector{
		settings:  backend,
		timestamp: time.Now().Unix(),
	}

	expvar.Do(func(kv expvar.KeyValue) {
		dataCollector.addExpVar(kv)
	})

	return dataCollector.dataPoints
}
1216937
func (connector *Forwarder) AddDatapoints(ctx context.Context, datapoints []*datapoint.Datapoint) error {
	start := time.Now()
	atomic.AddInt64(&connector.stats.pipeline, int64(len(datapoints)))
	defer atomic.AddInt64(&connector.stats.pipeline, -int64(len(datapoints)))
	defer connector.stats.requests.Add(float64(time.Now().Sub(start).Nanoseconds()))
	defer connector.stats.drainSize.Add(float64(len(datapoints)))
	atomic.AddInt64(&connector.stats.totalDatapointsForwarded, int64(len(datapoints)))
	datapoints = connector.emptyMetricNameFilter.FilterDatapoints(datapoints)
	datapoints = connector.FilterDatapoints(datapoints)
	if len(datapoints) == 0 {
		return nil
	}
	return connector.sink.AddDatapoints(ctx, datapoints)
}
1052495
func (w *WithDimensions) AddDatapoints(ctx context.Context, points []*datapoint.Datapoint, next Sink) error {
	return next.AddDatapoints(ctx, w.appendDimensions(points))
}
1005929
func sendDataPoints(data []dataPoint) error {
	json, err := json.Marshal(data)
	if err != nil {
		return err
	}

	resp, err := http.Post(*openTsdbURI, "application/json", bytes.NewReader(json))
	if err != nil {
		return err
	}
	resp.Body.Close()
	return nil
}
286540
func (l *ReaderLexer) BufferString() (str string) {
	guard := Mark("BufferString")
	defer func() {
		Trace("BufferString() -> l.pos = %d, l.buf = %q, return = %q\n", l.pos, l.buf, str)
		guard()
	}()

	Trace("l.buf = %q, l.pos = %d", l.buf, l.pos)

	total := 0
	for i := 0; len(l.buf) > i && i < l.pos+1; i++ {
		Trace("l.buf[%d] = %q\n", i, l.buf[i])
		if l.buf[i] == -1 {
			break
		}
		total += utf8.RuneLen(l.buf[i])
	}

	Trace("Expecting buffer to contain %d bytes", total)
	if total == 0 {
		str = ""
		return
	}

	strbuf := make([]byte, total)
	pos := 0
	for i := 0; len(l.buf) > i && i < l.pos+1; i++ {
		if l.buf[i] == -1 {
			break
		}
		Trace("Encoding rune %q into position %d", l.buf[i], pos)
		pos += utf8.EncodeRune(strbuf[pos:], l.buf[i])
	}
	Trace("%q (%d)\n", strbuf, len(strbuf))

	str = string(strbuf)
	return str
}
1337207
func (f *Facade) GetAllPublicEndpoints(ctx datastore.Context) ([]service.PublicEndpoint, error) {
	defer ctx.Metrics().Stop(ctx.Metrics().Start("GetAllPublicEndpoints"))
	return f.serviceStore.GetAllPublicEndpoints(ctx)
}
666188
func DataPoint(obs float64, vars []float64) *dataPoint {
	return &dataPoint{Observed: obs, Variables: vars}
}
32906
func (t *Term) Buffered() (int, error) {
	var n int
	err := termios.Tiocoutq(uintptr(t.fd), &n)
	return n, err
}
769459
func (w *MultiLevelSkipListWriter) BufferSkip(df int) error {
	assert(df%w.skipInterval == 0)
	numLevels := 1
	df /= w.skipInterval

	// determine max level
	for (df%w.skipMultiplier) == 0 && numLevels < w.numberOfSkipLevels {
		numLevels++
		df /= w.skipMultiplier
	}

	childPointer := int64(0)

	var err error
	for level := 0; level < numLevels; level++ {
		if err = w.spi.WriteSkipData(level, w.skipBuffer[level]); err != nil {
			return err
		}

		newChildPointer := w.skipBuffer[level].FilePointer()

		if level != 0 {
			// store child pointers for all levels except the lowest
			if err = w.skipBuffer[level].WriteVLong(childPointer); err != nil {
				return err
			}
		}

		// remember the childPointer for the next level
		childPointer = newChildPointer
	}
	return nil
}
812438
func (m *RequestCounter) Datapoints() []*datapoint.Datapoint {
	return []*datapoint.Datapoint{
		sfxclient.Cumulative("total_connections", nil, atomic.LoadInt64(&m.TotalConnections)),
		sfxclient.Cumulative("total_time_ns", nil, atomic.LoadInt64(&m.TotalProcessingTimeNs)),
		sfxclient.Gauge("active_connections", nil, atomic.LoadInt64(&m.ActiveConnections)),
	}
}
1294292
func (s *MetricAlarm) SetDatapointsToAlarm(v int64) *MetricAlarm {
	s.DatapointsToAlarm = &v
	return s
}
1220668
func (cmd *ImportCommand) bufferValues(ctx context.Context, useColumnKeys bool, path string) error {
	a := make([]pilosa.FieldValue, 0, cmd.BufferSize)

	var r *csv.Reader

	if path != "-" {
		// Open file for reading.
		f, err := os.Open(path)
		if err != nil {
			return errors.Wrap(err, "opening file")
		}
		defer f.Close()

		// Read rows as bits.
		r = csv.NewReader(f)
	} else {
		r = csv.NewReader(cmd.Stdin)
	}

	r.FieldsPerRecord = -1
	rnum := 0
	for {
		rnum++

		// Read CSV row.
		record, err := r.Read()
		if err == io.EOF {
			break
		} else if err != nil {
			return errors.Wrap(err, "reading")
		}

		// Ignore blank rows.
		if record[0] == "" {
			continue
		} else if len(record) < 2 {
			return fmt.Errorf("bad column count on row %d: col=%d", rnum, len(record))
		}

		var val pilosa.FieldValue

		// Parse column id.
		if useColumnKeys {
			val.ColumnKey = record[0]
		} else {
			if val.ColumnID, err = strconv.ParseUint(record[0], 10, 64); err != nil {
				return fmt.Errorf("invalid column id on row %d: %q", rnum, record[0])
			}
		}

		// Parse FieldValue.
		value, err := strconv.ParseInt(record[1], 10, 64)
		if err != nil {
			return fmt.Errorf("invalid value on row %d: %q", rnum, record[1])
		}
		val.Value = value

		a = append(a, val)

		// If we've reached the buffer size then import FieldValues.
		if len(a) == cmd.BufferSize {
			if err := cmd.importValues(ctx, useColumnKeys, a); err != nil {
				return err
			}
			a = a[:0]
		}
	}

	// If there are still values in the buffer then flush them.
	return cmd.importValues(ctx, useColumnKeys, a)
}
569418
func (qri *QueryRespItem) GetDataPoints() []*DataPoint {
	datapoints := make([]*DataPoint, 0)
	timestampStrs := qri.getSortedTimestampStrs()
	for _, timestampStr := range timestampStrs {
		timestamp, _ := strconv.ParseInt(timestampStr, 10, 64)
		datapoint := &DataPoint{
			Metric:    qri.Metric,
			Value:     qri.Dps[timestampStr],
			Tags:      qri.Tags,
			Timestamp: timestamp,
		}
		datapoints = append(datapoints, datapoint)
	}
	return datapoints
}
1016969
func BufferSize(bs int) Option {
	return func(s *Scroller) error {
		s.bufferSize = bs
		return nil
	}
}