add session lock

.github/workflows/test.yaml

@@ -5,9 +5,6 @@ on:
   pull_request:
     branches: [ master ]
 
-concurrency:
-  cancel-in-progress: true
-
 jobs:
   go-client:
     runs-on: ubuntu-latest
@@ -62,4 +59,4 @@ jobs:
       - name: down
         if: always()
         run: |
-          make down  
+          make down

session.go

@@ -10,6 +10,7 @@ package nebula_go
 
 import (
 	"fmt"
+	"sync"
 
 	"github.com/facebook/fbthrift/thrift/lib/go/thrift"
 	"github.com/vesoft-inc/nebula-go/v2/nebula"
@@ -26,11 +27,14 @@ type Session struct {
 	connection *connection
 	connPool   *ConnectionPool
 	log        Logger
+	mu         sync.Mutex
 	timezoneInfo
 }
 
 // Execute returns the result of the given query as a ResultSet
 func (session *Session) Execute(stmt string) (*ResultSet, error) {
+	session.mu.Lock()
+	defer session.mu.Unlock()
 	if session.connection == nil {
 		return nil, fmt.Errorf("failed to execute: Session has been released")
 	}
@@ -130,6 +134,8 @@ func (session *Session) Execute(stmt string) (*ResultSet, error) {
 //     ]
 // }
 func (session *Session) ExecuteJson(stmt string) ([]byte, error) {
+	session.mu.Lock()
+	defer session.mu.Unlock()
 	if session.connection == nil {
 		return nil, fmt.Errorf("failed to execute: Session has been released")
 	}
@@ -182,6 +188,8 @@ func (session *Session) Release() {
 	if session == nil {
 		return
 	}
+	session.mu.Lock()
+	defer session.mu.Unlock()
 	if session.connection == nil {
 		session.log.Warn("Session has been released")
 		return

session_test.go

@@ -0,0 +1,47 @@
+/* Copyright (c) 2021 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License.
+ */
+
+package nebula_go
+
+import (
+	"testing"
+	"time"
+)
+
+func TestSession_Execute(t *testing.T) {
+	config := PoolConfig{}
+	host := HostAddress{address, port}
+	pool, err := NewConnectionPool([]HostAddress{host}, config, DefaultLogger{})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	sess, err := pool.GetSession("root", "nebula")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	f := func(s *Session) {
+		time.Sleep(10 * time.Microsecond)
+		reps, err := s.Execute("yield 1")
+		if err != nil {
+			t.Fatal(err)
+		}
+		if !reps.IsSucceed() {
+			t.Fatal(reps.resp.ErrorMsg)
+		}
+	}
+	go func() {
+		for {
+			f(sess)
+		}
+	}()
+	go func() {
+		for {
+			f(sess)
+		}
+	}()
+	time.Sleep(300 * time.Millisecond)
+}