use concurrent map to solve fatal error: concurrent map writes #8

Author: oldthreefeng

cmd/dispose.go

@@ -24,7 +24,7 @@ func init() {
 }
 
 // Dispose is 处理函数
-func Dispose(concurrency, totalNumber uint64, request *pkg.Request) {
+func Dispose(concurrency int, totalNumber uint64, request *pkg.Request) {
 
 	// 设置接收数据缓存
 	ch := make(chan *pkg.RequestResults, 1000)
@@ -36,7 +36,7 @@ func Dispose(concurrency, totalNumber uint64, request *pkg.Request) {
 	wgReceiving.Add(1)
 	go pkg.ReceivingResults(concurrency, ch, &wgReceiving)
 
-	for i := uint64(0); i < concurrency; i++ {
+	for i := 0; i < concurrency; i++ {
 		wg.Add(1)
 		switch request.Form {
 		case pkg.FormTypeHttp:
@@ -59,7 +59,7 @@ func Dispose(concurrency, totalNumber uint64, request *pkg.Request) {
 				go pkg.WebSocket(i, ch, totalNumber, &wg, request, ws)
 			case 2:
 				// 并发建立长链接
-				go func(i uint64) {
+				go func(i int) {
 					// 连接以后再启动协程
 					ws := pkg.NewWebSocket(request.Url)
 					err := ws.GetConn()

cmd/root.go

@@ -98,7 +98,7 @@ func init() {
 	// will be global for your application.
 
 	rootCmd.PersistentFlags().StringVar(&cfgFile, "config", "", "config file for stress (default is $HOME/.stress.yaml)")
-	rootCmd.PersistentFlags().Uint64VarP(&pkg.Concurrency, "concurrency", "c", 1, "并发数")
+	rootCmd.PersistentFlags().IntVarP(&pkg.Concurrency, "concurrency", "c", 1, "并发数")
 	rootCmd.PersistentFlags().Uint64VarP(&pkg.Number, "number", "n", 1, "单协程的请求数")
 	rootCmd.PersistentFlags().StringVarP(&pkg.RequestUrl, "requestUrl", "u", "", "curl文件路径")
 	rootCmd.PersistentFlags().StringVarP(&pkg.Path, "path", "f", "", "read curl file to build test")

pkg/http.go

@@ -8,7 +8,7 @@ import (
 )
 
 // Http is http go link
-func Http(chanId uint64, ch chan<- *RequestResults, totalNumber uint64, wg *sync.WaitGroup, request *Request) {
+func Http(chanId int, ch chan<- *RequestResults, totalNumber uint64, wg *sync.WaitGroup, request *Request) {
 
 	defer func() {
 		wg.Done()

pkg/http_client.go

@@ -4,7 +4,6 @@ import (
 	"crypto/tls"
 	"fmt"
 	"github.com/oldthreefeng/stress/utils"
-	"github.com/orcaman/concurrent-map"
 	"io"
 	"net/http"
 	"time"
@@ -16,7 +15,7 @@ import (
 // body 请求的body
 // headers 请求头信息
 // timeout 请求超时时间
-func HttpRequest(method, url string, body io.Reader, headers cmap.ConcurrentMap, timeout time.Duration) (resp *http.Response, requestTime uint64, err error) {
+func HttpRequest(method, url string, body io.Reader, headers utils.ConcurrentMap, timeout time.Duration) (resp *http.Response, requestTime uint64, err error) {
 
 	// 跳过证书验证
 	tr := &http.Transport{
@@ -43,7 +42,7 @@ func HttpRequest(method, url string, body io.Reader, headers cmap.ConcurrentMap,
 
 	if _, ok := headers.Get("Content-Type"); !ok {
 		if headers == nil {
-			headers = cmap.New()
+			headers = utils.New()
 		}
 		headers.Set("Content-Type","application/x-www-form-urlencoded; charset=utf-8")
 	}

pkg/request.go

@@ -3,7 +3,7 @@ package pkg
 import (
 	"errors"
 	"fmt"
-	"github.com/orcaman/concurrent-map"
+	"github.com/oldthreefeng/stress/utils"
 	"io"
 	"net/http"
 	"strings"
@@ -68,7 +68,7 @@ type Request struct {
 	Url     string              // Url
 	Form    string              // http/webSocket/tcp
 	Method  string              // 方法 GET/POST/PUT
-	Headers cmap.ConcurrentMap // Headers
+	Headers utils.ConcurrentMap // Headers
 	Body    string              // body
 	Verify  string              // 验证的方法
 	Timeout time.Duration       // 请求超时时间
@@ -121,7 +121,7 @@ func NewRequest(url string, verify string, timeout time.Duration, debug bool, re
 
 	var (
 		method  = DefaultMethod
-		headers = cmap.New()
+		headers = utils.New()
 		body    string
 	)
 
@@ -213,7 +213,7 @@ func NewDefaultRequest() *Request {
 	}
 }
 
-func getHeaderValue(v string, headers cmap.ConcurrentMap) {
+func getHeaderValue(v string, headers utils.ConcurrentMap) {
 	index := strings.Index(v, ":")
 	if index < 0 {
 		return
@@ -273,14 +273,14 @@ func (r *Request) IsParameterLegal() (err error) {
 // RequestResults is  请求结果
 type RequestResults struct {
 	Id        string // 消息Id
-	ChanId    uint64 // 消息Id
+	ChanId    int // 消息Id
 	Time      uint64 // 请求时间 纳秒
 	IsSucceed bool   // 是否请求成功
 	ErrCode   int    // 错误码
 }
 
 // SetId is set chanId & id to request results
-func (r *RequestResults) SetId(chanId uint64, number uint64) {
+func (r *RequestResults) SetId(chanId int, number uint64) {
 	id := fmt.Sprintf("%d_%d", chanId, number)
 
 	r.Id = id

pkg/statistics.go renamed to pkg/satics.go

@@ -2,6 +2,7 @@ package pkg
 
 import (
 	"fmt"
+	"github.com/oldthreefeng/stress/utils"
 	"strings"
 	"sync"
 	"time"
@@ -15,7 +16,7 @@ var (
 // ReceivingResults is 接收结果并处理
 // 统计的时间都是纳秒，显示的时间 都是毫秒
 // concurrent 并发数
-func ReceivingResults(concurrent uint64, ch <-chan *RequestResults, wg *sync.WaitGroup) {
+func ReceivingResults(concurrent int, ch <-chan *RequestResults, wg *sync.WaitGroup) {
 
 	defer func() {
 		wg.Done()
@@ -34,13 +35,13 @@ func ReceivingResults(concurrent uint64, ch <-chan *RequestResults, wg *sync.Wai
 		successNum     uint64 // 成功处理数，code为0
 		failureNum     uint64 // 处理失败数，code不为0
 		chanIdLen      int    // 并发数
-		chanIds        = make(map[uint64]bool)
+		chanIds        = utils.New(Concurrency)
 	)
 
 	statTime := uint64(time.Now().UnixNano())
 
 	// 错误码/错误个数
-	var errCode = make(map[int]int)
+	var errCode = utils.New()
 
 	// 定时输出一次计算结果
 	ticker := time.NewTicker(exportStatisticsTime)
@@ -83,16 +84,18 @@ func ReceivingResults(concurrent uint64, ch <-chan *RequestResults, wg *sync.Wai
 		}
 
 		// 统计错误码
-		if value, ok := errCode[data.ErrCode]; ok {
-			errCode[data.ErrCode] = value + 1
+		strErrCode := fmt.Sprintf("%d", data.ErrCode)
+		if value, ok := errCode.Get(strErrCode); ok {
+			errCode.Set(strErrCode, value.(int)+1)
 		} else {
-			errCode[data.ErrCode] = 1
+			errCode.Set(strErrCode, 1)
 		}
-
-		if _, ok := chanIds[data.ChanId]; !ok {
-			chanIds[data.ChanId] = true
+		strChanId := fmt.Sprintf("%d", data.ChanId)
+		if _, ok := chanIds.Get(strChanId); !ok {
+			chanIds.Set(strChanId, true)
 			chanIdLen = len(chanIds)
 		}
+
 	}
 
 	// 数据全部接受完成，停止定时输出统计数据
@@ -117,7 +120,7 @@ func ReceivingResults(concurrent uint64, ch <-chan *RequestResults, wg *sync.Wai
 }
 
 // 计算数据
-func calculateData(concurrent, processingTime, requestTime, maxTime, minTime, successNum, failureNum uint64, chanIdLen int, errCode map[int]int) {
+func calculateData(concurrent int, processingTime, requestTime, maxTime, minTime, successNum, failureNum uint64, chanIdLen int, errCode utils.ConcurrentMap) {
 	if processingTime == 0 {
 		processingTime = 1
 	}
@@ -132,12 +135,12 @@ func calculateData(concurrent, processingTime, requestTime, maxTime, minTime, su
 
 	// 平均 每个协程成功数*总协程数据/总耗时 (每秒)
 	if processingTime != 0 {
-		qps = float64(successNum*1e9*concurrent) / float64(processingTime)
+		qps = float64(successNum*1e9*uint64(concurrent)) / float64(processingTime)
 	}
 
 	// 平均时长 总耗时/总请求数/并发数 纳秒=>毫秒
 	if successNum != 0 && concurrent != 0 {
-		averageTime = float64(processingTime) / float64(successNum*1e6*concurrent)
+		averageTime = float64(processingTime) / float64(successNum*1e6*uint64(concurrent))
 	}
 
 	// 纳秒=>毫秒
@@ -166,7 +169,7 @@ func header() {
 }
 
 // 打印表格
-func table(successNum, failureNum uint64, errCode map[int]int, qps, averageTime, maxTimeFloat, minTimeFloat, requestTimeFloat float64, chanIdLen int) {
+func table(successNum, failureNum uint64, errCode utils.ConcurrentMap, qps, averageTime, maxTimeFloat, minTimeFloat, requestTimeFloat float64, chanIdLen int) {
 	// 打印的时长都为毫秒
 	result := fmt.Sprintf("%4.0fs│%7d│%7d│%7d│%8.2f│%8.2f│%8.2f│%8.2f│%v", requestTimeFloat, chanIdLen, successNum, failureNum, qps, maxTimeFloat, minTimeFloat, averageTime, printMap(errCode))
 	fmt.Println(result)
@@ -175,13 +178,13 @@ func table(successNum, failureNum uint64, errCode map[int]int, qps, averageTime,
 }
 
 // 输出错误码、次数 节约字符(终端一行字符大小有限)
-func printMap(errCode map[int]int) (mapStr string) {
+func printMap(errCode utils.ConcurrentMap) (mapStr string) {
 
 	var (
 		mapArr []string
 	)
-	for key, value := range errCode {
-		mapArr = append(mapArr, fmt.Sprintf("%d:%d", key, value))
+	for key, value := range errCode.Items() {
+		mapArr = append(mapArr, fmt.Sprintf("%s:%d", key, value))
 	}
 
 	mapStr = strings.Join(mapArr, ";")

pkg/var.go

@@ -1,7 +1,7 @@
 package pkg
 
 var (
-	Concurrency uint64   // Concurrency is 并发请求数
+	Concurrency int   // Concurrency is 并发请求数
 	Number      uint64   // Number is 单个协程的请求总数
 	Debug       bool     // Debug is 是否开其调试模式
 	Path        string   // Path is curl 文件路径

pkg/websocket.go

@@ -22,7 +22,7 @@ func init() {
 }
 
 // WebSocket is  web socket go link
-func WebSocket(chanId uint64, ch chan<- *RequestResults, totalNumber uint64, wg *sync.WaitGroup, request *Request, ws *WebSocketC) {
+func WebSocket(chanId int, ch chan<- *RequestResults, totalNumber uint64, wg *sync.WaitGroup, request *Request, ws *WebSocketC) {
 
 	defer func() {
 		wg.Done()
@@ -69,7 +69,7 @@ end:
 }
 
 // webSocketRequest is 请求
-func webSocketRequest(chanId uint64, ch chan<- *RequestResults, i uint64, request *Request, ws *WebSocketC) {
+func webSocketRequest(chanId int, ch chan<- *RequestResults, i uint64, request *Request, ws *WebSocketC) {
 
 	var (
 		startTime = time.Now()