Stream thread-safe checkpoint.

Moving data structures to critical / protected section.

include/hcc_detail/hip_hcc.h

@@ -280,6 +280,12 @@ typedef std::mutex StreamMutex;
 typedef FakeMutex StreamMutex;
 #endif
 
+#if DEVICE_THREAD_SAFE
+typedef std::mutex DeviceMutex;
+#else
+typedef FakeMutex DeviceMutex;
+#warning "Device thread-safe disabled"
+#endif
 
 //
 //---
@@ -322,10 +328,46 @@ struct LockedBase {
 template <typename MUTEX_TYPE> 
 class ihipStreamCriticalBase_t : public LockedBase<MUTEX_TYPE> 
 {
-private:
-};
-typedef ihipStreamCriticalBase_t<StreamMutex> ihipStreamCritical_t;  
+public:
+    ihipStreamCriticalBase_t() :
+        _last_command_type(ihipCommandCopyH2H),
+        _last_copy_signal(NULL),
+        _signalCursor(0),
+        _oldest_live_sig_id(1),
+        _stream_sig_id(0)
+    {
+        _signalPool.resize(HIP_STREAM_SIGNALS > 0 ? HIP_STREAM_SIGNALS : 1);
+    };
 
+    ~ihipStreamCriticalBase_t() {
+        _signalPool.clear();
+    }
+
+    ihipStreamCriticalBase_t<StreamMutex>  * mlock() { LockedBase<MUTEX_TYPE>::lock(); return this;};
+
+
+public:
+    // Critical Data:
+    ihipCommand_t               _last_command_type;  // type of the last command
+
+    // signal of last copy command sent to the stream.
+    // May be NULL, indicating the previous command has completley finished and future commands don't need to create a dependency.
+    // Copy can be either H2D or D2H.
+    ihipSignal_t                *_last_copy_signal;
+
+    hc::completion_future       _last_kernel_future;  // Completion future of last kernel command sent to GPU.
+
+    // Signal pool:
+    int                         _signalCursor;
+    SIGSEQNUM                   _oldest_live_sig_id; // oldest live seq_id, anything < this can be allocated.
+    std::deque<ihipSignal_t>    _signalPool;   // Pool of signals for use by this stream.
+
+
+    SIGSEQNUM                   _stream_sig_id;      // Monotonically increasing unique signal id.
+};
+
+
+typedef ihipStreamCriticalBase_t<StreamMutex> ihipStreamCritical_t;  
 typedef LockedAccessor<ihipStreamCritical_t> Locked_ihipStreamCritical_t;
 
 
@@ -339,7 +381,9 @@ typedef uint64_t SeqNum_t ;
     ~ihipStream_t();
 
     // kind is hipMemcpyKind
-    void copySync (void* dst, const void* src, size_t sizeBytes, unsigned kind);
+    void copySync (Locked_ihipStreamCritical_t &crit, void* dst, const void* src, size_t sizeBytes, unsigned kind);
+    void locked_copySync (void* dst, const void* src, size_t sizeBytes, unsigned kind);
+
     void copyAsync(void* dst, const void* src, size_t sizeBytes, unsigned kind);
 
     //---
@@ -347,57 +391,46 @@ typedef uint64_t SeqNum_t ;
     bool                 preKernelCommand();
     void                 postKernelCommand(hc::completion_future &kernel_future);
 
-    int                  preCopyCommand(ihipSignal_t *lastCopy, hsa_signal_t *waitSignal, ihipCommand_t copyType);
+    int                  preCopyCommand(Locked_ihipStreamCritical_t &crit, ihipSignal_t *lastCopy, hsa_signal_t *waitSignal, ihipCommand_t copyType);
 
-    void                 reclaimSignals_ts(SIGSEQNUM sigNum);
-    void                 wait(bool assertQueueEmpty=false);
+    void                 locked_reclaimSignals(SIGSEQNUM sigNum);
+    void                 locked_wait(bool assertQueueEmpty=false);
+
+    // Use this if we already have the stream critical data mutex:
+    void                 wait(Locked_ihipStreamCritical_t &crit, bool assertQueueEmpty=false);
 
 
+    SIGSEQNUM            locked_lastCopySeqId() {Locked_ihipStreamCritical_t crit(_criticalData); return lastCopySeqId(crit); };
 
