ECS Workflow usage

[spatialos]

ECS Workflow usage

[jamiebrynes7]

Hey @No3371!

How does a worker entity comes from? If it's being automatically created, when?

This is a little confusing as this term is overloaded.

• There's a worker entity that is local only to your worker and is never replicated to the SpatialOS Runtime. It's created by the WorkerSystem when this system is created. (I assume you were talking about this one?)
• There's also the concept of the worker entities that the SpatialOS Runtime creates. This currently isn't used in the GDK at all.

It seems Spatial is designed to be 1 worker per program (is it?), however, the doc stated that "you can run multiple workers in Unity Editor", at the same time, it seems there's no way to distinguish what worker is connected by only query for "OnConnected" component.

There's no inherent constraint that there is 1 worker per process (in the GDK for Unity at least). For example, in the FPS project we run 10 simulated players (each their own worker!) in a single process.

Each SpatialOS worker creates its own ECS world when it connects to the SpatialOS Runtime. So you should never get more than one entity with an OnConnected component (that is, unless you yourself create one). The presence of the ECS world indicates that this worker is connected. You can get some information about this worker by inspecting the WorkerSystem on that world.

This world contains a worker entity, which can be uniquely identified by the WorkerEntityTag component attached to it Similar to question#2, does this means the worker entity is a SingletonEntity (Relatively new stuff in Unity ECS)?

Currently the worker entity is not a SingletonEntity. We're in the process of exploring some of the new ECS features and how best to integrate them into the GDK!

Example project part of the doc still says Burst should be disabled.

Nice catch, I'll raise a ticket to fix this (if you have the link to where this is stated that would be helpful 😄)! Since 0.3.5, Burst has cross-compile support 🎉

[No3371]

Link: Example project part of the doc still says Burst should be disabled

You can get some information about this worker by inspecting the WorkerSystem on that world.

Nice, the WorkerType is exactly what i need.

It's created by the WorkerSystem when this system is created.

This clearly answers my question.

And also thanks for all the detailed explanations 😃

[No3371]

So I started from WorkerSystem, following reference and jumping around, it seems a WorkerSystem is from a WorkerInWorld, which is from WorkerConnector, does this means we still stick with MonoBehaviour to start the ECS workflow?

If so, I found this confusing. In the documentation, the MonoBehaviour workflow section is totally separated with and at the same level of ECS workflow section. The UX indicates they are 2 different approaches, and the ECS workflow documentation lack of info about how to start a worker, so i thought we have some equivalent way to start the whole system in ECS style.

[No3371]

Just put together a SystemBase that query for entities that contains worker type info and connect the workers. Based on your code, tested with local launch.

Usage
Create the WorkerConnectorEntity directly or with a GameObject with the WorkerConnectorEntityAuthoring component.

[System.Serializable]
public struct WorkerConnectorEntity : IComponentData
{
    public Unity.Collections.NativeString64 workerType;
}

[RequiresEntityConversion]
public class WorkerConnectorEntityAuthoring : MonoBehaviour, IConvertGameObjectToEntity
{
    public string workerType;

    public void Convert(Entity entity, EntityManager dstManager, GameObjectConversionSystem conversionSystem)
    {
        dstManager.AddComponentData<WorkerConnectorEntity>(entity, new WorkerConnectorEntity { workerType = new Unity.Collections.NativeString64(workerType) });
    }
}

using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using System.Threading;
using System.Threading.Tasks;
using Improbable.Gdk.Core;
using Improbable.Worker.CInterop;
using Unity.Collections;
using Unity.Entities;
using Unity.Mathematics;
using UnityEngine;
using UnityEngine.LowLevel;

public class WorkerConnectorSystem : SystemBase
{
    /// <summary>
    ///     The number of connection attempts before giving up.
    /// </summary>
    public int MaxConnectionAttempts = 3;

    /// <summary>
    ///     Represents a SpatialOS worker.
    /// </summary>
    /// <remarks>
    ///    Only safe to access after the connection has succeeded.
    /// </remarks>
    public WorkerInWorld Worker;
    private List<Action<Worker>> workerConnectedCallbacks = new List<Action<Worker>>();
    private static readonly SemaphoreSlim WorkerConnectionSemaphore = new SemaphoreSlim(1, 1);

    EntityQuery query;
    List<Unity.Entities.Entity> connecting;

    protected override void OnCreate()
    {
        query = GetEntityQuery(
            ComponentType.ReadOnly<Unity.Transforms.Translation>(),
            ComponentType.ReadOnly<WorkerConnectorEntity>()
        );
        connecting = new List<Unity.Entities.Entity>();
    }

