Function testConcurrentQuery

scope_test.go:428–459

(c *cluster, u *user, cu *clusterUser, concurrency int, expectedSessionHostMap map[string]string)

}

func testConcurrentQuery(c *cluster, u *user, cu *clusterUser, concurrency int, expectedSessionHostMap map[string]string) error {
	ch := make(chan map[string]string, 10000)

	f := func(sessionId string) map[string]string {
		s := testGetScope(c, u, cu, sessionId)
		// user set sessionId
		s.sessionId = sessionId
		// concurrent task wait
		s.incQueued()
		time.Sleep(50 * time.Millisecond)
		s.dec()
		// return wake task's new host
		// same sessionId should get same host addr
		return map[string]string{sessionId: s.host.Host()}
	}

	for i := 0; i < concurrency; i++ {
		sessionId := strconv.Itoa(i % 4)
		go func() {
			ch <- f(sessionId)
		}()
	}
	for i := 0; i < concurrency; i++ {
		sessionHost := <-ch
		for sessionId, host := range sessionHost {
			if expectedSessionHostMap[sessionId] != host {
				return fmt.Errorf("incQueue waked task sessionId: %s,expected host: %v, get: %v", sessionId, expectedSessionHostMap[sessionId], host)
			}
		}
	}
	return nil
}

func testGetCluster() *cluster {
	c := &cluster{