Skip to content

Commit fbf90db

Browse files
mateofmateof
authored
Merge pull request #100 from mateof/feat/multi-connection-downloads
feat: multi-connection downloads to bypass Telegram's per-connection speed limit
2 parents a2129ee + 11181c1 commit fbf90db

4 files changed

Lines changed: 360 additions & 2 deletions

File tree

‎TelegramDownloader/Data/TelegramService.cs‎

Lines changed: 265 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -233,6 +233,263 @@ private async Task PrepareTransferClientAsync(int dcId)
233233
}
234234
}
235235

236+
#region Multi-connection downloads
237+
238+
// Telegram enforces its throughput limit PER CONNECTION (~5-6 MB/s), so a
239+
// single MTProto connection cannot go faster regardless of how many chunks
240+
// are pipelined. Official clients reach high speeds by opening several
241+
// sessions to the file's DC and splitting the file between them. This pool
242+
// holds extra authorized clients per DC (bootstrapped once from the main
243+
// client via auth.exportAuthorization/importAuthorization, persisted as
244+
// session files and reused across restarts).
245+
private class DcDownloadPool
246+
{
247+
public readonly SemaphoreSlim initLock = new SemaphoreSlim(1, 1);
248+
public readonly List<WTelegram.Client> clients = new List<WTelegram.Client>();
249+
public bool bootstrapFailed = false;
250+
}
251+
252+
private static readonly Dictionary<int, DcDownloadPool> downloadPools = new Dictionary<int, DcDownloadPool>();
253+
private const int MULTICONN_PART_SIZE = 1024 * 1024; // upload.getFile max limit per request
254+
private const int MULTICONN_BLOCK_SIZE = 4 * 1024 * 1024; // work unit assigned to a connection
255+
private const long MULTICONN_MIN_FILE_SIZE = 32L * 1024 * 1024;
256+
257+
public static int GetConfiguredDownloadConnections()
258+
{
259+
return Math.Clamp(GeneralConfigStatic.config?.DownloadConnections ?? 4, 2, 8);
260+
}
261+
262+
private static bool ShouldUseMultiConnection(TL.Document document)
263+
{
264+
GeneralConfig cfg = GeneralConfigStatic.config;
265+
return cfg != null && cfg.EnableMultiConnectionDownloads && document.size >= MULTICONN_MIN_FILE_SIZE;
266+
}
267+
268+
private async Task<List<WTelegram.Client>> GetDownloadPoolAsync(int dcId, int count)
269+
{
270+
DcDownloadPool pool;
271+
lock (downloadPools)
272+
{
273+
if (!downloadPools.TryGetValue(dcId, out pool))
274+
downloadPools[dcId] = pool = new DcDownloadPool();
275+
}
276+
if (pool.bootstrapFailed)
277+
return new List<WTelegram.Client>();
278+
await pool.initLock.WaitAsync();
279+
try
280+
{
281+
pool.clients.RemoveAll(c =>
282+
{
283+
if (!c.Disconnected) return false;
284+
try { c.Dispose(); } catch { }
285+
return true;
286+
});
287+
while (pool.clients.Count < count)
288+
{
289+
WTelegram.Client pc = await CreateDownloadPoolClientAsync(dcId, pool.clients.Count);
290+
if (pc == null)
291+
{
292+
// Do not retry the bootstrap on every download if the DC
293+
// refuses it (e.g. exportAuthorization not allowed).
294+
if (pool.clients.Count == 0)
295+
pool.bootstrapFailed = true;
296+
break;
297+
}
298+
pool.clients.Add(pc);
299+
}
300+
return pool.clients.Take(count).ToList();
301+
}
302+
finally
303+
{
304+
pool.initLock.Release();
305+
}
306+
}
307+
308+
private async Task<WTelegram.Client> CreateDownloadPoolClientAsync(int dcId, int index)
309+
{
310+
string sessionPath = Path.Combine(UserService.USERDATAFOLDER, $"WTelegram_dl_dc{dcId}_{index}.session");
311+
try
312+
{
313+
TL.Config tlConfig = await client.Help_GetConfig();
314+
DcOption dc = tlConfig.dc_options
315+
.Where(x => x.id == dcId && (x.flags & (DcOption.Flags.ipv6 | DcOption.Flags.cdn | DcOption.Flags.tcpo_only)) == 0)
316+
.OrderBy(x => (x.flags & DcOption.Flags.media_only) == 0 ? 0 : 1)
317+
.FirstOrDefault();
318+
if (dc == null)
319+
{
320+
_logger.LogWarning("No suitable address found for DC {Dc} - multi-connection download unavailable", dcId);
321+
return null;
322+
}
323+
string apiId = GeneralConfigStatic.tlconfig?.api_id ?? Environment.GetEnvironmentVariable("api_id");
324+
string apiHash = GeneralConfigStatic.tlconfig?.hash_id ?? Environment.GetEnvironmentVariable("hash_id");
325+
WTelegram.Client pc = new WTelegram.Client(what => what switch
326+
{
327+
"api_id" => apiId,
328+
"api_hash" => apiHash,
329+
"session_pathname" => sessionPath,
330+
"server_address" => $"{dc.ip_address}:{dc.port}",
331+
"device_model" => "TFM parallel download",
332+
_ => null
333+
});
334+
try
335+
{
336+
await pc.ConnectAsync();
337+
bool authorized = false;
338+
try
339+
{
340+
await pc.Users_GetUsers(InputUser.Self);
341+
authorized = true;
342+
}
343+
catch (RpcException)
344+
{
345+
// Fresh session, or a previously created one revoked from
346+
// the account's device list: (re)import the authorization.
347+
}
348+
if (!authorized)
349+
{
350+
Auth_ExportedAuthorization exported = await client.Auth_ExportAuthorization(dcId);
351+
await pc.Auth_ImportAuthorization(exported.id, exported.bytes);
352+
await pc.Users_GetUsers(InputUser.Self);
353+
}
354+
ApplyConfiguredParallelTransfers(pc);
355+
_logger.LogInformation("Download pool client {Index} ready for DC {Dc}", index, dcId);
356+
return pc;
357+
}
358+
catch
359+
{
360+
try { pc.Dispose(); } catch { }
361+
throw;
362+
}
363+
}
364+
catch (Exception ex)
365+
{
366+
_logger.LogWarning(ex, "Could not create download pool client {Index} for DC {Dc} - falling back to single-connection downloads", index, dcId);
367+
return null;
368+
}
369+
}
370+
371+
/// <summary>
372+
/// Downloads a document by splitting it in blocks served concurrently by
373+
/// several pool connections, writing each part at its absolute offset.
374+
/// Returns false (leaving the destination empty) when the pool is not
375+
/// available or the transfer failed in a recoverable way, so the caller
376+
/// can fall back to the standard sequential download.
377+
/// </summary>
378+
private async Task<bool> TryMultiConnectionDownloadAsync(TL.Document document, FileStream dest, DownloadModel model)
379+
{
380+
long size = document.size;
381+
List<WTelegram.Client> pool;
382+
try
383+
{
384+
pool = await GetDownloadPoolAsync(document.dc_id, GetConfiguredDownloadConnections());
385+
}
386+
catch (Exception ex)
387+
{
388+
_logger.LogWarning(ex, "Download pool unavailable for DC {Dc}", document.dc_id);
389+
return false;
390+
}
391+
if (pool.Count < 2)
392+
return false;
393+
394+
var location = new InputDocumentFileLocation
395+
{
396+
id = document.id,
397+
access_hash = document.access_hash,
398+
file_reference = document.file_reference,
399+
thumb_size = ""
400+
};
401+
402+
_logger.LogInformation("Multi-connection download - FileName: {Name}, Size: {SizeMB:F2}MB, Connections: {Connections}",
403+
model.name, size / (1024.0 * 1024.0), pool.Count);
404+
405+
long blockCount = (size + MULTICONN_BLOCK_SIZE - 1) / MULTICONN_BLOCK_SIZE;
406+
bool[] blockDone = new bool[blockCount];
407+
long confirmedBlocks = 0;
408+
long nextBlock = -1;
409+
object progressLock = new object();
410+
using CancellationTokenSource cts = new CancellationTokenSource();
411+
dest.SetLength(size);
412+
var handle = dest.SafeFileHandle;
413+
414+
void ReportPart(long block, int received, bool blockCompleted)
415+
{
416+
long confirmed;
417+
lock (progressLock)
418+
{
419+
if (blockCompleted)
420+
{
421+
blockDone[block] = true;
422+
while (confirmedBlocks < blockCount && blockDone[confirmedBlocks])
423+
confirmedBlocks++;
424+
}
425+
confirmed = Math.Min(size, confirmedBlocks * (long)MULTICONN_BLOCK_SIZE);
426+
}
427+
// Throws when the task gets canceled or paused, stopping the workers.
428+
model.ReportParallelProgress(confirmed, received, size);
429+
}
430+
431+
async Task Worker(WTelegram.Client pc)
432+
{
433+
while (!cts.IsCancellationRequested)
434+
{
435+
long block = Interlocked.Increment(ref nextBlock);
436+
if (block >= blockCount)
437+
return;
438+
long offset = block * (long)MULTICONN_BLOCK_SIZE;
439+
long end = Math.Min(size, offset + MULTICONN_BLOCK_SIZE);
440+
while (offset < end)
441+
{
442+
cts.Token.ThrowIfCancellationRequested();
443+
Upload_FileBase resp = null;
444+
for (int attempt = 1; ; attempt++)
445+
{
446+
try
447+
{
448+
resp = await pc.Upload_GetFile(location, offset, limit: MULTICONN_PART_SIZE);
449+
break;
450+
}
451+
catch (Exception) when (attempt < 3 && !cts.IsCancellationRequested)
452+
{
453+
await Task.Delay(1000 * attempt);
454+
}
455+
}
456+
if (resp is not Upload_File part)
457+
throw new InvalidOperationException($"Unexpected {resp?.GetType().Name} from Upload_GetFile (CDN-served files are not supported)");
458+
if (part.bytes.Length == 0)
459+
throw new InvalidOperationException($"Empty chunk at offset {offset}");
460+
RandomAccess.Write(handle, part.bytes, offset);
461+
offset += part.bytes.Length;
462+
ReportPart(block, part.bytes.Length, offset >= end);
463+
}
464+
}
465+
}
466+
467+
async Task GuardedWorker(WTelegram.Client pc)
468+
{
469+
try { await Worker(pc); }
470+
catch { cts.Cancel(); throw; }
471+
}
472+
473+
try
474+
{
475+
await Task.WhenAll(pool.Select(pc => Task.Run(() => GuardedWorker(pc))));
476+
await dest.FlushAsync();
477+
return true;
478+
}
479+
catch (Exception ex)
480+
{
481+
if (model.state == StateTask.Canceled || model.state == StateTask.Paused)
482+
throw;
483+
_logger.LogWarning(ex, "Multi-connection download failed, falling back to sequential - FileName: {Name}", model.name);
484+
dest.SetLength(0);
485+
dest.Position = 0;
486+
model._transmitted = 0;
487+
return false;
488+
}
489+
}
490+
491+
#endregion
492+
236493
public async Task<User> CallQrGenerator(Action<string> func, CancellationToken ct, bool logoutFirst = false)
237494
{
238495
return await client.LoginWithQRCode(func, logoutFirst: logoutFirst, ct: ct);
@@ -1430,8 +1687,14 @@ public async Task<string> DownloadFile(ChatMessages message, string fileName = n
14301687
_tis.addToDownloadList(model);
14311688
_logger.LogInformation("Starting file download to disk - FileName: {FileName}, Size: {SizeMB:F2}MB", filename, document.size / (1024.0 * 1024.0));
14321689
using var dest = new FileStream($"{Path.Combine(folder != null ? folder : Path.Combine(Environment.CurrentDirectory, "local", "temp"), filename)}", FileMode.Create, FileAccess.Write);
1433-
await PrepareTransferClientAsync(document.dc_id);
1434-
await client.DownloadFileAsync(document, dest, (PhotoSizeBase)null, model.ProgressCallback);
1690+
bool multiConnDone = false;
1691+
if (ShouldUseMultiConnection(document))
1692+
multiConnDone = await TryMultiConnectionDownloadAsync(document, dest, model);
1693+
if (!multiConnDone)
1694+
{
1695+
await PrepareTransferClientAsync(document.dc_id);
1696+
await client.DownloadFileAsync(document, dest, (PhotoSizeBase)null, model.ProgressCallback);
1697+
}
14351698
_logger.LogInformation("File download to disk completed - FileName: {FileName}", filename);
14361699
}
14371700
else if (message.message.media is MessageMediaPhoto { photo: Photo photo })

‎TelegramDownloader/Models/DownloadModel.cs‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -214,6 +214,46 @@ public void ProgressCallback(long transmitted, long totalSize)
214214
mutex.ReleaseMutex();
215215
}
216216

