GpuInterrupt.cpp

Go to the documentation of this file.

/*
 * Copyright 2017 MapD Technologies, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "DynamicWatchdog.h"
#include "Execute.h"

void Executor::registerActiveModule(void* module, const int device_id) const {
#ifdef HAVE_CUDA
  std::lock_guard<std::mutex> lock(gpu_active_modules_mutex_);
  CHECK_LT(device_id, max_gpu_count);
  gpu_active_modules_device_mask_ |= (1 << device_id);
  gpu_active_modules_[device_id] = module;
  VLOG(1) << "Executor " << this << ", mask 0x" << std::hex
          << gpu_active_modules_device_mask_ << ": Registered module " << module
          << " on device " << std::to_string(device_id);
#endif
}

void Executor::unregisterActiveModule(void* module, const int device_id) const {
#ifdef HAVE_CUDA
  std::lock_guard<std::mutex> lock(gpu_active_modules_mutex_);
  CHECK_LT(device_id, max_gpu_count);
  if ((gpu_active_modules_device_mask_ & (1 << device_id)) == 0) {
    return;
  }
  CHECK_EQ(gpu_active_modules_[device_id], module);
  gpu_active_modules_device_mask_ ^= (1 << device_id);
  VLOG(1) << "Executor " << this << ", mask 0x" << std::hex
          << gpu_active_modules_device_mask_ << ": Unregistered module " << module
          << " on device " << std::to_string(device_id);
#endif
}

void Executor::interrupt(const std::string& query_session,
                         const std::string& interrupt_session) {
  VLOG(1) << "Receive INTERRUPT request on the Executor " << this;
  interrupted_.store(true);
  if (g_enable_runtime_query_interrupt) {
    // We first check the query session as interrupted first.
    mapd_unique_lock<mapd_shared_mutex> session_write_lock(executor_session_mutex_);
    this->setQuerySessionAsInterrupted(query_session, session_write_lock);
    session_write_lock.unlock();
    // We have to cover interrupt request from *any* session because we don't know
    // whether the request is for the running query or pending query.
    // But pending query hangs on the executor until the running query is finished
    // to get the computing resources to execute the query
    // So, we just need to kill the pending query on the executor.
    mapd_shared_lock<mapd_shared_mutex> session_read_lock(executor_session_mutex_);
    bool isRunningSession =
        this->checkCurrentQuerySession(query_session, session_read_lock);
    session_read_lock.unlock();
    if (!isRunningSession) {
      interrupted_.store(false);
      return;
    }
  }

  bool CPU_execution_mode = true;

#ifdef HAVE_CUDA
  // The below code is basically for runtime query interrupt for GPU.
  // It is also possible that user forces to use CPU-mode even if the user has GPU(s).
  // In this case, we should not execute the code in below to avoid runtime failure
  CHECK_GE(Catalog_Namespace::SysCatalog::instance()
               .getDataMgr()
               .getCudaMgr()
               ->getDeviceCount(),
           1);
  std::lock_guard<std::mutex> lock(gpu_active_modules_mutex_);
  VLOG(1) << "Executor " << this << ": Interrupting Active Modules: mask 0x" << std::hex
          << gpu_active_modules_device_mask_;
  CUcontext old_cu_context;
  checkCudaErrors(cuCtxGetCurrent(&old_cu_context));
  for (int device_id = 0; device_id < max_gpu_count; device_id++) {
    if (gpu_active_modules_device_mask_ & (1 << device_id)) {
      void* module = gpu_active_modules_[device_id];
      auto cu_module = static_cast<CUmodule>(module);
      if (!cu_module) {
        continue;
      } else {
        VLOG(1) << "Try to interrupt the running query on GPU";
        CPU_execution_mode = false;
      }
      VLOG(1) << "Executor " << this << ": Interrupting Active Modules: mask 0x"
              << std::hex << gpu_active_modules_device_mask_ << " on device "
              << std::to_string(device_id);

      if (catalog_) {
        catalog_->getDataMgr().getCudaMgr()->setContext(device_id);
      } else {
        Catalog_Namespace::SysCatalog::instance().getDataMgr().getCudaMgr()->setContext(
            device_id);
      }

      // Create high priority non-blocking communication stream
      CUstream cu_stream1;
      checkCudaErrors(cuStreamCreateWithPriority(&cu_stream1, CU_STREAM_NON_BLOCKING, 1));

      CUevent start, stop;
      cuEventCreate(&start, 0);
      cuEventCreate(&stop, 0);
      cuEventRecord(start, cu_stream1);

      if (g_enable_dynamic_watchdog) {
        CUdeviceptr dw_abort;
        size_t dw_abort_size;
        if (cuModuleGetGlobal(&dw_abort, &dw_abort_size, cu_module, "dw_abort") ==
            CUDA_SUCCESS) {
          CHECK_EQ(dw_abort_size, sizeof(uint32_t));
          int32_t abort_val = 1;
          checkCudaErrors(cuMemcpyHtoDAsync(dw_abort,
                                            reinterpret_cast<void*>(&abort_val),
                                            sizeof(int32_t),
                                            cu_stream1));

          if (device_id == 0) {
            LOG(INFO) << "GPU: Async Abort submitted to Device "
                      << std::to_string(device_id);
          }
        }
      }

      if (g_enable_runtime_query_interrupt) {
        CUdeviceptr runtime_interrupt_flag;
        size_t runtime_interrupt_flag_size;
        auto status = cuModuleGetGlobal(&runtime_interrupt_flag,
                                        &runtime_interrupt_flag_size,
                                        cu_module,
                                        "runtime_interrupt_flag");
        if (status == CUDA_SUCCESS) {
          VLOG(1) << "Interrupt on GPU status: CUDA_SUCCESS";
          CHECK_EQ(runtime_interrupt_flag_size, sizeof(uint32_t));
          int32_t abort_val = 1;
          checkCudaErrors(cuMemcpyHtoDAsync(runtime_interrupt_flag,
                                            reinterpret_cast<void*>(&abort_val),
                                            sizeof(int32_t),
                                            cu_stream1));
          if (device_id == 0) {
            LOG(INFO) << "GPU: Async Abort submitted to Device "
                      << std::to_string(device_id);
          }
        } else if (status == CUDA_ERROR_NOT_FOUND) {
          std::runtime_error(
              "Runtime query interruption is failed: "
              "a interrupt flag on GPU does not be initialized.");
        } else {
          // if we reach here, query runtime interrupt is failed due to
          // one of the following error: CUDA_ERROR_NOT_INITIALIZED,
          // CUDA_ERROR_DEINITIALIZED. CUDA_ERROR_INVALID_CONTEXT, and
          // CUDA_ERROR_INVALID_VALUE. All those error codes are due to device failure.
          std::runtime_error("Runtime interrupt is failed due to device-related issue");
        }

        cuEventRecord(stop, cu_stream1);
        cuEventSynchronize(stop);
        float milliseconds = 0;
        cuEventElapsedTime(&milliseconds, start, stop);
        VLOG(1) << "Device " << std::to_string(device_id)
                << ": submitted async request to abort SUCCESS: "
                << std::to_string(milliseconds) << " ms\n";
        checkCudaErrors(cuStreamDestroy(cu_stream1));
      }
    }
    checkCudaErrors(cuCtxSetCurrent(old_cu_context));
  }
#endif
  if (g_enable_dynamic_watchdog) {
    dynamic_watchdog_init(static_cast<unsigned>(DW_ABORT));
  }

  if (g_enable_runtime_query_interrupt && CPU_execution_mode) {
    // turn interrupt flag on for CPU mode
    VLOG(1) << "Try to interrupt the running query on CPU";
    check_interrupt_init(static_cast<unsigned>(INT_ABORT));
  }
}

void Executor::resetInterrupt() {
#ifdef HAVE_CUDA
  std::lock_guard<std::mutex> lock(gpu_active_modules_mutex_);
#endif

  if (g_enable_dynamic_watchdog) {
    dynamic_watchdog_init(static_cast<unsigned>(DW_RESET));
  } else if (g_enable_runtime_query_interrupt) {
    check_interrupt_init(static_cast<unsigned>(INT_RESET));
  }

  if (g_cluster) {
    bool sessionLeft = false;
    mapd_shared_lock<mapd_shared_mutex> session_read_lock(executor_session_mutex_);
    std::string curSession = getCurrentQuerySession(session_read_lock);
    sessionLeft = checkIsQuerySessionInterrupted(curSession, session_read_lock);
    session_read_lock.unlock();
    if (curSession != "" || sessionLeft) {
      mapd_unique_lock<mapd_shared_mutex> session_write_lock(executor_session_mutex_);
      removeFromQuerySessionList(curSession, session_write_lock);
      invalidateQuerySession(session_write_lock);
      session_write_lock.unlock();
    }
  }

  if (interrupted_.load()) {
    interrupted_.store(false);
  }
  VLOG(1) << "RESET Executor " << this << " that had previously been interrupted";
}