Skip to content

Commit 89f378a

Browse files
Rob-HagueRob-Hague
authored
Keep APM and sync UploadFile/DownloadFile callbacks on the threadpool (sshnet#1805)
* Tweak internal IProgress usage for APM and sync UploadFile/DownloadFile Changes to support IProgress<> callback on UploadAsync/DownloadAsync meant wrapping the Action<> callback on existing methods in a Progress<>, which posts the callback onto the current synchronisation context rather than the threadpool. For the legacy APM methods (Begin[..]), let's just preserve their old behaviour. For the synchronous methods, posting to the synchronisation context is probably the worst choice (because if there is one, the method itself is running there). We can either revert to the threadpool as well, or take the opportunity to invoke the callback synchronously, which is a behavioural change but probably the least surprising behaviour for a synchronous method. * keep callbacks on the threadpool Actually, we could call the Download callback synchronously easily enough, but the Upload progress reports are being made on the message listener thread upon request ack. A more involved scheme could drain callbacks to fire during the read loop. For now just make it all the same behaviour as in 2025.1.0.
1 parent fa98d59 commit 89f378a

2 files changed

Lines changed: 123 additions & 63 deletions

File tree

‎src/Renci.SshNet/SftpClient.cs‎

Lines changed: 43 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -906,7 +906,7 @@ public void DownloadFile(string path, Stream output, Action<ulong>? downloadCall
906906

907907
if (downloadCallback != null)
908908
{
909-
downloadProgress = new Progress<DownloadFileProgressReport>(r => downloadCallback(r.TotalBytesDownloaded));
909+
downloadProgress = new ThreadPoolProgress<DownloadFileProgressReport>(r => downloadCallback(r.TotalBytesDownloaded));
910910
}
911911

912912
InternalDownloadFile(
@@ -935,7 +935,7 @@ public Task DownloadFileAsync(string path, Stream output, IProgress<DownloadFile
935935
path,
936936
output,
937937
asyncResult: null,
938-
downloadProgress: downloadProgress,
938+
downloadProgress,
939939
isAsync: true,
940940
cancellationToken);
941941
}
@@ -1012,7 +1012,11 @@ public IAsyncResult BeginDownloadFile(string path, Stream output, AsyncCallback?
10121012

10131013
if (downloadCallback != null)
10141014
{
1015-
downloadProgress = new Progress<DownloadFileProgressReport>(r => downloadCallback(r.TotalBytesDownloaded));
1015+
// The System.Progress<T> ctor captures the current synchronization context
1016+
// and posts the progress reports to it. For back-compat with previous
1017+
// versions which always posted the callback to the threadpool regardless of
1018+
// sync context, we use a custom IProgress<T> impl.
1019+
downloadProgress = new ThreadPoolProgress<DownloadFileProgressReport>(r => downloadCallback(r.TotalBytesDownloaded));
10161020
}
10171021

10181022
var asyncResult = new SftpDownloadAsyncResult(asyncCallback, state);
@@ -1090,7 +1094,7 @@ public void UploadFile(Stream input, string path, bool canOverride, Action<ulong
10901094

10911095
if (uploadCallback != null)
10921096
{
1093-
uploadProgress = new Progress<UploadFileProgressReport>(r => uploadCallback(r.TotalBytesUploaded));
1097+
uploadProgress = new ThreadPoolProgress<UploadFileProgressReport>(r => uploadCallback(r.TotalBytesUploaded));
10941098
}
10951099

10961100
InternalUploadFile(
@@ -1274,7 +1278,11 @@ public IAsyncResult BeginUploadFile(Stream input, string path, bool canOverride,
12741278

12751279
if (uploadCallback != null)
12761280
{
1277-
uploadProgress = new Progress<UploadFileProgressReport>(r => uploadCallback(r.TotalBytesUploaded));
1281+
// The System.Progress<T> ctor captures the current synchronization context
1282+
// and posts the progress reports to it. For back-compat with previous
1283+
// versions which always posted the callback to the threadpool regardless of
1284+
// sync context, we use a custom IProgress<T> impl.
1285+
uploadProgress = new ThreadPoolProgress<UploadFileProgressReport>(r => uploadCallback(r.TotalBytesUploaded));
12781286
}
12791287

12801288
var asyncResult = new SftpUploadAsyncResult(asyncCallback, state);
@@ -2418,16 +2426,10 @@ private async Task InternalDownloadFile(
24182426

24192427
asyncResult?.Update(totalBytesRead);
24202428

2421-
if (downloadProgress is not null)
2429+
downloadProgress?.Report(new DownloadFileProgressReport()
24222430
{
2423-
// Copy offset to ensure it's not modified between now and execution of callback
2424-
var report = new DownloadFileProgressReport()
2425-
{
2426-
TotalBytesDownloaded = totalBytesRead,
2427-
};
2428-
2429-
downloadProgress.Report(report);
2430-
}
2431+
TotalBytesDownloaded = totalBytesRead
2432+
});
24312433
}
24322434
}
24332435
finally
@@ -2546,16 +2548,10 @@ private async Task InternalUploadFile(
25462548

25472549
asyncResult?.Update(writtenBytes);
25482550

2549-
// Call callback to report number of bytes written
2550-
if (uploadProgress is not null)
2551+
uploadProgress?.Report(new UploadFileProgressReport()
25512552
{
2552-
UploadFileProgressReport report = new()
2553-
{
2554-
TotalBytesUploaded = writtenBytes,
2555-
};
2556-
2557-
uploadProgress.Report(report);
2558-
}
2553+
TotalBytesUploaded = writtenBytes
2554+
});
25592555
}
25602556
finally
25612557
{
@@ -2662,5 +2658,29 @@ private ISftpSession CreateAndConnectToSftpSession()
26622658
throw;
26632659
}
26642660
}
2661+
2662+
/// <summary>
2663+
/// An <see cref="IProgress{T}"/> implementation that posts callbacks to the threadpool.
2664+
/// </summary>
2665+
private sealed class ThreadPoolProgress<T> : IProgress<T>
2666+
{
2667+
private readonly Action<T> _handler;
2668+
2669+
public ThreadPoolProgress(Action<T> handler)
2670+
{
2671+
Debug.Assert(handler != null);
2672+
_handler = handler!;
2673+
}
2674+
2675+
void IProgress<T>.Report(T value)
2676+
{
2677+
_ = ThreadPool.QueueUserWorkItem(static state =>
2678+
{
2679+
var (handler, value) = ((Action<T>, T))state!;
2680+
handler(value);
2681+
},
2682+
(_handler, value));
2683+
}
2684+
}
26652685
}
26662686
}

