[core] Removed unnecessary post-cleaning of closed socket in the cleanup.

scripts/win-installer/build-win-installer.ps1

@@ -37,7 +37,8 @@
 param(
     [string]$Version = "",
     [switch]$NoBuild = $false,
-    [switch]$NoPause = $false
+    [switch]$NoPause = $false,
+	[string]$OnlyPlatform = ""
 )
 Write-Output "Building the SRT static libraries installer for Windows"
 
@@ -113,6 +114,9 @@ function ssl-lib($pf, $conf, $lib) { return "$(ssl-libdir $pf $conf)\$($LIBFILE.
 Write-Output "Searching OpenSSL libraries ..."
 $Missing = 0
 foreach ($Platform in $SSL.Keys) {
+	if ($OnlyPlatform -ne "" -and $Platform -ne $OnlyPlatform) {
+		continue
+	}
     if (-not (Test-Path -PathType Container $(ssl-include $Platform))) {
         Write-Output "**** Missing $(ssl-include $Platform)"
         $Missing = $Missing + 1
@@ -173,19 +177,32 @@ Write-Output "NSIS: $NSIS"
 #-----------------------------------------------------------------------------
 
 if (-not $NoBuild) {
+	Write-Output "BUILD ENABLED - configuring build"
+	if ($OnlyPlatform -ne "") {
+		Write-Output "LIMIT TO PLATFORM: $OnlyPlatform"
+	}
     foreach ($Platform in $ARCH.Keys) {
+		if ($OnlyPlatform -ne "" -and $Platform -ne $OnlyPlatform) {
+			Write-Output "Skipping platform $Platform."
+			continue
+		}
+		Write-Output "Preparing platform $Platform."
         # Build directory. Cleanup to force a fresh cmake config.
         $BuildDir = "$TmpDir\build.$Platform"
         Remove-Item -Recurse -Force -ErrorAction SilentlyContinue $BuildDir
         [void](New-Item -Path $BuildDir -ItemType Directory -Force)
 
         # Run CMake.
         Write-Output "Configuring build for platform $Platform ..."
-        & $CMake -S $RepoDir -B $BuildDir -A $Platform `
-            -DENABLE_STDCXX_SYNC=ON `
-            -DOPENSSL_INCLUDE_DIR="$(ssl-include $Platform)" `
-            -DOPENSSL_ROOT_DIR="$(ssl-libdir $Platform Release)"
-
+		$params = @("-S", "$RepoDir",
+					"-B", "$BuildDir",
+					"-A", "$Platform",
+					"-DENABLE_STDCXX_SYNC=ON",
+					"-DOPENSSL_INCLUDE_DIR=$(ssl-include $Platform)",
+					"-DOPENSSL_ROOT_DIR=$(ssl-libdir $Platform Release)"
+					)
+		Write-Output "Running: cmake $params"
+        & $CMake $params
         # Patch version string in version.h
         Get-Content "$BuildDir\version.h" |
             ForEach-Object {
@@ -197,7 +214,17 @@ if (-not $NoBuild) {
         # Compile SRT.
         Write-Output "Building for platform $Platform ..."
         foreach ($Conf in $LIBDIR.Keys) {
-            & $MSBuild "$BuildDir\SRT.sln" /nologo /maxcpucount /property:Configuration=$Conf /property:Platform=$Platform /target:srt_static
+           # The solution file format depends on the CMake version.
+           # Try new XML solution file format first.
+           $SolFile = "$BuildDir\SRT.slnx"
+           if (-not (Test-Path $SolFile)) {
+               # Try legacy solution file format.
+               $SolFile = "$BuildDir\SRT.sln"
+           }
+           if (-not (Test-Path $SolFile)) {
+               Exit-Script "Solution file $BuildDir\SRT.sln[x] not found, check CMake output"
+           }
+           & $MSBuild $SolFile /nologo /maxcpucount /property:Configuration=$Conf /property:Platform=$Platform /target:srt_static
         }
     }
 }
@@ -207,6 +234,9 @@ Write-Output "Checking compiled libraries ..."
 $Missing = 0
 foreach ($Conf in $LIBDIR.Keys) {
     foreach ($Platform in $SSL.Keys) {
+		if ($OnlyPlatform -ne "" -and $Platform -ne $OnlyPlatform) {
+			continue
+		}
         $Path = "$TmpDir\build.$Platform\$Conf\srt_static.lib"
         if (-not (Test-Path $Path)) {
             Write-Output "**** Missing $Path"

srtcore/api.cpp

@@ -2896,54 +2896,84 @@
         leaveCS(ls->second->m_AcceptLock);
     }
 
-    for (sockets_t::iterator j = m_ClosedSockets.begin(); j != m_ClosedSockets.end(); ++j)
-    {
-        CUDTSocket* ps = j->second;
-
-        // NOTE: There is still a hypothetical risk here that ps
-        // was made busy while the socket was already moved to m_ClosedSocket,
-        // if the socket was acquired through CUDTUnited::acquireSocket (that is,
-        // busy flag acquisition was done through the CUDTSocket* pointer rather
-        // than through the numeric ID). Therefore this way of busy acquisition
-        // should be done only if at the moment of acquisition there are certainly
-        // other conditions applying on the socket that prevent it from being deleted.
-        if (ps->isStillBusy())
+    if (!m_bGCStatus)
+    {
+        // The GC thread should have deleted all sockets by now; if there
+        // are any left, then likely the GC was interrupted before it could
+        // finish the job. We assume then that all threads have been wiped,
+        // so all we can do is to delete all sockets forcefully.
+
+        for (sockets_t::iterator j = m_ClosedSockets.begin(); j != m_ClosedSockets.end(); ++j)
         {
-            HLOGC(smlog.Debug, log << "checkBrokenSockets: @" << ps->m_SocketID << " is still busy, SKIPPING THIS CYCLE.");
-            continue;
-        }
+            CUDTSocket* ps = j->second;
+            CUDT* pu = &ps->core();
+            CRNode* rnode = pu->m_pRNode;
 
-        CUDT& u = ps->core();
+            // Ignore BUSY flag.
+            // Ignore Linger.
+            // Regard CRcvUList membership.
 
-        // HLOGC(smlog.Debug, log << "checking CLOSED socket: " << j->first);
-        if (!is_zero(u.m_tsLingerExpiration))
-        {
-            // asynchronous close:
-            if ((!u.m_pSndBuffer) || (0 == u.m_pSndBuffer->getCurrBufSize()) ||
-                (u.m_tsLingerExpiration <= steady_clock::now()))
+            if (rnode && rnode->m_bOnList)
             {
-                HLOGC(smlog.Debug, log << "checkBrokenSockets: marking CLOSED qualified @" << ps->m_SocketID);
-                u.m_tsLingerExpiration = steady_clock::time_point();
-                u.m_bClosing           = true;
-                ps->m_tsClosureTimeStamp        = steady_clock::now();
+                // Unstable rnode membership.
+                pu->m_pRcvQueue->forceRemove(pu);
             }
+
+            tbr.push_back(j->first);
         }
 
-        // timeout 1 second to destroy a socket AND it has been removed from
-        // RcvUList
-        const steady_clock::time_point now        = steady_clock::now();
-        const steady_clock::duration   closed_ago = now - ps->m_tsClosureTimeStamp.load();
-        if (closed_ago > seconds_from(1))
+    }
+    else
+    {
+        for (sockets_t::iterator j = m_ClosedSockets.begin(); j != m_ClosedSockets.end(); ++j)
         {
-            CRNode* rnode = u.m_pRNode;
-            if (!rnode || !rnode->m_bOnList)
+            CUDTSocket* ps = j->second;
+
+            // NOTE: There is still a hypothetical risk here that ps
+            // was made busy while the socket was already moved to m_ClosedSocket,
+            // if the socket was acquired through CUDTUnited::acquireSocket (that is,
+            // busy flag acquisition was done through the CUDTSocket* pointer rather
+            // than through the numeric ID). Therefore this way of busy acquisition
+            // should be done only if at the moment of acquisition there are certainly
+            // other conditions applying on the socket that prevent it from being deleted.
+            if (ps->isStillBusy())
             {
-                HLOGC(smlog.Debug,
-                      log << "checkBrokenSockets: @" << ps->m_SocketID << " closed "
-                          << FormatDuration(closed_ago) << " ago and removed from RcvQ - will remove");
+                HLOGC(smlog.Debug, log << "checkBrokenSockets: @" << ps->m_SocketID << " is still busy, SKIPPING THIS CYCLE.");
+                continue;
+            }
+
+            CUDT& u = ps->core();
+
+            // HLOGC(smlog.Debug, log << "checking CLOSED socket: " << j->first);
+            if (!is_zero(u.m_tsLingerExpiration))
+            {
+                // asynchronous close:
+                if ((!u.m_pSndBuffer) || (0 == u.m_pSndBuffer->getCurrBufSize()) ||
+                        (u.m_tsLingerExpiration <= steady_clock::now()))
+                {
+                    HLOGC(smlog.Debug, log << "checkBrokenSockets: marking CLOSED qualified @" << ps->m_SocketID);
+                    u.m_tsLingerExpiration = steady_clock::time_point();
+                    u.m_bClosing           = true;
+                    ps->m_tsClosureTimeStamp        = steady_clock::now();
+                }
+            }
+
+            // timeout 1 second to destroy a socket AND it has been removed from
+            // RcvUList
+            const steady_clock::time_point now        = steady_clock::now();
+            const steady_clock::duration   closed_ago = now - ps->m_tsClosureTimeStamp.load();
+            if (closed_ago > seconds_from(1))
+            {
+                CRNode* rnode = u.m_pRNode;
+                if (!rnode || !rnode->m_bOnList)
+                {
+                    HLOGC(smlog.Debug,
+                            log << "checkBrokenSockets: @" << ps->m_SocketID << " closed "
+                            << FormatDuration(closed_ago) << " ago and removed from RcvQ - will remove");
 
-                // HLOGC(smlog.Debug, log << "will unref socket: " << j->first);
-                tbr.push_back(j->first);
+                    // HLOGC(smlog.Debug, log << "will unref socket: " << j->first);
+                    tbr.push_back(j->first);
+                }
             }
         }
     }

srtcore/queue.cpp

@@ -1251,7 +1251,7 @@
     }
 }
 
-void* srt::CRcvQueue::worker(void* param)
+void* srt::CRcvQueue::worker(void* param) ATR_NOEXCEPT
 {
     CRcvQueue*   self = (CRcvQueue*)param;
     sockaddr_any sa(self->getIPversion());
@@ -1328,7 +1328,12 @@
                          << "CChannel reported ERROR DURING TRANSMISSION - IPE. INTERRUPTING worker anyway.");
             }
             cst = CONN_REJECT;
-            break;
+
+            // DO NOT interrupt though - the worker thread must run until all
+            // sockets are removed from the multiplexer. Alternatively you can forcefully
+            // remove all sockets from the receive U list.
+            continue;
+            // break;
         }
         // OTHERWISE: this is an "AGAIN" situation. No data was read, but the process should continue.
 
@@ -1350,10 +1355,7 @@
             {
                 HLOGC(qrlog.Debug,
                       log << CUDTUnited::CONID(u->m_SocketID) << " SOCKET broken, REMOVING FROM RCV QUEUE/MAP.");
-                // the socket must be removed from Hash table first, then RcvUList
-                self->m_pHash->remove(u->m_SocketID);
-                self->m_pRcvUList->remove(u);
-                u->m_pRNode->m_bOnList = false;
+                self->forceRemove(u);
             }
 
             ul = self->m_pRcvUList->m_pUList;
@@ -1842,6 +1844,14 @@
     }
 }
 
+void srt::CRcvQueue::forceRemove(CUDT* u)
+{
+    // the socket must be removed from Hash table first, then RcvUList
+    m_pHash->remove(u->m_SocketID);
+    m_pRcvUList->remove(u);
+    u->m_pRNode->m_bOnList = false;
+}
+
 void srt::CMultiplexer::resetAtFork()
 {
     if (m_pRcvQueue != NULL)

srtcore/queue.h

@@ -517,9 +517,11 @@ class CRcvQueue
 
     int getIPversion() { return m_iIPversion; }
 
+    void forceRemove(CUDT* u);
+
     void stop();
 private:
-    static void*  worker(void* param);
+    static void*  worker(void* param) ATR_NOEXCEPT;
     sync::CThread m_WorkerThread;
     // Subroutines of worker
     EReadStatus    worker_RetrieveUnit(int32_t& id, CUnit*& unit, sockaddr_any& sa);