PktSrc: Avoid calling ExtractNextPacketInternal() in GetNextTimeout()

This reworks 2aec7640dd (zeek/zeek#2039) to
avoid calling ExtractNextPacketInternal() within GetNextTimeout() for
the non-pseudo-realtime case. Also relates to zeek/zeek#2842.

The intention of the referenced change was to avoid a 0.00002 timeout when
a non-selectable packet source has more packets queued. This was implemented
by checking for a new packet within GetNextTimeout().

The proposed change switches to an predictive approach: Use the result of
the previous ExtractNextPacket() call (stored as had_packet) as an indication
whether more packets are to be expected.

Calling ExtractNextPacketInternal() within GetNextTimeout() may cause
surprising behavior as some packet source may block [1] or spent a significant
amount of time (e.g. applying BPF filters [2]) within ExtractNextPacket().
The result of GetNextTimeout() should be available immediately as guidance
for the main-loop and the actual work should happen within the ->Process()
method.

This change also attempts to separate the pseudo-realtime logic from the
non-pseudo-realtime in an attempt show pseudo-realtime as special.

[1] 00c4d657e0/src/Napatech.cc (L116)
[2] 58b25c8eba/src/Myricom.cc (L250)

src/iosource/PktSrc.cc

@@ -30,6 +30,8 @@ PktSrc::Properties::Properties()
 PktSrc::PktSrc()
 	{
 	have_packet = false;
+	// Small lie to make a new PktSrc look like the previous ExtractNextPacket() was successful.
+	had_packet = true;
 	errbuf = "";
 	SetClosed(true);
 	}
@@ -179,6 +181,8 @@ bool PktSrc::ExtractNextPacketInternal()
 
 	if ( ExtractNextPacket(&current_packet) )
 		{
+		had_packet = true;
+
 		if ( current_packet.time < 0 )
 			{
 			Weird("negative_packet_timestamp", &current_packet);
@@ -191,6 +195,10 @@ bool PktSrc::ExtractNextPacketInternal()
 		have_packet = true;
 		return true;
 		}
+	else
+		{
+		had_packet = false;
+		}
 
 	if ( run_state::pseudo_realtime && ! IsOpen() )
 		{
@@ -276,33 +284,14 @@ bool PktSrc::GetCurrentPacket(const Packet** pkt)
 
 double PktSrc::GetNextTimeout()
 	{
-	bool pkt_available = have_packet;
-
-	if ( props.selectable_fd == -1 || run_state::pseudo_realtime )
-		pkt_available = ExtractNextPacketInternal();
-
 	if ( run_state::is_processing_suspended() )
 		return -1;
 
-	// If there's no file descriptor for the source, which is the case for some interfaces like
+	// If we're in pseudo-realtime mode, find the next time that a packet is ready
-	// myricom, we can't rely on the polling mechanism to wait for data to be available. As gross
+	// and have poll block until then.
-	// as it is, just spin with a short timeout here so that it will continually poll the
+	if ( run_state::pseudo_realtime )
-	// interface. The old IOSource code had a 20 microsecond timeout between calls to select()
-	// so just use that.
-	if ( props.selectable_fd == -1 )
 		{
-		if ( ! pkt_available && ! run_state::pseudo_realtime )
+		ExtractNextPacketInternal();
-			return 0.00002;
-		}
-	// If we're live we want poll to do what it has to with the file descriptor. If we're not live
-	// but we're not in pseudo-realtime mode, let the loop just spin as fast as it can. If we're
-	// in pseudo-realtime mode, find the next time that a packet is ready and have poll block until
-	// then.
-	else if ( IsLive() )
-		return -1;
-
-	if ( ! run_state::pseudo_realtime )
-		return 0;
 
 		// This duplicates the calculation used in run_state::check_pseudo_time().
 		double pseudo_time = current_packet.time - run_state::detail::first_timestamp;
@@ -311,4 +300,25 @@ double PktSrc::GetNextTimeout()
 		return std::max(0.0, pseudo_time - ct);
 		}
 
+	// If there's no file descriptor for the source, which is the case for some interfaces
+	// like myricom, we can't rely on the polling mechanism to wait for data to be
+	// available. As gross as it is, just spin with a short timeout here so that it will
+	// continually poll the interface. The old IOSource code had a 20 microsecond timeout
+	// between calls to select() so just use that.
+
+	// A heuristic to avoid short sleeps when a non-selectable packet source has more
+	// packets queued is to return 0.0 if the source has yielded a packet on the
+	// last call to ExtractNextPacket().
+	if ( props.selectable_fd == -1 )
+		{
+		if ( have_packet || had_packet )
+			return 0.0;
+
+		return 0.00002;
+		}
+
+	// If there's an FD (offline or live) we want poll to do what it has to with it.
+	return -1.0;
+	}
+
 	} // namespace zeek::iosource

src/iosource/PktSrc.h

@@ -362,6 +362,8 @@ private:
 
 	bool have_packet;
 	Packet current_packet;
+	// Did the previous call to ExtractNextPacket() yield a packet.
+	bool had_packet;
 
 	// For BPF filtering support.
 	std::vector<detail::BPF_Program*> filters;