‎test/Renci.SshNet.IntegrationTests/OldIntegrationTests/SftpClientTest.Upload.cs‎

Lines changed: 80 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -324,77 +324,104 @@ public void Test_Sftp_Ensure_Async_Delegates_Called_For_BeginFileUpload_BeginFil
324324

325325
var remoteFileName = Path.GetRandomFileName();
326326
var localFileName = Path.GetRandomFileName();
327-
var uploadDelegateCalled = false;
328-
var downloadDelegateCalled = false;
329-
var listDirectoryDelegateCalled = false;
327+
using var uploadDelegateEvent = new ManualResetEventSlim();
328+
using var downloadDelegateEvent = new ManualResetEventSlim();
329+
using var listDirectoryDelegateEvent = new ManualResetEventSlim();
330+
using var uploadCallbackEvent = new ManualResetEventSlim();
331+
using var downloadCallbackEvent = new ManualResetEventSlim();
332+
using var listDirectoryCallbackEvent = new ManualResetEventSlim();
330333
IAsyncResult asyncResult;
331334

332335
// Test for BeginUploadFile.
333336

334337
CreateTestFile(localFileName, 1);
335338

336-
using (var fileStream = File.OpenRead(localFileName))
339+
var originalContext = SynchronizationContext.Current;
340+
try
337341
{
338-
asyncResult = sftp.BeginUploadFile(fileStream,
339-
remoteFileName,
340-
delegate (IAsyncResult ar)
341-
{
342-
sftp.EndUploadFile(ar);
343-
uploadDelegateCalled = true;
344-
},
345-
null);
346-
347-
while (!asyncResult.IsCompleted)
342+
// Set a throwing context to verify it's not captured by the callback
343+
SynchronizationContext.SetSynchronizationContext(new ThrowingSynchronizationContext());
344+
345+
using (var fileStream = File.OpenRead(localFileName))
348346
{
349-
Thread.Sleep(500);
347+
asyncResult = sftp.BeginUploadFile(fileStream,
348+
remoteFileName,
349+
delegate (IAsyncResult ar)
350+
{
351+
uploadDelegateEvent.Set();
352+
},
353+
state: null,
354+
uploadCallback: _ => uploadCallbackEvent.Set());
355+
356+
sftp.EndUploadFile(asyncResult);
350357
}
351358
}
359+
finally
360+
{
361+
SynchronizationContext.SetSynchronizationContext(originalContext);
362+
}
352363

