Merge pull request #960 from menzels/improve-multi-thread

Improve multi threading

Author: softins

.gitignore

@@ -10,6 +10,7 @@ Makefile
 .settings
 .idea
 .vscode
+.cache
 *.user
 *.user.*
 *.o

Jamulus.pro

@@ -403,6 +403,7 @@ HEADERS += src/buffer.h \
     src/global.h \
     src/protocol.h \
     src/recorder/jamcontroller.h \
+    src/threadpool.h \
     src/server.h \
     src/serverlist.h \
     src/serverlogging.h \

src/server.cpp

@@ -421,10 +421,24 @@ CServer::CServer ( const int          iNewMaxNumChan,
         vecChannels[i].SetEnable ( true );
     }
 
-    // introduced by kraney (#653): "increased the size of the thread pool"
-    if ( bUseMultithreading )
-    {
-        QThreadPool::globalInstance()->setMaxThreadCount ( QThread::idealThreadCount() * 4 );
+    int iAvailableCores = QThread::idealThreadCount();
+
+    // setup QThreadPool if multithreading is active and possible
+    if ( bUseMultithreading ) {
+        if ( iAvailableCores == 1 )
+        {
+            qDebug() << "found only one core, disabling multithreading";
+            bUseMultithreading = false;
+        }
+        else
+        {
+            // set maximum thread count to available cores; other threads will share at random
+            iMaxNumThreads = iAvailableCores;
+            qDebug() << "multithreading enabled, setting thread count to" << iMaxNumThreads;
+
+            pThreadPool = std::unique_ptr<CThreadPool>( new CThreadPool{static_cast<size_t>(iMaxNumThreads)} );
+            Futures.reserve( iMaxNumThreads );
+        }
     }
 
 
@@ -640,8 +654,8 @@ void CServer::OnNewConnection ( int          iChID,
     vecChannels[iChID].CreateReqChanInfoMes();
 
     // send welcome message (if enabled)
-    MutexWelcomeMessage.lock();
     {
+        QMutexLocker locker ( &MutexWelcomeMessage );
         if ( !strWelcomeMessage.isEmpty() )
         {
             // create formatted server welcome message and send it just to
@@ -652,7 +666,6 @@ void CServer::OnNewConnection ( int          iChID,
             vecChannels[iChID].CreateChatTextMes ( strWelcomeMessageFormated );
         }
     }
-    MutexWelcomeMessage.unlock();
 
     // send licence request message (if enabled)
     if ( eLicenceType != LT_NO_LICENCE )
@@ -707,17 +720,14 @@ void CServer::OnAboutToQuit()
     // if enabled, disconnect all clients on quit
     if ( bDisconnectAllClientsOnQuit )
     {
-        Mutex.lock();
+        QMutexLocker locker ( &Mutex );
+        for ( int i = 0; i < iMaxNumChannels; i++ )
         {
-            for ( int i = 0; i < iMaxNumChannels; i++ )
+            if ( vecChannels[i].IsConnected() )
             {
-                if ( vecChannels[i].IsConnected() )
-                {
-                    ConnLessProtocol.CreateCLDisconnection ( vecChannels[i].GetAddress() );
-                }
+                ConnLessProtocol.CreateCLDisconnection ( vecChannels[i].GetAddress() );
             }
         }
-        Mutex.unlock(); // release mutex
     }
 
     Stop();
@@ -805,13 +815,16 @@ static CTimingMeas JitterMeas ( 1000, "test2.dat" ); JitterMeas.Measure(); // TE
 */
     // Get data from all connected clients -------------------------------------
     // some inits
-    int iNumClients           = 0; // init connected client counter
+    int  iNumClients          = 0; // init connected client counter
+    bool bUseMT               = false;
+    int  iNumBlocks           = 0; // init number of blocks for multithreading
+    int  iMTBlockSize         = 0; // init block size for multithreading
     bChannelIsNowDisconnected = false; // note that the flag must be a member function since QtConcurrent::run can only take 5 params
 
-    // Make put and get calls thread safe. Do not forget to unlock mutex
-    // afterwards!
-    Mutex.lock();
     {
+        // Make put and get calls thread safe.
+        QMutexLocker locker ( &Mutex );
+
         // first, get number and IDs of connected channels
         for ( int i = 0; i < iMaxNumChannels; i++ )
         {
@@ -826,21 +839,23 @@ static CTimingMeas JitterMeas ( 1000, "test2.dat" ); JitterMeas.Measure(); // TE
             }
         }
 
+        // use multithreading for any non-zero number of clients
+        // (overhead is low and it is worth doing for all numbers)
+        bUseMT = bUseMultithreading && iNumClients > 0;
+
         // prepare and decode connected channels
-        if ( !bUseMultithreading )
+        if ( !bUseMT )
         {
-            for ( int iChanCnt = 0; iChanCnt < iNumClients; iChanCnt++ )
-            {
-                DecodeReceiveData ( iChanCnt, iNumClients );
-            }
+            // run the OPUS decoder for all data blocks
+            DecodeReceiveDataBlocks ( this, 0, iNumClients - 1, iNumClients );
         }
         else
         {
-            // processing with multithreading
-// TODO optimization of the MTBlockSize value
-            const int iMTBlockSize = 10; // every 10 users a new thread is created
-            const int iNumBlocks   = ( iNumClients - 1 ) / iMTBlockSize + 1;
+            // spread work equally among available threads
+            iNumBlocks   = std::min ( iNumClients, iMaxNumThreads );
+            iMTBlockSize = ( iNumClients - 1 ) / iNumBlocks + 1;
 
+            // processing with multithreading
             for ( int iBlockCnt = 0; iBlockCnt < iNumBlocks; iBlockCnt++ )
             {
                 // The work for OPUS decoding is distributed over all available processor cores.
@@ -849,16 +864,18 @@ static CTimingMeas JitterMeas ( 1000, "test2.dat" ); JitterMeas.Measure(); // TE
                 const int iStartChanCnt = iBlockCnt * iMTBlockSize;
                 const int iStopChanCnt  = std::min ( ( iBlockCnt + 1 ) * iMTBlockSize - 1, iNumClients - 1 );
 
-                FutureSynchronizer.addFuture ( QtConcurrent::run ( this,
-                                                                   &CServer::DecodeReceiveDataBlocks,
-                                                                   iStartChanCnt,
-                                                                   iStopChanCnt,
-                                                                   iNumClients ) );
+                Futures.push_back ( pThreadPool->enqueue ( CServer::DecodeReceiveDataBlocks,
+                                                           this,
+                                                           iStartChanCnt,
+                                                           iStopChanCnt,
+                                                           iNumClients ) );
             }
 
             // make sure all concurrent run threads have finished when we leave this function
-            FutureSynchronizer.waitForFinished();
-            FutureSynchronizer.clearFutures();
+            for ( auto& future : Futures ) {
+                future.wait();
+            }
+            Futures.clear();
         }
 
         // a channel is now disconnected, take action on it
@@ -868,7 +885,6 @@ static CTimingMeas JitterMeas ( 1000, "test2.dat" ); JitterMeas.Measure(); // TE
             CreateAndSendChanListForAllConChannels();
         }
     }
-    Mutex.unlock(); // release mutex
 
 
     // Process data ------------------------------------------------------------
@@ -909,7 +925,7 @@ static CTimingMeas JitterMeas ( 1000, "test2.dat" ); JitterMeas.Measure(); // TE
             }
 
             // processing without multithreading
