synchronized agent channel read/write

agent/agent_server.go

@@ -56,7 +56,7 @@ type AgentServer struct {
 	Port                  int
 	name2Store            map[string]*LiveDataStore
 	dir                   string
-	name2StoreLock        sync.Mutex
+	name2StoreCond        *sync.Cond
 	wg                    sync.WaitGroup
 	l                     net.Listener
 	computeResource       *resource.ComputeResource
@@ -71,12 +71,16 @@ func NewAgentServer(option *AgentServerOption) *AgentServer {
 	}
 	println("starting in", absoluteDir)
 	option.Dir = &absoluteDir
+
+	var lock sync.Mutex
+
 	as := &AgentServer{
-		Option:     option,
-		Master:     *option.Master,
-		Port:       *option.Port,
-		dir:        absoluteDir,
-		name2Store: make(map[string]*LiveDataStore),
+		Option:         option,
+		Master:         *option.Master,
+		Port:           *option.Port,
+		dir:            absoluteDir,
+		name2Store:     make(map[string]*LiveDataStore),
+		name2StoreCond: sync.NewCond(&lock),
 		computeResource: &resource.ComputeResource{
 			CPUCount: *option.MaxExecutor,
 			CPULevel: *option.CPULevel,

agent/agent_server_cmd.go

@@ -39,11 +39,11 @@ func (as *AgentServer) handleStart(conn net.Conn,
 	reply := &cmd.StartResponse{}
 
 	dir := path.Join(*as.Option.Dir, startRequest.GetDir())
-	os.MkdirAll(dir, os.ModeDir)
+	os.MkdirAll(dir, 0755)
 	err := rsync.FetchFilesTo(startRequest.GetHost()+":"+strconv.Itoa(int(startRequest.GetPort())), dir)
 	if err != nil {
 		log.Printf("Failed to download file: %v", err)
-		*reply.Error = err.Error()
+		reply.Error = proto.String(err.Error())
 		return reply
 	}
 
@@ -67,7 +67,7 @@ func (as *AgentServer) handleStart(conn net.Conn,
 	if err != nil {
 		log.Printf("Failed to start command %s under %s: %v",
 			cmd.Path, cmd.Dir, err)
-		*reply.Error = err.Error()
+		reply.Error = proto.String(err.Error())
 	} else {
 		reply.Pid = proto.Int32(int32(cmd.Process.Pid))
 	}

agent/agent_server_read.go

@@ -7,26 +7,25 @@ import (
 	"net"
 	"time"
 
-	"github.com/chrislusf/glow/io/store"
 	"github.com/chrislusf/glow/util"
 )
 
 func (as *AgentServer) handleLocalReadConnection(conn net.Conn, name string, offset int64) {
-	as.name2StoreLock.Lock()
-	ds, ok := as.name2Store[name]
-	if !ok {
-		s, err := store.NewLocalFileDataStore(as.dir, fmt.Sprintf("%s-%d", name, as.Port))
-		if err != nil {
-			// log.Printf("Failed to create queue on disk: %v", err)
-			as.name2StoreLock.Unlock()
-			return
+	var ds *LiveDataStore
+	var ok bool
+
+	as.name2StoreCond.L.Lock()
+	for {
+		ds, ok = as.name2Store[name]
+		if ok {
+			break
 		}
-		as.name2Store[name] = NewLiveDataStore(s)
-		ds = as.name2Store[name]
+		println(name, "is waiting to read...")
+		as.name2StoreCond.Wait()
 	}
-	as.name2StoreLock.Unlock()
+	as.name2StoreCond.L.Unlock()
 
-	// println(name, "start reading from:", offset)
+	println(name, "start reading from:", offset)
 
 	closeSignal := make(chan bool, 1)
 

agent/agent_server_write.go

@@ -12,20 +12,25 @@ import (
 
 func (as *AgentServer) handleWriteConnection(r io.Reader, name string) {
 
-	as.name2StoreLock.Lock()
+	as.name2StoreCond.L.Lock()
 	ds, ok := as.name2Store[name]
-	if !ok {
-		s, err := store.NewLocalFileDataStore(as.dir, fmt.Sprintf("%s-%d", name, as.Port))
-		if err != nil {
-			log.Printf("Failed to create a queue on disk: %v", err)
-			as.name2StoreLock.Unlock()
-			return
-		}
-		as.name2Store[name] = NewLiveDataStore(s)
-		ds = as.name2Store[name]
+	if ok {
+		as.doDelete(name)
+	}
 
+	s, err := store.NewLocalFileDataStore(as.dir, fmt.Sprintf("%s-%d", name, as.Port))
+	if err != nil {
+		log.Printf("Failed to create a queue on disk: %v", err)
+		as.name2StoreCond.L.Unlock()
+		return
 	}
-	as.name2StoreLock.Unlock()
+
+	as.name2Store[name] = NewLiveDataStore(s)
+	ds = as.name2Store[name]
+	println(name, "is broadcasting...")
+	as.name2StoreCond.Broadcast()
+
+	as.name2StoreCond.L.Unlock()
 
 	//register stream
 	go client.NewHeartBeater(as.Port, *as.Option.Master).StartChannelHeartBeat(ds.killHeartBeater, name)
@@ -52,15 +57,20 @@ func (as *AgentServer) handleWriteConnection(r io.Reader, name string) {
 
 func (as *AgentServer) handleDelete(name string) {
 
-	as.name2StoreLock.Lock()
+	as.name2StoreCond.L.Lock()
+	defer as.name2StoreCond.L.Unlock()
+
+	as.doDelete(name)
+}
+
+func (as *AgentServer) doDelete(name string) {
+
 	ds, ok := as.name2Store[name]
 	if !ok {
-		as.name2StoreLock.Unlock()
 		return
 	}
 
 	delete(as.name2Store, name)
-	as.name2StoreLock.Unlock()
 
 	ds.Destroy()
 }

driver/context_driver.go

@@ -131,5 +131,8 @@ func (fcd *FlowContextDriver) CloseOutputChannels(fc *flow.FlowContext) {
 }
 
 func (option *DriverOption) RelatedFileNames() []string {
-	return strings.Split(option.RelatedFiles, strconv.QuoteRune(os.PathListSeparator))
+	if option.RelatedFiles != "" {
+		return strings.Split(option.RelatedFiles, strconv.QuoteRune(os.PathListSeparator))
+	}
+	return []string{}
 }

driver/rsync/fetch_url.go

@@ -5,10 +5,13 @@ import (
 	"fmt"
 	"io/ioutil"
 	"path/filepath"
+	"sync"
 
 	"github.com/chrislusf/glow/util"
 )
 
+var fetchLock sync.Mutex
+
 type ListFileResult struct {
 	Files []FileHash `json:"files,omitempty"`
 }
@@ -27,6 +30,8 @@ func ListFiles(server string) ([]FileHash, error) {
 }
 
 func FetchFilesTo(driverAddress string, dir string) error {
+	fetchLock.Lock()
+	defer fetchLock.Unlock()
 
 	fileList, err := ListFiles(driverAddress)
 	if err != nil {

driver/rsync/http_server.go

@@ -33,13 +33,13 @@ func NewRsyncServer(file string, relatedFiles []string) (*RsyncServer, error) {
 		RelatedFiles:   relatedFiles,
 	}
 	if fh, err := GenerateFileHash(file); err != nil {
-		log.Printf("Failed to read %s: %v", file, err)
+		log.Printf("Failed1 to read %s: %v", file, err)
 	} else {
 		rs.fileHashes = append(rs.fileHashes, *fh)
 	}
 	for _, f := range rs.RelatedFiles {
 		if fh, err := GenerateFileHash(f); err != nil {
-			log.Printf("Failed to read %s: %v", f, err)
+			log.Printf("Failed2 to read %s: %v", f, err)
 		} else {
 			rs.fileHashes = append(rs.fileHashes, *fh)
 		}

glow.go

@@ -81,8 +81,9 @@ func main() {
 		wg.Wait()
 
 	case receiver.FullCommand():
+		target := r.FindTarget(*receiveFromChanName, *receiverMaster)
 		rc := r.NewReceiveChannel(*receiveFromChanName, 0)
-		recvChan, err := rc.GetDirectChannel(*receiverMaster)
+		recvChan, err := rc.GetDirectChannel(target)
 		if err != nil {
 			panic(err)
 		}

io/store/rotating_file_store.go

@@ -355,6 +355,7 @@ func (l *RotatingFileStore) Destroy() {
 // oldLogFiles returns the list of backup log files stored in the same
 // directory as the current log file, sorted by ModTime
 func (l *RotatingFileStore) listOldLogFiles() ([]logInfo, error) {
+	os.MkdirAll(l.dir(), 0755)
 	files, err := ioutil.ReadDir(l.dir())
 	if err != nil {
 		return nil, fmt.Errorf("can't read log file directory: %s", err)