package datagramsession import ( "context" "fmt" "io" "time" "github.com/google/uuid" "github.com/lucas-clemente/quic-go" "github.com/rs/zerolog" "golang.org/x/sync/errgroup" ) const ( requestChanCapacity = 16 defaultReqTimeout = time.Second * 5 ) var ( errSessionManagerClosed = fmt.Errorf("session manager closed") ) // Manager defines the APIs to manage sessions from the same transport. type Manager interface { // Serve starts the event loop Serve(ctx context.Context) error // RegisterSession starts tracking a session. Caller is responsible for starting the session RegisterSession(ctx context.Context, sessionID uuid.UUID, dstConn io.ReadWriteCloser) (*Session, error) // UnregisterSession stops tracking the session and terminates it UnregisterSession(ctx context.Context, sessionID uuid.UUID, message string, byRemote bool) error // UpdateLogger updates the logger used by the Manager UpdateLogger(log *zerolog.Logger) } type manager struct { registrationChan chan *registerSessionEvent unregistrationChan chan *unregisterSessionEvent datagramChan chan *newDatagram closedChan chan struct{} transport transport sessions map[uuid.UUID]*Session log *zerolog.Logger // timeout waiting for an API to finish. This can be overriden in test timeout time.Duration } func NewManager(transport transport, log *zerolog.Logger) *manager { return &manager{ registrationChan: make(chan *registerSessionEvent), unregistrationChan: make(chan *unregisterSessionEvent), // datagramChan is buffered, so it can read more datagrams from transport while the event loop is processing other events datagramChan: make(chan *newDatagram, requestChanCapacity), closedChan: make(chan struct{}), transport: transport, sessions: make(map[uuid.UUID]*Session), log: log, timeout: defaultReqTimeout, } } func (m *manager) UpdateLogger(log *zerolog.Logger) { // Benign data race, no problem if the old pointer is read or not concurrently. m.log = log } func (m *manager) Serve(ctx context.Context) error { errGroup, ctx := errgroup.WithContext(ctx) errGroup.Go(func() error { for { sessionID, payload, err := m.transport.ReceiveFrom() if err != nil { if aerr, ok := err.(*quic.ApplicationError); ok && uint64(aerr.ErrorCode) == uint64(quic.NoError) { return nil } else { return err } } datagram := &newDatagram{ sessionID: sessionID, payload: payload, } select { case <-ctx.Done(): return ctx.Err() // Only the event loop routine can update/lookup the sessions map to avoid concurrent access // Send the datagram to the event loop. It will find the session to send to case m.datagramChan <- datagram: } } }) errGroup.Go(func() error { for { select { case <-ctx.Done(): return nil case datagram := <-m.datagramChan: m.sendToSession(datagram) case registration := <-m.registrationChan: m.registerSession(ctx, registration) // TODO: TUN-5422: Unregister inactive session upon timeout case unregistration := <-m.unregistrationChan: m.unregisterSession(unregistration) } } }) err := errGroup.Wait() close(m.closedChan) m.shutdownSessions(err) return err } func (m *manager) shutdownSessions(err error) { if err == nil { err = errSessionManagerClosed } closeSessionErr := &errClosedSession{ message: err.Error(), // Usually connection with remote has been closed, so set this to true to skip unregistering from remote byRemote: true, } for _, s := range m.sessions { s.close(closeSessionErr) } } func (m *manager) RegisterSession(ctx context.Context, sessionID uuid.UUID, originProxy io.ReadWriteCloser) (*Session, error) { ctx, cancel := context.WithTimeout(ctx, m.timeout) defer cancel() event := newRegisterSessionEvent(sessionID, originProxy) select { case <-ctx.Done(): m.log.Error().Msg("Datagram session registration timeout") return nil, ctx.Err() case m.registrationChan <- event: session := <-event.resultChan return session, nil // Once closedChan is closed, manager won't accept more registration because nothing is // reading from registrationChan and it's an unbuffered channel case <-m.closedChan: return nil, errSessionManagerClosed } } func (m *manager) registerSession(ctx context.Context, registration *registerSessionEvent) { session := m.newSession(registration.sessionID, registration.originProxy) m.sessions[registration.sessionID] = session registration.resultChan <- session } func (m *manager) newSession(id uuid.UUID, dstConn io.ReadWriteCloser) *Session { return &Session{ ID: id, transport: m.transport, dstConn: dstConn, // activeAtChan has low capacity. It can be full when there are many concurrent read/write. markActive() will // drop instead of blocking because last active time only needs to be an approximation activeAtChan: make(chan time.Time, 2), // capacity is 2 because close() and dstToTransport routine in Serve() can write to this channel closeChan: make(chan error, 2), log: m.log, } } func (m *manager) UnregisterSession(ctx context.Context, sessionID uuid.UUID, message string, byRemote bool) error { ctx, cancel := context.WithTimeout(ctx, m.timeout) defer cancel() event := &unregisterSessionEvent{ sessionID: sessionID, err: &errClosedSession{ message: message, byRemote: byRemote, }, } select { case <-ctx.Done(): m.log.Error().Msg("Datagram session unregistration timeout") return ctx.Err() case m.unregistrationChan <- event: return nil case <-m.closedChan: return errSessionManagerClosed } } func (m *manager) unregisterSession(unregistration *unregisterSessionEvent) { session, ok := m.sessions[unregistration.sessionID] if ok { delete(m.sessions, unregistration.sessionID) session.close(unregistration.err) } } func (m *manager) sendToSession(datagram *newDatagram) { session, ok := m.sessions[datagram.sessionID] if !ok { m.log.Error().Str("sessionID", datagram.sessionID.String()).Msg("session not found") return } // session writes to destination over a connected UDP socket, which should not be blocking, so this call doesn't // need to run in another go routine _, err := session.transportToDst(datagram.payload) if err != nil { m.log.Err(err).Str("sessionID", datagram.sessionID.String()).Msg("Failed to write payload to session") } }