Function Test_CompletionCount

transport/obj_test.go:572–633

(t *testing.T)

}

func Test_CompletionCount(t *testing.T) {
	tutils.CheckSkip(t, tutils.SkipTestArgs{Long: true})
	var (
		numSent                   int64
		numCompleted, numReceived atomic.Int64
	)

	receive := func(hdr transport.ObjHdr, objReader io.Reader, err error) error {
		cos.Assert(err == nil)
		written, _ := io.Copy(io.Discard, objReader)
		cos.Assert(written == hdr.ObjAttrs.Size)
		numReceived.Inc()
		return nil
	}
	callback := func(_ transport.ObjHdr, _ io.ReadCloser, _ interface{}, _ error) {
		numCompleted.Inc()
	}

	ts := httptest.NewServer(objmux)
	defer ts.Close()

	trname := "cmpl-cnt"
	err := transport.HandleObjStream(trname, receive)
	tassert.CheckFatal(t, err)
	defer transport.Unhandle(trname)
	httpclient := transport.NewIntraDataClient()
	url := ts.URL + transport.ObjURLPath(trname)
	t.Setenv("AIS_STREAM_BURST_NUM", "256")
	stream := transport.NewObjStream(httpclient, url, cos.GenTie(), nil) // provide for sizeable queue at any point
	random := newRand(mono.NanoTime())
	rem := int64(0)
	for idx := 0; idx < 10000; idx++ {
		if idx%7 == 0 {
			hdr := genStaticHeader(random)
			hdr.ObjAttrs.Size = 0
			hdr.Opaque = []byte(strconv.FormatInt(104729*int64(idx), 10))
			stream.Send(&transport.Obj{Hdr: hdr, Callback: callback})
			rem = random.Int63() % 13
		} else {
			hdr, rr := makeRandReader(random, false)
			stream.Send(&transport.Obj{Hdr: hdr, Reader: rr, Callback: callback})
		}
		numSent++
		if numSent > 5000 && rem == 3 {
			stream.Stop()
			break
		}
	}
	// collect all pending completions until timeout
	started := time.Now()
	for numCompleted.Load() < numSent {
		time.Sleep(time.Millisecond * 10)
		if time.Since(started) > time.Second*10 {
			break
		}
	}
	if numSent == numCompleted.Load() {
		tlog.Logf("sent %d = %d completed, %d received\n", numSent, numCompleted.Load(), numReceived.Load())