217+
/// <summary>
218+
/// Progress reporting for multi-connection downloads, where parts arrive
219+
/// out of order. <paramref name="confirmedPrefix"/> is the contiguous
220+
/// number of bytes completed from the start of the file (the only safe
221+
/// resume offset for persistence), while <paramref name="chunkBytes"/> is
222+
/// the size of the part just received, used for live speed accounting.
223+
/// Throws like ProgressCallback when the task is canceled or paused.
224+
/// </summary>
225+
public void ReportParallelProgress(long confirmedPrefix, int chunkBytes, long totalSize)
226+
{
227+
if (state == StateTask.Canceled)
228+
throw new Exception($"Canceled {name}");
229+
if (state == StateTask.Paused)
230+
{
231+
state = StateTask.Working;
232+
tis.deleteDownloadInList(this);
233+
throw new Exception($"Paused {name}");
234+
}
235+
tis.addDownloadBytes(chunkBytes);
236+
mutex.WaitOne();
237+
_transmitted = confirmedPrefix;
238+
_sizeString = HelperService.SizeSuffix(totalSize);
239+
_transmittedString = HelperService.SizeSuffix(confirmedPrefix);
240+
progress = Convert.ToInt32(confirmedPrefix * 100 / totalSize);
241+
EventChanged?.Invoke(this, new DownloadEventArgs());
242+
243+
OnProgressPersist?.Invoke(_transmitted, progress, state);
244+
245+
if (confirmedPrefix == totalSize)
246+
{
247+
endnDate = DateTime.Now;
248+
state = StateTask.Completed;
249+
EventStatechanged?.Invoke(this, EventArgs.Empty);
250+
NotificationModel nm = new NotificationModel();
251+
nm.sendEvent(new Notification($"Download {name} completed", "Download Completed", NotificationTypes.Success));
252+
tis.CheckPendingDownloads();
253+
}
254+
mutex.ReleaseMutex();
255+
}
256+
217257
public void Cancel()
218258
{
219259
mutex.WaitOne();

‎TelegramDownloader/Models/GeneralConfig.cs‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -189,6 +189,25 @@ public StreamingMode GetEffectiveStreamingMode()
189189
/// </summary>
190190
public int ParallelTransfers { get; set; } = 4;
191191

192+
// Multi-connection download settings
193+
/// <summary>
194+
/// EXPERIMENTAL: download large files using several parallel MTProto
195+
/// connections, the same technique Telegram Desktop uses to reach high
196+
/// speeds. Telegram limits throughput per connection (~5-6 MB/s), so a
197+
/// single connection cannot go faster no matter how many chunks are in
198+
/// flight; multiple connections each get their own allowance. Enabling
199+
/// this creates up to DownloadConnections extra sessions on the account
200+
/// (visible in Telegram's device list); they are created once, stored
201+
/// next to the main session file and reused across restarts.
202+
/// </summary>
203+
public bool EnableMultiConnectionDownloads { get; set; } = false;
204+
205+
/// <summary>
206+
/// Number of parallel connections used per file download (2-8).
207+
/// Only used when EnableMultiConnectionDownloads is true.
208+
/// </summary>
209+
public int DownloadConnections { get; set; } = 4;
210+
192211
}
193212

