Basic implementation of send now works. (Rev 79)
authortzachid <tzachid@ad392aa1-c5ef-ae45-8dd8-e69d62a5ef86>
Mon, 21 Nov 2005 12:09:40 +0000 (12:09 +0000)
committertzachid <tzachid@ad392aa1-c5ef-ae45-8dd8-e69d62a5ef86>
Mon, 21 Nov 2005 12:09:40 +0000 (12:09 +0000)
git-svn-id: svn://openib.tc.cornell.edu/gen1/trunk@164 ad392aa1-c5ef-ae45-8dd8-e69d62a5ef86

ulp/sdp/kernel/Precompile.h
ulp/sdp/kernel/SOURCES
ulp/sdp/kernel/SdpBufferPool.cpp [new file with mode: 0644]
ulp/sdp/kernel/SdpBufferPool.h [new file with mode: 0644]
ulp/sdp/kernel/SdpGenUtils.cpp
ulp/sdp/kernel/SdpGenUtils.h
ulp/sdp/kernel/SdpLock.h [new file with mode: 0644]
ulp/sdp/kernel/SdpSocket.cpp
ulp/sdp/kernel/SdpSocket.h
ulp/sdp/kernel/SdpTrace.cpp
ulp/sdp/kernel/SdpTrace.h

index ca6a3bc..d2079c2 100644 (file)
@@ -16,12 +16,14 @@ class SdpArp;
 #include "ib_al.h"\r
 \r
 #include "sdpMsgs.h"\r
+#include "SdpGenUtils.h"\r
 #include "SdpTrace.h"\r
+#include "sdpLock.h"\r
 #include "RefCount.h"\r
 #include "sdpdriver.h"\r
 #include "SdpShared.h"\r
 #include "SdpUserFile.h"\r
-#include "SdpGenUtils.h"\r
+#include "SdpBufferPool.h"\r
 #include "SdpSocket.h"\r
 #include "SdpArp.h"\r
 \r
index 1b775bc..79448a6 100644 (file)
@@ -7,6 +7,7 @@ SOURCES= SdpDriver.cpp          \
         SdpGenUtils.cpp        \\r
         SdpSocket.cpp          \\r
         SdpArp.cpp             \\r
+        SdpBufferPool.cpp      \\r
         SdpTrace.cpp\r
 \r
 INCLUDES=..\include;\\r
