Skip to content

refactor: rpcqueue #433

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
0xFA11 merged 10 commits into from
Jan 13, 2021
Merged

refactor: rpcqueue #433

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
cadac6a
MTT-327
NoelStephensUnity Jan 7, 2021
d130e99
Merge branch 'develop' into feature/rpcqueuerevisited
NoelStephensUnity Jan 7, 2021
00eda77
MTT-367 MTT-327
NoelStephensUnity Jan 8, 2021
e1dc42f
Minor update
NoelStephensUnity Jan 8, 2021
320d90f
Merge branch 'develop' into feature/rpcqueuerevisited
0xFA11 Jan 9, 2021
dbeb061
Merge branch 'develop' into feature/rpcqueuerevisited
NoelStephensUnity Jan 12, 2021
dde21b4
fix: for renamed RPCQueueManager and camelCase
NoelStephensUnity Jan 12, 2021
af6b2d4
fix: more naming update fixes
NoelStephensUnity Jan 12, 2021
7b2040f
Merge branch 'develop' into feature/rpcqueuerevisited
0xFA11 Jan 13, 2021
850c991
fix: small fixes
0xFA11 Jan 13, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 13 additions & 13 deletions com.unity.multiplayer.mlapi/Runtime/Core/MessageBatcher.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,15 +78,15 @@ private int PopLength(in BitStream messageStream)
/// <param name="list">the list to fill</param>
private static void FillTargetList(in FrameQueueItem item, ref ulong[] list)
{
switch (item.QueueItemType)
switch (item.queueItemType)
{
case RPCQueueManager.QueueItemType.ServerRpc:
case RpcQueueContainer.QueueItemType.ServerRpc:
Array.Resize(ref list, 1);
list[0] = item.NetworkId;
list[0] = item.networkId;
break;
case RPCQueueManager.QueueItemType.ClientRpc:
case RpcQueueContainer.QueueItemType.ClientRpc:
// copy the list
list = item.ClientIds.ToArray();
list = item.clientIds.ToArray();
break;
default:
break;
Expand Down Expand Up @@ -120,32 +120,32 @@ public void QueueItem(in FrameQueueItem item)
Writer.WriteBit(false); // Encrypted
Writer.WriteBit(false); // Authenticated

switch (item.QueueItemType)
switch (item.queueItemType)
{
// 6 bits are used for the message type, which is an MLAPIConstants
case RPCQueueManager.QueueItemType.ServerRpc:
case RpcQueueContainer.QueueItemType.ServerRpc:
Writer.WriteBits(MLAPIConstants.MLAPI_SERVER_RPC, 6); // MessageType
break;
case RPCQueueManager.QueueItemType.ClientRpc:
case RpcQueueContainer.QueueItemType.ClientRpc:
Writer.WriteBits(MLAPIConstants.MLAPI_CLIENT_RPC, 6); // MessageType
break;
}
}

// write the amounts of bytes that are coming up
PushLength(item.MessageData.Count, ref Writer);
PushLength(item.messageData.Count, ref Writer);

// write the message to send
// todo: is there a faster alternative to .ToArray()
Writer.WriteBytes(item.MessageData.ToArray(), item.MessageData.Count);
Writer.WriteBytes(item.messageData.ToArray(), item.messageData.Count);

ProfilerStatManager.bytesSent.Record((int)item.MessageData.Count);
ProfilerStatManager.bytesSent.Record((int)item.messageData.Count);
ProfilerStatManager.rpcsSent.Record();
}
}

public delegate void SendCallbackType(ulong clientId, SendStream messageStream);
public delegate void ReceiveCallbackType(BitStream messageStream, MLAPI.RPCQueueManager.QueueItemType messageType, ulong clientId, float time);
public delegate void ReceiveCallbackType(BitStream messageStream, MLAPI.RpcQueueContainer.QueueItemType messageType, ulong clientId, float time);

