session.go 18 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721
  1. package yamux
  2. import (
  3. "bufio"
  4. "bytes"
  5. "fmt"
  6. "io"
  7. "io/ioutil"
  8. "log"
  9. "math"
  10. "net"
  11. "strings"
  12. "sync"
  13. "sync/atomic"
  14. "time"
  15. )
  16. // Session is used to wrap a reliable ordered connection and to
  17. // multiplex it into multiple streams.
  18. type Session struct {
  19. // remoteGoAway indicates the remote side does
  20. // not want futher connections. Must be first for alignment.
  21. remoteGoAway int32
  22. // localGoAway indicates that we should stop
  23. // accepting futher connections. Must be first for alignment.
  24. localGoAway int32
  25. // nextStreamID is the next stream we should
  26. // send. This depends if we are a client/server.
  27. nextStreamID uint32
  28. // config holds our configuration
  29. config *Config
  30. // logger is used for our logs
  31. logger *log.Logger
  32. // conn is the underlying connection
  33. conn io.ReadWriteCloser
  34. // bufRead is a buffered reader
  35. bufRead *bufio.Reader
  36. // pings is used to track inflight pings
  37. pings map[uint32]chan struct{}
  38. pingID uint32
  39. pingLock sync.Mutex
  40. // streams maps a stream id to a stream, and inflight has an entry
  41. // for any outgoing stream that has not yet been established. Both are
  42. // protected by streamLock.
  43. streams map[uint32]*Stream
  44. inflight map[uint32]struct{}
  45. streamLock sync.Mutex
  46. // synCh acts like a semaphore. It is sized to the AcceptBacklog which
  47. // is assumed to be symmetric between the client and server. This allows
  48. // the client to avoid exceeding the backlog and instead blocks the open.
  49. synCh chan struct{}
  50. // acceptCh is used to pass ready streams to the client
  51. acceptCh chan *Stream
  52. // sendCh is used to mark a stream as ready to send,
  53. // or to send a header out directly.
  54. sendCh chan sendReady
  55. // recvDoneCh is closed when recv() exits to avoid a race
  56. // between stream registration and stream shutdown
  57. recvDoneCh chan struct{}
  58. // shutdown is used to safely close a session
  59. shutdown bool
  60. shutdownErr error
  61. shutdownCh chan struct{}
  62. shutdownLock sync.Mutex
  63. }
  64. // sendReady is used to either mark a stream as ready
  65. // or to directly send a header
  66. type sendReady struct {
  67. Hdr []byte
  68. mu *sync.Mutex // Protects Body from unsafe reads.
  69. Body []byte
  70. Err chan error
  71. }
  72. // newSession is used to construct a new session
  73. func newSession(config *Config, conn io.ReadWriteCloser, client bool) *Session {
  74. logger := config.Logger
  75. if logger == nil {
  76. logger = log.New(config.LogOutput, "", log.LstdFlags)
  77. }
  78. s := &Session{
  79. config: config,
  80. logger: logger,
  81. conn: conn,
  82. bufRead: bufio.NewReader(conn),
  83. pings: make(map[uint32]chan struct{}),
  84. streams: make(map[uint32]*Stream),
  85. inflight: make(map[uint32]struct{}),
  86. synCh: make(chan struct{}, config.AcceptBacklog),
  87. acceptCh: make(chan *Stream, config.AcceptBacklog),
  88. sendCh: make(chan sendReady, 64),
  89. recvDoneCh: make(chan struct{}),
  90. shutdownCh: make(chan struct{}),
  91. }
  92. if client {
  93. s.nextStreamID = 1
  94. } else {
  95. s.nextStreamID = 2
  96. }
  97. go s.recv()
  98. go s.send()
  99. if config.EnableKeepAlive {
  100. go s.keepalive()
  101. }
  102. return s
  103. }
  104. // IsClosed does a safe check to see if we have shutdown
  105. func (s *Session) IsClosed() bool {
  106. select {
  107. case <-s.shutdownCh:
  108. return true
  109. default:
  110. return false
  111. }
  112. }
  113. // CloseChan returns a read-only channel which is closed as
  114. // soon as the session is closed.
  115. func (s *Session) CloseChan() <-chan struct{} {
  116. return s.shutdownCh
  117. }
  118. // NumStreams returns the number of currently open streams
  119. func (s *Session) NumStreams() int {
  120. s.streamLock.Lock()
  121. num := len(s.streams)
  122. s.streamLock.Unlock()
  123. return num
  124. }
  125. // Open is used to create a new stream as a net.Conn
  126. func (s *Session) Open() (net.Conn, error) {
  127. conn, err := s.OpenStream()
  128. if err != nil {
  129. return nil, err
  130. }
  131. return conn, nil
  132. }
  133. // OpenStream is used to create a new stream
  134. func (s *Session) OpenStream() (*Stream, error) {
  135. if s.IsClosed() {
  136. return nil, ErrSessionShutdown
  137. }
  138. if atomic.LoadInt32(&s.remoteGoAway) == 1 {
  139. return nil, ErrRemoteGoAway
  140. }
  141. // Block if we have too many inflight SYNs
  142. select {
  143. case s.synCh <- struct{}{}:
  144. case <-s.shutdownCh:
  145. return nil, ErrSessionShutdown
  146. }
  147. GET_ID:
  148. // Get an ID, and check for stream exhaustion
  149. id := atomic.LoadUint32(&s.nextStreamID)
  150. if id >= math.MaxUint32-1 {
  151. return nil, ErrStreamsExhausted
  152. }
  153. if !atomic.CompareAndSwapUint32(&s.nextStreamID, id, id+2) {
  154. goto GET_ID
  155. }
  156. // Register the stream
  157. stream := newStream(s, id, streamInit)
  158. s.streamLock.Lock()
  159. s.streams[id] = stream
  160. s.inflight[id] = struct{}{}
  161. s.streamLock.Unlock()
  162. if s.config.StreamOpenTimeout > 0 {
  163. go s.setOpenTimeout(stream)
  164. }
  165. // Send the window update to create
  166. if err := stream.sendWindowUpdate(); err != nil {
  167. select {
  168. case <-s.synCh:
  169. default:
  170. s.logger.Printf("[ERR] yamux: aborted stream open without inflight syn semaphore")
  171. }
  172. return nil, err
  173. }
  174. return stream, nil
  175. }
  176. // setOpenTimeout implements a timeout for streams that are opened but not established.
  177. // If the StreamOpenTimeout is exceeded we assume the peer is unable to ACK,
  178. // and close the session.
  179. // The number of running timers is bounded by the capacity of the synCh.
  180. func (s *Session) setOpenTimeout(stream *Stream) {
  181. timer := time.NewTimer(s.config.StreamOpenTimeout)
  182. defer timer.Stop()
  183. select {
  184. case <-stream.establishCh:
  185. return
  186. case <-s.shutdownCh:
  187. return
  188. case <-timer.C:
  189. // Timeout reached while waiting for ACK.
  190. // Close the session to force connection re-establishment.
  191. s.logger.Printf("[ERR] yamux: aborted stream open (destination=%s): %v", s.RemoteAddr().String(), ErrTimeout.err)
  192. s.Close()
  193. }
  194. }
  195. // Accept is used to block until the next available stream
  196. // is ready to be accepted.
  197. func (s *Session) Accept() (net.Conn, error) {
  198. conn, err := s.AcceptStream()
  199. if err != nil {
  200. return nil, err
  201. }
  202. return conn, err
  203. }
  204. // AcceptStream is used to block until the next available stream
  205. // is ready to be accepted.
  206. func (s *Session) AcceptStream() (*Stream, error) {
  207. select {
  208. case stream := <-s.acceptCh:
  209. if err := stream.sendWindowUpdate(); err != nil {
  210. return nil, err
  211. }
  212. return stream, nil
  213. case <-s.shutdownCh:
  214. return nil, s.shutdownErr
  215. }
  216. }
  217. // Close is used to close the session and all streams.
  218. // Attempts to send a GoAway before closing the connection.
  219. func (s *Session) Close() error {
  220. s.shutdownLock.Lock()
  221. defer s.shutdownLock.Unlock()
  222. if s.shutdown {
  223. return nil
  224. }
  225. s.shutdown = true
  226. if s.shutdownErr == nil {
  227. s.shutdownErr = ErrSessionShutdown
  228. }
  229. close(s.shutdownCh)
  230. s.conn.Close()
  231. <-s.recvDoneCh
  232. s.streamLock.Lock()
  233. defer s.streamLock.Unlock()
  234. for _, stream := range s.streams {
  235. stream.forceClose()
  236. }
  237. return nil
  238. }
  239. // exitErr is used to handle an error that is causing the
  240. // session to terminate.
  241. func (s *Session) exitErr(err error) {
  242. s.shutdownLock.Lock()
  243. if s.shutdownErr == nil {
  244. s.shutdownErr = err
  245. }
  246. s.shutdownLock.Unlock()
  247. s.Close()
  248. }
  249. // GoAway can be used to prevent accepting further
  250. // connections. It does not close the underlying conn.
  251. func (s *Session) GoAway() error {
  252. return s.waitForSend(s.goAway(goAwayNormal), nil)
  253. }
  254. // goAway is used to send a goAway message
  255. func (s *Session) goAway(reason uint32) header {
  256. atomic.SwapInt32(&s.localGoAway, 1)
  257. hdr := header(make([]byte, headerSize))
  258. hdr.encode(typeGoAway, 0, 0, reason)
  259. return hdr
  260. }
  261. // Ping is used to measure the RTT response time
  262. func (s *Session) Ping() (time.Duration, error) {
  263. // Get a channel for the ping
  264. ch := make(chan struct{})
  265. // Get a new ping id, mark as pending
  266. s.pingLock.Lock()
  267. id := s.pingID
  268. s.pingID++
  269. s.pings[id] = ch
  270. s.pingLock.Unlock()
  271. // Send the ping request
  272. hdr := header(make([]byte, headerSize))
  273. hdr.encode(typePing, flagSYN, 0, id)
  274. if err := s.waitForSend(hdr, nil); err != nil {
  275. return 0, err
  276. }
  277. // Wait for a response
  278. start := time.Now()
  279. select {
  280. case <-ch:
  281. case <-time.After(s.config.ConnectionWriteTimeout):
  282. s.pingLock.Lock()
  283. delete(s.pings, id) // Ignore it if a response comes later.
  284. s.pingLock.Unlock()
  285. return 0, ErrTimeout
  286. case <-s.shutdownCh:
  287. return 0, ErrSessionShutdown
  288. }
  289. // Compute the RTT
  290. return time.Now().Sub(start), nil
  291. }
  292. // keepalive is a long running goroutine that periodically does
  293. // a ping to keep the connection alive.
  294. func (s *Session) keepalive() {
  295. for {
  296. select {
  297. case <-time.After(s.config.KeepAliveInterval):
  298. _, err := s.Ping()
  299. if err != nil {
  300. if err != ErrSessionShutdown {
  301. s.logger.Printf("[ERR] yamux: keepalive failed: %v", err)
  302. s.exitErr(ErrKeepAliveTimeout)
  303. }
  304. return
  305. }
  306. case <-s.shutdownCh:
  307. return
  308. }
  309. }
  310. }
  311. // waitForSendErr waits to send a header, checking for a potential shutdown
  312. func (s *Session) waitForSend(hdr header, body []byte) error {
  313. errCh := make(chan error, 1)
  314. return s.waitForSendErr(hdr, body, errCh)
  315. }
  316. // waitForSendErr waits to send a header with optional data, checking for a
  317. // potential shutdown. Since there's the expectation that sends can happen
  318. // in a timely manner, we enforce the connection write timeout here.
  319. func (s *Session) waitForSendErr(hdr header, body []byte, errCh chan error) error {
  320. t := timerPool.Get()
  321. timer := t.(*time.Timer)
  322. timer.Reset(s.config.ConnectionWriteTimeout)
  323. defer func() {
  324. timer.Stop()
  325. select {
  326. case <-timer.C:
  327. default:
  328. }
  329. timerPool.Put(t)
  330. }()
  331. ready := sendReady{Hdr: hdr, mu: &sync.Mutex{}, Body: body, Err: errCh}
  332. select {
  333. case s.sendCh <- ready:
  334. case <-s.shutdownCh:
  335. return ErrSessionShutdown
  336. case <-timer.C:
  337. return ErrConnectionWriteTimeout
  338. }
  339. bodyCopy := func() {
  340. if body == nil {
  341. return // A nil body is ignored.
  342. }
  343. // In the event of session shutdown or connection write timeout,
  344. // we need to prevent `send` from reading the body buffer after
  345. // returning from this function since the caller may re-use the
  346. // underlying array.
  347. ready.mu.Lock()
  348. defer ready.mu.Unlock()
  349. if ready.Body == nil {
  350. return // Body was already copied in `send`.
  351. }
  352. newBody := make([]byte, len(body))
  353. copy(newBody, body)
  354. ready.Body = newBody
  355. }
  356. select {
  357. case err := <-errCh:
  358. return err
  359. case <-s.shutdownCh:
  360. bodyCopy()
  361. return ErrSessionShutdown
  362. case <-timer.C:
  363. bodyCopy()
  364. return ErrConnectionWriteTimeout
  365. }
  366. }
  367. // sendNoWait does a send without waiting. Since there's the expectation that
  368. // the send happens right here, we enforce the connection write timeout if we
  369. // can't queue the header to be sent.
  370. func (s *Session) sendNoWait(hdr header) error {
  371. t := timerPool.Get()
  372. timer := t.(*time.Timer)
  373. timer.Reset(s.config.ConnectionWriteTimeout)
  374. defer func() {
  375. timer.Stop()
  376. select {
  377. case <-timer.C:
  378. default:
  379. }
  380. timerPool.Put(t)
  381. }()
  382. select {
  383. case s.sendCh <- sendReady{Hdr: hdr}:
  384. return nil
  385. case <-s.shutdownCh:
  386. return ErrSessionShutdown
  387. case <-timer.C:
  388. return ErrConnectionWriteTimeout
  389. }
  390. }
  391. // send is a long running goroutine that sends data
  392. func (s *Session) send() {
  393. var bodyBuf bytes.Buffer
  394. for {
  395. bodyBuf.Reset()
  396. select {
  397. case ready := <-s.sendCh:
  398. // Send a header if ready
  399. if ready.Hdr != nil {
  400. _, err := s.conn.Write(ready.Hdr)
  401. if err != nil {
  402. s.logger.Printf("[ERR] yamux: Failed to write header: %v", err)
  403. asyncSendErr(ready.Err, err)
  404. s.exitErr(err)
  405. return
  406. }
  407. }
  408. if ready.mu != nil {
  409. ready.mu.Lock()
  410. if ready.Body != nil {
  411. // Copy the body into the buffer to avoid
  412. // holding a mutex lock during the write.
  413. _, err := bodyBuf.Write(ready.Body)
  414. if err != nil {
  415. ready.Body = nil
  416. ready.mu.Unlock()
  417. s.logger.Printf("[ERR] yamux: Failed to copy body into buffer: %v", err)
  418. asyncSendErr(ready.Err, err)
  419. s.exitErr(err)
  420. return
  421. }
  422. ready.Body = nil
  423. }
  424. ready.mu.Unlock()
  425. }
  426. if bodyBuf.Len() > 0 {
  427. // Send data from a body if given
  428. _, err := s.conn.Write(bodyBuf.Bytes())
  429. if err != nil {
  430. s.logger.Printf("[ERR] yamux: Failed to write body: %v", err)
  431. asyncSendErr(ready.Err, err)
  432. s.exitErr(err)
  433. return
  434. }
  435. }
  436. // No error, successful send
  437. asyncSendErr(ready.Err, nil)
  438. case <-s.shutdownCh:
  439. return
  440. }
  441. }
  442. }
  443. // recv is a long running goroutine that accepts new data
  444. func (s *Session) recv() {
  445. if err := s.recvLoop(); err != nil {
  446. s.exitErr(err)
  447. }
  448. }
  449. // Ensure that the index of the handler (typeData/typeWindowUpdate/etc) matches the message type
  450. var (
  451. handlers = []func(*Session, header) error{
  452. typeData: (*Session).handleStreamMessage,
  453. typeWindowUpdate: (*Session).handleStreamMessage,
  454. typePing: (*Session).handlePing,
  455. typeGoAway: (*Session).handleGoAway,
  456. }
  457. )
  458. // recvLoop continues to receive data until a fatal error is encountered
  459. func (s *Session) recvLoop() error {
  460. defer close(s.recvDoneCh)
  461. hdr := header(make([]byte, headerSize))
  462. for {
  463. // Read the header
  464. if _, err := io.ReadFull(s.bufRead, hdr); err != nil {
  465. if err != io.EOF && !strings.Contains(err.Error(), "closed") && !strings.Contains(err.Error(), "reset by peer") {
  466. s.logger.Printf("[ERR] yamux: Failed to read header: %v", err)
  467. }
  468. return err
  469. }
  470. // Verify the version
  471. if hdr.Version() != protoVersion {
  472. s.logger.Printf("[ERR] yamux: Invalid protocol version: %d", hdr.Version())
  473. return ErrInvalidVersion
  474. }
  475. mt := hdr.MsgType()
  476. if mt < typeData || mt > typeGoAway {
  477. return ErrInvalidMsgType
  478. }
  479. if err := handlers[mt](s, hdr); err != nil {
  480. return err
  481. }
  482. }
  483. }
  484. // handleStreamMessage handles either a data or window update frame
  485. func (s *Session) handleStreamMessage(hdr header) error {
  486. // Check for a new stream creation
  487. id := hdr.StreamID()
  488. flags := hdr.Flags()
  489. if flags&flagSYN == flagSYN {
  490. if err := s.incomingStream(id); err != nil {
  491. return err
  492. }
  493. }
  494. // Get the stream
  495. s.streamLock.Lock()
  496. stream := s.streams[id]
  497. s.streamLock.Unlock()
  498. // If we do not have a stream, likely we sent a RST
  499. if stream == nil {
  500. // Drain any data on the wire
  501. if hdr.MsgType() == typeData && hdr.Length() > 0 {
  502. s.logger.Printf("[WARN] yamux: Discarding data for stream: %d", id)
  503. if _, err := io.CopyN(ioutil.Discard, s.bufRead, int64(hdr.Length())); err != nil {
  504. s.logger.Printf("[ERR] yamux: Failed to discard data: %v", err)
  505. return nil
  506. }
  507. } else {
  508. s.logger.Printf("[WARN] yamux: frame for missing stream: %v", hdr)
  509. }
  510. return nil
  511. }
  512. // Check if this is a window update
  513. if hdr.MsgType() == typeWindowUpdate {
  514. if err := stream.incrSendWindow(hdr, flags); err != nil {
  515. if sendErr := s.sendNoWait(s.goAway(goAwayProtoErr)); sendErr != nil {
  516. s.logger.Printf("[WARN] yamux: failed to send go away: %v", sendErr)
  517. }
  518. return err
  519. }
  520. return nil
  521. }
  522. // Read the new data
  523. if err := stream.readData(hdr, flags, s.bufRead); err != nil {
  524. if sendErr := s.sendNoWait(s.goAway(goAwayProtoErr)); sendErr != nil {
  525. s.logger.Printf("[WARN] yamux: failed to send go away: %v", sendErr)
  526. }
  527. return err
  528. }
  529. return nil
  530. }
  531. // handlePing is invokde for a typePing frame
  532. func (s *Session) handlePing(hdr header) error {
  533. flags := hdr.Flags()
  534. pingID := hdr.Length()
  535. // Check if this is a query, respond back in a separate context so we
  536. // don't interfere with the receiving thread blocking for the write.
  537. if flags&flagSYN == flagSYN {
  538. go func() {
  539. hdr := header(make([]byte, headerSize))
  540. hdr.encode(typePing, flagACK, 0, pingID)
  541. if err := s.sendNoWait(hdr); err != nil {
  542. s.logger.Printf("[WARN] yamux: failed to send ping reply: %v", err)
  543. }
  544. }()
  545. return nil
  546. }
  547. // Handle a response
  548. s.pingLock.Lock()
  549. ch := s.pings[pingID]
  550. if ch != nil {
  551. delete(s.pings, pingID)
  552. close(ch)
  553. }
  554. s.pingLock.Unlock()
  555. return nil
  556. }
  557. // handleGoAway is invokde for a typeGoAway frame
  558. func (s *Session) handleGoAway(hdr header) error {
  559. code := hdr.Length()
  560. switch code {
  561. case goAwayNormal:
  562. atomic.SwapInt32(&s.remoteGoAway, 1)
  563. case goAwayProtoErr:
  564. s.logger.Printf("[ERR] yamux: received protocol error go away")
  565. return fmt.Errorf("yamux protocol error")
  566. case goAwayInternalErr:
  567. s.logger.Printf("[ERR] yamux: received internal error go away")
  568. return fmt.Errorf("remote yamux internal error")
  569. default:
  570. s.logger.Printf("[ERR] yamux: received unexpected go away")
  571. return fmt.Errorf("unexpected go away received")
  572. }
  573. return nil
  574. }
  575. // incomingStream is used to create a new incoming stream
  576. func (s *Session) incomingStream(id uint32) error {
  577. // Reject immediately if we are doing a go away
  578. if atomic.LoadInt32(&s.localGoAway) == 1 {
  579. hdr := header(make([]byte, headerSize))
  580. hdr.encode(typeWindowUpdate, flagRST, id, 0)
  581. return s.sendNoWait(hdr)
  582. }
  583. // Allocate a new stream
  584. stream := newStream(s, id, streamSYNReceived)
  585. s.streamLock.Lock()
  586. defer s.streamLock.Unlock()
  587. // Check if stream already exists
  588. if _, ok := s.streams[id]; ok {
  589. s.logger.Printf("[ERR] yamux: duplicate stream declared")
  590. if sendErr := s.sendNoWait(s.goAway(goAwayProtoErr)); sendErr != nil {
  591. s.logger.Printf("[WARN] yamux: failed to send go away: %v", sendErr)
  592. }
  593. return ErrDuplicateStream
  594. }
  595. // Register the stream
  596. s.streams[id] = stream
  597. // Check if we've exceeded the backlog
  598. select {
  599. case s.acceptCh <- stream:
  600. return nil
  601. default:
  602. // Backlog exceeded! RST the stream
  603. s.logger.Printf("[WARN] yamux: backlog exceeded, forcing connection reset")
  604. delete(s.streams, id)
  605. hdr := header(make([]byte, headerSize))
  606. hdr.encode(typeWindowUpdate, flagRST, id, 0)
  607. return s.sendNoWait(hdr)
  608. }
  609. }
  610. // closeStream is used to close a stream once both sides have
  611. // issued a close. If there was an in-flight SYN and the stream
  612. // was not yet established, then this will give the credit back.
  613. func (s *Session) closeStream(id uint32) {
  614. s.streamLock.Lock()
  615. if _, ok := s.inflight[id]; ok {
  616. select {
  617. case <-s.synCh:
  618. default:
  619. s.logger.Printf("[ERR] yamux: SYN tracking out of sync")
  620. }
  621. }
  622. delete(s.streams, id)
  623. s.streamLock.Unlock()
  624. }
  625. // establishStream is used to mark a stream that was in the
  626. // SYN Sent state as established.
  627. func (s *Session) establishStream(id uint32) {
  628. s.streamLock.Lock()
  629. if _, ok := s.inflight[id]; ok {
  630. delete(s.inflight, id)
  631. } else {
  632. s.logger.Printf("[ERR] yamux: established stream without inflight SYN (no tracking entry)")
  633. }
  634. select {
  635. case <-s.synCh:
  636. default:
  637. s.logger.Printf("[ERR] yamux: established stream without inflight SYN (didn't have semaphore)")
  638. }
  639. s.streamLock.Unlock()
  640. }