In such cases, another DPC is immediately scheduled, on the
understanding that DPC's for higher level drivers are likely to consume
those packets and free up resources in the interim.
In such a backpressure state, the RX ring is not drained until all
packets in the intermediate queue have been indicated up the stack.
Signed off by: Martin Harvey <martin.harvey@xxxxxxxxxx>
---
src/xenvif/receiver.c | 93 +++++++++++++++++++++++++++++++++----------
1 file changed, 73 insertions(+), 20 deletions(-)
diff --git a/src/xenvif/receiver.c b/src/xenvif/receiver.c
index 7295646..964207b 100644
--- a/src/xenvif/receiver.c
+++ b/src/xenvif/receiver.c
@@ -99,11 +99,14 @@ typedef struct _XENVIF_RECEIVER_RING {
BOOLEAN Connected;
BOOLEAN Enabled;
BOOLEAN Stopped;
+ BOOLEAN Backpressured;
+ BOOLEAN FinalFlush;
XENVIF_VIF_OFFLOAD_OPTIONS OffloadOptions;
ULONG BackfillSize;
PXENBUS_DEBUG_CALLBACK DebugCallback;
PXENVIF_THREAD WatchdogThread;
PLIST_ENTRY PacketQueue;
+ LONG PacketQueueEstSize; //Conservative estimate.
KDPC QueueDpc;
ULONG QueueDpcs;
LIST_ENTRY PacketComplete;
@@ -147,6 +150,8 @@ struct _XENVIF_RECEIVER {
#define XENVIF_RECEIVER_TAG 'ECER'
+#define RX_BUFFERING_MAX 2048 //X-ref XenNet IN_NDIS_MAX (half in NDIS half elsewhere)
+
static FORCEINLINE PVOID
__ReceiverAllocate(
IN ULONG Length
@@ -1337,43 +1342,56 @@ __ReceiverRingSwizzle(
PXENVIF_VIF_CONTEXT Context;
LIST_ENTRY List;
PLIST_ENTRY ListEntry;
+ BOOLEAN AllFlushed;
+ ULONG PushedUpstream;
Receiver = Ring->Receiver;
Frontend = Receiver->Frontend;
Context = PdoGetVifContext(FrontendGetPdo(Frontend));
+ AllFlushed = TRUE;
+ PushedUpstream = 0;
InitializeListHead(&List);
- ListEntry = InterlockedExchangePointer(&Ring->PacketQueue, NULL);
+ if (IsListEmpty(&Ring->PacketComplete) || Ring->FinalFlush)
+ {
+ Ring->PacketQueueEstSize = 0;
+ ListEntry = InterlockedExchangePointer(&Ring->PacketQueue, NULL);
- // Packets are held in the queue in reverse order so that the most
- // recent is always head of the list. This is necessary to allow
- // addition to the list to be done atomically.
+ // Packets are held in the queue in reverse order so that the most
+ // recent is always head of the list. This is necessary to allow
+ // addition to the list to be done atomically.
- while (ListEntry != NULL) {
- PLIST_ENTRY NextEntry;
+ while (ListEntry != NULL) {
+ PLIST_ENTRY NextEntry;
- NextEntry = ListEntry->Blink;
- ListEntry->Flink = ListEntry->Blink = ListEntry;
+ NextEntry = ListEntry->Blink;
+ ListEntry->Flink = ListEntry->Blink = ListEntry;
- InsertHeadList(&List, ListEntry);
+ InsertHeadList(&List, ListEntry);
- ListEntry = NextEntry;
- }
+ ListEntry = NextEntry;
+ }
- while (!IsListEmpty(&List)) {
- PXENVIF_RECEIVER_PACKET Packet;
+ while (!IsListEmpty(&List)) {
+ PXENVIF_RECEIVER_PACKET Packet;
- ListEntry = RemoveHeadList(&List);
- ASSERT3P(ListEntry, !=, &List);
+ ListEntry = RemoveHeadList(&List);
+ ASSERT3P(ListEntry, !=, &List);
- RtlZeroMemory(ListEntry, sizeof (LIST_ENTRY));
+ RtlZeroMemory(ListEntry, sizeof (LIST_ENTRY));
- Packet = CONTAINING_RECORD(ListEntry, XENVIF_RECEIVER_PACKET, ListEntry);
- ReceiverRingProcessPacket(Ring, Packet);
+ Packet = CONTAINING_RECORD(ListEntry, XENVIF_RECEIVER_PACKET,
ListEntry);
+ ReceiverRingProcessPacket(Ring, Packet);
+ }
+ }
+ else
+ {
+ AllFlushed = FALSE;
}
- while (!IsListEmpty(&Ring->PacketComplete)) {
+ while ((!IsListEmpty(&Ring->PacketComplete))
+ && ((PushedUpstream < RX_BUFFERING_MAX) || Ring->FinalFlush)) {
PXENVIF_RECEIVER_PACKET Packet;
PXENVIF_PACKET_INFO Info;
PUCHAR BaseVa;
@@ -1381,6 +1399,8 @@ __ReceiverRingSwizzle(
PETHERNET_ADDRESS DestinationAddress;
ETHERNET_ADDRESS_TYPE Type;
+ PushedUpstream++;
+
ListEntry = RemoveHeadList(&Ring->PacketComplete);
ASSERT3P(ListEntry, !=, &Ring->PacketComplete);
@@ -1536,9 +1556,33 @@ __ReceiverRingSwizzle(
Packet->TagControlInformation,
&Packet->Info,
&Packet->Hash,
- !IsListEmpty(&Ring->PacketComplete) ? TRUE :
FALSE,
+ ((!IsListEmpty(&Ring->PacketComplete))
+ && ((PushedUpstream < RX_BUFFERING_MAX) ||
Ring->FinalFlush)) ? TRUE : FALSE,
Packet);
}
+
+ if (!IsListEmpty(&Ring->PacketComplete))
+ AllFlushed = FALSE;
+
+ if (!AllFlushed)
+ {
+ //Re-run remainder from back of DPC queue.
+ Ring->Backpressured = TRUE;
+ if (KeInsertQueueDpc(&Ring->QueueDpc, NULL, NULL))
+ Ring->QueueDpcs++;
+ }
+ else
+ {
+ if ((Ring->Backpressured) && !Ring->FinalFlush)
+ {
+ //Not any more - restart dataflow from initial ring poll.
+ Ring->Backpressured = FALSE;
+
+ //PollDpc zeroed before final flush, don't queue it here.
+ if (KeInsertQueueDpc(&Ring->PollDpc, NULL, NULL))
+ Ring->PollDpcs++;
+ }
+ }
}
static FORCEINLINE VOID
@@ -1973,6 +2017,8 @@ __ReceiverRingQueuePacket(
ListEntry->Blink = Ring->PacketQueue;
New = ListEntry;
} while (InterlockedCompareExchangePointer(&Ring->PacketQueue,
(PVOID)New, (PVOID)Old) != Old);
+
+ InterlockedIncrement(&Ring->PacketQueueEstSize);
}
static DECLSPEC_NOINLINE ULONG
@@ -1992,6 +2038,9 @@ ReceiverRingPoll(
if (!Ring->Enabled)
goto done;
+ if ((Ring->PacketQueueEstSize >= RX_BUFFERING_MAX) && !Ring->FinalFlush)
+ goto done;
+
for (;;) {
BOOLEAN Error;
BOOLEAN Extra;
@@ -2949,8 +2998,12 @@ __ReceiverRingTeardown(
Ring->BackfillSize = 0;
Ring->OffloadOptions.Value = 0;
+ Ring->FinalFlush = TRUE;
+ KeInsertQueueDpc(&Ring->QueueDpc, NULL, NULL);
KeFlushQueuedDpcs();
RtlZeroMemory(&Ring->QueueDpc, sizeof (KDPC));
+ Ring->Backpressured = FALSE;
+ Ring->FinalFlush = FALSE;
ThreadAlert(Ring->WatchdogThread);
ThreadJoin(Ring->WatchdogThread);
