hjha-server/ServerCore/Fiber/ThreadPoolScheduler.cs

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading;
using ActorCore;
using MrWu.Debug;

namespace Server.Core
{
    public class ThreadPoolScheduler : IScheduler
    {
        private readonly List<Thread> _threads;

        private readonly ConcurrentQueue<int> idQueue = new ConcurrentQueue<int>();
        
        private readonly ActorManager _actorManager;
        
        public ThreadPoolScheduler(ActorManager actorManager)
        {
            _actorManager = actorManager;
            int threadCount = Environment.ProcessorCount;
            if (threadCount > 2)
            {
                threadCount = 2;
            }
            this._threads = new List<Thread>(threadCount);
            for (int i = 0; i < threadCount; i++)
            {
                Thread thread = new Thread(this.Loop);
                this._threads.Add(thread);
                thread.Start();
            }
        }

        private void Loop()
        {
            int count = 0;

            uint threadId = ThreadHelper.GetOSThreadId();
            Debug.Info($"ThreadStart ThreadPoolScheduler {threadId}");
            
            while (true)
            {
                if (count <= 0)
                {
                    Thread.Sleep(1);
                    count = this._actorManager.Count() / this._threads.Count + 1;
                }

                --count;
                if (this._actorManager.IsDisposed())
                {
                    return;
                }

                if (!this.idQueue.TryDequeue(out int id))
                {
                    Thread.Sleep(1);
                    continue;
                }

                IActor actor = this._actorManager.Get(id);
                if (actor == null)
                {
                    continue;
                }

                if (actor.IsDisposed)
                {
                    continue;
                }
                
                ActorManager.Current = actor;
                SynchronizationContext.SetSynchronizationContext(actor.ThreadSynchronizationContext);
                actor.IUpdate();
                actor.ILateUpdate();
                actor.ThreadSynchronizationContext.Update();
                SynchronizationContext.SetSynchronizationContext(null);
                ActorManager.Current = null;
                
                this.idQueue.Enqueue(id);
            }
        }

        public void Add(int actorTypeId)
        {
            this.idQueue.Enqueue(actorTypeId);
        }

        public void Dispose()
        {
            foreach (Thread thread in this._threads)
            {
                thread.Join();
            }
        }
    }
}