background processing

aca/Controllers/BlobController.cs

@@ -13,16 +13,19 @@ public class BlobController : ControllerBase
     private readonly ILogger<BlobController> _logger;
     private IConfiguration _configuration;
 
+    private IBackgroundTaskQueue _queue;
+
     const string BLOB = "b";
     const string CONTAINER = "c";
     const string SAMPLE = "s";
 
     const string TEMP_LOC = "sample";
 
-    public BlobController(ILogger<BlobController> logger, IConfiguration configuration)
+    public BlobController(ILogger<BlobController> logger, IConfiguration configuration, IBackgroundTaskQueue queue)
     {
         _logger = logger;
         _configuration = configuration;
+        _queue = queue;
     }
 
 
@@ -52,12 +55,25 @@ public async Task<ActionResult<string>> CopyBlob(BlobRequest item)
             // copy single file
         }else if(CONTAINER.Equals(item.RequestType)){
             _logger.LogInformation($"BlobController::CopyBlob::Copy entire container content.");
+            // creating a background task
+            var workItem = new Func<CancellationToken, ValueTask>(async token =>
+                {
+                    _logger.LogInformation(
+                        $"Starting work item {item.RequestType} at: {DateTimeOffset.Now}");
+                    // do the copy here
+                    await CopyContainer(sourceBlobClient,targetBlobClient,5000,sas);
+                    _logger.LogInformation($"Work item {item.RequestType} completed at: {DateTimeOffset.Now}");
+
+                });
+            await _queue.QueueBackgroundWorkItemAsync(workItem);
+
+            return Accepted($"Copy container task Created: {item.SourceContainer} to {item.TargetContainer}");
             // copy entire container
-            await CopyContainer(sourceBlobClient,targetBlobClient,5000,sas);
-            return $"Copied container {item.SourceContainer} to {item.TargetContainer}";
+            // await CopyContainer(sourceBlobClient,targetBlobClient,5000,sas);
+            // return $"Copied container {item.SourceContainer} to {item.TargetContainer}";
         }else if(SAMPLE.Equals(item.RequestType) && !string.IsNullOrEmpty(item.BlobName)){
             _logger.LogInformation($"BlobController::CopyBlob::Creating samples.");   
-            long btime = DateTime.Now.Ticks; 
+            // long btime = DateTime.Now.Ticks; 
             // download the file to a temporary location (sample container)        
             BlobContainerClient localBlobClient = new BlobContainerClient(sourceCS,TEMP_LOC);
             localBlobClient.CreateIfNotExists();
@@ -66,10 +82,23 @@ public async Task<ActionResult<string>> CopyBlob(BlobRequest item)
             BlobClient localBlob = localBlobClient.GetBlobClient(localFileTemp);            
             Uri uri = new Uri(item.BlobName);            
             localBlob.StartCopyFromUri(uri);     
-            _logger.LogInformation($"BlobController::CopyBlob::Creating samples::local copy completed - copy to designated container started");  
-            await CreateSample(localBlob,targetBlobClient,item.SampleSize);
-            long etime = DateTime.Now.Ticks;            
-            return $"Created {item.SampleSize} samples in {item.TargetContainer}, took {(etime-btime)/TimeSpan.TicksPerSecond} seconds ";
+            _logger.LogInformation($"BlobController::CopyBlob::Creating samples::local copy completed - copy to designated container task starting.");  
+            // creating a background task
+            var workItem = new Func<CancellationToken, ValueTask>(async token =>
+                {
+                    _logger.LogInformation(
+                        $"Starting work item {item.RequestType} at: {DateTimeOffset.Now}");
+                    // do the copy here
+                    await CreateSample(localBlob,targetBlobClient,item.SampleSize);
+                    _logger.LogInformation($"Work item {item.RequestType} completed at: {DateTimeOffset.Now}");
+
+                });
+            await _queue.QueueBackgroundWorkItemAsync(workItem);
+
+            return Accepted($"Sample task Created: creating {item.SampleSize} samples in {item.TargetContainer}");
+
+            // long etime = DateTime.Now.Ticks;            
+            // return $"Created {item.SampleSize} samples in {item.TargetContainer}, took {(etime-btime)/TimeSpan.TicksPerSecond} seconds ";
         }else{
 
             // wrong type passed

aca/Program.cs

@@ -1,14 +1,13 @@
 using Azure.Identity;
-// using Microsoft.AspNetCore.HttpOverrides;
 
-// using Microsoft.OpenApi.Models;
+
 
 
 var builder = WebApplication.CreateBuilder(args);
 
-// // when running localy and your user is granted right rbac on the keyvault, u can use this:
+// when running localy and your user is granted right rbac on the keyvault, u can use this:
 // builder.Configuration.AddAzureKeyVault(
-//         new Uri($"https://{builder.Configuration["keyvault"]}.vault.azure.net/"),
+//         new Uri($"https://ext-lake-kv.vault.azure.net/"),
 //         new DefaultAzureCredential());
 
 builder.Configuration.AddAzureKeyVault(
@@ -21,6 +20,14 @@
 
 // Add services to the container.
 
+
+builder.Services.AddSingleton<IBackgroundTaskQueue>(ctx =>
+        {
+            return new BackgroundTaskQueue(100);
+        });
+
+builder.Services.AddHostedService<BackgroundWorker>();
+
 builder.Services.AddControllers();
 // Learn more about configuring Swagger/OpenAPI at https://aka.ms/aspnetcore/swashbuckle
 builder.Services.AddEndpointsApiExplorer();

aca/Workers/BackgroundWorker.cs

@@ -0,0 +1,52 @@
+
+public class BackgroundWorker : BackgroundService
+{
+    private readonly ILogger<BackgroundWorker> _logger;
+
+    public BackgroundWorker(IBackgroundTaskQueue taskQueue, 
+            ILogger<BackgroundWorker> logger)
+        {
+            TaskQueue = taskQueue;
+            _logger = logger;
+        }
+
+        public IBackgroundTaskQueue TaskQueue { get; }
+
+        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+        {
+            _logger.LogInformation("Queued Hosted Service running.");              
+            await BackgroundProcessing(stoppingToken);
+        }
+
+        private async Task BackgroundProcessing(CancellationToken stoppingToken)
+        {
+            while (!stoppingToken.IsCancellationRequested)
+            {
+
+
+                var workItem = 
+                    await TaskQueue.DequeueAsync(stoppingToken);
+                    _logger.LogInformation("Accepted task at: {time}", DateTimeOffset.Now);
+
+                try
+                {
+                    await workItem(stoppingToken);
+                    _logger.LogInformation("Completed task at: {time}", DateTimeOffset.Now);
+                }
+                catch (Exception ex)
+                {
+                    _logger.LogError(ex, 
+                        "Error occurred executing {WorkItem}.", nameof(workItem));
+                }
+            }
+        }
+
+        public override async Task StopAsync(CancellationToken stoppingToken)
+        {
+            _logger.LogInformation("Queued Hosted Service is stopping.");
+
+            await base.StopAsync(stoppingToken);
+        }
+}
+
+

aca/Workers/IbackgroundTaskQueue.cs

@@ -0,0 +1,48 @@
+using System.Threading.Channels;
+
+    public interface IBackgroundTaskQueue
+    {
+        ValueTask QueueBackgroundWorkItemAsync(Func<CancellationToken, ValueTask> workItem);
+
+        ValueTask<Func<CancellationToken, ValueTask>> DequeueAsync(
+            CancellationToken cancellationToken);
+    }
+
+    public class BackgroundTaskQueue : IBackgroundTaskQueue
+    {
+        private readonly Channel<Func<CancellationToken, ValueTask>> _queue;
+
+        public BackgroundTaskQueue(int capacity)
+        {
+            // Capacity should be set based on the expected application load and
+            // number of concurrent threads accessing the queue.            
+            // BoundedChannelFullMode.Wait will cause calls to WriteAsync() to return a task,
+            // which completes only when space became available. This leads to backpressure,
+            // in case too many publishers/calls start accumulating.
+            var options = new BoundedChannelOptions(capacity)
+            {
+                FullMode = BoundedChannelFullMode.Wait
+            };
+            _queue = Channel.CreateBounded<Func<CancellationToken, ValueTask>>(options);
+        }
+
+        public async ValueTask QueueBackgroundWorkItemAsync(
+            Func<CancellationToken, ValueTask> workItem)
+        {
+            if (workItem == null)
+            {
+                throw new ArgumentNullException(nameof(workItem));
+            }
+
+            await _queue.Writer.WriteAsync(workItem);
+        }
+
+        public async ValueTask<Func<CancellationToken, ValueTask>> DequeueAsync(
+            CancellationToken cancellationToken)
+        {
+            var workItem = await _queue.Reader.ReadAsync(cancellationToken);
+
+            return workItem;
+        }
+    }
+

aca/aca.csproj

@@ -6,7 +6,7 @@
     <ImplicitUsings>enable</ImplicitUsings>
   </PropertyGroup>
 
-  <ItemGroup>
+  <ItemGroup>    
     <PackageReference Include="Swashbuckle.AspNetCore" Version="6.2.3" />
     <PackageReference Include="Azure.Extensions.AspNetCore.Configuration.Secrets" Version="1.2.2" />
     <PackageReference Include="Azure.Identity" Version="1.6.1" />

aca/test.rest

@@ -12,6 +12,20 @@ Content-Type: application/json
   "requestType": "c"
 }
 
+### test background
+POST https://localhost:7179/api/Blob
+accept: text/plain
+Content-Type: application/json
+
+{
+  "sourceCS": "sourcecs",
+  "targetCS": "targetcs",
+  "blobName": "test2/auth4.gif",
+  "sourceContainer": "1000files",
+  "targetContainer": "test1000",
+  "requestType": "c"
+}
+
 ###
 # single blob copy
 POST https://localhost:7179/api/Blob
@@ -40,7 +54,7 @@ Content-Type: application/json
   "sourceContainer": "testdir",
   "targetContainer": "5stream",
   "requestType": "s",
-  "sampleSize": 5000
+  "sampleSize": 500
 }
 
 
@@ -59,3 +73,19 @@ Content-Type: application/json
   "requestType": "s",
   "SampleSize": 5
 }
+
+### test aca
+
+POST https://second--80qa8q9.jollyrock-6da7e977.northeurope.azurecontainerapps.io/api/Blob
+accept: text/plain
+Content-Type: application/json
+
+{
+  "sourceCS": "sourcecs",
+  "targetCS": "sourcecs",
+  "blobName": "https://raw.githubusercontent.com/soferreira/copy-alternatives/main/sample_data/daily.data",
+  "sourceContainer": "testdir",
+  "targetContainer": "batchingtest",
+  "requestType": "s",
+  "sampleSize": 200
+}