<commit_before>﻿// -----------------------------------------------------------------------
// <copyright file="ClusterCacheInvalidation.cs" company="Asynkron AB">
//      Copyright (C) 2015-2020 Asynkron AB All rights reserved
// </copyright>
// -----------------------------------------------------------------------
using System;
using System.Collections;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Proto.Extensions;

namespace Proto.Cluster.Cache
{
    public class ClusterCacheInvalidation : IActorSystemExtension<ClusterCacheInvalidation>
    {
        private const string ActorName = "$invalidator";

        private ConcurrentDictionary<string, int> AddressRefs { get; } = new();
        private ActorSystem ActorSystem { get; }
        private int NextId { get; set; }

        public ClusterCacheInvalidation(Cluster cluster)
        {
            ActorSystem = cluster.System;
            ActorSystem.Extensions.Register(this);

            ActorSystem.EventStream.Subscribe<ClusterTopology>(topology => {
                    foreach (var member in topology.Joined)
                    {
                        if (member.Id.Equals(ActorSystem.Id)) continue;

                        AddressRefs.TryAdd(member.Address, NextId++);
                    }

                    foreach (var member in topology.Left)
                    {
                        AddressRefs.Remove(member.Address, out _);
                    }
                }
            );

            ActorSystem.Root.SpawnNamed(
                Props.FromFunc(context => {
                        if (context.Message is ActivationTerminated terminated)
                        {
                            cluster.PidCache.RemoveByVal(terminated.ClusterIdentity, terminated.Pid);
                        }

                        return Task.CompletedTask;
                    }
                ),
                ActorName
            );
        }

        private bool IsRemote(PID? sender) => sender?.Address != null && !sender.Address.Equals(ActorSystem.Address);

        private void Invalidate(ClusterIdentity identity, PID activation, BitArray activeRemotes)
        {
            var message = new ActivationTerminated
            {
                ClusterIdentity = identity,
                Pid = activation
            };
            var remotesToInvalidate = AddressRefs
                .Where(pair => activeRemotes[pair.Value])
                .Select(pair => pair.Key);

            foreach (var address in remotesToInvalidate)
            {
                ActorSystem.Root.Send(PID.FromAddress(address, ActorName), message);
            }
        }

        private void AddRemote(PID sender, BitArray activeRemotes)
        {
            if (AddressRefs.TryGetValue(sender.Address, out var index))
            {
                if (index > activeRemotes.Length)
                {
                    activeRemotes.Length = index + 1;
                }

                activeRemotes[index] = true;<commit_msg>Bugfix<commit_after>﻿// -----------------------------------------------------------------------
// <copyright file="ClusterCacheInvalidation.cs" company="Asynkron AB">
//      Copyright (C) 2015-2020 Asynkron AB All rights reserved
// </copyright>
// -----------------------------------------------------------------------
using System;
using System.Collections;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Proto.Extensions;

namespace Proto.Cluster.Cache
{
    public class ClusterCacheInvalidation : IActorSystemExtension<ClusterCacheInvalidation>
    {
        private const string ActorName = "$invalidator";

        private ConcurrentDictionary<string, int> AddressRefs { get; } = new();
        private ActorSystem ActorSystem { get; }
        private int NextId { get; set; }

        public ClusterCacheInvalidation(Cluster cluster)
        {
            ActorSystem = cluster.System;
            ActorSystem.Extensions.Register(this);

            ActorSystem.EventStream.Subscribe<ClusterTopology>(topology => {
                    foreach (var member in topology.Joined)
                    {
                        if (member.Id.Equals(ActorSystem.Id)) continue;

                        AddressRefs.TryAdd(member.Address, NextId++);
                    }

                    foreach (var member in topology.Left)
                    {
                        AddressRefs.Remove(member.Address, out _);
                    }
                }
            );

            ActorSystem.Root.SpawnNamed(
                Props.FromFunc(context => {
                        if (context.Message is ActivationTerminated terminated)
                        {
                            cluster.PidCache.RemoveByVal(terminated.ClusterIdentity, terminated.Pid);
                        }

                        return Task.CompletedTask;
                    }
                ),
                ActorName
            );
        }

        private bool IsRemote(PID? sender) => sender?.Address != null && !sender.Address.Equals(ActorSystem.Address);

        private void Invalidate(ClusterIdentity identity, PID activation, BitArray activeRemotes)
        {
            var message = new ActivationTerminated
            {
                ClusterIdentity = identity,
                Pid = activation
            };
            var remotesToInvalidate = AddressRefs
                .Where(pair => activeRemotes[pair.Value])
                .Select(pair => pair.Key);

            foreach (var address in remotesToInvalidate)
            {
                ActorSystem.Root.Send(PID.FromAddress(address, ActorName), message);
            }
        }

        private void AddRemote(PID sender, BitArray activeRemotes)
        {
            if (AddressRefs.TryGetValue(sender.Address, out var index))
            {
                if (index >= activeRemotes.Length)
                {
                    activeRemotes.Length = index + 1;
                }

                activeRemotes[index] = true;
{'fraction_non_alphanumeric': 0.12512139851084494, 'fraction_numerical': 0.0029135642602784073, 'mean_word_length': 1.407089988313206, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2007946', 'n_tokens_mistral': 1489, 'n_tokens_neox': 1348, 'n_words': 293}