Changeset 26396 in vbox

Timestamp:

Feb 9, 2010 6:32:26 PM (15 years ago)

Author:

vboxsync

Message:

Web service: make no. of threads a cmd line param, encapsulate queue code

File:

• trunk/src/VBox/Main/webservice/vboxweb.cpp (modified) (11 diffs)

trunk/src/VBox/Main/webservice/vboxweb.cpp

@@ -142 +142 @@
         { "--timeout",          't', RTGETOPT_REQ_UINT32 },
         { "--check-interval",   'i', RTGETOPT_REQ_UINT32 },
+        { "--threads",          'T', RTGETOPT_REQ_UINT32 },
         { "--verbose",          'v', RTGETOPT_REQ_NOTHING },
         { "--logfile",          'F', RTGETOPT_REQ_STRING },
@@ -182 +183 @@
             case 't':
                 pcszDescr = "Session timeout in seconds; 0 = disable timeouts (" DEFAULT_TIMEOUT_SECS_STRING ").";
+            break;
+
+            case 'T':
+                pcszDescr = "Number of worker threads to run in parallel (5).";
             break;
 
@@ -295 +300 @@
     }
 
+    /**
+     * Adds the given socket to the SOAP queue and posts the
+     * member event sem to wake up the workers.
+     * @param s Socket from soap_accept() which has work to do.
+     */
+    uint32_t add(int s)
+    {
+        uint32_t cItems;
+        // enqueue the socket of this connection and post eventsem so
+        // that one of our threads can pick it up
+        util::AutoWriteLock qlock(m_mutex COMMA_LOCKVAL_SRC_POS);
+        m_llSocketsQ.push_back(s);
+        cItems = m_llSocketsQ.size();
+        qlock.release();
+
+        // unblock one of the worker threads
+        RTSemEventMultiSignal(m_event);
+
+        return cItems;
+    }
+
+    /**
+     * Blocks the current thread until work comes in; then returns
+     * the SOAP socket which has work to do. This reduces m_cIdleThreads
+     * by one, and the caller MUST call done() when it's done processing.
+     * @return
+     */
+    int get()
+    {
+        while (1)
+        {
+            // wait for something to happen
+            RTSemEventMultiWait(m_event, RT_INDEFINITE_WAIT);
+
+            util::AutoWriteLock qlock(m_mutex COMMA_LOCKVAL_SRC_POS);
+            if (m_llSocketsQ.size())
+            {
+                int socket = m_llSocketsQ.front();
+                m_llSocketsQ.pop_front();
+                --m_cIdleThreads;
+
+                // reset the multi event only if the queue is now empty; otherwise
+                // another thread will also wake up when we release the mutex and
+                // process another one
+                if (m_llSocketsQ.size() == 0)
+                    RTSemEventMultiReset(m_event);
+
+                qlock.release();
+
+                return socket;
+            }
+
+            // nothing to do: keep looping
+        }
+    }
+
+    /**
+     * To be called by a thread after fetching an item from the
+     * queue via get() and having finished its lengthy processing.
+     */
+    void done()
+    {
+        util::AutoWriteLock qlock(m_mutex COMMA_LOCKVAL_SRC_POS);
+        ++m_cIdleThreads;
+    }
+
     util::WriteLockHandle   m_mutex;
     RTSEMEVENTMULTI         m_event;
@@ -305 +376 @@
 };
 
