VirtualLiveLab · shosatojp · Apr 8, 2021 · Apr 6, 2021 · Apr 6, 2021 · Apr 6, 2021
diff --git a/LoopLibrary/BaseLoop.cs b/LoopLibrary/BaseLoop.cs
@@ -3,16 +3,20 @@
 using System.Threading.Tasks;
 using DebugPrintLibrary;
 using EventServerCore;
+using System.Linq;
+using System.Collections.Generic;

 namespace LoopLibrary
 {
    public abstract class BaseLoop<T>
    {
        public readonly ulong id;
-        private readonly int _interval;
+        private int _interval;
        private readonly CancellationTokenSource _cts = new CancellationTokenSource();
        private int _count;
        public CancellationTokenSource Cts => _cts;
+        private readonly int _maxTimespans = 100;
+        private Queue<double> _timespans = new Queue<double>();

        protected BaseLoop(int interval, ulong id = 1)
        {
@@ -34,9 +38,19 @@ private async Task<Result<T>> Loop(CancellationToken token)
                {
                    var delay = Task.Delay(_interval, token);

-                    await Update(_count);
-                    _count++;
-
+                    var start = DateTime.Now;
+                    await Update(_count++);
+                    var timespan = DateTime.Now - start;
+
+                    lock (_timespans)
+                    {
+                        _timespans.Enqueue(timespan.TotalMilliseconds);
+                        if (_timespans.Count > _maxTimespans)
+                        {
+                            _timespans.Dequeue();
+                        }
+                    }
+
                    await delay;
                }
            }
@@ -57,7 +71,7 @@ private async Task<Result<T>> Loop(CancellationToken token)
            }
        }

-        protected virtual void Start(){}
+        protected virtual void Start() { }

        protected abstract Task Update(int count);

@@ -68,12 +82,32 @@ protected virtual void OnCancel()

        public void Done(T result)
        {
-            throw new OperationCompletedException<Result<T>>(new Result<T>(true ,result));
+            throw new OperationCompletedException<Result<T>>(new Result<T>(true, result));
        }

        public void Cancel()
        {
            throw new OperationCanceledException();
        }
+
+        public double GetFps()
+        {
+            lock (_timespans)
+            {
+                if (_timespans.Count > 0)
+                {
+                    return 1000 / _timespans.Average();
+                }
+                else
+                {
+                    return 0;
+                }
+            }
+        }
+
+        public void SetInterval(int ms)
+        {
+            _interval = ms;
+        }
    }
 }
diff --git a/StreamClient/StreamClient.csproj b/StreamClient/StreamClient.csproj
@@ -4,6 +4,7 @@
        <OutputType>Exe</OutputType>
        <TargetFramework>netcoreapp3.1</TargetFramework>
        <Nullable>enable</Nullable>
+        <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
    </PropertyGroup>

    <ItemGroup>

diff --git a/StreamServer.Test/PacketTest.cs b/StreamServer.Test/PacketTest.cs
@@ -1,4 +1,5 @@
 using System.Collections.Generic;
+using System.Linq;
 using CommonLibrary;
 using Xunit;
 using Xunit.Abstractions;
@@ -20,7 +21,8 @@ public void PacketTest1()
            var packet = new MinimumAvatarPacket(1, new Vector3(2, 5, 6), 10,
                new Vector4(2, 4, 3, 120), 0);
            Assert.True(packet.CheckRange(), "packet.CheckRange()");
-            var buff = Utility.PacketsToBuffer(new List<MinimumAvatarPacket> {packet});
+            var packets = new List<MinimumAvatarPacket> {packet};
+            var buff = Utility.PacketsToBuffer(ref packets);
            var decodedPacket = Utility.BufferToPackets(buff);
            Assert.Equal(packet.PaketId, decodedPacket![0].PaketId);
            Assert.Equal(packet.Position, decodedPacket![0].Position);
