Fix permission denied in gdbserver container

Signed-off-by: quicksilver <zhifeng.zhang@zilliz.com>

build/builder.sh

@@ -15,6 +15,7 @@ fi
 
 if [ "${1-}" = "gdbserver" ]; then
     mkdir -p "${DOCKER_VOLUME_DIRECTORY:-.docker}/amd64-ubuntu18.04-gdbserver-cache"
+    mkdir -p "${DOCKER_VOLUME_DIRECTORY:-.docker}/amd64-ubuntu18.04-gdbserver-home"
     chmod -R 777 "${DOCKER_VOLUME_DIRECTORY:-.docker}"
 
     docker-compose pull --ignore-pull-failures gdbserver
@@ -25,6 +26,11 @@ if [ "${1-}" = "gdbserver" ]; then
     exit 0
 fi
 
+if [ "${1-}" = "down" ]; then
+    docker-compose down
+    exit 0
+fi
+
 # Attempt to run in the container with the same UID/GID as we have on the host,
 # as this results in the correct permissions on files created in the shared
 # volumes. This isn't always possible, however, as IDs less than 100 are

internal/reader/collection_replica.go

@@ -29,9 +29,7 @@ import (
  * is up-to-date.
  */
 type collectionReplica interface {
-	// tSafe
-	getTSafe() Timestamp
-	setTSafe(t Timestamp)
+	getTSafe() *tSafe
 
 	// collection
 	getCollectionNum() int
@@ -57,27 +55,18 @@ type collectionReplica interface {
 }
 
 type collectionReplicaImpl struct {
-	tSafeMu sync.Mutex
-	tSafe   Timestamp
-
 	mu          sync.RWMutex
 	collections []*Collection
 	segments    map[UniqueID]*Segment
+
+	tSafe *tSafe
 }
 
 //----------------------------------------------------------------------------------------------------- tSafe
-func (colReplica *collectionReplicaImpl) getTSafe() Timestamp {
-	colReplica.tSafeMu.Lock()
-	defer colReplica.tSafeMu.Unlock()
+func (colReplica *collectionReplicaImpl) getTSafe() *tSafe {
 	return colReplica.tSafe
 }
 
-func (colReplica *collectionReplicaImpl) setTSafe(t Timestamp) {
-	colReplica.tSafeMu.Lock()
-	colReplica.tSafe = t
-	colReplica.tSafeMu.Unlock()
-}
-
 //----------------------------------------------------------------------------------------------------- collection
 func (colReplica *collectionReplicaImpl) getCollectionNum() int {
 	colReplica.mu.RLock()

internal/reader/flow_graph_service_time_node.go

@@ -28,7 +28,7 @@ func (stNode *serviceTimeNode) Operate(in []*Msg) []*Msg {
 	}
 
 	// update service time
-	(*stNode.replica).setTSafe(serviceTimeMsg.timeRange.timestampMax)
+	(*(*stNode.replica).getTSafe()).set(serviceTimeMsg.timeRange.timestampMax)
 	return nil
 }
 

internal/reader/query_node.go

@@ -33,9 +33,13 @@ func NewQueryNode(ctx context.Context, queryNodeID uint64) *QueryNode {
 	segmentsMap := make(map[int64]*Segment)
 	collections := make([]*Collection, 0)
 
+	tSafe := newTSafe()
+
 	var replica collectionReplica = &collectionReplicaImpl{
 		collections: collections,
 		segments:    segmentsMap,
+
+		tSafe: tSafe,
 	}
 
 	return &QueryNode{

internal/reader/search_service.go

@@ -20,7 +20,9 @@ type searchService struct {
 	ctx    context.Context
 	cancel context.CancelFunc
 
-	replica               *collectionReplica
+	replica      *collectionReplica
+	tSafeWatcher *tSafeWatcher
+
 	searchMsgStream       *msgstream.MsgStream
 	searchResultMsgStream *msgstream.MsgStream
 }
@@ -63,7 +65,9 @@ func newSearchService(ctx context.Context, replica *collectionReplica) *searchSe
 		ctx:    searchServiceCtx,
 		cancel: searchServiceCancel,
 
-		replica:               replica,
+		replica:      replica,
+		tSafeWatcher: newTSafeWatcher(),
+
 		searchMsgStream:       &inputStream,
 		searchResultMsgStream: &outputStream,
 	}
@@ -101,6 +105,18 @@ func (ss *searchService) close() {
 	ss.cancel()
 }
 
+func (ss *searchService) register() {
+	tSafe := (*(ss.replica)).getTSafe()
+	(*tSafe).registerTSafeWatcher(ss.tSafeWatcher)
+}
+
+func (ss *searchService) waitNewTSafe() Timestamp {
+	// block until dataSyncService updating tSafe
+	ss.tSafeWatcher.hasUpdate()
+	timestamp := (*(*ss.replica).getTSafe()).get()
+	return timestamp
+}
+
 func (ss *searchService) search(searchMessages []msgstream.TsMsg) error {
 
 	type SearchResult struct {

internal/reader/tsafe.go

@@ -0,0 +1,63 @@
+package reader
+
+import (
+	"sync"
+)
+
+type tSafeWatcher struct {
+	notifyChan chan bool
+}
+
+func newTSafeWatcher() *tSafeWatcher {
+	return &tSafeWatcher{
+		notifyChan: make(chan bool),
+	}
+}
+
+func (watcher *tSafeWatcher) notify() {
+	if len(watcher.notifyChan) == 0 {
+		watcher.notifyChan <- true
+	}
+}
+
+func (watcher *tSafeWatcher) hasUpdate() {
+	<-watcher.notifyChan
+}
+
+type tSafe interface {
+	get() Timestamp
+	set(t Timestamp)
+	registerTSafeWatcher(t *tSafeWatcher)
+}
+
+type tSafeImpl struct {
+	tSafeMu     sync.Mutex
+	tSafe       Timestamp
+	watcherList []*tSafeWatcher
+}
+
+func newTSafe() *tSafe {
+	var t tSafe = &tSafeImpl{
+		watcherList: make([]*tSafeWatcher, 0),
+	}
+	return &t
+}
+
+func (ts *tSafeImpl) registerTSafeWatcher(t *tSafeWatcher) {
+	ts.watcherList = append(ts.watcherList, t)
+}
+
+func (ts *tSafeImpl) get() Timestamp {
+	ts.tSafeMu.Lock()
+	defer ts.tSafeMu.Unlock()
+	return ts.tSafe
+}
+
+func (ts *tSafeImpl) set(t Timestamp) {
+	ts.tSafeMu.Lock()
+	ts.tSafe = t
+	ts.tSafeMu.Unlock()
+	for _, watcher := range ts.watcherList {
+		watcher.notify()
+	}
+}

internal/reader/tsafe_test.go

@@ -0,0 +1,21 @@
+package reader
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestTSafe_GetAndSet(t *testing.T) {
+	tSafe := newTSafe()
+	watcher := newTSafeWatcher()
+	(*tSafe).registerTSafeWatcher(watcher)
+
+	go func() {
+		watcher.hasUpdate()
+		timestamp := (*tSafe).get()
+		assert.Equal(t, timestamp, Timestamp(1000))
+	}()
+
+	(*tSafe).set(Timestamp(1000))
+}