Simulating movement and communication that data

Having explored how to represent data as a Graph and navigate routes within it, the next question is how to effectively move along these routes. Several factors were considered in making this decision. One option was to delegate this computational task to the frontend, utilizing the user's CPU. This approach would create independent simulations on each frontend, leading to a lack of synchronization among different clients. Additionally, the potential for client disconnections raises concerns about losing entire simulation groups. Moreover, managing extensive metadata for route calculations and movement factors on the client side poses significant challenges. Given these considerations, I concluded that a more efficient approach involves centralizing the computation on the backend. This setup allows for the calculation and broadcasting of messages, ensuring consistent and reliable simulations across all clients.

Movement

So the first thing we doing once we perform an A* algorithm is create a journey is send to a channel. This can be done via this syntax

newJourney := dto.NewJourney(
	dto.WithId(inQueueJourney.Id),
	dto.WithStartingPointNode(inQueueJourney.StartingPointNode),
	dto.WithEndingPointNode(inQueueJourney.EndingPointNode),
	dto.WithCurrentPointNode(inQueueJourney.StartingPointNode),
	dto.WithDistance(inQueueJourney.Distance),
	dto.WithPath(*path),
	dto.WithTotalTripCost(totalCost),
	dto.WithStatus(inQueueJourney.Status),
	dto.WithDateCreate(time.Now().Format(time.RFC3339)),
)
newJourneyChannel <- newJourney

This is a global channel create in the main that the simulator listens and apply movement, the following code snippet

simulationTick := time.NewTicker(time.Millisecond * 400) // adjust duration as needed
	for {
		select {
		case newJourney := <-newJourneyChannel:
			activeJourneys = append(activeJourneys, newJourney)
			journeyTimeToCompleteMap[newJourney.Id] = time.Now()
		case <-simulationTick.C:
			// Simulate movement for all active journeys
			var movedJourneys []*dto.Journey // Assuming Journey is the type of the journeys
			for _, j := range activeJourneys {
				processJourney(j)
				message, err := json.Marshal(j)
				if err != nil {
					log.Println("Failed to marshal buffer:", err)
					observibility.GetMetrics().LogFailedSimulatedJourney()
					continue
				}
				err = websocketManager.Send(message)
				if err != nil {
					log.Println("Failed to send via websocket, error stack:", err)
					observibility.GetMetrics().LogFailedSimulatedJourney()
				} else if reachedEnd(j) {
					log.Printf("Journey with id %d has reach its end\n", j.Id)
					go updateStatusForJourney([]int32{j.Id}, string(dto.Finished))
					observibility.GetMetrics().LogTimeToFinishSimulation(time.Since(journeyTimeToCompleteMap[j.Id]))
					observibility.GetMetrics().LogJourneyDistance(j.Distance)
					observibility.GetMetrics().LogSuccessfulSimulatedJourney()
					// Skipping appending this journey as it has reached its end
					continue
				}
				movedJourneys = append(movedJourneys, j)
			}
			// Remove completed journeys or update list as needed...
			activeJourneys = movedJourneys
		}
	}

This code snippet outlines a simulation loop for managing and updating the status of journeys over time, using Go's concurrency features and time scheduling. Here's a structured breakdown:

1. Setting Up a Timer: A ticker named simulationTick is created, set to trigger every 400 milliseconds. This ticker will regularly initiate simulation updates for all active journeys.

2. Infinite Loop for Event Handling: The code enters an infinite loop using a select statement to handle two types of events:

   • New Journey Event: When a new journey is received on the newJourneyChannel, it is added to the activeJourneys slice.
   • Simulation Tick Event: Triggered by the simulationTick ticker. On each tick, the loop simulates movement for all active journeys by:
     • Processing each journey individually through a function call to processJourney(j).
     • Marshalling the journey object into a JSON string for potential logging or communication purposes. If marshalling fails, an error is logged, and the loop continues to the next journey.
     • Checking if a journey has reached its end using a function reachedEnd(j). If the journey has reached its end:
       • A log message is printed to indicate completion.
       • The journey's status is asynchronously updated to Finished using a goroutine call to updateStatusForJourney.
       • The journey is not added to the movedJourneys slice, effectively removing it from active tracking.
     • If the journey hasn't reached its end, it is added to the movedJourneys slice.
   • After iterating through all active journeys, the activeJourneys slice is updated to only include those that haven't reached their destination.

