- 
                Notifications
    You must be signed in to change notification settings 
- Fork 92
Implement incremental rate type #428
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
          
     Open
      
      
            albertoperdomo2
  wants to merge
  1
  commit into
  vllm-project:main
  
    
      
        
          
  
    
      Choose a base branch
      
     
    
      
        
      
      
        
          
          
        
        
          
            
              
              
              
  
           
        
        
          
            
              
              
           
        
       
     
  
        
          
            
          
            
          
        
       
    
      
from
albertoperdomo2:feature/incremental-load-reimpl
  
      
      
   
  
    
  
  
  
 
  
      
    base: main
Could not load branches
            
              
  
    Branch not found: {{ refName }}
  
            
                
      Loading
              
            Could not load tags
            
            
              Nothing to show
            
              
  
            
                
      Loading
              
            Are you sure you want to change the base?
            Some commits from the old base branch may be removed from the timeline,
            and old review comments may become outdated.
          
          
  
     Open
                    Changes from all commits
      Commits
    
    
  File filter
Filter by extension
Conversations
          Failed to load comments.   
        
        
          
      Loading
        
  Jump to
        
          Jump to file
        
      
      
          Failed to load files.   
        
        
          
      Loading
        
  Diff view
Diff view
There are no files selected for viewing
  
    
      This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
      Learn more about bidirectional Unicode characters
    
  
  
    
              
  
    
      This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
      Learn more about bidirectional Unicode characters
    
  
  
    
              
  
    
      This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
      Learn more about bidirectional Unicode characters
    
  
  
    
              
  
    
      This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
      Learn more about bidirectional Unicode characters
    
  
  
    
              
  
    
      This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
      Learn more about bidirectional Unicode characters
    
  
  
    
              
  
    
      This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
      Learn more about bidirectional Unicode characters
    
  
  
    
              | Original file line number | Diff line number | Diff line change | 
