package io.ray.runtime;

import io.ray.api.runtime.RayRuntime;
import io.ray.api.runtime.RayRuntimeFactory;
import io.ray.runtime.config.RayConfig;
import io.ray.runtime.config.RunMode;
import io.ray.runtime.generated.Common.WorkerType;
import io.ray.runtime.util.LoggingUtil;
import io.ray.runtime.util.SystemConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/** The default Ray runtime factory. It produces an instance of RayRuntime. */
public class DefaultRayRuntimeFactory implements RayRuntimeFactory {

  @Override
  public RayRuntime createRayRuntime() {
    RayConfig rayConfig = RayConfig.create();
    LoggingUtil.setupLogging(rayConfig);
    SystemConfig.setup(rayConfig);
    Logger logger = LoggerFactory.getLogger(DefaultRayRuntimeFactory.class);

    if (rayConfig.workerMode == WorkerType.WORKER) {
      // Handle the uncaught exceptions thrown from user-spawned threads.
      Thread.setDefaultUncaughtExceptionHandler(
          (Thread t, Throwable e) -> {
            logger.error(String.format("Uncaught worker exception in thread %s", t), e);
          });
    }

    try {
      logger.debug("Initializing runtime with config: {}", rayConfig);
      AbstractRayRuntime runtime =
          rayConfig.runMode == RunMode.LOCAL
              ? new RayDevRuntime(rayConfig)
              : new RayNativeRuntime(rayConfig);
      runtime.start();
      return runtime;
    } catch (Exception e) {
      logger.error("Failed to initialize ray runtime, with config " + rayConfig, e);
      throw new RuntimeException("Failed to initialize ray runtime", e);
    }
  }
}
