func (p *multiplexer) bestSend(data []byte, action_desc string) bool {
	var buf = make([]byte, FRAME_HEADER_LEN+len(data))
	_frame(buf, FRAME_ACTION_TOKENS, 0, data)
	var tun *Conn
	for i := 1; i <= 3; i++ {
		if p.status < 0 /* MUX_CLOSED */ || p.pool == nil {
			if log.V(4) {
				log.Warningln("abandon sending data of", action_desc)
			}
			break
		}
		tun = p.pool.Select()
		if tun != nil {
			if tunWrite1(tun, buf) == nil {
				return true
			}
		} else {
			time.Sleep(time.Millisecond * 200 * time.Duration(i))
		}
	}
	if log.V(3) {
		log.Warningln("failed to send data of", action_desc)
	}
	return false
}

func (c *Client) initialNegotiation() (tun *Conn) {
	var newParams = new(tunParams)
	var err error
	tun, err = c.nego.negotiate(newParams)
	if err != nil {
		if log.V(1) == true || DEBUG {
			log.Errorf("Connection failed %s, Error: %s. Retry after %s",
				c.nego.RemoteName(), err, RETRY_INTERVAL)
		} else {
			log.Errorf("Connection failed %s. Retry after %s",
				c.nego.RemoteName(), RETRY_INTERVAL)
		}
		if strings.Contains(err.Error(), "closed") {
			log.Warningln(string(bytes.Repeat([]byte{'+'}, 30)))
			log.Warningln("Maybe your clock is inaccurate, or your client credential is invalid.")
			log.Warningln(string(bytes.Repeat([]byte{'+'}, 30)))
			os.Exit(2)
		}
		return nil
	}
	c.params = newParams
	c.token = newParams.token

	tun.identifier = c.nego.RemoteName()
	log.Infof("Login to the gateway %s successfully", tun.identifier)
	return
}

func (c *Client) ClientServe(conn net.Conn) {
	var done bool
	defer func() {
		if e := recover(); e != nil {
			log.Warningln(e)
		}
		if !done {
			SafeClose(conn)
		}
	}()

	reqNum := atomic.AddInt32(&c.reqCnt, 1)
	pbConn := NewPushbackInputStream(conn)
	proto, e := detectProtocol(pbConn)
	if e != nil {
		// chrome will make some advance connections and then aborted
		// cause a EOF
		if e != io.EOF && e != io.ErrUnexpectedEOF {
			log.Warningln(e)
		}
		return
	}
	switch proto {
	case REQ_PROT_SOCKS5:
		s5 := s5Handler{conn: pbConn}
		s5.handshake()
		if !s5.handshakeResponse() {
			literalTarget := s5.parseRequest()
			if !s5.finalResponse() {
				c.mux.HandleRequest("SOCKS5", conn, literalTarget)
				done = true
			}
		}
	case REQ_PROT_HTTP:
		prot, literalTarget := httpProxyHandshake(pbConn)
		if prot == REQ_PROT_HTTP { // plain http
			c.mux.HandleRequest("HTTP", pbConn, literalTarget)
		} else { // http tunnel
			c.mux.HandleRequest("HTTP/T", conn, literalTarget)
		}
		done = true
	default:
		log.Warningln("unrecognized request from", conn.RemoteAddr())
		time.Sleep(REST_INTERVAL)
	}
	// client setSeed at every 32 req
	if reqNum&0x1f == 0x1f {
		myRand.setSeed(0)
	}
}

func (c *Client) getToken() ([]byte, error) {
	c.lock.Lock()

	var tlen = len(c.token) / TKSZ
	if tlen <= TOKENS_FLOOR {
		c.asyncRequestTokens()
	}
	for len(c.token) < TKSZ {
		// release lock for waiting of pendingTK()
		c.lock.Unlock()
		log.Warningln("Waiting for token. Maybe the requests are coming too fast.")
		if !c.pendingTK.acquire(RETRY_INTERVAL * 2) {
			return nil, ERR_REQ_TK_TIMEOUT
		}
		if atomic.LoadInt32(&c.state) < CLT_WORKING {
			return nil, ERR_REQ_TK_ABORTED
		}
		// recover lock status
		c.lock.Lock()
	}
	var token = c.token[:TKSZ]
	c.token = c.token[TKSZ:]
	// finally release
	c.lock.Unlock()
	return token, nil
}

func (nego *d5SNegotiation) dataSession(hConn *hashedConn, buf []byte) (session *Session, err error) {
	token := buf[:TKSZ]
	if session := nego.sessionMgr.take(token); session != nil {
		session.identifyConn(hConn.Conn)
		nego.tokenBuf = buf
		return session, DATATUN_SESSION
	}
	log.Warningln("Incorrect token from", hConn.RemoteAddr())
	return nil, VALIDATION_FAILED
}

