Make the czar protocol supports 65536 streams

Author: mohanson

README.md

@@ -89,7 +89,7 @@ $ daze client ... -p baboon
 
 **Czar**
 
-Protocol czar is an implementation of the ashe protocol based on TCP multiplexing. Multiplexing involves reusing a single TCP connection for multiple ashe protocols, which saves time on the TCP three-way handshake. However, this may result in a slight decrease in data transfer rate (approximately 0.19%). In most cases, using Protocol czar provides a better user experience compared to using the ashe protocol directly.
+Protocol czar is an implementation of the ashe protocol based on TCP multiplexing. Multiplexing involves reusing a single TCP connection for multiple ashe protocols, which saves time on the TCP three-way handshake. However, this may result in a slight decrease in data transfer rate (approximately 0.29%). In most cases, using Protocol czar provides a better user experience compared to using the ashe protocol directly.
 
 ```sh
 $ daze server ... -p czar

protocol/czar/engine.go

@@ -27,25 +27,25 @@ import (
 // To open a stream:
 //
 // +-----+-----+-----+-----+
-// |  0  | Sid |    Rsv    |
+// | 0x0 | Rsv |    Sid    |
 // +-----+-----+-----+-----+
 //
 // Both server and client can push data to each other.
 //
-// +-----+-----+-----+-----+-----+-----+
-// |  1  | Sid |    Len    |    Msg    |
-// +-----+-----+-----+-----+-----+-----+
+// +-----+-----+-----+-----+-----+-----+-----+-----+
+// | 0x1 | Rsv |    Sid    |    Len    |    Msg    |
+// +-----+-----+-----+-----+-----+-----+-----+-----+
 //
 // Close the specified stream.
 //
 // +-----+-----+-----+-----+
-// |  2  | Sid | 0/1 | Rsv |
+// | 0x2 | 0/1 |    Sid    |
 // +-----+-----+-----+-----+
 //
 // Keep alive probe and reply.
 //
 // +-----+-----+-----+-----+
-// |  3  | 0/1 |    Rsv    |
+// | 0x3 | 0/1 |    Rsv    |
 // +-----+-----+-----+-----+
 
 // Conf is acting as package level configuration.
@@ -54,14 +54,16 @@ var Conf = struct {
 	IdleProbeDuration time.Duration
 	// If no data is read for more than this time, the connection is closed.
 	IdleReplyDuration time.Duration
-	// Packet size. Since the size of the packet header is 4, this value must be greater than 4. If the value is too
+	// Packet size. Since the size of the packet header is 6, this value must be greater than 6. If the value is too
 	// small, the transmission efficiency will be reduced, and if it is too large, the concurrency capability of mux
 	// will be reduced.
 	PacketSize int
+	StreamPool int
 }{
 	IdleProbeDuration: time.Second * 32,
 	IdleReplyDuration: time.Second * 48,
 	PacketSize:        2048,
+	StreamPool:        256,
 }
 
 // Server implemented the czar protocol.
@@ -163,7 +165,7 @@ func (c *Client) Dial(ctx *daze.Context, network string, address string) (io.Rea
 		if err != nil {
 			return nil, err
 		}
-		log.Printf("czar: mux slot stream id=0x%02x", srv.idx)
+		log.Printf("czar: mux slot stream id=0x%04x", srv.idx)
 		spy := &ashe.Client{Cipher: c.Cipher}
 		con, err := spy.Estab(ctx, srv, network, address)
 		if err != nil {

protocol/czar/mux.go

@@ -12,7 +12,7 @@ import (
 
 // A Stream managed by the multiplexer.
 type Stream struct {
-	idx uint8
+	idx uint16
 	mux *Mux
 	rbf []byte
 	rch chan []byte
@@ -28,7 +28,9 @@ func (s *Stream) Close() error {
 	s.wer.Put(io.ErrClosedPipe)
 	s.zo0.Do(func() {
 		s.mux.pri.Pri(0, func() error {
-			s.mux.con.Write([]byte{0x02, s.idx, 0x00, 0x00})
+			buf := [4]byte{0x02, 0x00}
+			binary.BigEndian.PutUint16(buf[2:4], s.idx)
+			s.mux.con.Write(buf[:])
 			return nil
 		})
 	})
@@ -41,7 +43,9 @@ func (s *Stream) Esolc() error {
 	s.wer.Put(io.ErrClosedPipe)
 	s.zo0.Do(func() {
 		s.mux.pri.Pri(0, func() error {
-			s.mux.con.Write([]byte{0x02, s.idx, 0x01, 0x00})
+			buf := [4]byte{0x02, 0x01}
+			binary.BigEndian.PutUint16(buf[2:4], s.idx)
+			s.mux.con.Write(buf[:])
 			return nil
 		})
 	})
@@ -89,19 +93,20 @@ func (s *Stream) Write(p []byte) (int, error) {
 	)
 	for {
 		switch {
-		case len(p) >= Conf.PacketSize-4:
+		case len(p) >= Conf.PacketSize-6:
 			buf = make([]byte, Conf.PacketSize)
-			l = Conf.PacketSize - 4
+			l = Conf.PacketSize - 6
 		case len(p) >= 1:
-			buf = make([]byte, 4+len(p))
+			buf = make([]byte, 6+len(p))
 			l = len(p)
 		case len(p) >= 0:
 			return n, nil
 		}
 		buf[0] = 0x01
-		buf[1] = s.idx
-		binary.BigEndian.PutUint16(buf[2:4], uint16(l))
-		copy(buf[4:], p[:l])
+		buf[1] = 0x00
+		binary.BigEndian.PutUint16(buf[2:4], s.idx)
+		binary.BigEndian.PutUint16(buf[4:6], uint16(l))
+		copy(buf[6:], p[:l])
 		p = p[l:]
 		err := s.mux.pri.Pri(1, func() error {
 			if err := s.wer.Get(); err != nil {
@@ -122,7 +127,7 @@ func (s *Stream) Write(p []byte) (int, error) {
 }
 
 // NewStream returns a new Stream.
-func NewStream(idx uint8, mux *Mux) *Stream {
+func NewStream(idx uint16, mux *Mux) *Stream {
 	return &Stream{
 		idx: idx,
 		mux: mux,
@@ -136,7 +141,7 @@ func NewStream(idx uint8, mux *Mux) *Stream {
 }
 
 // NewWither returns a new Stream. Stream has been automatically closed, used as a placeholder.
-func NewWither(idx uint8, mux *Mux) *Stream {
+func NewWither(idx uint16, mux *Mux) *Stream {
 	stm := NewStream(idx, mux)
 	stm.zo0.Do(func() {})
 	stm.zo1.Do(func() {})
@@ -169,15 +174,17 @@ func (m *Mux) Close() error {
 func (m *Mux) Open() (*Stream, error) {
 	var (
 		err error
-		idx uint8
+		idx uint16
 		stm *Stream
 	)
 	idx, err = m.idp.Get()
 	if err != nil {
 		return nil, err
 	}
 	err = m.pri.Pri(0, func() error {
-		return doa.Err(m.con.Write([]byte{0x00, idx, 0x00, 0x00}))
+		buf := [4]byte{0x00, 0x00}
+		binary.BigEndian.PutUint16(buf[2:4], idx)
+		return doa.Err(m.con.Write(buf[:]))
 	})
 	if err != nil {
 		m.idp.Put(idx)
@@ -194,8 +201,9 @@ func (m *Mux) Recv() {
 		bsz uint16
 		buf = make([]byte, 4)
 		cmd uint8
+		cme uint8
 		err error
-		idx uint8
+		idx uint16
 		msg []byte
 		old *Stream
 		prb = time.AfterFunc(Conf.IdleProbeDuration, func() {
@@ -219,7 +227,8 @@ func (m *Mux) Recv() {
 			break
 		}
 		cmd = buf[0]
-		idx = buf[1]
+		cme = buf[1]
+		idx = binary.BigEndian.Uint16(buf[2:4])
 		if cmd >= 0x04 {
 			// Packet format error, connection closed.
 			m.con.Close()
@@ -238,7 +247,12 @@ func (m *Mux) Recv() {
 			m.usb[idx] = stm
 			m.ach <- stm
 		case 0x01:
-			bsz = binary.BigEndian.Uint16(buf[2:4])
+			_, err = io.ReadFull(m.con, buf[0:2])
+			if err != nil {
+				m.con.Close()
+				break
+			}
+			bsz = binary.BigEndian.Uint16(buf[0:2])
 			msg = make([]byte, bsz)
 			_, err = io.ReadFull(m.con, msg)
 			if err != nil {
@@ -259,7 +273,7 @@ func (m *Mux) Recv() {
 			old = NewWither(idx, m)
 			m.usb[idx] = old
 		case 0x03:
-			switch idx {
+			switch cme {
 			case 0x00:
 				m.pri.Pri(0, func() error {
 					return doa.Err(m.con.Write([]byte{0x03, 0x01, 0x00, 0x00}))
@@ -279,16 +293,16 @@ func NewMux(conn io.ReadWriteCloser) *Mux {
 		idp: NewSip(),
 		pri: priority.NewPriority(2),
 		rer: NewErr(),
-		usb: make([]*Stream, 256),
+		usb: make([]*Stream, Conf.StreamPool),
 	}
 	return mux
 }
 
 // NewMuxServer returns a new MuxServer.
 func NewMuxServer(conn io.ReadWriteCloser) *Mux {
 	mux := NewMux(conn)
-	for i := range 256 {
-		mux.usb[i] = NewWither(uint8(i), mux)
+	for i := range Conf.StreamPool {
+		mux.usb[i] = NewWither(uint16(i), mux)
 	}
 	go mux.Recv()
 	return mux

protocol/czar/sip.go

@@ -16,28 +16,28 @@ type Sip struct {
 }
 
 // Get selects an stream id from the pool, removes it from the pool, and returns it to the caller.
-func (s *Sip) Get() (uint8, error) {
+func (s *Sip) Get() (uint16, error) {
 	s.m.Lock()
 	defer s.m.Unlock()
 	n := big.NewInt(0).Not(s.i)
 	m := n.TrailingZeroBits()
-	if m == 256 {
+	if m == uint(Conf.StreamPool) {
 		return 0, errors.New("daze: out of stream")
 	}
 	s.i.SetBit(s.i, int(m), 1)
-	return uint8(m), nil
+	return uint16(m), nil
 }
 
 // Put adds x to the pool.
-func (s *Sip) Put(x uint8) {
+func (s *Sip) Put(x uint16) {
 	s.m.Lock()
 	defer s.m.Unlock()
 	doa.Doa(s.i.Bit(int(x)) == 1)
 	s.i = s.i.SetBit(s.i, int(x), 0)
 }
 
 // Set removes x from the pool.
-func (s *Sip) Set(x uint8) {
+func (s *Sip) Set(x uint16) {
 	s.m.Lock()
 	defer s.m.Unlock()
 	doa.Doa(s.i.Bit(int(x)) == 0)

protocol/czar/sip_test.go

@@ -8,8 +8,8 @@ import (
 
 func TestProtocolCzarSip(t *testing.T) {
 	sid := NewSip()
-	for i := range 256 {
-		doa.Doa(doa.Try(sid.Get()) == uint8(i))
+	for i := range Conf.StreamPool {
+		doa.Doa(doa.Try(sid.Get()) == uint16(i))
 	}
 	doa.Doa(doa.Err(sid.Get()) != nil)
 	sid.Put(65)