-    // Non-threadsafe accessors - must be protected by high-level stream lock:
-    SIGSEQNUM            lastCopySeqId() { return _last_copy_signal ? _last_copy_signal->_sig_id : 0; };
-    ihipSignal_t *              allocSignal();
+    // Non-threadsafe accessors - must be protected by high-level stream lock with accessor passed to function.
+    SIGSEQNUM            lastCopySeqId(Locked_ihipStreamCritical_t &crit) { return crit->_last_copy_signal ? crit->_last_copy_signal->_sig_id : 0; };
+    ihipSignal_t *       allocSignal(Locked_ihipStreamCritical_t &crit);
 
 
     //-- Non-racy accessors:
     // These functions access fields set at initialization time and are non-racy (so do not acquire mutex)
     ihipDevice_t *              getDevice() const;
 
+
+
     //---
     //Member vars - these are set at initialization:
     SeqNum_t                    _id;   // monotonic sequence ID
     hc::accelerator_view        _av;
     unsigned                    _flags;
+
 private:
     ihipStreamCritical_t        _criticalData;
 
 private:
     void                        enqueueBarrier(hsa_queue_t* queue, ihipSignal_t *depSignal);
-    void                 waitCopy(ihipSignal_t *signal);
+    void                        waitCopy(Locked_ihipStreamCritical_t &crit, ihipSignal_t *signal);
 
     // The unsigned return is hipMemcpyKind
     unsigned resolveMemcpyDirection(bool srcInDeviceMem, bool dstInDeviceMem);
     void setCopyAgents(unsigned kind, ihipCommand_t *commandType, hsa_agent_t *srcAgent, hsa_agent_t *dstAgent);
 
-    //---
-
-    unsigned                    _device_index;
-
-    // Critical Data:
-    ihipCommand_t               _last_command_type;  // type of the last command
-
-    // signal of last copy command sent to the stream.
-    // May be NULL, indicating the previous command has completley finished and future commands don't need to create a dependency.
-    // Copy can be either H2D or D2H.
-    ihipSignal_t                *_last_copy_signal;
-    hc::completion_future       _last_kernel_future;  // Completion future of last kernel command sent to GPU.
-
-    int                         _signalCursor;
-
-    SIGSEQNUM                   _stream_sig_id;      // Monotonically increasing unique signal id.
-    SIGSEQNUM                   _oldest_live_sig_id; // oldest live seq_id, anything < this can be allocated.
-    std::deque<ihipSignal_t>    _signalPool;   // Pool of signals for use by this stream.
-
+    unsigned                    _device_index;       // index into the g_device array 
 
     friend std::ostream& operator<<(std::ostream& os, const ihipStream_t& s);
 };
@@ -465,13 +498,10 @@ private:
     ihipStream_t::SeqNum_t   _stream_id;
 };
 
-#if DEVICE_THREAD_SAFE
-typedef ihipDeviceCriticalBase_t<std::mutex> ihipDeviceCritical_t;  // Use real mutex
-#else
-#warning "Device thread-safe disabled"
-typedef ihipDeviceCriticalBase_t<FakeMutex>  ihipDeviceCritical_t; // Fake mutex, for testing
-#endif
+// Note Mutex selected based on DeviceMutex
+typedef ihipDeviceCriticalBase_t<DeviceMutex> ihipDeviceCritical_t;  
 
+// This type is used by functions that need access to the critical device structures.
 typedef LockedAccessor<ihipDeviceCritical_t> Locked_ihipDeviceCritical_t;
 
 

src/hip_event.cpp

@@ -91,7 +91,8 @@ hipError_t hipEventRecord(hipEvent_t event, hipStream_t stream)
             // Clear timestamps
             eh->_timestamp = 0;
             eh->_marker = stream->_av.create_marker();
-            eh->_copy_seq_id = stream->lastCopySeqId();
+            
+            eh->_copy_seq_id = stream->locked_lastCopySeqId();
 
             return ihipLogStatus(hipSuccess);
         }
@@ -139,7 +140,7 @@ hipError_t hipEventSynchronize(hipEvent_t event)
 #else
             eh->_marker.wait();
 #endif
