| | | 1 | | using System; |
| | | 2 | | using UnityEngine; |
| | | 3 | | |
| | | 4 | | // The mailbox handles the communication latencies for agent-to-agent communication. The mailbox |
| | | 5 | | // owns the message queue and releases messages to be delivered after a set latency. |
| | | 6 | | public class Mailbox { |
| | 0 | 7 | | private static readonly Configs.LinkConfig _defaultLinkConfig = new Configs.LinkConfig { |
| | | 8 | | PacketDeliveryRatio = 1f, |
| | | 9 | | }; |
| | | 10 | | |
| | | 11 | | // Message queue. |
| | 0 | 12 | | private readonly PriorityQueue<PendingMessage> _messageQueue = |
| | | 13 | | new PriorityQueue<PendingMessage>(); |
| | | 14 | | |
| | | 15 | | // Enqueue a message to be delivered after a set latency. |
| | 0 | 16 | | public void Send(Message message) { |
| | 0 | 17 | | if (message == null) { |
| | 0 | 18 | | return; |
| | | 19 | | } |
| | | 20 | | |
| | 0 | 21 | | if (CommsManager.Instance.ContainsNode(message.Receiver)) { |
| | 0 | 22 | | Configs.LinkConfig config = GetLinkConfig(message); |
| | | 23 | | |
| | | 24 | | // TODO(Joseph0120): Set the packet delivery ratio to config.PacketDeliveryRatio. |
| | 0 | 25 | | float packetDeliveryRatio = Mathf.Clamp01(1); |
| | 0 | 26 | | if (UnityEngine.Random.value >= packetDeliveryRatio) { |
| | 0 | 27 | | return; |
| | | 28 | | } |
| | | 29 | | |
| | 0 | 30 | | float latency = config.LatencySeconds; |
| | 0 | 31 | | float jitter = Utilities.SampleStandardNormal() * config.LatencyStdSeconds; |
| | 0 | 32 | | float totalLatency = Math.Max(0f, latency + jitter); |
| | 0 | 33 | | float deliverAt = SimManager.Instance.ElapsedTime + totalLatency; |
| | | 34 | | |
| | | 35 | | // Deliver the message immediately if it is already due. |
| | 0 | 36 | | if (deliverAt <= SimManager.Instance.ElapsedTime) { |
| | 0 | 37 | | if (CommsManager.Instance.ContainsNode(message.Receiver)) { |
| | 0 | 38 | | message.Receiver.Receive(message); |
| | 0 | 39 | | } |
| | 0 | 40 | | } else { |
| | 0 | 41 | | var pendingMessage = new PendingMessage(message, deliverAt); |
| | 0 | 42 | | _messageQueue.Enqueue(pendingMessage, pendingMessage.DeliverAt); |
| | 0 | 43 | | } |
| | 0 | 44 | | } |
| | 0 | 45 | | } |
| | | 46 | | |
| | | 47 | | // Deliver should be called at every fixed update to check for due messages to be delivered. |
| | 77 | 48 | | public void Deliver() { |
| | 77 | 49 | | while (!_messageQueue.IsEmpty() && |
| | 0 | 50 | | _messageQueue.Peek().DeliverAt <= SimManager.Instance.ElapsedTime) { |
| | 0 | 51 | | PendingMessage pendingMessage = _messageQueue.Dequeue(); |
| | 0 | 52 | | if (CommsManager.Instance.ContainsNode(pendingMessage.Receiver)) { |
| | 0 | 53 | | pendingMessage.Receiver.Receive(pendingMessage.Message); |
| | 0 | 54 | | } |
| | 0 | 55 | | } |
| | 77 | 56 | | } |
| | | 57 | | |
| | | 58 | | // Clear all pending messages. |
| | 23 | 59 | | public void Clear() { |
| | 23 | 60 | | _messageQueue.Clear(); |
| | 23 | 61 | | } |
| | | 62 | | |
| | 0 | 63 | | private Configs.LinkConfig GetLinkConfig(Message message) { |
| | 0 | 64 | | Configs.CommunicationConfig communicationConfig = |
| | | 65 | | SimManager.Instance.SimulationConfig?.CommunicationConfig; |
| | 0 | 66 | | if (communicationConfig == null) { |
| | 0 | 67 | | return _defaultLinkConfig; |
| | | 68 | | } |
| | | 69 | | |
| | 0 | 70 | | Configs.AgentType senderType = message.Sender.EndpointType; |
| | 0 | 71 | | Configs.AgentType receiverType = message.Receiver.EndpointType; |
| | 0 | 72 | | foreach (Configs.LinkOverride linkOverride in communicationConfig.LinkOverrides) { |
| | 0 | 73 | | if (linkOverride.From == senderType && linkOverride.To == receiverType) { |
| | 0 | 74 | | return linkOverride.LinkConfig; |
| | | 75 | | } |
| | 0 | 76 | | } |
| | 0 | 77 | | return communicationConfig.LinkConfig ?? _defaultLinkConfig; |
| | 0 | 78 | | } |
| | | 79 | | } |