diff --git a/ulp/sdp/kernel/SdpBufferPool.cpp b/ulp/sdp/kernel/SdpBufferPool.cpp
new file mode 100644 (file)
index 0000000..b71a6be
--- /dev/null
@@ -0,0 +1,321 @@
+/* Copyright mellanox */\r
+#pragma warning(disable: 4244 ) \r
+\r
+#include "preCompile.h"\r
+\r
+NTSTATUS \r
+BufferPool::Init(\r
+    int MaxBuffers, \r
+    int MaxConcurrentSends, \r
+    int MaxMessageSize,\r
+    ib_pd_handle_t pd,\r
+    ib_qp_handle_t qp\r
+    )\r
+{\r
+    SDP_PRINT(SDP_TRACE, SDP_BUFFER_POOL, ("this = 0x%p \n",this));\r
+    m_MaxBuffers = MaxBuffers;\r
+    m_MaxConcurrentSends = MaxConcurrentSends;\r
+    m_MaxMessageSize = MaxMessageSize;        \r
+    m_ClientBeingServed = false;\r
+    m_CurrentlySentBuffers = 0;\r
+    m_CurrentlyAllocated = 0;\r
+    m_ClientWaiting = false;\r
+    KeInitializeEvent(&m_WaitingClients, NotificationEvent, FALSE);\r
+    ASSERT(pd != NULL);\r
+    m_pd = pd;\r
+    ASSERT(qp != NULL);    \r
+    m_qp = qp;\r
+\r
+    return STATUS_SUCCESS;\r
+}\r
+\r
+/*\r
+    This function is being called by a thread that wants to do a send in order\r
+    to have a buffer that he can copy the data to.\r
+    FirstBuffer tells if this is the first buffer that he wants.\r
+    If it is true, this means that no other request will be handled before\r
+    this client will indicate that he has finished queing his data.\r
+    If an event is returned this means that the caller has to wait on the\r
+    event before the request will be staisfied.\r
+\r
+    This function is being called under a lock\r
+\r
+*/\r
+NTSTATUS \r
+BufferPool::GetBuffer(\r
+    BufferDescriptor **ppBufferDescriptor,    \r
+    KEVENT **ppEvent,\r
+    bool FirstBuffer\r
+    )\r
+{\r
+    SDP_PRINT(SDP_TRACE, SDP_BUFFER_POOL, ("this = 0x%p FirstBuffer = %s\n",this,\r
+        FirstBuffer ? "TRUE" : "FALSE"));\r
+    NTSTATUS rc = STATUS_SUCCESS;\r
+    *ppBufferDescriptor = NULL;\r
+\r
+    if (m_ClientBeingServed == true && (FirstBuffer != false)) {\r
+        // The request can not be staisfied right now. We need to hold it\r
+        // until our request is being freed\r
+        // BUGBUG: iMPLMENT: create event and put it in the queue\r
+        ASSERT(FALSE);\r
+    }\r
+\r
+    if (FirstBuffer == true) {\r
+        m_ClientBeingServed = true; \r
+    }\r
+\r
+    // Can we supply a buffer right now ?\r
+    if (m_CurrentlyAllocated < m_MaxBuffers) {\r
+        // yes, supply a buffer\r
+        if (m_FreePackets.Size() > 0) {\r
+            LIST_ENTRY *item = m_FreePackets.RemoveHeadList();\r
+            *ppBufferDescriptor = CONTAINING_RECORD(item, BufferDescriptor , BuffersList);\r
+            goto Cleanup;\r
+        } else {\r
+            // we need to alocate a new buffer\r
+            rc = AllocateBuffer(ppBufferDescriptor);\r
+            if (!NT_SUCCESS(rc)) {\r
+                SDP_PRINT(SDP_ERR, SDP_BUFFER_POOL, ("AllocateBuffer failed rc = 0x%x\n", rc ));\r
+                ASSERT(*ppBufferDescriptor == NULL);\r
+                goto Cleanup;\r
+            }\r
+            m_CurrentlyAllocated++;\r
+            goto Cleanup;\r
+        }        \r
+    } else {\r
+        // No buffers available, we have to wait\r
+        ASSERT(m_ClientWaiting == false);\r
+        KeClearEvent(&m_WaitingClients);\r
+        m_ClientWaiting = true;\r
+        *ppEvent = &m_WaitingClients;\r
+    }\r
+\r
+Cleanup:    \r
+    return rc;\r
+}\r
+\r
+NTSTATUS \r
+BufferPool::AddBufferToQueuedList(BufferDescriptor *pBufferDescriptor)\r
+{\r
+    SDP_PRINT(SDP_TRACE, SDP_BUFFER_POOL, ("this = 0x%p pBufferDescriptor = 0x%x\n",this,\r
+        pBufferDescriptor));\r
+    NTSTATUS rc = STATUS_SUCCESS;\r
+\r
+    if ((m_CurrentlySentBuffers < m_MaxConcurrentSends) && \r
+        (m_QueuedPackets.Size() == 0 )){\r
+        // we can send right away (no need to wait for anything)\r
+        rc = SendBuffer(pBufferDescriptor);\r
+        goto Cleanup;\r
+    } else {\r
+        // we put the buffer in the queued list\r
+        m_QueuedPackets.InsertTailList(&pBufferDescriptor->BuffersList);\r
+    }\r
+\r
+Cleanup:    \r
+    return rc;\r
+\r
+}\r
+\r
+/* \r
+    This function is being called by a client that has asked for some buffers\r
+    when he has recieved all it's data\r
+*/\r
+VOID \r
+BufferPool::AllowOthersToGet()\r
+{\r
+    ASSERT(m_ClientBeingServed == true);\r
+    m_ClientBeingServed = false;\r
+\r
+    // BUGBUG: this means that we should free the next waiter (Once we support more\r
+    // than one thread).\r
+}\r
+\r
+\r
+VOID \r
+BufferPool::ReturnBuffer(BufferDescriptor *pBufferDescriptor)\r
+{\r
+    SDP_PRINT(SDP_TRACE, SDP_BUFFER_POOL, ("this = 0x%p buffer=0x%p\n",this, pBufferDescriptor));\r
+    m_FreePackets.InsertTailList(&pBufferDescriptor->BuffersList);\r
+    // Is there a client waiting ?\r
+    if ( m_ClientWaiting) {\r
+        KeSetEvent( &m_WaitingClients, IO_NO_INCREMENT, FALSE );\r
+        m_ClientWaiting = false;        \r
+    }\r
+    m_CurrentlySentBuffers--;\r
+    ASSERT(m_CurrentlySentBuffers >= 0);\r
+}\r
+\r
+/*\r
+    This function goes over the list of packets that we can send, and sends\r
+    them. It is called under the lock, and might be called also from a DPC\r
+    context.\r
+\r
+*/\r
+NTSTATUS\r
+BufferPool::SendBuffersIfCan()\r
+{\r
+    SDP_PRINT(SDP_TRACE, SDP_BUFFER_POOL, ("this = 0x%p \n",this));\r
+    NTSTATUS rc = STATUS_SUCCESS;\r
+\r
+    while ((m_QueuedPackets.Size() > 0) && \r
+           (m_CurrentlySentBuffers < m_MaxConcurrentSends)) {\r
+        // we can now send the next buffer\r
+        LIST_ENTRY *item = m_QueuedPackets.RemoveHeadList();\r
+        BufferDescriptor *pBufferDescriptor = CONTAINING_RECORD(item, BufferDescriptor , BuffersList);\r
+        rc = SendBuffer(pBufferDescriptor);\r
+        if (!NT_SUCCESS(rc)) {\r
+            SDP_PRINT(SDP_ERR, SDP_BUFFER_POOL, ("SendBuffer failed rc = 0x%x\n", rc ));\r
+            goto Cleanup;\r
+        }        \r
+    }\r
+\r
+Cleanup:    \r
+    return rc;\r
+\r
+}\r
+\r
+/*\r
+    This function is being called from under the lock and is the last one to be called.\r
+    It frees all resources\r
+\r
+*/\r
+VOID \r
+BufferPool::ShutDown()\r
+{\r
+    SDP_PRINT(SDP_TRACE, SDP_BUFFER_POOL, ("this = 0x%p \n",this));\r
+    BufferDescriptor *pBufferDescriptor = NULL;\r
+    LIST_ENTRY *item = NULL;\r
+\r
+    while (m_FreePackets.Size() > 0 ) {\r
+        item = m_FreePackets.RemoveHeadList();\r
+        pBufferDescriptor = CONTAINING_RECORD(item, BufferDescriptor , BuffersList);\r
+        DeAllocateBuffer(pBufferDescriptor);\r
+    }\r
+\r
+    while (m_QueuedPackets.Size() > 0 ) {\r
+        item = m_QueuedPackets.RemoveHeadList();\r
+        pBufferDescriptor = CONTAINING_RECORD(item, BufferDescriptor , BuffersList);\r
+        DeAllocateBuffer(pBufferDescriptor);\r
+    }\r
+\r
+}\r
+\r
+NTSTATUS \r
+BufferPool::AllocateBuffer(BufferDescriptor ** ppBufferDescriptor)\r
+{\r
+    SDP_PRINT(SDP_TRACE, SDP_BUFFER_POOL, ("this = 0x%p \n",this));\r
+    NTSTATUS rc = STATUS_SUCCESS;\r
+    BufferDescriptor *pBufferDescriptor = NULL;\r
+    ib_mr_create_t mr_create;\r
+    uint32_t rkey;\r
+\r
+    // Allocate the buffer descriptor\r
+    pBufferDescriptor = \r
+        (BufferDescriptor *)\r
+            ExAllocatePoolWithTag(\r
+                            NonPagedPool ,\r
+                            sizeof BufferDescriptor, \r
+                            SEND_BUFFERS_ALLOCATION_TAG\r
+                            );\r
+    if (pBufferDescriptor == NULL) {\r
+        SDP_PRINT(SDP_ERR, SDP_BUFFER_POOL, ("ExAllocatePoolWithTag failed \n"));\r
+        rc = STATUS_NO_MEMORY;\r
+        goto Cleanup;\r
+    }\r
+\r
+    // Allocate the buffer itself\r
+    pBufferDescriptor->pBuffer = \r
+        ExAllocatePoolWithTag(\r
+                        NonPagedPool ,\r
+                        m_MaxMessageSize, \r
+                        SEND_BUFFERS_ALLOCATION_TAG\r
+                        );\r
+\r
+    if (pBufferDescriptor->pBuffer == NULL) {\r
+        SDP_PRINT(SDP_ERR, SDP_BUFFER_POOL, ("ExAllocatePoolWithTag failed \n"));\r
+        rc = STATUS_NO_MEMORY;\r
+        goto Cleanup;\r
+    }\r
+\r
+    pBufferDescriptor->BufferSize = m_MaxMessageSize;\r
+    pBufferDescriptor->DataSize = 0;\r
+    pBufferDescriptor->mr_handle = NULL;\r
+    \r
+    // Now we need to register this memory with the hardware\r
+    mr_create.vaddr = pBufferDescriptor->pBuffer;\r
+    mr_create.length = pBufferDescriptor->BufferSize;\r
+    mr_create.access_ctrl = IB_AC_LOCAL_WRITE;\r
+\r
+    ib_api_status_t ib_status = ib_reg_mem( m_pd, &mr_create, &pBufferDescriptor->ds_array.lkey, &rkey, &pBufferDescriptor->mr_handle );\r
+    if( ib_status != IB_SUCCESS ) {\r
+        SDP_PRINT(SDP_ERR, SDP_BUFFER_POOL, ("ib_reg_mem failed ib_status = 0x%d\n", ib_status ));\r
+        rc = IB2Status(ib_status);\r
+        goto Cleanup;\r
+    }\r
+\r
+Cleanup:\r
+    if (!NT_SUCCESS(rc)) {\r
+        if (pBufferDescriptor != NULL) {\r
+            if (pBufferDescriptor->pBuffer != NULL) {\r
+                ExFreePoolWithTag(pBufferDescriptor->pBuffer, SEND_BUFFERS_ALLOCATION_TAG);\r
+            }\r
+            ExFreePoolWithTag(pBufferDescriptor, SEND_BUFFERS_ALLOCATION_TAG);\r
+        }        \r
+    }\r
+    *ppBufferDescriptor = pBufferDescriptor;\r
+    return rc;\r
+}\r
+\r
+VOID \r
+BufferPool::DeAllocateBuffer(BufferDescriptor *pBufferDescriptor)\r
+{\r
+    //????? clear the memory here. \r
+    // ?????\r
+\r
+}\r
+\r
+NTSTATUS\r
+BufferPool::SendBuffer(BufferDescriptor *pBufferDescriptor)\r
+{\r
+    SDP_PRINT(SDP_TRACE, SDP_BUFFER_POOL, ("this = 0x%p \n",this));\r
+    NTSTATUS rc = STATUS_SUCCESS;\r
+\r
+    msg_hdr_bsdh *pHeader = (msg_hdr_bsdh *) pBufferDescriptor->pBuffer;\r
+\r
+    pHeader->recv_bufs = QP_ATTRIB_RQ_DEPTH; //?????recv_bufs = conn->l_advt_bf;\r
+    pHeader->size = pBufferDescriptor->DataSize + sizeof msg_hdr_bsdh;\r
+    pHeader->seq_num = 1;//?????++conn->send_seq;\r
+    pHeader->seq_ack = 0;//????conn->advt_seq;\r
+    pHeader->mid = SDP_MID_DATA;\r
+    pHeader->flags = SDP_MSG_FLAG_NON_FLAG;\r
+    /*\r
+     * endian swap\r
+     */\r
+    sdp_msg_swap_bsdh(pHeader);\r
+\r
+    ib_send_wr_t    send_wr;\r
+\r
+    send_wr.p_next = NULL;\r
+    send_wr.wr_id = (uintn_t)pBufferDescriptor;//??? buff->wrid;//?????(uint64_t) (uintptr_t) wr;\r
+    send_wr.wr_type = WR_SEND;\r
+    send_wr.send_opt = IB_SEND_OPT_SIGNALED;//socket_info->send_opt;\r
+\r
+    pBufferDescriptor->ds_array.length = pBufferDescriptor->DataSize + sizeof msg_hdr_bsdh;\r
+    pBufferDescriptor->ds_array.vaddr = (uint64_t)(void* __ptr64) pBufferDescriptor->pBuffer;\r
+\r
+    send_wr.num_ds = 1;\r
+    send_wr.ds_array = &pBufferDescriptor->ds_array;\r
+    \r
+    ib_api_status_t ib_status = ib_post_send(m_qp, &send_wr, NULL);\r
+    if( ib_status != IB_SUCCESS ) {\r
+        SDP_PRINT(SDP_ERR, SDP_BUFFER_POOL, ("ib_post_send failed ib_status = 0x%d\n", ib_status ));\r
+        rc = IB2Status(ib_status);\r
+        goto Cleanup;\r
+    }\r
+    m_CurrentlySentBuffers ++;\r
+\r
+Cleanup:\r
+    return rc;\r
+}\r
+\r
+\r
diff --git a/ulp/sdp/kernel/SdpBufferPool.h b/ulp/sdp/kernel/SdpBufferPool.h
new file mode 100644 (file)
index 0000000..a88fd35
--- /dev/null
@@ -0,0 +1,135 @@
+/* Copyright mellanox */\r
+\r
+#ifndef H_SDP_BUFFER_POOL_H\r
+#define H_SDP_BUFFER_POOL_H \r
+\r
+\r
+// This is simply a wrapper to the LIST_ENTRY class that allows \r
+// easier work with this list\r
+class LinkedList {\r
+\r
+public:\r
+    LinkedList() {\r
+       size = 0;\r
+       InitializeListHead(&m_Data);\r
+    }\r
+\r
+    int Size() {return size;}\r
+\r
+    LIST_ENTRY *RemoveHeadList() {\r
+        LIST_ENTRY *pTemp;\r
+        ASSERT(size > 0);\r
+        ASSERT(!IsListEmpty(&m_Data));\r
+        pTemp = ::RemoveHeadList(&m_Data);\r
+        size--;\r
+        return pTemp;        \r
+    }\r
+    \r
+    VOID InsertTailList (LIST_ENTRY *Item) {\r
+        ::InsertTailList(&m_Data, Item);\r
+        size++;\r
+    }\r
+\r
+private:\r
+    int size;\r
+    LIST_ENTRY m_Data;\r
+};\r
+\r
+\r
+// The defenition of the function that we use to report back errors\r
+typedef void (* SendErrorCB )(NTSTATUS Error, VOID *Context);\r
+\r
+\r
+// Each buffer starts with msg_hdr_bsdh and is followed by the actual data\r
+class BufferDescriptor {\r
+public:\r
+    NTSTATUS WriteData(char *pData, uint32_t size) {\r
+        NTSTATUS rc = STATUS_SUCCESS;\r
+        ASSERT(size <= BufferSize - sizeof msg_hdr_bsdh);\r
+        char *pStart = (char *) pBuffer + sizeof msg_hdr_bsdh;\r
+        rc = CopyFromUser(pStart, pData, size);\r
+        if (!NT_SUCCESS(rc)) {\r
+            SDP_PRINT(SDP_ERR, SDP_BUFFER_POOL, ("CopyFromUser failed rc = 0x%x\n", rc ));\r
+            goto Cleanup;\r
+        }                \r
+        DataSize = size;\r
+        Cleanup:\r
+        return rc;\r
+    }\r
+\r
+    // Each buffer starts with bsdh_hdr structure\r
+    VOID    *pBuffer;            // A pointer to the actual place that we put the data\r
+    int     BufferSize;          // The total size of the buffer\r
+    int     DataSize;            // The size of the data that we have allocated\r
+    LIST_ENTRY BuffersList;      // The place to hold the list of the buffers\r
+    ib_mr_handle_t mr_handle;    // A handle to the registared memory,\r
+\r
+    ib_local_ds_t ds_array;      // Used for sending the buffer\r
+\r
+};\r
+\r
+class BufferPool {\r
+\r
+public:\r
+\r
+    NTSTATUS Init(\r
+        int MaxBuffers, \r
+        int MaxConcurrentSends, \r
+        int MaxMessageSize,\r
+        ib_pd_handle_t pd,\r
+        ib_qp_handle_t qp\r
+        );\r
+\r
+    NTSTATUS GetBuffer(\r
+        BufferDescriptor ** ppBufferDescriptor, \r
+        KEVENT **ppEvent,\r
+        bool FirstBuffer\r
+        );\r
+\r
+    NTSTATUS AddBufferToQueuedList(BufferDescriptor *pBufferDescriptor);    \r
+\r
+    VOID AllowOthersToGet();    \r
+\r
+    VOID ReturnBuffer(BufferDescriptor *pBufferDescriptor);\r
+\r
+    NTSTATUS SendBuffersIfCan();\r
+\r
+    VOID ShutDown();\r
+    \r
+private:\r
+\r
+    NTSTATUS AllocateBuffer(BufferDescriptor ** ppBufferDescriptor);\r
+\r
+    VOID     DeAllocateBuffer(BufferDescriptor *pBufferDescriptor);\r
+\r
+    NTSTATUS SendBuffer(BufferDescriptor *pBufferDescriptor);\r
+\r
+    // Global data about this connection\r
+    int m_MaxBuffers;           // The maximum number of buffers that we allow for this QP\r
+    int m_MaxConcurrentSends;   // The total numbers of sends that are allowd for the QP\r
+    int m_MaxMessageSize;       // The maximum buffer size that we allw\r
+\r
+    int m_CurrentlySentBuffers; // Number of buffers that we have sent, and didn't get an ack yet\r
+    int m_CurrentlyAllocated;   // The number of buffers that we have allocated\r
+\r
+    bool m_ClientBeingServed;   // true if we have already started giving buffers to a client\r
+\r
+    LinkedList m_FreePackets; // This packets are free and might be used\r
+    LinkedList m_QueuedPackets; // This packets were filled with data and should be filled\r
+    \r
+\r
+    // TODO: A queue of events for threads that are waiting for buffers.\r
+\r
+    // IBAL constants from the main socket structure \r
+    // TODO: Should they stay here and be used like this ?\r
+    ib_pd_handle_t          m_pd;\r
+    ib_qp_handle_t          m_qp;\r
+\r
+    // A list of events that the users has to wait on. ???? currently only one\r
+    KEVENT m_WaitingClients; // switch to a linked list\r
+    bool   m_ClientWaiting;\r
+\r
+};\r
+\r
+#endif // H_SDP_BUFFER_POOL_H\r
+\r
index b99b18b..bfccea6 100644 (file)
@@ -67,6 +67,29 @@ NTSTATUS
     return rc;\r
 }\r
 \r
