[Cl] Fix a possible deadlock in extreme copy. Fix xrootd#2619

Author: smithdh

src/XrdCl/XrdClXCpCtx.cc

@@ -37,7 +37,7 @@ XCpCtx::XCpCtx( const std::vector<std::string> &urls, uint64_t blockSize, uint8_
       pUrls( std::deque<std::string>( urls.begin(), urls.end() ) ), pBlockSize( blockSize ),
       pParallelSrc( parallelSrc ), pChunkSize( chunkSize ), pParallelChunks( parallelChunks ),
       pOffset( 0 ), pFileSize( -1 ), pFileSizeCV( 0 ), pDataReceived( 0 ), pDone( false ),
-      pDoneCV( 0 ), pRefCount( 1 )
+      pDoneCV( 0 ), pRefCount( 1 ), pDeleteCV( 0 ), pDelete( false )
 {
   SetFileSize( fileSize );
 }
@@ -65,11 +65,12 @@ bool XCpCtx::GetNextUrl( std::string & url )
 
 XCpSrc* XCpCtx::WeakestLink( XCpSrc *exclude )
 {
-  XrdSysMutexHelper lck( pMtx );
   uint64_t transferRate = -1; // set transferRate to max uint64 value
   XCpSrc *ret = 0;
 
   std::list<XCpSrc*>::iterator itr;
+  XrdSysMutexHelper lck( pMtx );
+
   for( itr = pSources.begin() ; itr != pSources.end() ; ++itr )
   {
     XCpSrc *src = *itr;
@@ -82,7 +83,8 @@ XCpSrc* XCpCtx::WeakestLink( XCpSrc *exclude )
     }
   }
 
-  return ret;
+  if( !ret ) return ret;
+  return ret->Self();
 }
 
 void XCpCtx::PutChunk( PageInfo* chunk )
