private static readonly TimeSpan RETRY_DELAY = TimeSpan.FromSeconds(5); // Pause 5 seconds between forwards to let the membership directory settle down

        internal RemoteGrainDirectory(LocalGrainDirectory r, GrainId id)
            : base(id, r.MyAddress)
        {
            router = r;
            partition = r.DirectoryPartition;
            logger = TraceLogger.GetLogger("Orleans.GrainDirectory.CacheValidator", TraceLogger.LoggerType.Runtime);
        }

 public ClusterGrainDirectory(LocalGrainDirectory r, GrainId grainId, string clusterId, bool lowPriority)
     : base(grainId, r.MyAddress, lowPriority)
 {
     this.router = r;        
     this.clusterId = clusterId;
     this.logger = r.Logger;
 }

        public async Task<RemoteClusterActivationResponse> ProcessActivationRequest(GrainId grain, string requestClusterId, int hopCount = 0)
        {
            // check if the requesting cluster id is in the current configuration view of this cluster
            // if not, reject the message.
            var multiClusterConfiguration = Runtime.Silo.CurrentSilo.LocalMultiClusterOracle?.GetMultiClusterConfiguration();
            if (multiClusterConfiguration == null || !multiClusterConfiguration.Clusters.Contains(requestClusterId))       
            {
                logger.Warn(ErrorCode.GlobalSingleInstance_WarningInvalidOrigin, 
                    "GSIP:Rsp {0} Origin={1} GSI request rejected because origin is not in MC configuration", grain.ToString(), requestClusterId);

                return new RemoteClusterActivationResponse(ActivationResponseStatus.Failed);
            }

            var forwardAddress = router.CheckIfShouldForward(grain, 0, "ProcessActivationRequest");

            // on all silos other than first, we insert a retry delay and recheck owner before forwarding
            if (hopCount > 0 && forwardAddress != null)
            {
                await Task.Delay(LocalGrainDirectory.RETRY_DELAY);
                forwardAddress = router.CheckIfShouldForward(grain, hopCount, "ProcessActivationRequest(recheck)");
            }

            if (forwardAddress == null)
            {
                return ProcessRequestLocal(grain, requestClusterId);
            }
            else
            {
                if (logger.IsVerbose2)
                    logger.Verbose("GSIP:Rsp {0} Origin={1} forward to {2}", grain.ToString(), requestClusterId, forwardAddress);

                var clusterGrainDir = InsideRuntimeClient.Current.InternalGrainFactory.GetSystemTarget<IClusterGrainDirectory>(Constants.ClusterDirectoryServiceId, forwardAddress);
                return await clusterGrainDir.ProcessActivationRequest(grain, requestClusterId, hopCount + 1);
            }
        }

        public static ActivationAddress GetAddress(SiloAddress silo, GrainId grain, ActivationId activation, MultiClusterStatus status = MultiClusterStatus.Owned)
        {
            // Silo part is not mandatory
            if (grain == null) throw new ArgumentNullException("grain");

            return new ActivationAddress(silo, grain, activation, status);
        }

        internal PersistentStreamPullingAgent(
            GrainId id, 
            string strProviderName,
            IStreamProviderRuntime runtime,
            IStreamPubSub streamPubSub,
            QueueId queueId,
            PersistentStreamProviderConfig config)
            : base(id, runtime.ExecutingSiloAddress, true)
        {
            if (runtime == null) throw new ArgumentNullException("runtime", "PersistentStreamPullingAgent: runtime reference should not be null");
            if (strProviderName == null) throw new ArgumentNullException("runtime", "PersistentStreamPullingAgent: strProviderName should not be null");

            QueueId = queueId;
            streamProviderName = strProviderName;
            providerRuntime = runtime;
            pubSub = streamPubSub;
            pubSubCache = new Dictionary<StreamId, StreamConsumerCollection>();
            safeRandom = new SafeRandom();
            this.config = config;
            numMessages = 0;

            logger = providerRuntime.GetLogger(GrainId + "-" + streamProviderName);
            logger.Info((int)ErrorCode.PersistentStreamPullingAgent_01, 
                "Created {0} {1} for Stream Provider {2} on silo {3} for Queue {4}.",
                GetType().Name, GrainId.ToDetailedString(), streamProviderName, Silo, QueueId.ToStringWithHashCode());

            string statUniquePostfix = strProviderName + "." + QueueId;
            numReadMessagesCounter = CounterStatistic.FindOrCreate(new StatisticName(StatisticNames.STREAMS_PERSISTENT_STREAM_NUM_READ_MESSAGES, statUniquePostfix));
            numSentMessagesCounter = CounterStatistic.FindOrCreate(new StatisticName(StatisticNames.STREAMS_PERSISTENT_STREAM_NUM_SENT_MESSAGES, statUniquePostfix));
            IntValueStatistic.FindOrCreate(new StatisticName(StatisticNames.STREAMS_PERSISTENT_STREAM_PUBSUB_CACHE_SIZE, statUniquePostfix), () => pubSubCache.Count);
            // TODO: move queue cache size statistics tracking into queue cache implementation once Telemetry APIs and LogStatistics have been reconciled.
            //IntValueStatistic.FindOrCreate(new StatisticName(StatisticNames.STREAMS_PERSISTENT_STREAM_QUEUE_CACHE_SIZE, statUniquePostfix), () => queueCache != null ? queueCache.Size : 0);
        }

 public ProxiedMessageCenter(ClientConfiguration config, IPAddress localAddress, int gen, GrainId clientId, IGatewayListProvider gatewayListProvider)
 {
     lockable = new object();
     MyAddress = SiloAddress.New(new IPEndPoint(localAddress, 0), gen);
     ClientId = clientId;
     Running = false;
     MessagingConfiguration = config;
     GatewayManager = new GatewayManager(config, gatewayListProvider);
     PendingInboundMessages = new RuntimeQueue<Message>();
     gatewayConnections = new Dictionary<Uri, GatewayConnection>();
     numMessages = 0;
     grainBuckets = new WeakReference[config.ClientSenderBuckets];
     logger = TraceLogger.GetLogger("Messaging.ProxiedMessageCenter", TraceLogger.LoggerType.Runtime);
     if (logger.IsVerbose) logger.Verbose("Proxy grain client constructed");
     IntValueStatistic.FindOrCreate(StatisticNames.CLIENT_CONNECTED_GATEWAY_COUNT, () =>
         {
             lock (gatewayConnections)
             {
                 return gatewayConnections.Values.Count(conn => conn.IsLive);
             }
         });
     if (StatisticsCollector.CollectQueueStats)
     {
         queueTracking = new QueueTrackingStatistic("ClientReceiver");
     }
 }

        public static ActivationAddress GetAddress(SiloAddress silo, GrainId grain, ActivationId activation)
        {
            // Silo part is not mandatory
            if (grain == null) throw new ArgumentNullException("grain");

            return new ActivationAddress(silo, grain, activation);
        }

 private ActivationAddress(SiloAddress silo, GrainId grain, ActivationId activation, MultiClusterStatus status)
 {
     Silo = silo;
     Grain = grain;
     Activation = activation;
     Status = status;
 }

        internal override Task<PlacementResult> OnSelectActivation(
            PlacementStrategy strategy, GrainId target, IPlacementContext context)
        {
            if (target.IsClient)
                throw new InvalidOperationException("Cannot use StatelessWorkerStrategy to route messages to client grains.");

            // If there are available (not busy with a request) activations, it returns the first one.
            // If all are busy and the number of local activations reached or exceeded MaxLocal, it randomly returns one of them.
            // Otherwise, it requests creation of a new activation.
            List<ActivationData> local;

            if (!context.LocalLookup(target, out local) || local.Count == 0)
                return Task.FromResult((PlacementResult)null);

            var placement = (StatelessWorkerPlacement)strategy;

            foreach (var activation in local)
            {
                ActivationData info;
                if (!context.TryGetActivationData(activation.ActivationId, out info) ||
                    info.State != ActivationState.Valid || !info.IsInactive) continue;

                return Task.FromResult(PlacementResult.IdentifySelection(ActivationAddress.GetAddress(context.LocalSilo, target, activation.ActivationId)));
            }

            if (local.Count >= placement.MaxLocal)
            {
                var id = local[local.Count == 1 ? 0 : random.Next(local.Count)].ActivationId;
                return Task.FromResult(PlacementResult.IdentifySelection(ActivationAddress.GetAddress(context.LocalSilo, target, id)));
            }

            return Task.FromResult((PlacementResult)null);
        }

        public override async Task<PlacementResult> OnSelectActivation(PlacementStrategy strategy, GrainId target, IPlacementContext context)
        {
            // first, check if we can find an activation for this client in the cache or local directory partition
            AddressesAndTag addresses;
            if (context.FastLookup(target, out addresses))
                return ChooseRandomActivation(addresses.Addresses, context);

            // we need to look up the directory entry for this grain on a remote silo
            switch (target.Category)
            {
                case UniqueKey.Category.Client:
                    {
                        addresses = await context.FullLookup(target);
                        return ChooseRandomActivation(addresses.Addresses, context);
                    }

                case UniqueKey.Category.GeoClient:
                    {
                        // we need to look up the activations in the remote cluster
                        addresses = await context.LookupInCluster(target, target.Key.ClusterId);
                        return ChooseRandomActivation(addresses.Addresses, context);
                    }

                default:
                    throw new InvalidOperationException("Unsupported client type. Grain " + target);
            }
        }

        internal PersistentStreamPullingAgent(
            GrainId id, 
            string strProviderName,
            IStreamProviderRuntime runtime,
            QueueId queueId, 
            TimeSpan queueGetPeriod,
            TimeSpan initQueueTimeout,
            TimeSpan maxDeliveryTime)
            : base(id, runtime.ExecutingSiloAddress, true)
        {
            if (runtime == null) throw new ArgumentNullException("runtime", "PersistentStreamPullingAgent: runtime reference should not be null");
            if (strProviderName == null) throw new ArgumentNullException("runtime", "PersistentStreamPullingAgent: strProviderName should not be null");

            QueueId = queueId;
            streamProviderName = strProviderName;
            providerRuntime = runtime;
            pubSub = runtime.PubSub(StreamPubSubType.GrainBased);
            pubSubCache = new Dictionary<StreamId, StreamConsumerCollection>();
            safeRandom = new SafeRandom();
            this.queueGetPeriod = queueGetPeriod;
            this.initQueueTimeout = initQueueTimeout;
            this.maxDeliveryTime = maxDeliveryTime;
            numMessages = 0;

            logger = providerRuntime.GetLogger(GrainId + "-" + streamProviderName);
            logger.Info((int)ErrorCode.PersistentStreamPullingAgent_01, 
                "Created {0} {1} for Stream Provider {2} on silo {3} for Queue {4}.",
                GetType().Name, GrainId.ToDetailedString(), streamProviderName, Silo, QueueId.ToStringWithHashCode());

            numReadMessagesCounter = CounterStatistic.FindOrCreate(new StatisticName(StatisticNames.STREAMS_PERSISTENT_STREAM_NUM_READ_MESSAGES, strProviderName));
            numSentMessagesCounter = CounterStatistic.FindOrCreate(new StatisticName(StatisticNames.STREAMS_PERSISTENT_STREAM_NUM_SENT_MESSAGES, strProviderName));
        }

 protected SystemTarget(GrainId grainId, SiloAddress silo, bool lowPriority)
 {
     GrainId = grainId;
     Silo = silo;
     ActivationId = ActivationId.GetSystemActivation(grainId, silo);
     SchedulingContext = new SchedulingContext(this, lowPriority);
 }

 internal override Task<PlacementResult> OnAddActivation(
     PlacementStrategy strategy, GrainId grain, IPlacementContext context)
 {
     var grainType = context.GetGrainTypeName(grain);
     var allSilos = context.AllActiveSilos;
     return Task.FromResult(
         PlacementResult.SpecifyCreation(allSilos[random.Next(allSilos.Count)], strategy, grainType));
 }

 internal void ClientDropped(GrainId clientId)
 {
     var addr = GetClientActivationAddress(clientId);
     scheduler.QueueTask(
         () => ExecuteWithRetries(() => grainDirectory.UnregisterAsync(addr, force:true), ErrorCode.ClientRegistrarFailedToUnregister, String.Format("Directory.UnRegisterAsync {0} failed.", addr)), 
         this.SchedulingContext)
                 .Ignore();
 }

        private GlobalSingleInstanceResponseTracker(Task<RemoteClusterActivationResponse>[] responsePromises, GrainId grain, Logger logger)
        {
            this.responsePromises = responsePromises;
            this.grain = grain;
            this.logger = logger;

            CheckIfDone();
        }

        internal async Task Start(StatisticsProviderManager statsManager, IMessageCenter transport, GrainId clientId)
        {
            runtimeStats.Start();

            // Configure Metrics
            IProvider statsProvider = null;
            if (!string.IsNullOrEmpty(config.StatisticsProviderName))
            {
                var extType = config.StatisticsProviderName;
                statsProvider = statsManager.GetProvider(extType);
                var metricsDataPublisher = statsProvider as IClientMetricsDataPublisher;
                if (metricsDataPublisher == null)
                {
                    var msg = String.Format("Trying to create {0} as a metrics publisher, but the provider is not configured."
                        , extType);
                    throw new ArgumentException(msg, "ProviderType (configuration)");
                }
                var configurableMetricsDataPublisher = metricsDataPublisher as IConfigurableClientMetricsDataPublisher;
                if (configurableMetricsDataPublisher != null)
                {
                    configurableMetricsDataPublisher.AddConfiguration(
                        config.DeploymentId, config.DNSHostName, clientId.ToString(), transport.MyAddress.Endpoint.Address);
                }
                tableStatistics = new ClientTableStatistics(transport, metricsDataPublisher, runtimeStats)
                {
                    MetricsTableWriteInterval = config.StatisticsMetricsTableWriteInterval
                };
            }
            else if (config.UseAzureSystemStore)
            {
                // Hook up to publish client metrics to Azure storage table
                var publisher = AssemblyLoader.LoadAndCreateInstance<IClientMetricsDataPublisher>(Constants.ORLEANS_AZURE_UTILS_DLL, logger);
                await publisher.Init(config, transport.MyAddress.Endpoint.Address, clientId.ToParsableString());
                tableStatistics = new ClientTableStatistics(transport, publisher, runtimeStats)
                {
                    MetricsTableWriteInterval = config.StatisticsMetricsTableWriteInterval
                };
            }

            // Configure Statistics
            if (config.StatisticsWriteLogStatisticsToTable)
            {
                if (statsProvider != null)
                {
                    logStatistics.StatsTablePublisher = statsProvider as IStatisticsPublisher;
                    // Note: Provider has already been Init-ialized above.
                }
                else if (config.UseAzureSystemStore)
                {
                    var statsDataPublisher = AssemblyLoader.LoadAndCreateInstance<IStatisticsPublisher>(Constants.ORLEANS_AZURE_UTILS_DLL, logger);
                    await statsDataPublisher.Init(false, config.DataConnectionString, config.DeploymentId,
                        transport.MyAddress.Endpoint.ToString(), clientId.ToParsableString(), config.DNSHostName);
                    logStatistics.StatsTablePublisher = statsDataPublisher;
                }
            }
            logStatistics.Start();
        }

 internal void ClientAdded(GrainId clientId)
 {
     // Use a ActivationId that is hashed from clientId, and not random ActivationId.
     // That way, when we refresh it in the directiry, it's the same one.
     var addr = GetClientActivationAddress(clientId);
     scheduler.QueueTask(
         () => ExecuteWithRetries(() => grainDirectory.RegisterAsync(addr), ErrorCode.ClientRegistrarFailedToRegister, String.Format("Directory.RegisterAsync {0} failed.", addr)), 
         this.SchedulingContext)
                 .Ignore();
 }

 internal List<ActivationAddress> GetHandedOffInfo(GrainId grain)
 {
     foreach (var partition in directoryPartitionsMap.Values)
     {
         var result = partition.LookUpGrain(grain);
         if (result.Addresses != null)
             return result.Addresses;
     }
     return null;
 }

        private Task<PlacementResult> MakePlacement(PlacementStrategy strategy, GrainId grain, IPlacementContext context, CachedLocalStat minLoadedSilo)
        {
            // Increment placement by number of silos instead of by one.
            // This is our trick to get more balanced placement, accounting to the probable
            // case when multiple silos place on the same silo at the same time, before stats are refreshed.
            minLoadedSilo.IncrementActivationCount(localCache.Count);

            return Task.FromResult(PlacementResult.SpecifyCreation(
                minLoadedSilo.Address,
                strategy,
                context.GetGrainTypeName(grain)));
        }

 internal List<ActivationAddress> GetHandedOffInfo(GrainId grain)
 {
     foreach (var partition in directoryPartitionsMap.Values)
     {
         var result = partition.LookUpGrain(grain);
         if (result != null)
         {
             // Force the list to be created in order to avoid race conditions
             return result.Item1.Select(pair => ActivationAddress.GetAddress(pair.Item1, grain, pair.Item2)).ToList();
         }
     }
     return null;
 }