func (t *Session) DataTunServe(fconn *Conn, isNewSession bool) {
	atomic.AddInt32(&t.activeCnt, 1)
	defer func() {
		var (
			offline bool
			err     = recover()
		)
		if atomic.AddInt32(&t.activeCnt, -1) <= 0 {
			offline = true
			t.mgr.clearTokens(t)
			t.mux.destroy()
		}
		if log.V(1) {
			log.Infof("Tun=%s was disconnected. %v\n", fconn.identifier, nvl(err, NULL))
			if offline {
				log.Infof("Client=%s was offline\n", t.cid)
			}
		}
		if DEBUG {
			ex.CatchException(err)
		}
	}()
	if isNewSession {
		log.Infof("Client=%s is online\n", t.cid)
	}

	if log.V(1) {
		log.Infof("Tun=%s is established\n", fconn.identifier)
	}
	t.mux.Listen(fconn, t.eventHandler, DT_PING_INTERVAL)
}

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 (t *Session) DataTunServe(fconn *Conn, buf []byte) {
	defer func() {
		var offline bool
		if atomic.AddInt32(&t.activeCnt, -1) <= 0 {
			offline = true
			t.mgr.clearTokens(t)
			t.mux.destroy()
		}
		var err = recover()
		if log.V(1) {
			log.Infof("Tun=%s was disconnected. %v\n", fconn.identifier, nvl(err, NULL))
			if offline {
				log.Infof("Client=%s was offline\n", t.cid)
			}
		}
		if DEBUG {
			ex.CatchException(err)
		}
	}()
	atomic.AddInt32(&t.activeCnt, 1)

	if buf != nil {
		token := buf[:TKSZ]
		fconn.cipher = t.cipherFactory.NewCipher(token)
		buf = nil
	} else { // first negotiation had initialized cipher, the buf will be null
		log.Infof("Client=%s is online\n", t.cid)
	}

	if log.V(1) {
		log.Infof("Tun=%s is established\n", fconn.identifier)
	}
	t.mux.Listen(fconn, t.eventHandler, DT_PING_INTERVAL)
}

func acquire(s *semaphore, id int, t *testing.T) {
	if s.acquire(tmo) {
		atomic.AddInt32(&cnt1, 1)
		if log.V(3) {
			t.Log("\tacquired", id)
		}
	} else {
		atomic.AddInt32(&cnt2, 1)
		if log.V(3) {
			t.Log("\tacquired timeout", id)
		}
	}
}

func (p *multiplexer) connectToDest(frm *frame, key string, tun *Conn) {
	defer func() {
		ex.CatchException(recover())
	}()
	var (
		dstConn net.Conn
		err     error
		target  = string(frm.data)
	)
	dstConn, err = net.DialTimeout("tcp", target, GENERAL_SO_TIMEOUT)
	frm.length = 0
	if err != nil {
		p.router.removePreRegistered(key)
		log.Errorf("Cannot connect to [%s] for %s error: %s\n", target, key, err)
		frm.action = FRAME_ACTION_OPEN_N
		tunWrite2(tun, frm)
	} else {
		if log.V(1) {
			log.Infoln("OPEN", target, "for", key)
		}
		dstConn.SetReadDeadline(ZERO_TIME)
		edge := p.router.register(key, target, tun, dstConn, false) // write edge
		frm.action = FRAME_ACTION_OPEN_Y
		if tunWrite2(tun, frm) == nil {
			p.relay(edge, tun, frm.sid) // read edge
		} else { // send open_y failed
			SafeClose(tun)
		}
	}
}

// close for ending of queued task
func (q *equeue) _close(force bool, close_code uint) {
	q.lock.Lock()
	defer q.lock.Unlock()
	e := q.edge
	if log.V(4) {
		switch close_code {
		case CLOSED_BY_ERR:
			log.Infoln("terminate", e.dest)
		case CLOSED_FORCE:
			log.Infoln("close", e.dest)
		case CLOSED_WRITE:
			log.Infof("closeW %s by peer\n", e.dest)
		}
	}

	for i, e := q.buffer.Len(), q.buffer.Front(); i > 0; i, e = i-1, e.Next() {
		f := e.Value.(*frame)
		if f != nil {
			f.free()
		}
	}

	q.buffer = nil
	if force {
		atomic.StoreUint32(&e.closed, TCP_CLOSED)
		SafeClose(e.conn)
	} else {
		closeW(e.conn)
	}
}

