<commit_before>/*
 * Copyright 2016 The BigDL Authors.
 *
 * 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.
 */

package com.intel.analytics.bigdl.utils

import java.io.{FileOutputStream, InputStream, PrintWriter}
import java.util.Locale
import java.util.concurrent.atomic.AtomicBoolean

import org.apache.log4j.Logger
import org.apache.spark._
import com.intel.analytics.bigdl.mkl.MKL
import org.apache.spark.utils.SparkUtils
import py4j.GatewayServer

import scala.util.control.{ControlThrowable, NonFatal}

/**
 * define engine type trait
 */
sealed trait EngineType

case object MklBlas extends EngineType


object Engine {
  @deprecated(
    "See https://bigdl-project.github.io/master/#APIGuide/Engine/",
    "0.1.0")
  def init(nExecutor: Int,
           executorCores: Int,
           onSpark: Boolean): Option[SparkConf] = {
    logger.warn("Engine.init(nExecutor, executorCores, onSpark) is deprecated. " +
      "Please refer to " +
      "https://bigdl-project.github.io/master/#APIGuide/Engine/")
    setNodeAndCore(nExecutor, executorCores)
    val res = if (onSpark) {
      require(localMode == false,
        s"Engine.init: bigdl.localMode should not be set while onSpark is " +
          s"true. Please set correct java property.")
      Some(createSparkConf())
    } else {
      require(localMode == true,
        s"Engine.init: bigdl.localMode should be set while onSpark is " +
          s"false. Please set correct java property.")
      None
    }
    res
  }

  /**
   * BigDL need some Spark conf values to be set correctly to have better performance.
   *
   * This method will create a SparkConf, or use the existing one if you provide one.
   * Populate it with correct values.
   *
   * We recommend you use this method instead of setting spark conf values directly. This can
   * make the Spark conf values changes transparent to you. However, if you use spark-shell or
   * Jupyter notebook, as the Spark context is created before your code, you have to
   * set them directly (through command line options or properties file)
   *
   * @return
   */
  def createSparkConf(exisitingConf : SparkConf = null) : SparkConf = {
    var _conf = exisitingConf
    if (_conf == null) {
      _conf = new SparkConf()
    }
    readConf.foreach(c => _conf.set(c._1, c._2))
    _conf
  }

  /**
   * This method should be call before any BigDL procedure and after the Spark context is created.
   *
   * BigDL needs some Spark conf values to be set correctly to have a better performance. There's
   * also multi-thread engines so executor number and core number per executor need to be known
   * to set the parameter of these engines correctly.
   *
   * The method can set parameters of multi-thread engines, verify spark conf values of an
   * existing spark context.
   */
  def init: Unit = this.synchronized {
    if (localMode) {
      logger.info("Detect bigdl.localMode is set. Run workload without spark")
      // The physical core number should have been initialized
      // by java property -Dbigdl.coreNumber=xx
      setNodeAndCore(1, getCoreNumberFromProperty)
    } else {
      logger.info("Auto detect executor number and executor cores number")
      val (nExecutor, executorCores) = sparkExecutorAndCore().get
      logger.info(s"Executor number is $nExecutor and executor cores number is $executorCores")
      setNodeAndCore(nExecutor, executorCores)
      checkSparkContext
    }
  }

  private val logger = Logger.getLogger(getClass)
  private val singletonCounter = new AtomicBoolean()
  private var physicalCoreNumber = -1
  private var nodeNum: Int = -1

  @volatile
  private var gatewayServer: py4j.GatewayServer = null
  private val driverPortFileCreated = new AtomicBoolean()

  private def createGatewayPortFile(port: Int): Unit = {
    val file = new java.io.File(SparkFiles.getRootDirectory(), "gateway_port")
    logger.debug(s"Creating JavaGatewayServer port file" +
      s" on executor-${SparkEnv.get.executorId}:${file.getAbsolutePath}")
    if (file.exists()) {
      file.delete()
    }
    file.createNewFile()
    val out = new PrintWriter(file)
    try {
      out.print(port)
      out.flush()
    } finally {
      out.close()
    }
  }