func (p *multiplexer) bestSend(data []byte, action_desc string) bool {
	var buf = make([]byte, FRAME_HEADER_LEN+len(data))
	pack(buf, FRAME_ACTION_TOKENS, 0, data)

	for i := 1; i <= 3; i++ {
		if atomic.LoadInt32(&p.status) < 0 /* MUX_CLOSED */ || p.pool == nil {
			log.Warningln("abandon sending data of", action_desc)
			break
		}
		tun := p.pool.Select()
		if tun != nil {
			if frameWriteBuffer(tun, buf) == nil {
				return true
			}
		} else {
			time.Sleep(time.Millisecond * 200 * time.Duration(i))
		}
	}
	log.Warningln("failed to send data of", action_desc)
	return false
}

func (c *Client) ClientServe(conn net.Conn) {
	var done bool
	defer func() {
		if e := recover(); e != nil {
			log.Warningln(e)
		}
		if !done {
			SafeClose(conn)
		}
	}()

	pbConn := NewPushbackInputStream(conn)
	switch detectProtocol(pbConn) {
	case REQ_PROT_SOCKS5:
		s5 := S5Step1{conn: pbConn}
		s5.Handshake()
		if !s5.HandshakeAck() {
			literalTarget := s5.parseSocks5Request()
			if !s5.respondSocks5() {
				c.mux.HandleRequest("SOCKS5", conn, literalTarget)
				done = true
			}
		}
	case REQ_PROT_HTTP:
		prot, literalTarget := httpProxyHandshake(pbConn)
		if prot == REQ_PROT_HTTP { // plain http
			c.mux.HandleRequest("HTTP", pbConn, literalTarget)
		} else { // http tunnel
			c.mux.HandleRequest("HTTP/T", conn, literalTarget)
		}
		done = true
	default:
		log.Warningln("unrecognized request from", conn.RemoteAddr())
		time.Sleep(REST_INTERVAL)
	}

}

// step1 response
// return: True=Denied
func (s *s5Handler) handshakeResponse() bool {
	msg := []byte{5, 0}
	if s.err != nil {
		// handshake error feedback
		log.Warningln(s.err)
		if ex, y := s.err.(*exception.Exception); y {
			msg[1] = byte(ex.Code())
		} else {
			msg[1] = 0xff
		}
		setWTimeout(s.conn)
		s.conn.Write(msg)
		return true
	}

	// accept
	setWTimeout(s.conn)
	_, err := s.conn.Write(msg)
	if err != nil {
		log.Warningln(err)
		return true
	}
	return false
}

// new connection
func (n *dbcSerNego) handshakeSession(hConn *hashedConn) (session *Session, err error) {
	defer func() {
		// free ibHash
		n.ibHash = nil
		if e, y := exception.ErrorOf(recover()); y {
			log.Warningln("handshake error", e)
			err = e
		}
	}()
	var skey = n.verifyThenDHExchange(hConn)
	var cf = NewCipherFactory(n.Cipher, skey)

	hConn.cipher = cf.InitCipher(n.ibHash)
	session = NewSession(hConn.Conn, cf, n)
	n.isNewSession = true
	n.respondTestWithToken(hConn, session)
	return
}

// quick resume session
func (n *dbcSerNego) dataSession(hConn *hashedConn) (session *Session, err error) {
	token := make([]byte, TKSZ)
	setRTimeout(hConn)
	nr, err := hConn.Read(token)
	// read buf ok
	if nr == len(token) && err == nil {
		// check token ok
		if session := n.sessionMgr.take(token); session != nil {
			// init cipher of new connection
			hConn.cipher = session.cipherFactory.InitCipher(token)
			// check and set identify
			session.identifyConn(hConn.Conn)
			return session, nil
		}
	}
	log.Warningln("Incorrect token from", n.clientAddr, nvl(err, NULL))
	return nil, VALIDATION_FAILED
}

func CatchException(e interface{}) bool {
	if e != nil {
		if s, y := e.(string); y {
			log.Warningln(s)
		} else if ex, y := e.(*Exception); y && ex.warning {
			log.Errorln(ex.msg)
			return true
		} else {
			log.Errorln(e)
		}
		if DEBUG || bool(log.V(3)) {
			buf := make([]byte, 1600)
			n := runtime.Stack(buf, false)
			log.DirectPrintln(string(buf[:n]))
		}
		return true
	}
	return false
}

func (s *S5Step1) Handshake() {
	var buf = make([]byte, 2)
	_, err := io.ReadFull(s.conn, buf)
	if err != nil {
		s.err = INVALID_SOCKS5_HEADER.Apply(err)
		return
	}
	ver, nmethods := buf[0], int(buf[1])
	if ver != SOCKS5_VER || nmethods < 1 {
		s.err = INVALID_SOCKS5_HEADER.Apply(fmt.Sprintf("[% x]", buf[:2]))
		return
	}
	buf = make([]byte, nmethods+1) // consider method non-00
	n, err := io.ReadAtLeast(s.conn, buf, nmethods)
	if err != nil || n != nmethods {
		s.err = INVALID_SOCKS5_HEADER
		log.Warningln("invalid socks5 header:", hex.EncodeToString(buf))
	}
}