func (nego *d5SNegotiation) verifyThenDHExchange(conn net.Conn, credBuf []byte) (key []byte) {
	userIdentity, err := RSADecrypt(credBuf, nego.RSAKeys.priv)
	ThrowErr(err)
	clientIdentity := string(userIdentity)
	if log.V(2) {
		log.Infoln("Auth clientIdentity:", SubstringBefore(clientIdentity, IDENTITY_SEP), "***")
	}
	allow, ex := nego.AuthSys.Authenticate(userIdentity)
	cDHPub, err := ReadFullByLen(2, conn)
	if !allow { // invalid user indentity
		log.Warningf("Auth %s failed: %v\n", clientIdentity, ex)
		conn.Write([]byte{0, 1, 0xff})
		panic(ex)
	}
	nego.clientIdentity = clientIdentity
	key = takeSharedKey(nego.dhKeys, cDHPub)
	//	if log.V(5) {
	//		dumpHex("Sharedkey", key)
	//	}
	buf := new(bytes.Buffer)
	buf.Write(nego.dhKeys.pubLen)
	buf.Write(nego.dhKeys.pub)
	_, err = buf.WriteTo(conn)
	return
}

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 TestSingleRequest(t *testing.T) {
	startEmulation()
	conn, e := net.Dial("tcp", cltAddr)
	ThrowErr(e)
	rest(1)
	assertLength(t, "client.registry", client.router.registry, 1)
	buf0 := make([]byte, 0xffff)
	buf1 := make([]byte, 0xffff)
	for i := 0; i < 10; i++ {
		n := randomBuffer(buf0)
		nw, e := conn.Write(buf0[:n])
		ThrowErr(e)
		nr, e := io.ReadFull(conn, buf1[:n-2])
		ThrowErr(e)
		if log.V(3) {
			fmt.Printf("\tsend=%d recv=%d\n", nw, nr)
		}
		if !bytes.Equal(buf0[2:n], buf1[:nr]) {
			t.Errorf("sent is inconsistent with recv. nw=%d nr=%d\n", nw, nr)
		}
	}
	conn.Close()
	rest(2)
	checkFinishedLength(t)
}

// return header=1 + TKSZ*many
func (s *SessionMgr) createTokens(session *Session, many int) []byte {
	s.lock.Lock()
	defer s.lock.Unlock()
	var (
		tokens  = make([]byte, 1+many*TKSZ)
		i64buf  = make([]byte, 8)
		_tokens = tokens[1:]
		sha     = sha1.New()
	)
	rand.Seed(time.Now().UnixNano())
	sha.Write([]byte(session.uid))
	for i := 0; i < many; i++ {
		binary.BigEndian.PutUint64(i64buf, uint64(rand.Int63()))
		sha.Write(i64buf)
		binary.BigEndian.PutUint64(i64buf, uint64(time.Now().UnixNano()))
		sha.Write(i64buf)
		pos := i * TKSZ
		sha.Sum(_tokens[pos:pos])
		token := _tokens[pos : pos+TKSZ]
		key := fmt.Sprintf("%x", token)
		if _, y := s.container[key]; y {
			i--
			continue
		}
		s.container[key] = session
		session.tokens[key] = true
	}
	if log.V(4) {
		log.Errorf("sessionMap created=%d len=%d\n", many, len(s.container))
	}
	return tokens
}

func sendFrame(frm *frame) bool {
	dst := frm.conn.conn
	if log.V(5) {
		log.Infoln("SEND queue", frm)
	}
	dst.SetWriteDeadline(time.Now().Add(GENERAL_SO_TIMEOUT))
	nw, ew := dst.Write(frm.data)
	if nw == int(frm.length) && ew == nil {
		return false
	}
	// an error occured
	if log.V(1) {
		log.Warningf("Write edge(%s) error(%v). %s\n", frm.conn.dest, ew, frm)
	}
	return true
}

func (s *Server) updateNow() {
	tc := calculateTimeCounter(true)
	// write atomically
	atomic.StorePointer(&s.tcPool, unsafe.Pointer(&tc))
	if log.V(4) {
		log.Infoln("updateTimeCounterThread", len(tc))
	}
}

// async request
func (c *Client) asyncRequestTokens() {
	// don't require if shutdown
	if atomic.LoadInt32(&c.state) >= CLT_WORKING {
		go c.mux.bestSend([]byte{FRAME_ACTION_TOKEN_REQUEST}, "asyncRequestTokens")
		if log.V(3) {
			log.Infof("Request new tokens, pool=%d\n", len(c.token)/TKSZ)
		}
	}
}

