<commit_before>/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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 org.apache.spark.streaming

import java.io.{IOException, ObjectInputStream, ObjectOutputStream}
import java.util.concurrent.ConcurrentHashMap

import org.apache.spark.internal.Logging
import org.apache.spark.streaming.dstream.{DStream, InputDStream, ReceiverInputDStream}
import org.apache.spark.streaming.event.{Event, EventSource, TimerEventSource}
import org.apache.spark.streaming.scheduler.Job
import org.apache.spark.util.Utils

import scala.collection.JavaConversions._
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

final private[streaming] class DStreamGraph extends Serializable with Logging {

  private val inputStreams = new ArrayBuffer[InputDStream[_]]()
  private val outputStreams = new ArrayBuffer[DStream[_]]()

  val eventSourceToBoundStreams = new ConcurrentHashMap[EventSource, mutable.HashSet[DStream[_]]]

  var rememberDuration: Duration = null
  var checkpointInProgress = false

  var zeroTime: Time = null
  var startTime: Time = null
  var batchDuration: Duration = null
  var defaultTimer: TimerEventSource = null

  def eventSources = {
    eventSourceToBoundStreams.keySet().toSet
  }

  def start(time: Time) {
    this.synchronized {
      require(zeroTime == null, "DStream graph computation already started")
      zeroTime = time
      startTime = time
      defaultTimer.setStartTime(time + batchDuration)
      outputStreams.foreach(_.initialize(zeroTime))
      outputStreams.foreach(_.remember(rememberDuration))
      outputStreams.foreach(_.validateAtStart)
      outputStreams.filter(_.boundEventSources.isEmpty)
        .foreach(_.bind(defaultTimer))
      inputStreams.par.foreach(_.start())
      eventSources.par.foreach(_.start())
    }
  }

  def restart(time: Time) {
    this.synchronized {
      startTime = time
      eventSources.par.foreach(_.restart())
    }
  }

  def stop() {
    this.synchronized {
      eventSources.par.foreach(_.stop())
      inputStreams.par.foreach(_.stop())
    }
  }

  def setContext(ssc: StreamingContext) {
    this.synchronized {
      eventSources.foreach(_.setContext(ssc))
      outputStreams.foreach(_.setContext(ssc))
    }
  }

  def setBatchDuration(duration: Duration, ssc: StreamingContext) {
    this.synchronized {
      require(batchDuration == null,
        s"Batch duration already set as $batchDuration. Cannot set it again.")
      batchDuration = duration
      defaultTimer = ssc.timer(Time(0), Time(Long.MaxValue), batchDuration, "DefaultTimer")
    }
  }

  def remember(duration: Duration) {
    this.synchronized {
      require(rememberDuration == null,
        s"Remember duration already set as $rememberDuration. Cannot set it again.")
      rememberDuration = duration
    }
  }

  def addInputStream(inputStream: InputDStream[_]) {
    this.synchronized {
      inputStream.setGraph(this)
      inputStreams += inputStream
    }
  }

  def addOutputStream(outputStream: DStream[_]) {
    this.synchronized {
      outputStream.setGraph(this)
      outputStreams += outputStream
    }
  }

  def bind(stream: DStream[_], eventSource: EventSource) {
    val boundStreams = Option(eventSourceToBoundStreams.get(eventSource))
      .getOrElse {
        val emptySet = mutable.HashSet.empty[DStream[_]]
        eventSourceToBoundStreams.put(eventSource, emptySet)
        emptySet
      }
    boundStreams += stream
  }

  def getInputStreams(): Array[InputDStream[_]] = this.synchronized { inputStreams.toArray }

  def getOutputStreams(): Array[DStream[_]] = this.synchronized { outputStreams.toArray }

  def getReceiverInputStreams(): Array[ReceiverInputDStream[_]] = this.synchronized {
    inputStreams.filter(_.isInstanceOf[ReceiverInputDStream[_]])
      .map(_.asInstanceOf[ReceiverInputDStream[_]])
      .toArray
  }

  def getBoundStreams(eventSource: EventSource): Seq[DStream[_]] = {
    eventSourceToBoundStreams.getOrDefault(eventSource, mutable.HashSet.empty).toSeq
  }

  def getInputStreamName(streamId: Int): Option[String] = synchronized {
    inputStreams.find(_.id == streamId).map(_.name)
  }

  def generateJobs(event: Event): Seq[Job] = {
    logDebug(s"Generating jobs for event $event")
    val jobs = this.synchronized {
      getBoundStreams(event.eventSource)
        .flatMap { stream =>
          val jobOption = stream.generateJob(event)
          jobOption.foreach(_.setCallSite(stream.creationSite))<commit_msg>Set the graph of the stream after bind<commit_after>/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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 org.apache.spark.streaming

import java.io.{IOException, ObjectInputStream, ObjectOutputStream}
import java.util.concurrent.ConcurrentHashMap

import org.apache.spark.internal.Logging
import org.apache.spark.streaming.dstream.{DStream, InputDStream, ReceiverInputDStream}
import org.apache.spark.streaming.event.{Event, EventSource, TimerEventSource}
import org.apache.spark.streaming.scheduler.Job
import org.apache.spark.util.Utils

import scala.collection.JavaConversions._
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

final private[streaming] class DStreamGraph extends Serializable with Logging {

  private val inputStreams = new ArrayBuffer[InputDStream[_]]()
  private val outputStreams = new ArrayBuffer[DStream[_]]()

  val eventSourceToBoundStreams = new ConcurrentHashMap[EventSource, mutable.HashSet[DStream[_]]]

  var rememberDuration: Duration = null
  var checkpointInProgress = false

  var zeroTime: Time = null
  var startTime: Time = null
  var batchDuration: Duration = null
  var defaultTimer: TimerEventSource = null

  def eventSources = {
    eventSourceToBoundStreams.keySet().toSet
  }

  def start(time: Time) {
    this.synchronized {
      require(zeroTime == null, "DStream graph computation already started")
      zeroTime = time
      startTime = time
      defaultTimer.setStartTime(time + batchDuration)
      outputStreams.foreach(_.initialize(zeroTime))
      outputStreams.foreach(_.remember(rememberDuration))
      outputStreams.foreach(_.validateAtStart)
      outputStreams.filter(_.boundEventSources.isEmpty)
        .foreach(_.bind(defaultTimer))
      inputStreams.par.foreach(_.start())
      eventSources.par.foreach(_.start())
    }
  }

  def restart(time: Time) {
    this.synchronized {
      startTime = time
      eventSources.par.foreach(_.restart())
    }
  }

  def stop() {
    this.synchronized {
      eventSources.par.foreach(_.stop())
      inputStreams.par.foreach(_.stop())
    }
  }

  def setContext(ssc: StreamingContext) {
    this.synchronized {
      eventSources.foreach(_.setContext(ssc))
      outputStreams.foreach(_.setContext(ssc))
    }
  }

  def setBatchDuration(duration: Duration, ssc: StreamingContext) {
    this.synchronized {
      require(batchDuration == null,
        s"Batch duration already set as $batchDuration. Cannot set it again.")
      batchDuration = duration
      defaultTimer = ssc.timer(Time(0), Time(Long.MaxValue), batchDuration, "DefaultTimer")
    }
  }

  def remember(duration: Duration) {
    this.synchronized {
      require(rememberDuration == null,
        s"Remember duration already set as $rememberDuration. Cannot set it again.")
      rememberDuration = duration
    }
  }

  def addInputStream(inputStream: InputDStream[_]) {
    this.synchronized {
      inputStream.setGraph(this)
      inputStreams += inputStream
    }
  }

  def addOutputStream(outputStream: DStream[_]) {
    this.synchronized {
      outputStream.setGraph(this)
      outputStreams += outputStream
    }
  }

  def bind(stream: DStream[_], eventSource: EventSource) {
    val boundStreams = Option(eventSourceToBoundStreams.get(eventSource))
      .getOrElse {
        val emptySet = mutable.HashSet.empty[DStream[_]]
        eventSourceToBoundStreams.put(eventSource, emptySet)
        emptySet
      }
    boundStreams += stream
    if (stream.graph == null) {
      stream.setGraph(this)
    }
  }

  def getInputStreams(): Array[InputDStream[_]] = this.synchronized { inputStreams.toArray }

  def getOutputStreams(): Array[DStream[_]] = this.synchronized { outputStreams.toArray }

  def getReceiverInputStreams(): Array[ReceiverInputDStream[_]] = this.synchronized {
    inputStreams.filter(_.isInstanceOf[ReceiverInputDStream[_]])
      .map(_.asInstanceOf[ReceiverInputDStream[_]])
      .toArray
  }

  def getBoundStreams(eventSource: EventSource): Seq[DStream[_]] = {
    eventSourceToBoundStreams.getOrDefault(eventSource, mutable.HashSet.empty).toSeq
  }

  def getInputStreamName(streamId: Int): Option[String] = synchronized {
    inputStreams.find(_.id == streamId).map(_.name)
  }

  def generateJobs(event: Event): Seq[Job] = {
    logDebug(s"Generating jobs for event $event")
    val jobs = this.synchronized {
      getBoundStreams(event.eventSource)
        .flatMap { stream =>
          val jobOption = stream.generateJob(event)
          jobOption.foreach(_.setCallSite(stream.creationSite))
{'fraction_non_alphanumeric': 0.09622515926595036, 'fraction_numerical': 0.0009508414947228297, 'mean_word_length': 3.871699861046781, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, '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': '3873839', 'n_tokens_mistral': 3043, 'n_tokens_neox': 2978, 'n_words': 724}