|---|---|---|
|  | @@ -12,6 +12,7 @@ | |
| from __future__ import annotations | ||
|  | ||
| import asyncio | ||
| import math | ||
| import random | ||
| import time | ||
| from abc import abstractmethod | ||
|  | @@ -25,6 +26,7 @@ | |
|  | ||
| __all__ = [ | ||
| "AsyncConstantStrategy", | ||
| "AsyncIncrementalStrategy", | ||
| "AsyncPoissonStrategy", | ||
| "ConcurrentStrategy", | ||
| "SchedulingStrategy", | ||
|  | @@ -36,7 +38,9 @@ | |
|  | ||
|  | ||
| StrategyType = Annotated[ | ||
| Literal["synchronous", "concurrent", "throughput", "constant", "poisson"], | ||
| Literal[ | ||
| "synchronous", "concurrent", "throughput", "constant", "poisson", "incremental" | ||
| ], | ||
| "Valid strategy type identifiers for scheduling request patterns", | ||
| ] | ||
|  | ||
|  | @@ -517,3 +521,114 @@ def request_completed(self, request_info: RequestInfo): | |
| :param request_info: Completed request metadata (unused) | ||
| """ | ||
| _ = request_info # request_info unused for async poisson strategy | ||
|  | ||
|  | ||
| @SchedulingStrategy.register("incremental") | ||
| class AsyncIncrementalStrategy(ThroughputStrategy): | ||
| """ | ||
| Incremental rate scheduling with gradual load increase over time. | ||
|  | ||
| Schedules requests starting at a base rate and incrementally increasing | ||
| the rate by a factor over time until reaching an optional rate limit. | ||
| Supports initial burst mode to quickly reach the target starting rate. | ||
| Useful for finding system saturation points or progressive load testing. | ||
| """ | ||
|  | ||
| type_: Literal["incremental"] = "incremental" # type: ignore[assignment] | ||
| start_rate: float = Field( | ||
| description="Initial rate at which to schedule requests in requests/second", | ||
| gt=0, | ||
| ) | ||
| increment_factor: float = Field( | ||
| description="Factor by which to increase the rate over time", | ||
| gt=0, | ||
| ) | ||
| rate_limit: int | None = Field( | ||
| default=None, | ||
| description="Maximum rate cap after which load remains constant", | ||
| gt=0, | ||
| ) | ||
| initial_burst: bool = Field( | ||
| default=True, | ||
| description=( | ||
| "Whether to send initial burst of math.floor(start_rate) requests " | ||
| "to reach target rate" | ||
| ), | ||
| ) | ||
|  | ||
| _process_offset: float | None = PrivateAttr(None) | ||
| _burst_sent: bool = PrivateAttr(False) | ||
|  | ||
| def __str__(self) -> str: | ||
| """ | ||
| :return: String identifier with start rate and increment factor | ||
| """ | ||
| return f"incremental@{self.start_rate:.2f}+{self.increment_factor:.2f}" | ||
|  | ||
| def init_processes_timings( | ||
| self, | ||
| worker_count: int, | ||
| max_concurrency: int, | ||
| startup_duration: float, | ||
| ): | ||
| """ | ||
| Initialize incremental-specific timing state. | ||
|  | ||
| :param worker_count: Number of worker processes to coordinate | ||
| :param max_concurrency: Maximum number of concurrent requests allowed | ||
| :param startup_duration: Duration in seconds for request startup ramping | ||
| """ | ||
| super().init_processes_timings(worker_count, max_concurrency, startup_duration) | ||
| with self._processes_lock: | ||
| self._process_offset = None | ||
|  | ||
| async def next_request_time(self, offset: int) -> float: | ||
| """ | ||
| Calculate next request time with incremental rate increase. | ||
|  | ||
| Implements gradual rate increase: rate = start_rate + (increment_factor * elapsed_time) | ||
| Optionally sends initial burst and caps at rate_limit. | ||
|  | ||
| :param offset: Unused for incremental strategy | ||
| :return: Next request time based on incremental rate calculation | ||
| """ | ||
| _ = offset # offset unused for incremental strategy | ||
| start_time = await self.get_processes_start_time() | ||
|  | ||
| # Handle initial burst if enabled | ||
| if self.initial_burst and not self._burst_sent: | ||
| self._burst_sent = True | ||
| burst_count = math.floor(self.start_rate) | ||
| for _ in range(burst_count): | ||
| pass | ||
| if self._process_offset is None: | ||
| self._process_offset = start_time | ||
|  | ||
| if self._process_offset is None: | ||
| self._process_offset = start_time | ||
|  | ||
| current_time = time.time() | ||
| if current_time <= start_time: | ||
| return start_time | ||
|  | ||
| # Calculate current rate based on elapsed time | ||
| elapsed_time = current_time - start_time | ||
| next_rate = self.start_rate + (self.increment_factor * elapsed_time) | ||
|  | ||
| # Cap at rate limit if specified | ||
| if self.rate_limit and next_rate >= self.rate_limit: | ||
| increment = 1.0 / self.rate_limit | ||
| else: | ||
| increment = 1.0 / next_rate | ||
|  | ||
| self._process_offset += increment | ||
| There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Strategies are shared across threads/processes, any assignments need to be atomic or locked with  | ||
|  | ||
| return self._process_offset | ||
|  | ||
| def request_completed(self, request_info: RequestInfo): | ||
| """ | ||
| Handle request completion (no-op for incremental strategy). | ||
|  | ||
| :param request_info: Completed request metadata (unused) | ||
| """ | ||
| _ = request_info # request_info unused for async incremental strategy | ||
      
      Oops, something went wrong.
        
    
  
  Add this suggestion to a batch that can be applied as a single commit.
  This suggestion is invalid because no changes were made to the code.
  Suggestions cannot be applied while the pull request is closed.
  Suggestions cannot be applied while viewing a subset of changes.
  Only one suggestion per line can be applied in a batch.
  Add this suggestion to a batch that can be applied as a single commit.
  Applying suggestions on deleted lines is not supported.
  You must change the existing code in this line in order to create a valid suggestion.
  Outdated suggestions cannot be applied.
  This suggestion has been applied or marked resolved.
  Suggestions cannot be applied from pending reviews.
  Suggestions cannot be applied on multi-line comments.
  Suggestions cannot be applied while the pull request is queued to merge.
  Suggestion cannot be applied right now. Please check back later.
  
    
  
    
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Generally we don't like to add top-level arguments that only map to a specific use-case. Many of the other components have a "kwargs" argument so it might make sense to add a
--profile-kwargs. Also--rateneeds to map to something so maybe make itincrement_factor(orstart_rate; whichever it makes more sense to sweep over).There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Another option might be to modify
--rateto take tuples of(start_rate, increment_factor).cc: @markurtz @jaredoconnell to weigh in on this.