transport: add GatedMaListener type #3186
Conversation
| // Upgrader is a multistream upgrader that can upgrade an underlying connection | ||
| // to a full transport connection (secure and multiplexed). | ||
| type Upgrader interface { | ||
| // UpgradeListener upgrades the passed multiaddr-net listener into a full libp2p-transport listener. | ||
| // | ||
| // Deprecated: Use UpgradeGatedManetListener(upgrader.GateManetListener(manet.Listener)) instead. | ||
| UpgradeListener(Transport, manet.Listener) Listener |
There was a problem hiding this comment.
Is anyone besides us using this? Should we just remove it?
| l.mx.Lock() | ||
| l.scopes[conn] = scope | ||
| // TODO: This should be 10 seconds? | ||
| time.AfterFunc(time.Minute, func() { |
There was a problem hiding this comment.
Is this in case the connection is closed or dropped? This is kind of hacky
sukunrt
force-pushed
the
sukun/tcpreuse-refactor
branch
2 times, most recently
from
a3eadf2 to
78cd22b
Compare
sukunrt
changed the title
sukunrt
force-pushed
the
sukun/tcpreuse-refactor
branch
3 times, most recently
from
ae94a2d to
6b3b600
Compare
sukunrt
force-pushed
the
sukun/tcpreuse-refactor
branch
from
6b3b600 to
512a7d5
Compare
This introduces a new GatedMaListener type which gates conns accepted from a manet.Listener with a gater and creates the rcmgr scope for it. Explicitly passing the scope allows for many guardrails that the previous interface assertion didn't. This breaks the previous responsibility of the upgradeListener method into two, one gating the connection initially, and the other upgrading the connection with a security and muxer selection. This split makes it easy to gate the connection with the resource manager as early as possible. This is especially true for websocket because we want to gate the connection just after the TCP connection is established, and not after the tls handshake + websocket upgrade is completed.
sukunrt
force-pushed
the
sukun/tcpreuse-refactor
branch
from
512a7d5 to
f1b1ee9
Compare
| if err != nil { | ||
| return nil, err | ||
| } | ||
| gmal = upgrader.GateMaListener(mal) |
There was a problem hiding this comment.
Note: This changes the address reported to the resource manager and connection gater from:
/ip4/a.b.c.d/tcp/xx/ws to /ip4/a.b.c.d/tcp/xx
We are not providing the /ws component anymore. This is the same behavior as what the shared tcp listener does as it doesn't know whether the connection will be a websocket or tcp one.
This is easy enough to fix, but that would make the behavior between the shared tcp listener and unshared one inconsistent.
I don't think the resource manager cares, as we don't have a limit of connections per transport.
The conn gater might care about this. If that's the case, we can move the conngater.InterceptAccept into GateMaListener and have the gatedMaListener only manage resource manager scope.
There was a problem hiding this comment.
I don't think I understand. Are we not including the ws component in the
connection's Multiaddr methods? We probably should since it's often useful for
debugging and understanding where a connection came from.
| return nil, nil, err | ||
| } | ||
|
|
||
| // TODO(sukunrt): Should we log and ignore this error? We can proceed. We just won't have metrics for this connection. |
There was a problem hiding this comment.
Log and ignore seems reasonable to me
| if nc, ok := c.(*negotiatingConn); ok { | ||
| return context.WithValue(ctx, connKey{}, nc) | ||
| } | ||
| log.Errorf("BUG: expected *websocket.negotiatingConn in context: got %T", c) |
There was a problem hiding this comment.
Shouldn't this say: "expected *websocket.negotiatingConn type of NetConn" or similar?
| @@ -107,19 +107,44 @@ func New(security []sec.SecureTransport, muxers []StreamMuxer, psk ipnet.PSK, rc | |||
| func (u *upgrader) UpgradeListener(t transport.Transport, list manet.Listener) transport.Listener { | |||
There was a problem hiding this comment.
Shouldn't this just call return u.UpgradeGatedMaListener(t, u.GateMaListener(list))
| } | ||
|
|
||
| if t.enableMetrics { | ||
| list = newTracingListener(&tcpListener{list, 0}, t.metricsCollector) | ||
| // TODO: Fix this: The tcpListener wrapping should happen on both enableMetrics and disabledMetrics path |
There was a problem hiding this comment.
Why would it happen on the disable metrics path?
| if !connType.IsKnown() { | ||
| fmt.Println("connType", connType) |
| @@ -167,6 +148,7 @@ func (c *Conn) closeOnceFn() error { | |||
| time.Now().Add(GracefulCloseTimeout), | |||
| ) | |||
| err2 := c.Conn.Close() | |||
| c.Scope.Done() | |||
There was a problem hiding this comment.
Doesn't the upgrader close the scope?
| if err != nil { | ||
| return nil, err | ||
| } | ||
| gmal = upgrader.GateMaListener(mal) |
There was a problem hiding this comment.
I don't think I understand. Are we not including the ws component in the
connection's Multiaddr methods? We probably should since it's often useful for
debugging and understanding where a connection came from.
This introduces a new GatedMaListener type which gates conns
accepted from a manet.Listener with a gater and creates the rcmgr
scope for it. Explicitly passing the scope allows for many guardrails
that the previous interface assertion didn't.
This breaks the previous responsibility of the upgradeListener method
into two, one gating the connection initially, and the other upgrading
the connection with a security and muxer selection.
This split makes it easy to gate the connection with the resource
manager as early as possible. This is especially true for websocket
because we want to gate the connection just after the TCP connection is
established, and not after the tls handshake + websocket upgrade is
completed.
fixes: #3182