func (s *Server) Serve(ln net.Listener) error {
	var lastOverflowErrorTime time.Time
	var lastPerIPErrorTime time.Time
	var c net.Conn
	var err error

	s.mu.Lock()
	{
		if s.ln != nil {
			s.mu.Unlock()
			return ErrAlreadyServing
		}

		s.ln = ln
		s.done = make(chan struct{})
	}
	s.mu.Unlock()

	maxWorkersCount := s.getConcurrency()
	s.concurrencyCh = make(chan struct{}, maxWorkersCount)
	wp := &workerPool{
		WorkerFunc:      s.serveConn,
		MaxWorkersCount: maxWorkersCount,
		LogAllErrors:    s.LogAllErrors,
		Logger:          s.logger(),
		connState:       s.setState,
	}
	wp.Start()

	// Count our waiting to accept a connection as an open connection.
	// This way we can't get into any weird state where just after accepting
	// a connection Shutdown is called which reads open as 0 because it isn't
	// incremented yet.
	atomic.AddInt32(&s.open, 1)
	defer atomic.AddInt32(&s.open, -1)

	for {
		if c, err = acceptConn(s, ln, &lastPerIPErrorTime); err != nil {
			wp.Stop()
			if err == io.EOF {
				return nil
			}
			return err
		}
		s.setState(c, StateNew)
		atomic.AddInt32(&s.open, 1)
		if !wp.Serve(c) {
			atomic.AddInt32(&s.open, -1)
			s.writeFastError(c, StatusServiceUnavailable,
				"The connection cannot be served because Server.Concurrency limit exceeded")
			c.Close()
			s.setState(c, StateClosed)
			if time.Since(lastOverflowErrorTime) > time.Minute {
				s.logger().Printf("The incoming connection cannot be served, because %d concurrent connections are served. "+
					"Try increasing Server.Concurrency", maxWorkersCount)
				lastOverflowErrorTime = time.Now()
			}

			// The current server reached concurrency limit,
			// so give other concurrently running servers a chance
			// accepting incoming connections on the same address.
			//
			// There is a hope other servers didn't reach their
			// concurrency limits yet :)
			//
			// See also: https://github.com/valyala/fasthttp/pull/485#discussion_r239994990
			if s.SleepWhenConcurrencyLimitsExceeded > 0 {
				time.Sleep(s.SleepWhenConcurrencyLimitsExceeded)
			}
		}
		c = nil
	}
}
func (s *Server) Serve(ln net.Listener) error {
	var lastOverflowErrorTime time.Time
	var lastPerIPErrorTime time.Time
	var c net.Conn
	var err error

	s.mu.Lock()
	{
		if s.ln != nil {
			s.mu.Unlock()
			return ErrAlreadyServing
		}

		s.ln = ln
		s.done = make(chan struct{})
	}
	s.mu.Unlock()

	maxWorkersCount := s.getConcurrency()
	s.concurrencyCh = make(chan struct{}, maxWorkersCount)
	wp := &workerPool{
		WorkerFunc:      s.serveConn,
		MaxWorkersCount: maxWorkersCount,
		LogAllErrors:    s.LogAllErrors,
		Logger:          s.logger(),
		connState:       s.setState,
	}
	wp.Start()

	
	
	
	
	atomic.AddInt32(&s.open, 1)
	defer atomic.AddInt32(&s.open, -1)

	for {
		if c, err = acceptConn(s, ln, &lastPerIPErrorTime); err != nil {
			wp.Stop()
			if err == io.EOF {
				return nil
			}
			return err
		}
		s.setState(c, StateNew)
		atomic.AddInt32(&s.open, 1)
		if !wp.Serve(c) {
			atomic.AddInt32(&s.open, -1)
			s.writeFastError(c, StatusServiceUnavailable,
				"The connection cannot be served because Server.Concurrency limit exceeded")
			c.Close()
			s.setState(c, StateClosed)
			if time.Since(lastOverflowErrorTime) > time.Minute {
				s.logger().Printf("The incoming connection cannot be served, because %d concurrent connections are served. "+
					"Try increasing Server.Concurrency", maxWorkersCount)
				lastOverflowErrorTime = time.Now()
			}

			
			
			
			
			
			
			
			
			if s.SleepWhenConcurrencyLimitsExceeded > 0 {
				time.Sleep(s.SleepWhenConcurrencyLimitsExceeded)
			}
		}
		c = nil
	}
}
func ( s * Server ) Serve ( ln net . Listener ) error { var lastOverflowErrorTime time . Time var lastPerIPErrorTime time . Time var c net . Conn var err error s . mu . Lock ( ) { if s . ln != nil { s . mu . Unlock ( ) return ErrAlreadyServing } s . ln = ln s . done = make ( chan struct { } ) } s . mu . Unlock ( ) maxWorkersCount := s . getConcurrency ( ) s . concurrencyCh = make ( chan struct { } , maxWorkersCount ) wp := & workerPool { WorkerFunc : s . serveConn , MaxWorkersCount : maxWorkersCount , LogAllErrors : s . LogAllErrors , Logger : s . logger ( ) , connState : s . setState , } wp . Start ( ) // Count our waiting to accept a connection as an open connection. // This way we can't get into any weird state where just after accepting // a connection Shutdown is called which reads open as 0 because it isn't // incremented yet. atomic . AddInt32 ( & s . open , 1 ) defer atomic . AddInt32 ( & s . open , - 1 ) for { if c , err = acceptConn ( s , ln , & lastPerIPErrorTime ) ; err != nil { wp . Stop ( ) if err == io . EOF { return nil } return err } s . setState ( c , StateNew ) atomic . AddInt32 ( & s . open , 1 ) if ! wp . Serve ( c ) { atomic . AddInt32 ( & s . open , - 1 ) s . writeFastError ( c , StatusServiceUnavailable , " " ) c . Close ( ) s . setState ( c , StateClosed ) if time . Since ( lastOverflowErrorTime ) > time . Minute { s . logger ( ) . Printf ( " " + " " , maxWorkersCount ) lastOverflowErrorTime = time . Now ( ) } // The current server reached concurrency limit, // so give other concurrently running servers a chance // accepting incoming connections on the same address. // // There is a hope other servers didn't reach their // concurrency limits yet :) // // See also: https://github.com/valyala/fasthttp/pull/485#discussion_r239994990 if s . SleepWhenConcurrencyLimitsExceeded > 0 { time . Sleep ( s . SleepWhenConcurrencyLimitsExceeded ) } } c = nil } }
// Serve serves incoming connections from the given listener.
//
// Serve blocks until the given listener returns permanent error.
Serve serves incoming connections from the given listener . Serve blocks until the given listener returns permanent error .