diff --git a/hipamd/src/CMakeLists.txt b/hipamd/src/CMakeLists.txt index 3dc305725d..ac88ddf5d4 100644 --- a/hipamd/src/CMakeLists.txt +++ b/hipamd/src/CMakeLists.txt @@ -98,6 +98,7 @@ target_sources(amdhip64 PRIVATE hip_device.cpp hip_error.cpp hip_event.cpp + hip_event_ipc.cpp hip_fatbin.cpp hip_global.cpp hip_graph_internal.cpp diff --git a/hipamd/src/hip_event.cpp b/hipamd/src/hip_event.cpp index e31f8f6856..3fa8837fee 100644 --- a/hipamd/src/hip_event.cpp +++ b/hipamd/src/hip_event.cpp @@ -25,44 +25,6 @@ #include #endif -void ipcEventCallback(hipStream_t stream, hipError_t status, void* user_data) -{ - std::atomic *signal = reinterpret_cast*>(user_data); - signal->store(0); - return; -} - -// ================================================================================================ -bool createIpcEventShmemIfNeeded(hip::Event::ihipIpcEvent_t& ipc_evt) { -#if !defined(_MSC_VER) - if (ipc_evt.ipc_shmem_) { - // ipc_shmem_ already created, no need to create it again - return true; - } - char name_template[] = "/tmp/eventXXXXXX"; - int temp_fd = mkstemp(name_template); - - ipc_evt.ipc_name_ = name_template; - ipc_evt.ipc_name_.replace(0, 5, "/hip_"); - if (!amd::Os::MemoryMapFileTruncated(ipc_evt.ipc_name_.c_str(), - const_cast (reinterpret_cast(&(ipc_evt.ipc_shmem_))), - sizeof(hip::ihipIpcEventShmem_t))) { - return false; - } - ipc_evt.ipc_shmem_->owners = 1; - ipc_evt.ipc_shmem_->read_index = -1; - ipc_evt.ipc_shmem_->write_index = 0; - for (uint32_t sig_idx = 0; sig_idx < IPC_SIGNALS_PER_EVENT; ++sig_idx) { - ipc_evt.ipc_shmem_->signal[sig_idx] = 0; - } - - close(temp_fd); - return true; -#else - return false; -#endif -} - namespace hip { bool Event::ready() { @@ -126,7 +88,7 @@ hipError_t Event::elapsedTime(Event& eStop, float& ms) { } amd::ScopedLock stopLock(eStop.lock_); - if (event_ == nullptr || eStop.event_ == nullptr) { + if (event_ == nullptr || eStop.event_ == nullptr) { return hipErrorInvalidHandle; } @@ -145,14 +107,15 @@ hipError_t Event::elapsedTime(Event& eStop, float& ms) { amd::Command* command = new amd::Marker(*event_->command().queue(), kMarkerDisableFlush); command->enqueue(); command->awaitCompletion(); - ms = static_cast(static_cast(command->event().profilingInfo().end_) - time())/1000000.f; + ms = static_cast(static_cast(command->event().profilingInfo().end_) - time()) / + 1000000.f; command->release(); } else { // Note: with direct dispatch eStop.ready() relies on HW event, but CPU status can be delayed. // Hence for now make sure CPU status is updated by calling awaitCompletion(); event_->awaitCompletion(); eStop.event_->awaitCompletion(); - ms = static_cast(eStop.time() - time())/1000000.f; + ms = static_cast(eStop.time() - time()) / 1000000.f; } return hipSuccess; } @@ -167,13 +130,10 @@ int64_t Event::time() const { } hipError_t Event::streamWaitCommand(amd::Command*& command, amd::HostQueue* queue) { - if (flags & hipEventInterprocess) { - command = new amd::Marker(*queue, false); - } else { - amd::Command::EventWaitList eventWaitList; - eventWaitList.push_back(event_); - command = new amd::Marker(*queue, kMarkerDisableFlush, eventWaitList); - } + amd::Command::EventWaitList eventWaitList; + eventWaitList.push_back(event_); + command = new amd::Marker(*queue, kMarkerDisableFlush, eventWaitList); + if (command == NULL) { return hipErrorOutOfMemory; } @@ -181,20 +141,7 @@ hipError_t Event::streamWaitCommand(amd::Command*& command, amd::HostQueue* queu } hipError_t Event::enqueueStreamWaitCommand(hipStream_t stream, amd::Command* command) { - if (flags & hipEventInterprocess) { - auto t{new CallbackData{ipc_evt_.ipc_shmem_->read_index, ipc_evt_.ipc_shmem_}}; - StreamCallback* cbo = new StreamCallback( - stream, reinterpret_cast(WaitThenDecrementSignal), t, command); - if (!command->setCallback(CL_COMPLETE, ihipStreamCallback, cbo)) { - command->release(); - return hipErrorInvalidHandle; - } - command->enqueue(); - command->awaitCompletion(); - return hipSuccess; - } else { - command->enqueue(); - } + command->enqueue(); return hipSuccess; } @@ -205,26 +152,24 @@ hipError_t Event::streamWait(hipStream_t stream, uint flags) { if ((event_ == nullptr) || (event_->command().queue() == queue) || ready()) { return hipSuccess; } - - if (!(this->flags & hipEventInterprocess)) { - if (!event_->notifyCmdQueue()) { - return hipErrorLaunchOutOfResources; - } + if (!event_->notifyCmdQueue()) { + return hipErrorLaunchOutOfResources; } amd::Command* command; - streamWaitCommand(command, queue); - enqueueStreamWaitCommand(stream, command); - if (!(this->flags & hipEventInterprocess)) { - command->release(); + hipError_t status = streamWaitCommand(command, queue); + if (status != hipSuccess) { + return status; } + status = enqueueStreamWaitCommand(stream, command); + if (status != hipSuccess) { + return status; + } + command->release(); return hipSuccess; } hipError_t Event::recordCommand(amd::Command*& command, amd::HostQueue* queue) { - bool recorded = isRecorded(); - if ((flags & hipEventInterprocess) && !recorded) { - command = new amd::Marker(*queue, kMarkerDisableFlush); - } else if (command == nullptr) { + if (command == nullptr) { static constexpr bool kRecordExplicitGpuTs = true; // Always submit a EventMarker. command = new hip::EventMarker(*queue, !kMarkerDisableFlush, kRecordExplicitGpuTs); @@ -233,54 +178,14 @@ hipError_t Event::recordCommand(amd::Command*& command, amd::HostQueue* queue) { } hipError_t Event::enqueueRecordCommand(hipStream_t stream, amd::Command* command, bool record) { - amd::HostQueue* queue = hip::getQueue(stream); - bool recorded = isRecorded(); - if ((flags & hipEventInterprocess) && !recorded) { - amd::Event& tEvent = command->event(); - createIpcEventShmemIfNeeded(ipc_evt_); - int write_index = ipc_evt_.ipc_shmem_->write_index++; - int offset = write_index % IPC_SIGNALS_PER_EVENT; - while (ipc_evt_.ipc_shmem_->signal[offset] != 0) { - amd::Os::sleep(1); - } - // Lock signal. - ipc_evt_.ipc_shmem_->signal[offset] = 1; - ipc_evt_.ipc_shmem_->owners_device_id = deviceId(); - - std::atomic* signal = &ipc_evt_.ipc_shmem_->signal[offset]; - StreamCallback* cbo = new StreamCallback( - stream, reinterpret_cast(ipcEventCallback), signal, command); - if (!tEvent.setCallback(CL_COMPLETE, ihipStreamCallback, cbo)) { - command->release(); - return hipErrorInvalidHandle; - } - command->enqueue(); - tEvent.notifyCmdQueue(); - - // Add the new barrier to stall the stream, until the callback is done - amd::Command::EventWaitList eventWaitList; - eventWaitList.push_back(command); - amd::Command* block_command = new amd::Marker(*queue, !kMarkerDisableFlush, eventWaitList); - if (block_command == nullptr) { - return hipErrorInvalidValue; - } - block_command->enqueue(); - block_command->release(); - - // Update read index to indicate new signal. - int expected = write_index - 1; - while (!ipc_evt_.ipc_shmem_->read_index.compare_exchange_weak(expected, write_index)) { - amd::Os::sleep(1); - } - } else { - command->enqueue(); - if (event_ == &command->event()) return hipSuccess; - if (event_ != nullptr) { - event_->release(); - } - event_ = &command->event(); - recorded_ = record; + command->enqueue(); + if (event_ == &command->event()) return hipSuccess; + if (event_ != nullptr) { + event_->release(); } + event_ = &command->event(); + recorded_ = record; + return hipSuccess; } @@ -297,26 +202,29 @@ hipError_t Event::addMarker(hipStream_t stream, amd::Command* command, bool reco } } // namespace hip - +// ================================================================================================ hipError_t ihipEventCreateWithFlags(hipEvent_t* event, unsigned flags) { if (event == nullptr) { return hipErrorInvalidValue; } #if !defined(_MSC_VER) unsigned supportedFlags = hipEventDefault | hipEventBlockingSync | hipEventDisableTiming | - hipEventReleaseToDevice | hipEventReleaseToSystem | hipEventInterprocess; + hipEventReleaseToDevice | hipEventReleaseToSystem | hipEventInterprocess; #else unsigned supportedFlags = hipEventDefault | hipEventBlockingSync | hipEventDisableTiming | - hipEventReleaseToDevice | hipEventReleaseToSystem; + hipEventReleaseToDevice | hipEventReleaseToSystem; #endif const unsigned releaseFlags = (hipEventReleaseToDevice | hipEventReleaseToSystem); - - const bool illegalFlags = - (flags & ~supportedFlags) || // can't set any unsupported flags. - (flags & releaseFlags) == releaseFlags; // can't set both release flags - + // can't set any unsupported flags. + // can't set both release flags + const bool illegalFlags = (flags & ~supportedFlags) || (flags & releaseFlags) == releaseFlags; if (!illegalFlags) { - hip::Event* e = new hip::Event(flags); + hip::Event* e = nullptr; + if (flags & hipEventInterprocess) { + e = new hip::IPCEvent(); + } else { + e = new hip::Event(flags); + } if (e == nullptr) { return hipErrorOutOfMemory; } @@ -327,24 +235,6 @@ hipError_t ihipEventCreateWithFlags(hipEvent_t* event, unsigned flags) { return hipSuccess; } -hipError_t ihipEventQuery(hipEvent_t event) { - if (event == nullptr) { - return hipErrorInvalidHandle; - } - - hip::Event* e = reinterpret_cast(event); - if ((e->flags & hipEventInterprocess) && (e->ipc_evt_.ipc_shmem_)) { - int prev_read_idx = e->ipc_evt_.ipc_shmem_->read_index; - int offset = (prev_read_idx % IPC_SIGNALS_PER_EVENT); - if (e->ipc_evt_.ipc_shmem_->read_index < prev_read_idx+IPC_SIGNALS_PER_EVENT && e->ipc_evt_.ipc_shmem_->signal[offset] != 0) { - return hipErrorNotReady; - } - return hipSuccess; - } else { - return e->query(); - } -} - hipError_t hipEventCreateWithFlags(hipEvent_t* event, unsigned flags) { HIP_INIT_API(hipEventCreateWithFlags, event, flags); HIP_RETURN(ihipEventCreateWithFlags(event, flags), *event); @@ -363,19 +253,11 @@ hipError_t hipEventDestroy(hipEvent_t event) { } hip::Event* e = reinterpret_cast(event); - if ((e->flags & hipEventInterprocess) && (e->ipc_evt_.ipc_shmem_)) { - int owners = -- e->ipc_evt_.ipc_shmem_->owners; - // Make sure event is synchronized - hipEventSynchronize(event); - if (!amd::Os::MemoryUnmapFile(e->ipc_evt_.ipc_shmem_,sizeof(hip::ihipIpcEventShmem_t))) { - HIP_RETURN(hipErrorInvalidHandle); - } - } delete e; HIP_RETURN(hipSuccess); } -hipError_t hipEventElapsedTime(float *ms, hipEvent_t start, hipEvent_t stop) { +hipError_t hipEventElapsedTime(float* ms, hipEvent_t start, hipEvent_t stop) { HIP_INIT_API(hipEventElapsedTime, ms, start, stop); if (ms == nullptr) { @@ -387,7 +269,7 @@ hipError_t hipEventElapsedTime(float *ms, hipEvent_t start, hipEvent_t stop) { } hip::Event* eStart = reinterpret_cast(start); - hip::Event* eStop = reinterpret_cast(stop); + hip::Event* eStop = reinterpret_cast(stop); if (eStart->deviceId() != eStop->deviceId()) { HIP_RETURN(hipErrorInvalidHandle); @@ -396,7 +278,6 @@ hipError_t hipEventElapsedTime(float *ms, hipEvent_t start, hipEvent_t stop) { HIP_RETURN(eStart->elapsedTime(*eStop, *ms), "Elapsed Time = ", *ms); } -// ================================================================================================ hipError_t hipEventRecord(hipEvent_t event, hipStream_t stream) { HIP_INIT_API(hipEventRecord, event, stream); @@ -417,7 +298,6 @@ hipError_t hipEventRecord(hipEvent_t event, hipStream_t stream) { HIP_RETURN(e->addMarker(stream, nullptr, true)); } -// ================================================================================================ hipError_t hipEventSynchronize(hipEvent_t event) { HIP_INIT_API(hipEventSynchronize, event); @@ -426,82 +306,19 @@ hipError_t hipEventSynchronize(hipEvent_t event) { } hip::Event* e = reinterpret_cast(event); - if ((e->flags & hipEventInterprocess) && (e->ipc_evt_.ipc_shmem_)) { - int prev_read_idx = e->ipc_evt_.ipc_shmem_->read_index; - if (prev_read_idx >= 0) { - int offset = (prev_read_idx % IPC_SIGNALS_PER_EVENT); - while ((e->ipc_evt_.ipc_shmem_->read_index < prev_read_idx + IPC_SIGNALS_PER_EVENT) - && (e->ipc_evt_.ipc_shmem_->signal[offset] != 0)) { - amd::Os::sleep(1); - } - } - HIP_RETURN(hipSuccess); - } else { - HIP_RETURN(e->synchronize()); + HIP_RETURN(e->synchronize()); +} + +hipError_t ihipEventQuery(hipEvent_t event) { + if (event == nullptr) { + return hipErrorInvalidHandle; } + + hip::Event* e = reinterpret_cast(event); + return e->query(); } hipError_t hipEventQuery(hipEvent_t event) { HIP_INIT_API(hipEventQuery, event); HIP_RETURN(ihipEventQuery(event)); } - -hipError_t hipIpcGetEventHandle(hipIpcEventHandle_t* handle, hipEvent_t event) { - HIP_INIT_API(hipIpcGetEventHandle, handle, event); -#if !defined(_MSC_VER) - if (handle == nullptr || event == nullptr) { - HIP_RETURN(hipErrorInvalidValue); - } - hip::Event* e = reinterpret_cast(event); - if (!(e->flags & hipEventInterprocess)) { - HIP_RETURN(hipErrorInvalidConfiguration); - } - if (!createIpcEventShmemIfNeeded(e->ipc_evt_)) { - HIP_RETURN(hipErrorInvalidConfiguration); - } - e->ipc_evt_.ipc_shmem_->owners_device_id = e->deviceId(); - e->ipc_evt_.ipc_shmem_->owners_process_id = getpid(); - ihipIpcEventHandle_t* iHandle = reinterpret_cast(handle); - memset(iHandle->shmem_name, 0, HIP_IPC_HANDLE_SIZE); - e->ipc_evt_.ipc_name_.copy(iHandle->shmem_name, std::string::npos); - HIP_RETURN(hipSuccess); -#else - assert(0 && "Unimplemented"); - HIP_RETURN(hipErrorNotSupported); -#endif -} - -hipError_t hipIpcOpenEventHandle(hipEvent_t* event, hipIpcEventHandle_t handle) { - HIP_INIT_API(NONE, event, handle); -#if !defined(_MSC_VER) - hipError_t hip_err = hipSuccess; - if (event == nullptr) { - HIP_RETURN(hipErrorInvalidValue); - } - hip_err = ihipEventCreateWithFlags(event, hipEventDisableTiming | hipEventInterprocess); - if (hip_err != hipSuccess) { - HIP_RETURN(hip_err); - } - hip::Event* e = reinterpret_cast(*event); - ihipIpcEventHandle_t* iHandle = reinterpret_cast(&handle); - hip::Event::ihipIpcEvent_t& ipc_evt = e->ipc_evt_; - ipc_evt.ipc_name_ = iHandle->shmem_name; - if (!amd::Os::MemoryMapFileTruncated(ipc_evt.ipc_name_.c_str(), - (const void**) &(ipc_evt.ipc_shmem_), sizeof(hip::ihipIpcEventShmem_t))) { - HIP_RETURN(hipErrorInvalidValue); - } - - if (getpid() == ipc_evt.ipc_shmem_->owners_process_id.load()) { - // If this is in the same process, return error. - HIP_RETURN(hipErrorInvalidContext); - } - - ipc_evt.ipc_shmem_->owners += 1; - e->setDeviceId(ipc_evt.ipc_shmem_->owners_device_id.load()); - - HIP_RETURN(hipSuccess); -#else - assert(0 && "Unimplemented"); - HIP_RETURN(hipErrorNotSupported); -#endif -} diff --git a/hipamd/src/hip_event.hpp b/hipamd/src/hip_event.hpp index 7f5e7352fe..f727aff4a8 100644 --- a/hipamd/src/hip_event.hpp +++ b/hipamd/src/hip_event.hpp @@ -24,36 +24,23 @@ #include "hip_internal.hpp" #include "thread/monitor.hpp" + // Internal structure for stream callback handler class StreamCallback { - public: - StreamCallback(hipStream_t stream, hipStreamCallback_t callback, void* userData, - amd::Command* command) - : stream_(stream), callBack_(callback), - userData_(userData), command_(command) { - }; - hipStream_t stream_; - hipStreamCallback_t callBack_; - void* userData_; - amd::Command* command_; + public: + StreamCallback(hipStream_t stream, hipStreamCallback_t callback, void* userData, + amd::Command* command) + : stream_(stream), callBack_(callback), userData_(userData), command_(command){}; + hipStream_t stream_; + hipStreamCallback_t callBack_; + void* userData_; + amd::Command* command_; }; void CL_CALLBACK ihipStreamCallback(cl_event event, cl_int command_exec_status, void* user_data); - namespace hip { -class EventMarker: public amd::Marker { -public: - EventMarker(amd::HostQueue& queue, bool disableFlush, bool markerTs = false) - : amd::Marker(queue, disableFlush) { - profilingInfo_.enabled_ = true; - profilingInfo_.callback_ = nullptr; - profilingInfo_.marker_ts_ = markerTs; - profilingInfo_.clear(); - } -}; - #define IPC_SIGNALS_PER_EVENT 32 typedef struct ihipIpcEventShmem_s { std::atomic owners; @@ -64,6 +51,17 @@ typedef struct ihipIpcEventShmem_s { std::atomic signal[IPC_SIGNALS_PER_EVENT]; } ihipIpcEventShmem_t; +class EventMarker : public amd::Marker { + public: + EventMarker(amd::HostQueue& queue, bool disableFlush, bool markerTs = false) + : amd::Marker(queue, disableFlush) { + profilingInfo_.enabled_ = true; + profilingInfo_.callback_ = nullptr; + profilingInfo_.marker_ts_ = markerTs; + profilingInfo_.clear(); + } +}; + class Event { /// event recorded on stream where capture is active bool onCapture_; @@ -73,31 +71,32 @@ class Event { std::vector nodesPrevToRecorded_; public: - Event(unsigned int flags) : flags(flags), lock_("hipEvent_t", true), - event_(nullptr), recorded_(false) { + Event(unsigned int flags) + : flags(flags), lock_("hipEvent_t", true), event_(nullptr), recorded_(false) { // No need to init event_ here as addMarker does that onCapture_ = false; - device_id_ = hip::getCurrentDevice()->deviceId(); // Created in current device ctx + device_id_ = hip::getCurrentDevice()->deviceId(); // Created in current device ctx } - ~Event() { + virtual ~Event() { if (event_ != nullptr) { event_->release(); } } unsigned int flags; - hipError_t query(); - hipError_t synchronize(); + virtual hipError_t query(); + virtual hipError_t synchronize(); hipError_t elapsedTime(Event& stop, float& ms); - hipError_t streamWaitCommand(amd::Command*& command, amd::HostQueue* queue); - hipError_t enqueueStreamWaitCommand(hipStream_t stream, amd::Command* command); - hipError_t streamWait(hipStream_t stream, uint flags); + virtual hipError_t streamWaitCommand(amd::Command*& command, amd::HostQueue* queue); + virtual hipError_t enqueueStreamWaitCommand(hipStream_t stream, amd::Command* command); + virtual hipError_t streamWait(hipStream_t stream, uint flags); - hipError_t recordCommand(amd::Command*& command, amd::HostQueue* queue); - hipError_t enqueueRecordCommand(hipStream_t stream, amd::Command* command, bool record); + virtual hipError_t recordCommand(amd::Command*& command, amd::HostQueue* queue); + virtual hipError_t enqueueRecordCommand(hipStream_t stream, amd::Command* command, bool record); hipError_t addMarker(hipStream_t stream, amd::Command* command, bool record); + void BindCommand(amd::Command& command, bool record) { amd::ScopedLock lock(lock_); if (event_ != nullptr) { @@ -135,20 +134,14 @@ class Event { void SetNodesPrevToRecorded(std::vector& graphNode) { nodesPrevToRecorded_ = graphNode; } + virtual hipError_t GetHandle(ihipIpcEventHandle_t* handle) { + return hipErrorInvalidConfiguration; + } + virtual hipError_t OpenHandle(ihipIpcEventHandle_t* handle) { + return hipErrorInvalidConfiguration; + } - // IPC Events - struct ihipIpcEvent_t { - std::string ipc_name_; - int ipc_fd_; - ihipIpcEventShmem_t* ipc_shmem_; - ihipIpcEvent_t(): ipc_name_("dummy"), ipc_fd_(0), ipc_shmem_(nullptr) { - } - void setipcname(const char* name) { - ipc_name_ = std::string(name); - } - }; - ihipIpcEvent_t ipc_evt_; -private: + protected: amd::Monitor lock_; amd::HostQueue* stream_; amd::Event* event_; @@ -162,11 +155,48 @@ private: int64_t time() const; }; +class IPCEvent : public Event { + // IPC Events + struct ihipIpcEvent_t { + std::string ipc_name_; + int ipc_fd_; + ihipIpcEventShmem_t* ipc_shmem_; + ihipIpcEvent_t() : ipc_name_("dummy"), ipc_fd_(0), ipc_shmem_(nullptr) {} + void setipcname(const char* name) { ipc_name_ = std::string(name); } + }; + ihipIpcEvent_t ipc_evt_; + + public: + ~IPCEvent() { + if (ipc_evt_.ipc_shmem_) { + int owners = --ipc_evt_.ipc_shmem_->owners; + // Make sure event is synchronized + hipError_t status = synchronize(); + if (!amd::Os::MemoryUnmapFile(ipc_evt_.ipc_shmem_, sizeof(hip::ihipIpcEventShmem_t))) { + // print hipErrorInvalidHandle; + } + } + } + IPCEvent() : Event(hipEventInterprocess) {} + bool createIpcEventShmemIfNeeded(); + hipError_t GetHandle(ihipIpcEventHandle_t* handle); + hipError_t OpenHandle(ihipIpcEventHandle_t* handle); + hipError_t synchronize(); + hipError_t query(); + + hipError_t streamWaitCommand(amd::Command*& command, amd::HostQueue* queue); + hipError_t enqueueStreamWaitCommand(hipStream_t stream, amd::Command* command); + hipError_t streamWait(hipStream_t stream, uint flags); + + hipError_t recordCommand(amd::Command*& command, amd::HostQueue* queue); + hipError_t enqueueRecordCommand(hipStream_t stream, amd::Command* command, bool record); }; +}; // namespace hip + struct CallbackData { int previous_read_index; hip::ihipIpcEventShmem_t* shmem; }; -#endif // HIP_EVEMT_H +#endif // HIP_EVEMT_H diff --git a/hipamd/src/hip_event_ipc.cpp b/hipamd/src/hip_event_ipc.cpp new file mode 100644 index 0000000000..dea4a0e5f4 --- /dev/null +++ b/hipamd/src/hip_event_ipc.cpp @@ -0,0 +1,255 @@ +/* Copyright (c) 2015 - 2021 Advanced Micro Devices, Inc. + + Permission is hereby granted, free of charge, to any person obtaining a copy + of this software and associated documentation files (the "Software"), to deal + in the Software without restriction, including without limitation the rights + to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + copies of the Software, and to permit persons to whom the Software is + furnished to do so, subject to the following conditions: + + The above copyright notice and this permission notice shall be included in + all copies or substantial portions of the Software. + + THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + THE SOFTWARE. */ + +#include + +#include "hip_event.hpp" +#if !defined(_MSC_VER) +#include +#endif + +void ipcEventCallback(hipStream_t stream, hipError_t status, void* user_data) { + std::atomic* signal = reinterpret_cast*>(user_data); + signal->store(0); + return; +} +// ================================================================================================ + +hipError_t ihipEventCreateWithFlags(hipEvent_t* event, unsigned flags); + +namespace hip { + +bool IPCEvent::createIpcEventShmemIfNeeded() { +#if !defined(_MSC_VER) + if (ipc_evt_.ipc_shmem_) { + // ipc_shmem_ already created, no need to create it again + return true; + } + char name_template[] = "/tmp/eventXXXXXX"; + int temp_fd = mkstemp(name_template); + + ipc_evt_.ipc_name_ = name_template; + ipc_evt_.ipc_name_.replace(0, 5, "/hip_"); + if (!amd::Os::MemoryMapFileTruncated( + ipc_evt_.ipc_name_.c_str(), + const_cast(reinterpret_cast(&(ipc_evt_.ipc_shmem_))), + sizeof(hip::ihipIpcEventShmem_t))) { + return false; + } + ipc_evt_.ipc_shmem_->owners = 1; + ipc_evt_.ipc_shmem_->read_index = -1; + ipc_evt_.ipc_shmem_->write_index = 0; + for (uint32_t sig_idx = 0; sig_idx < IPC_SIGNALS_PER_EVENT; ++sig_idx) { + ipc_evt_.ipc_shmem_->signal[sig_idx] = 0; + } + + close(temp_fd); + return true; +#else + return false; +#endif +} + +hipError_t IPCEvent::query() { + if (ipc_evt_.ipc_shmem_) { + int prev_read_idx = ipc_evt_.ipc_shmem_->read_index; + int offset = (prev_read_idx % IPC_SIGNALS_PER_EVENT); + if (ipc_evt_.ipc_shmem_->read_index < prev_read_idx + IPC_SIGNALS_PER_EVENT && + ipc_evt_.ipc_shmem_->signal[offset] != 0) { + return hipErrorNotReady; + } + } + return hipSuccess; +} + +hipError_t IPCEvent::synchronize() { + if (ipc_evt_.ipc_shmem_) { + int prev_read_idx = ipc_evt_.ipc_shmem_->read_index; + if (prev_read_idx >= 0) { + int offset = (prev_read_idx % IPC_SIGNALS_PER_EVENT); + while ((ipc_evt_.ipc_shmem_->read_index < prev_read_idx + IPC_SIGNALS_PER_EVENT) && + (ipc_evt_.ipc_shmem_->signal[offset] != 0)) { + amd::Os::sleep(1); + } + } + } + return hipSuccess; +} + +hipError_t IPCEvent::streamWaitCommand(amd::Command*& command, amd::HostQueue* queue) { + command = new amd::Marker(*queue, false); + if (command == NULL) { + return hipErrorOutOfMemory; + } + return hipSuccess; +} + +hipError_t IPCEvent::enqueueStreamWaitCommand(hipStream_t stream, amd::Command* command) { + auto t{new CallbackData{ipc_evt_.ipc_shmem_->read_index, ipc_evt_.ipc_shmem_}}; + StreamCallback* cbo = new StreamCallback( + stream, reinterpret_cast(WaitThenDecrementSignal), t, command); + if (!command->setCallback(CL_COMPLETE, ihipStreamCallback, cbo)) { + command->release(); + return hipErrorInvalidHandle; + } + command->enqueue(); + command->awaitCompletion(); + return hipSuccess; +} + +hipError_t IPCEvent::streamWait(hipStream_t stream, uint flags) { + amd::HostQueue* queue = hip::getQueue(stream); + // Access to event_ object must be lock protected + amd::ScopedLock lock(lock_); + if ((event_ == nullptr) || (event_->command().queue() == queue) || ready()) { + return hipSuccess; + } + amd::Command* command; + hipError_t status = streamWaitCommand(command, queue); + if (status != hipSuccess) { + return status; + } + status = enqueueStreamWaitCommand(stream, command); + return status; +} + +hipError_t IPCEvent::recordCommand(amd::Command*& command, amd::HostQueue* queue) { + bool recorded = isRecorded(); + if (!recorded) { + command = new amd::Marker(*queue, kMarkerDisableFlush); + } else { + return Event::recordCommand(command, queue); + } + return hipSuccess; +} + +hipError_t IPCEvent::enqueueRecordCommand(hipStream_t stream, amd::Command* command, bool record) { + amd::HostQueue* queue = hip::getQueue(stream); + bool recorded = isRecorded(); + if (!recorded) { + amd::Event& tEvent = command->event(); + createIpcEventShmemIfNeeded(); + int write_index = ipc_evt_.ipc_shmem_->write_index++; + int offset = write_index % IPC_SIGNALS_PER_EVENT; + while (ipc_evt_.ipc_shmem_->signal[offset] != 0) { + amd::Os::sleep(1); + } + // Lock signal. + ipc_evt_.ipc_shmem_->signal[offset] = 1; + ipc_evt_.ipc_shmem_->owners_device_id = deviceId(); + + std::atomic* signal = &ipc_evt_.ipc_shmem_->signal[offset]; + StreamCallback* cbo = new StreamCallback( + stream, reinterpret_cast(ipcEventCallback), signal, command); + if (!tEvent.setCallback(CL_COMPLETE, ihipStreamCallback, cbo)) { + command->release(); + return hipErrorInvalidHandle; + } + command->enqueue(); + tEvent.notifyCmdQueue(); + + // Add the new barrier to stall the stream, until the callback is done + amd::Command::EventWaitList eventWaitList; + eventWaitList.push_back(command); + amd::Command* block_command = new amd::Marker(*queue, !kMarkerDisableFlush, eventWaitList); + if (block_command == nullptr) { + return hipErrorInvalidValue; + } + block_command->enqueue(); + block_command->release(); + + // Update read index to indicate new signal. + int expected = write_index - 1; + while (!ipc_evt_.ipc_shmem_->read_index.compare_exchange_weak(expected, write_index)) { + amd::Os::sleep(1); + } + } else { + return Event::enqueueRecordCommand(stream, command, record); + } + return hipSuccess; +} + +hipError_t IPCEvent::GetHandle(ihipIpcEventHandle_t* handle) { + if (!createIpcEventShmemIfNeeded()) { + return hipErrorInvalidConfiguration; + } + ipc_evt_.ipc_shmem_->owners_device_id = deviceId(); + ipc_evt_.ipc_shmem_->owners_process_id = getpid(); + memset(handle->shmem_name, 0, HIP_IPC_HANDLE_SIZE); + ipc_evt_.ipc_name_.copy(handle->shmem_name, std::string::npos); + return hipSuccess; +} + +hipError_t IPCEvent::OpenHandle(ihipIpcEventHandle_t* handle) { + ipc_evt_.ipc_name_ = handle->shmem_name; + if (!amd::Os::MemoryMapFileTruncated(ipc_evt_.ipc_name_.c_str(), + (const void**)&(ipc_evt_.ipc_shmem_), + sizeof(ihipIpcEventShmem_t))) { + return hipErrorInvalidValue; + } + + if (getpid() == ipc_evt_.ipc_shmem_->owners_process_id.load()) { + // If this is in the same process, return error. + return hipErrorInvalidContext; + } + + ipc_evt_.ipc_shmem_->owners += 1; + setDeviceId(ipc_evt_.ipc_shmem_->owners_device_id.load()); + + return hipSuccess; +} + +} // namespace hip + +// ================================================================================================ + +hipError_t hipIpcGetEventHandle(hipIpcEventHandle_t* handle, hipEvent_t event) { + HIP_INIT_API(hipIpcGetEventHandle, handle, event); +#if !defined(_MSC_VER) + if (handle == nullptr || event == nullptr) { + HIP_RETURN(hipErrorInvalidValue); + } + hip::Event* e = reinterpret_cast(event); + HIP_RETURN(e->GetHandle(reinterpret_cast(handle))); +#else + assert(0 && "Unimplemented"); + HIP_RETURN(hipErrorNotSupported); +#endif +} + +hipError_t hipIpcOpenEventHandle(hipEvent_t* event, hipIpcEventHandle_t handle) { + HIP_INIT_API(NONE, event, handle); +#if !defined(_MSC_VER) + hipError_t hip_err = hipSuccess; + if (event == nullptr) { + HIP_RETURN(hipErrorInvalidValue); + } + hip_err = ihipEventCreateWithFlags(event, hipEventDisableTiming | hipEventInterprocess); + if (hip_err != hipSuccess) { + HIP_RETURN(hip_err); + } + hip::Event* e = reinterpret_cast(*event); + ihipIpcEventHandle_t* iHandle = reinterpret_cast(&handle); + HIP_RETURN(e->OpenHandle(iHandle)); +#else + assert(0 && "Unimplemented"); + HIP_RETURN(hipErrorNotSupported); +#endif +}