func (t *Server) TunnelServe(conn *net.TCPConn) {
	fconn := NewConnWithHash(conn)
	defer func() {
		fconn.FreeHash()
		ex.CatchException(recover())
	}()
	nego := &d5SNegotiation{Server: t}
	session, err := nego.negotiate(fconn)

	if err == nil || err == DATATUN_SESSION { // dataTunnel
		go session.DataTunServe(fconn.Conn, nego.tokenBuf)
	} else {
		log.Warningln("Close abnormal connection from", conn.RemoteAddr(), err)
		SafeClose(conn)
		if session != nil {
			t.sessionMgr.clearTokens(session)
		}
	}
}

func (c *Client) initialNegotiation() (tun *Conn) {
	var tp = new(tunParams)
	var err error
	tun, err = c.nego.negotiate(tp)
	if err != nil {
		log.Errorf("Failed to connect %s, Retry after %s\n", c.nego.RemoteName(), RETRY_INTERVAL)
		if strings.Contains(err.Error(), "closed") {
			log.Warningln("Maybe your clock is inaccurate, or your client credential is invalid.")
			os.Exit(2)
		}
		return nil
	}
	c.token, c.tp = tp.token, tp
	tp.token = nil
	tun.identifier = c.nego.RemoteName()

	log.Infoln("Login to the gateway", c.nego.RemoteName(), "successfully")
	return tun
}

func (c *Client) getToken() []byte {
	c.lock.Lock()
	defer c.lock.Unlock()

	var tlen = len(c.token) / TKSZ
	if tlen <= TOKENS_FLOOR {
		c.requireTokens()
	}
	for len(c.token) < TKSZ {
		log.Warningln("waiting for token. May be the requests are coming too fast.")
		if !c.pendingTK.acquire(RETRY_INTERVAL * 2) { // discarded request
			panic("Aborted")
		}
		if atomic.LoadInt32(&c.State) < CLT_WORKING {
			panic("Abandon the request by shutdown.")
		}
	}
	var token = c.token[:TKSZ]
	c.token = c.token[TKSZ:]
	return token
}

func CatchException(e interface{}) bool {
	if ex, y := e.(*Exception); y && ex.warning {
		log.Errorln(ex.msg)
		return true
	} else if e != nil {
		if DEBUG {
			buf := make([]byte, 1600)
			runtime.Stack(buf, false)
			fmt.Println(e)
			fmt.Println(string(buf))
		} else {
			if s, y := e.(string); y {
				log.Warningln(s)
			} else {
				log.Errorln(e)
			}
		}
		return true
	}
	return false
}

