5.2 Stream API with nano‐second timeout precision

Nanosecond-Precision Stream Timeout Implementation

Overview

This implementation enhances the StreamProcessor with high-precision timeout detection for missing or invalid stream segments. It provides nanosecond-level precision for timing measurement and supports fine-grained control over stream lifecycle management.

Table of Contents

• Core Components
• Timeout Detection
• Configuration
• Usage Examples
• Advanced Features
• Performance Considerations
• Implementation Details

Core Components

StreamSegmentTracker

The primary component responsible for tracking segment gaps and timeouts:

public class StreamSegmentTracker {
    private final NavigableMap<Long, StreamSegment> segments;
    private final long timeoutNanos;
    private volatile long expectedSequence;
    private volatile long lastActivityNanos;
}

Key responsibilities:

• Maintains ordered sequence of stream segments
• Tracks timing information at nanosecond precision
• Detects gaps in sequence space
• Monitors inactivity periods

SequenceGap

Represents missing segments in the stream:

public record SequenceGap(
    long startSequence,
    long length
) implements Comparable<SequenceGap>

StreamTimeoutEvent

Encapsulates timeout information:

public record StreamTimeoutEvent(
    StreamKey key,
    StreamMetadata metadata,
    Set<SequenceGap> gaps,
    long lastActivityTime
) {
    public Duration getInactiveDuration() {
        return Duration.ofNanos(System.nanoTime() - lastActivityTime);
    }
}

Timeout Detection

Mechanism

1. Activity Tracking

   private void updateActivity() {
       lastActivityNanos = System.nanoTime();
   }

2. Timeout Checking

   public boolean hasTimedOut() {
       long now = System.nanoTime();
       return (now - lastActivityNanos) > timeoutNanos;
   }

3. Periodic Monitoring

   timeoutExecutor.scheduleAtFixedRate(
       this::checkTimeouts,
       timeoutNanos, 
       timeoutNanos, 
       TimeUnit.NANOSECONDS
   );

Gap Detection

The system maintains a sorted map of segments and identifies gaps:

public Set<SequenceGap> getGaps() {
    Set<SequenceGap> gaps = new TreeSet<>();
    if (segments.isEmpty()) return gaps;
    
    long current = expectedSequence;
    for (Map.Entry<Long, StreamSegment> entry : segments.entrySet()) {
        if (entry.getKey() > current) {
            gaps.add(new SequenceGap(current, entry.getKey() - current));
        }
        current = entry.getKey() + entry.getValue().getLength();
    }
    return gaps;
}

Configuration

Timeout Settings

// Create with 1-second timeout
StreamProcessor processor = new StreamProcessor(
    TimeUnit.SECONDS.toNanos(1)
);

// Create with custom timeout
long timeoutNanos = TimeUnit.MILLISECONDS.toNanos(500);
StreamProcessor processor = new StreamProcessor(timeoutNanos);

Thread Configuration

ThreadFactory factory = r -> {
    Thread t = new Thread(r, "Stream-Timeout-Monitor");
    t.setDaemon(true);
    return t;
};

ScheduledExecutorService executor = 
    Executors.newSingleThreadScheduledExecutor(factory);

Usage Examples

Basic Usage

// Create processor
StreamProcessor processor = new StreamProcessor(
    TimeUnit.SECONDS.toNanos(1)
);

// Add timeout handler
processor.addTimeoutListener(event -> {
    System.out.printf(
        "Stream %s timed out after %s%n",
        event.key(),
        event.getInactiveDuration()
    );
    
    // Log gap information
    event.gaps().forEach(gap -> 
        System.out.printf(
            "Gap at sequence %d, length %d%n",
            gap.startSequence(),
            gap.length()
        )
    );
});

// Process packets
processor.processPacket(packet);

Recovery Handling

processor.addTimeoutListener(event -> {
    if (event.gaps().size() <= 3) {
        // Request retransmission for small gaps
        requestRetransmission(event.key(), event.gaps());
    } else {
        // Close stream for larger gaps
        processor.closeStream(event.key());
    }
});

Advanced Features

Partial Stream Recovery

public Stream buildPartial() {
    return new StreamImpl(
        packets,
        ByteBuffer.wrap(streamData.toByteArray()),
        buildMetadata(),
        false  // incomplete flag
    );
}

Custom Timeout Strategies

public interface TimeoutStrategy {
    boolean shouldTimeout(
        StreamContext context,
        Set<SequenceGap> gaps,
        Duration inactivity
    );
}

Performance Considerations

Memory Usage

• Uses ConcurrentSkipListMap for efficient concurrent access
• Implements weak references for packet storage
• Provides configurable buffer sizes

Thread Safety

• Thread-safe segment tracking
• Non-blocking timeout detection
• Lock-free activity updates

Timing Precision

• Uses System.nanoTime() for high-precision timing
• Accounts for system clock resolution
• Handles timer wraparound

Implementation Details

Timeout Detection Loop

private void checkTimeouts() {
    long now = System.nanoTime();
    activeStreams.forEach((key, context) -> {
        if (context.hasTimedOut()) {
            handleTimeout(key, context);
        }
    });
}

Segment Processing

private void processContiguousSegments() {
    while (!segments.isEmpty()) {
        Map.Entry<Long, StreamSegment> entry = segments.firstEntry();
        if (entry.getKey() != expectedSequence) {
            break;
        }
        StreamSegment segment = segments.remove(entry.getKey());
        processSegment(segment);
        expectedSequence = segment.getSequenceNumber() + 
                          segment.getLength();
    }
}

Clean Shutdown

public void shutdown() {
    timeoutExecutor.shutdown();
    try {
        if (!timeoutExecutor.awaitTermination(
                timeoutNanos, 
                TimeUnit.NANOSECONDS)) {
            timeoutExecutor.shutdownNow();
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        timeoutExecutor.shutdownNow();
    }
}

Error Handling

Invalid Segments

try {
    StreamSegment segment = extractSegment(packet);
    context.getSegmentTracker().addSegment(segment);
} catch (InvalidSegmentException e) {
    handleInvalidSegment(key, packet, e);
}

Timeout Recovery

private void handleTimeout(StreamKey key, StreamContext context) {
    Set<SequenceGap> gaps = context.getSegmentTracker().getGaps();
    StreamTimeoutEvent event = new StreamTimeoutEvent(
        key,
        context.getMetadata(),
        gaps,
        context.getLastActivityTime()
    );
    
    notifyTimeoutListeners(event);
    
    if (shouldCloseStream(event)) {
        closeStream(key);
    }
}

Would you like me to:

1. Add monitoring and metrics documentation?
2. Expand the error handling section?
3. Include integration examples with different network protocols?