    protected async override void OnUpdate()
    {
        if (wantsDispose)
        {
            Dispose();
            return;
        }

        var pendingNative = query.ToEntityArray(Allocator.Persistent);
        for (int i = 0; i < pendingNative.Length; i++)
        {
            var capture = pendingNative[i];
            if (connecting.Contains(capture)) continue;
            connecting.Add(capture);
            var wce = EntityManager.GetComponentData<WorkerConnectorEntity>(pendingNative[i]);
            Debug.Log("Found a connector entity: " + wce.workerType);
            var transalation = EntityManager.GetComponentData<Unity.Transforms.Translation>(pendingNative[i]);
            var connParams = CreateConnectionParameters(wce.workerType);

            var builder = new SpatialOSConnectionHandlerBuilder()
                .SetConnectionParameters(connParams);

            if (!Application.isEditor)
            {
                var initializer = new CommandLineConnectionFlowInitializer();
                switch (initializer.GetConnectionService())
                {
                    case ConnectionService.Receptionist:
                        builder.SetConnectionFlow(new ReceptionistFlow(CreateNewWorkerId(wce.workerType), initializer));
                        break;
                    case ConnectionService.Locator:
                        builder.SetConnectionFlow(new LocatorFlow(initializer));
                        break;
                    default:
                        throw new ArgumentOutOfRangeException();
                }
            }
            else
            {
                builder.SetConnectionFlow(new ReceptionistFlow(CreateNewWorkerId(wce.workerType)));
            }

            try
            {
                await Connect(builder, new ForwardingDispatcher(), transalation.Value).ConfigureAwait(true);
            }
            catch (Exception e)
            {
                Debug.Log("Failed to connect: " + wce.workerType + "(" + e);
            }
            finally
            {
                EntityManager.DestroyEntity(capture);
                connecting.Remove(capture);
            }
        }
        pendingNative.Dispose();
    }

    protected static string CreateNewWorkerId(NativeString64 workerType)
    {
        return $"{workerType}-{Guid.NewGuid().GetHashCode():x}";
    }

    protected ConnectionParameters CreateConnectionParameters(NativeString64 workerType, IConnectionParameterInitializer initializer = null)
    {
        var @params = new ConnectionParameters
        {
            WorkerType = workerType.ToString(),
            DefaultComponentVtable = new ComponentVtable(),
            Network =
            {
                ConnectionType = NetworkConnectionType.ModularKcp,
                ModularKcp =
                {
                    DownstreamCompression = new CompressionParameters(),
                    UpstreamCompression = new CompressionParameters(),
                }
            }
        };

        initializer?.Initialize(@params);

        return @params;
    }
    
