Revert "Reduce http proxy client overhead by 1RTT"

pull/2350/head
Kslr 5 years ago committed by GitHub
parent 3feec68a87
commit bb939ae03f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

@ -7,7 +7,6 @@ import (
"time" "time"
"github.com/golang/protobuf/proto" "github.com/golang/protobuf/proto"
"v2ray.com/core/common/buf"
"v2ray.com/core/common/mux" "v2ray.com/core/common/mux"
"v2ray.com/core/common/net" "v2ray.com/core/common/net"
"v2ray.com/core/common/session" "v2ray.com/core/common/session"
@ -120,13 +119,6 @@ func NewBridgeWorker(domain string, tag string, d routing.Dispatcher) (*BridgeWo
tag: tag, tag: tag,
} }
// Initialize the connection by sending a Keepalive frame
keepalive := buf.New()
mux.FrameMetadata{SessionStatus: mux.SessionStatusKeepAlive}.WriteTo(keepalive)
err = link.Writer.WriteMultiBuffer(buf.MultiBuffer{keepalive})
if err != nil {
return nil, err
}
worker, err := mux.NewServerWorker(context.Background(), w, link) worker, err := mux.NewServerWorker(context.Background(), w, link)
if err != nil { if err != nil {
return nil, err return nil, err

@ -3,14 +3,10 @@
package http package http
import ( import (
"bufio"
"context" "context"
"encoding/base64" "encoding/base64"
"io" "io"
"net/http"
"strings" "strings"
"sync"
"time"
"v2ray.com/core" "v2ray.com/core"
"v2ray.com/core/common" "v2ray.com/core/common"
@ -26,7 +22,6 @@ import (
"v2ray.com/core/transport/internet" "v2ray.com/core/transport/internet"
) )
// Client is a inbound handler for HTTP protocol
type Client struct { type Client struct {
serverPicker protocol.ServerPicker serverPicker protocol.ServerPicker
policyManager policy.Manager policyManager policy.Manager
@ -95,7 +90,9 @@ func (c *Client) Process(ctx context.Context, link *transport.Link, dialer inter
p = c.policyManager.ForLevel(user.Level) p = c.policyManager.ForLevel(user.Level)
} }
conn = setUpHTTPTunnel(conn, &destination, user) if err := setUpHttpTunnel(conn, conn, &destination, user); err != nil {
return err
}
ctx, cancel := context.WithCancel(ctx) ctx, cancel := context.WithCancel(ctx)
timer := signal.CancelAfterInactivity(ctx, cancel, p.Timeouts.ConnectionIdle) timer := signal.CancelAfterInactivity(ctx, cancel, p.Timeouts.ConnectionIdle)
@ -106,15 +103,7 @@ func (c *Client) Process(ctx context.Context, link *transport.Link, dialer inter
} }
responseFunc := func() error { responseFunc := func() error {
defer timer.SetTimeout(p.Timeouts.UplinkOnly) defer timer.SetTimeout(p.Timeouts.UplinkOnly)
bc := bufio.NewReader(conn) return buf.Copy(buf.NewReader(conn), link.Writer, buf.UpdateActivity(timer))
resp, err := http.ReadResponse(bc, nil)
if err != nil {
return err
}
if resp.StatusCode != http.StatusOK {
return newError(resp.Status)
}
return buf.Copy(buf.NewReader(bc), link.Writer, buf.UpdateActivity(timer))
} }
var responseDonePost = task.OnSuccess(responseFunc, task.Close(link.Writer)) var responseDonePost = task.OnSuccess(responseFunc, task.Close(link.Writer))
@ -125,8 +114,8 @@ func (c *Client) Process(ctx context.Context, link *transport.Link, dialer inter
return nil return nil
} }
// setUpHTTPTunnel will create a socket tunnel via HTTP CONNECT method // setUpHttpTunnel will create a socket tunnel via HTTP CONNECT method
func setUpHTTPTunnel(conn internet.Connection, destination *net.Destination, user *protocol.MemoryUser) *tunConn { func setUpHttpTunnel(reader io.Reader, writer io.Writer, destination *net.Destination, user *protocol.MemoryUser) error {
var headers []string var headers []string
destNetAddr := destination.NetAddr() destNetAddr := destination.NetAddr()
headers = append(headers, "CONNECT "+destNetAddr+" HTTP/1.1") headers = append(headers, "CONNECT "+destNetAddr+" HTTP/1.1")
@ -140,62 +129,16 @@ func setUpHTTPTunnel(conn internet.Connection, destination *net.Destination, use
b := buf.New() b := buf.New()
b.WriteString(strings.Join(headers, "\r\n") + "\r\n\r\n") b.WriteString(strings.Join(headers, "\r\n") + "\r\n\r\n")
return newTunConn(conn, b, 5 * time.Millisecond) if err := buf.WriteAllBytes(writer, b.Bytes()); err != nil {
} return err
// tunConn is a connection that writes header before content,
// the header will be written during the next Write call or after
// specified delay.
type tunConn struct {
internet.Connection
header *buf.Buffer
once sync.Once
timer *time.Timer
}
func newTunConn(conn internet.Connection, header *buf.Buffer, delay time.Duration) *tunConn {
tc := &tunConn{
Connection: conn,
header: header,
} }
if delay > 0 {
tc.timer = time.AfterFunc(delay, func() { b.Clear()
tc.Write([]byte{}) if _, err := b.ReadFrom(reader); err != nil {
}) return err
} }
return tc
}
func (c *tunConn) Write(b []byte) (n int, err error) { return nil
// fallback to normal write if header is sent
if c.header == nil {
return c.Connection.Write(b)
}
// Prevent timer and writer race condition
c.once.Do(func() {
if c.timer != nil {
c.timer.Stop()
c.timer = nil
}
lenheader := c.header.Len()
// Concate header and b
common.Must2(c.header.Write(b))
// Write buffer
var nc int64
nc, err = io.Copy(c.Connection, c.header)
c.header.Release()
c.header = nil
n = int(nc) - int(lenheader)
if n < 0 { n = 0 }
b = b[n:]
})
// Write Trailing bytes
if len(b) > 0 && err == nil {
var nw int
nw, err = c.Connection.Write(b)
n += nw
}
return n, err
} }
func init() { func init() {

Loading…
Cancel
Save