-            if ( !bUseMultithreading )
+            if ( !bUseMT )
             {
                 // generate a separate mix for each channel, OPUS encode the
                 // audio data and transmit the network packet
@@ -918,14 +934,8 @@ static CTimingMeas JitterMeas ( 1000, "test2.dat" ); JitterMeas.Measure(); // TE
         }
 
         // processing with multithreading
-        if ( bUseMultithreading )
+        if ( bUseMT )
         {
-            // introduced by kraney (#653): each thread must complete within the 1 or 2ms time budget for the timer
-// TODO determine at startup by running a small benchmark
-            const int iMaximumMixOpsInTimeBudget = 500; // approximate limit as observed on GCP e2-standard instance
-            const int iMTBlockSize = iMaximumMixOpsInTimeBudget / iNumClients; // number of ops = block size * total number of clients
-            const int iNumBlocks   = ( iNumClients - 1 ) / iMTBlockSize + 1;
-
             for ( int iBlockCnt = 0; iBlockCnt < iNumBlocks; iBlockCnt++ )
             {
                 // Generate a separate mix for each channel, OPUS encode the
@@ -936,16 +946,18 @@ static CTimingMeas JitterMeas ( 1000, "test2.dat" ); JitterMeas.Measure(); // TE
                 const int iStartChanCnt = iBlockCnt * iMTBlockSize;
                 const int iStopChanCnt  = std::min ( ( iBlockCnt + 1 ) * iMTBlockSize - 1, iNumClients - 1 );
 
-                FutureSynchronizer.addFuture ( QtConcurrent::run ( this,
-                                                                   &CServer::MixEncodeTransmitDataBlocks,
-                                                                   iStartChanCnt,
-                                                                   iStopChanCnt,
-                                                                   iNumClients ) );
+                Futures.push_back ( pThreadPool->enqueue ( CServer::MixEncodeTransmitDataBlocks,
+                                                           this,
+                                                           iStartChanCnt,
+                                                           iStopChanCnt,
+                                                           iNumClients ) );
             }
 
             // make sure all concurrent run threads have finished when we leave this function
-            FutureSynchronizer.waitForFinished();
-            FutureSynchronizer.clearFutures();
+            for ( auto& fFuture : Futures ) {
+                fFuture.wait();
+            }
+            Futures.clear();
         }
         if ( bDelayPan )
         {
@@ -966,25 +978,31 @@ static CTimingMeas JitterMeas ( 1000, "test2.dat" ); JitterMeas.Measure(); // TE
     }
 }
 
