save file

src/crane/crane_master.go

@@ -38,6 +38,7 @@ var workerConn map[int]net.Conn // map worker servNum to its corresponding conne
 var workerDataConn map[int]net.Conn
 var workerPort int
 var needRestart bool
+var tempData []interface{}
 
 var activeListMux sync.Mutex
 var dataSaveMux sync.Mutex
@@ -77,6 +78,7 @@ func InitializeCraneMaster() {
 	possibleServerIdx = 0
 	workerConn = make(map[int]net.Conn)
 	workerDataConn = make(map[int]net.Conn)
+
 	childrenBolts = nil
 	// Check if I am backup
 	if shared.GetOwnServerNumber() == MASTER_ONE {
@@ -295,6 +297,7 @@ func startNewJob() {
 			}
 		}
 	}
+	tempData = make([]interface{})
 	//wg.Wait()
 	// start spout goroutine
 	go runSpout()
@@ -418,7 +421,7 @@ func runSpout() {
 			return
 		}
 	}
-
+	fmt.Println("Finish reading all tuples")
 	if err := scanner.Err(); err != nil {
 		fmt.Println(err)
 	}
@@ -442,17 +445,20 @@ func handleWorkerConnection(conn net.Conn) {
 
 			// TODO: Process data
 			// fmt.Println(*data)
-			debugCount += 1
+			debugCount++
 			if debugCount%10000 == 0 {
 				fmt.Println(debugCount)
 			}
+			tempNewLine := []interface{}{"\n"}
+			tempData = append(tempData, tempNewLine)
+			tempData = append(tempData, data.Tuple)
 			//saveData(data.Tuple)
 		}
 	}
 	conn.Close()
 }
 
-func saveData(tuple []interface{}) {
+func appendDataToFile(tuple []interface{}) {
 	dataSaveMux.Lock()
 	f, err := os.OpenFile(craneOutput, os.O_APPEND|os.O_WRONLY|os.O_CREATE, 0600)
 	if err != nil {
@@ -469,6 +475,23 @@ func saveData(tuple []interface{}) {
 	return
 }
 
+func saveTempData() {
+	dataSaveMux.Lock()
+	f, err := os.Create(craneOutput)
+	if err != nil {
+		panic(err)
+	}
+	defer f.Close()
+	for _, elem := range tempData {
+		tmpStr := fmt.Sprintf("%v\n", elem)
+		if _, err = f.WriteString(tmpStr); err != nil {
+			fmt.Println(err)
+		}
+	}
+	dataSaveMux.Unlock()
+	return
+}
+
 // Save current output to a file in sdfs
 func SaveToFile() {
 	dataSaveMux.Lock()