+/**
+ * Thread function for each of the SOAP queue worker threads. This keeps
+ * running, blocks on the event semaphore in SoapThread.SoapQ and picks
+ * up a socket from the queue therein, which has been put there by
+ * beginProcessing().
+ *
+ * @param pThread
+ * @param pvThread
+ * @return
+ */
 int fntSoapQueue(RTTHREAD pThread, void *pvThread)
 {
@@ -313 +394 @@
     while (1)
     {
-        // wait for something to happen
-        RTSemEventMultiWait(pst->pQ->m_event, RT_INDEFINITE_WAIT);
-
-        util::AutoWriteLock qlock(pst->pQ->m_mutex COMMA_LOCKVAL_SRC_POS);
-        if (pst->pQ->m_llSocketsQ.size())
-        {
-            pst->soap->socket = pst->pQ->m_llSocketsQ.front();
-            pst->pQ->m_llSocketsQ.pop_front();
-            --pst->pQ->m_cIdleThreads;
-
-            // reset the multi event only if the queue is now empty; otherwise
-            // another thread will also wake up when we release the mutex and
-            // process another one
-            if (pst->pQ->m_llSocketsQ.size() == 0)
-                RTSemEventMultiReset(pst->pQ->m_event);
-
-            qlock.release();
-
-            WebLog("Thread %d is handling connection from IP=%lu.%lu.%lu.%lu socket=%d (%d threads idle)",
-                   pst->u,
-                   (pst->soap->ip>>24)&0xFF,
-                   (pst->soap->ip>>16)&0xFF,
-                   (pst->soap->ip>>8)&0xFF,
-                   pst->soap->ip&0xFF,
-                   pst->soap->socket,
-                   pst->pQ->m_cIdleThreads);
-
-            // process the request; this goes into the COM code in methodmaps.cpp
-            soap_serve(pst->soap);
-
-            soap_destroy(pst->soap); // clean up class instances
-            soap_end(pst->soap); // clean up everything and close socket
-
-            qlock.acquire();
-            ++pst->pQ->m_cIdleThreads;
-        }
+        // wait for a socket to arrive on the queue
+        pst->soap->socket = pst->pQ->get();
+
+        WebLog("T%d handles connection from IP=%lu.%lu.%lu.%lu socket=%d (%d thr idle)\n",
+                pst->u,
+                (pst->soap->ip>>24)&0xFF,
+                (pst->soap->ip>>16)&0xFF,
+                (pst->soap->ip>>8)&0xFF,
+                pst->soap->ip&0xFF,
+                pst->soap->socket,
+                pst->pQ->m_cIdleThreads);
+
+        // process the request; this goes into the COM code in methodmaps.cpp
+        soap_serve(pst->soap);
+
+        soap_destroy(pst->soap); // clean up class instances
+        soap_end(pst->soap); // clean up everything and close socket
+
+        // tell the queue we're idle again
+        pst->pQ->done();
     }
 
@@ -356 +421 @@
 /**
  * Called from main(). This implements the loop that takes SOAP calls
- * from HTTP and serves them, calling the COM method implementations
- * in the generated methodmaps.cpp code.
- */
-void beginProcessing()
+ * from HTTP and serves them by handing sockets to the SOAP queue
+ * worker threads.
+ */
+void beginProcessing(size_t cThreads)
 {
     // set up gSOAP
@@ -382 +447 @@
                m);
 
-        // initialize thread queue, mutex and eventsem, create 10 threads
-        SoapQ soapq(10, &soap);
+        // initialize thread queue, mutex and eventsem, create worker threads
+        SoapQ soapq(cThreads, &soap);
 
         for (uint64_t i = 1;
@@ -397 +462 @@
             }
 
-            // enqueue the socket of this connection and post eventsem so
-            // that one of our threads can pick it up
-            util::AutoWriteLock qlock(soapq.m_mutex COMMA_LOCKVAL_SRC_POS);
-            soapq.m_llSocketsQ.push_back(s);
-            qlock.release();
-
-            WebLog("Request %llu on socket %d queued for processing\n", i, s);
-            // unblock one of the worker threads
-            RTSemEventMultiSignal(soapq.m_event);
+            // add the socket to the queue and tell worker threads to
+            // pick up the jobn
+            size_t cItemsOnQ = soapq.add(s);
+            WebLog("Request %llu on socket %d queued for processing (%d items on Q)\n", i, s, cItemsOnQ);
 
             // we have to process main event queue
@@ -427 +487 @@
 {
     int rc;
+
+    uint32_t cWorkerThreads = 5;
 
     // intialize runtime
@@ -471 +533 @@
                        "Opened log file \"%s\"\n", VBOX_VERSION_STRING, ValueUnion.psz);
             }
+            break;
+
+            case 'T':
+                cWorkerThreads = ValueUnion.u32;
             break;
 
@@ -577 +643 @@
     }
 
-    beginProcessing();
+    beginProcessing(cWorkerThreads);
 
     com::Shutdown();