-void CServer::DecodeReceiveDataBlocks ( const int iStartChanCnt,
+// This is a static method used as a callback, and does not inherit a "this" pointer,
+// so it is necessary for the server instance to be passed as a parameter.
+void CServer::DecodeReceiveDataBlocks ( CServer*  pServer,
+                                        const int iStartChanCnt,
                                         const int iStopChanCnt,
                                         const int iNumClients )
 {
     // loop over all channels in the current block, needed for multithreading support
     for ( int iChanCnt = iStartChanCnt; iChanCnt <= iStopChanCnt; iChanCnt++ )
     {
-        DecodeReceiveData ( iChanCnt, iNumClients );
+        pServer->DecodeReceiveData ( iChanCnt, iNumClients );
     }
 }
 
-void CServer::MixEncodeTransmitDataBlocks ( const int iStartChanCnt,
+// This is a static method used as a callback, and does not inherit a "this" pointer,
+// so it is necessary for the server instance to be passed as a parameter.
+void CServer::MixEncodeTransmitDataBlocks ( CServer*  pServer,
+                                            const int iStartChanCnt,
                                             const int iStopChanCnt,
                                             const int iNumClients )
 {
     // loop over all channels in the current block, needed for multithreading support
     for ( int iChanCnt = iStartChanCnt; iChanCnt <= iStopChanCnt; iChanCnt++ )
     {
-        MixEncodeTransmitData ( iChanCnt, iNumClients );
+        pServer->MixEncodeTransmitData ( iChanCnt, iNumClients );
     }
 }
 
@@ -1088,7 +1106,7 @@ void CServer::DecodeReceiveData ( const int iChanCnt,
         // get current number of OPUS coded bytes
         const int iCeltNumCodedBytes = vecChannels[iCurChanID].GetCeltNumCodedBytes();
 
-        for ( int iB = 0; iB < vecNumFrameSizeConvBlocks[iChanCnt]; iB++ )
+        for ( size_t iB = 0; iB < (size_t)vecNumFrameSizeConvBlocks[iChanCnt]; iB++ )
         {
             // get data
             const EGetDataStat eGetStat = vecChannels[iCurChanID].GetData ( vecvecbyCodedData[iChanCnt], iCeltNumCodedBytes );
@@ -1390,26 +1408,26 @@ void CServer::MixEncodeTransmitData ( const int iChanCnt,
             DoubleFrameSizeConvBufOut[iCurChanID].GetAll ( vecsSendData, DOUBLE_SYSTEM_FRAME_SIZE_SAMPLES * vecNumAudioChannels[iChanCnt] );
         }
 
-        for ( int iB = 0; iB < vecNumFrameSizeConvBlocks[iChanCnt]; iB++ )
+        // OPUS encoding
+        if ( pCurOpusEncoder != nullptr )
         {
-            // OPUS encoding
-            if ( pCurOpusEncoder != nullptr )
-            {
 // TODO find a better place than this: the setting does not change all the time so for speed
 //      optimization it would be better to set it only if the network frame size is changed
 opus_custom_encoder_ctl ( pCurOpusEncoder, OPUS_SET_BITRATE ( CalcBitRateBitsPerSecFromCodedBytes ( iCeltNumCodedBytes, iClientFrameSizeSamples ) ) );
 
+            for ( size_t iB = 0; iB < (size_t)vecNumFrameSizeConvBlocks[iChanCnt]; iB++ )
+            {
                 iUnused = opus_custom_encode ( pCurOpusEncoder,
                                                &vecsSendData[iB * SYSTEM_FRAME_SIZE_SAMPLES * vecNumAudioChannels[iChanCnt]],
                                                iClientFrameSizeSamples,
                                                &vecvecbyCodedData[iChanCnt][0],
                                                iCeltNumCodedBytes );
-            }
 
-            // send separate mix to current clients
-            vecChannels[iCurChanID].PrepAndSendPacket ( &Socket,
-                                                        vecvecbyCodedData[iChanCnt],
-                                                        iCeltNumCodedBytes );
+                // send separate mix to current clients
+                vecChannels[iCurChanID].PrepAndSendPacket ( &Socket,
+                                                            vecvecbyCodedData[iChanCnt],
+                                                            iCeltNumCodedBytes );
+            }
         }
     }