-            eh->_stream->reclaimSignals_ts(eh->_copy_seq_id);
+            eh->_stream->locked_reclaimSignals(eh->_copy_seq_id);
 
             return ihipLogStatus(hipSuccess);
         }

src/hip_hcc.cpp

@@ -128,40 +128,32 @@ ihipStream_t::ihipStream_t(unsigned device_index, hc::accelerator_view av, unsig
     _id(0), // will be set by add function.
     _av(av),
     _flags(flags),
-    _device_index(device_index),
-    _last_command_type(ihipCommandCopyH2D),
-    _last_copy_signal(NULL),
-    _signalCursor(0),
-    _stream_sig_id(0),
-    _oldest_live_sig_id(1)
+    _device_index(device_index)
 {
     tprintf(DB_SYNC, " streamCreate: stream=%p\n", this);
-    _signalPool.resize(HIP_STREAM_SIGNALS > 0 ? HIP_STREAM_SIGNALS : 1);
-
 };
 
 
 //---
 ihipStream_t::~ihipStream_t()
-{
-    _signalPool.clear();
-    // Hack to catch memory issues, in particular accesses to _acc after it has been destroyed.
-    //memset (&_av, 0x0, sizeof(hc::accelerator_view&));
+{ 
 }
 
 
 
 //---
-void ihipStream_t::reclaimSignals_ts(SIGSEQNUM sigNum)
+void ihipStream_t::locked_reclaimSignals(SIGSEQNUM sigNum)
 {
+    Locked_ihipStreamCritical_t crit(_criticalData);
+
     tprintf(DB_SIGNAL, "reclaim signal #%lu\n", sigNum);
-    // Mark all signals older and including this one as available for
-    _oldest_live_sig_id = sigNum+1;
+    // Mark all signals older and including this one as available for re-allocation.
+    crit->_oldest_live_sig_id = sigNum+1;
 }
 
 
 //---
-void ihipStream_t::waitCopy(ihipSignal_t *signal)
+void ihipStream_t::waitCopy(Locked_ihipStreamCritical_t &crit, ihipSignal_t *signal)
 {
     hsa_signal_wait_acquire(signal->_hsa_signal, HSA_SIGNAL_CONDITION_LT, 1, UINT64_MAX, HSA_WAIT_STATE_ACTIVE);
 
@@ -169,32 +161,43 @@ void ihipStream_t::waitCopy(ihipSignal_t *signal)
 
     tprintf(DB_SIGNAL, "waitCopy reclaim signal #%lu\n", sigNum);
     // Mark all signals older and including this one as available for reclaim
-    if (sigNum > _oldest_live_sig_id) {
-        _oldest_live_sig_id = sigNum+1; // TODO, +1 here seems dangerous.
+    if (sigNum > crit->_oldest_live_sig_id) {
+        crit->_oldest_live_sig_id = sigNum+1; // TODO, +1 here seems dangerous.
     }
 
 }
 
-
-//---
 //Wait for all kernel and data copy commands in this stream to complete.
-void ihipStream_t::wait(bool assertQueueEmpty)
+//This signature should be used in routines that already have locked the stream mutex
+void ihipStream_t::wait(Locked_ihipStreamCritical_t &crit, bool assertQueueEmpty)
 {
     if (! assertQueueEmpty) {
         tprintf (DB_SYNC, "stream %p wait for queue-empty..\n", this);
         _av.wait();
     }
-    if (_last_copy_signal) {
-        tprintf (DB_SYNC, "stream %p wait for lastCopy:#%lu...\n", this, _last_copy_signal ? _last_copy_signal->_sig_id: 0x0 );
-        this->waitCopy(_last_copy_signal);
+    if (crit->_last_copy_signal) {
+        tprintf (DB_SYNC, "stream %p wait for lastCopy:#%lu...\n", this, lastCopySeqId(crit) );
+        this->waitCopy(crit, crit->_last_copy_signal);
     }
 
     // Reset the stream to "empty" - next command will not set up an inpute dependency on any older signal.
-    _last_command_type = ihipCommandCopyH2D;
-    _last_copy_signal = NULL;
+    crit->_last_command_type = ihipCommandCopyH2D;
+    crit->_last_copy_signal = NULL;
+}
+
+
+//---
+//Wait for all kernel and data copy commands in this stream to complete.
+void ihipStream_t::locked_wait(bool assertQueueEmpty)
+{
+    Locked_ihipStreamCritical_t crit(_criticalData);
+
+    wait(crit, assertQueueEmpty);
+
 };
 
 
+
 //---
 ihipDevice_t * ihipStream_t::getDevice() const
 {
@@ -210,26 +213,26 @@ ihipDevice_t * ihipStream_t::getDevice() const
 // Allocate a new signal from the signal pool.
 // Returned signals have value of 0.
 // Signals are intended for use in this stream and are always reclaimed "in-order".
-ihipSignal_t *ihipStream_t::allocSignal()
+ihipSignal_t *ihipStream_t::allocSignal(Locked_ihipStreamCritical_t &crit)
 {
-    int numToScan = _signalPool.size();
+    int numToScan = crit->_signalPool.size();
     do {
-        auto thisCursor = _signalCursor;
-        if (++_signalCursor == _signalPool.size()) {
-            _signalCursor = 0;
+        auto thisCursor = crit->_signalCursor;
+        if (++crit->_signalCursor == crit->_signalPool.size()) {
+            crit->_signalCursor = 0;
         }
 
-        if (_signalPool[thisCursor]._sig_id < _oldest_live_sig_id) {
-            SIGSEQNUM oldSigId = _signalPool[thisCursor]._sig_id;
-            _signalPool[thisCursor]._index = thisCursor;
-            _signalPool[thisCursor]._sig_id  =  ++_stream_sig_id;  // allocate it.
+        if (crit->_signalPool[thisCursor]._sig_id < crit->_oldest_live_sig_id) {
+            SIGSEQNUM oldSigId = crit->_signalPool[thisCursor]._sig_id;
+            crit->_signalPool[thisCursor]._index = thisCursor;
+            crit->_signalPool[thisCursor]._sig_id  =  ++crit->_stream_sig_id;  // allocate it.
             tprintf(DB_SIGNAL, "allocatSignal #%lu at pos:%i (old sigId:%lu < oldest_live:%lu)\n",
-                    _signalPool[thisCursor]._sig_id,
-                    thisCursor, oldSigId, _oldest_live_sig_id);
+                    crit->_signalPool[thisCursor]._sig_id,
+                    thisCursor, oldSigId, crit->_oldest_live_sig_id);
 
 
 
-            return &_signalPool[thisCursor];
+            return &crit->_signalPool[thisCursor];
         }
 
     } while (--numToScan) ;
@@ -237,13 +240,13 @@ ihipSignal_t *ihipStream_t::allocSignal()
     assert(numToScan == 0);
 
     // Have to grow the pool:
-    _signalCursor = _signalPool.size(); // set to the beginning of the new entries:
-    if (_signalCursor > 10000) {
-        fprintf (stderr, "warning: signal pool size=%d, may indicate runaway number of inflight commands\n", _signalCursor);
+    crit->_signalCursor = crit->_signalPool.size(); // set to the beginning of the new entries:
+    if (crit->_signalCursor > 10000) {
+        fprintf (stderr, "warning: signal pool size=%d, may indicate runaway number of inflight commands\n", crit->_signalCursor);
     }
-    _signalPool.resize(_signalPool.size() * 2);
-    tprintf (DB_SIGNAL, "grow signal pool to %zu entries, cursor=%d\n", _signalPool.size(), _signalCursor);
-    return allocSignal();  // try again,
+    crit->_signalPool.resize(crit->_signalPool.size() * 2);
+    tprintf (DB_SIGNAL, "grow signal pool to %zu entries, cursor=%d\n", crit->_signalPool.size(), crit->_signalCursor);
+    return allocSignal(crit);  // try again,
 
     // Should never reach here.
     assert(0);
@@ -288,30 +291,31 @@ void ihipStream_t::enqueueBarrier(hsa_queue_t* queue, ihipSignal_t *depSignal)
 //
 bool ihipStream_t::preKernelCommand()
 {
-    _criticalData.lock();// will be unlocked in postKernelCommand
+    ihipStreamCritical_t *critData =_criticalData.mlock();// will be unlocked in postKernelCommand
 
     bool addedSync = false;
     // If switching command types, we need to add a barrier packet to synchronize things.
-    if (_last_command_type != ihipCommandKernel) {
-        if (_last_copy_signal) {
+    if (critData->_last_command_type != ihipCommandKernel) {
+        if (critData->_last_copy_signal) {
             addedSync = true;
 
             hsa_queue_t * q =  (hsa_queue_t*)_av.get_hsa_queue();
             if (HIP_DISABLE_HW_KERNEL_DEP == 0) {
-                this->enqueueBarrier(q, _last_copy_signal);
+                this->enqueueBarrier(q, critData->_last_copy_signal);
                 tprintf (DB_SYNC, "stream %p switch %s to %s (barrier pkt inserted with wait on #%lu)\n",
-                        this, ihipCommandName[_last_command_type], ihipCommandName[ihipCommandKernel], _last_copy_signal->_sig_id)
+                        this, ihipCommandName[critData->_last_command_type], ihipCommandName[ihipCommandKernel], critData->_last_copy_signal->_sig_id)
 
             } else if (HIP_DISABLE_HW_KERNEL_DEP>0) {
                     tprintf (DB_SYNC, "stream %p switch %s to %s (HOST wait for previous...)\n",
-                            this, ihipCommandName[_last_command_type], ihipCommandName[ihipCommandKernel]);
-                    this->waitCopy(_last_copy_signal);
+                            this, ihipCommandName[critData->_last_command_type], ihipCommandName[ihipCommandKernel]);
+                    assert(0); // Fix/enable next line.  TODO
+                    //this->waitCopy(critData, critData->_last_copy_signal);
             } else if (HIP_DISABLE_HW_KERNEL_DEP==-1) {
                 tprintf (DB_SYNC, "stream %p switch %s to %s (IGNORE dependency)\n",
-                        this, ihipCommandName[_last_command_type], ihipCommandName[ihipCommandKernel]);
+                        this, ihipCommandName[critData->_last_command_type], ihipCommandName[ihipCommandKernel]);
             }
         }
-        _last_command_type = ihipCommandKernel;
+        critData->_last_command_type = ihipCommandKernel;
     }
 
     return addedSync;
@@ -321,7 +325,8 @@ bool ihipStream_t::preKernelCommand()
 //---
 void ihipStream_t::postKernelCommand(hc::completion_future &kernelFuture)
 {
-    _last_kernel_future = kernelFuture;
+    // We locked _criticalData in the preKernelCommand() so OK to access here:
+    _criticalData._last_kernel_future = kernelFuture;
 
     _criticalData.unlock(); // paired with lock from preKernelCommand
 };
@@ -331,7 +336,7 @@ void ihipStream_t::postKernelCommand(hc::completion_future &kernelFuture)
 //---
 // Called whenever a copy command is set to the stream.
 // Examines the last command sent to this stream and returns a signal to wait on, if required.
-int ihipStream_t::preCopyCommand(ihipSignal_t *lastCopy, hsa_signal_t *waitSignal, ihipCommand_t copyType)
+int ihipStream_t::preCopyCommand(Locked_ihipStreamCritical_t &crit, ihipSignal_t *lastCopy, hsa_signal_t *waitSignal, ihipCommand_t copyType)
 {
     int needSync = 0;
 
@@ -340,24 +345,24 @@ int ihipStream_t::preCopyCommand(ihipSignal_t *lastCopy, hsa_signal_t *waitSigna
     //_mutex.lock(); // will be unlocked in postCopyCommand
 
     // If switching command types, we need to add a barrier packet to synchronize things.
-    if (FORCE_SAMEDIR_COPY_DEP || (_last_command_type != copyType)) {
+    if (FORCE_SAMEDIR_COPY_DEP || (crit->_last_command_type != copyType)) {
 
 
-        if (_last_command_type == ihipCommandKernel) {
+        if (crit->_last_command_type == ihipCommandKernel) {
             tprintf (DB_SYNC, "stream %p switch %s to %s (async copy dep on prev kernel)\n",
-                    this, ihipCommandName[_last_command_type], ihipCommandName[copyType]);
+                    this, ihipCommandName[crit->_last_command_type], ihipCommandName[copyType]);
             needSync = 1;
-            hsa_signal_t *hsaSignal = (static_cast<hsa_signal_t*> (_last_kernel_future.get_native_handle()));
+            hsa_signal_t *hsaSignal = (static_cast<hsa_signal_t*> (crit->_last_kernel_future.get_native_handle()));
             if (hsaSignal) {
                 *waitSignal = * hsaSignal;
             } else {
                 assert(0); // if NULL signal, and we return 1, hsa_amd_memory_copy_async will fail.  Confirm this never happens.
             }
-        } else if (_last_copy_signal) {
+        } else if (crit->_last_copy_signal) {
             needSync = 1;
             tprintf (DB_SYNC, "stream %p switch %s to %s (async copy dep on other copy #%lu)\n",
-                    this, ihipCommandName[_last_command_type], ihipCommandName[copyType], _last_copy_signal->_sig_id);
-            *waitSignal = _last_copy_signal->_hsa_signal;
+                    this, ihipCommandName[crit->_last_command_type], ihipCommandName[copyType], crit->_last_copy_signal->_sig_id);
+            *waitSignal = crit->_last_copy_signal->_hsa_signal;
         }
 
         if (HIP_DISABLE_HW_COPY_DEP && needSync) {
@@ -372,10 +377,10 @@ int ihipStream_t::preCopyCommand(ihipSignal_t *lastCopy, hsa_signal_t *waitSigna
             }
         }
 
-        _last_command_type = copyType;
+        crit->_last_command_type = copyType;
     }
 
-    _last_copy_signal = lastCopy;
+    crit->_last_copy_signal = lastCopy;
 
     return needSync;
 }
@@ -694,7 +699,7 @@ void ihipDevice_t::locked_syncDefaultStream(bool waitOnSelf)
             if (waitOnSelf || (stream != _default_stream)) {
                 // TODO-hcc - use blocking or active wait here?
                 // TODO-sync - cudaDeviceBlockingSync
-                stream->wait();
+                stream->locked_wait();
             }
         }
     }
@@ -726,7 +731,7 @@ void ihipDevice_t::locked_waitAllStreams()
 
     tprintf(DB_SYNC, "waitAllStream\n");
     for (auto streamI=l->const_streams().begin(); streamI!=l->const_streams().end(); streamI++) {
-        (*streamI)->wait();
+        (*streamI)->locked_wait();
     }
 }
 
@@ -969,7 +974,7 @@ hipStream_t ihipSyncAndResolveStream(hipStream_t stream)
         // Have to wait for legacy default stream to be empty:
         if (!(stream->_flags & hipStreamNonBlocking))  {
             tprintf(DB_SYNC, "stream %p wait default stream\n", stream);
-            stream->getDevice()->_default_stream->wait();
+            stream->getDevice()->_default_stream->locked_wait();
         }
 
         return stream;
@@ -1101,7 +1106,7 @@ void ihipStream_t::setCopyAgents(unsigned kind, ihipCommand_t *commandType, hsa_
 }
 
 
-void ihipStream_t::copySync(void* dst, const void* src, size_t sizeBytes, unsigned kind)
+void ihipStream_t::copySync(Locked_ihipStreamCritical_t &crit, void* dst, const void* src, size_t sizeBytes, unsigned kind)
 {
     ihipDevice_t *device = this->getDevice();
 
@@ -1127,7 +1132,7 @@ void ihipStream_t::copySync(void* dst, const void* src, size_t sizeBytes, unsign
     hsa_signal_t depSignal;
 
     if ((kind == hipMemcpyHostToDevice) && (!srcTracked)) {
-        int depSignalCnt = preCopyCommand(NULL, &depSignal, ihipCommandCopyH2D);
+        int depSignalCnt = preCopyCommand(crit, NULL, &depSignal, ihipCommandCopyH2D);
         if (HIP_STAGING_BUFFERS) {
             tprintf(DB_COPY1, "D2H && !dstTracked: staged copy H2D dst=%p src=%p sz=%zu\n", dst, src, sizeBytes);
 
@@ -1138,7 +1143,7 @@ void ihipStream_t::copySync(void* dst, const void* src, size_t sizeBytes, unsign
             }
 
             // The copy waits for inputs and then completes before returning so can reset queue to empty:
-            this->wait(true);
+            this->wait(crit, true);
         } else {
             // TODO - remove, slow path.
             tprintf(DB_COPY1, "H2D && ! srcTracked: am_copy dst=%p src=%p sz=%zu\n", dst, src, sizeBytes);
@@ -1149,14 +1154,14 @@ void ihipStream_t::copySync(void* dst, const void* src, size_t sizeBytes, unsign
 #endif
         }
     } else if ((kind == hipMemcpyDeviceToHost) && (!dstTracked)) {
-        int depSignalCnt = preCopyCommand(NULL, &depSignal, ihipCommandCopyD2H);
+        int depSignalCnt = preCopyCommand(crit, NULL, &depSignal, ihipCommandCopyD2H);
         if (HIP_STAGING_BUFFERS) {
             tprintf(DB_COPY1, "D2H && !dstTracked: staged copy D2H dst=%p src=%p sz=%zu\n", dst, src, sizeBytes);
             //printf ("staged-copy- read dep signals\n");
             device->_staging_buffer[1]->CopyDeviceToHost(dst, src, sizeBytes, depSignalCnt ? &depSignal : NULL);
 
-            // The copy waits for inputs and then completes before returning so can reset queue to empty:
-            this->wait(true);
+            // The copy completes before returning so can reset queue to empty:
+            this->wait(crit, true);
 
         } else {
             // TODO - remove, slow path.
@@ -1168,7 +1173,7 @@ void ihipStream_t::copySync(void* dst, const void* src, size_t sizeBytes, unsign
 #endif
         }
     } else if (kind == hipMemcpyHostToHost)  {
-        int depSignalCnt = preCopyCommand(NULL, &depSignal, ihipCommandCopyH2H);
+        int depSignalCnt = preCopyCommand(crit, NULL, &depSignal, ihipCommandCopyH2H);
 
         if (depSignalCnt) {
             // host waits before doing host memory copy.
@@ -1183,10 +1188,10 @@ void ihipStream_t::copySync(void* dst, const void* src, size_t sizeBytes, unsign
         hsa_agent_t srcAgent, dstAgent;
         setCopyAgents(kind, &commandType, &srcAgent, &dstAgent);
 
-        int depSignalCnt = preCopyCommand(NULL, &depSignal, commandType);
+        int depSignalCnt = preCopyCommand(crit, NULL, &depSignal, commandType);
 
         // Get a completion signal:
-        ihipSignal_t *ihipSignal = allocSignal();
+        ihipSignal_t *ihipSignal = allocSignal(crit);
         hsa_signal_t copyCompleteSignal = ihipSignal->_hsa_signal;
 
         hsa_signal_store_relaxed(copyCompleteSignal, 1);
@@ -1197,7 +1202,7 @@ void ihipStream_t::copySync(void* dst, const void* src, size_t sizeBytes, unsign
 
         // This is sync copy, so let's wait for copy right here:
         if (hsa_status == HSA_STATUS_SUCCESS) {
-            waitCopy(ihipSignal); // wait for copy, and return to pool.
+            waitCopy(crit, ihipSignal); // wait for copy, and return to pool.
         } else {
             throw ihipException(hipErrorInvalidValue);
         }
@@ -1205,10 +1210,19 @@ void ihipStream_t::copySync(void* dst, const void* src, size_t sizeBytes, unsign
 }
 
 
+// Sync copy that acquires lock:
+void ihipStream_t::locked_copySync(void* dst, const void* src, size_t sizeBytes, unsigned kind)
+{
+    Locked_ihipStreamCritical_t crit (_criticalData);
+    copySync(crit, dst, src, sizeBytes, kind);
+}
+
 
 
 void ihipStream_t::copyAsync(void* dst, const void* src, size_t sizeBytes, unsigned kind)
 {
+    Locked_ihipStreamCritical_t crit(_criticalData);
+
     ihipDevice_t *device = this->getDevice();
 
     if (device == NULL) {
@@ -1222,7 +1236,7 @@ void ihipStream_t::copyAsync(void* dst, const void* src, size_t sizeBytes, unsig
         /* As this is a CPU op, we need to wait until all
         the commands in current stream are finished.
         */
-        this->wait();
+        this->wait(crit);
 
         memcpy(dst, src, sizeBytes);
 
@@ -1249,8 +1263,7 @@ void ihipStream_t::copyAsync(void* dst, const void* src, size_t sizeBytes, unsig
         }
 
 
-
-        ihipSignal_t *ihip_signal = allocSignal();
+        ihipSignal_t *ihip_signal = allocSignal(crit);
         hsa_signal_store_relaxed(ihip_signal->_hsa_signal, 1);
 
 
@@ -1261,7 +1274,7 @@ void ihipStream_t::copyAsync(void* dst, const void* src, size_t sizeBytes, unsig
             setCopyAgents(kind, &commandType, &srcAgent, &dstAgent);
 
             hsa_signal_t depSignal;
-            int depSignalCnt = preCopyCommand(ihip_signal, &depSignal, commandType);
+            int depSignalCnt = preCopyCommand(crit, ihip_signal, &depSignal, commandType);
 
             tprintf (DB_SYNC, " copy-async, waitFor=%lu completion=#%lu(%lu)\n", depSignalCnt? depSignal.handle:0x0, ihip_signal->_sig_id, ihip_signal->_hsa_signal.handle);
 
@@ -1271,7 +1284,7 @@ void ihipStream_t::copyAsync(void* dst, const void* src, size_t sizeBytes, unsig
             if (hsa_status == HSA_STATUS_SUCCESS) {
                 if (HIP_LAUNCH_BLOCKING) {
                     tprintf(DB_SYNC, "LAUNCH_BLOCKING for completion of hipMemcpyAsync(%zu)\n", sizeBytes);
-                    this->wait();
+                    this->wait(crit);
                 }
             } else {
                 // This path can be hit if src or dst point to unpinned host memory.
@@ -1279,7 +1292,7 @@ void ihipStream_t::copyAsync(void* dst, const void* src, size_t sizeBytes, unsig
                 throw ihipException(hipErrorInvalidValue);
             }
         } else {
-            copySync(dst, src, sizeBytes, kind);
+            copySync(crit, dst, src, sizeBytes, kind);
         }
     }
 }
@@ -1332,6 +1345,7 @@ hipError_t hipHccGetAcceleratorView(hipStream_t stream, hc::accelerator_view **a
 //        Caps convention _ or camelCase
 //        if { }
 //        Should use ihip* data structures inside code rather than app-facing hip.  For example, use ihipDevice_t (rather than hipDevice_t), ihipStream_t (rather than hipStream_t).
+//        locked_
 // TODO - describe MT strategy
 //
 //// TODO - add identifier numbers for streams and devices to help with debugging.

src/hip_memory.cpp

@@ -307,17 +307,14 @@ hipError_t hipMemcpy(void* dst, const void* src, size_t sizeBytes, hipMemcpyKind
     hipError_t e = hipSuccess;
 
     try {
-        stream->copySync(dst, src, sizeBytes, kind);
+
+        stream->locked_copySync(dst, src, sizeBytes, kind);
     }
     catch (ihipException ex) {
         e = ex._code;
     }
 
 
-    if (HIP_LAUNCH_BLOCKING) {
-        tprintf(DB_SYNC, "LAUNCH_BLOCKING for completion of hipMemcpy\n");
-        stream->wait();
-    }
 
     return ihipLogStatus(e);
 }

src/hip_stream.cpp

@@ -82,7 +82,7 @@ hipError_t hipStreamWaitEvent(hipStream_t stream, hipEvent_t event, unsigned int
         // TODO-hcc Convert to use create_blocking_marker(...) functionality.
         // Currently we have a super-conservative version of this - block on host, and drain the queue.
         // This should create a barrier packet in the target queue.
-        stream->wait();
+        stream->locked_wait();
         e = hipSuccess;
     }
 
@@ -101,7 +101,7 @@ hipError_t hipStreamSynchronize(hipStream_t stream)
         ihipDevice_t *device = ihipGetTlsDefaultDevice();
         device->locked_syncDefaultStream(true/*waitOnSelf*/);
     } else {
-        stream->wait();
+        stream->locked_wait();
         e = hipSuccess;
     }
 
@@ -125,7 +125,7 @@ hipError_t hipStreamDestroy(hipStream_t stream)
         ihipDevice_t *device = ihipGetTlsDefaultDevice();
         device->locked_syncDefaultStream(true/*waitOnSelf*/);
     } else {
-        stream->wait();
+        stream->locked_wait();
         e = hipSuccess;
     }