@@ -36,7 +38,8 @@ public void PacketTest2()
            var packet2 = new MinimumAvatarPacket(2, new Vector3((short) 7.5f, (short) 2.6f, (short) 9.2f), 99,
                new Vector4(1, 9, 6, -1), 0);
            Assert.True(packet1.CheckRange() && packet2.CheckRange(), "packet1.CheckRange() && packet2.CheckRange()");
-            var buff = Utility.PacketsToBuffer(new List<MinimumAvatarPacket> {packet1, packet2});
+            var packets = new List<MinimumAvatarPacket> {packet1, packet2};
+            var buff = Utility.PacketsToBuffer(ref packets);
            var decodedPacket = Utility.BufferToPackets(buff);

            Assert.Equal(packet1.PaketId, decodedPacket![0].PaketId);
@@ -56,7 +59,8 @@ public void PacketTest3()
            var packet = new MinimumAvatarPacket(3, new Vector3(), 0,
                new Vector4(), 0);
            Assert.True(packet.CheckRange(), "packet.CheckRange()");
-            var buff = Utility.PacketsToBuffer(new List<MinimumAvatarPacket> {packet});
+            var packets = new List<MinimumAvatarPacket> {packet};
+            var buff = Utility.PacketsToBuffer(ref packets);
            var decodedPacket = Utility.BufferToPackets(buff);
            Assert.Equal(packet.PaketId, decodedPacket![0].PaketId);
            Assert.Equal(packet.Position, decodedPacket![0].Position);
@@ -77,7 +81,7 @@ public static void PacketTest4()
                Assert.True(packet.CheckRange(), $"packet.CheckRange({i})");
            }

-            var buffs = Utility.PacketsToBuffers(packets);
+            var buffs = Utility.PacketsToBuffers(ref packets).ToList();

            const int nSize = 29;
            for (int i = 0; i < buffs.Count; ++i)

diff --git a/StreamServer/Source/StreamServer/Entry.cs b/StreamServer/Source/StreamServer/Entry.cs
@@ -11,8 +11,8 @@ static void Main(string[] args)
        {
            UdpClient udpClient = new UdpClient(5577);
            var input = new InputLoop(udpClient, 2, 1);
-            var output = new OutputLoop(udpClient, 1000, 2);
-            var statusCheck = new StatusCheckLoop(1000, 3);
+            var output = new OutputLoop(udpClient, 10, 2);
+            var statusCheck = new StatusCheckLoop(1000, 3, input, output);
            input.Run();
            output.Run();
            statusCheck.Run();
@@ -28,6 +28,19 @@ static void Main(string[] args)
                    Thread.Sleep(100);
                    break;
                }
+                else
+                {
+                    try
+                    {
+                        var interval = Int32.Parse(line);
+                        output.SetInterval(interval);
+                        Printer.PrintDbg($"Set output interval to {interval}");
+                    }
+                    catch (System.Exception)
+                    {
+                        continue;
+                    }
+                }
            }
        }
    }

diff --git a/StreamServer/Source/StreamServer/InputLoop.cs b/StreamServer/Source/StreamServer/InputLoop.cs
@@ -18,7 +18,7 @@ public InputLoop(UdpClient udpClient, int interval, ulong id)
        {
            udp = udpClient;
        }
-        
+
        protected override void Start()
        {
            IPEndPoint localEndPoint = (IPEndPoint)udp.Client.LocalEndPoint;
@@ -32,13 +32,12 @@ protected override async Task Update(int count)
            {
                try
                {
-                    UdpReceiveResult res;
-                    res = await udp.ReceiveAsync();
-                    var process = PacketProcessor.Process(res);
-                    tasks.Add(process);
-                } catch (SocketException e)
+                    UdpReceiveResult res = await udp.ReceiveAsync();
+                    tasks.Add(Task.Run(() => PacketProcessor.Process(res)));
+                }
+                catch (SocketException e)
                {
-                    if (e.ErrorCode != 10054) //Client Disconnected.
+                    if (e.ErrorCode != 10054) // Client Disconnected.
                        Printer.PrintDbg(e, id);
                }
            }

