// Throttle Writer silently throws away messages over the limit.
329237
func (t ThrottleWriter) Write(p []byte) (n int, err error) {

	now := time.Now().UTC()
	if now.Sub(t.Start) < t.Within {
		if t.Count > t.Limit {
			if t.Callback != nil {
				t.Callback()
			}
			return 0, nil
		}
		t.Count++
	} else {
		t.Start = now
		t.Count = 0
	}

	return t.Writer.Write(p)
}
95898
func (d *Driver) Throttle(speed float32) {
	d.mutex.Lock()
	defer d.mutex.Unlock()
	d.cmd[2] = floatToCmdByte(speed, 0x7e, 0x7e)
	d.cmd[10] = checksum(d.cmd)
}
505681
func (engine *Engine) Limit(limit int, start ...int) *Session {
	session := engine.NewSession()
	session.IsAutoClose = true
	return session.Limit(limit, start...)
}
1158851
func WriteMessage(w io.Writer, m *Message) error {
	_, err := w.Write(m.raw)
	return err
}
449398
func NewWriterCallback(onWrite CountCallback, writer io.Writer) *Writer {
	return &Writer{
		writer:  writer,
		onWrite: onWrite,
	}
}
1200593
func (cn *connection) writer(keepAliveTimeout time.Duration) {
	var (
		lastWrite      time.Time = time.Now()
		keepAliveTimer *time.Timer
	)
	keepAliveTimer = time.AfterFunc(keepAliveTimeout, func() {
		cn.mu().Lock()
		defer cn.mu().Unlock()
		if time.Since(lastWrite) >= keepAliveTimeout {
			cn.tickleWriter()
		}
		keepAliveTimer.Reset(keepAliveTimeout)
	})
	cn.mu().Lock()
	defer cn.mu().Unlock()
	defer cn.Close()
	defer keepAliveTimer.Stop()
	frontBuf := new(bytes.Buffer)
	for {
		if cn.closed.IsSet() {
			return
		}
		if cn.writeBuffer.Len() == 0 {
			cn.fillWriteBuffer(func(msg pp.Message) bool {
				cn.wroteMsg(&msg)
				cn.writeBuffer.Write(msg.MustMarshalBinary())
				torrent.Add(fmt.Sprintf("messages filled of type %s", msg.Type.String()), 1)
				return cn.writeBuffer.Len() < 1<<16 // 64KiB
			})
		}
		if cn.writeBuffer.Len() == 0 && time.Since(lastWrite) >= keepAliveTimeout {
			cn.writeBuffer.Write(pp.Message{Keepalive: true}.MustMarshalBinary())
			postedKeepalives.Add(1)
		}
		if cn.writeBuffer.Len() == 0 {
			// TODO: Minimize wakeups....
			cn.writerCond.Wait()
			continue
		}
		// Flip the buffers.
		frontBuf, cn.writeBuffer = cn.writeBuffer, frontBuf
		cn.mu().Unlock()
		n, err := cn.w.Write(frontBuf.Bytes())
		cn.mu().Lock()
		if n != 0 {
			lastWrite = time.Now()
			keepAliveTimer.Reset(keepAliveTimeout)
		}
		if err != nil {
			return
		}
		if n != frontBuf.Len() {
			panic("short write")
		}
		frontBuf.Reset()
	}
}
194879
func (p *IOThrottlerPool) AddReadWriter(readWriter io.ReadWriteCloser) (io.ReadWriteCloser, error) {
	// Register with the pool
	err := twiddleClientCount(p, 1)
	if err != nil {
		return nil, err
	}

	return &throttledReadWriteCloser{throttledReadCloser{readWriter, p},
		throttledWriteCloser{readWriter, p}}, err
}
940416
func (log logNull) Write(message []byte) (int, error) {
	return len(message), nil
}
746252
func (writer GlogWriter) Write(data []byte) (n int, err error) {
	glog.Info(string(data))
	return len(data), nil
}
153112
func (p *IOThrottlerPool) AddWriter(writer io.WriteCloser) (io.WriteCloser, error) {
	// Register with the pool
	err := twiddleClientCount(p, 1)
	if err != nil {
		return nil, err
	}

	return &throttledWriteCloser{writer, p}, nil
}
581266
func throttled(threshold int64, h http.Handler) http.Handler {
	var active int64

	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		current := atomic.AddInt64(&active, 1)
		defer atomic.AddInt64(&active, -1)

		if current-1 >= threshold {
			log.Warnf("Reached threshold of %d concurrent active clients", threshold)
			http.Error(w, "429 Too Many Requests", 429)
			return
		}

		h.ServeHTTP(w, r)
	})
}
1235998
func (l *Limiter) SetOnLimitReached(fn func(w http.ResponseWriter, r *http.Request)) *Limiter {
	l.Lock()
	l.onLimitReached = fn
	l.Unlock()

	return l
}
1326731
func (s *Server) MaxRates(_ context.Context, request *throttlerdatapb.MaxRatesRequest) (_ *throttlerdatapb.MaxRatesResponse, err error) {
	defer servenv.HandlePanic("throttler", &err)

	rates := s.manager.MaxRates()
	return &throttlerdatapb.MaxRatesResponse{
		Rates: rates,
	}, nil
}
862652
func (rl *RateLimiter) Limit() bool {
	// Calculate the number of ns that have passed since our last call
	now := unixNano()
	passed := now - atomic.SwapUint64(&rl.lastCheck, now)

	// Add them to our allowance
	rate := atomic.LoadUint64(&rl.rate)
	current := atomic.AddUint64(&rl.allowance, passed*rate)

	// Ensure our allowance is not over maximum
	if max := atomic.LoadUint64(&rl.max); current > max {
		atomic.AddUint64(&rl.allowance, max-current)
		current = max
	}

	// If our allowance is less than one unit, rate-limit!
	if current < rl.unit {
		return true
	}

	// Not limited, subtract a unit
	atomic.AddUint64(&rl.allowance, -rl.unit)
	return false
}
14617
func RateLimit(q Quota, vary *VaryBy, store GCRAStore) *Throttler {
	count, period := q.Quota()

	if count < 1 {
		count = 1
	}
	if period <= 0 {
		period = time.Second
	}

	rate := Rate{period: period / time.Duration(count)}
	limiter, err := NewGCRARateLimiter(store, RateQuota{rate, count - 1})

	// This panic in unavoidable because the original interface does
	// not support returning an error.
	if err != nil {
		panic(err)
	}

	return &Throttler{
		HTTPRateLimiter{
			RateLimiter: limiter,
			VaryBy:      vary,
		},
	}
}
88335
func (g *Generator) DropLinuxResourcesBlockIOThrottleWriteBpsDevice(major int64, minor int64) {
	if g.Config == nil || g.Config.Linux == nil || g.Config.Linux.Resources == nil || g.Config.Linux.Resources.BlockIO == nil {
		return
	}

	throttleDevices := dropBlockIOThrottleDevice(g.Config.Linux.Resources.BlockIO.ThrottleWriteBpsDevice, major, minor)
	g.Config.Linux.Resources.BlockIO.ThrottleWriteBpsDevice = throttleDevices
}
1193070
func newMessageWriter(w http.ResponseWriter, getProgressFunc func() (bytesScanned, bytesProcessed int64)) *messageWriter {
	writer := &messageWriter{
		writer:          w,
		getProgressFunc: getProgressFunc,

		payloadBuffer: make([]byte, bufLength),
		payloadCh:     make(chan []byte),

		errCh:  make(chan []byte),
		doneCh: make(chan struct{}),
	}
	go writer.start()
	return writer
}