1+ using ByteSync . Interfaces . Controls . Communications ;
2+
3+ namespace ByteSync . Services . Communications . Transfers . Uploading ;
4+
5+ public class AdaptiveUploadController : IAdaptiveUploadController
6+ {
7+ // Initial configuration
8+ private const int INITIAL_CHUNK_SIZE_BYTES = 500 * 1024 ; // 500 KB
9+ private const int MIN_PARALLELISM = 2 ;
10+ private const int MAX_PARALLELISM = 4 ;
11+
12+ // Thresholds
13+ private static readonly TimeSpan UpscaleThreshold = TimeSpan . FromSeconds ( 25 ) ;
14+ private static readonly TimeSpan DownscaleThreshold = TimeSpan . FromSeconds ( 30 ) ;
15+
16+ // Chunk size thresholds for parallelism increases
17+ private const int FOUR_MB = 4 * 1024 * 1024 ;
18+ private const int EIGHT_MB = 8 * 1024 * 1024 ;
19+
20+ // State
21+ private int _currentChunkSizeBytes ;
22+ private int _currentParallelism ;
23+ private readonly Queue < TimeSpan > _recentDurations ;
24+ private readonly Queue < int > _recentPartNumbers ;
25+ private readonly Queue < bool > _recentSuccesses ;
26+ private int _successesInWindow ;
27+ private int _windowSize ;
28+ private readonly ILogger < AdaptiveUploadController > _logger ;
29+
30+ public AdaptiveUploadController ( ILogger < AdaptiveUploadController > logger )
31+ {
32+ _logger = logger ;
33+ _currentChunkSizeBytes = INITIAL_CHUNK_SIZE_BYTES ;
34+ _currentParallelism = MIN_PARALLELISM ;
35+ _recentDurations = new Queue < TimeSpan > ( ) ;
36+ _recentPartNumbers = new Queue < int > ( ) ;
37+ _recentSuccesses = new Queue < bool > ( ) ;
38+ _windowSize = _currentParallelism ;
39+ }
40+
41+ public int CurrentChunkSizeBytes => _currentChunkSizeBytes ;
42+ public int CurrentParallelism => _currentParallelism ;
43+
44+ public int GetNextChunkSizeBytes ( )
45+ {
46+ return _currentChunkSizeBytes ;
47+ }
48+
49+ public void RecordUploadResult ( TimeSpan elapsed , bool isSuccess , int partNumber , int ? statusCode = null , Exception ? exception = null )
50+ {
51+ // Track window of last N uploads where N == current parallelism
52+ _recentDurations . Enqueue ( elapsed ) ;
53+ _recentPartNumbers . Enqueue ( partNumber ) ;
54+ _recentSuccesses . Enqueue ( isSuccess ) ;
55+ if ( isSuccess ) { _successesInWindow += 1 ; }
56+ while ( _recentDurations . Count > _windowSize )
57+ {
58+ _recentDurations . Dequeue ( ) ;
59+ if ( _recentPartNumbers . Count > 0 )
60+ {
61+ _recentPartNumbers . Dequeue ( ) ;
62+ }
63+ if ( _recentSuccesses . Count > 0 )
64+ {
65+ var removedSuccess = _recentSuccesses . Dequeue ( ) ;
66+ if ( removedSuccess && _successesInWindow > 0 )
67+ {
68+ _successesInWindow -= 1 ;
69+ }
70+ }
71+ }
72+
73+ // If error indicates bandwidth problems, reset chunk size
74+ if ( ! isSuccess && statusCode != null )
75+ {
76+ if ( statusCode == 429 || statusCode == 503 || statusCode == 507 )
77+ {
78+ _logger . LogWarning ( "Adaptive: bandwidth error status {Status}. Resetting chunk size to {InitialKb} KB (was {PrevKb} KB)" , statusCode , INITIAL_CHUNK_SIZE_BYTES / 1024 , _currentChunkSizeBytes / 1024 ) ;
79+ _currentChunkSizeBytes = INITIAL_CHUNK_SIZE_BYTES ;
80+ return ;
81+ }
82+ }
83+
84+ if ( _recentDurations . Count < _windowSize )
85+ {
86+ return ; // not enough data yet
87+ }
88+
89+ var maxElapsed = TimeSpan . Zero ;
90+ foreach ( var d in _recentDurations )
91+ {
92+ if ( d > maxElapsed ) maxElapsed = d ;
93+ }
94+
95+ _logger . LogDebug (
96+ "Adaptive: maxElapsedMs={MaxElapsedMs}, parallelism={Parallelism}, chunkKB={ChunkKb}" ,
97+ maxElapsed . TotalMilliseconds , _currentParallelism , _currentChunkSizeBytes / 1024 ) ;
98+
99+ // Downscale path first
100+ if ( maxElapsed > DownscaleThreshold )
101+ {
102+ // Adjust chunk size first (incremental), then parallelism in the same evaluation
103+ _currentChunkSizeBytes = ( int ) Math . Max ( 64 * 1024 , _currentChunkSizeBytes * 0.75 ) ;
104+ _logger . LogInformation (
105+ "Adaptive: Downscale. maxElapsedMs={MaxElapsedMs} > {ThresholdMs}. New chunkKB={ChunkKb}." ,
106+ maxElapsed . TotalMilliseconds , DownscaleThreshold . TotalMilliseconds , _currentChunkSizeBytes / 1024 ) ;
107+ if ( _currentParallelism > MIN_PARALLELISM )
108+ {
109+ _logger . LogInformation (
110+ "Adaptive: Downscale. Reducing parallelism {Prev} -> {Next}. Resetting window." ,
111+ _currentParallelism , _currentParallelism - 1 ) ;
112+ _currentParallelism -= 1 ;
113+ _windowSize = _currentParallelism ;
114+ }
115+ // Logging before resetting window done above; now reset
116+ ResetWindow ( ) ;
117+ return ;
118+ }
119+
120+ // Upscale when stable and fast (<= 25s) and all in window were successful
121+ if ( maxElapsed <= UpscaleThreshold && _successesInWindow >= _windowSize )
122+ {
123+ // Increase chunk size up to 25% per step, but target towards 25s heuristically
124+ // Simple rule: +25%
125+ var increased = ( int ) ( _currentChunkSizeBytes * 1.25 ) ;
126+ _currentChunkSizeBytes = increased ;
127+ _logger . LogInformation (
128+ "Adaptive: Upscale. maxElapsedMs={MaxElapsedMs} <= {ThresholdMs}. New chunkKB={ChunkKb}." ,
129+ maxElapsed . TotalMilliseconds , UpscaleThreshold . TotalMilliseconds , _currentChunkSizeBytes / 1024 ) ;
130+
131+ // Increase parallelism at thresholds of chunk size
132+ if ( _currentChunkSizeBytes >= EIGHT_MB )
133+ {
134+ var prev = _currentParallelism ;
135+ _currentParallelism = Math . Max ( _currentParallelism , 4 ) ;
136+ if ( _currentParallelism != prev )
137+ {
138+ _logger . LogInformation ( "Adaptive: Upscale. Increasing parallelism {Prev} -> {Next} due to chunk>=8MB." , prev , _currentParallelism ) ;
139+ }
140+ }
141+ else if ( _currentChunkSizeBytes >= FOUR_MB )
142+ {
143+ var prev = _currentParallelism ;
144+ _currentParallelism = Math . Max ( _currentParallelism , 3 ) ;
145+ if ( _currentParallelism != prev )
146+ {
147+ _logger . LogInformation ( "Adaptive: Upscale. Increasing parallelism {Prev} -> {Next} due to chunk>=4MB." , prev , _currentParallelism ) ;
148+ }
149+ }
150+ _currentParallelism = Math . Min ( _currentParallelism , MAX_PARALLELISM ) ;
151+ _windowSize = _currentParallelism ;
152+ }
153+ }
154+
155+ private void ResetWindow ( )
156+ {
157+ while ( _recentDurations . Count > 0 )
158+ {
159+ _recentDurations . Dequeue ( ) ;
160+ }
161+ while ( _recentPartNumbers . Count > 0 )
162+ {
163+ _recentPartNumbers . Dequeue ( ) ;
164+ }
165+ while ( _recentSuccesses . Count > 0 )
166+ {
167+ _recentSuccesses . Dequeue ( ) ;
168+ }
169+ _successesInWindow = 0 ;
170+ }
171+ }
0 commit comments