  private[bigdl] def createJavaGateway(driverPort: Int): Unit = {
    if (SparkUtils.isDriver) {
      if (driverPortFileCreated.compareAndSet(false, true)) {
        try {
          createGatewayPortFile(driverPort)
        } catch {
          case NonFatal(e) =>
            throw new Exception("Could not create java gateway port file", e)
        }
      }
      return
    }<commit_msg>use a seperate java gateway (#2561)<commit_after>/*
 * Copyright 2016 The BigDL Authors.
 *
 * 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.
 */

package com.intel.analytics.bigdl.utils

import java.io.{FileOutputStream, InputStream, PrintWriter}
import java.util.Locale
import java.util.concurrent.atomic.AtomicBoolean

import org.apache.log4j.Logger
import org.apache.spark._
import com.intel.analytics.bigdl.mkl.MKL
import org.apache.spark.utils.SparkUtils
import py4j.GatewayServer

import scala.util.control.{ControlThrowable, NonFatal}

/**
 * define engine type trait
 */
sealed trait EngineType

case object MklBlas extends EngineType


object Engine {
  @deprecated(
    "See https://bigdl-project.github.io/master/#APIGuide/Engine/",
    "0.1.0")
  def init(nExecutor: Int,
           executorCores: Int,
           onSpark: Boolean): Option[SparkConf] = {
    logger.warn("Engine.init(nExecutor, executorCores, onSpark) is deprecated. " +
      "Please refer to " +
      "https://bigdl-project.github.io/master/#APIGuide/Engine/")
    setNodeAndCore(nExecutor, executorCores)
    val res = if (onSpark) {
      require(localMode == false,
        s"Engine.init: bigdl.localMode should not be set while onSpark is " +
          s"true. Please set correct java property.")
      Some(createSparkConf())
    } else {
      require(localMode == true,
        s"Engine.init: bigdl.localMode should be set while onSpark is " +
          s"false. Please set correct java property.")
      None
    }
    res
  }

  /**
   * BigDL need some Spark conf values to be set correctly to have better performance.
   *
   * This method will create a SparkConf, or use the existing one if you provide one.
   * Populate it with correct values.
   *
   * We recommend you use this method instead of setting spark conf values directly. This can
   * make the Spark conf values changes transparent to you. However, if you use spark-shell or
   * Jupyter notebook, as the Spark context is created before your code, you have to
   * set them directly (through command line options or properties file)
   *
   * @return
   */
  def createSparkConf(exisitingConf : SparkConf = null) : SparkConf = {
    var _conf = exisitingConf
    if (_conf == null) {
      _conf = new SparkConf()
    }
    readConf.foreach(c => _conf.set(c._1, c._2))
    _conf
  }

  /**
   * This method should be call before any BigDL procedure and after the Spark context is created.
   *
   * BigDL needs some Spark conf values to be set correctly to have a better performance. There's
   * also multi-thread engines so executor number and core number per executor need to be known
   * to set the parameter of these engines correctly.
   *
   * The method can set parameters of multi-thread engines, verify spark conf values of an
   * existing spark context.
   */
  def init: Unit = this.synchronized {
    if (localMode) {
      logger.info("Detect bigdl.localMode is set. Run workload without spark")
      // The physical core number should have been initialized
      // by java property -Dbigdl.coreNumber=xx
      setNodeAndCore(1, getCoreNumberFromProperty)
    } else {
      logger.info("Auto detect executor number and executor cores number")
      val (nExecutor, executorCores) = sparkExecutorAndCore().get
      logger.info(s"Executor number is $nExecutor and executor cores number is $executorCores")
      setNodeAndCore(nExecutor, executorCores)
      checkSparkContext
    }
  }

  private val logger = Logger.getLogger(getClass)
  private val singletonCounter = new AtomicBoolean()
  private var physicalCoreNumber = -1
  private var nodeNum: Int = -1

  @volatile
  private var gatewayServer: py4j.GatewayServer = null

  private def createGatewayPortFile(port: Int): Unit = {
    val file = new java.io.File(SparkFiles.getRootDirectory(), "gateway_port")
    logger.debug(s"Creating JavaGatewayServer port file" +
      s" on executor-${SparkEnv.get.executorId}:${file.getAbsolutePath}")
    if (file.exists()) {
      file.delete()
    }
    file.createNewFile()
    val out = new PrintWriter(file)
    try {
      out.print(port)
      out.flush()
    } finally {
      out.close()
    }
  }

  private[bigdl] def createJavaGateway(driverPort: Int): Unit = {
{'fraction_non_alphanumeric': 0.08764455264759587, 'fraction_numerical': 0.00426049908703591, 'mean_word_length': 3.15290648694187, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 6, 'https://': 4, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3753685', 'n_tokens_mistral': 2887, 'n_tokens_neox': 2821, 'n_words': 985}