modify the process order and change the feedin to send once has data

api/driver.go

@@ -115,23 +115,23 @@ func (d *driverImpl) removeFinishedFeedInTasks() {
 func (d *driverImpl) doOneFeedInTask(task *feedInTask) bool {
 	madeProgress := false
 
-	canSendAll := true
-	for _, port := range task.localPorts {
+	for i, port := range task.localPorts {
+		if task.portRounds[i] >= task.rounds {
+			continue
+		}
 		if !port.CanSend() {
-			canSendAll = false
-			break
+			continue
 		}
-	}
 
-	if !canSendAll {
-		return false
-	}
+		dataIndex := task.portRounds[i]*task.stride + i
+		if dataIndex < 0 || dataIndex >= len(task.data) {
+			continue
+		}
 
-	for i, port := range task.localPorts {
 		msg := cgra.MoveMsgBuilder{}.
 			WithSrc(port.AsRemote()).
 			WithDst(task.remotePorts[i]).
-			WithData(cgra.NewScalar(task.data[task.round*task.stride+i])).
+			WithData(cgra.NewScalar(task.data[dataIndex])).
 			WithColor(task.color).
 			WithSendTime(d.Engine.CurrentTime()). // Set the current engine time here
 			Build()
@@ -145,15 +145,15 @@ func (d *driverImpl) doOneFeedInTask(task *feedInTask) bool {
 		core.Trace("DataFlow",
 			"Behavior", "FeedIn",
 			slog.Float64("Time", float64(d.Engine.CurrentTime()*1e9)),
-			"Data", task.data[task.round*task.stride+i],
+			"Data", task.data[dataIndex],
 			"Color", task.color,
 			"From", port.Name(),
 			"To", task.remotePorts[i],
 		)
+		task.portRounds[i]++
 		madeProgress = true
 	}
 
-	task.round++
 	return madeProgress
 }
 
@@ -171,31 +171,45 @@ func (d *driverImpl) doCollect() bool {
 }
 
 func (d *driverImpl) doOneCollectTask(task *collectTask) bool {
-	if !d.allDataReady(task) {
-		return false
-	}
+	madeProgress := false
+	for i, port := range task.ports {
+		if task.portRounds[i] >= task.rounds {
+			continue
+		}
+		item := port.PeekIncoming()
+		if item == nil {
+			continue
+		}
 
-	//fmt.Printf("\033[31mCollect Task: %v\033[0m\n", task)
+		msg, ok := item.(*cgra.MoveMsg)
+		if !ok {
+			continue
+		}
+		if msg.Color != task.color {
+			continue
+		}
 
-	for i, port := range task.ports {
-		msg := port.RetrieveIncoming().(*cgra.MoveMsg)
-		task.data[task.round*task.stride+i] = msg.Data.First()
-		// in red
+		port.RetrieveIncoming()
+		dataIndex := task.portRounds[i]*task.stride + i
+		if dataIndex >= 0 && dataIndex < len(task.data) {
+			task.data[dataIndex] = msg.Data.First()
+		}
 
 		core.Trace("DataFlow",
 			"Behavior", "Collect",
 			slog.Float64("Time", float64(d.Engine.CurrentTime()*1e9)),
-			"Data", task.data[task.round*task.stride+i],
+			"Data", msg.Data.First(),
 			"Pred", msg.Data.Pred,
 			"Color", task.color,
 			"From", task.ports[i].Name(),
 			"To", "None",
 		)
-	}
 
-	task.round++
+		task.portRounds[i]++
+		madeProgress = true
+	}
 
-	return true
+	return madeProgress
 }
 
 func (*driverImpl) allDataReady(task *collectTask) bool {
@@ -298,11 +312,19 @@ type feedInTask struct {
 
 	stride int
 	color  int
-	round  int
+	rounds int
+
+	// Port-wise progress allows opportunistic feed-in and avoids global barriers.
+	portRounds []int
 }
 
 func (t *feedInTask) isFinished() bool {
-	return t.round >= len(t.data)/t.stride
+	for _, r := range t.portRounds {
+		if r < t.rounds {
+			return false
+		}
+	}
+	return true
 }
 
 func (d *driverImpl) FeedIn(
@@ -318,6 +340,8 @@ func (d *driverImpl) FeedIn(
 		stride:     stride,
 		color:      d.getColorIndex(color),
 	}
+	task.rounds = len(task.data) / task.stride
+	task.portRounds = make([]int, len(task.localPorts))
 
 	sideIndex := int(side)
 	d.feedInTasks[sideIndex] = append(d.feedInTasks[sideIndex], task)
@@ -360,11 +384,19 @@ type collectTask struct {
 	ports  []sim.Port
 	stride int
 	color  int
-	round  int
+	rounds int
+
+	// Port-wise progress allows opportunistic collect and avoids global barriers.
+	portRounds []int
 }
 
 func (t *collectTask) isFinished() bool {
-	return t.round >= len(t.data)/t.stride
+	for _, r := range t.portRounds {
+		if r < t.rounds {
+			return false
+		}
+	}
+	return true
 }
 
 func (d *driverImpl) Collect(
@@ -380,6 +412,8 @@ func (d *driverImpl) Collect(
 		stride: stride,
 		color:  d.getColorIndex(color),
 	}
+	task.rounds = len(task.data) / task.stride
+	task.portRounds = make([]int, len(task.ports))
 
 	sideIndex := int(side)
 	//fmt.Println(color)

api/driver_internal_test.go

@@ -165,7 +165,8 @@ var _ = Describe("Driver", func() {
 				remotePorts: remotePorts,
 				stride:      3,
 				color:       0, // R
-				round:       0,
+				rounds:      2,
+				portRounds:  []int{0, 0, 0},
 			},
 		}
 
@@ -203,22 +204,20 @@ var _ = Describe("Driver", func() {
 				ports:  ports,
 				stride: 3,
 				color:  0, // R
-				round:  0,
+				rounds: 2,
+				portRounds: []int{
+					0, 0, 0,
+				},
 			},
 		}
 
-		// Mock PeekIncoming and RetrieveIncoming for first round
-		// Note: allDataReady checks PeekIncoming multiple times (once per port),
-		// then doOneCollectTask calls RetrieveIncoming for each port
+		// Mock PeekIncoming and RetrieveIncoming for first round.
 		msg1 := cgra.MoveMsgBuilder{}.WithData(cgra.NewScalar(1)).Build()
 		msg2 := cgra.MoveMsgBuilder{}.WithData(cgra.NewScalar(2)).Build()
 		msg3 := cgra.MoveMsgBuilder{}.WithData(cgra.NewScalar(3)).Build()
-		// allDataReady will call PeekIncoming for all ports (at least once each)
-		// Since allDataReady may be called multiple times, we use AnyTimes
-		localPort1.EXPECT().PeekIncoming().Return(msg1).AnyTimes()
-		localPort2.EXPECT().PeekIncoming().Return(msg2).AnyTimes()
-		localPort3.EXPECT().PeekIncoming().Return(msg3).AnyTimes()
-		// Then doOneCollectTask will call RetrieveIncoming for each port
+		localPort1.EXPECT().PeekIncoming().Return(msg1).Times(1)
+		localPort2.EXPECT().PeekIncoming().Return(msg2).Times(1)
+		localPort3.EXPECT().PeekIncoming().Return(msg3).Times(1)
 		localPort1.EXPECT().RetrieveIncoming().Return(msg1).Times(1)
 		localPort2.EXPECT().RetrieveIncoming().Return(msg2).Times(1)
 		localPort3.EXPECT().RetrieveIncoming().Return(msg3).Times(1)
@@ -229,9 +228,9 @@ var _ = Describe("Driver", func() {
 		msg4 := cgra.MoveMsgBuilder{}.WithData(cgra.NewScalar(4)).Build()
 		msg5 := cgra.MoveMsgBuilder{}.WithData(cgra.NewScalar(5)).Build()
 		msg6 := cgra.MoveMsgBuilder{}.WithData(cgra.NewScalar(6)).Build()
-		localPort1.EXPECT().PeekIncoming().Return(msg4).AnyTimes()
-		localPort2.EXPECT().PeekIncoming().Return(msg5).AnyTimes()
-		localPort3.EXPECT().PeekIncoming().Return(msg6).AnyTimes()
+		localPort1.EXPECT().PeekIncoming().Return(msg4).Times(1)
+		localPort2.EXPECT().PeekIncoming().Return(msg5).Times(1)
+		localPort3.EXPECT().PeekIncoming().Return(msg6).Times(1)
 		localPort1.EXPECT().RetrieveIncoming().Return(msg4).Times(1)
 		localPort2.EXPECT().RetrieveIncoming().Return(msg5).Times(1)
 		localPort3.EXPECT().RetrieveIncoming().Return(msg6).Times(1)

core/core.go

@@ -85,11 +85,11 @@ func (c *Core) MapProgram(program interface{}, x int, y int) {
 
 // Tick runs the program for one cycle.
 func (c *Core) Tick() (madeProgress bool) {
-	madeProgress = c.doSend() || madeProgress
+	madeProgress = c.doRecv() || madeProgress
 	// madeProgress = c.AlwaysPart() || madeProgress
 	// madeProgress = c.emu.runRoutingRules(&c.state) || madeProgress
 	madeProgress = c.runProgram() || madeProgress
-	madeProgress = c.doRecv() || madeProgress
+	madeProgress = c.doSend() || madeProgress
 	return madeProgress
 }