This loop continuously updates and manages the state of journeys, simulating their progress and dynamically adjusting the list of active journeys based on their completion status.

Movement Factor

Now we saw the gist of movement, do we move all journeys the same way in each cycle ? The answer is no there is nuance to it. Lets go through an example :

Imagine a vehicle on a path from Point A to Point B, then to Point C. Each update cycle, the system calculates how far the vehicle can move from A towards B based on the time allotted per cycle and the cost of the journey between A and B. If, after several cycles, the accumulated movement is enough to reach B, the vehicle's position updates to B, the journey cost is updated, and the process repeats for the next segment from B to C.

How do we achieve this behavior code wise ? The following code snippet does this

func updateVehiclePosition(j *dto.Journey) {
	key := fmt.Sprintf("%f,%f", j.CurrentPointNode.X, j.CurrentPointNode.Y)
	nextCurrentPair := j.Path.GetPair(key).Next()
	movementFactor := calculateMovementFactor(&j.CurrentPointNode, &nextCurrentPair.Value, 200)
	j.AccumulatedMovement += movementFactor

	if nextCurrentPair != nil && j.AccumulatedMovement >= 1 {
		moveVehicleToNextPoint(j, nextCurrentPair.Value)
		updateCost(j)
		j.AccumulatedMovement -= 1
	}
}

So we calculate the movement factor and added to the AccumulatedMovement, once that accrued value 1 or more we can move to the next point. Now how do we calculate movement factor

func calculateMovementFactor(currentPoint, nextPoint *dto.PointNode, cycleTimeMs int) float64 {
	edgeCost := getEdgeCost(currentPoint, nextPoint)

	// Assuming effortPerCycle is a predetermined constant that represents
	// how much 'effort' a vehicle can exert in one cycle.
	// This constant can be derived based on system averages or empirical data.
	effortPerCycle := calculateEffortPerCycle(cycleTimeMs)

	movementFactor := effortPerCycle / edgeCost

	// Ensuring the movement factor is not more than 1
	if movementFactor > 1 {
		movementFactor = 1
	}

	// Check if the fractional part of the movement factor is close to 1
	if movementFactor-math.Floor(movementFactor) >= 0.9 && movementFactor-math.Floor(movementFactor) <= 1.0 {
		movementFactor = math.Ceil(movementFactor)
	}

	return movementFactor
}

func getEdgeCost(currentNode, nextNode *dto.PointNode) float64 {
	return datastructures.RoadMapEdgeCostGraph[dto.Edge{From: *currentNode, To: *nextNode}]
}

func calculateEffortPerCycle(cycleTimeMs int) float64 {
	return 1.0 / float64(cycleTimeMs)
}

This code snippet calculates a "movement factor" that determines how far a vehicle can move towards the next point in its path during a single update cycle. The process involves three main steps:

1. Calculating Edge Cost (getEdgeCost): It retrieves the cost of moving from the current position to the next position. This cost is looked up from a pre-defined graph (RoadMapEdgeCostGraph) that maps the cost between every two connected points.

2. Calculating Effort Per Cycle (calculateEffortPerCycle): The function calculates how much effort a vehicle can exert in one update cycle, based on the cycle's duration in milliseconds. The effort is inversely proportional to the cycle time, implying that shorter cycles allow for greater effort within each cycle.

3. Determining Movement Factor (calculateMovementFactor): The movement factor is the ratio of the effort a vehicle can exert in one cycle to the cost of moving from the current point to the next point. This factor indicates how much of the distance to the next point the vehicle can cover in one cycle. To ensure logical movement progression, the movement factor is capped at 1, meaning the vehicle cannot move beyond the next point in a single cycle. Additionally, if the movement factor's fractional part is very close to 1 (between 0.9 and 1.0), it is rounded up to ensure the vehicle moves to the next point without getting stuck due to fractional calculations.

This mechanism allows for a dynamic calculation of vehicle progress in a simulation, factoring in the varying costs of movement across different segments of a route and the temporal granularity of the simulation cycles.

Sending to the frontend

Once the journey has been processed (which involves either updating or maintaining the current position), we then transmit this information to the frontend. How is this achieved? The only component capable of interfacing with the frontend is the api. Therefore, the simulator sends a message to the api using websockets, and the api, in turn, relays this message to the frontend (fe) also through websockets. The choice of websockets is deliberate; they are highly effective for real-time updates, which is our goal. Additionally, the occasional loss of a few messages is not critical; we simply proceed without them.