[INLONG-11601][SDK] Fix data race in Golang SDK (#11602)

Co-authored-by: gunli <[email protected]>
apache · Dec 16, 2024 · 73e6b2a · 73e6b2a
1 parent 64b3531
commit 73e6b2a
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 5 deletions.
diff --git a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/dataproxy/request.go b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/dataproxy/request.go
@@ -139,8 +139,9 @@ func (b *batchReq) done(err error) {
 	}
 
 	if b.pool != nil {
-		b.pool.Put(b)
+		pool := b.pool
 		b.pool = nil
+		pool.Put(b)
 	}
 }
 
@@ -368,8 +369,9 @@ func (s *sendDataReq) done(err error, errCode string) {
 	}
 
 	if s.pool != nil {
-		s.pool.Put(s)
+		pool := s.pool
 		s.pool = nil
+		pool.Put(s)
 	}
 }
 

diff --git a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/dataproxy/worker.go b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/dataproxy/worker.go
@@ -130,7 +130,7 @@ type worker struct {
 	metrics            *metrics                 // metrics
 	bufferPool         bufferpool.BufferPool    // buffer pool
 	bytePool           bufferpool.BytePool      // byte pool
-	stop               bool                     // stop the worker
+	stop               chan struct{}            // stop the worker
 }
 
 func newWorker(cli *client, index int, opts *Options) (*worker, error) {
@@ -162,6 +162,7 @@ func newWorker(cli *client, index int, opts *Options) (*worker, error) {
 		bufferPool:         opts.BufferPool,
 		bytePool:           opts.BytePool,
 		log:                opts.Logger,
+		stop:               make(chan struct{}),
 	}
 
 	// set to init state
@@ -197,8 +198,10 @@ func (w *worker) start() {
 			}
 		}()
 
-		for !w.stop {
+		for {
 			select {
+			case <-w.stop:
+				return
 			case req, ok := <-w.cmdChan:
 				if !ok {
 					continue
@@ -640,7 +643,7 @@ func (w *worker) close() {
 
 	// wait for the close request done
 	<-req.doneCh
-	w.stop = true
+	close(w.stop)
 }
 
 func (w *worker) handleClose(req *closeReq) {