session.go 18 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719
  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, 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. ready.mu.Lock()
  409. if ready.Body != nil {
  410. // Copy the body into the buffer to avoid
  411. // holding a mutex lock during the write.
  412. _, err := bodyBuf.Write(ready.Body)
  413. if err != nil {
  414. ready.Body = nil
  415. ready.mu.Unlock()
  416. s.logger.Printf("[ERR] yamux: Failed to copy body into buffer: %v", err)
  417. asyncSendErr(ready.Err, err)
  418. s.exitErr(err)
  419. return
  420. }
  421. ready.Body = nil
  422. }
  423. ready.mu.Unlock()
  424. if bodyBuf.Len() > 0 {
  425. // Send data from a body if given
  426. _, err := s.conn.Write(bodyBuf.Bytes())
  427. if err != nil {
  428. s.logger.Printf("[ERR] yamux: Failed to write body: %v", err)
  429. asyncSendErr(ready.Err, err)
  430. s.exitErr(err)
  431. return
  432. }
  433. }
  434. // No error, successful send
  435. asyncSendErr(ready.Err, nil)
  436. case <-s.shutdownCh:
  437. return
  438. }
  439. }
  440. }
  441. // recv is a long running goroutine that accepts new data
  442. func (s *Session) recv() {
  443. if err := s.recvLoop(); err != nil {
  444. s.exitErr(err)
  445. }
  446. }
  447. // Ensure that the index of the handler (typeData/typeWindowUpdate/etc) matches the message type
  448. var (
  449. handlers = []func(*Session, header) error{
  450. typeData: (*Session).handleStreamMessage,
  451. typeWindowUpdate: (*Session).handleStreamMessage,
  452. typePing: (*Session).handlePing,
  453. typeGoAway: (*Session).handleGoAway,
  454. }
  455. )
  456. // recvLoop continues to receive data until a fatal error is encountered
  457. func (s *Session) recvLoop() error {
  458. defer close(s.recvDoneCh)
  459. hdr := header(make([]byte, headerSize))
  460. for {
  461. // Read the header
  462. if _, err := io.ReadFull(s.bufRead, hdr); err != nil {
  463. if err != io.EOF && !strings.Contains(err.Error(), "closed") && !strings.Contains(err.Error(), "reset by peer") {
  464. s.logger.Printf("[ERR] yamux: Failed to read header: %v", err)
  465. }
  466. return err
  467. }
  468. // Verify the version
  469. if hdr.Version() != protoVersion {
  470. s.logger.Printf("[ERR] yamux: Invalid protocol version: %d", hdr.Version())
  471. return ErrInvalidVersion
  472. }
  473. mt := hdr.MsgType()
  474. if mt < typeData || mt > typeGoAway {
  475. return ErrInvalidMsgType
  476. }
  477. if err := handlers[mt](s, hdr); err != nil {
  478. return err
  479. }
  480. }
  481. }
  482. // handleStreamMessage handles either a data or window update frame
  483. func (s *Session) handleStreamMessage(hdr header) error {
  484. // Check for a new stream creation
  485. id := hdr.StreamID()
  486. flags := hdr.Flags()
  487. if flags&flagSYN == flagSYN {
  488. if err := s.incomingStream(id); err != nil {
  489. return err
  490. }
  491. }
  492. // Get the stream
  493. s.streamLock.Lock()
  494. stream := s.streams[id]
  495. s.streamLock.Unlock()
  496. // If we do not have a stream, likely we sent a RST
  497. if stream == nil {
  498. // Drain any data on the wire
  499. if hdr.MsgType() == typeData && hdr.Length() > 0 {
  500. s.logger.Printf("[WARN] yamux: Discarding data for stream: %d", id)
  501. if _, err := io.CopyN(ioutil.Discard, s.bufRead, int64(hdr.Length())); err != nil {
  502. s.logger.Printf("[ERR] yamux: Failed to discard data: %v", err)
  503. return nil
  504. }
  505. } else {
  506. s.logger.Printf("[WARN] yamux: frame for missing stream: %v", hdr)
  507. }
  508. return nil
  509. }
  510. // Check if this is a window update
  511. if hdr.MsgType() == typeWindowUpdate {
  512. if err := stream.incrSendWindow(hdr, flags); err != nil {
  513. if sendErr := s.sendNoWait(s.goAway(goAwayProtoErr)); sendErr != nil {
  514. s.logger.Printf("[WARN] yamux: failed to send go away: %v", sendErr)
  515. }
  516. return err
  517. }
  518. return nil
  519. }
  520. // Read the new data
  521. if err := stream.readData(hdr, flags, s.bufRead); err != nil {
  522. if sendErr := s.sendNoWait(s.goAway(goAwayProtoErr)); sendErr != nil {
  523. s.logger.Printf("[WARN] yamux: failed to send go away: %v", sendErr)
  524. }
  525. return err
  526. }
  527. return nil
  528. }
  529. // handlePing is invokde for a typePing frame
  530. func (s *Session) handlePing(hdr header) error {
  531. flags := hdr.Flags()
  532. pingID := hdr.Length()
  533. // Check if this is a query, respond back in a separate context so we
  534. // don't interfere with the receiving thread blocking for the write.
  535. if flags&flagSYN == flagSYN {
  536. go func() {
  537. hdr := header(make([]byte, headerSize))
  538. hdr.encode(typePing, flagACK, 0, pingID)
  539. if err := s.sendNoWait(hdr); err != nil {
  540. s.logger.Printf("[WARN] yamux: failed to send ping reply: %v", err)
  541. }
  542. }()
  543. return nil
  544. }
  545. // Handle a response
  546. s.pingLock.Lock()
  547. ch := s.pings[pingID]
  548. if ch != nil {
  549. delete(s.pings, pingID)
  550. close(ch)
  551. }
  552. s.pingLock.Unlock()
  553. return nil
  554. }
  555. // handleGoAway is invokde for a typeGoAway frame
  556. func (s *Session) handleGoAway(hdr header) error {
  557. code := hdr.Length()
  558. switch code {
  559. case goAwayNormal:
  560. atomic.SwapInt32(&s.remoteGoAway, 1)
  561. case goAwayProtoErr:
  562. s.logger.Printf("[ERR] yamux: received protocol error go away")
  563. return fmt.Errorf("yamux protocol error")
  564. case goAwayInternalErr:
  565. s.logger.Printf("[ERR] yamux: received internal error go away")
  566. return fmt.Errorf("remote yamux internal error")
  567. default:
  568. s.logger.Printf("[ERR] yamux: received unexpected go away")
  569. return fmt.Errorf("unexpected go away received")
  570. }
  571. return nil
  572. }
  573. // incomingStream is used to create a new incoming stream
  574. func (s *Session) incomingStream(id uint32) error {
  575. // Reject immediately if we are doing a go away
  576. if atomic.LoadInt32(&s.localGoAway) == 1 {
  577. hdr := header(make([]byte, headerSize))
  578. hdr.encode(typeWindowUpdate, flagRST, id, 0)
  579. return s.sendNoWait(hdr)
  580. }
  581. // Allocate a new stream
  582. stream := newStream(s, id, streamSYNReceived)
  583. s.streamLock.Lock()
  584. defer s.streamLock.Unlock()
  585. // Check if stream already exists
  586. if _, ok := s.streams[id]; ok {
  587. s.logger.Printf("[ERR] yamux: duplicate stream declared")
  588. if sendErr := s.sendNoWait(s.goAway(goAwayProtoErr)); sendErr != nil {
  589. s.logger.Printf("[WARN] yamux: failed to send go away: %v", sendErr)
  590. }
  591. return ErrDuplicateStream
  592. }
  593. // Register the stream
  594. s.streams[id] = stream
  595. // Check if we've exceeded the backlog
  596. select {
  597. case s.acceptCh <- stream:
  598. return nil
  599. default:
  600. // Backlog exceeded! RST the stream
  601. s.logger.Printf("[WARN] yamux: backlog exceeded, forcing connection reset")
  602. delete(s.streams, id)
  603. hdr := header(make([]byte, headerSize))
  604. hdr.encode(typeWindowUpdate, flagRST, id, 0)
  605. return s.sendNoWait(hdr)
  606. }
  607. }
  608. // closeStream is used to close a stream once both sides have
  609. // issued a close. If there was an in-flight SYN and the stream
  610. // was not yet established, then this will give the credit back.
  611. func (s *Session) closeStream(id uint32) {
  612. s.streamLock.Lock()
  613. if _, ok := s.inflight[id]; ok {
  614. select {
  615. case <-s.synCh:
  616. default:
  617. s.logger.Printf("[ERR] yamux: SYN tracking out of sync")
  618. }
  619. }
  620. delete(s.streams, id)
  621. s.streamLock.Unlock()
  622. }
  623. // establishStream is used to mark a stream that was in the
  624. // SYN Sent state as established.
  625. func (s *Session) establishStream(id uint32) {
  626. s.streamLock.Lock()
  627. if _, ok := s.inflight[id]; ok {
  628. delete(s.inflight, id)
  629. } else {
  630. s.logger.Printf("[ERR] yamux: established stream without inflight SYN (no tracking entry)")
  631. }
  632. select {
  633. case <-s.synCh:
  634. default:
  635. s.logger.Printf("[ERR] yamux: established stream without inflight SYN (didn't have semaphore)")
  636. }
  637. s.streamLock.Unlock()
  638. }