jgaunt/rclone
1
0
Fork 0
mirror of https://github.com/rclone/rclone.git synced 2026-01-02 08:33:50 +00:00
Code Issues Projects Releases Wiki Activity

fshttp: add DSCP support with --dscp for QoS with differentiated services

Browse Source
This commit is contained in:
Max Sum
2021-01-29 11:41:50 +08:00
committed by Nick Craig-Wood
parent dfc63eb8f1
commit edfe183ba2
7 changed files with 240 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
fs/config.go
View File

@@ -122,6 +122,7 @@ type ConfigInfo struct {
Headers []*HTTPOption
RefreshTimes bool
NoConsole bool
TrafficClass uint8
}
// NewConfig creates a new config with everything set to the default

68
fs/config/configflags/configflags.go
View File

@@ -7,6 +7,7 @@ import (
"log"
"net"
"path/filepath"
"strconv"
"strings"
"github.com/rclone/rclone/fs"
@@ -29,6 +30,7 @@ var (
deleteAfter bool
bindAddr string
disableFeatures string
dscp string
uploadHeaders []string
downloadHeaders []string
headers []string
@@ -125,6 +127,7 @@ func AddFlags(ci *fs.ConfigInfo, flagSet *pflag.FlagSet) {
flags.StringArrayVarP(flagSet, &headers, "header", "", nil, "Set HTTP header for all transactions")
flags.BoolVarP(flagSet, &ci.RefreshTimes, "refresh-times", "", ci.RefreshTimes, "Refresh the modtime of remote files.")
flags.BoolVarP(flagSet, &ci.NoConsole, "no-console", "", ci.NoConsole, "Hide console window. Supported on Windows only.")
flags.StringVarP(flagSet, &dscp, "dscp", "", "", "Set DSCP value to connections. Can be value or names, eg. CS1, LE, DF, AF21.")
}
// ParseHeaders converts the strings passed in via the header flags into HTTPOptions
@@ -254,6 +257,13 @@ func SetFlags(ci *fs.ConfigInfo) {
if len(headers) != 0 {
ci.Headers = ParseHeaders(headers)
}
if len(dscp) != 0 {
if value, ok := parseDSCP(dscp); ok {
ci.TrafficClass = value << 2
} else {
log.Fatalf("--dscp: Invalid DSCP name: %v", dscp)
}
}
// Make the config file absolute
configPath, err := filepath.Abs(config.ConfigPath)
@@ -266,3 +276,61 @@ func SetFlags(ci *fs.ConfigInfo) {
ci.MultiThreadSet = multiThreadStreamsFlag != nil && multiThreadStreamsFlag.Changed
}
// parseHeaders converts DSCP names to value
func parseDSCP(dscp string) (uint8, bool) {
if s, err := strconv.ParseUint(dscp, 10, 6); err == nil {
return uint8(s), true
}
dscp = strings.ToUpper(dscp)
switch dscp {
case "BE":
fallthrough
case "DF":
fallthrough
case "CS0":
return 0x00, true
case "CS1":
return 0x08, true
case "AF11":
return 0x0A, true
case "AF12":
return 0x0C, true
case "AF13":
return 0x0E, true
case "CS2":
return 0x10, true
case "AF21":
return 0x12, true
case "AF22":
return 0x14, true
case "AF23":
return 0x16, true
case "CS3":
return 0x18, true
case "AF31":
return 0x1A, true
case "AF32":
return 0x1C, true
case "AF33":
return 0x1E, true
case "CS4":
return 0x20, true
case "AF41":
return 0x22, true
case "AF42":
return 0x24, true
case "AF43":
return 0x26, true
case "CS5":
return 0x28, true
case "EF":
return 0x2E, true
case "CS6":
return 0x30, true
case "LE":
return 0x01, true
default:
return 0, false
}
}

114
fs/fshttp/dialer.go Normal file
View File

@@ -0,0 +1,114 @@
package fshttp
import (
"context"
"net"
"time"
"github.com/rclone/rclone/fs"
"golang.org/x/net/ipv4"
"golang.org/x/net/ipv6"
)
func dialContext(ctx context.Context, network, address string, ci *fs.ConfigInfo) (net.Conn, error) {
return NewDialer(ctx).DialContext(ctx, network, address)
}
// Dialer structure contains default dialer and timeout, tclass support
type Dialer struct {
net.Dialer
timeout time.Duration
tclass int
}
// NewDialer creates a Dialer structure with Timeout, Keepalive,
// LocalAddr and DSCP set from rclone flags.
func NewDialer(ctx context.Context) *Dialer {
ci := fs.GetConfig(ctx)
dialer := &Dialer{
Dialer: net.Dialer{
Timeout: ci.ConnectTimeout,
KeepAlive: 30 * time.Second,
},
timeout: ci.Timeout,
tclass: int(ci.TrafficClass),
}
if ci.BindAddr != nil {
dialer.Dialer.LocalAddr = &net.TCPAddr{IP: ci.BindAddr}
}
return dialer
}
// Dial connects to the address on the named network.
func (d *Dialer) Dial(network, address string) (net.Conn, error) {
return d.DialContext(context.Background(), network, address)
}
// DialContext connects to the address on the named network using
// the provided context.
func (d *Dialer) DialContext(ctx context.Context, network, address string) (net.Conn, error) {
c, err := d.Dialer.DialContext(ctx, network, address)
if err != nil {
return c, err
}
if d.tclass != 0 {
if addr, ok := c.RemoteAddr().(*net.IPAddr); ok {
if addr.IP.To16() != nil && addr.IP.To4() == nil {
err = ipv6.NewConn(c).SetTrafficClass(d.tclass)
} else {
err = ipv4.NewConn(c).SetTOS(d.tclass)
}
if err != nil {
return c, err
}
}
}
return newTimeoutConn(c, d.timeout)
}
// A net.Conn that sets a deadline for every Read or Write operation
type timeoutConn struct {
net.Conn
timeout time.Duration
}
// create a timeoutConn using the timeout
func newTimeoutConn(conn net.Conn, timeout time.Duration) (c *timeoutConn, err error) {
c = &timeoutConn{
Conn: conn,
timeout: timeout,
}
err = c.nudgeDeadline()
return
}
// Nudge the deadline for an idle timeout on by c.timeout if non-zero
func (c *timeoutConn) nudgeDeadline() (err error) {
if c.timeout == 0 {
return nil
}
when := time.Now().Add(c.timeout)
return c.Conn.SetDeadline(when)
}
// readOrWrite bytes doing idle timeouts
func (c *timeoutConn) readOrWrite(f func([]byte) (int, error), b []byte) (n int, err error) {
n, err = f(b)
// Don't nudge if no bytes or an error
if n == 0 || err != nil {
return
}
// Nudge the deadline on successful Read or Write
err = c.nudgeDeadline()
return
}
// Read bytes doing idle timeouts
func (c *timeoutConn) Read(b []byte) (n int, err error) {
return c.readOrWrite(c.Conn.Read, b)
}
// Write bytes doing idle timeouts
func (c *timeoutConn) Write(b []byte) (n int, err error) {
return c.readOrWrite(c.Conn.Write, b)
}

78
fs/fshttp/http.go
View File

@@ -33,68 +33,6 @@ var (
logMutex sync.Mutex
)
// A net.Conn that sets a deadline for every Read or Write operation
type timeoutConn struct {
net.Conn
timeout time.Duration
}
// create a timeoutConn using the timeout
func newTimeoutConn(conn net.Conn, timeout time.Duration) (c *timeoutConn, err error) {
c = &timeoutConn{
Conn: conn,
timeout: timeout,
}
err = c.nudgeDeadline()
return
}
// Nudge the deadline for an idle timeout on by c.timeout if non-zero
func (c *timeoutConn) nudgeDeadline() (err error) {
if c.timeout == 0 {
return nil
}
when := time.Now().Add(c.timeout)
return c.Conn.SetDeadline(when)
}
// Read bytes doing idle timeouts
func (c *timeoutConn) Read(b []byte) (n int, err error) {
// Ideally we would LimitBandwidth(len(b)) here and replace tokens we didn't use
n, err = c.Conn.Read(b)
accounting.TokenBucket.LimitBandwidth(accounting.TokenBucketSlotTransportRx, n)
// Don't nudge if no bytes or an error
if n == 0 || err != nil {
return
}
// Nudge the deadline on successful Read or Write
err = c.nudgeDeadline()
return n, err
}
// Write bytes doing idle timeouts
func (c *timeoutConn) Write(b []byte) (n int, err error) {
accounting.TokenBucket.LimitBandwidth(accounting.TokenBucketSlotTransportTx, len(b))
n, err = c.Conn.Write(b)
// Don't nudge if no bytes or an error
if n == 0 || err != nil {
return
}
// Nudge the deadline on successful Read or Write
err = c.nudgeDeadline()
return n, err
}
// dial with context and timeouts
func dialContextTimeout(ctx context.Context, network, address string, ci *fs.ConfigInfo) (net.Conn, error) {
dialer := NewDialer(ctx)
c, err := dialer.DialContext(ctx, network, address)
if err != nil {
return c, err
}
return newTimeoutConn(c, ci.Timeout)
}
// ResetTransport resets the existing transport, allowing it to take new settings.
// Should only be used for testing.
func ResetTransport() {
@@ -150,7 +88,7 @@ func NewTransportCustom(ctx context.Context, customize func(*http.Transport)) ht
t.DisableCompression = ci.NoGzip
t.DialContext = func(ctx context.Context, network, addr string) (net.Conn, error) {
return dialContextTimeout(ctx, network, addr, ci)
return dialContext(ctx, network, addr, ci)
}
t.IdleConnTimeout = 60 * time.Second
t.ExpectContinueTimeout = ci.ExpectContinueTimeout
@@ -346,17 +284,3 @@ func (t *Transport) RoundTrip(req *http.Request) (resp *http.Response, err error
}
return resp, err
}
// NewDialer creates a net.Dialer structure with Timeout, Keepalive
// and LocalAddr set from rclone flags.
func NewDialer(ctx context.Context) *net.Dialer {
ci := fs.GetConfig(ctx)
dialer := &net.Dialer{
Timeout: ci.ConnectTimeout,
KeepAlive: 30 * time.Second,
}
if ci.BindAddr != nil {
dialer.LocalAddr = &net.TCPAddr{IP: ci.BindAddr}
}
return dialer
}