datapath-windows: Packet subscribe handler

This change includes the following:
[1] Handler for subscribe/unsubscribe to a packet queue associated with a
    socket pid.
[2] Allocation of per socket packet queue on a packet subscription.
[3] Removal of static allocated queues.
[4] Freeing the packet queue (on user mode process termination).

Signed-off-by: Eitan Eliahu <eliahue@vmware.com>
Acked-by: Ankur Sharma <ankursharma@vmware.com>
Signed-off-by: Ben Pfaff <blp@nicira.com>

datapath-windows/include/OvsDpInterfaceExt.h

@@ -78,6 +78,7 @@ enum ovs_win_control_cmd {
     OVS_CTRL_CMD_WIN_GET_PID,
     OVS_CTRL_CMD_WIN_PEND_REQ,
     OVS_CTRL_CMD_MC_SUBSCRIBE_REQ,
+    OVS_CTRL_CMD_PACKET_SUBSCRIBE_REQ,
 
     /* This command is logically belong to the Vport family */
     OVS_CTRL_CMD_EVENT_NOTIFY,
@@ -86,8 +87,10 @@ enum ovs_win_control_cmd {
 
 /* NL Attributes for joining/unjoining an MC group */
 enum ovs_nl_mcast_attr {
-    OVS_NL_ATTR_MCAST_GRP,   /* (UINT32) Join an MC group */
+    OVS_NL_ATTR_MCAST_GRP,        /* (UINT32) Join an MC group */
-    OVS_NL_ATTR_MCAST_JOIN,  /* (UINT8) 1/0 - Join/Unjoin */
+    OVS_NL_ATTR_MCAST_JOIN,       /* (UINT8) 1/0 - Join/Unjoin */
+    OVS_NL_ATTR_PACKET_SUBSCRIBE, /* (UNINT8): 1/0 - subscribe/unsubscribe */
+    OVS_NL_ATTR_PACKET_PID,       /* (UNINT32) netlink PID to receive upcalls */
     __OVS_NL_ATTR_CTRL_MAX
 };
 #define OVS_WIN_CONTROL_ATTR_MAX (__OVS_NL_ATTR_CTRL_MAX - 1)

datapath-windows/ovsext/Datapath.c

@@ -23,6 +23,8 @@
 #if defined OVS_USE_NL_INTERFACE && OVS_USE_NL_INTERFACE == 1
 
 #include "precomp.h"
+#include "Switch.h"
+#include "User.h"
 #include "Datapath.h"
 #include "Jhash.h"
 #include "Switch.h"
@@ -90,6 +92,7 @@ typedef struct _NETLINK_FAMILY {
 static NetlinkCmdHandler OvsGetPidCmdHandler,
                          OvsPendEventCmdHandler,
                          OvsSubscribeEventCmdHandler,
+                         OvsSubscribePacketCmdHandler,
                          OvsReadEventCmdHandler,
                          OvsReadPacketCmdHandler,
                          OvsNewDpCmdHandler,
@@ -133,6 +136,11 @@ NETLINK_CMD nlControlFamilyCmdOps[] = {
       .supportedDevOp = OVS_WRITE_DEV_OP,
       .validateDpIndex = TRUE,
     },
+    { .cmd = OVS_CTRL_CMD_PACKET_SUBSCRIBE_REQ,
+      .handler = OvsSubscribePacketCmdHandler,
+      .supportedDevOp = OVS_WRITE_DEV_OP,
+      .validateDpIndex = TRUE,
+    },
     { .cmd = OVS_CTRL_CMD_EVENT_NOTIFY,
       .handler = OvsReadEventCmdHandler,
       .supportedDevOp = OVS_READ_EVENT_DEV_OP,
@@ -344,7 +352,6 @@ OvsInit()
     gOvsCtrlLock = &ovsCtrlLockObj;
     NdisAllocateSpinLock(gOvsCtrlLock);
     OvsInitEventQueue();
-    OvsUserInit();
 }
 
 VOID
@@ -355,7 +362,6 @@ OvsCleanup()
         NdisFreeSpinLock(gOvsCtrlLock);
         gOvsCtrlLock = NULL;
     }
-    OvsUserCleanup();
 }
 
 VOID
@@ -1070,7 +1076,7 @@ OvsSubscribeEventCmdHandler(POVS_USER_PARAMS_CONTEXT usrParamsCtx,
     POVS_MESSAGE msgIn = (POVS_MESSAGE)usrParamsCtx->inputBuffer;
 
     rc = NlAttrParse(&msgIn->nlMsg, sizeof (*msgIn),
-         NlMsgAttrsLen((PNL_MSG_HDR)msgIn), policy, attrs, 2);
+         NlMsgAttrsLen((PNL_MSG_HDR)msgIn), policy, attrs, ARRAY_SIZE(attrs));
     if (!rc) {
         status = STATUS_INVALID_PARAMETER;
         goto done;
@@ -2343,4 +2349,54 @@ OvsReadPacketCmdHandler(POVS_USER_PARAMS_CONTEXT usrParamsCtx,
                             usrParamsCtx->outputLength, replyLen);
     return status;
 }
+
+/*
+ * --------------------------------------------------------------------------
+ *  Handler for the subscription for a packet queue
+ * --------------------------------------------------------------------------
+ */
+static NTSTATUS
+OvsSubscribePacketCmdHandler(POVS_USER_PARAMS_CONTEXT usrParamsCtx,
+                            UINT32 *replyLen)
+{
+    NDIS_STATUS status;
+    BOOLEAN rc;
+    UINT8 join;
+    UINT32 pid;
+    const NL_POLICY policy[] =  {
+        [OVS_NL_ATTR_PACKET_PID] = {.type = NL_A_U32 },
+        [OVS_NL_ATTR_PACKET_SUBSCRIBE] = {.type = NL_A_U8 }
+        };
+    PNL_ATTR attrs[ARRAY_SIZE(policy)];
+
+    UNREFERENCED_PARAMETER(replyLen);
+
+    POVS_OPEN_INSTANCE instance =
+        (POVS_OPEN_INSTANCE)usrParamsCtx->ovsInstance;
+    POVS_MESSAGE msgIn = (POVS_MESSAGE)usrParamsCtx->inputBuffer;
+
+    rc = NlAttrParse(&msgIn->nlMsg, sizeof (*msgIn),
+         NlMsgAttrsLen((PNL_MSG_HDR)msgIn), policy, attrs, ARRAY_SIZE(attrs));
+    if (!rc) {
+        status = STATUS_INVALID_PARAMETER;
+        goto done;
+    }
+
+    join = NlAttrGetU8(attrs[OVS_NL_ATTR_PACKET_PID]);
+    pid = NlAttrGetU32(attrs[OVS_NL_ATTR_PACKET_PID]);
+
+    /* The socket subscribed with must be the same socket we perform receive*/
+    ASSERT(pid == instance->pid);
+
+    status = OvsSubscribeDpIoctl(instance, pid, join);
+
+    /*
+     * XXX Need to add this instance to a global data structure
+     * which hold all packet based instances. The data structure (hash)
+     * should be searched through the pid field of the instance for
+     * placing the missed packet into the correct queue
+     */
+done:
+    return status;
+}
 #endif /* OVS_USE_NL_INTERFACE */

datapath-windows/ovsext/Datapath.h

@@ -52,6 +52,10 @@ typedef struct _OVS_DEVICE_EXTENSION {
     INT pidCount;
 } OVS_DEVICE_EXTENSION, *POVS_DEVICE_EXTENSION;
 
+// forward declaration
+typedef struct _OVS_USER_PACKET_QUEUE OVS_USER_PACKET_QUEUE,
+                                      *POVS_USER_PACKET_QUEUE;
+
 /*
  * Private context for each handle on the device.
  */
@@ -59,7 +63,7 @@ typedef struct _OVS_OPEN_INSTANCE {
     UINT32 cookie;
     PFILE_OBJECT fileObject;
     PVOID eventQueue;
-    PVOID packetQueue;
+    POVS_USER_PACKET_QUEUE packetQueue;
     UINT32 pid;
 
     /*

datapath-windows/ovsext/Driver.c

@@ -16,6 +16,7 @@
 
 #include "precomp.h"
 #include "Switch.h"
+#include "User.h"
 #include "Datapath.h"
 
 #ifdef OVS_DBG_MOD

datapath-windows/ovsext/Event.c

@@ -16,8 +16,9 @@
 
 #include "precomp.h"
 
-#include "Datapath.h"
 #include "Switch.h"
+#include "User.h"
+#include "Datapath.h"
 #include "Vport.h"
 #include "Event.h"
 

datapath-windows/ovsext/Ioctl.c

@@ -109,7 +109,6 @@ OvsInit()
 {
     OvsInitIoctl();
     OvsInitEventQueue();
-    OvsUserInit();
 }
 
 VOID
@@ -117,7 +116,6 @@ OvsCleanup()
 {
     OvsCleanupEventQueue();
     OvsCleanupIoctl();
-    OvsUserCleanup();
 }
 
 VOID

datapath-windows/ovsext/User.c

@@ -22,11 +22,11 @@
 
 #include "precomp.h"
 
-#include "Datapath.h"
 #include "Switch.h"
 #include "Vport.h"
 #include "Event.h"
 #include "User.h"
+#include "Datapath.h"
 #include "PacketIO.h"
 #include "Checksum.h"
 #include "NetProto.h"
@@ -39,8 +39,6 @@
 #define OVS_DBG_MOD OVS_DBG_USER
 #include "Debug.h"
 
-OVS_USER_PACKET_QUEUE ovsPacketQueues[OVS_MAX_NUM_PACKET_QUEUES];
-
 POVS_PACKET_QUEUE_ELEM OvsGetNextPacket(POVS_OPEN_INSTANCE instance);
 extern PNDIS_SPIN_LOCK gOvsCtrlLock;
 extern POVS_SWITCH_CONTEXT gOvsSwitchContext;
@@ -50,34 +48,6 @@ static VOID _MapNlAttrToOvsPktExec(PNL_ATTR *nlAttrs, PNL_ATTR *keyAttrs,
                                    OvsPacketExecute  *execute);
 extern NL_POLICY nlFlowKeyPolicy[];
 
-NTSTATUS
-OvsUserInit()
-{
-    UINT32 i;
-    POVS_USER_PACKET_QUEUE queue;
-    for (i = 0; i < OVS_MAX_NUM_PACKET_QUEUES; i++) {
-        queue = &ovsPacketQueues[i];
-        RtlZeroMemory(queue, sizeof (*queue));
-        InitializeListHead(&queue->packetList);
-        NdisAllocateSpinLock(&queue->queueLock);
-    }
-    return STATUS_SUCCESS;
-}
-
-VOID
-OvsUserCleanup()
-{
-    UINT32 i;
-    POVS_USER_PACKET_QUEUE queue;
-    for (i = 0; i < OVS_MAX_NUM_PACKET_QUEUES; i++) {
-        queue = &ovsPacketQueues[i];
-        ASSERT(IsListEmpty(&queue->packetList));
-        ASSERT(queue->instance == NULL);
-        ASSERT(queue->pendingIrp == NULL);
-        NdisFreeSpinLock(&queue->queueLock);
-    }
-}
-
 static VOID
 OvsPurgePacketQueue(POVS_USER_PACKET_QUEUE queue,
                     POVS_OPEN_INSTANCE instance)
@@ -105,7 +75,6 @@ OvsPurgePacketQueue(POVS_USER_PACKET_QUEUE queue,
     }
 }
 
-
 VOID
 OvsCleanupPacketQueue(POVS_OPEN_INSTANCE instance)
 {
@@ -120,8 +89,11 @@ OvsCleanupPacketQueue(POVS_OPEN_INSTANCE instance)
     if (queue) {
         PDRIVER_CANCEL cancelRoutine;
         NdisAcquireSpinLock(&queue->queueLock);
+        ASSERT(queue->instance == instance);
+        /* XXX Should not happen */
         if (queue->instance != instance) {
             NdisReleaseSpinLock(&queue->queueLock);
+            NdisFreeSpinLock(&queue->queueLock);
             return;
         }
 
@@ -130,7 +102,6 @@ OvsCleanupPacketQueue(POVS_OPEN_INSTANCE instance)
             queue->numPackets = 0;
         }
         queue->instance = NULL;
-        queue->queueId = OVS_MAX_NUM_PACKET_QUEUES;
         instance->packetQueue = NULL;
         irp = queue->pendingIrp;
         queue->pendingIrp = NULL;
@@ -141,6 +112,7 @@ OvsCleanupPacketQueue(POVS_OPEN_INSTANCE instance)
             }
         }
         NdisReleaseSpinLock(&queue->queueLock);
