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

@@ -26,42 +26,43 @@ import (
 //
 // To open a stream:
 //
-// +-----+-----+-----+-----+
-// | 0x0 | Sid |    Rsv    |
-// +-----+-----+-----+-----+
+// +-----+-----+-----+-----+-----+-----+-----+-----+
+// | 0x0 | Rsv |    Sid    |          Rsv          |
+// +-----+-----+-----+-----+-----+-----+-----+-----+
 //
 // Both server and client can push data to each other.
 //
-// +-----+-----+-----+-----+-----+-----+
-// | 0x1 | Sid |    Len    |    Msg    |
-// +-----+-----+-----+-----+-----+-----+
+// +-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+
+// | 0x1 | Rsv |    Sid    |    Len    |    Rsv    |    Msg    |
+// +-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+
 //
 // Close the specified stream.
 //
-// +-----+-----+-----+-----+
-// | 0x2 | Sid | 0/1 | Rsv |
-// +-----+-----+-----+-----+
+// +-----+-----+-----+-----+-----+-----+-----+-----+
+// | 0x2 | 0/1 |    Sid    |          Rsv          |
+// +-----+-----+-----+-----+-----+-----+-----+-----+
 //
 // Keep alive probe and reply.
 //
-// +-----+-----+-----+-----+
-// | 0x3 | 0/1 |    Rsv    |
-// +-----+-----+-----+-----+
+// +-----+-----+-----+-----+-----+-----+-----+-----+
+// | 0x3 | 0/1 |                Rsv                |
+// +-----+-----+-----+-----+-----+-----+-----+-----+
 
 // Conf is acting as package level configuration.
 var Conf = struct {
 	// The duration a connection needs to be idle before mux begins sending out keep-alive probe.
 	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
-	// small, the transmission efficiency will be reduced, and if it is too large, the concurrency capability of mux
-	// will be reduced.
+	// Packet size. The 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 +164,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,10 +28,10 @@ func (s *Stream) Close() error {
 	s.wer.Put(io.ErrClosedPipe)
 	s.zo0.Do(func() {
 		s.mux.pri.Pri(0, func() error {
-			mph := make([]byte, 4)
+			mph := make([]byte, 8)
 			mph[0] = 0x02
-			mph[1] = s.idx
-			mph[2] = 0x00
+			mph[1] = 0x00
+			binary.BigEndian.PutUint16(mph[2:4], s.idx)
 			s.mux.con.Write(mph)
 			return nil
 		})
@@ -45,10 +45,10 @@ func (s *Stream) Esolc() error {
 	s.wer.Put(io.ErrClosedPipe)
 	s.zo0.Do(func() {
 		s.mux.pri.Pri(0, func() error {
-			mph := make([]byte, 4)
+			mph := make([]byte, 8)
 			mph[0] = 0x02
-			mph[1] = s.idx
-			mph[2] = 0x01
+			mph[1] = 0x01
+			binary.BigEndian.PutUint16(mph[2:4], s.idx)
 			s.mux.con.Write(mph)
 			return nil
 		})
@@ -97,19 +97,20 @@ func (s *Stream) Write(p []byte) (int, error) {
 	)
 	for {
 		switch {
-		case len(p) >= Conf.PacketSize-4:
+		case len(p) >= Conf.PacketSize-8:
 			buf = make([]byte, Conf.PacketSize)
-			l = Conf.PacketSize - 4
+			l = Conf.PacketSize - 8
 		case len(p) >= 1:
-			buf = make([]byte, 4+len(p))
+			buf = make([]byte, 8+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[8:], p[:l])
 		p = p[l:]
 		err := s.mux.pri.Pri(1, func() error {
 			if err := s.wer.Get(); err != nil {
@@ -130,7 +131,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,
@@ -144,7 +145,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() {})
@@ -177,17 +178,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 {
-		mph := make([]byte, 4)
+		mph := make([]byte, 8)
 		mph[0] = 0x00
-		mph[1] = idx
+		binary.BigEndian.PutUint16(mph[2:4], idx)
 		return doa.Err(m.con.Write(mph))
 	})
 	if err != nil {
@@ -203,15 +204,16 @@ func (m *Mux) Open() (*Stream, error) {
 func (m *Mux) Recv() {
 	var (
 		bsz uint16
-		buf = make([]byte, 4)
+		buf = make([]byte, 8)
 		cmd uint8
+		cmz uint8
 		err error
-		idx uint8
+		idx uint16
 		msg []byte
 		old *Stream
 		prb = time.AfterFunc(Conf.IdleProbeDuration, func() {
 			if m.pri.Pri(0, func() error {
-				mph := make([]byte, 4)
+				mph := make([]byte, 8)
 				mph[0] = 0x03
 				mph[1] = 0x00
 				return doa.Err(m.con.Write(mph))
@@ -227,13 +229,14 @@ func (m *Mux) Recv() {
 	for {
 		prb.Reset(Conf.IdleProbeDuration)
 		rst.Reset(Conf.IdleReplyDuration)
-		_, err = io.ReadFull(m.con, buf[:4])
+		_, err = io.ReadFull(m.con, buf)
 		if err != nil {
 			m.rer.Put(err)
 			break
 		}
 		cmd = buf[0]
-		idx = buf[1]
+		cmz = buf[1]
+		idx = binary.BigEndian.Uint16(buf[2:4])
 		if cmd >= 0x04 {
 			// Packet format error, connection closed.
 			m.con.Close()
@@ -252,7 +255,7 @@ func (m *Mux) Recv() {
 			m.usb[idx] = stm
 			m.ach <- stm
 		case 0x01:
-			bsz = binary.BigEndian.Uint16(buf[2:4])
+			bsz = binary.BigEndian.Uint16(buf[4:6])
 			msg = make([]byte, bsz)
 			_, err = io.ReadFull(m.con, msg)
 			if err != nil {
@@ -273,10 +276,10 @@ func (m *Mux) Recv() {
 			old = NewWither(idx, m)
 			m.usb[idx] = old
 		case 0x03:
-			switch idx {
+			switch cmz {
 			case 0x00:
 				m.pri.Pri(0, func() error {
-					mph := make([]byte, 4)
+					mph := make([]byte, 8)
 					mph[0] = 0x03
 					mph[1] = 0x01
 					return doa.Err(m.con.Write(mph))
@@ -296,16 +299,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/mux_test.go

@@ -145,9 +145,9 @@ func TestProtocolCzarMuxServerReopen(t *testing.T) {
 	cli := doa.Try(net.Dial("tcp", DazeTesterListenOn))
 	defer cli.Close()
 
-	mph := make([]byte, 4)
-	mph[0] = 0x00 // Cmd open stream
-	mph[1] = 0x00 // Sid
+	mph := make([]byte, 8)
+	mph[0] = 0x00                              // Cmd open stream
+	binary.BigEndian.PutUint16(mph[2:4], 0x00) // Sid
 	cli.Write(mph)
 	cli.Write(mph)
 	buf := make([]byte, 1)

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)