+NTSTATUS\r
+CopyFromUser(\r
+    IN  void* const         p_dest,\r
+    IN  const void* const   p_src,\r
+    IN  const size_t        count )\r
+{\r
+    /*\r
+     * The memory copy must be done within a try/except block as the\r
+     * memory could be changing while the buffer is copied.\r
+     */\r
+    __try\r
+    {\r
+        ProbeForRead( (void*)p_src, count, 1 );\r
+        RtlCopyMemory( p_dest, p_src, count );\r
+        return STATUS_SUCCESS;\r
+    }\r
+    __except(EXCEPTION_EXECUTE_HANDLER)\r
+    {\r
+        SDP_PRINT(SDP_ERR, SDP_SOCKET, ("copying memory from user failed\n"));\r
+        ASSERT(FALSE);        \r
+        return STATUS_ACCESS_DENIED;\r
+    }\r
+}\r
 \r
 \r
 void* __cdecl operator new(size_t n ) throw() {\r
index 30deb83..622eae0 100644 (file)
@@ -5,6 +5,7 @@
 \r
 \r
 #define GLOBAL_ALLOCATION_TAG ' pdS'\r
+#define SEND_BUFFERS_ALLOCATION_TAG 'SpdS'\r
 \r
 \r
 class CSpinLockWrapper {\r
@@ -43,6 +44,11 @@ NTSTATUS IB2Status (ib_api_status_t ib_status);
 \r
 USHORT nthos(USHORT in);\r
 \r
+NTSTATUS\r
+CopyFromUser(\r
+    IN  void* const         p_dest,\r
+    IN  const void* const   p_src,\r
+    IN  const size_t        count );\r
 \r
 NTSTATUS \r
   MyKeWaitForSingleObject(\r
diff --git a/ulp/sdp/kernel/SdpLock.h b/ulp/sdp/kernel/SdpLock.h
new file mode 100644 (file)
index 0000000..9ec2ea8
--- /dev/null
@@ -0,0 +1,245 @@
+/* Copyright mellanox */\r
+#ifndef _SDP_LOCK_H\r
+#define _SDP_LOCK_H\r
+\r
+/*\r
+The goal of this lock is to be a user mode lock that will allow us to synchronize\r
+both "user" operations at PASSIVE level as well as DPC's at DPC level.\r
+\r
+The main problem that we have is that we have many functions that we can only call at \r
+passive level, and therefore can not be called under a spinlock.\r
+\r
+We might, however, receive notifications at DPC level. Example of such are send and \r
+receive completions. As always, shutdown might appear at any time (at any level?).\r
+\r
+Bottom line of this is that the lock will be implemented as an event. DPC level \r
+callers that will call us will only mark our state as send/received/shutdown arrived.\r
+\r
+Once one tries to take/free the lock from passive level, he will have to handle this \r
+events first.\r
+\r
+Callers at DPC level, (send receive call backs) will only signal if the lock is taken\r
+or do the actual job if it is not taken.\r
+\r
+There will therefore be a spinlock that will protect the event. \r
+\r
+*/\r
+\r
+\r
+// Still Need to make sure that all errors are handled when they should ??????\r
+\r
+typedef NTSTATUS (* SendCBHandler )(SdpSocket *);\r
+\r
+const int SEND_CB_CALLED        = 0x00000001;\r
+const int RECV_CB_CALLED        = 0x00000002;\r
+const int SHUTDOWN_SIGNALLED    = 0x00000004;\r
+const int SHUTDOWN_HANDELED     = 0x00000008;\r
+const int ERROR_SIGNALLED       = 0x00000010;\r
+\r
+const int DPC_FLAGS = SEND_CB_CALLED | SEND_CB_CALLED;\r
+inline void ResetFlags(int &Flags)\r
+{\r
+    Flags &= (!(SEND_CB_CALLED | RECV_CB_CALLED));\r
+}\r
+\r
+inline void ResetDpcFlags(int &Flags)\r
+{\r
+    // Currently this function is just like the one above it. It will probably\r
+    // change in the future\r
+    Flags &= (!(DPC_FLAGS));\r
+}\r
+\r
+inline bool SomethingToHandle(int flags)\r
+{\r
+    if (flags & SEND_CB_CALLED) return true;\r
+    if (flags & RECV_CB_CALLED) return true;\r
+    if ((flags & SHUTDOWN_SIGNALLED) && !(flags & SHUTDOWN_HANDELED) ) return true;\r
+\r
+    return false;\r
+}\r
+\r
+class SdpLock {\r
+public:\r
+    SdpLock() {\r
+        m_InUse = false;\r
+        m_flags = 0;\r
+        KeInitializeEvent(&m_Event, NotificationEvent , TRUE);\r
+        KeInitializeSpinLock(&m_SpinLock);\r
+        m_SendCBHandler = NULL;\r
+    }\r
+\r
+    VOID Init(SendCBHandler SendCB, SdpSocket *pSdpSocket)\r
+    {\r
+        m_SendCBHandler = SendCB;\r
+        m_pSdpSocket = pSdpSocket;\r
+    }\r
+\r
+    /*\r
+        Lock should handle recieve_cb/send_cb without user knowledge.\r
+        for shutdown, it should return false and not continue\r
+\r
+        return value of false means that the lock can not be taken (eitheir\r
+        shutdown or STATUS_ALERTED, or some error has happend)\r
+    */\r
+    bool Lock() {\r
+        KIRQL  OldIrql;\r
+        int OldFlags = 0;\r
+        NTSTATUS rc = STATUS_SUCCESS;\r
+        ASSERT(KeGetCurrentIrql() == PASSIVE_LEVEL);\r
+        bool Locked = false;\r
+        do {\r
+            KeAcquireSpinLock(&m_SpinLock, &OldIrql);\r
+\r
+            if (m_InUse) {\r
+                // We have to release the spinlock and wait on the event\r
+                KeReleaseSpinLock(&m_SpinLock, OldIrql);\r
+                rc = MyKeWaitForSingleObject(&m_Event, UserRequest, UserMode, false, NULL);\r
+                if (( rc == STATUS_ALERTED ) ||( rc == STATUS_USER_APC )) {\r
+                    SDP_PRINT(SDP_WARN, SDP_LOCK, ("MyKeWaitForSingleObject was alerted = 0x%x\n", rc ));\r
+                    rc = STATUS_UNEXPECTED_IO_ERROR;\r
+                    SignalShutdown();\r
+                    Locked = false;\r
+                    goto Cleanup;\r
+                }                \r
+                continue;\r
+            }\r
+            m_InUse = true;\r
+            KeClearEvent(&m_Event);\r
+            OldFlags = m_flags;\r
+            ResetFlags(m_flags);\r
+            KeReleaseSpinLock(&m_SpinLock, OldIrql);\r
+            rc = HandleFlags(OldFlags);\r
+            if (!NT_SUCCESS(rc)) {\r
+                // We have to signal the error to the calling side\r
+                SDP_PRINT(SDP_ERR, SDP_LOCK, ("HandleFlags failed rc = 0x%x\n", rc ));\r
+                Locked = false;\r
+                ASSERT(m_flags & ERROR_SIGNALLED);\r
+                KeAcquireSpinLock(&m_SpinLock, &OldIrql);\r
+                m_InUse = false;\r
+                // Release whoever is waiting\r
+                KeSetEvent(&m_Event, IO_NO_INCREMENT, FALSE);\r
+                KeReleaseSpinLock(&m_SpinLock, OldIrql);                \r
+                goto Cleanup;\r
+            }\r
+            // Exit the loop\r
+            Locked = true;\r
+            goto Cleanup;            \r
+        } while (true);\r
+        \r
+Cleanup:\r
+        SDP_PRINT(SDP_DEBUG, SDP_LOCK,("Lock is returing %s\n", Locked ? "true" : "false"));\r
+        return Locked;\r
+    }\r
+\r
+    /*\r
+        Frees the lock and handle any events that might happen there.\r
+        Please note that the lock is freed no metter what the error code is.\r
+        An error means that there was some error in the sockets.\r
+    */\r
+    NTSTATUS Unlock()\r
+    {\r
+        KIRQL  OldIrql;\r
+        int OldFlags = 0;\r
+        NTSTATUS rc = STATUS_SUCCESS;\r
+\r
+        while (true) {\r
+            ASSERT(m_InUse);\r
+            ASSERT(KeGetCurrentIrql() == PASSIVE_LEVEL);\r
+            KeAcquireSpinLock(&m_SpinLock, &OldIrql);\r
+            OldFlags = m_flags;\r
+            ResetFlags(m_flags);\r
+            if (!SomethingToHandle(OldFlags)) {\r
+                // We can safely quit the lock\r
+                m_InUse = false;\r
+            }\r
+            KeReleaseSpinLock(&m_SpinLock, OldIrql);\r
+            if (SomethingToHandle(OldFlags)) {\r
+                rc = HandleFlags(OldFlags); \r
+                if (!NT_SUCCESS(rc)) {\r
+                    // We have to signal the error to the calling side\r
+                    SDP_PRINT(SDP_ERR, SDP_LOCK, ("HandleFlags failed rc = 0x%x\n", rc ));\r
+                    ASSERT(m_flags & ERROR_SIGNALLED);\r
+                }\r
+                // At the time that we were handeling the flags, someone might have \r
+                // signaled something, so we have to try again\r
+                continue;\r
+            }\r
+            break;\r
+        }\r
+        \r
+        // Release whoever is waiting\r
+        KeSetEvent(&m_Event, IO_NO_INCREMENT, FALSE);\r
+        return rc;\r
+    }\r
+/*\r
+    This function is being called at DPC level. It has some message of a call back.\r
+    to tell us. Once called, it will try to take the lock. If it succeeds, it will \r
+    do the actual work, if not it will only signal. Once it returns the lock is freed \r
+    again\r
+*/\r
+    bool SignalCB(int flags)\r
+    {\r
+        KIRQL  OldIrql;\r
+        int OldFlags = 0;\r
+        NTSTATUS rc = STATUS_SUCCESS;\r
+        ASSERT(KeGetCurrentIrql() == DISPATCH_LEVEL);\r
+        KeAcquireSpinLock(&m_SpinLock, &OldIrql);\r
+        if (m_InUse) {\r
+            m_flags |= flags;\r
+            KeReleaseSpinLock(&m_SpinLock, OldIrql);\r
+            return false;\r
+        }\r
+        m_InUse = true;\r
+        // In this lock, we only handle DPC events\r
+        OldFlags = (m_flags & DPC_FLAGS) | flags;\r
+        ResetDpcFlags(m_flags);\r
+        KeClearEvent(&m_Event);\r
+        KeReleaseSpinLock(&m_SpinLock, OldIrql);\r
+        rc = HandleFlags(OldFlags); \r
+        if (!NT_SUCCESS(rc)) {\r
+            // We have to signal the error to the calling side\r
+            SDP_PRINT(SDP_ERR, SDP_LOCK, ("HandleFlags failed rc = 0x%x\n", rc ));\r
+            ASSERT(m_flags & ERROR_SIGNALLED);\r
+        }\r
+        KeAcquireSpinLock(&m_SpinLock, &OldIrql);\r
+        // Release whoever is waiting\r
+        m_InUse = false;\r
+        KeSetEvent(&m_Event, IO_NO_INCREMENT, FALSE);\r
+        KeReleaseSpinLock(&m_SpinLock, OldIrql);\r
+        return true;\r
+    }\r
+\r
+    /*\r
+        This function is responsible for handling the flags that we might get.\r
+        Currently it can be called from passive or DPC level, and handle only "DPC" events\r
+    */\r
+    NTSTATUS HandleFlags(int flags) {\r
+        NTSTATUS rc = STATUS_SUCCESS;\r
+        if (flags & SEND_CB_CALLED) {\r
+            // We need to handle the send CB\r
+            rc = m_SendCBHandler(m_pSdpSocket);\r
+            if (!NT_SUCCESS(rc)) {\r
+                SDP_PRINT(SDP_ERR, SDP_BUFFER_POOL, ("SendBuffer failed rc = 0x%x\n", rc ));\r
+                m_flags |= ERROR_SIGNALLED;\r
+                // We continue from here since, there might be other things to handle,\r
+                // and this might be in a DPC context\r
+            }        \r
+        }\r
+        return rc;\r
+    }\r
+\r
+    VOID SignalShutdown() {ASSERT (FALSE);} //????????????? Make sure this is used\r
+    VOID SignalError(NTSTATUS rc) {ASSERT (FALSE);} //?????????????       \r
+\r
+    KEVENT  m_Event;         // the event for passive level threads\r
+    KSPIN_LOCK  m_SpinLock;  // The real guard of the lock\r
+    SendCBHandler m_SendCBHandler;\r
+\r
+\r
+    bool m_InUse;  // Tells if this lock has any user\r
+    int  m_flags;      // call backs that were recieved\r
+\r
+    SdpSocket *m_pSdpSocket; // The socket that this class depends on\r
+};\r
+\r
+#endif // _SDP_LOCK_H\r
index 33cd8b2..a185360 100644 (file)
@@ -5,11 +5,12 @@
 #pragma warning(disable: 4244 ) \r
 \r
 NTSTATUS sdp_cm_hello_ack_check(struct sdp_msg_hello_ack *hello_ack);\r
+static NTSTATUS __send_cb2(SdpSocket * pSdpSocket);\r
 \r
 static void AL_API\r
 cm_rej_callback(IN ib_cm_rej_rec_t *p_cm_rej_rec )\r
 {\r
-    SDP_PRINT(SDP_TRACE, SDP_SOCKET, ("cm_rej_callback called"));\r
+    SDP_PRINT(SDP_TRACE, SDP_SOCKET, ("dispatch level = %d\n", KeGetCurrentIrql()));\r
     // BUGBUG: This should be used to return error to the connecting side\r
 }\r
 \r
@@ -55,7 +56,7 @@ cm_apr_callback(
 static void AL_API\r
 cm_dreq_callback(IN ib_cm_dreq_rec_t    *p_cm_dreq_rec )\r
 {\r
-    SDP_PRINT(SDP_TRACE, SDP_SOCKET, ("cm_dreq_callback called"));\r
+    SDP_PRINT(SDP_TRACE, SDP_SOCKET, ("dispatch level = %d\n", KeGetCurrentIrql()));\r
     ASSERT(FALSE);\r
 }\r
 SdpSocket::SdpSocket()\r
@@ -70,8 +71,6 @@ SdpSocket::SdpSocket()
     m_scq = NULL;\r
     m_qp = NULL;\r
 \r
-    m_shutdown = false;\r
-\r
     m_state = SS_IDLE;\r
 }\r
 \r
@@ -83,13 +82,158 @@ NTSTATUS SdpSocket::Init(
     SDP_PRINT(SDP_TRACE, SDP_SOCKET, ("SdpSocket::Init this = 0x%p\n", this));\r
 \r
     m_CreationFlags = pSocketInParam->dwFlags;\r
+\r
+    m_Lock.Init(__send_cb2, this);\r
     pSocketOutParam->Errno = 0;// No error\r
     pSocketOutParam->pSocket = this; // give the user a handle to the socket\r
-    KeInitializeSpinLock(&m_Lock);\r
 \r
     return rc;\r
 }\r
 \r
+struct sdpc_buff {\r
+//    struct sdpc_buff   *next;\r
+//    struct sdpc_buff   *prev;\r
+//    u32                 type; /* element type. (for generic queue) */\r
+//    struct sdpc_buff_q *pool; /* pool currently holding this buffer. */\r
+//    int (*release)(struct sdpc_buff *buff); /* release the object */\r
+    /*\r
+     * primary generic data pointers\r
+     */\r
+    void *head; /* first byte of data buffer */\r
+    void *data; /* first byte of valid data in buffer */\r
+    void *tail; /* last byte of valid data in buffer */\r
+    void *end;  /* last byte of data buffer */\r
+    /*\r
+     * Experimental\r
+     */\r
+    uint32_t flags;  /* Buffer flags */\r
+    /*\r
+     * Protocol specific data\r
+     */\r
+    struct msg_hdr_bsdh *bsdh_hdr; /* SDP header (BSDH) */\r
+    uint32_t data_size;                 /* size of just data in the buffer */\r
+    uint64_t wrid;                   /* IB work request ID */\r
+    /*\r
+     * IB specific data (The main buffer pool sets the lkey when \r
+     * it is created)\r
+     */\r
+    uint64_t real; /* component of scather/gather list (address) */\r
+    uint32_t size; /* component of scather/gather list (lenght)  */\r
+    uint32_t lkey; /* component of scather/gather list (key) */\r
+};\r
+\r
+const int BUFFER_SIZE = 4000 + 16;//65536;\r
+\r
+#define SDP_BUFF_F_UNSIG    0x0001     /* unsignalled buffer */\r
+\r
+#define SDP_BUFF_F_GET_UNSIG(buff) ((buff)->flags &    SDP_BUFF_F_UNSIG)\r
+#define SDP_BUFF_F_SET_UNSIG(buff) ((buff)->flags |=   SDP_BUFF_F_UNSIG)\r
+#define SDP_BUFF_F_CLR_UNSIG(buff) ((buff)->flags &= (~SDP_BUFF_F_UNSIG))\r
+\r
+\r
+NTSTATUS SdpSocket::WSPSend(\r
+        WspSendIn    *pWspSendIn,\r
+        WspSendOut   *pWspSendOut\r
+        )\r
+{    \r
+    SDP_PRINT(SDP_TRACE, SDP_SOCKET, ("this = 0x%p \n",this));\r
+    char temp[4000];\r
+    memcpy(temp,"abcd",5);\r
+\r
+    NTSTATUS rc = STATUS_SUCCESS;\r
+    BufferDescriptor * pBufferDescriptor = NULL;\r
+    bool First = true;\r
+    ULONG Coppied = 0;\r
+    bool Locked = false;\r
+    PRKEVENT  pBuffersEvent = NULL;\r
+\r
+    while (Coppied < pWspSendIn->BufferSize) {\r
+        if ((Locked == false) && !m_Lock.Lock()) {\r
+            SDP_PRINT(SDP_ERR, SDP_SOCKET, ("Failed to lock this = 0x%p \n",this));\r
+            rc = STATUS_SHUTDOWN_IN_PROGRESS;\r
+            goto Cleanup;\r
+        }\r
+        Locked = true;\r
+\r
+        rc = m_SendBufferPool.GetBuffer(&pBufferDescriptor, &pBuffersEvent, First);\r
+        if (!NT_SUCCESS(rc)) {\r
+            SDP_PRINT(SDP_ERR, SDP_SOCKET, ("m_SendBufferPool.GetBuffer failed rc = 0x%x\n", rc ));\r
+            m_Lock.Unlock(); // Error ignored as this is already an error pass\r
+            Locked = false;\r
+            goto Cleanup;\r
+        }\r
+        First = false;\r
+        \r
+        if (pBuffersEvent != NULL) {\r
+            // We are told to wait on this event\r
+            rc = m_Lock.Unlock();\r
+            Locked = false;\r
+            if (!NT_SUCCESS(rc)) {\r
+                SDP_PRINT(SDP_ERR, SDP_SOCKET, ("m_Lock.Unlock() failed rc = 0x%x\n", rc ));\r
+                goto Cleanup;\r
+            }\r
+\r
+            rc = MyKeWaitForSingleObject(\r
+                pBuffersEvent,\r
+                UserRequest,\r
+                UserMode,\r
+                FALSE,\r
+                NULL\r
+                );    \r
+\r
+            if (( rc == STATUS_ALERTED ) ||( rc == STATUS_USER_APC )) {\r
+                // BUGBUG: Think what to do here, we should be able to stop the\r
+                // connect, and quit (probably shutdown should be enough)\r
+                SDP_PRINT(SDP_WARN, SDP_SOCKET, ("MyKeWaitForSingleObject was alerted = 0x%x\n", rc ));\r
+                rc = STATUS_UNEXPECTED_IO_ERROR;\r
+                //pWspConnectOut->Errno = WSAENETUNREACH; // BUGBUG: verify this error\r
+                Shutdown();\r
+                goto Cleanup;\r
+            }\r
+            // try getting the buffer again\r
+            continue;\r
+        }\r
+\r
+        // copy the data from the user mode to the buffers \r
+        ULONG CopySize =  pBufferDescriptor->BufferSize - sizeof msg_hdr_bsdh;\r
+        CopySize = min(CopySize, pWspSendIn->BufferSize - Coppied);\r
+        \r
+        pBufferDescriptor->WriteData(pWspSendIn->pData + Coppied, CopySize);\r
+        Coppied += CopySize;\r
+        \r
+        // return the data to the buffer\r
+        rc = m_SendBufferPool.AddBufferToQueuedList(pBufferDescriptor);\r
+        if (!NT_SUCCESS(rc)) {\r
+            SDP_PRINT(SDP_ERR, SDP_SOCKET, ("m_SendBufferPool.AddBufferToQueuedList failed rc = 0x%x\n", rc ));\r
+            // free the buffer that you have\r
+            m_SendBufferPool.ReturnBuffer(pBufferDescriptor);\r
+            m_Lock.Unlock(); // Error ignored as this is already an error pass        \r
+            goto Cleanup;\r
+        }\r
+    }\r
+    ASSERT(Locked == true);\r
+    rc = m_Lock.Unlock();\r
+    if (!NT_SUCCESS(rc)) {\r
+        SDP_PRINT(SDP_ERR, SDP_SOCKET, ("m_Lock.Unlock() failed rc = 0x%x\n", rc ));\r
+        goto Cleanup;\r
+    }\r
+    \r
+    m_SendBufferPool.AllowOthersToGet();\r
+   \r
+Cleanup:\r
+    if (NT_SUCCESS(rc) ) {\r
+        pWspSendOut->Errno = 0;\r
+        pWspSendOut->NumberOfBytesSent = pWspSendIn->BufferSize;\r
+    } else {\r
+        // Make sure that we have the error setted\r
+        ASSERT(pWspSendOut->Errno != 0); // BUGBUG: Need to make sure that this\r
+        // is indeed the case.\r
+    }\r
+    return rc;\r
+}\r
+\r
+#if 0\r
+//Naive send implmentation.\r
 NTSTATUS SdpSocket::WSPSend(\r
         WspSendIn    *pWspSendIn,\r
         WspSendOut   *pWspSendOut\r
@@ -98,13 +242,147 @@ NTSTATUS SdpSocket::WSPSend(
     NTSTATUS rc = STATUS_SUCCESS;\r
     SDP_PRINT(SDP_TRACE, SDP_SOCKET, ("this = 0x%p \n",this));\r
 \r
+    ib_mr_create_t mr_create;\r
+\r
+    /* Memory registration parameters, returned by ib_reg_mem. */\r
+    char *BufferStart = NULL;\r
+    uint32_t lkey;\r
+    uint32_t rkey;\r
+    ib_mr_handle_t mr_handle = NULL;\r
+\r
+    \r
+\r
+    // First allocate a buffer and a buffer descriptor\r
+    sdpc_buff *buff = new sdpc_buff;\r
+    ASSERT(buff != NULL);\r
+    BufferStart = new CHAR [BUFFER_SIZE];    \r
+    ASSERT(BufferStart != NULL);\r
+    buff->head = BufferStart;\r
+\r
+    // we leave enough space for holding the header of the request\r
+    buff->end  = (CHAR *)(buff->head) + BUFFER_SIZE;\r
+    buff->head = (char *)(buff->head) + 0x10;\r
+    \r
+    buff->data    = buff->head;\r
+    buff->tail    = buff->head;\r
+    buff->lkey    = 0;\r
+    buff->real    = 0;\r
+    buff->size    = 0;\r
+\r
+    // Copy the data to the buffer\r
+    memcpy(buff->data, "5678",5);\r
+    buff->tail = (char *)(buff->tail) + 5;\r
+    \r
+\r
+    // Register the buffer\r
+    mr_create.vaddr = BufferStart;\r
+    mr_create.length = BUFFER_SIZE;\r
+    mr_create.access_ctrl = IB_AC_LOCAL_WRITE;\r
+\r
+    ib_api_status_t ib_status = ib_reg_mem( m_pd, &mr_create, &lkey, &rkey, &mr_handle );\r
+    ASSERT(ib_status == IB_SUCCESS);\r
+    \r
+\r
+\r
+    \r
+\r
+    // Send the buffer.\r
+    buff->data = (char *)(buff->head) - sizeof(struct msg_hdr_bsdh);\r
+    buff->bsdh_hdr = (struct msg_hdr_bsdh *) buff->data;\r
+    buff->bsdh_hdr->mid = SDP_MID_DATA;\r
+    buff->bsdh_hdr->flags = SDP_MSG_FLAG_NON_FLAG;\r
+    buff->bsdh_hdr->size = BUFFER_SIZE;\r
+\r
+\r
+    /*\r
+     * signalled? With no delay turned off, data transmission may be\r
+     * waiting for a send completion.\r
+     */\r
+    SDP_BUFF_F_SET_UNSIG(buff);\r
+\r
+    buff->wrid = 0;//conn->send_wrid++;\r
+\r
+    buff->lkey = lkey;\r
+    buff->bsdh_hdr->recv_bufs = QP_ATTRIB_RQ_DEPTH; //?????recv_bufs = conn->l_advt_bf;\r
+//?????? put this in    buff->bsdh_hdr->size = (char *)buff->tail - (char *)buff->data;\r
+    buff->bsdh_hdr->seq_num = 1;//?????++conn->send_seq;\r
+    buff->bsdh_hdr->seq_ack = 0;//????conn->advt_seq;\r
+\r
+    /*\r
+     * endian swap\r
+     */\r
+    sdp_msg_swap_bsdh(buff->bsdh_hdr);\r
+    buff->real = (uint64_t)(void* __ptr64)BufferStart;\r
+    buff->size = BUFFER_SIZE;\r
+    \r
+    /*\r
+     * save the buffer for the event handler.\r
+     */\r
+#if 0\r
+    result = sdp_buff_q_put_tail(&conn->send_post, buff);\r
+    if (result < 0) {\r
+        sdp_dbg_warn(conn, "Error <%d> queueing send buffer", result);\r
+        goto done;\r
+    }\r
+#endif    \r
+    /*\r
+     * post send\r
+     */\r
+/*     \r
+    buff->size = buff->tail - buff->data;\r
+       buff->real = dma_map_single(conn->ca->dma_device,\r
+                           buff->data,\r
+                                   buff->size,\r
+                                   PCI_DMA_TODEVICE);\r
+    send_param.next    = NULL;\r
+    send_param.wr_id   = buff->wrid;\r
+       send_param.sg_list = (struct ib_sge *)&buff->real;\r
+    send_param.num_sge = 1;\r
+    send_param.opcode  = IB_WR_SEND;\r
+*/\r
+    ib_send_wr_t    send_wr;\r
+\r
+    send_wr.p_next = NULL;\r
+    send_wr.wr_id = buff->wrid;//?????(uint64_t) (uintptr_t) wr;\r
+    send_wr.wr_type = WR_SEND;\r
+    send_wr.send_opt = IB_SEND_OPT_SIGNALED;//IB_SEND_OPT_INLINE;//socket_info->send_opt;\r
+\r
+\r
+    ib_local_ds_t ds_array;\r
+    ds_array.length = buff->size;\r
+    ds_array.lkey = buff->lkey;\r
+    ds_array.vaddr = buff->real;\r
+\r
+    send_wr.num_ds = 1;\r
+    send_wr.ds_array = &ds_array;\r
+    \r
+    ib_status = ib_post_send(m_qp, &send_wr, NULL);\r
+    ASSERT(ib_status == IB_SUCCESS);\r
+\r
+    \r
+\r
+    // Wait for the notification of send compleated ?????\r
+    rc = MyKeWaitForSingleObject(\r
+                &m_SendCompleteEvent,\r
+                UserRequest,\r
+                UserMode,\r
+                FALSE,\r
+                NULL);    \r
+    KeResetEvent(&m_SendCompleteEvent);\r
+\r
+//Cleanup:\r
+    if (mr_handle != NULL) {\r
+        ib_dereg_mr(mr_handle);\r
+    }\r
+    delete [] BufferStart;\r
+    delete buff; \r
     pWspSendOut->Errno = 0;\r
     pWspSendOut->NumberOfBytesSent = pWspSendIn->BufferSize;\r
 \r
-    return rc;\r
-    \r
+    return rc;   \r
 }\r
 \r
+#endif\r
 \r
 \r
 NTSTATUS SdpSocket::WSPConnect(\r
@@ -117,7 +395,6 @@ NTSTATUS SdpSocket::WSPConnect(
     ib_net64_t SrcPortGuid;\r
     ib_net64_t DestPortGuid;\r
     ib_path_rec_t path_rec;\r
-    CSpinLockWrapper Lock(m_Lock);\r
         \r
     SDP_PRINT(SDP_TRACE, SDP_SOCKET, ("this = 0x%p remote addresses ip=%d.%d.%d.%d:%d\n",\r
         this,\r
@@ -136,13 +413,14 @@ NTSTATUS SdpSocket::WSPConnect(
 \r
     // check socket state\r
     // BUGBUG: Do a better work here\r
-    Lock.Lock();\r
+    m_Lock.Lock();//??? retval\r
     if (m_state != SS_IDLE) {\r
         // We can not connect in this state \r
         SDP_PRINT(SDP_ERR, SDP_SOCKET, ("Invalid Socket state %s\n", SS2String(m_state)));\r
         pWspConnectOut->Errno = WSAEINVAL;\r
+        m_Lock.Unlock(); //?????retval\r
         goto Cleanup;\r
-        Lock.Unlock();\r
+        \r
     }\r
 \r
     //\r
@@ -156,7 +434,7 @@ NTSTATUS SdpSocket::WSPConnect(
         if (!NT_SUCCESS(rc)) {\r
             SDP_PRINT(SDP_ERR, SDP_SOCKET, ("m_pSdpArp->SourceAddrFromDestAddr failed rc = 0x%x\n", rc ));\r
             pWspConnectOut->Errno = WSAENETUNREACH;\r
-            Lock.Unlock();\r
+            m_Lock.Unlock(); // Error ignored as this is already an error pass\r
             goto Cleanup;\r
         }        \r
     }\r
@@ -167,7 +445,7 @@ NTSTATUS SdpSocket::WSPConnect(
        if (!NT_SUCCESS(rc)) {\r
             SDP_PRINT(SDP_ERR, SDP_SOCKET, ("m_pSdpArp->SourcePortGidFromIP failed rc = 0x%x\n", rc ));\r
             pWspConnectOut->Errno = WSAENETUNREACH; // BUGBUG: verify this error\r
-            Lock.Unlock();\r
+            m_Lock.Unlock(); // Error ignored as this is already an error pass\r
             goto Cleanup;\r
         }    \r
     }\r
@@ -175,7 +453,7 @@ NTSTATUS SdpSocket::WSPConnect(
     if (!NT_SUCCESS(rc)) {\r
         SDP_PRINT(SDP_ERR, SDP_SOCKET, ("m_pSdpArp->SourcePortGidFromIP failed rc = 0x%x\n", rc ));\r
         pWspConnectOut->Errno = WSAENETUNREACH; // BUGBUG: verify this error\r
-        Lock.Unlock();\r
+        m_Lock.Unlock(); // Error ignored as this is already an error pass\r
         goto Cleanup;\r
     }        \r
 \r
@@ -183,7 +461,7 @@ NTSTATUS SdpSocket::WSPConnect(
     if (!NT_SUCCESS(rc)) {\r
         SDP_PRINT(SDP_ERR, SDP_SOCKET, ("m_pSdpArp->DestPortGidFromIP failed rc = 0x%x\n", rc ));\r
         pWspConnectOut->Errno = WSAENETUNREACH; // BUGBUG: verify this error\r
-        Lock.Unlock();\r
+        m_Lock.Unlock(); // Error ignored as this is already an error pass\r
         goto Cleanup;\r
     }        \r
 \r
@@ -193,7 +471,7 @@ NTSTATUS SdpSocket::WSPConnect(
 \r
     // Since this is a function that might wait we do it without the lock\r
     m_state = SS_CONNECTING_QPR_SENT;\r
-    Lock.Unlock();\r
+    m_Lock.Unlock(); //?????\r
 \r
     rc = g_pSdpDriver->m_pSdpArp->QueryPathRecord( SrcPortGuid, DestPortGuid, &path_rec );\r
     if (!NT_SUCCESS(rc)) {\r
@@ -217,10 +495,6 @@ NTSTATUS SdpSocket::WSPConnect(
         goto Cleanup;\r
     }    \r
 \r
-//    Lock.Lock(); // Do we really need the lock ?\r
-\r
-\r
-\r
     // We need to prepare the hello mesage for the CM\r
     sdp_msg_hello hello_msg;\r
     CreateHelloHeader(&hello_msg, pWspConnectIn->IP);\r
@@ -230,7 +504,7 @@ NTSTATUS SdpSocket::WSPConnect(
     CreateCmRequest(&cm_req, &hello_msg, &path_rec, pWspConnectIn->Port);\r
 \r
     // Create the event to wait on to the connection request to end:\r
-    KeInitializeEvent(&m_ConnectCmCompleteEvent, NotificationEvent, FALSE );   \r
+    KeInitializeEvent(&m_ConnectCmCompleteEvent, NotificationEvent , FALSE );\r
 \r
     m_state = SS_CONNECTING_REQ_SENT;\r
     \r
@@ -273,18 +547,18 @@ NTSTATUS SdpSocket::WSPConnect(
     }    \r
 \r
     // we should now complete the request\r
-    Lock.Lock();    \r
+    m_Lock.Lock(); //?????  retval\r
     if (m_state == SS_CONNECTED) {\r
         pWspConnectOut->Errno = 0;\r
         ASSERT(rc == STATUS_SUCCESS);\r
-        Lock.Unlock();\r
+        m_Lock.Unlock(); //?????  retval\r
         goto Cleanup;\r
     } else {\r
         // There probably was some error or some kind of shutdown, we \r
         // need to return an error.\r
         rc = STATUS_UNEXPECTED_IO_ERROR;\r
         pWspConnectOut->Errno = WSAENETUNREACH; // BUGBUG: verify this error\r
-        Lock.Unlock();\r
+        m_Lock.Unlock(); // Error ignored as this is already an error pass\r
         goto Cleanup;\r
     }\r
 \r
@@ -327,7 +601,15 @@ NTSTATUS SdpSocket::CmSendRTU()
     if (!NT_SUCCESS(rc)) {\r
         SDP_PRINT(SDP_ERR, SDP_SOCKET, ("sdp_cm_hello_ack_check failed rc = 0x%x\n", rc ));\r
         goto Cleanup;\r
-    }         \r
+    }\r
+\r
+    int MaxMessageSize = min(m_hello_ack.hah.l_rcv_size, MAX_SEND_BUFFER_SIZE);\r
+\r
+    rc = m_SendBufferPool.Init(MAX_SEND_PACKETS, QP_ATTRIB_SQ_DEPTH, MaxMessageSize, m_pd, m_qp);\r
+    if (!NT_SUCCESS(rc)) {\r
+        SDP_PRINT(SDP_ERR, SDP_SOCKET, ("m_SendBufferPool.Init failed rc = 0x%x\n", rc ));\r
+        goto Cleanup;\r
+    }\r
 \r
 #if 0\r
     /*\r
@@ -378,6 +660,21 @@ NTSTATUS SdpSocket::CmSendRTU()
     // How should this be locked ??\r
     m_state = SS_CONNECTED;\r
 \r
+    // we now arm the CQs\r
+    ib_status = ib_rearm_cq(m_rcq, FALSE);\r
+     if( ib_status != IB_SUCCESS ) {\r
+        SDP_PRINT(SDP_ERR, SDP_SOCKET, ("ib_rearm_cq failed ib_status = 0x%d\n", ib_status ));\r
+        rc = IB2Status(ib_status);\r
+        goto Cleanup;\r
+    }\r
+\r
+    ib_status = ib_rearm_cq(m_scq, FALSE);\r
+    if( ib_status != IB_SUCCESS ) {\r
+        SDP_PRINT(SDP_ERR, SDP_SOCKET, ("ib_rearm_cq failed ib_status = 0x%d\n", ib_status ));\r
+        rc = IB2Status(ib_status);\r
+        goto Cleanup;\r
+    }\r
+\r
 Cleanup:    \r
     return rc;\r
 }\r
@@ -414,6 +711,104 @@ __recv_cb1(
     ASSERT(FALSE);\r
 }\r
 \r
+// TODO: Clear the callback functions mess\r
+void\r
+SdpSocket::__send_cb1(\r
+    IN  const   ib_cq_handle_t  h_cq,\r
+    IN  void    *cq_context )\r
+{\r
+    SdpSocket *pSocket = (SdpSocket *) cq_context;\r
+    pSocket->m_Lock.SignalCB(SEND_CB_CALLED);\r
+}\r
+\r
+// This function is here so it's addresses can be taken\r
+static NTSTATUS __send_cb2(SdpSocket * pSdpSocket)\r
+{\r
+    return pSdpSocket->send_cb();\r
+}\r
+\r
+NTSTATUS SdpSocket::send_cb()\r
+{\r
+    SDP_PRINT(SDP_TRACE, SDP_SOCKET, ("called this =0x%x\n", this));\r
+    NTSTATUS rc = STATUS_SUCCESS;\r
+    ib_api_status_t     ib_status;\r
+    ib_wc_t             wc[QP_ATTRIB_SQ_DEPTH], *p_wc, *p_free;\r
+    size_t              i;\r
+    BufferDescriptor *pBufferDescriptor = NULL;\r
+\r
+    for( i = 0; i < QP_ATTRIB_SQ_DEPTH; i++ ) {\r
+        wc[i].p_next = &wc[i + 1];\r
+    }\r
+    wc[QP_ATTRIB_SQ_DEPTH - 1].p_next = NULL;\r
+\r
+    do \r
+    {\r
+        p_free = wc;\r
+        ib_status = ib_poll_cq( m_scq, &p_free, &p_wc );\r
+        ASSERT( ib_status == IB_SUCCESS || ib_status == IB_NOT_FOUND);\r
+        if (ib_status != IB_SUCCESS) {            \r
+            SDP_PRINT(SDP_ERR, SDP_SOCKET, ("ib_poll_cq failed ib_status=%d, this =0x%x\n", ib_status,this));\r
+            ASSERT(ib_status == IB_INVALID_CQ_HANDLE || ib_status == IB_NOT_FOUND);\r
+            rc = IB2Status(ib_status);\r
+            goto Cleanup;\r
+        }\r
+\r
+        while( p_wc )\r
+        {\r
+            ASSERT( p_wc->wc_type == IB_WC_SEND );\r
+            pBufferDescriptor = (BufferDescriptor*)(uintn_t)p_wc->wr_id;\r
+            m_SendBufferPool.ReturnBuffer(pBufferDescriptor);\r
+            \r
+            switch( p_wc->status )\r
+            {\r
+            case IB_WCS_SUCCESS:\r
+                // Nothing to do here\r
+                break;\r
+\r
+            case IB_WCS_WR_FLUSHED_ERR:\r
+                SDP_PRINT(SDP_ERR, SDP_SOCKET, ("Flushed send completion. this =0x%x\n", this));\r
+                // Intentainly fall down\r
+            default:\r
+                SDP_PRINT( SDP_ERR, SDP_SOCKET, ("Send failed with %s\n",\r
+                ib_get_wc_status_str( p_wc->status )) );\r
+                m_Lock.SignalError(IB2Status(ib_status));\r
+            }\r
+/* Do we need this ????\r
+        free the memory that was used for the send\r
+               if( p_send_buf )\r
+               {\r
+                       cl_perf_start( FreeSendBuf );\r
+                       ExFreeToNPagedLookasideList( &p_port->buf_mgr.send_buf_list,\r
+                               p_send_buf );\r
+                       cl_perf_stop( &p_port->p_adapter->perf, FreeSendBuf );\r
+               }\r
+*/\r
+\r
+            p_wc = p_wc->p_next;\r
+        }\r
+        /* If we didn't use up every WC, break out. */\r
+    } while( !p_free );\r
+\r
+\r
+    /* Rearm the CQ. */\r
+    ib_status = ib_rearm_cq(m_scq, FALSE );\r
+    if( ib_status != IB_SUCCESS ) {\r
+        SDP_PRINT(SDP_ERR, SDP_SOCKET, ("ib_rearm_cq failed ib_status = 0x%d\n", ib_status ));\r
+        rc = IB2Status(ib_status);\r
+        goto Cleanup;\r
+    }\r
+\r
+    /* Resume any sends awaiting resources. */\r
+    rc = m_SendBufferPool.SendBuffersIfCan();\r
+    if (!NT_SUCCESS(rc)) {\r
+        SDP_PRINT(SDP_ERR, SDP_SOCKET, ("m_SendBufferPool.Init SendBuffersIfCan rc = 0x%x\n", rc ));\r
+        goto Cleanup;\r
+    }\r
+\r
+Cleanup:\r
+    return rc;\r
+\r
+}\r
 \r
 \r
 // BUGBUG: This code is based on __cq_event, find out what it realy does\r
@@ -501,7 +896,7 @@ NTSTATUS SdpSocket::CreateQp()
 \r
     /* Allocate send CQ. */\r
     cq_create.size = QP_ATTRIB_SQ_DEPTH;\r
-    cq_create.pfn_comp_cb = __recv_cb1; // ???? We are not doing anything there ??? why bother\r
+    cq_create.pfn_comp_cb = SdpSocket::__send_cb1;\r
 \r
     ib_status = ib_create_cq(\r
                     mh_Ca, \r
@@ -628,7 +1023,7 @@ VOID SdpSocket::CreateCmRequest(
         else if( cm_req->local_resp_timeout < CM_MIN_LOCAL_TIMEOUT )\r
         cm_req->local_resp_timeout = CM_MIN_LOCAL_TIMEOUT;\r
 \r
-    cm_req->rnr_nak_timeout = 6;//???QP_ATTRIB_RNR_NAK_TIMEOUT;\r
+    cm_req->rnr_nak_timeout = 6;//6;//???QP_ATTRIB_RNR_NAK_TIMEOUT;\r
     cm_req->rnr_retry_cnt = 6;//????QP_ATTRIB_RNR_RETRY;\r
     cm_req->retry_cnt = 6;//????QP_ATTRIB_RETRY_COUNT;\r
 \r
@@ -641,8 +1036,7 @@ VOID SdpSocket::CreateCmRequest(
 VOID SdpSocket::Shutdown()\r
 {\r
     //???? locking\r
-    // if(m_shutdown) ???\r
-    m_shutdown = true;\r
+    // if(m_shutdown - on the lock) ???\r
 \r
     SDP_PRINT(SDP_TRACE, SDP_SOCKET, ("SdpSocket::Shutdown called this = 0x%p\n", this));\r
 \r
@@ -665,6 +1059,9 @@ VOID SdpSocket::Shutdown()
     if (mh_Ca != NULL) {\r
         ib_close_ca(mh_Ca, NULL); //?????? CALL BACK ??? IMPLMENT\r
     }\r
+\r
+    // Now that all ibal operations have finished we can free the memory\r
+    m_SendBufferPool.ShutDown();\r
 }\r
 \r
 /*\r
index 1850583..adb4420 100644 (file)
@@ -11,6 +11,10 @@ It keeps a list of all the objects so we know when to remove them.
 #ifndef _SDP_SOCKET_H\r
 #define _SDP_SOCKET_H\r
 \r
+const int MAX_SEND_BUFFER_SIZE          = 32768; // This is the maximum send packet size\r
+const int MAX_SEND_PACKETS              = 40;    // This is the maximum number of packets allocated per send\r
+\r
+                                                    \r
 \r
 #define QP_ATTRIB_SQ_DEPTH                             16\r
 #define QP_ATTRIB_SQ_SGE                               1       /* Set based on inline data requirements */\r
@@ -35,6 +39,7 @@ enum SocketStates {
 };\r
 \r
 \r
+\r
 class SdpSocket : public RefCountImpl {\r
 private: \r
    \r
@@ -47,22 +52,30 @@ private:
     USHORT m_SrcPort;\r
     ULONG  m_SrcIp;\r
 \r
-\r
-    KSPIN_LOCK  m_Lock;\r
-    bool        m_shutdown;\r
+    bool                    m_shutdown; // Make sure this is synced w\r
+    SdpLock                 m_Lock;\r
 \r
 \r
     // A handle to the ca that is being used (in connect) and its guid\r
     ib_ca_handle_t          mh_Ca;\r
     net64_t                 m_CaGuid;\r
-\r
     ib_pd_handle_t          m_pd;\r
     ib_cq_handle_t          m_rcq;\r
+\r
     ib_cq_handle_t          m_scq;\r
+   \r
     ib_qp_handle_t          m_qp;\r
 \r
+    BufferPool              m_SendBufferPool;\r
+\r
     KEVENT                  m_ConnectCmCompleteEvent;\r
 \r
+    VOID SignalShutdown();\r
+\r
+    static VOID __send_cb1(\r
+    IN  const   ib_cq_handle_t  h_cq,\r
+    IN  void    *cq_context );\r
+\r
 public:\r
     SdpSocket();\r
 \r
@@ -105,7 +118,7 @@ public:
     struct sdp_msg_hello_ack m_hello_ack;\r
     ib_cm_handle_t m_cm_handle_t; // BUGBUG: Check how this is used / locked\r
 \r
-    \r
+    NTSTATUS send_cb();\r
 \r
     // Used to allow the user file to remember us\r
     LIST_ENTRY m_UserFileList;\r
index ee93583..d69c2a2 100644 (file)
@@ -5,6 +5,7 @@
 \r
 BOOLEAN CheckCondition(int sev, int top, char *file, int line, char * func)\r
 {\r
-    DbgPrint ("%s ", func);\r
+//    return FALSE;\r
+    DbgPrint ("%s: ", func);\r
     return TRUE;\r
 }
\ No newline at end of file
index 77b1554..27526bd 100644 (file)
@@ -15,6 +15,9 @@
 #define SDP_DRIVER          0x000004\r
 #define SDP_SOCKET          0x000008\r
 #define SDP_ARP             0x000010\r
+#define SDP_BUFFER_POOL     0x000020\r
+#define SDP_LOCK            0x000040\r
+\r
 \r
 // BUGBUG: CONVERT TO A FUNCTION\r
 \r