in src/prod/src/Reliability/Replication/ComReplicator.cpp [453:796]
HRESULT ComReplicator::UpdateReplicatorSettings(
/* [in] */ FABRIC_REPLICATOR_SETTINGS const * replicatorSettings)
{
if (replicatorSettings == NULL) { return ComUtility::OnPublicApiReturn(E_POINTER); }
FABRIC_REPLICATOR_SETTINGS_EX1 * replicatorSettingsEx1 = (FABRIC_REPLICATOR_SETTINGS_EX1 *) replicatorSettings->Reserved;
if (replicatorSettingsEx1 != NULL)
{
FABRIC_REPLICATOR_SETTINGS_EX2 * replicatorSettingsEx2 = (FABRIC_REPLICATOR_SETTINGS_EX2 *)replicatorSettingsEx1->Reserved;
if (replicatorSettingsEx2 != NULL)
{
FABRIC_REPLICATOR_SETTINGS_EX3 * replicatorSettingsEx3 = (FABRIC_REPLICATOR_SETTINGS_EX3 *)replicatorSettingsEx2->Reserved;
if (replicatorSettingsEx3 != NULL)
{
FABRIC_REPLICATOR_SETTINGS_EX4 * replicatorSettingsEx4 = (FABRIC_REPLICATOR_SETTINGS_EX4 *)replicatorSettingsEx3->Reserved;
FAIL_IF_OPTIONAL_PARAM_RESERVED_FIELD_NOT_NULL(replicatorSettingsEx4);
}
}
}
std::wstring errorCause;
Common::StringWriter errorWriter(errorCause);
errorWriter.WriteLine("Security settings is the only supported setting that can be changed using UpdateReplicatorSettings");
// Check if any non upgradable settings have changed
bool hasBreakingChangesInNewSettings = false;
{
DWORD effective_initialPrimaryReplicationQueueSize = 0;
DWORD effective_maxPrimaryReplicationQueueSize = 0;
DWORD effective_maxPrimaryReplicationQueueMemorySize = 0;
DWORD effective_initialSecondaryReplicationQueueSize = 0;
DWORD effective_maxSecondaryReplicationQueueSize = 0;
DWORD effective_maxSecondaryReplicationQueueMemorySize = 0;
DWORD flags = replicatorSettings->Flags;
REInternalSettingsSPtr config = replicator_.Config;
ReplicatorSettingsUPtr inputSettings;
auto error = ReplicatorSettings::FromPublicApi(
*replicatorSettings,
inputSettings);
if (!error.IsSuccess())
{
return error.ToHResult();
}
ASSERT_IFNOT(flags == inputSettings->Flags, "Input Flags should be equal to calculated flags");
if (!hasBreakingChangesInNewSettings &&
(flags & FABRIC_REPLICATOR_ADDRESS) != 0)
{
std::wstring replicatorAddress(inputSettings->ReplicatorAddress);
hasBreakingChangesInNewSettings = !StringUtility::AreEqualCaseInsensitive(config->ReplicatorAddress, replicatorAddress);
if (hasBreakingChangesInNewSettings)
errorWriter.WriteLine("{0} mismatch {1}, {2}", "ReplicatorAddress", replicatorAddress, config->ReplicatorAddress);
}
// NOTE: Intentionally not comparing listen and publish address during update and ignoring it because we have the merged setting
// while user might provide older values
if (!hasBreakingChangesInNewSettings &&
(flags & FABRIC_REPLICATOR_RETRY_INTERVAL) != 0)
{
hasBreakingChangesInNewSettings = config->RetryInterval != inputSettings->RetryInterval ?
true :
false;
if (hasBreakingChangesInNewSettings)
errorWriter.WriteLine("{0} mismatch {1}, {2}", "RetryInterval", inputSettings->RetryInterval, config->RetryInterval);
}
if (!hasBreakingChangesInNewSettings &&
(flags & FABRIC_REPLICATOR_BATCH_ACKNOWLEDGEMENT_INTERVAL) != 0)
{
hasBreakingChangesInNewSettings = config->BatchAcknowledgementInterval != inputSettings->BatchAcknowledgementInterval ?
true :
false;
if (hasBreakingChangesInNewSettings)
errorWriter.WriteLine("{0} mismatch {1}, {2}", "BatchAcknowledgementInterval", inputSettings->BatchAcknowledgementInterval, config->BatchAcknowledgementInterval);
}
if (!hasBreakingChangesInNewSettings &&
(flags & FABRIC_REPLICATOR_REQUIRE_SERVICE_ACK) != 0)
{
hasBreakingChangesInNewSettings = config->RequireServiceAck != inputSettings->RequireServiceAck ?
true :
false;
if (hasBreakingChangesInNewSettings)
errorWriter.WriteLine("{0} mismatch {1}, {2}", "RequireServiceAck", inputSettings->RequireServiceAck, config->RequireServiceAck);
}
if (!hasBreakingChangesInNewSettings &&
(flags & FABRIC_REPLICATOR_REPLICATION_QUEUE_INITIAL_SIZE) != 0)
{
effective_initialPrimaryReplicationQueueSize = static_cast<DWORD>(inputSettings->InitialReplicationQueueSize);
effective_initialSecondaryReplicationQueueSize = static_cast<DWORD>(inputSettings->InitialReplicationQueueSize);
errorWriter.WriteLine("Setting Initial Primary and Secondary Queue Size from config 'InitialReplicationQueueSize' value {0}", inputSettings->InitialReplicationQueueSize);
}
if (!hasBreakingChangesInNewSettings &&
(flags & FABRIC_REPLICATOR_REPLICATION_QUEUE_MAX_SIZE) != 0)
{
effective_maxPrimaryReplicationQueueSize = static_cast<DWORD>(inputSettings->MaxReplicationQueueSize);
effective_maxSecondaryReplicationQueueSize = static_cast<DWORD>(inputSettings->MaxReplicationQueueSize);
errorWriter.WriteLine("Setting Max Primary and Secondary Queue Size from config 'MaxReplicationQueueSize' value {0}", inputSettings->MaxReplicationQueueSize);
}
if (!hasBreakingChangesInNewSettings &&
(flags & FABRIC_REPLICATOR_REPLICATION_QUEUE_MAX_MEMORY_SIZE) != 0)
{
effective_maxPrimaryReplicationQueueMemorySize = static_cast<DWORD>(inputSettings->MaxReplicationQueueMemorySize);
effective_maxSecondaryReplicationQueueMemorySize = static_cast<DWORD>(inputSettings->MaxReplicationQueueMemorySize);
errorWriter.WriteLine("Setting Max Primary and Secondary Queue MemorySize from config 'MaxReplicationQueueMemorySize' value {0}", inputSettings->MaxReplicationQueueMemorySize);
if ((flags & FABRIC_REPLICATOR_REPLICATION_QUEUE_MAX_SIZE) == 0)
{
effective_maxPrimaryReplicationQueueSize = 0;
effective_maxSecondaryReplicationQueueSize = 0;
errorWriter.WriteLine("Setting Max Primary and Secondary Queue Size to 0 since max memory size is configured");
}
}
if (!hasBreakingChangesInNewSettings &&
(flags & FABRIC_REPLICATOR_COPY_QUEUE_INITIAL_SIZE) != 0)
{
hasBreakingChangesInNewSettings = static_cast<DWORD>(config->InitialCopyQueueSize) != inputSettings->InitialCopyQueueSize ?
true :
false;
if (hasBreakingChangesInNewSettings)
errorWriter.WriteLine("{0} mismatch {1}, {2}", "InitialCopyQueueSize", inputSettings->InitialCopyQueueSize, config->InitialCopyQueueSize);
}
if (!hasBreakingChangesInNewSettings &&
(flags & FABRIC_REPLICATOR_COPY_QUEUE_MAX_SIZE) != 0)
{
hasBreakingChangesInNewSettings = static_cast<DWORD>(config->MaxCopyQueueSize) != inputSettings->MaxCopyQueueSize ?
true :
false;
if (hasBreakingChangesInNewSettings)
errorWriter.WriteLine("{0} mismatch {1}, {2}", "MaxCopyQueueSize", inputSettings->MaxCopyQueueSize, config->MaxCopyQueueSize);
}
if (!hasBreakingChangesInNewSettings &&
(flags & FABRIC_REPLICATOR_SECONDARY_CLEAR_ACKNOWLEDGED_OPERATIONS) != 0)
{
hasBreakingChangesInNewSettings = config->SecondaryClearAcknowledgedOperations != inputSettings->SecondaryClearAcknowledgedOperations ?
true :
false;
if (hasBreakingChangesInNewSettings)
errorWriter.WriteLine("{0} mismatch {1}, {2}", "SecondaryClearAcknowledgedOperations", inputSettings->SecondaryClearAcknowledgedOperations, config->SecondaryClearAcknowledgedOperations);
}
if (!hasBreakingChangesInNewSettings &&
(flags & FABRIC_REPLICATOR_REPLICATION_MESSAGE_MAX_SIZE) != 0)
{
hasBreakingChangesInNewSettings = static_cast<DWORD>(config->MaxReplicationMessageSize) != inputSettings->MaxReplicationMessageSize ?
true :
false;
if (hasBreakingChangesInNewSettings)
errorWriter.WriteLine("{0} mismatch {1}, {2}", "MaxReplicationMessageSize", inputSettings->MaxReplicationMessageSize, config->MaxReplicationMessageSize);
}
if (!hasBreakingChangesInNewSettings &&
(flags & FABRIC_REPLICATOR_USE_STREAMFAULTS_AND_ENDOFSTREAM_OPERATIONACK) != 0)
{
hasBreakingChangesInNewSettings = config->UseStreamFaultsAndEndOfStreamOperationAck != inputSettings->UseStreamFaultsAndEndOfStreamOperationAck ?
true :
false;
if (hasBreakingChangesInNewSettings)
errorWriter.WriteLine("{0} mismatch {1}, {2}", "UseStreamFaultsAndEndOfStreamOperationAck", inputSettings->UseStreamFaultsAndEndOfStreamOperationAck, config->UseStreamFaultsAndEndOfStreamOperationAck);
}
if (!hasBreakingChangesInNewSettings &&
(flags & FABRIC_REPLICATOR_PRIMARY_REPLICATION_QUEUE_INITIAL_SIZE) != 0)
{
effective_initialPrimaryReplicationQueueSize = static_cast<DWORD>(inputSettings->InitialPrimaryReplicationQueueSize);
errorWriter.WriteLine("Setting Initial Primary Queue MemorySize from config 'InitialPrimaryReplicationQueueSize' value {0}", inputSettings->InitialPrimaryReplicationQueueSize);
}
if (!hasBreakingChangesInNewSettings &&
(flags & FABRIC_REPLICATOR_PRIMARY_REPLICATION_QUEUE_MAX_SIZE) != 0)
{
effective_maxPrimaryReplicationQueueSize = static_cast<DWORD>(inputSettings->MaxPrimaryReplicationQueueSize);
errorWriter.WriteLine("Setting Max Primary Queue Size from config 'MaxPrimaryReplicationQueueSize' value {0}", inputSettings->MaxPrimaryReplicationQueueSize);
}
if (!hasBreakingChangesInNewSettings &&
(flags & FABRIC_REPLICATOR_PRIMARY_REPLICATION_QUEUE_MAX_MEMORY_SIZE) != 0)
{
effective_maxPrimaryReplicationQueueMemorySize = static_cast<DWORD>(inputSettings->MaxPrimaryReplicationQueueMemorySize);
errorWriter.WriteLine("Setting Max Primary Queue MemorySize from config 'MaxPrimaryReplicationQueueMemorySize' value {0}", inputSettings->MaxPrimaryReplicationQueueMemorySize);
}
if (!hasBreakingChangesInNewSettings &&
(flags & FABRIC_REPLICATOR_SECONDARY_REPLICATION_QUEUE_INITIAL_SIZE) != 0)
{
effective_initialSecondaryReplicationQueueSize = static_cast<DWORD>(inputSettings->InitialSecondaryReplicationQueueSize);
errorWriter.WriteLine("Setting Initial Secondary Queue MemorySize from config 'InitialSecondaryReplicationQueueSize' value {0}", inputSettings->InitialSecondaryReplicationQueueSize);
}
if (!hasBreakingChangesInNewSettings &&
(flags & FABRIC_REPLICATOR_SECONDARY_REPLICATION_QUEUE_MAX_SIZE) != 0)
{
effective_maxSecondaryReplicationQueueSize = static_cast<DWORD>(inputSettings->MaxSecondaryReplicationQueueSize);
errorWriter.WriteLine("Setting Max Secondary Queue Size from config 'MaxSecondaryReplicationQueueSize' value {0}", inputSettings->MaxSecondaryReplicationQueueSize);
}
if (!hasBreakingChangesInNewSettings &&
(flags & FABRIC_REPLICATOR_SECONDARY_REPLICATION_QUEUE_MAX_MEMORY_SIZE) != 0)
{
effective_maxSecondaryReplicationQueueMemorySize = static_cast<DWORD>(inputSettings->MaxSecondaryReplicationQueueMemorySize);
errorWriter.WriteLine("Setting Max Secondary Queue MemorySize from config 'MaxSecondaryReplicationQueueMemorySize' value {0}", inputSettings->MaxSecondaryReplicationQueueMemorySize);
}
if (!hasBreakingChangesInNewSettings &&
(flags & FABRIC_REPLICATOR_PRIMARY_WAIT_FOR_PENDING_QUORUMS_TIMEOUT) != 0)
{
hasBreakingChangesInNewSettings = config->PrimaryWaitForPendingQuorumsTimeout != inputSettings->PrimaryWaitForPendingQuorumsTimeout ?
true :
false;
if (hasBreakingChangesInNewSettings)
errorWriter.WriteLine("{0} mismatch {1}, {2}", "PrimaryWaitForPendingQuorumsTimeout", inputSettings->PrimaryWaitForPendingQuorumsTimeout, config->PrimaryWaitForPendingQuorumsTimeout);
}
// Do this in the end, after all effective queue sizes are calculated above
if (!hasBreakingChangesInNewSettings &&
((flags & FABRIC_REPLICATOR_REPLICATION_QUEUE_INITIAL_SIZE) || (flags & FABRIC_REPLICATOR_PRIMARY_REPLICATION_QUEUE_INITIAL_SIZE)))
{
hasBreakingChangesInNewSettings = config->InitialPrimaryReplicationQueueSize != effective_initialPrimaryReplicationQueueSize ?
true :
false;
if (hasBreakingChangesInNewSettings)
errorWriter.WriteLine("Current Initial Primary Queue Size {0} is not equal to input Initial Primary Queue Size {1}", config->InitialPrimaryReplicationQueueSize, effective_initialPrimaryReplicationQueueSize);
}
if (!hasBreakingChangesInNewSettings &&
((flags & FABRIC_REPLICATOR_REPLICATION_QUEUE_MAX_SIZE) || (flags & FABRIC_REPLICATOR_PRIMARY_REPLICATION_QUEUE_MAX_SIZE)))
{
hasBreakingChangesInNewSettings = config->MaxPrimaryReplicationQueueSize != effective_maxPrimaryReplicationQueueSize ?
true :
false;
if (hasBreakingChangesInNewSettings)
errorWriter.WriteLine("Current Max Primary Queue Size {0} is not equal to input Max Primary Queue Size {1}", config->MaxPrimaryReplicationQueueSize, effective_maxPrimaryReplicationQueueSize);
}
if (!hasBreakingChangesInNewSettings &&
((flags & FABRIC_REPLICATOR_REPLICATION_QUEUE_MAX_MEMORY_SIZE) || (flags & FABRIC_REPLICATOR_PRIMARY_REPLICATION_QUEUE_MAX_MEMORY_SIZE)))
{
hasBreakingChangesInNewSettings = config->MaxPrimaryReplicationQueueMemorySize != effective_maxPrimaryReplicationQueueMemorySize ?
true :
false;
if (hasBreakingChangesInNewSettings)
errorWriter.WriteLine("Current Max Primary Queue MemorySize {0} is not equal to input Max Primary Queue MemorySize {1}", config->MaxPrimaryReplicationQueueMemorySize, effective_maxPrimaryReplicationQueueMemorySize);
}
if (!hasBreakingChangesInNewSettings &&
((flags & FABRIC_REPLICATOR_REPLICATION_QUEUE_INITIAL_SIZE) || (flags & FABRIC_REPLICATOR_SECONDARY_REPLICATION_QUEUE_INITIAL_SIZE)))
{
hasBreakingChangesInNewSettings = config->InitialSecondaryReplicationQueueSize != effective_initialSecondaryReplicationQueueSize ?
true :
false;
if (hasBreakingChangesInNewSettings)
errorWriter.WriteLine("Current Initial Secondary Queue Size {0} is not equal to input Initial Secondary Queue Size {1}", config->InitialSecondaryReplicationQueueSize, effective_initialSecondaryReplicationQueueSize);
}
if (!hasBreakingChangesInNewSettings &&
((flags & FABRIC_REPLICATOR_REPLICATION_QUEUE_MAX_SIZE) || (flags & FABRIC_REPLICATOR_SECONDARY_REPLICATION_QUEUE_MAX_SIZE)))
{
hasBreakingChangesInNewSettings = config->MaxSecondaryReplicationQueueSize != effective_maxSecondaryReplicationQueueSize ?
true :
false;
if (hasBreakingChangesInNewSettings)
errorWriter.WriteLine("Current Max Secondary Queue Size {0} is not equal to input Max Secondary Queue Size {1}", config->MaxSecondaryReplicationQueueSize, effective_maxSecondaryReplicationQueueSize);
}
if (!hasBreakingChangesInNewSettings &&
((flags & FABRIC_REPLICATOR_REPLICATION_QUEUE_MAX_MEMORY_SIZE) || (flags & FABRIC_REPLICATOR_SECONDARY_REPLICATION_QUEUE_MAX_MEMORY_SIZE)))
{
hasBreakingChangesInNewSettings = config->MaxSecondaryReplicationQueueMemorySize != effective_maxSecondaryReplicationQueueMemorySize ?
true :
false;
if (hasBreakingChangesInNewSettings)
errorWriter.WriteLine("Current Max Secondary Queue MemorySize {0} is not equal to input Max Secondary Queue MemorySize {1}", config->MaxSecondaryReplicationQueueMemorySize, effective_maxSecondaryReplicationQueueMemorySize);
}
}
if (!hasBreakingChangesInNewSettings)
{
ErrorCode error;
if ((replicatorSettings->Flags & FABRIC_REPLICATOR_SECURITY) != 0)
{
if (NULL == replicatorSettings->SecurityCredentials)
{
ReplicatorEventSource::Events->ReplicatorFactoryGeneralConfigError(
ReplicatorEngine.PartitionId,
ReplicatorEngine.ReplicaId,
L"Security Credentials is NULL, but FABRIC_REPLICATOR_SECURITY flag is enabled");
return ComUtility::OnPublicApiReturn(E_INVALIDARG);
}
Transport::SecuritySettings securitySettings;
error = Transport::SecuritySettings::FromPublicApi(*(replicatorSettings->SecurityCredentials), securitySettings);
if (error.IsSuccess())
{
error = replicator_.UpdateSecuritySettings(securitySettings);
}
}
return ComUtility::OnPublicApiReturn(error.ToHResult());
}
else
{
ReplicatorEventSource::Events->ReplicatorFactoryGeneralConfigError(
ReplicatorEngine.PartitionId,
ReplicatorEngine.ReplicaId,
errorCause);
return ComUtility::OnPublicApiReturn(E_INVALIDARG);
}
}