Проблема с параллелизмом функций Azure OrchestrationTrigger ⇐ C#

[Anonymous]

Я создаю приложение-функцию оркестрации Azure, которое имеет четыре разные функции. Первая функция — это функция задачи запуска, которая удаляет контейнеры из базы данных Cosmos DB. Вторая функция, с которой у меня возникла проблема, сканирует контейнер Cosmos DB, состоящий из нескольких ссылок, и извлекает некоторые из них. Затем он сохраняет ссылки внутри нового контейнера Cosmos DB.
Внутри функции оркестрации я использую цикл for, который запускает эту функцию 4 раза с разными параметрами. На самом деле я будет около 9 или более: я хочу, чтобы приложение взяло A, B, C, D и завершило все. Теперь это происходит взад и вперед.
Спасибо за вашу помощь.
Это моя функция оркестрации:

Код: Выделить всё

[Function(nameof(CosmosDBFunction))]
public async Task RunOrchestrator(
[OrchestrationTrigger] TaskOrchestrationContext context)
{
ILogger logger = context.CreateReplaySafeLogger(nameof(CosmosDBFunction));
logger.LogInformation("----Function Started-----");

try
{
await context.CallActivityAsync("StartupTask");
}
catch (Exception ex)
{
logger.LogError($"Error happened StartupTask: {ex.Message}");
}

var parallelTasks = new List();
var sites = new List() {
"A", "B", "C", "D"
};

for (int i = 0; i < sites.Count; i++)
{
try
{
var task = context.CallActivityAsync("ScanSafetoAContainer", sites[i]);
parallelTasks.Add(task);
}
catch (TaskFailedException ex) when (ex.InnerException is CosmosException cosmosException)
{
_logger.LogError($"Error calling for site {sites[i]}: {ex.Message}");
}
catch (TaskFailedException ex)
{
_logger.LogError($"Error calling {sites[i]}: {ex.Message}");
}
}

await Task.WhenAll(parallelTasks);

context.CallActivityAsync("SaveToAContainer");

try
{
logger.LogInformation("----Function B Started-----");
await context.CallActivityAsync("SaveToXContainer");
}
catch (TaskFailedException ex)
{
logger.LogError($"Error happened : {ex.Message}");
}

try
{
logger.LogInformation("----Function clinksToBlinks Started-----");
await context.CallActivityAsync("SaveToYContainer");
}
catch (TaskFailedException ex)
{
logger.LogError($"Error happened: {ex.Message}");
}

logger.LogInformation("----Function Finished-----");

try
{
logger.LogInformation($"Operation is completed");
return "Operation is completed";
}
catch (Exception ex)
{
return $"Operation is completed: exception at the end line {ex.Message}";
}
}

Это функция, которая была вызвана внутри цикла:

Код: Выделить всё

[Function(nameof(Something))]
[CosmosDBOutput(databaseName: "DB", containerName: "AB",
Connection = "CosmosConnection"
, CreateIfNotExists = true, PartitionKey = "/something")]
public async Task SaveAToContainer
([ActivityTrigger] string site,
FunctionContext executionContext)
{
ConcurrentBag links = new ConcurrentBag();
ILogger _logger = executionContext
.GetLogger(nameof(Something));

var container = await CreateContainer();
var client = _httpClientFactory;
client.Timeout = TimeSpan.FromSeconds(60);

using FeedIterator object = container.GetItemLinqQueryable
()
.Where(s =>  s.prop == prop)
.ToFeedIterator();

while (pages.HasMoreResults)
{
long counter = 0;

foreach (var page in await pages.ReadNextAsync())
{
string href = string.Empty;

try
{
HttpResponseMessage responseFromUrl = await client.GetAsync(page.Url);
var htmlContent = await responseFromUrl.Content.ReadAsStringAsync();
var doc = new HtmlDocument();
doc.LoadHtml(htmlContent);

HtmlNode divNode = doc.DocumentNode.SelectSingleNode("//div[@id='ga-maincontent']");
HtmlNodeCollection anchorTags = doc.DocumentNode
.SelectNodes("//div[@id='ga-maincontent']//a[@href]");

if (anchorTags != null)
{
foreach (HtmlNode anchorTag in anchorTags)
{
href = anchorTag.GetAttributeValue("href", string.Empty);

if (href.StartsWith("http"))
{
long uniqueId = Interlocked.Increment(ref counter);
links.Add(new Link
{
Id = $"{Guid.NewGuid().ToString()}{Interlocked.Increment(ref counter)}",
PropA= something
});
}
}
}
}
catch (Exception ex)
{
links.Add(new Link
{
Id = $"{Guid.NewGuid().ToString()}{Interlocked.Increment(ref counter)}",
PropA= something
});

_logger.LogInformation($"Http request exception :" +
$" Message :{ex.Message} | Status Code {ex.Source} | {ex.Source} ");
}

//if (i >= pages.) break;
}
}

string scannedLinks = string.Empty;

try
{
ss = JsonConvert.SerializeObject(links);
}
catch (Exception ex)
{
_logger.LogInformation($"Parallel Func Insert Links : {ex.Message} ");
}

return ss;
}

Раньше я использовал список вместо ConcurrentBag, но потом подумал, что это не потокобезопасно, почему приложения не выполняются так, как я ожидал. Тем не менее, я чувствую, что использую ключевые слова async и await, чтобы убедиться, что приложение запускается и ожидает завершения задач.
Что я действительно хочу сделать, так это: когда я зацикливаюсь через задачу: A, B, C и D будут завершены и все будут готовы сохранены в базе данных космоса, а затем мы перейдем к третьей функции.
Любое предположение о том, кто я поступать неправильно очень ценится.

Подробнее здесь: https://stackoverflow.com/questions/787 ... llel-issue