// async request
func (c *Client) requireTokens() {
	// non-working state can't require anything
	if atomic.CompareAndSwapInt32(&c.State, CLT_WORKING, CLT_PENDING) {
		if log.V(3) {
			log.Infof("Request new tokens, pool=%d\n", len(c.token)/TKSZ)
		}
		go c.mux.bestSend([]byte{FRAME_ACTION_TOKEN_REQUEST}, "requireTokens")
	}
}

func TestConcurrency(t *testing.T) {
	var wg sync.WaitGroup
	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func(j int) {
			defer wg.Done()
			conn, e := net.Dial("tcp", cltAddr)
			ThrowErr(e)
			if log.V(2) {
				fmt.Printf("\tthread=%d/ start\n", j)
			}
			defer conn.Close()
			buf0 := make([]byte, 0xffff)
			buf1 := make([]byte, 0xffff)
			for k := 0; k < 99; k++ {
				n := randomBuffer(buf0)
				nw, e := conn.Write(buf0[:n])
				ThrowErr(e)
				ThrowIf(nw != int(n), fmt.Sprintf("nr=%d nw=%d", n, nw))
				conn.SetReadDeadline(time.Now().Add(time.Second * 4))
				nr, e := io.ReadFull(conn, buf1[:n-2])
				if e != nil {
					if ne, y := e.(net.Error); y && ne.Timeout() {
						continue
					} else {
						ThrowErr(e)
					}
				}
				if log.V(2) {
					fmt.Printf("\tthread=%d/%d send=%d recv=%d\n", j, k, nw, nr)
				}
				if !bytes.Equal(buf0[2:n], buf1[:nr]) {
					t.Errorf("thread=%d/ sent != recv. nw=%d nr=%d\n", j, nw, nr)
				}
			}
			if log.V(2) {
				fmt.Printf("\tthread=%d/ done\n", j)
			}
		}(i)
	}
	wg.Wait()
	rest(3)
	checkFinishedLength(t)
}

func (p *multiplexer) HandleRequest(prot string, client net.Conn, target string) {
	sid := next_sid()
	if log.V(1) {
		log.Infof("%s->[%s] from=%s sid=%d\n", prot, target, ipAddr(client.RemoteAddr()), sid)
	}
	tun := p.pool.Select()
	ThrowIf(tun == nil, "No tun to deliveries request")
	key := sessionKey(tun, sid)
	edge := p.router.register(key, target, tun, client, true) // write edge
	p.relay(edge, tun, sid)                                   // read edge
}

func NewGeoIPFilter(keyword string) (f *GeoIPFilter, e error) {
	if len(keyword) != 2 {
		return nil, fmt.Errorf("filter keyword must be 2-byte country_iso_code")
	}
	f = new(GeoIPFilter)
	f.keyword = StoU16(strings.ToUpper(keyword))
	f.tab = deserialize(buildGeoDB())
	if log.V(1) {
		log.Infoln("Init GeoIPFilter with target keyword", keyword)
	}
	return
}

func (t *Session) DataTunServe(tun *Conn, isNewSession bool) {
	defer func() {
		tun.cipher.Cleanup()
		if atomic.AddInt32(&t.activeCnt, -1) <= 0 {
			t.destroy()
			log.Infof("Client %s was offline", t.cid)
		}
	}()

	if isNewSession {
		log.Infof("Client %s is online", t.cid)
	}
	if log.V(1) {
		log.Infof("Tun %s is established", tun.identifier)
	}
	cnt := atomic.AddInt32(&t.activeCnt, 1)
	// mux will output error log
	t.mux.Listen(tun, t.eventHandler, DT_PING_INTERVAL+int(cnt))
	if log.V(1) {
		log.Infof("Tun %s was disconnected", tun.identifier)
	}
}

func (h *ConnPool) Select() *Conn {
	h.lock.Lock()
	defer h.lock.Unlock()
	if h.pool.Len() < 1 {
		return nil
	}
	sort.Sort(h.pool)
	if log.V(5) {
		log.Infoln("selected tun", h.pool[0].LocalAddr())
	}
	selected := h.pool[0]
	atomic.AddInt64(&selected.priority.rank, -1)
	return selected
}

func (c *Client) saveTokens(data []byte) {
	var tokens []byte
	switch data[0] {
	case FRAME_ACTION_TOKEN_REQUEST:
		log.Warningf("unexpected token request")
		return
	case FRAME_ACTION_TOKEN_REPLY:
		tokens = data[1:]
	}
	c.lock.Lock()
	c.token = append(c.token, tokens...)
	c.lock.Unlock()
	// wakeup waiting
	c.pendingTK.notifyAll()
	if log.V(3) {
		log.Infof("Recv tokens=%d pool=%d\n", len(tokens)/TKSZ, len(c.token)/TKSZ)
	}
}