Skip to content

Commit 7f1a2e1

Browse files
TorstenDittmannTorstenDittmann
authored
Merge pull request #95 from appwrite/concurrent-chunk-uploads-1-9-x-minimal
feat: support concurrent chunk uploads
2 parents dc22cae + 67d86d0 commit 7f1a2e1

1 file changed

Lines changed: 171 additions & 40 deletions

File tree

‎Appwrite/Client.cs‎

Lines changed: 171 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
using System.Text;
88
using System.Text.Json;
99
using System.Text.Json.Serialization;
10+
using System.Threading;
1011
using System.Threading.Tasks;
1112
using Appwrite.Converters;
1213
using Appwrite.Extensions;
@@ -26,6 +27,7 @@ public class Client
2627
private string _endpoint;
2728

2829
private static readonly int ChunkSize = 5 * 1024 * 1024;
30+
private static readonly int MaxConcurrentUploads = 8;
2931

3032
public static JsonSerializerOptions DeserializerOptions { get; set; } = new JsonSerializerOptions
3133
{
@@ -267,14 +269,14 @@ private HttpRequestMessage PrepareRequest(
267269
{
268270
if (header.Key.Equals("content-type", StringComparison.OrdinalIgnoreCase))
269271
{
270-
_http.DefaultRequestHeaders.Accept.Add(new MediaTypeWithQualityHeaderValue(header.Value));
272+
request.Headers.Accept.Add(new MediaTypeWithQualityHeaderValue(header.Value));
271273
}
272274
else
273275
{
274-
if (_http.DefaultRequestHeaders.Contains(header.Key)) {
275-
_http.DefaultRequestHeaders.Remove(header.Key);
276+
if (request.Headers.Contains(header.Key)) {
277+
request.Headers.Remove(header.Key);
276278
}
277-
_http.DefaultRequestHeaders.Add(header.Key, header.Value);
279+
request.Headers.Add(header.Key, header.Value);
278280
}
279281
}
280282

@@ -454,7 +456,6 @@ public async Task<T> ChunkedUpload<T>(
454456
{
455457
case "path":
456458
var info = new FileInfo(input.Path);
457-
input.Data = info.OpenRead();
458459
size = info.Length;
459460
break;
460461
case "stream":
@@ -480,6 +481,8 @@ public async Task<T> ChunkedUpload<T>(
480481
switch(input.SourceType)
481482
{
482483
case "path":
484+
buffer = System.IO.File.ReadAllBytes(input.Path);
485+
break;
483486
case "stream":
484487
var dataStream = input.Data as Stream;
485488
if (dataStream == null)
@@ -509,87 +512,215 @@ public async Task<T> ChunkedUpload<T>(
509512
);
510513
}
511514

515+
var uploadId = string.Empty;
516+
512517
if (!string.IsNullOrEmpty(idParamName))
513518
{
514519
try
515520
{
516-
// Make a request to check if a file already exists
517-
var current = await Call<Dictionary<string, object?>>(
518-
method: "GET",
519-
path: $"{path}/{parameters[idParamName!]}",
520-
new Dictionary<string, string> { { "content-type", "application/json" } },
521-
parameters: new Dictionary<string, object?>()
522-
);
523-
if (current.TryGetValue("chunksUploaded", out var chunksUploadedValue) && chunksUploadedValue != null)
524-
{
525-
offset = Convert.ToInt64(chunksUploadedValue) * ChunkSize;
521+
// Make a request to check if a file already exists
522+
var current = await Call<Dictionary<string, object?>>(
523+
method: "GET",
524+
path: $"{path}/{parameters[idParamName!]}",
525+
new Dictionary<string, string> { { "content-type", "application/json" } },
526+
parameters: new Dictionary<string, object?>()
527+
);
528+
if (current.TryGetValue("chunksUploaded", out var chunksUploadedValue) && chunksUploadedValue != null)
529+
{
530+
offset = Convert.ToInt64(chunksUploadedValue) * ChunkSize;
531+
}
532+
result = current;
533+
uploadId = parameters[idParamName!]?.ToString() ?? string.Empty;
526534
}
527-
}
528535
catch
529536
{
530537
// ignored as it mostly means file not found
531538
}
532539
}
533540

534-
while (offset < size)
541+
var readLock = new object();
542+
543+
async Task<byte[]> ReadChunkAsync(long start, long end)
535544
{
545+
var length = (int)(end - start);
546+
var chunk = new byte[length];
547+
536548
switch(input.SourceType)
537549
{
538550
case "path":
551+
using (var chunkStream = System.IO.File.OpenRead(input.Path))
552+
{
553+
chunkStream.Seek(start, SeekOrigin.Begin);
554+
var read = 0;
555+
while (read < length)
556+
{
557+
var count = await chunkStream.ReadAsync(chunk, read, length - read);
558+
if (count == 0)
559+
break;
560+
read += count;
561+
}
562+
}
563+
break;
539564
case "stream":
540565
var stream = input.Data as Stream;
541566
if (stream == null)
542567
throw new InvalidOperationException("Stream data is null");
543-
stream.Seek(offset, SeekOrigin.Begin);
544-
await stream.ReadAsync(buffer, 0, ChunkSize);
568+
lock (readLock)
569+
{
570+
stream.Seek(start, SeekOrigin.Begin);
571+
var read = 0;
572+
while (read < length)
573+
{
574+
var count = stream.Read(chunk, read, length - read);
575+
if (count == 0)
576+
break;
577+
read += count;
578+
}
579+
}
545580
break;
546581
case "bytes":
547-
buffer = ((byte[])input.Data)
548-
.Skip((int)offset)
549-
.Take((int)Math.Min(size - offset, ChunkSize - 1))
550-
.ToArray();
582+
Buffer.BlockCopy((byte[])input.Data, (int)start, chunk, 0, length);
551583
break;
552584
}
553585

554-
var content = new MultipartFormDataContent {
555-
{ new ByteArrayContent(buffer), paramName, input.Filename }
586+
return chunk;
587+
}
588+
589+
async Task<Dictionary<string, object?>> UploadChunkAsync(int index, long start, long end, bool includeUploadId)
590+
{
591+
var chunkHeaders = new Dictionary<string, string>(headers)
592+
{
593+
["Content-Range"] = $"bytes {start}-{end - 1}/{size}"
556594
};
557595

558-
parameters[paramName] = content;
596+
if (includeUploadId && !string.IsNullOrEmpty(uploadId))
597+
{
598+
chunkHeaders["x-appwrite-id"] = uploadId;
599+
}
600+
601+
var content = new MultipartFormDataContent {
602+
{ new ByteArrayContent(await ReadChunkAsync(start, end)), paramName, input.Filename }
603+
};
559604

560-
headers["Content-Range"] =
561-
$"bytes {offset}-{Math.Min(offset + ChunkSize - 1, size - 1)}/{size}";
605+
var chunkParameters = new Dictionary<string, object?>(parameters)
606+
{
607+
[paramName] = content
608+
};
562609

563-
result = await Call<Dictionary<string, object?>>(
610+
var chunkResult = await Call<Dictionary<string, object?>>(
564611
method: "POST",
565612
path,
566-
headers,
567-
parameters
613+
chunkHeaders,
614+
chunkParameters
568615
);
569616

570-
offset += ChunkSize;
617+
if (index == 0)
618+
{
619+
uploadId = chunkResult.ContainsKey("$id")
620+
? chunkResult["$id"]?.ToString() ?? string.Empty
621+
: string.Empty;
622+
}
623+
624+
return chunkResult;
625+
}
626+
627+
if (offset == 0)
628+
{
629+
var firstChunkEnd = Math.Min(ChunkSize, size);
630+
result = await UploadChunkAsync(0, 0, firstChunkEnd, false);
631+
offset = firstChunkEnd;
571632

572-
var id = result.ContainsKey("$id")
573-
? result["$id"]?.ToString() ?? string.Empty
574-
: string.Empty;
575633
var chunksTotal = result.TryGetValue("chunksTotal", out var chunksTotalValue) && chunksTotalValue != null
576634
? Convert.ToInt64(chunksTotalValue)
577-
: 0L;
635+
: (long)Math.Ceiling(size / (double)ChunkSize);
578636
var chunksUploaded = result.TryGetValue("chunksUploaded", out var chunksUploadedValue) && chunksUploadedValue != null
579637
? Convert.ToInt64(chunksUploadedValue)
580638
: 0L;
581639

582-
headers["x-appwrite-id"] = id;
583-
584640
onProgress?.Invoke(
585641
new UploadProgress(
586-
id: id,
587-
progress: Math.Min(offset, size) / size * 100,
642+
id: uploadId,
643+
progress: Math.Min(offset, size) / (double)size * 100,
588644
sizeUploaded: Math.Min(offset, size),
589645
chunksTotal: chunksTotal,
590646
chunksUploaded: chunksUploaded));
591647
}
592648

649+
var chunks = new List<(int Index, long Start, long End)>();
650+
var chunkOffset = offset;
651+
var totalChunks = (long)Math.Ceiling(size / (double)ChunkSize);
652+
while (chunkOffset < size)
653+
{
654+
var end = Math.Min(chunkOffset + ChunkSize, size);
655+
chunks.Add(((int)(chunkOffset / ChunkSize), chunkOffset, end));
656+
chunkOffset = end;
657+
}
658+
659+
if (chunks.Count > 0)
660+
{
661+
var nextChunk = -1;
662+
var completedChunks = (long)(offset / ChunkSize);
663+
var uploadedBytes = offset;
664+
var resultLock = new object();
665+
var lastResult = result;
666+
Dictionary<string, object?>? completedResult = null;
667+
668+
bool IsUploadComplete(Dictionary<string, object?> chunkResult)
669+
{
670+
if (!chunkResult.TryGetValue("chunksUploaded", out var chunksUploadedValue) || chunksUploadedValue == null)
671+
{
672+
return false;
673+
}
674+
675+
var chunksUploaded = Convert.ToInt64(chunksUploadedValue);
676+
var chunksTotal = chunkResult.TryGetValue("chunksTotal", out var chunksTotalValue) && chunksTotalValue != null
677+
? Convert.ToInt64(chunksTotalValue)
678+
: totalChunks;
679+
680+
return chunksUploaded >= chunksTotal;
681+
}
682+
683+
var workers = Enumerable.Range(0, Math.Min(MaxConcurrentUploads, chunks.Count))
684+
.Select(async _ =>
685+
{
686+
while (true)
687+
{
688+
var chunkIndex = Interlocked.Increment(ref nextChunk);
689+
if (chunkIndex >= chunks.Count)
690+
break;
691+
692+
var chunk = chunks[chunkIndex];
693+
var chunkResult = await UploadChunkAsync(chunk.Index, chunk.Start, chunk.End, true);
694+
695+
lock (resultLock)
696+
{
697+
lastResult = chunkResult;
698+
if (IsUploadComplete(chunkResult))
699+
{
700+
completedResult = chunkResult;
701+
}
702+
}
703+
704+
var chunksUploaded = Interlocked.Increment(ref completedChunks);
705+
var sizeUploaded = Interlocked.Add(ref uploadedBytes, chunk.End - chunk.Start);
706+
var chunksTotal = chunkResult.TryGetValue("chunksTotal", out var chunksTotalValue) && chunksTotalValue != null
707+
? Convert.ToInt64(chunksTotalValue)
708+
: totalChunks;
709+
710+
onProgress?.Invoke(
711+
new UploadProgress(
712+
id: uploadId,
713+
progress: Math.Min(sizeUploaded, size) / (double)size * 100,
714+
sizeUploaded: Math.Min(sizeUploaded, size),
715+
chunksTotal: chunksTotal,
716+
chunksUploaded: chunksUploaded));
717+
}
718+
});
719+
720+
await Task.WhenAll(workers);
721+
result = completedResult ?? lastResult;
722+
}
723+
593724
// Convert to non-nullable dictionary for converter
594725
var nonNullableResult = result.Where(kvp => kvp.Value != null)
595726
.ToDictionary(kvp => kvp.Key, kvp => kvp.Value!);

0 commit comments

Comments
 (0)