commit

aarch64/pjisuv/common/service/disk_clean.go

@@ -80,11 +80,12 @@ func getIndexToRemoveForLRU() int {
 	i := len(lru) - 1
 	for i >= 0 {
 		for i2, window := range entity.TimeWindowConsumerQueue {
-			if masterConfig.LabelMapTriggerId[window.FaultTime] == lru[i] {
-				return i2
+			for _, label := range window.Labels {
+				if masterConfig.LabelMapTriggerId[label] == lru[i] {
+					return i2
+				}
 			}
 		}
 	}
 	return -1
-
 }

aarch64/pjisuv/master/package/service/move_bag_and_send_window.go

@@ -11,11 +11,11 @@ import (
 	"time"
 )
 
-// RunTimeWindowProducerQueue 将时间窗口内的包全部move出去，并等待当前时间窗口结束触发上传
+// RunTimeWindowProducerQueue 依次处理时间窗口，从data目录移动到copy目录
 func RunTimeWindowProducerQueue() {
 	c_log.GlobalLogger.Info("生产者队列goroutine - 启动")
 	for {
-		// 收到自杀信号
+		// 1 监控自杀信号
 		select {
 		case signal := <-commonService.ChannelKillMove:
 			if signal == 1 {
@@ -30,58 +30,51 @@ func RunTimeWindowProducerQueue() {
 			}
 		default:
 		}
-
 		time.Sleep(time.Duration(1) * time.Second)
-		if len(entity.TimeWindowProducerQueue) > 0 {
-			bags, _ := util.ListAbsolutePathWithSuffixAndSort(commonConfig.CloudConfig.BagDataDir, ".bag")
-			currentTimeWindow := entity.TimeWindowProducerQueue[0]
-			move := false
-			bigger := false
-			for _, bag := range bags {
-				bagTime := util.GetBagTime(bag)
-				// 2 如果bag不小于timeWindowBegin不大于timeWindowEnd，则移动
-				compare1 := util.TimeCustom1GreaterEqualThanTimeCustom2(bagTime, currentTimeWindow.TimeWindowBegin)
-				compare2 := util.TimeCustom1LessEqualThanTimeCustom2(bagTime, currentTimeWindow.TimeWindowEnd)
-				if compare1 && compare2 {
-					// 将bag包移动到Copy目录
-					domain.MoveFromDataToCopy(currentTimeWindow.FaultTime, commonConfig.CloudConfig.BagDataDir, bag, commonConfig.CloudConfig.BagCopyDir)
-					move = true
-				} else {
-					if util.TimeCustom1GreaterEqualThanTimeCustom2(bagTime, currentTimeWindow.TimeWindowBegin) {
-						// 必须已经生成了窗口之后的包才算窗口结束了
-						bigger = true
-						break
-					}
+		// 2 处理
+		if len(entity.TimeWindowProducerQueue) == 0 {
+			continue
+		}
+		currentTimeWindow := entity.TimeWindowProducerQueue[0]                                         // 当前窗口
+		bags, _ := util.ListAbsolutePathWithSuffixAndSort(commonConfig.CloudConfig.BagDataDir, ".bag") // data目录的bag包列表
+		move := false
+		bigger := false
+		for _, bag := range bags {
+			bagTime := util.GetBagTime(bag) // bag包对应的时间
+			// 3 如果bag时间在窗口时间区间内，则从data移动到copy
+			if util.BagTimeInInterval(bagTime, currentTimeWindow.TimeWindowBegin, currentTimeWindow.TimeWindowEnd) {
+				domain.MoveFromDataToCopy(currentTimeWindow.FaultTime, commonConfig.CloudConfig.BagDataDir, bag, commonConfig.CloudConfig.BagCopyDir)
+				move = true
+			} else {
+				// 4 如果已经生成了新的包，则当前窗口已经移动完成了
+				if util.TimeCustom1GreaterEqualThanTimeCustom2(bagTime, currentTimeWindow.TimeWindowBegin) {
+					bigger = true
+					break
 				}
 			}
-			// 如果没有包可以供当前窗口移动，且已经生成了更新的包，则当前窗口已经可以上传
-			if !move && bigger {
-				// 1 如果第一个已经大于了timeWindowEnd，则触发上传并删除
-				// 将时间窗口发送给从节点
-				currentTimeWindow.CanUpload = "yes"
-				c_log.GlobalLogger.Info("将已完成的窗口发送给从节点：", currentTimeWindow.CanUpload)
-				domain.SupplyCopyBags(commonConfig.CloudConfig.BagDataDir, commonConfig.CloudConfig.BagCopyDir, currentTimeWindow)
+		}
+		// 5 如果没有包可以供当前窗口移动，且已经生成了更新的包，则当前窗口已经可以上传
+		if !move && bigger {
+			currentTimeWindow.CanUpload = "yes"
+			// 如果第一个已经大于了timeWindowEnd，则触发上传并删除
+			domain.SupplyCopyBags(commonConfig.CloudConfig.BagDataDir, commonConfig.CloudConfig.BagCopyDir, currentTimeWindow)
+			entity.RefreshTcpSendTime()
+			go sendTimeWindowByTcp(currentTimeWindow)
+			entity.RemoveHeadOfdTimeWindowProducerQueue()
+			entity.AddTimeWindowToTimeWindowConsumerQueue(currentTimeWindow)
+			// 获取copy目录下的字典json，key为触发时间，value为label
+			timeToLabelJson, _ := util.ReadFile(commonConfig.CloudConfig.TimeToLabelJsonPath)
+			timeToLabelMap, _ := util.JsonStringToMap(timeToLabelJson)
+			timeToLabelMap[currentTimeWindow.FaultTime] = util.ToString(currentTimeWindow.Labels)
+			timeToLabelJson, _ = util.MapToJsonString(timeToLabelMap)
+			_ = util.WriteFile(timeToLabelJson, commonConfig.CloudConfig.TimeToLabelJsonPath)
+			continue
+		} else { // 保证当前窗口只发送一次，每间隔5秒发一次非yes窗口
+			if int(time.Since(entity.TcpSendTime).Seconds()) > commonConfig.CloudConfig.TimeWindowSendGap {
+				currentTimeWindow.CanUpload = "no"
+				c_log.GlobalLogger.Info("每隔", commonConfig.CloudConfig.TimeWindowSendGap, "秒发送一次tcp消息")
 				entity.RefreshTcpSendTime()
 				go sendTimeWindowByTcp(currentTimeWindow)
-				// 将时间窗口移出准备队列
-				entity.RemoveHeadOfdTimeWindowProducerQueue()
-				// 将时间窗口加入运行队列
-				entity.AddTimeWindowToTimeWindowConsumerQueue(currentTimeWindow)
-				// 获取copy目录下的字典json，key为触发时间，value为label
-				timeToLabelJson, _ := util.ReadFile(commonConfig.CloudConfig.TimeToLabelJsonPath)
-				timeToLabelMap, _ := util.JsonStringToMap(timeToLabelJson)
-				timeToLabelMap[currentTimeWindow.FaultTime] = util.ToString(currentTimeWindow.Labels)
-				timeToLabelJson, _ = util.MapToJsonString(timeToLabelMap)
-				_ = util.WriteFile(timeToLabelJson, commonConfig.CloudConfig.TimeToLabelJsonPath)
-				continue
-			} else { // 保证当前窗口只发送一次，每间隔5秒发一次
-				if int(time.Since(entity.TcpSendTime).Seconds()) > commonConfig.CloudConfig.TimeWindowSendGap {
-					c_log.GlobalLogger.Info("每隔", commonConfig.CloudConfig.TimeWindowSendGap, "秒发送一次tcp消息")
-					entity.RefreshTcpSendTime()
-					// 2 如果第一个不大于timeWindowEnd，则发送不可上传的窗口信息。
-					currentTimeWindow.CanUpload = "no"
-					go sendTimeWindowByTcp(currentTimeWindow)
-				}
 			}
 		}
 	}

common/domain/d_service.go

@@ -28,19 +28,22 @@ func SupplyCopyBags(bagDataDir string, bagCopyDir string, currentTimeWindow enti
 	// 如果bag包没有达到length，补充几个
 	copyBags, _ := util.ListAbsolutePathWithSuffixAndSort(GetCopyDir(bagCopyDir, currentTimeWindow.FaultTime), ".bag")
 	copyBagsLength := len(copyBags)
-	if copyBagsLength < currentTimeWindow.Length {
-		time.Sleep(time.Duration(copyBagsLength) * time.Second)
-		dataBags, _ := util.ListAbsolutePathWithSuffixAndSort(bagDataDir, ".bag")
-		gap := currentTimeWindow.Length - copyBagsLength
-		c_log.GlobalLogger.Info("故障 ", currentTimeWindow.FaultTime, "需要补充 ", gap, " 个 bag 包")
-		for _, bag := range dataBags {
-			bagTime := util.GetBagTime(bag)
-			if util.TimeCustom1GreaterEqualThanTimeCustom2(bagTime, currentTimeWindow.FaultTime) {
-				MoveFromDataToCopy(currentTimeWindow.FaultTime, bagDataDir, bag, bagCopyDir)
-				gap = gap - 1
-				if gap == 0 {
-					break
-				}
+	if copyBagsLength >= currentTimeWindow.Length {
+		return
+	}
+	// 等待差距时间
+	gap := currentTimeWindow.Length - copyBagsLength
+	time.Sleep(time.Duration(gap) * time.Second)
+	// 获取data目录下的包列表
+	dataBags, _ := util.ListAbsolutePathWithSuffixAndSort(bagDataDir, ".bag")
+	c_log.GlobalLogger.Info("故障 ", currentTimeWindow.FaultTime, "需要补充 ", gap, " 个 bag 包")
+	for _, bag := range dataBags {
+		bagTime := util.GetBagTime(bag)
+		if util.TimeCustom1GreaterEqualThanTimeCustom2(bagTime, currentTimeWindow.FaultTime) {
+			MoveFromDataToCopy(currentTimeWindow.FaultTime, bagDataDir, bag, bagCopyDir)
+			gap = gap - 1
+			if gap == 0 {
+				break
 			}
 		}
 	}

common/util/u_time.go

@@ -11,6 +11,13 @@ func GetNowTimeCustom() string {
 	formattedTime := currentTime.Format("2006-01-02-15-04-05")
 	return formattedTime
 }
+
+func BagTimeInInterval(bagTime string, begin string, end string) bool {
+	compare1 := TimeCustom1GreaterEqualThanTimeCustom2(bagTime, begin)
+	compare2 := TimeCustom1LessEqualThanTimeCustom2(bagTime, end)
+	return compare1 && compare2
+}
+
 func TimeCustom1GreaterEqualThanTimeCustom2(timeCustom1 string, timeCustom2 string) bool {
 	timeInt1, _ := strconv.Atoi(strings.Replace(timeCustom1, "-", "", -1))
 	timeInt2, _ := strconv.Atoi(strings.Replace(timeCustom2, "-", "", -1))

kinglong/common/svc/disk_clean.go

@@ -79,8 +79,10 @@ func getIndexToRemoveForLRU() int {
 	i := len(lru) - 1
 	for i >= 0 {
 		for i2, window := range global.TimeWindowConsumerQueue {
-			if masterConfig.LabelMapTriggerId[window.FaultTime] == lru[i] {
-				return i2
+			for _, label := range window.Labels {
+				if masterConfig.LabelMapTriggerId[label] == lru[i] {
+					return i2
+				}
 			}
 		}
 	}

pji/common/svc/disk_clean.go

@@ -78,8 +78,10 @@ func getIndexToRemoveForLRU() int {
 	i := len(lru) - 1
 	for i >= 0 {
 		for i2, window := range global.TimeWindowConsumerQueue {
-			if masterConfig.LabelMapTriggerId[window.FaultTime] == lru[i] {
-				return i2
+			for _, label := range window.Labels {
+				if masterConfig.LabelMapTriggerId[label] == lru[i] {
+					return i2
+				}
 			}
 		}
 	}