+        NdisFreeSpinLock(&queue->queueLock);
     }
     LIST_FORALL_SAFE(&tmp, link, next) {
         RemoveEntryList(link);
@@ -150,44 +122,38 @@ OvsCleanupPacketQueue(POVS_OPEN_INSTANCE instance)
     if (irp) {
         OvsCompleteIrpRequest(irp, 0, STATUS_SUCCESS);
     }
+    if (queue) {
+        OvsFreeMemory(queue);
+    }
 }
 
 NTSTATUS
-OvsSubscribeDpIoctl(PFILE_OBJECT fileObject,
+OvsSubscribeDpIoctl(PVOID instanceP,
-                    PVOID inputBuffer,
+                    UINT32 pid,
-                    UINT32 inputLength)
+                    UINT8 join)
 {
-    POVS_OPEN_INSTANCE instance = (POVS_OPEN_INSTANCE)fileObject->FsContext;
-    UINT32 queueId;
     POVS_USER_PACKET_QUEUE queue;
-    if (inputLength < sizeof (UINT32)) {
+    POVS_OPEN_INSTANCE instance = (POVS_OPEN_INSTANCE)instanceP;
-        return STATUS_INVALID_PARAMETER;
+
-    }
+    if (instance->packetQueue && !join) {
-    queueId = *(UINT32 *)inputBuffer;
+        /* unsubscribe */
-    if (instance->packetQueue && queueId >= OVS_MAX_NUM_PACKET_QUEUES) {
-        /*
-         * unsubscribe
-         */
         OvsCleanupPacketQueue(instance);
-    } else if (instance->packetQueue == NULL &&
+    } else if (instance->packetQueue == NULL && join) {
-               queueId < OVS_MAX_NUM_PACKET_QUEUES) {
+        queue = (POVS_USER_PACKET_QUEUE) OvsAllocateMemory(sizeof *queue);
-        queue = &ovsPacketQueues[queueId];
+        if (queue == NULL) {
-        NdisAcquireSpinLock(&queue->queueLock);
+            return STATUS_NO_MEMORY;
-        if (ovsPacketQueues[queueId].instance) {
-             if (ovsPacketQueues[queueId].instance != instance) {
-                 NdisReleaseSpinLock(&queue->queueLock);
-                 return STATUS_INSUFFICIENT_RESOURCES;
-             } else {
-                 NdisReleaseSpinLock(&queue->queueLock);
-                 return STATUS_SUCCESS;
-             }
         }
-        queue->queueId = queueId;
+        instance->packetQueue = queue;
+        RtlZeroMemory(queue, sizeof (*queue));
+        NdisAllocateSpinLock(&queue->queueLock);
+        NdisAcquireSpinLock(&queue->queueLock);
+        InitializeListHead(&queue->packetList);
+        queue->pid = pid;
         queue->instance = instance;
         instance->packetQueue = queue;
-        ASSERT(IsListEmpty(&queue->packetList));
         NdisReleaseSpinLock(&queue->queueLock);
     } else {
+        /* user mode should call only once for subscribe */
         return STATUS_INVALID_PARAMETER;
     }
     return STATUS_SUCCESS;
@@ -623,14 +589,12 @@ OvsGetNextPacket(POVS_OPEN_INSTANCE instance)
 
 
 POVS_USER_PACKET_QUEUE
-OvsGetQueue(UINT32 queueId)
+OvsGetQueue(UINT32 pid)
 {
-    POVS_USER_PACKET_QUEUE queue;
+    /* XXX To be implemented. Return the queue assoiated with the pid*/
-    if (queueId >= OVS_MAX_NUM_PACKET_QUEUES) {
+    UNREFERENCED_PARAMETER(pid);
-        return NULL;
+    ASSERT(FALSE);
-    }
+    return NULL;
-    queue = &ovsPacketQueues[queueId];
-    return queue->instance != NULL ? queue : NULL;
 }
 
 VOID

datapath-windows/ovsext/User.h

@@ -21,6 +21,7 @@
 #ifndef __USER_H_
 #define __USER_H_ 1
 
+
 #include "Datapath.h"
 
 /*
@@ -28,7 +29,6 @@
  * more than 32 queues for processing packets to
  * userspace
  */
-#define OVS_MAX_NUM_PACKET_QUEUES 32
 #define OVS_DEFAULT_PACKET_QUEUE 1
 #define OVS_MAX_PACKET_QUEUE_LEN  4096
 
@@ -40,7 +40,7 @@
 #define OVS_MAX_PACKETS_PER_TUNNEL 1024
 
 typedef struct _OVS_USER_PACKET_QUEUE {
-    UINT32 queueId;
+    UINT32 pid;
     UINT32 numPackets;
     LIST_ENTRY  packetList;
     PVOID instance;
@@ -67,10 +67,6 @@ typedef struct _OVS_USER_STATS {
     UINT32 l4Csum;
 } OVS_USER_STATS, *POVS_USER_STATS;
 
-
-NTSTATUS OvsUserInit();
-VOID OvsUserCleanup();
-
 VOID OvsCleanupPacketQueue(struct _OVS_OPEN_INSTANCE *instance);
 
 POVS_PACKET_QUEUE_ELEM OvsCreateQueueNlPacket(PVOID userData,
@@ -96,14 +92,14 @@ NTSTATUS OvsCreateAndAddPackets(PVOID userData,
                                 LIST_ENTRY *list,
                                 UINT32 *num);
 
-NTSTATUS OvsSubscribeDpIoctl(PFILE_OBJECT fileObject,
+NTSTATUS OvsSubscribeDpIoctl(PVOID instanceP,
-                             PVOID inputBuffer,
+                             UINT32 pid,
-                             UINT32 inputLength);
+                             UINT8 join);
 
 NTSTATUS OvsReadDpIoctl(PFILE_OBJECT fileObject,
-                              PVOID outputBuffer,
+                        PVOID outputBuffer,
-                              UINT32 outputLength,
+                        UINT32 outputLength,
-                              UINT32 *replyLen);
+                        UINT32 *replyLen);
 NTSTATUS OvsExecuteDpIoctl(OvsPacketExecute *execute);
 NTSTATUS OvsPurgeDpIoctl(PFILE_OBJECT fileObject);