194213
public class TLConfig

‎TelegramDownloader/Pages/Config.razor‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -245,6 +245,42 @@
245245
</div>
246246
</div>
247247

248+
<div class="config-item">
249+
<div class="config-item-info">
250+
<div class="config-item-label">
251+
<i class="bi bi-diagram-3"></i>
252+
Multi-Connection Downloads
253+
</div>
254+
<div class="config-item-description">
255+
Download large files (&gt;32MB) using several parallel connections, like Telegram Desktop does.
256+
Telegram limits speed per connection (~5-6 MB/s), so this is the way to go faster.
257+
Creates extra sessions on your account (visible in Telegram's device list as "TFM parallel download"), created once and reused.
258+
<span class="badge bg-warning text-dark ms-2">Experimental</span>
259+
</div>
260+
</div>
261+
<div class="config-item-control">
262+
<InputCheckbox @bind-Value="Model!.EnableMultiConnectionDownloads" class="form-check-input config-switch" />
263+
</div>
264+
</div>
265+
266+
@if (Model!.EnableMultiConnectionDownloads)
267+
{
268+
<div class="config-item">
269+
<div class="config-item-info">
270+
<div class="config-item-label">
271+
<i class="bi bi-ethernet"></i>
272+
Download Connections
273+
</div>
274+
<div class="config-item-description">
275+
Number of parallel connections used per file download (2-8)
276+
</div>
277+
</div>
278+
<div class="config-item-control">
279+
<NumberInput TValue="int" @bind-Value="Model!.DownloadConnections" Min="2" Max="8" EnableMinMax="true" class="form-control" style="width: 80px;" />
280+
</div>
281+
</div>
282+
}
283+
248284
<div class="config-item">
249285
<div class="config-item-info">
250286
<div class="config-item-label">

0 commit comments

Comments
 (0)