    /// <summary>
    ///     Asynchronously connects a worker to the SpatialOS runtime.
    /// </summary>
    /// <param name="builder">Describes how to create a <see cref="IConnectionHandler"/> for this worker.</param>
    /// <param name="logger">The logger for the worker to use.</param>
    /// <returns></returns>
    protected async Task Connect(IConnectionHandlerBuilder builder, ILogDispatcher logger, float3 origin)
    {
        if (builder == null)
        {
            throw new ArgumentException("Builder cannot be null.", nameof(builder));
        }

        // Check that other workers have finished trying to connect before this one starts.
        // This prevents races on the workers starting and races on when we start ticking systems.
        await WorkerConnectionSemaphore.WaitAsync();
        try
        {
            // A check is needed for the case that play mode is exited before the semaphore was released.
            if (!Application.isPlaying)
            {
                return;
            }

            Worker = await ConnectWithRetries(builder, MaxConnectionAttempts, logger, builder.WorkerType, origin);

            Worker.OnDisconnect += OnDisconnected;

            if (!Application.isPlaying)
            {
                Dispose();
                throw new ConnectionFailedException("Editor application stopped",
                    ConnectionErrorReason.EditorApplicationStopped);
            }

            HandleWorkerConnectionEstablished();

            // Update PlayerLoop
            // Extracted Code From PlayerLoopUtils //
            // Create simulation system for the default group
            var simulationSystemGroup = Worker.World.GetOrCreateSystem<SimulationSystemGroup>();

            var systems = new List<ComponentSystemBase>();
            foreach (var system in Worker.World.Systems)
            {
                systems.Add(system);
            }

            var uniqueSystemTypes = new HashSet<Type>(systems.Select(s => s.GetType()));

            // Add systems to their groups, based on the [UpdateInGroup] attribute.
            for (var i = 0; i < systems.Count; i++)
            {
                var system = systems[i];
                var type = system.GetType();

                // Skip the root-level systems
                if (type == typeof(InitializationSystemGroup) ||
                    type == typeof(SimulationSystemGroup) ||
                    type == typeof(PresentationSystemGroup))
                {
                    continue;
                }

                // Add to default group if none is defined
                var groupAttributes = type.GetCustomAttributes(typeof(UpdateInGroupAttribute), true);
                if (groupAttributes.Length == 0)
                {
                    simulationSystemGroup.AddSystemToUpdateList(system);
                }

                foreach (UpdateInGroupAttribute groupAttr in groupAttributes)
                {
                    if (!typeof(ComponentSystemGroup).IsAssignableFrom(groupAttr.GroupType))
                    {
                        Debug.LogError(
                            $"Invalid [UpdateInGroup] attribute for {type}: {groupAttr.GroupType} must be derived from ComponentSystemGroup.");
                        continue;
                    }

                    var systemGroup = (ComponentSystemGroup) Worker.World.GetOrCreateSystem(groupAttr.GroupType);
                    systemGroup.AddSystemToUpdateList(Worker.World.GetOrCreateSystem(type));
                    if (!uniqueSystemTypes.Contains(groupAttr.GroupType))
                    {
                        uniqueSystemTypes.Add(groupAttr.GroupType);
                        systems.Add(systemGroup);
                    }
                }
            }

            // Sort all root groups, sorts depth first
            foreach (var system in systems)
            {
                var type = system.GetType();
                if (type == typeof(InitializationSystemGroup) ||
                    type == typeof(SimulationSystemGroup) ||
                    type == typeof(PresentationSystemGroup))
                {
                    var groupSystem = system as ComponentSystemGroup;
                    groupSystem.SortSystemUpdateList();
                }
            }
            // Extracted Code From PlayerLoopUtils //
            ScriptBehaviourUpdateOrder.UpdatePlayerLoop(Worker.World, PlayerLoop.GetCurrentPlayerLoop());
        }
        catch (Exception)
        {
#if UNITY_EDITOR
            // Temporary warning to be replaced when we can reliably detect if a local runtime is running, or not.
            logger.HandleLog(LogType.Warning,
                new LogEvent(
                        "Is a local runtime running? If not, you can start one from 'SpatialOS -> Local launch' or by pressing Cmd/Ctrl-L")
                    .WithField("Reason", "A worker running in the Editor failed to connect"));
#endif
            // A check is needed for the case that play mode is exited before the connection can complete.
            if (Application.isPlaying)
            {
                Dispose();
            }

            throw;
        }
        finally
        {
            WorkerConnectionSemaphore.Release();
        }

#if !UNITY_EDITOR && DEVELOPMENT_BUILD && !UNITY_ANDROID && !UNITY_IPHONE
        try
        {
            var port = GetPlayerConnectionPort();
            Worker.SendLogMessage(LogLevel.Info, $"Unity PlayerConnection port: {port}.", Worker.WorkerId, null);
        }
        catch (Exception e)
        {
            logger.HandleLog(LogType.Exception, new LogEvent("Could not find the Unity PlayerConnection port.").WithException(e));
        }
#endif

        foreach (var callback in workerConnectedCallbacks)
        {
            callback(Worker);
        }
    }

    private static async Task<WorkerInWorld> ConnectWithRetries(IConnectionHandlerBuilder connectionHandlerBuilder, int maxAttempts,
        ILogDispatcher logger, string workerType, float3 origin)
    {
        var remainingAttempts = maxAttempts;
        while (remainingAttempts > 0)
        {
            if (!Application.isPlaying)
            {
                throw new ConnectionFailedException("Editor application stopped", ConnectionErrorReason.EditorApplicationStopped);
            }

            try
            {
                using (var tokenSource = new CancellationTokenSource())
                {
                    Action cancelTask = delegate { tokenSource.Cancel(); };
                    Application.quitting += cancelTask;

                    var workerInWorld = await WorkerInWorld.CreateWorkerInWorldAsync(connectionHandlerBuilder, workerType, logger, origin, tokenSource.Token);

                    Application.quitting -= cancelTask;
                    return workerInWorld;
                }
            }
            catch (ConnectionFailedException e)
            {
                if (e.Reason == ConnectionErrorReason.EditorApplicationStopped)
                {
                    throw;
                }

                --remainingAttempts;
                logger.HandleLog(LogType.Error,
                    new LogEvent($"Failed attempt {maxAttempts - remainingAttempts} to create worker")
                        .WithField("WorkerType", workerType)
                        .WithField("Message", e.Message));
            }
        }

        throw new ConnectionFailedException(
            $"Tried to connect {maxAttempts} times - giving up.",
            ConnectionErrorReason.ExceededMaximumRetries);
    }
    