// TODO notify peer to slow down when queue increased too fast
func (p *multiplexer) Listen(tun *Conn, handler event_handler, interval int) {
	tun.priority = &TSPriority{0, 1e9}
	p.pool.Push(tun)
	defer p.onTunDisconnected(tun, handler)
	tun.SetSockOpt(1, 2, 1)
	var (
		header = make([]byte, FRAME_HEADER_LEN)
		router = p.router
		idle   = NewIdler(interval, p.isClient)
		nr     int
		er     error
		frm    *frame
		key    string
	)
	if !p.isClient {
		// first ping frame will let client to be aware of using a valid token.
		idle.ping(tun)
	}
	for {
		idle.newRound(tun)
		nr, er = io.ReadFull(tun, header)
		if nr == FRAME_HEADER_LEN {
			frm = _parseFrameHeader(header)
			if frm.length > 0 {
				nr, er = io.ReadFull(tun, frm.data)
			}
		}
		if er != nil {
			switch idle.consumeError(er) {
			case ERR_NEW_PING:
				if idle.ping(tun) == nil {
					continue
				}
			case ERR_PING_TIMEOUT:
				if log.V(2) {
					log.Warningln("Peer was unresponsive then close", tun.identifier)
				}
			default:
				if log.V(2) {
					log.Warningln("Read tunnel", tun.identifier, er)
				}
			}
			return // error, abandon tunnel
		}
		key = sessionKey(tun, frm.sid)

		switch frm.action {
		case FRAME_ACTION_CLOSE_W:
			if edge, _ := router.getRegistered(key); edge != nil {
				edge.bitwiseCompareAndSet(TCP_CLOSE_W)
				edge.deliver(frm)
			}
		case FRAME_ACTION_CLOSE_R:
			if edge, _ := router.getRegistered(key); edge != nil {
				edge.bitwiseCompareAndSet(TCP_CLOSE_R)
				closeR(edge.conn)
			}
		case FRAME_ACTION_DATA:
			edge, pre := router.getRegistered(key)
			if edge != nil {
				edge.deliver(frm)
			} else if pre {
				router.preDeliver(key, frm)
			} else {
				if log.V(2) {
					log.Warningln("peer send data to an unexisted socket.", key, frm)
				}
				// trigger sending close to notice peer.
				_frame(header, FRAME_ACTION_CLOSE_R, frm.sid, nil)
				if tunWrite1(tun, header) != nil {
					return
				}
			}
		case FRAME_ACTION_OPEN:
			router.preRegister(key)
			go p.connectToDest(frm, key, tun)
		case FRAME_ACTION_OPEN_N, FRAME_ACTION_OPEN_Y:
			edge, _ := router.getRegistered(key)
			if edge != nil {
				if log.V(4) {
					if frm.action == FRAME_ACTION_OPEN_Y {
						log.Infoln(p.role, "recv OPEN_Y", frm)
					} else {
						log.Infoln(p.role, "recv OPEN_N", frm)
					}
				}
				edge.ready <- frm.action
				close(edge.ready)
			} else {
				if log.V(2) {
					log.Warningln("peer send OPENx to an unexisted socket.", key, frm)
				}
			}
		case FRAME_ACTION_PING:
			if idle.pong(tun) == nil {
				atomic.AddInt32(&p.pingCnt, 1)
			} else { // reply pong failed
				return
			}
//.........这里部分代码省略.........

// TODO notify peer to slow down when queue increased too fast
func (p *multiplexer) Listen(tun *Conn, handler event_handler, interval int) {
	tun.priority = &TSPriority{0, 1e9}
	p.pool.Push(tun)
	defer p.onTunDisconnected(tun, handler)
	tun.SetSockOpt(1, 0, 1)
	var (
		header = make([]byte, FRAME_HEADER_LEN)
		idle   = NewIdler(interval, p.isClient)
		router = p.router
		nr     int
		er     error
		frm    *frame
		key    string
	)
	if !p.isClient {
		// server first ping client
		// make client aware of using a valid token.
		idle.ping(tun)
	}
	for {
		idle.newRound(tun)
		nr, er = io.ReadFull(tun, header)
		if nr == FRAME_HEADER_LEN {
			frm, er = parse_frame(header)
			if er == nil && len(frm.data) > 0 {
				// read All and discard tail random
				nr, er = io.ReadFull(tun, frm.data)
				frm.data = frm.data[:frm.length]
			}
		}
		if er != nil {
			// shutdown
			if atomic.LoadInt32(&p.status) < 0 {
				time.Sleep(time.Second)
				return
			}
			switch idle.consumeError(er) {
			case ERR_NEW_PING:
				if idle.ping(tun) == nil {
					continue
				}
			case ERR_PING_TIMEOUT:
				if log.V(1) {
					log.Errorln("Peer was unresponsive then close", tun.identifier)
				}
			default:
				if log.V(1) {
					log.Errorln("Error", tun.identifier, er)
				}
			}
			// abandon this connection
			return
		}
		key = sessionKey(tun, frm.sid)

		switch frm.action {
		case FRAME_ACTION_CLOSE_W:
			if edge, _ := router.getRegistered(key); edge != nil {
				edge.bitwiseCompareAndSet(TCP_CLOSE_W)
				edge.deliver(frm)
			}
		case FRAME_ACTION_CLOSE_R:
			if edge, _ := router.getRegistered(key); edge != nil {
				edge.bitwiseCompareAndSet(TCP_CLOSE_R)
				closeR(edge.conn)
			}
		case FRAME_ACTION_DATA:
			edge, pre := router.getRegistered(key)
			if edge != nil {
				edge.deliver(frm)
			} else if pre {
				router.preDeliver(key, frm)
			} else {
				if log.V(2) {
					log.Warningln("peer send data to an unexisted socket.", key, frm)
				}
				// trigger sending close to notice peer.
				pack(header, FRAME_ACTION_CLOSE_R, frm.sid, nil)
				if frameWriteBuffer(tun, header) != nil {
					return
				}
			}
		case FRAME_ACTION_OPEN:
			router.preRegister(key)
			go p.connectToDest(frm, key, tun)
		case FRAME_ACTION_OPEN_N, FRAME_ACTION_OPEN_Y, FRAME_ACTION_OPEN_DENIED:
			edge, _ := router.getRegistered(key)
			if edge != nil {
				if log.V(4) {
					log.Infoln(p.role, "recv OPEN_x", frm)
				}
				edge.ready <- frm.action
				close(edge.ready)
			} else {
				if log.V(2) {
					log.Warningln("peer send OPEN_x to an unexisted socket.", key, frm)
				}
			}
		case FRAME_ACTION_PING:
//.........这里部分代码省略.........