/// <summary>
/// SendItems
Expand Down Expand Up @@ -185,7 +185,7 @@ public void SendItems(int threshold, SendCallbackType sendCallback)
/// <param name="messageType"> the message type to pass back to callback</param>
/// <param name="clientId"> the clientId to pass back to callback</param>
/// <param name="receiveTime"> the packet receive time to pass back to callback</param>
public int ReceiveItems(in BitStream messageStream, ReceiveCallbackType receiveCallback, MLAPI.RPCQueueManager.QueueItemType messageType, ulong clientId, float receiveTime)
public int ReceiveItems(in BitStream messageStream, ReceiveCallbackType receiveCallback, MLAPI.RpcQueueContainer.QueueItemType messageType, ulong clientId, float receiveTime)
{
do
{
Expand Down
4 changes: 4 additions & 0 deletions com.unity.multiplayer.mlapi/Runtime/Core/NetworkUpdateManager.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,7 @@
/// About the Network Update Loop
/// The NetworkUpdateEngine is a temporary solution for the network update loop implementation.
/// This will be revised with a more robust and modular implementation in the near future.
Comment thread
NoelStephensUnity marked this conversation as resolved.

using System;
using System.Text;
using UnityEngine;
Expand Down
15 changes: 5 additions & 10 deletions com.unity.multiplayer.mlapi/Runtime/Core/NetworkedBehaviour.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,9 @@ internal enum NExec
Client = 2
}

/// <summary>
/// This is a temporary solution for channel names and the below hardcoded value might not be mandatory in the future.
/// </summary>
Comment thread
NoelStephensUnity marked this conversation as resolved.
private const string StandardRpc_ChannelName = "STDRPC";

#pragma warning disable 414
Expand All @@ -49,12 +52,8 @@ internal enum NExec
internal BitWriter BeginSendServerRpc(ServerRpcSendParams sendParams, bool isReliable)
{
var rpcQueueMananger = NetworkingManager.Singleton.RpcQueueManager;
Comment thread
NoelStephensUnity marked this conversation as resolved.
Outdated
if (rpcQueueMananger == null)
{
return null;
}

var writer = rpcQueueMananger.BeginAddQueueItemToOutboundFrame(RPCQueueManager.QueueItemType.ServerRpc, Time.realtimeSinceStartup, StandardRpc_ChannelName, 0, NetworkingManager.Singleton.ServerClientId, null);
var writer = rpcQueueMananger.BeginAddQueueItemToOutboundFrame(RpcQueueContainer.QueueItemType.ServerRpc, Time.realtimeSinceStartup, StandardRpc_ChannelName, 0, NetworkingManager.Singleton.ServerClientId, null);
writer.WriteUInt64Packed(NetworkId); // NetworkObjectId
writer.WriteUInt16Packed(GetBehaviourId()); // NetworkBehaviourId
return writer;
Expand All @@ -74,12 +73,8 @@ internal BitWriter BeginSendClientRpc(ClientRpcSendParams sendParams, bool isRel
{
//This will start a new queue item entry and will then return the writer to the current frame's stream
var rpcQueueMananger = NetworkingManager.Singleton.RpcQueueManager;
if (rpcQueueMananger == null)
{
return null;
}

var writer = rpcQueueMananger.BeginAddQueueItemToOutboundFrame(RPCQueueManager.QueueItemType.ClientRpc, Time.realtimeSinceStartup, StandardRpc_ChannelName, 0, NetworkId, sendParams.TargetClientIds ?? NetworkingManager.Singleton.ConnectedClientsList.Select(c => c.ClientId).ToArray());
var writer = rpcQueueMananger.BeginAddQueueItemToOutboundFrame(RpcQueueContainer.QueueItemType.ClientRpc, Time.realtimeSinceStartup, StandardRpc_ChannelName, 0, NetworkId, sendParams.TargetClientIds ?? NetworkingManager.Singleton.ConnectedClientsList.Select(c => c.ClientId).ToArray());
writer.WriteUInt64Packed(NetworkId); // NetworkObjectId
writer.WriteUInt16Packed(GetBehaviourId()); // NetworkBehaviourId
return writer;
Expand Down
28 changes: 14 additions & 14 deletions com.unity.multiplayer.mlapi/Runtime/Core/NetworkingManager.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,7 @@ public class NetworkingManager : MonoBehaviour
[HideInInspector]
public bool LoopbackEnabled;

public RPCQueueManager RpcQueueManager { get; private set; }
public RpcQueueContainer RpcQueueManager { get; private set; }

/// <summary>
/// A synchronized time, represents the time in seconds since the server application started. Is replicated across all clients
Expand Down Expand Up @@ -689,10 +689,10 @@ private void Shutdown()

private void Awake()
{
RpcQueueManager = new RPCQueueManager(LoopbackEnabled);
RpcQueueManager = new RpcQueueContainer(LoopbackEnabled);
//Note: Since frame history is not being used, this is set to 0
//To test frame history, increase the number to (n) where n > 0
RpcQueueManager?.Initialize(0);
RpcQueueManager.Initialize(0);

NetworkUpdateManager.RegisterNetworkUpdateAction(NetworkPreUpdate, NetworkUpdateManager.NetworkUpdateStages.PREUPDATE);
NetworkUpdateManager.RegisterNetworkUpdateAction(NetworkFixedUpdate, NetworkUpdateManager.NetworkUpdateStages.FIXEDUPDATE);
Expand Down Expand Up @@ -766,7 +766,7 @@ private void NetworkPreUpdate()

private void NetworkFixedUpdate()
{
RpcQueueManager?.ProcessAndFlushRPCQueue(RPCQueueManager.RPCQueueProcessingTypes.Receive);
RpcQueueManager?.ProcessAndFlushRPCQueue(RpcQueueContainer.RPCQueueProcessingTypes.Receive);
}

/// <summary>
Expand Down Expand Up @@ -856,7 +856,7 @@ private void NetworkUpdate()
/// </summary>
private void NetworkLateUpdate()
{
RpcQueueManager?.ProcessAndFlushRPCQueue(RPCQueueManager.RPCQueueProcessingTypes.Send);
RpcQueueManager?.ProcessAndFlushRPCQueue(RpcQueueContainer.RPCQueueProcessingTypes.Send);
Comment thread
NoelStephensUnity marked this conversation as resolved.
Outdated
}

internal void UpdateNetworkTime(ulong clientId, float netTime, float receiveTime, bool warp = false)
Expand Down Expand Up @@ -1165,7 +1165,7 @@ internal void HandleIncomingData(ulong clientId, string channelName, ArraySegmen
{
if (IsServer)
{
batcher.ReceiveItems(messageStream, ReceiveCallback, RPCQueueManager.QueueItemType.ServerRpc, clientId, receiveTime);
batcher.ReceiveItems(messageStream, ReceiveCallback, RpcQueueContainer.QueueItemType.ServerRpc, clientId, receiveTime);
}
ProfilerStatManager.rpcBatchesRcvd.Record();
break;
Expand All @@ -1174,7 +1174,7 @@ internal void HandleIncomingData(ulong clientId, string channelName, ArraySegmen
{
if (IsClient)
{
batcher.ReceiveItems(messageStream, ReceiveCallback, RPCQueueManager.QueueItemType.ClientRpc, clientId, receiveTime);
batcher.ReceiveItems(messageStream, ReceiveCallback, RpcQueueContainer.QueueItemType.ClientRpc, clientId, receiveTime);
}
ProfilerStatManager.rpcBatchesRcvd.Record();
break;
Expand All @@ -1192,10 +1192,10 @@ internal void HandleIncomingData(ulong clientId, string channelName, ArraySegmen
#endif
}

private static void ReceiveCallback(BitStream messageStream, MLAPI.RPCQueueManager.QueueItemType messageType, ulong clientId, float receiveTime)
private static void ReceiveCallback(BitStream messageStream, MLAPI.RpcQueueContainer.QueueItemType messageType, ulong clientId, float receiveTime)
{
#if DEVELOPMENT_BUILD || UNITY_EDITOR
if (messageType == RPCQueueManager.QueueItemType.ServerRpc)
if (messageType == RpcQueueContainer.QueueItemType.ServerRpc)
{
s_MLAPIServerSTDRPCQueued.Begin();
}
Expand All @@ -1206,7 +1206,7 @@ private static void ReceiveCallback(BitStream messageStream, MLAPI.RPCQueueManag
#endif
InternalMessageHandler.RPCReceiveQueueItem(clientId, messageStream, receiveTime, messageType);
#if DEVELOPMENT_BUILD || UNITY_EDITOR
if (messageType == RPCQueueManager.QueueItemType.ServerRpc)
if (messageType == RpcQueueContainer.QueueItemType.ServerRpc)
{
s_MLAPIServerSTDRPCQueued.End();
}
Expand All @@ -1227,9 +1227,9 @@ public static void InvokeRpc(FrameQueueItem queueItem)
#if DEVELOPMENT_BUILD || UNITY_EDITOR
s_InvokeRPC.Begin();
#endif
var networkObjectId = queueItem.StreamReader.ReadUInt64Packed();
var networkBehaviourId = queueItem.StreamReader.ReadUInt16Packed();
var networkMethodId = queueItem.StreamReader.ReadUInt32Packed();
var networkObjectId = queueItem.streamReader.ReadUInt64Packed();
var networkBehaviourId = queueItem.streamReader.ReadUInt16Packed();
var networkMethodId = queueItem.streamReader.ReadUInt32Packed();

if (__ntable.ContainsKey(networkMethodId))
{
Expand All @@ -1239,7 +1239,7 @@ public static void InvokeRpc(FrameQueueItem queueItem)
var networkBehaviour = networkObject.GetBehaviourAtOrderIndex(networkBehaviourId);
if (ReferenceEquals(networkBehaviour, null)) return;

__ntable[networkMethodId](networkBehaviour, queueItem.StreamReader, queueItem.NetworkId);
__ntable[networkMethodId](networkBehaviour, queueItem.streamReader, queueItem.networkId);
}

#if DEVELOPMENT_BUILD || UNITY_EDITOR
Expand Down
24 changes: 13 additions & 11 deletions com.unity.multiplayer.mlapi/Runtime/Core/RPCQueue/FrameQueueItem.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,19 +8,21 @@ namespace MLAPI
/// FrameQueueItem
/// Container structure for RPCs written to the Queue Frame
/// Used for both Inbound and Outbound RPCs
/// NOTE: This could eventually become obsolete as other systems mature
/// NOTE: This structure will change in the near future and is in a state of flux.
/// This will include removing specific properties or changing property types (i.e. Channel could become a byte value)
Comment thread
NoelStephensUnity marked this conversation as resolved.
/// </summary>
public struct FrameQueueItem
{
public RPCQueueManager.QueueItemType QueueItemType;
public SecuritySendFlags SendFlags;
public ulong NetworkId; //Sender's network Identifier
public string Channel;
public ulong[] ClientIds; //Server invoked Client RPCs only
public long StreamSize;
public PooledBitWriter StreamWriter;
public PooledBitReader StreamReader;
public PooledBitStream ItemStream;
public ArraySegment<byte> MessageData;
public RpcQueueContainer.QueueItemType queueItemType;
public SecuritySendFlags sendFlags;
public ulong networkId; //Sender's network Identifier
public string channel;
public ulong[] clientIds; //Server invoked Client RPCs only
public long streamSize;
public float timeStamp;
public PooledBitWriter streamWriter;
public PooledBitReader streamReader;
public PooledBitStream itemStream;
public ArraySegment<byte> messageData;
}
}
Loading