    protected virtual void HandleWorkerConnectionEstablished() {}

    private void OnDisconnected(string reason)
    {
        Worker.LogDispatcher.HandleLog(LogType.Log, new LogEvent($"Worker disconnected")
            .WithField("WorkerId", Worker.WorkerId)
            .WithField("Reason", reason));
        DeferredDisposeWorker();
    }

    bool wantsDispose = false;

    protected void DeferredDisposeWorker()
    {
        // Remove the world from the loop early, to avoid errors during the delay frame
        RemoveFromPlayerLoop();
        wantsDispose = true;
    }


    public virtual void Dispose()
    {
        RemoveFromPlayerLoop();
        Worker?.Dispose();
        Worker = null;
    }

    private void RemoveFromPlayerLoop()
    {
        if (Worker?.World != null)
        {
            // Remove root systems from the disposing world from the PlayerLoop
            // This only affects the loop next frame
            // Extracted Code From PlayerLoopUtils //
            var playerLoop = PlayerLoop.GetCurrentPlayerLoop();

            //Reflection to get world from PlayerLoopSystem
            var wrapperType =
                typeof(ScriptBehaviourUpdateOrder).Assembly.GetType(
                    "Unity.Entities.ScriptBehaviourUpdateOrder+DummyDelegateWrapper");
            var systemField = wrapperType.GetField("m_System", BindingFlags.NonPublic | BindingFlags.Instance);

            for (var i = 0; i < playerLoop.subSystemList.Length; ++i)
            {
                ref var playerLoopSubSystem = ref playerLoop.subSystemList[i];
                playerLoopSubSystem.subSystemList = playerLoopSubSystem.subSystemList.Where(s =>
                {
                    if (s.updateDelegate != null && s.updateDelegate.Target.GetType() == wrapperType)
                    {
                        var targetSystem = systemField.GetValue(s.updateDelegate.Target) as ComponentSystemBase;
                        return targetSystem.World != Worker.World;
                    }

                    return true;
                }).ToArray();
            }

            // Update PlayerLoop
            PlayerLoop.SetPlayerLoop(playerLoop);
            // Extracted Code From PlayerLoopUtils //
        }
    }
}

[No3371]

I am new to all stuff about ECS and SpatialOS, so I was exploring if it's possible to do a pure ECS with SpatialOS, now after a day trying get my head around the design, I see that maybe it's still good to use MonoBehaviour to connect the worker since the connection logic may vary and there's always a default scene usable even we go for pure ECS.

However, that makes ECS workflow more puzzling, let's say, we use the WorkerConnector to connect the worker then we await. When the await finished, the worker is connected, isn't it? So why do we need the OnConnected queries?

According to my limited ECS knowledge, querying for OnConnected only make sense when we can put systems into worker world before we start the connection process, but it seems there's no way we can access the WorkerInWorld.World before the connection finished.

My opinion is the CreateWorkerInWorldAsync() have to expose a parameter to specify a list of system types that has to be created on a worker world right after the world get created.

Would love to see your insights on this or maybe let me know I misunderstood anything, thanks.

[jamiebrynes7]

Hey, sorry I meant to reply yesterday, but it slipped my mind!

You are correct that the workers are currently expected to be started from the WorkerConnector MonoBehaviours, even if all the game logic exists in ECS. These are a way to 'bootstrap' your worker so to speak.

To be completely honest with you, I forgot that OnConnected still existed. I think historically (like version ~ 0.1) this was used in some of the player lifecycle systems as a way of triggering a player spawn immediately upon connection (the worker bootstrapping & systems worked a little differently back then).

I think you're correct that it doesn't have a strong use case anymore since OnCreate fulfills that purpose of knowing when the worker is connected.

My opinion is the CreateWorkerInWorldAsync() have to expose a parameter to specify a list of system types that has to be created on a worker world right after the world get created.

Traditionally, this is what the HandleWorkerConnectionEstablished is used for. This is called once the worker connects, but before we setup the PlayerLoop for that ECS world. For example, the UnityGameLogic worker uses this callback to add some systems - https://github.com/spatialos/gdk-for-unity-fps-starter-project/blob/master/workers/unity/Assets/Fps/Scripts/WorkerConnectors/GameLogicWorkerConnector.cs#L66

[No3371]

Traditionally, this is what the HandleWorkerConnectionEstablished is used for.

Nice. I was kinda lost since I keep jumping around that WorkerConnectorSystem script and WorkerConnector... guess I'd better stop thinking about the ECS connector idea for now... that kind of pure pure ECS stuff looks tempting though 😆