Skip to content

fix: remove mid-stream TPOT predictions from predicted-latency producer #2037

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.

Sign up for GitHub

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
MicheleCampi wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix: remove mid-stream TPOT predictions from predicted-latency producer #2037

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
111 changes: 0 additions & 111 deletions ...pp/framework/plugins/requestcontrol/dataproducer/predictedlatency/decode_token_sampler.go
View file
Open in desktop

This file was deleted.

96 changes: 0 additions & 96 deletions ...amework/plugins/requestcontrol/dataproducer/predictedlatency/decode_token_sampler_test.go
View file
Open in desktop

This file was deleted.

18 changes: 7 additions & 11 deletions pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/plugin.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -225,7 +225,10 @@ func (pl *PredictedLatency) readInFlightLoad(endpoint fwksched.Endpoint) inFligh
}

type Config struct {
SamplingMean float64 `json:"samplingMean,omitempty"`
// Deprecated: no longer used. Mid-stream TPOT predictions were removed;
// the value is accepted for config compatibility and ignored.
SamplingMean float64 `json:"samplingMean,omitempty"`
// Deprecated: no longer used. Accepted for config compatibility and ignored.
MaxDecodeTokenSamplesForPrediction int `json:"maxDecodeTokenSamplesForPrediction,omitempty"`
SLOBufferFactor float64 `json:"sloBufferFactor,omitempty"`
ContextTTL time.Duration `json:"contextTTL,omitempty"`
Expand Down Expand Up @@ -267,6 +270,9 @@ func PredictedLatencyFactory(name string, rawParameters *json.Decoder, handle pl
if handle == nil {
return nil, errors.New("plugin handle is required")
}
if parameters.SamplingMean != DefaultConfig.SamplingMean || parameters.MaxDecodeTokenSamplesForPrediction != DefaultConfig.MaxDecodeTokenSamplesForPrediction {
log.FromContext(handle.Context()).Info("Deprecated: samplingMean and maxDecodeTokenSamplesForPrediction are ignored; mid-stream TPOT predictions were removed")
}
if err := registerMetrics(handle.Metrics()); err != nil {
return nil, err
}
Expand All @@ -282,14 +288,6 @@ func PredictedLatencyFactory(name string, rawParameters *json.Decoder, handle pl
func (c *Config) validate() error {
var errs []error

if c.SamplingMean <= 0 {
errs = append(errs, fmt.Errorf("samplingMean must be > 0, got %f", c.SamplingMean))
}

if c.MaxDecodeTokenSamplesForPrediction < 0 {
errs = append(errs, fmt.Errorf("maxDecodeTokenSamplesForPrediction must be >= 0, got %d", c.MaxDecodeTokenSamplesForPrediction))
}

if c.SLOBufferFactor <= 0 {
errs = append(errs, fmt.Errorf("sloBufferFactor must be > 0, got %f", c.SLOBufferFactor))
}
Expand Down Expand Up @@ -370,8 +368,6 @@ type predictedLatencyCtx struct {
predictedTTFT float64
avgTPOT float64
avgPredictedTPOT float64
decodeTokenSampler *decodeTokenSampler
tpotObservations []float64
predictedTPOTObservations []float64

inputTokenCount int
Expand Down
14 changes: 4 additions & 10 deletions pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/plugin_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -391,16 +391,10 @@ func TestPredictedLatencyFactory(t *testing.T) {
expectErr: false,
},
{
name: "invalid samplingMean <= 0",
pluginName: "bad-sampling-mean",
jsonParams: `{"samplingMean": -1.0}`,
expectErr: true,
},
{
name: "invalid maxSampledTokens < 0",
pluginName: "bad-max-tokens",
jsonParams: `{"maxDecodeTokenSamplesForPrediction": -1}`,
expectErr: true,
name: "deprecated sampling params are accepted and ignored",
pluginName: "deprecated-sampling-params",
jsonParams: `{"samplingMean": -1.0, "maxDecodeTokenSamplesForPrediction": -1}`,
expectErr: false,
},
{
name: "invalid sloBufferFactor <= 0",
Expand Down
69 changes: 5 additions & 64 deletions ...pp/framework/plugins/requestcontrol/dataproducer/predictedlatency/requestcontrol_hooks.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -149,15 +149,15 @@ func (pl *PredictedLatency) ResponseBody(ctx context.Context, request *fwksched.

if predictedLatencyCtx.ttft == 0 {
if pl.config.StreamingMode && !response.EndOfStream {
processFirstTokenForLatencyPrediction(ctx, pl.latencypredictor, pl.config.StreamingMode, pl.config.EndpointRoleLabel, predictedLatencyCtx, now, pl.config.SamplingMean, pl.config.MaxDecodeTokenSamplesForPrediction)
processFirstTokenForLatencyPrediction(ctx, pl.latencypredictor, pl.config.StreamingMode, pl.config.EndpointRoleLabel, predictedLatencyCtx, now)
}
} else {
processTokenForLatencyPrediction(ctx, pl.typedName.Name, pl.typedName.Type, pl.latencypredictor, pl.config.EndpointRoleLabel, predictedLatencyCtx, targetMetadata, now, pl.config.SamplingMean, pl.config.MaxDecodeTokenSamplesForPrediction)
processTokenForLatencyPrediction(ctx, predictedLatencyCtx, now)
}

if response.EndOfStream {
if !pl.config.StreamingMode {
processFirstTokenForLatencyPrediction(ctx, pl.latencypredictor, pl.config.StreamingMode, pl.config.EndpointRoleLabel, predictedLatencyCtx, now, pl.config.SamplingMean, pl.config.MaxDecodeTokenSamplesForPrediction)
processFirstTokenForLatencyPrediction(ctx, pl.latencypredictor, pl.config.StreamingMode, pl.config.EndpointRoleLabel, predictedLatencyCtx, now)
}

if predictedLatencyCtx.ttft > 0 {
Expand Down Expand Up @@ -247,12 +247,9 @@ func processFirstTokenForLatencyPrediction(
endpointRoleLabel string,
predictedLatencyCtx *predictedLatencyCtx,
now time.Time,
samplingMean float64,
maxDecodeTokenSamplesForPrediction int,
) {
logger := log.FromContext(ctx)

initializeSampler(ctx, predictedLatencyCtx, samplingMean, maxDecodeTokenSamplesForPrediction)
predictedLatencyCtx.ttft = float64(now.Sub(predictedLatencyCtx.requestReceivedTimestamp).Milliseconds())
predictedLatencyCtx.generatedTokenCount = 1

Expand Down Expand Up @@ -286,15 +283,6 @@ func processFirstTokenForLatencyPrediction(
refreshLastSeenMetrics(ctx, predictedLatencyCtx)
}

func initializeSampler(ctx context.Context, predictedLatencyCtx *predictedLatencyCtx, samplingMean float64, maxDecodeTokenSamplesForPrediction int) {
if predictedLatencyCtx.decodeTokenSampler == nil {
logger := log.FromContext(ctx)
requestID := predictedLatencyCtx.schedulingRequest.Headers[reqcommon.RequestIDHeaderKey]
predictedLatencyCtx.decodeTokenSampler = newDecodeTokenSampler(requestID, samplingMean, maxDecodeTokenSamplesForPrediction)
logger.V(logutil.DEBUG).Info("Initialized token sampler for first token", "request_id", requestID, "next_prediction_token", predictedLatencyCtx.decodeTokenSampler.getNextSampleToken())
}
}

func predictFirstTPOT(ctx context.Context, predictedLatencyCtx *predictedLatencyCtx) {
logger := log.FromContext(ctx)
targetName := predictedLatencyCtx.targetMetadata.NamespacedName.Name
Expand All @@ -309,67 +297,20 @@ func predictFirstTPOT(ctx context.Context, predictedLatencyCtx *predictedLatency
}
}

// processTokenForLatencyPrediction records actual inter-token latency, sampled predictions, and advances timestamp.
// processTokenForLatencyPrediction records the actual TPOT for the token and advances the timestamp.
func processTokenForLatencyPrediction(
ctx context.Context,
pluginName, pluginType string,
predictor latencypredictor.PredictorInterface,
endpointRoleLabel string,
predictedLatencyCtx *predictedLatencyCtx,
targetEndpointMetadata *fwkdl.EndpointMetadata,
now time.Time,
samplingMean float64,
maxDecodeTokenSamplesForPrediction int,
) {
logger := log.FromContext(ctx)

if predictedLatencyCtx.decodeTokenSampler == nil {
requestID := predictedLatencyCtx.schedulingRequest.Headers[reqcommon.RequestIDHeaderKey]
predictedLatencyCtx.decodeTokenSampler = newDecodeTokenSampler(requestID, samplingMean, maxDecodeTokenSamplesForPrediction)
logger.V(logutil.DEBUG).Info("Initialized token sampler for subsequent tokens", "request_id", requestID, "next_prediction_token", predictedLatencyCtx.decodeTokenSampler.getNextSampleToken())
}

latencyMs := float64(now.Sub(predictedLatencyCtx.lastTokenTimestamp).Milliseconds())
predictedLatencyCtx.generatedTokenCount++

if predictedLatencyCtx.generatedTokenCount == 2 || predictedLatencyCtx.decodeTokenSampler.shouldPredict(predictedLatencyCtx.generatedTokenCount) {
predictedLatencyCtx.tpotObservations = append(predictedLatencyCtx.tpotObservations, latencyMs)
}
if predictedLatencyCtx.generatedTokenCount == 2 {
logger.V(logutil.DEBUG).Info("First inter-token latency observed",
"actual_tpot_ms", latencyMs,
"predicted_tpot_ms", predictedLatencyCtx.avgPredictedTPOT)
}

m, err := getLatestMetricsForProfile(predictedLatencyCtx, "")
if err != nil {
logger.V(logutil.DEBUG).Info("Skipping TPOT prediction due to missing metrics or schedulingResult", "error", err)
return
}

if predictedLatencyCtx.decodeTokenSampler.shouldPredict(predictedLatencyCtx.generatedTokenCount) {
in := buildPredictionRequest(
endpointRoleLabel,
targetEndpointMetadata,
m,
predictedLatencyCtx.inputTokenCount,
predictedLatencyCtx.generatedTokenCount,
0,
)
start := time.Now()
p, err := predictor.Predict(ctx, in)
dur := time.Since(start)
if err != nil || p == nil {
logger.V(logutil.DEBUG).Error(err, "TPOT predict failed", "duration_ms", dur.Milliseconds())
predictedLatencyCtx.predictedTPOTObservations = append(predictedLatencyCtx.predictedTPOTObservations, 0)
predictedLatencyCtx.avgPredictedTPOT = calculateRunningAverage(predictedLatencyCtx.avgPredictedTPOT, 0, len(predictedLatencyCtx.predictedTPOTObservations))
} else {
logger.V(logutil.DEBUG).Info("TPOT predict succeeded", "value_ms", p.TPOT, "duration_ms", dur.Milliseconds())
predictedLatencyCtx.predictedTPOTObservations = append(predictedLatencyCtx.predictedTPOTObservations, p.TPOT)
predictedLatencyCtx.avgPredictedTPOT = calculateRunningAverage(predictedLatencyCtx.avgPredictedTPOT, p.TPOT, len(predictedLatencyCtx.predictedTPOTObservations))
}
recordRequestTPOTPredictionDuration(ctx, pluginName, pluginType, predictedLatencyCtx.schedulingRequest.TargetModel, predictedLatencyCtx.incomingModelName, dur.Seconds())
predictedLatencyCtx.decodeTokenSampler.recordPrediction(predictedLatencyCtx.generatedTokenCount)
"actual_tpot_ms", latencyMs)
}

predictedLatencyCtx.lastTokenTimestamp = now
Expand Down
Loading
Loading