@@ -104,10 +106,10 @@ std::pair<uint64_t, uint64_t> XCpCtx::GetBlock()
 
 void XCpCtx::SetFileSize( int64_t size )
 {
-  XrdSysMutexHelper lck( pMtx );
+  XrdSysCondVarHelper lckcv( pFileSizeCV );
+  XrdSysMutexHelper lckmtx( pMtx );
   if( pFileSize < 0 && size >= 0 )
   {
-    XrdSysCondVarHelper lck( pFileSizeCV );
     pFileSize = size;
     pFileSizeCV.Broadcast();
 
@@ -125,10 +127,23 @@ XRootDStatus XCpCtx::Initialize()
   {
     XCpSrc *src = new XCpSrc( pChunkSize, pParallelChunks, pFileSize, this );
     pSources.push_back( src );
-    src->Start();
   }
 
-  if( pSources.empty() )
+  auto scpy = pSources;
+  bool ok   = false;
+  for(auto src: scpy) {
+    if( src->Start() )
+    {
+      // src destructor will remove src from pSources
+      src->Delete();
+    }
+    else
+    {
+      ok = true;
+    }
+  }
+
+  if( !ok )
   {
     Log *log = DefaultEnv::GetLog();
     log->Error( UtilityMsg, "Failed to initialize (failed to create new threads)" );
@@ -190,6 +205,8 @@ size_t XCpCtx::GetRunning()
   // count active sources
   size_t nbRunning = 0;
   std::list<XCpSrc*>::iterator itr;
+  XrdSysMutexHelper lck( pMtx );
+
   for( itr = pSources.begin() ; itr != pSources.end() ; ++ itr)
     if( (*itr)->IsRunning() )
       ++nbRunning;

src/XrdCl/XrdClXCpCtx.hh

@@ -56,16 +56,44 @@ class XCpCtx
     XCpCtx( const std::vector<std::string> &urls, uint64_t blockSize, uint8_t parallelSrc, uint64_t chunkSize, uint64_t parallelChunks, int64_t fileSize );
 
     /**
-     * Deletes the instance if the reference counter reached 0.
+     * Decrements the reference count and then waits for it to reach
+     * zero, then deletes the instance. Should only be called once.
      */
     void Delete()
     {
-      XrdSysMutexHelper lck( pMtx );
+      XrdSysMutexHelper lckmtx( pMtx );
       --pRefCount;
       if( !pRefCount )
       {
-        lck.UnLock();
+        lckmtx.UnLock();
         delete this;
+        return;
+      }
+      lckmtx.UnLock();
+
+      XrdSysCondVarHelper lckcv( pDoneCV );
+      pDone = true;
+      pDoneCV.Broadcast();
+      lckcv.UnLock();
+
+      lckcv.Lock( &pDeleteCV );
+      while( !pDelete ) pDeleteCV.Wait();
+      lckcv.UnLock();
+      delete this;
+    }
+
+    /**
+     * Decrements the reference count and signal when we reach 0
+     */
+    void Release()
+    {
+      XrdSysMutexHelper lck( pMtx );
+      --pRefCount;
+      if( !pRefCount )
+      {
+        XrdSysCondVarHelper lckcv( pDeleteCV );
+        pDelete = true;
+        pDeleteCV.Broadcast();
       }
     }
 
@@ -257,6 +285,7 @@ class XCpCtx
     /**
      * File Size conditional variable.
      * (notifies waiters if the file size has been set)
+     * If both required, lock order is pFileSizeCV then pMtx.
      */
     XrdSysCondVar              pFileSizeCV;
 
@@ -298,6 +327,18 @@ class XCpCtx
      * Reference counter
      */
     size_t                     pRefCount;
+
+    /**
+     * A condition variable, signals Delete() in case we had to block
+     * the deleter until all sources had finished.
+     * If both required, lock order is pMtx and then pDeleteCV.
+     */
+    XrdSysCondVar              pDeleteCV;
+
+    /**
+     * Predicate for pDeleteCV
+     */
+    bool                       pDelete;
 };
 
 } /* namespace XrdCl */

src/XrdCl/XrdClXCpSrc.cc

@@ -117,19 +117,17 @@ XCpSrc::XCpSrc( uint32_t chunkSize, uint8_t parallel, int64_t fileSize, XCpCtx *
 XCpSrc::~XCpSrc()
 {
   pCtx->RemoveSrc( this );
-  pCtx->Delete();
+  // we release ctx, it is always Delete() by its creator
+  // not by us.
+  pCtx->Release();
 }
 
-void XCpSrc::Start()
+int XCpSrc::Start()
 {
   pRunning = true;
   int rc = pthread_create( &pThread, 0, Run, this );
-  if( rc )
-  {
-    pRunning = false;
-    pCtx->RemoveSrc( this );
-    pCtx->Delete();
-  }
+  if( rc ) pRunning = false;
+  return rc;
 }
 
 void* XCpSrc::Run( void* arg )
@@ -466,7 +464,19 @@ void XCpSrc::Steal( XCpSrc *src )
 {
   if( !src ) return;
 
-  XrdSysMutexHelper lck1( pMtx ), lck2( src->pMtx );
+  // use the address of the mutex to form an
+  // order for acquiring the locks.
+  XrdSysMutexHelper lck1, lck2;
+  if ( std::less{}(&pMtx, &src->pMtx) )
+  {
+    lck2.Lock( &src->pMtx );
+    lck1.Lock( &pMtx );
+  }
+  else
+  {
+    lck1.Lock( &pMtx );
+    lck2.Lock( &src->pMtx );
+  }
 
   Log *log = DefaultEnv::GetLog();
   std::string myHost = URL( pUrl ).GetHostName(), srcHost = URL( src->pUrl ).GetHostName();
@@ -572,8 +582,10 @@ XRootDStatus XCpSrc::GetWork()
     return XRootDStatus();
   }
 
+  // WeakestLink() increases ref count on wLink, so decrease after
   XCpSrc *wLink = pCtx->WeakestLink( this );
   Steal( wLink );
+  if( wLink ) wLink->Delete();
 
   // if we managed to steal something declare success
   if( pCurrentOffset < pBlkEnd || !pRecovered.empty() ) return XRootDStatus();

src/XrdCl/XrdClXCpSrc.hh

@@ -29,6 +29,8 @@
 #include "XrdCl/XrdClSyncQueue.hh"
 #include "XrdSys/XrdSysPthread.hh"
 
+#include <atomic>
+
 namespace XrdCl
 {
 
@@ -55,7 +57,7 @@ class XCpSrc
     /**
      * Creates new thread with XCpSrc::Run as the start routine.
      */
-    void Start();
+    int Start();
 
     /**
      * Stops the thread.
@@ -87,6 +89,9 @@ class XCpSrc
     XCpSrc* Self()
     {
       XrdSysMutexHelper lck( pMtx );
+      // if Ctx is trying to increase our ref count it is possible
+      // we are already in our destrutor, waiting for RemoveSrc().
+      if( !pRefCount ) return nullptr;
       ++pRefCount;
       return this;
     }
@@ -311,7 +316,7 @@ class XCpSrc
     /**
      * Total number of data transferred from this source.
      */
-    uint64_t                      pDataTransfered;
+    std::atomic<uint64_t>         pDataTransfered;
 
     /**
      * A map of ongoing transfers (the offset is the key,
@@ -348,7 +353,7 @@ class XCpSrc
      * false means the source has been stopped,
      * or failed.
      */
-    bool                          pRunning;
+    std::atomic<bool>             pRunning;
 
     /**
      * The time when we started / restarted  chunks