diff --git a/StreamServer/Source/StreamServer/ModelManager.cs b/StreamServer/Source/StreamServer/ModelManager.cs
@@ -5,10 +5,9 @@ namespace StreamServer
 {
    public class ModelManager
    {
-        private ModelManager(){}
+        private ModelManager() { }
        private static ModelManager? _instance;
        public static ModelManager Instance => _instance ??= new ModelManager();
-
        public readonly ConcurrentDictionary<ulong, User> Users = new ConcurrentDictionary<ulong, User>();
    }
 }
diff --git a/StreamServer/Source/StreamServer/OutputLoop.cs b/StreamServer/Source/StreamServer/OutputLoop.cs
@@ -3,7 +3,9 @@
 using System.Net;
 using System.Net.Sockets;
 using System.Threading.Tasks;
+using System.Linq;
 using CommonLibrary;
+using CommonLibrary.ExtensionMethod;
 using DebugPrintLibrary;
 using EventServerCore;
 using LoopLibrary;
@@ -20,7 +22,7 @@ public OutputLoop(UdpClient udpClient, int interval, ulong id)
        {
            udp = udpClient;
        }
-        
+
        protected override void Start()
        {
            var localEndPoint = udp.Client.LocalEndPoint as IPEndPoint;
@@ -35,11 +37,15 @@ protected override async Task Update(int count)
            List<User> users = new List<User>();
            foreach (var kvp in ModelManager.Instance.Users)
            {
-                if(kvp.Value == null) continue;
+                if (kvp.Value == null)
+                    continue;
                var user = kvp.Value;
+
                MinimumAvatarPacket? packet = user.CurrentPacket;
                {
-                    if (user.IsConnected && packet != null && DateTime.Now - user.DateTimeBox!.LastUpdated > new TimeSpan(0, 0, 1))
+                    // 最終パケットが１秒以上前だったら切断したものとする
+                    if (user.IsConnected && packet != null
+                        && DateTime.Now - user.DateTimeBox!.LastUpdated > new TimeSpan(0, 0, 1))
                    {
 #if DEBUG
                        Printer.PrintDbg($"Disconnected: [{user.UserId.ToString()}] " +
@@ -50,15 +56,22 @@ protected override async Task Update(int count)
                        ModelManager.Instance.Users.TryRemove(kvp.Key, out var dummy);
                    }
                }
-                if (user.IsConnected) users.Add(user);
-                if (packet != null) packets.Add(packet);
+
+                // 有効なユーザとその最新のパケットをまとめる
+                if (user.IsConnected)
+                    users.Add(user);
+                if (packet != null)
+                    packets.Add(packet);
            }
-            List<Task> tasks = new List<Task>();
-            foreach (var user in users)
+
+            // 論理プロセッサ分に分割して並列処理
+            await Task.WhenAll(users.SplitInto(Environment.ProcessorCount).Select(userList => Task.Run(async () =>
            {
-                tasks.Add(PacketSender.Send(user, packets, udp));
-            }
-            await Task.WhenAll(tasks);
+                foreach (var user in userList)
+                {
+                    await PacketSender.Send(user, packets, udp);
+                }
+            })));
        }
    }
 }
