pass_chan_stress.go raw

   1  package main
   2  
   3  import "io"
   4  import "moxie"
   5  
   6  // B.0.4 stress test: sustained codec round-trip under load.
   7  // 10000 messages with mixed-size variable payloads. Exercises the
   8  // Codec serialization path repeatedly to catch GC/heap interactions
   9  // that single-shot tests miss.
  10  
  11  type Msg struct {
  12  	Seq  uint32
  13  	Sum  uint32
  14  	Body []byte
  15  }
  16  
  17  func (m *Msg) EncodeTo(w io.Writer) (encErr error) {
  18  	var hdr [8]byte
  19  	hdr[0] = byte(m.Seq)
  20  	hdr[1] = byte(m.Seq >> 8)
  21  	hdr[2] = byte(m.Seq >> 16)
  22  	hdr[3] = byte(m.Seq >> 24)
  23  	hdr[4] = byte(m.Sum)
  24  	hdr[5] = byte(m.Sum >> 8)
  25  	hdr[6] = byte(m.Sum >> 16)
  26  	hdr[7] = byte(m.Sum >> 24)
  27  	_, encErr = w.Write(hdr[:])
  28  	if encErr != nil {
  29  		return encErr
  30  	}
  31  	var sz [4]byte
  32  	l := uint32(len(m.Body))
  33  	sz[0] = byte(l)
  34  	sz[1] = byte(l >> 8)
  35  	sz[2] = byte(l >> 16)
  36  	sz[3] = byte(l >> 24)
  37  	_, encErr = w.Write(sz[:])
  38  	if encErr != nil {
  39  		return encErr
  40  	}
  41  	_, encErr = w.Write(m.Body)
  42  	return encErr
  43  }
  44  
  45  func (m *Msg) DecodeFrom(r io.Reader) (decErr error) {
  46  	var hdr [8]byte
  47  	_, decErr = io.ReadFull(r, hdr[:])
  48  	if decErr != nil {
  49  		return decErr
  50  	}
  51  	m.Seq = uint32(hdr[0]) | uint32(hdr[1])<<8 | uint32(hdr[2])<<16 | uint32(hdr[3])<<24
  52  	m.Sum = uint32(hdr[4]) | uint32(hdr[5])<<8 | uint32(hdr[6])<<16 | uint32(hdr[7])<<24
  53  	var sz [4]byte
  54  	_, decErr = io.ReadFull(r, sz[:])
  55  	if decErr != nil {
  56  		return decErr
  57  	}
  58  	l := uint32(sz[0]) | uint32(sz[1])<<8 | uint32(sz[2])<<16 | uint32(sz[3])<<24
  59  	m.Body = make([]byte, l)
  60  	_, decErr = io.ReadFull(r, m.Body)
  61  	return decErr
  62  }
  63  
  64  func mirror(in chan Msg, out chan Msg, done chan moxie.Uint32) {
  65  	count := uint32(0)
  66  	for {
  67  		m := <-in
  68  		if m.Seq == 0xFFFFFFFF {
  69  			break
  70  		}
  71  		// Recompute sum on body to verify integrity.
  72  		s := uint32(0)
  73  		for i := int32(0); i < int32(len(m.Body)); i++ {
  74  			s += uint32(m.Body[i])
  75  		}
  76  		out <- Msg{Seq: m.Seq, Sum: s, Body: m.Body}
  77  		count++
  78  	}
  79  	done <- moxie.Uint32(count)
  80  }
  81  
  82  func main() {
  83  	in := make(chan Msg)
  84  	out := make(chan Msg)
  85  	done := make(chan moxie.Uint32)
  86  	spawn(mirror, in, out, done)
  87  
  88  	const N = uint32(10000)
  89  	for i := uint32(0); i < N; i++ {
  90  		// Varying body size: 0, 1, 2, ..., 1023, 0, 1, ...
  91  		bodyLen := int32(i % 1024)
  92  		body := make([]byte, bodyLen)
  93  		expectedSum := uint32(0)
  94  		for j := int32(0); j < bodyLen; j++ {
  95  			body[j] = byte((i + uint32(j)) & 0xFF)
  96  			expectedSum += uint32(body[j])
  97  		}
  98  		in <- Msg{Seq: i, Sum: 0, Body: body}
  99  		got := <-out
 100  		if got.Seq != i || got.Sum != expectedSum {
 101  			println("FAIL:stress", "i", i, "got.Seq", got.Seq, "got.Sum", got.Sum, "want.Sum", expectedSum)
 102  			return
 103  		}
 104  		if i%1000 == 0 && i > 0 {
 105  			println("progress:", i)
 106  		}
 107  	}
 108  	in <- Msg{Seq: 0xFFFFFFFF, Sum: 0, Body: nil}
 109  	count := uint32(<-done)
 110  	if count != N {
 111  		println("FAIL:stress", "count", count, "want", N)
 112  		return
 113  	}
 114  	println("PASS:stress-codec", N, "msgs")
 115  }
 116