package webtorrent
import (
+ "context"
"expvar"
"fmt"
"io"
"sync"
"time"
+ "github.com/anacrolix/log"
"github.com/anacrolix/missinggo/v2/pproffd"
"github.com/pion/datachannel"
+ "github.com/pion/webrtc/v3"
+ "go.opentelemetry.io/otel"
+ "go.opentelemetry.io/otel/attribute"
+ "go.opentelemetry.io/otel/codes"
+ "go.opentelemetry.io/otel/trace"
+)
- "github.com/pion/webrtc/v2"
+const (
+ dataChannelLabel = "webrtc-datachannel"
)
var (
metrics = expvar.NewMap("webtorrent")
api = func() *webrtc.API {
// Enable the detach API (since it's non-standard but more idiomatic).
- s := webrtc.SettingEngine{}
s.DetachDataChannels()
return webrtc.NewAPI(webrtc.WithSettingEngine(s))
}()
- config = webrtc.Configuration{ICEServers: []webrtc.ICEServer{{URLs: []string{"stun:stun.l.google.com:19302"}}}}
newPeerConnectionMu sync.Mutex
)
type wrappedPeerConnection struct {
*webrtc.PeerConnection
+ closeMu sync.Mutex
pproffd.CloseWrapper
+ span trace.Span
+ ctx context.Context
}
-func (me wrappedPeerConnection) Close() error {
- return me.CloseWrapper.Close()
+func (me *wrappedPeerConnection) Close() error {
+ me.closeMu.Lock()
+ defer me.closeMu.Unlock()
+ err := me.CloseWrapper.Close()
+ me.span.End()
+ return err
}
-func newPeerConnection() (wrappedPeerConnection, error) {
+func newPeerConnection(logger log.Logger, iceServers []string) (*wrappedPeerConnection, error) {
newPeerConnectionMu.Lock()
defer newPeerConnectionMu.Unlock()
- pc, err := api.NewPeerConnection(config)
+ ctx, span := otel.Tracer(tracerName).Start(context.Background(), "PeerConnection")
+
+ pcConfig := webrtc.Configuration{ICEServers: []webrtc.ICEServer{{URLs: iceServers}}}
+
+ pc, err := api.NewPeerConnection(pcConfig)
if err != nil {
- return wrappedPeerConnection{}, err
+ span.SetStatus(codes.Error, err.Error())
+ span.RecordError(err)
+ span.End()
+ return nil, err
}
- return wrappedPeerConnection{
- pc,
- pproffd.NewCloseWrapper(pc),
- }, nil
+ wpc := &wrappedPeerConnection{
+ PeerConnection: pc,
+ CloseWrapper: pproffd.NewCloseWrapper(pc),
+ ctx: ctx,
+ span: span,
+ }
+ // If the state change handler intends to call Close, it should call it on the wrapper.
+ wpc.OnConnectionStateChange(func(state webrtc.PeerConnectionState) {
+ logger.Levelf(log.Warning, "webrtc PeerConnection state changed to %v", state)
+ span.AddEvent("connection state changed", trace.WithAttributes(attribute.String("state", state.String())))
+ })
+ return wpc, nil
}
-// newOffer creates a transport and returns a WebRTC offer to be announced
-func newOffer() (
- peerConnection wrappedPeerConnection,
+func setAndGatherLocalDescription(peerConnection *wrappedPeerConnection, sdp webrtc.SessionDescription) (_ webrtc.SessionDescription, err error) {
+ gatherComplete := webrtc.GatheringCompletePromise(peerConnection.PeerConnection)
+ peerConnection.span.AddEvent("setting local description")
+ err = peerConnection.SetLocalDescription(sdp)
+ if err != nil {
+ err = fmt.Errorf("setting local description: %w", err)
+ return
+ }
+ <-gatherComplete
+ peerConnection.span.AddEvent("gathering complete")
+ return *peerConnection.LocalDescription(), nil
+}
+
+// newOffer creates a transport and returns a WebRTC offer to be announced. See
+// https://github.com/pion/webrtc/blob/master/examples/data-channels/jsfiddle/main.go for what this is modelled on.
+func (tc *TrackerClient) newOffer(
+ logger log.Logger,
+ offerId string,
+ infoHash [20]byte,
+) (
+ peerConnection *wrappedPeerConnection,
dataChannel *webrtc.DataChannel,
offer webrtc.SessionDescription,
err error,
) {
- peerConnection, err = newPeerConnection()
+ peerConnection, err = newPeerConnection(logger, tc.ICEServers)
if err != nil {
return
}
- dataChannel, err = peerConnection.CreateDataChannel("webrtc-datachannel", nil)
+
+ peerConnection.span.SetAttributes(attribute.String(webrtcConnTypeKey, "offer"))
+
+ dataChannel, err = peerConnection.CreateDataChannel(dataChannelLabel, nil)
if err != nil {
+ err = fmt.Errorf("creating data channel: %w", err)
peerConnection.Close()
- return
}
+ initDataChannel(dataChannel, peerConnection, func(dc datachannel.ReadWriteCloser, dcCtx context.Context, dcSpan trace.Span) {
+ metrics.Add("outbound offers answered with datachannel", 1)
+ tc.mu.Lock()
+ tc.stats.ConvertedOutboundConns++
+ tc.mu.Unlock()
+ tc.OnConn(dc, DataChannelContext{
+ OfferId: offerId,
+ LocalOffered: true,
+ InfoHash: infoHash,
+ peerConnection: peerConnection,
+ Context: dcCtx,
+ Span: dcSpan,
+ })
+ })
+
offer, err = peerConnection.CreateOffer(nil)
if err != nil {
+ dataChannel.Close()
peerConnection.Close()
return
}
- err = peerConnection.SetLocalDescription(offer)
+
+ offer, err = setAndGatherLocalDescription(peerConnection, offer)
if err != nil {
+ dataChannel.Close()
peerConnection.Close()
- return
}
return
}
-func initAnsweringPeerConnection(
- peerConnection wrappedPeerConnection,
- offerId string,
- offer webrtc.SessionDescription,
- onOpen onDataChannelOpen,
+type onDetachedDataChannelFunc func(detached datachannel.ReadWriteCloser, ctx context.Context, span trace.Span)
+
+func (tc *TrackerClient) initAnsweringPeerConnection(
+ peerConn *wrappedPeerConnection,
+ offerContext offerContext,
) (answer webrtc.SessionDescription, err error) {
- err = peerConnection.SetRemoteDescription(offer)
- if err != nil {
- return
- }
- answer, err = peerConnection.CreateAnswer(nil)
- if err != nil {
- return
- }
- err = peerConnection.SetLocalDescription(answer)
- if err != nil {
- return
- }
+ peerConn.span.SetAttributes(attribute.String(webrtcConnTypeKey, "answer"))
+
timer := time.AfterFunc(30*time.Second, func() {
+ peerConn.span.SetStatus(codes.Error, "answer timeout")
metrics.Add("answering peer connections timed out", 1)
- peerConnection.Close()
+ peerConn.Close()
})
- peerConnection.OnDataChannel(func(d *webrtc.DataChannel) {
- setDataChannelOnOpen(d, peerConnection, func(dc datachannel.ReadWriteCloser) {
+ peerConn.OnDataChannel(func(d *webrtc.DataChannel) {
+ initDataChannel(d, peerConn, func(detached datachannel.ReadWriteCloser, ctx context.Context, span trace.Span) {
timer.Stop()
metrics.Add("answering peer connection conversions", 1)
- onOpen(dc, DataChannelContext{answer, offer, offerId, false})
+ tc.mu.Lock()
+ tc.stats.ConvertedInboundConns++
+ tc.mu.Unlock()
+ tc.OnConn(detached, DataChannelContext{
+ OfferId: offerContext.Id,
+ LocalOffered: false,
+ InfoHash: offerContext.InfoHash,
+ peerConnection: peerConn,
+ Context: ctx,
+ Span: span,
+ })
})
})
+
+ err = peerConn.SetRemoteDescription(offerContext.SessDesc)
+ if err != nil {
+ return
+ }
+ answer, err = peerConn.CreateAnswer(nil)
+ if err != nil {
+ return
+ }
+
+ answer, err = setAndGatherLocalDescription(peerConn, answer)
return
}
-// getAnswerForOffer creates a transport from a WebRTC offer and and returns a WebRTC answer to be
-// announced.
-func getAnswerForOffer(
- offer webrtc.SessionDescription, onOpen onDataChannelOpen, offerId string,
+// newAnsweringPeerConnection creates a transport from a WebRTC offer and returns a WebRTC answer to be announced.
+func (tc *TrackerClient) newAnsweringPeerConnection(
+ offerContext offerContext,
) (
- answer webrtc.SessionDescription, err error,
+ peerConn *wrappedPeerConnection, answer webrtc.SessionDescription, err error,
) {
- peerConnection, err := newPeerConnection()
+ peerConn, err = newPeerConnection(tc.Logger, tc.ICEServers)
if err != nil {
- err = fmt.Errorf("failed to peer connection: %w", err)
+ err = fmt.Errorf("failed to create new connection: %w", err)
return
}
- answer, err = initAnsweringPeerConnection(peerConnection, offerId, offer, onOpen)
+ answer, err = tc.initAnsweringPeerConnection(peerConn, offerContext)
if err != nil {
- peerConnection.Close()
+ peerConn.span.RecordError(err)
+ peerConn.Close()
}
return
}
-func (t *outboundOffer) setAnswer(answer webrtc.SessionDescription, onOpen func(datachannel.ReadWriteCloser)) error {
- setDataChannelOnOpen(t.dataChannel, t.peerConnection, onOpen)
- err := t.peerConnection.SetRemoteDescription(answer)
- if err == nil {
- // TODO: Maybe grab this inside the onOpen callback and mark the offer used there.
- t.answered = true
- }
- return err
-}
-
type datachannelReadWriter interface {
datachannel.Reader
datachannel.Writer
return me()
}
-func setDataChannelOnOpen(
+func initDataChannel(
dc *webrtc.DataChannel,
- pc wrappedPeerConnection,
- onOpen func(closer datachannel.ReadWriteCloser),
+ pc *wrappedPeerConnection,
+ onOpen onDetachedDataChannelFunc,
) {
+ var span trace.Span
+ dc.OnClose(func() {
+ span.End()
+ })
dc.OnOpen(func() {
+ pc.span.AddEvent("data channel opened")
+ var ctx context.Context
+ ctx, span = otel.Tracer(tracerName).Start(pc.ctx, "DataChannel")
raw, err := dc.Detach()
if err != nil {
// This shouldn't happen if the API is configured correctly, and we call from OnOpen.
panic(err)
}
- onOpen(hookDataChannelCloser(raw, pc))
+ onOpen(hookDataChannelCloser(raw, pc, span, dc), ctx, span)
})
}
-func hookDataChannelCloser(dcrwc datachannel.ReadWriteCloser, pc wrappedPeerConnection) datachannel.ReadWriteCloser {
+// Hooks the datachannel's Close to Close the owning PeerConnection. The datachannel takes ownership
+// and responsibility for the PeerConnection.
+func hookDataChannelCloser(
+ dcrwc datachannel.ReadWriteCloser,
+ pc *wrappedPeerConnection,
+ dataChannelSpan trace.Span,
+ originalDataChannel *webrtc.DataChannel,
+) datachannel.ReadWriteCloser {
return struct {
datachannelReadWriter
io.Closer
}{
dcrwc,
- ioCloserFunc(pc.Close),
+ ioCloserFunc(func() error {
+ dcrwc.Close()
+ pc.Close()
+ originalDataChannel.Close()
+ dataChannelSpan.End()
+ return nil
+ }),
}
}