AI/rocm-systems
2
0
Garpu 0
Kode Masalah Tarik Permintaan Actions Packages Projects Rilis Wiki Kegiatan

SWDEV-381402 - Derive hip::Stream from amd::HostQueue

Telusuri Sumber 
Change-Id: I6c1aca5eb350c32d974ae4ffcc725705355956d8
This commit is contained in:
Ioannis Assiouras
2023-02-08 20:18:11 +00:00
orang tua aaf1c19609
melakukan e3633dc8f4
19 mengubah file dengan 403 tambahan dan 732 penghapusan
Download Patch File Download Diff File Expand all files Collapse all files
hipamd/src/hip_stream.cpp
+53 -89
Melihat File
@@ -31,7 +31,8 @@ namespace hip {
// ================================================================================================
Stream::Stream(hip::Device* dev, Priority p, unsigned int f, bool null_stream,
const std::vector<uint32_t>& cuMask, hipStreamCaptureStatus captureStatus)
: queue_(nullptr),
: amd::HostQueue(*dev->asContext(), *dev->devices()[0], 0, amd::CommandQueue::RealTimeDisabled,
convertToQueuePriority(p), cuMask),
lock_("Stream Callback lock"),
device_(dev),
priority_(p),
@@ -40,18 +41,11 @@ Stream::Stream(hip::Device* dev, Priority p, unsigned int f, bool null_stream,
cuMask_(cuMask),
captureStatus_(captureStatus),
originStream_(false),
captureID_(0) {}
// ================================================================================================
Stream::~Stream() {
if (queue_ != nullptr) {
amd::ScopedLock lock(streamSetLock);
streamSet.erase(this);
queue_->release();
queue_ = nullptr;
}
}
captureID_(0)
{
amd::ScopedLock lock(streamSetLock);
streamSet.insert(this);
}
// ================================================================================================
hipError_t Stream::EndCapture() {
@@ -77,38 +71,16 @@ hipError_t Stream::EndCapture() {
// ================================================================================================
bool Stream::Create() {
amd::CommandQueue::Priority p;
switch (priority_) {
case Priority::High:
p = amd::CommandQueue::Priority::High;
break;
case Priority::Low:
p = amd::CommandQueue::Priority::Low;
break;
case Priority::Normal:
default:
p = amd::CommandQueue::Priority::Normal;
break;
}
amd::HostQueue* queue = new amd::HostQueue(*device_->asContext(), *device_->devices()[0],
0, amd::CommandQueue::RealTimeDisabled,
p, cuMask_);
return create();
}
// Create a host queue
bool result = (queue != nullptr) ? queue->create() : false;
// Insert just created stream into the list of the blocking queues
if (result) {
// ================================================================================================
bool Stream::terminate() {
{
amd::ScopedLock lock(streamSetLock);
streamSet.insert(this);
queue_ = queue;
device_->SaveQueue(queue);
} else if (queue != nullptr) {
// Queue creation has failed, and virtual device associated with the queue may not be created.
// Just need to delete the queue instance.
delete queue;
streamSet.erase(this);
}
return result;
return HostQueue::terminate();
}
// ================================================================================================
@@ -130,29 +102,6 @@ bool isValid(hipStream_t& stream) {
return true;
}
// ================================================================================================
amd::HostQueue* Stream::asHostQueue(bool skip_alloc) {
if (queue_ != nullptr) {
return queue_;
}
// Access to the stream object is lock protected, because possible allocation
amd::ScopedLock l(Lock());
if (queue_ == nullptr) {
// Create the host queue for the first time
if (!skip_alloc) {
Create();
}
}
return queue_;
}
// ================================================================================================
void Stream::Finish() const {
if (queue_ != nullptr) {
queue_->finish();
}
}
// ================================================================================================
int Stream::DeviceId() const {
return device_->deviceId();
@@ -176,7 +125,7 @@ void Stream::syncNonBlockingStreams(int deviceId) {
for (auto& it : streamSet) {
if (it->Flags() & hipStreamNonBlocking) {
if (it->DeviceId() == deviceId) {
it->asHostQueue()->finish();
it->finish();
}
}
}
@@ -203,7 +152,7 @@ void Stream::destroyAllStreams(int deviceId) {
}
}
for (auto& it : toBeDeleted) {
delete it;
it->release();
}
}
@@ -211,36 +160,48 @@ bool Stream::StreamCaptureOngoing(void) {
return (g_allCapturingStreams.empty() == true) ? false : true;
}
bool Stream::existsActiveStreamForDevice(hip::Device* device) {
amd::ScopedLock lock(streamSetLock);
for (const auto& active_stream : streamSet) {
if ((active_stream->GetDevice() == device) &&
active_stream->GetQueueStatus()) {
return true;
}
}
return false;
}
};// hip namespace
// ================================================================================================
void iHipWaitActiveStreams(amd::HostQueue* blocking_queue, bool wait_null_stream) {
void iHipWaitActiveStreams(hip::Stream* blocking_stream, bool wait_null_stream) {
amd::Command::EventWaitList eventWaitList(0);
bool submitMarker = 0;
{
amd::ScopedLock lock(streamSetLock);
for (const auto& stream : streamSet) {
amd::HostQueue* active_queue = stream->asHostQueue();
for (const auto& active_stream : streamSet) {
// If it's the current device
if ((&active_queue->device() == &blocking_queue->device()) &&
if ((&active_stream->device() == &blocking_stream->device()) &&
// Make sure it's a default stream
((stream->Flags() & hipStreamNonBlocking) == 0) &&
((active_stream->Flags() & hipStreamNonBlocking) == 0) &&
// and it's not the current stream
(active_queue != blocking_queue) &&
(active_stream != blocking_stream) &&
// check for a wait on the null stream
(stream->Null() == wait_null_stream)) {
(active_stream->Null() == wait_null_stream)) {
// Get the last valid command
amd::Command* command = active_queue->getLastQueuedCommand(true);
amd::Command* command = active_stream->getLastQueuedCommand(true);
if (command != nullptr) {
amd::Event& event = command->event();
// Check HW status of the ROCcrl event.
// Note: not all ROCclr modes support HW status
bool ready = active_queue->device().IsHwEventReady(event);
bool ready = active_stream->device().IsHwEventReady(event);
if (!ready) {
ready = (command->status() == CL_COMPLETE);
}
submitMarker |= active_queue->vdev()->isFenceDirty();
submitMarker |= active_stream->vdev()->isFenceDirty();
// Check the current active status
if (!ready) {
command->notifyCmdQueue();
@@ -259,7 +220,7 @@ void iHipWaitActiveStreams(amd::HostQueue* blocking_queue, bool wait_null_stream
// Check if we have to wait anything
if (eventWaitList.size() > 0 || submitMarker) {
amd::Command* command = new amd::Marker(*blocking_queue, kMarkerDisableFlush, eventWaitList);
amd::Command* command = new amd::Marker(*blocking_stream, kMarkerDisableFlush, eventWaitList);
if (command != nullptr) {
command->enqueue();
command->release();
@@ -288,8 +249,11 @@ static hipError_t ihipStreamCreate(hipStream_t* stream,
}
hip::Stream* hStream = new hip::Stream(hip::getCurrentDevice(), priority, flags, false, cuMask);
if (hStream == nullptr || !hStream->Create()) {
delete hStream;
if (hStream == nullptr) {
return hipErrorOutOfMemory;
}
else if (!hStream->Create()) {
hStream->release();
return hipErrorOutOfMemory;
}
@@ -310,7 +274,7 @@ stream_per_thread::stream_per_thread() {
stream_per_thread::~stream_per_thread() {
for (auto &stream:m_streams) {
if (stream != nullptr && hip::isValid(stream)) {
delete reinterpret_cast<hip::Stream*>(stream);
reinterpret_cast<hip::Stream*>(stream)->release();
stream = nullptr;
}
}
@@ -449,7 +413,7 @@ hipError_t hipStreamSynchronize_common(hipStream_t stream) {
}
}
// Wait for the current host queue
hip::getQueue(stream)->finish();
hip::getStream(stream)->finish();
return hipSuccess;
}
@@ -498,7 +462,7 @@ hipError_t hipStreamDestroy(hipStream_t stream) {
if (l_it != hip::tls.capture_streams_.end()) {
hip::tls.capture_streams_.erase(l_it);
}
delete s;
s->release();
HIP_RETURN(hipSuccess);
}
@@ -564,9 +528,9 @@ hipError_t hipStreamQuery_common(hipStream_t stream) {
HIP_RETURN(hipErrorStreamCaptureUnsupported);
}
}
amd::HostQueue* hostQueue = hip::getQueue(stream);
hip::Stream* hip_stream = hip::getStream(stream);
amd::Command* command = hostQueue->getLastQueuedCommand(true);
amd::Command* command = hip_stream->getLastQueuedCommand(true);
if (command == nullptr) {
// Nothing was submitted to the queue
return hipSuccess;
@@ -604,13 +568,13 @@ hipError_t streamCallback_common(hipStream_t stream, StreamCallback* cbo, void*
return hipErrorContextIsDestroyed;
}
amd::HostQueue* hostQueue = hip::getQueue(stream);
amd::Command* last_command = hostQueue->getLastQueuedCommand(true);
hip::Stream* hip_stream = hip::getStream(stream);
amd::Command* last_command = hip_stream->getLastQueuedCommand(true);
amd::Command::EventWaitList eventWaitList;
if (last_command != nullptr) {
eventWaitList.push_back(last_command);
}
amd::Command* command = new amd::Marker(*hostQueue, !kMarkerDisableFlush, eventWaitList);
amd::Command* command = new amd::Marker(*hip_stream, !kMarkerDisableFlush, eventWaitList);
if (command == nullptr) {
return hipErrorInvalidValue;
}
@@ -630,7 +594,7 @@ hipError_t streamCallback_common(hipStream_t stream, StreamCallback* cbo, void*
// Add the new barrier to stall the stream, until the callback is done
eventWaitList.clear();
eventWaitList.push_back(command);
amd::Command* block_command = new amd::Marker(*hostQueue, !kMarkerDisableFlush, eventWaitList);
amd::Command* block_command = new amd::Marker(*hip_stream, !kMarkerDisableFlush, eventWaitList);
if (block_command == nullptr) {
return hipErrorInvalidValue;
}