|
- // Copyright 2015 Google Inc. All rights reserved.
- //
- // Licensed under the Apache License, Version 2.0 (the "License");
- // you may not use this file except in compliance with the License.
- // You may obtain a copy of the License at
- //
- // http://www.apache.org/licenses/LICENSE-2.0
- //
- // Unless required by applicable law or agreed to in writing, software
- // distributed under the License is distributed on an "AS IS" BASIS,
- // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- // See the License for the specific language governing permissions and
- // limitations under the License.
-
- package trafficshape
-
- import (
- "io"
- "net"
- "sort"
- "sync"
- "time"
-
- "github.com/google/martian/log"
- )
-
- // ErrForceClose is an error that communicates the need to close the connection.
- type ErrForceClose struct {
- message string
- }
-
- func (efc *ErrForceClose) Error() string {
- return efc.message
- }
-
- // DefaultBitrate represents the bitrate that will be for all url regexs for which a shape
- // has not been specified.
- var DefaultBitrate int64 = 500000000000 // 500Gbps (unlimited)
-
- // urlShape contains a rw lock protected shape of a url_regex.
- type urlShape struct {
- sync.RWMutex
- Shape *Shape
- }
-
- // urlShapes contains a rw lock protected map of url regexs to their URLShapes.
- type urlShapes struct {
- sync.RWMutex
- M map[string]*urlShape
- LastModifiedTime time.Time
- }
-
- // Buckets contains the read and write buckets for a url_regex.
- type Buckets struct {
- ReadBucket *Bucket
- WriteBucket *Bucket
- }
-
- // NewBuckets returns a *Buckets with the specified up and down bandwidths.
- func NewBuckets(up int64, down int64) *Buckets {
- return &Buckets{
- ReadBucket: NewBucket(up, time.Second),
- WriteBucket: NewBucket(down, time.Second),
- }
- }
-
- // ThrottleContext represents whether we are currently in a throttle interval for a particular
- // url_regex. If ThrottleNow is true, only then will the current throttle 'Bandwidth' be set
- // correctly.
- type ThrottleContext struct {
- ThrottleNow bool
- Bandwidth int64
- }
-
- // NextActionInfo represents whether there is an upcoming action. Only if ActionNext is true will the
- // Index and ByteOffset be set correctly.
- type NextActionInfo struct {
- ActionNext bool
- Index int64
- ByteOffset int64
- }
-
- // Context represents the current information that is needed while writing back to the client.
- // Only if Shaping is true, that is we are currently writing back a response that matches a certain
- // url_regex will the other values be set correctly. If so, the Buckets represent the buckets
- // to be used for the current url_regex. NextActionInfo tells us whether there is an upcoming action
- // that needs to be performed, and ThrottleContext tells us whether we are currently in a throttle
- // interval (according to the RangeStart). Note, the ThrottleContext is only used once in the start
- // to determine the beginning bandwidth. It need not be updated after that. This
- // is because the subsequent throttles are captured in the upcoming ChangeBandwidth actions.
- // Byte Offset represents the absolute byte offset of response data that we are currently writing back.
- // It does not account for the header data.
- type Context struct {
- Shaping bool
- RangeStart int64
- URLRegex string
- Buckets *Buckets
- GlobalBucket *Bucket
- ThrottleContext *ThrottleContext
- NextActionInfo *NextActionInfo
- ByteOffset int64
- HeaderLen int64
- HeaderBytesWritten int64
- }
-
- // Listener wraps a net.Listener and simulates connection latency and bandwidth
- // constraints.
- type Listener struct {
- net.Listener
-
- ReadBucket *Bucket
- WriteBucket *Bucket
-
- mu sync.RWMutex
- latency time.Duration
- GlobalBuckets map[string]*Bucket
- Shapes *urlShapes
- defaults *Default
- }
-
- // Conn wraps a net.Conn and simulates connection latency and bandwidth
- // constraints. Shapes represents the traffic shape map inherited from the listener.
- // Established is the time that this connection was established. LocalBuckets represents a map from
- // the url_regexes to their dedicated buckets.
- type Conn struct {
- net.Conn
- ReadBucket *Bucket // Shared by listener.
- WriteBucket *Bucket // Shared by listener.
- latency time.Duration
- ronce sync.Once
- wonce sync.Once
- Shapes *urlShapes
- GlobalBuckets map[string]*Bucket
- LocalBuckets map[string]*Buckets
- Established time.Time
- Context *Context
- DefaultBandwidth Bandwidth
- Listener *Listener
- }
-
- // NewListener returns a new bandwidth constrained listener. Defaults to
- // DefaultBitrate (uncapped).
- func NewListener(l net.Listener) *Listener {
- return &Listener{
- Listener: l,
- ReadBucket: NewBucket(DefaultBitrate/8, time.Second),
- WriteBucket: NewBucket(DefaultBitrate/8, time.Second),
- Shapes: &urlShapes{M: make(map[string]*urlShape)},
- GlobalBuckets: make(map[string]*Bucket),
- defaults: &Default{
- Bandwidth: Bandwidth{
- Up: DefaultBitrate / 8,
- Down: DefaultBitrate / 8,
- },
- Latency: 0,
- },
- }
- }
-
- // ReadBitrate returns the bitrate in bits per second for reads.
- func (l *Listener) ReadBitrate() int64 {
- return l.ReadBucket.Capacity() * 8
- }
-
- // SetReadBitrate sets the bitrate in bits per second for reads.
- func (l *Listener) SetReadBitrate(bitrate int64) {
- l.ReadBucket.SetCapacity(bitrate / 8)
- }
-
- // WriteBitrate returns the bitrate in bits per second for writes.
- func (l *Listener) WriteBitrate() int64 {
- return l.WriteBucket.Capacity() * 8
- }
-
- // SetWriteBitrate sets the bitrate in bits per second for writes.
- func (l *Listener) SetWriteBitrate(bitrate int64) {
- l.WriteBucket.SetCapacity(bitrate / 8)
- }
-
- // SetDefaults sets the default traffic shaping parameters for the listener.
- func (l *Listener) SetDefaults(defaults *Default) {
- l.mu.Lock()
- defer l.mu.Unlock()
-
- l.defaults = defaults
- }
-
- // Defaults returns the default traffic shaping parameters for the listener.
- func (l *Listener) Defaults() *Default {
- l.mu.RLock()
- defer l.mu.RUnlock()
-
- return l.defaults
- }
-
- // Latency returns the latency for connections.
- func (l *Listener) Latency() time.Duration {
- l.mu.Lock()
- defer l.mu.Unlock()
-
- return l.latency
- }
-
- // SetLatency sets the initial latency for connections.
- func (l *Listener) SetLatency(latency time.Duration) {
- l.mu.Lock()
- defer l.mu.Unlock()
-
- l.latency = latency
- }
-
- // GetTrafficShapedConn takes in a normal connection and returns a traffic shaped connection.
- func (l *Listener) GetTrafficShapedConn(oc net.Conn) *Conn {
- if tsconn, ok := oc.(*Conn); ok {
- return tsconn
- }
- urlbuckets := make(map[string]*Buckets)
- globalurlbuckets := make(map[string]*Bucket)
-
- l.Shapes.RLock()
- defaults := l.Defaults()
- latency := l.Latency()
- defaultBandwidth := defaults.Bandwidth
- for regex, shape := range l.Shapes.M {
- // It should be ok to not acquire the read lock on shape, since WriteBucket is never mutated.
- globalurlbuckets[regex] = shape.Shape.WriteBucket
- urlbuckets[regex] = NewBuckets(DefaultBitrate/8, shape.Shape.MaxBandwidth)
- }
-
- l.Shapes.RUnlock()
-
- curinfo := &Context{}
-
- lc := &Conn{
- Conn: oc,
- latency: latency,
- ReadBucket: l.ReadBucket,
- WriteBucket: l.WriteBucket,
- Shapes: l.Shapes,
- GlobalBuckets: globalurlbuckets,
- LocalBuckets: urlbuckets,
- Context: curinfo,
- Established: time.Now(),
- DefaultBandwidth: defaultBandwidth,
- Listener: l,
- }
- return lc
- }
-
- // Accept waits for and returns the next connection to the listener.
- func (l *Listener) Accept() (net.Conn, error) {
- oc, err := l.Listener.Accept()
- if err != nil {
- log.Errorf("trafficshape: failed accepting connection: %v", err)
- return nil, err
- }
-
- if tconn, ok := oc.(*net.TCPConn); ok {
- log.Debugf("trafficshape: setting keep-alive for TCP connection")
- tconn.SetKeepAlive(true)
- tconn.SetKeepAlivePeriod(3 * time.Minute)
- }
- return l.GetTrafficShapedConn(oc), nil
- }
-
- // Close closes the read and write buckets along with the underlying listener.
- func (l *Listener) Close() error {
- defer log.Debugf("trafficshape: closed read/write buckets and connection")
-
- l.ReadBucket.Close()
- l.WriteBucket.Close()
-
- return l.Listener.Close()
- }
-
- // Read reads bytes from connection into b, optionally simulating connection
- // latency and throttling read throughput based on desired bandwidth
- // constraints.
- func (c *Conn) Read(b []byte) (int, error) {
- c.ronce.Do(c.sleepLatency)
-
- n, err := c.ReadBucket.FillThrottle(func(remaining int64) (int64, error) {
- max := remaining
- if l := int64(len(b)); max > l {
- max = l
- }
-
- n, err := c.Conn.Read(b[:max])
- return int64(n), err
- })
- if err != nil && err != io.EOF {
- log.Errorf("trafficshape: error on throttled read: %v", err)
- }
-
- return int(n), err
- }
-
- // ReadFrom reads data from r until EOF or error, optionally simulating
- // connection latency and throttling read throughput based on desired bandwidth
- // constraints.
- func (c *Conn) ReadFrom(r io.Reader) (int64, error) {
- c.ronce.Do(c.sleepLatency)
-
- var total int64
- for {
- n, err := c.ReadBucket.FillThrottle(func(remaining int64) (int64, error) {
- return io.CopyN(c.Conn, r, remaining)
- })
-
- total += n
-
- if err == io.EOF {
- log.Debugf("trafficshape: exhausted reader successfully")
- return total, nil
- } else if err != nil {
- log.Errorf("trafficshape: failed copying from reader: %v", err)
- return total, err
- }
- }
- }
-
- // WriteTo writes data to w from the connection, optionally simulating
- // connection latency and throttling write throughput based on desired
- // bandwidth constraints.
- func (c *Conn) WriteTo(w io.Writer) (int64, error) {
- c.wonce.Do(c.sleepLatency)
-
- var total int64
- for {
- n, err := c.WriteBucket.FillThrottle(func(remaining int64) (int64, error) {
- return io.CopyN(w, c.Conn, remaining)
- })
-
- total += n
-
- if err != nil {
- if err != io.EOF {
- log.Errorf("trafficshape: failed copying to writer: %v", err)
- }
- return total, err
- }
- }
- }
-
- func min(x, y int64) int64 {
- if x < y {
- return x
- }
- return y
- }
-
- // CheckExistenceAndValidity checks that the current url regex is present in the map, and that
- // the connection was established before the url shape map was last updated. We do not allow the
- // updated url shape map to traffic shape older connections.
- // Important: Assumes you have acquired the required locks and will release them youself.
- func (c *Conn) CheckExistenceAndValidity(URLRegex string) bool {
- shapeStillValid := c.Shapes.LastModifiedTime.Before(c.Established)
- _, p := c.Shapes.M[URLRegex]
- return p && shapeStillValid
- }
-
- // GetCurrentThrottle uses binary search to determine if the current byte offset ('start')
- // lies within a throttle interval. If so, also returns the bandwidth specified for that interval.
- func (c *Conn) GetCurrentThrottle(start int64) *ThrottleContext {
- c.Shapes.RLock()
- defer c.Shapes.RUnlock()
-
- if !c.CheckExistenceAndValidity(c.Context.URLRegex) {
- log.Debugf("existence check failed")
- return &ThrottleContext{
- ThrottleNow: false,
- }
- }
-
- c.Shapes.M[c.Context.URLRegex].RLock()
- defer c.Shapes.M[c.Context.URLRegex].RUnlock()
-
- throttles := c.Shapes.M[c.Context.URLRegex].Shape.Throttles
-
- if l := len(throttles); l != 0 {
- // ind is the first index in throttles with ByteStart > start.
- // Once we get ind, we can check the previous throttle, if any,
- // to see if its ByteEnd is after 'start'.
- ind := sort.Search(len(throttles),
- func(i int) bool { return throttles[i].ByteStart > start })
-
- // All throttles have Bytestart > start, hence not in throttle.
- if ind == 0 {
- return &ThrottleContext{
- ThrottleNow: false,
- }
- }
-
- // No throttle has Bytestart > start, so check the last throttle to
- // see if it ends after 'start'. Note: the last throttle is special
- // since it can have -1 (meaning infinity) as the ByteEnd.
- if ind == l {
- if throttles[l-1].ByteEnd > start || throttles[l-1].ByteEnd == -1 {
- return &ThrottleContext{
- ThrottleNow: true,
- Bandwidth: throttles[l-1].Bandwidth,
- }
- }
- return &ThrottleContext{
- ThrottleNow: false,
- }
- }
-
- // Check the previous throttle to see if it ends after 'start'.
- if throttles[ind-1].ByteEnd > start {
- return &ThrottleContext{
- ThrottleNow: true,
- Bandwidth: throttles[ind-1].Bandwidth,
- }
- }
-
- return &ThrottleContext{
- ThrottleNow: false,
- }
- }
-
- return &ThrottleContext{
- ThrottleNow: false,
- }
- }
-
- // GetNextActionFromByte takes in a byte offset and uses binary search to determine the upcoming
- // action, i.e the first action after the byte that still has a non zero count.
- func (c *Conn) GetNextActionFromByte(start int64) *NextActionInfo {
- c.Shapes.RLock()
- defer c.Shapes.RUnlock()
-
- if !c.CheckExistenceAndValidity(c.Context.URLRegex) {
- log.Debugf("existence check failed")
- return &NextActionInfo{
- ActionNext: false,
- }
- }
-
- c.Shapes.M[c.Context.URLRegex].RLock()
- defer c.Shapes.M[c.Context.URLRegex].RUnlock()
-
- actions := c.Shapes.M[c.Context.URLRegex].Shape.Actions
-
- if l := len(actions); l != 0 {
- ind := sort.Search(len(actions),
- func(i int) bool { return actions[i].getByte() >= start })
-
- return c.GetNextActionFromIndex(int64(ind))
- }
-
- return &NextActionInfo{
- ActionNext: false,
- }
- }
-
- // GetNextActionFromIndex takes in an index and returns the first action after the index that
- // has a non zero count, if there is one.
- func (c *Conn) GetNextActionFromIndex(ind int64) *NextActionInfo {
- c.Shapes.RLock()
- defer c.Shapes.RUnlock()
-
- if !c.CheckExistenceAndValidity(c.Context.URLRegex) {
- return &NextActionInfo{
- ActionNext: false,
- }
- }
-
- c.Shapes.M[c.Context.URLRegex].RLock()
- defer c.Shapes.M[c.Context.URLRegex].RUnlock()
-
- actions := c.Shapes.M[c.Context.URLRegex].Shape.Actions
-
- if l := int64(len(actions)); l != 0 {
-
- for ind < l && (actions[ind].getCount() == 0) {
- ind++
- }
-
- if ind >= l {
- return &NextActionInfo{
- ActionNext: false,
- }
- }
- return &NextActionInfo{
- ActionNext: true,
- Index: ind,
- ByteOffset: actions[ind].getByte(),
- }
- }
- return &NextActionInfo{
- ActionNext: false,
- }
- }
-
- // WriteDefaultBuckets writes bytes from b to the connection, optionally simulating
- // connection latency and throttling write throughput based on desired
- // bandwidth constraints. It uses the WriteBucket inherited from the listener.
- func (c *Conn) WriteDefaultBuckets(b []byte) (int, error) {
- c.wonce.Do(c.sleepLatency)
-
- var total int64
- for len(b) > 0 {
- var max int64
-
- n, err := c.WriteBucket.FillThrottle(func(remaining int64) (int64, error) {
- max = remaining
- if l := int64(len(b)); remaining >= l {
- max = l
- }
-
- n, err := c.Conn.Write(b[:max])
- return int64(n), err
- })
-
- total += n
-
- if err != nil {
- if err != io.EOF {
- log.Errorf("trafficshape: failed write: %v", err)
- }
- return int(total), err
- }
-
- b = b[max:]
- }
-
- return int(total), nil
- }
-
- // Write writes bytes from b to the connection, while enforcing throttles and performing actions.
- // It uses and updates the Context in the connection.
- func (c *Conn) Write(b []byte) (int, error) {
- if !c.Context.Shaping {
- return c.WriteDefaultBuckets(b)
- }
- c.wonce.Do(c.sleepLatency)
- var total int64
-
- // Write the header if needed, without enforcing any traffic shaping, and without updating
- // ByteOffset.
- if headerToWrite := c.Context.HeaderLen - c.Context.HeaderBytesWritten; headerToWrite > 0 {
- writeAmount := min(int64(len(b)), headerToWrite)
-
- n, err := c.Conn.Write(b[:writeAmount])
-
- if err != nil {
- if err != io.EOF {
- log.Errorf("trafficshape: failed write: %v", err)
- }
- return int(n), err
- }
- c.Context.HeaderBytesWritten += writeAmount
- total += writeAmount
- b = b[writeAmount:]
- }
-
- var amountToWrite int64
-
- for len(b) > 0 {
- var max int64
-
- // Determine the amount to be written up till the next action.
- amountToWrite = int64(len(b))
- if c.Context.NextActionInfo.ActionNext {
- amountTillNextAction := c.Context.NextActionInfo.ByteOffset - c.Context.ByteOffset
- if amountTillNextAction <= amountToWrite {
- amountToWrite = amountTillNextAction
- }
- }
-
- // Write into both the local and global buckets, as well as the underlying connection.
- n, err := c.Context.Buckets.WriteBucket.FillThrottleLocked(func(remaining int64) (int64, error) {
- max = min(remaining, amountToWrite)
-
- if max == 0 {
- return 0, nil
- }
-
- return c.Context.GlobalBucket.FillThrottleLocked(func(rem int64) (int64, error) {
- max = min(rem, max)
- n, err := c.Conn.Write(b[:max])
-
- return int64(n), err
- })
- })
-
- if err != nil {
- if err != io.EOF {
- log.Errorf("trafficshape: failed write: %v", err)
- }
- return int(total), err
- }
-
- // Update the current byte offset.
- c.Context.ByteOffset += n
- total += n
-
- b = b[max:]
-
- // Check if there was an upcoming action, and that the byte offset matches the action's byte.
- if c.Context.NextActionInfo.ActionNext &&
- c.Context.ByteOffset >= c.Context.NextActionInfo.ByteOffset {
- // Note here, we check again that the url shape map is still valid and that the action still has
- // a non zero count, since that could have been modified since the last time we checked.
- ind := c.Context.NextActionInfo.Index
- c.Shapes.RLock()
- if !c.CheckExistenceAndValidity(c.Context.URLRegex) {
- c.Shapes.RUnlock()
- // Write the remaining b using default buckets, and set Shaping as false
- // so that subsequent calls to Write() also use default buckets
- // without performing any actions.
- c.Context.Shaping = false
- writeTotal, e := c.WriteDefaultBuckets(b)
- return int(total) + writeTotal, e
- }
- c.Shapes.M[c.Context.URLRegex].Lock()
- actions := c.Shapes.M[c.Context.URLRegex].Shape.Actions
- if actions[ind].getCount() != 0 {
- // Update the action count, determine the type of action and perform it.
- actions[ind].decrementCount()
- switch action := actions[ind].(type) {
- case *Halt:
- d := action.Duration
- log.Debugf("trafficshape: Sleeping for time %d ms for urlregex %s at byte offset %d",
- d, c.Context.URLRegex, c.Context.ByteOffset)
- c.Shapes.M[c.Context.URLRegex].Unlock()
- c.Shapes.RUnlock()
- time.Sleep(time.Duration(d) * time.Millisecond)
- case *CloseConnection:
- log.Infof("trafficshape: Closing connection for urlregex %s at byte offset %d",
- c.Context.URLRegex, c.Context.ByteOffset)
- c.Shapes.M[c.Context.URLRegex].Unlock()
- c.Shapes.RUnlock()
- return int(total), &ErrForceClose{message: "Forcing close connection"}
- case *ChangeBandwidth:
- bw := action.Bandwidth
- log.Infof("trafficshape: Changing connection bandwidth to %d for urlregex %s at byte offset %d",
- bw, c.Context.URLRegex, c.Context.ByteOffset)
- c.Shapes.M[c.Context.URLRegex].Unlock()
- c.Shapes.RUnlock()
- c.Context.Buckets.WriteBucket.SetCapacity(bw)
- default:
- c.Shapes.M[c.Context.URLRegex].Unlock()
- c.Shapes.RUnlock()
- }
- } else {
- c.Shapes.M[c.Context.URLRegex].Unlock()
- c.Shapes.RUnlock()
- }
- // Get the next action to be performed, if any.
- c.Context.NextActionInfo = c.GetNextActionFromIndex(ind + 1)
- }
- }
- return int(total), nil
- }
-
- func (c *Conn) sleepLatency() {
- log.Debugf("trafficshape: simulating latency: %s", c.latency)
- time.Sleep(c.latency)
- }
|