353364
File.Delete(localFileName);
354365

355-
Assert.IsTrue(uploadDelegateCalled, "BeginUploadFile");
366+
Assert.IsTrue(uploadDelegateEvent.Wait(1000));
367+
Assert.IsTrue(uploadCallbackEvent.Wait(1000));
356368

357369
// Test for BeginDownloadFile.
358370

359371
asyncResult = null;
360-
using (var fileStream = File.OpenWrite(localFileName))
372+
try
361373
{
362-
asyncResult = sftp.BeginDownloadFile(remoteFileName,
363-
fileStream,
364-
delegate (IAsyncResult ar)
365-
{
366-
sftp.EndDownloadFile(ar);
367-
downloadDelegateCalled = true;
368-
},
369-
null);
374+
// Set a throwing context to verify it's not captured by the callback
375+
SynchronizationContext.SetSynchronizationContext(new ThrowingSynchronizationContext());
370376

371-
while (!asyncResult.IsCompleted)
377+
using (var fileStream = File.OpenWrite(localFileName))
372378
{
373-
Thread.Sleep(500);
379+
asyncResult = sftp.BeginDownloadFile(remoteFileName,
380+
fileStream,
381+
delegate (IAsyncResult ar)
382+
{
383+
downloadDelegateEvent.Set();
384+
},
385+
state: null,
386+
downloadCallback: _ => downloadCallbackEvent.Set());
387+
388+
sftp.EndDownloadFile(asyncResult);
374389
}
375390
}
391+
finally
392+
{
393+
SynchronizationContext.SetSynchronizationContext(originalContext);
394+
}
376395

377396
File.Delete(localFileName);
378397

379-
Assert.IsTrue(downloadDelegateCalled, "BeginDownloadFile");
398+
Assert.IsTrue(downloadDelegateEvent.Wait(1000));
399+
Assert.IsTrue(downloadCallbackEvent.Wait(1000));
380400

381401
// Test for BeginListDirectory.
382402

383-
asyncResult = null;
384-
asyncResult = sftp.BeginListDirectory(sftp.WorkingDirectory,
385-
delegate (IAsyncResult ar)
386-
{
387-
_ = sftp.EndListDirectory(ar);
388-
listDirectoryDelegateCalled = true;
389-
},
390-
null);
391-
392-
while (!asyncResult.IsCompleted)
403+
try
393404
{
394-
Thread.Sleep(500);
405+
// Set a throwing context to verify it's not captured by the callback
406+
SynchronizationContext.SetSynchronizationContext(new ThrowingSynchronizationContext());
407+
408+
asyncResult = sftp.BeginListDirectory(sftp.WorkingDirectory,
409+
delegate (IAsyncResult ar)
410+
{
411+
listDirectoryDelegateEvent.Set();
412+
},
413+
state: null,
414+
listCallback: _ => listDirectoryCallbackEvent.Set());
415+
416+
_ = sftp.EndListDirectory(asyncResult);
417+
}
418+
finally
419+
{
420+
SynchronizationContext.SetSynchronizationContext(originalContext);
395421
}
396422

397-
Assert.IsTrue(listDirectoryDelegateCalled, "BeginListDirectory");
423+
Assert.IsTrue(listDirectoryDelegateEvent.Wait(1000));
424+
Assert.IsTrue(listDirectoryCallbackEvent.Wait(1000));
398425
}
399426
}
400427

@@ -482,5 +509,18 @@ public async Task Test_Sftp_UploadFileAsync_UploadProgress()
482509
Assert.IsTrue(callbackCalled);
483510
}
484511
}
512+
513+
private sealed class ThrowingSynchronizationContext : SynchronizationContext
514+
{
515+
public override void Post(SendOrPostCallback d, object state)
516+
{
517+
throw new InvalidOperationException();
518+
}
519+
520+
public override void Send(SendOrPostCallback d, object state)
521+
{
522+
throw new InvalidOperationException();
523+
}
524+
}
485525
}
486526
}

0 commit comments

Comments
 (0)