session.go 18 KB

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