diff --git a/StreamServer/Source/StreamServer/PacketProcessor.cs b/StreamServer/Source/StreamServer/PacketProcessor.cs
@@ -9,38 +9,28 @@ namespace StreamServer
 {
    public static class PacketProcessor
    {
-        public static async Task Process(UdpReceiveResult res)
+        public static void Process(UdpReceiveResult res)
        {
-            try
+            var packet = Utility.BufferToPacket(res.Buffer);
+            if (!ValidatePacket(packet))
            {
-                await Task.Run(() =>
-                {
-                    var packet = Utility.BufferToPacket(res.Buffer);
-                    if (!ValidatePacket(packet))
-                    {
-                        return;
-                    }
-                    var users = ModelManager.Instance.Users;
-
-                    User? user;
-                    if (!users.TryGetValue(packet.PaketId, out user))
-                    {
-                        // RemoteEndPointから来た最初のパケットの場合
-                        user = users[packet.PaketId] = new User(packet.PaketId);
-                        user.RemoteEndPoint = res.RemoteEndPoint;
-                        Printer.PrintDbg($"Connected: [{user.UserId.ToString()}] " +
-                                         $"({res.RemoteEndPoint.Address}: {res.RemoteEndPoint.Port.ToString()})");
-                    }
-
-                    user.CurrentPacket = packet;
-                    user.DateTimeBox = new DateTimeBox(DateTime.Now);
-                    user.IsConnected = true;
-                });
+                return;
            }
-            catch (Exception e)
+            var users = ModelManager.Instance.Users;
+
+            User? user;
+            if (!users.TryGetValue(packet.PaketId, out user))
            {
-                Printer.PrintDbg(e);
+                // RemoteEndPointから来た最初のパケットの場合
+                user = users[packet.PaketId] = new User(packet.PaketId);
+                user.RemoteEndPoint = res.RemoteEndPoint;
+                Printer.PrintDbg($"Connected: [{user.UserId.ToString()}] " +
+                                 $"({res.RemoteEndPoint.Address}: {res.RemoteEndPoint.Port.ToString()})");
            }
+
+            user.CurrentPacket = packet;
+            user.DateTimeBox = new DateTimeBox(DateTime.Now);
+            user.IsConnected = true;
        }

        private static bool ValidatePacket(in MinimumAvatarPacket? packet)

diff --git a/StreamServer/Source/StreamServer/PacketSender.cs b/StreamServer/Source/StreamServer/PacketSender.cs
@@ -1,4 +1,4 @@
-using System;
+using System;
 using System.Collections.Generic;
 using System.Linq;
 using System.Net.Sockets;
@@ -14,40 +14,28 @@ public static class PacketSender
    {
        public static async Task Send(User user, List<MinimumAvatarPacket> packets, UdpClient udp)
        {
-            await Task.Run(async () =>
+            var packetCopy = packets.ToList();
+            const int maxCount = 100;
+            if (user.CurrentPacket != null)
            {
-                var packetCopy = packets.ToList();
-                if (user.CurrentPacket != null)
-                {
-                    var selfPosition = user.CurrentPacket.Position;
-
-                    packetCopy.HeapSort((a, b) =>
-                    {
-                        var aSquare = Vector3.Square(a.Position, selfPosition);
-                        var bSquare = Vector3.Square(b.Position, selfPosition);
-                        var comp = aSquare < bSquare ? -1 : 1;
-                        return comp;
-                    });
-                }
+                var selfPosition = user.CurrentPacket.Position;

-                if (packetCopy.Count > 100)
-                    packetCopy = packetCopy.GetRange(0, 100);
-                var buffs = Utility.PacketsToBuffers(packetCopy);
-                var tasks = new List<Task>();
-                foreach (var buf in buffs)
+                packetCopy.HeapSort((a, b) =>
                {
-                    tasks.Add(udp.SendAsync(buf, buf.Length, user.RemoteEndPoint));
-                }
+                    var aSquare = Vector3.Square(a.Position, selfPosition);
+                    var bSquare = Vector3.Square(b.Position, selfPosition);
+                    var comp = aSquare < bSquare ? -1 : 1;
+                    return comp;
+                }, maxCount);
+            }

-                try
-                {
-                    await Task.WhenAll(tasks);
-                }
-                catch (Exception e)
-                {
-                    Printer.PrintDbg(e);
-                }
-            });
+            if (packetCopy.Count > maxCount)
+                packetCopy = packetCopy.GetRange(0, maxCount);
+            var buffs = Utility.PacketsToBuffers(ref packetCopy);
+            foreach (var buf in buffs)
+            {
+                await udp.SendAsync(buf, buf.Length, user.RemoteEndPoint);
+            }
        }
    }
 }