<commit_before>//
//  Future.swift
//  Based originially on swiftz code by <NAME>
//  With GCD additions by <NAME>
//

import Foundation

private let sharedProcessingQueue = dispatch_queue_create("llama.future.shared-processing", DISPATCH_QUEUE_CONCURRENT)

public class Future<T> {
  private var _value: T?
  var onCompleteHandlers: [(T -> ())] = []

  // The resultQueue is used to read the result. It begins suspended
  // and is resumed once a result exists.
  // FIXME: Would like to add a uniqueid to the label
  let resultReadyGroup = dispatch_group_create()

  let mutateQueue = dispatch_queue_create("llama.future.value", DISPATCH_QUEUE_SERIAL)

  let processingQueue: dispatch_queue_t

  public init(queue: dispatch_queue_t = sharedProcessingQueue) {
    self.processingQueue = queue
    dispatch_group_enter(self.resultReadyGroup)
  }

  public convenience init(_ f: () -> T, queue: dispatch_queue_t = sharedProcessingQueue) {
    self.init(queue: queue)
    dispatch_async(self.processingQueue) { self.completeWith(f()) }
  }

  public func isCompleted() -> Bool {
    var isCompleted: Bool = false
    dispatch_sync(self.mutateQueue) {
      isCompleted = self._value != nil
    }
    return isCompleted
  }

  public func onComplete(f: T -> ()) {
    dispatch_async(mutateQueue) {
      self.onCompleteHandlers += [f]
    }
  }

  internal func completeWith(x: T) {
    dispatch_async(mutateQueue) {
      precondition(self._value == nil, "Future cannot complete more than once")
      self._value = x
      for handler in self.onCompleteHandlers { handler(x) }
      dispatch_group_leave(self.resultReadyGroup)
    }
  }

  public func result() -> T {
    return self.waitResult()!
  }

  public func waitResult(timeout: dispatch_time_t = DISPATCH_TIME_FOREVER) -> T? {
    if dispatch_group_wait(self.resultReadyGroup, timeout) == 0 {
      return self._value!
    } else {
      return nil
    }
  }

  public func map<U>(f: T -> U) -> Future<U> {
    return future { f(self.result()) }
  }

  public func flatMap<U>(f: T -> Future<U>) -> Future<U> {
    return future { f(self.result()).result() }
  }
}

func sequence<T>(futures: [Future<T>]) -> Future<[T]> {
  return future {
    futures.reduce([]) { acc, f in
      return acc + [f.result()]
    }
  }
}

// FIXME: This should be combinable with the Result version.
// But not sure how to define Functor without forcing Result and Future to be subclasses
// Result is an enum, so that's hard.
func <**><T,U>(x: Future<T>, f: T -> U) -> Future<U> {
  return x.map(f)
}

func future<T>(f: () -> T) -> Future<T> {
  return Future(f)
}<commit_msg>Try using dispatch_group for the completion handlers, too.<commit_after>//
//  Future.swift
//  Based originially on swiftz code by <NAME>
//  With GCD additions by <NAME>
//

import Foundation

private let sharedProcessingQueue = dispatch_queue_create("llama.future.shared-processing", DISPATCH_QUEUE_CONCURRENT)

public class Future<T> {
  private var _value: Result<T>?

  // The resultQueue is used to read the result. It begins suspended
  // and is resumed once a result exists.
  // FIXME: Would like to add a uniqueid to the label
  let resultReadyGroup = dispatch_group_create()

  let mutateQueue = dispatch_queue_create("llama.future.value", DISPATCH_QUEUE_SERIAL)

  let processingQueue: dispatch_queue_t

  public init(queue: dispatch_queue_t = sharedProcessingQueue) {
    self.processingQueue = queue
    dispatch_group_enter(self.resultReadyGroup)
  }

  public convenience init(_ f: () -> Result<T>, queue: dispatch_queue_t = sharedProcessingQueue) {
    self.init(queue: queue)
    dispatch_async(self.processingQueue) { self.completeWith(f()) }
  }

  public func isCompleted() -> Bool {
    var isCompleted: Bool = false
    dispatch_sync(self.mutateQueue) {
      isCompleted = self._value != nil
    }
    return isCompleted
  }

  // FIXME: Use dispatch_group_notify to schedule
  public func onComplete(f: Result<T> -> ()) {
    dispatch_group_notify(self.resultReadyGroup, processingQueue) { f(self._value!) }
  }

  // FIXME: Implement on top of onComplete w/ switch
  public func onSuccess(f: T -> ()) {
    dispatch_group_notify(self.resultReadyGroup, self.processingQueue) {
      switch self._value! {
      case .Success(let box):
        dispatch_group_notify(self.resultReadyGroup, self.processingQueue) { f(box.unbox) }
      case .Failure(_): return
      }
    }
  }

  // FIXME: Implement on top of onComplete w/ switch
  public func onFailure(f: NSError -> ()) {
    dispatch_group_notify(self.resultReadyGroup, self.processingQueue) {
      switch self._value! {
      case .Success(_): return
      case .Failure(let err): dispatch_group_notify(self.resultReadyGroup, self.processingQueue) { f(err) }
      }
    }
  }

  internal func completeWith(x: Result<T>) {
    dispatch_async(mutateQueue) {
      precondition(self._value == nil, "Future cannot complete more than once")
      self._value = x
//      for handler in self.onCompleteHandlers { handler(success(x)) }
      dispatch_group_leave(self.resultReadyGroup)
    }
  }

  public func result() -> Result<T> {
    return self.waitResult()!
  }

  public func waitResult(timeout: dispatch_time_t = DISPATCH_TIME_FOREVER) -> Result<T>? {
    if dispatch_group_wait(self.resultReadyGroup, timeout) == 0 {
      return self._value!
    } else {
      return nil
    }
  }

  public func map<U>(f: T -> U) -> Future<U> {
    let newFuture = Future<U>()
    self.onComplete { r in
      switch r {
      case .Success(let box): newFuture.completeWith(success(f(box.unbox)))
      case .Failure(let err): newFuture.completeWith(failure(err))
      }
    }
    return newFuture
  }

  public func flatMap<U>(f: T -> Future<U>) -> Future<U> {
    let newFuture = Future<U>()
    self.onComplete { r in
      switch r {
      case .Success(let box): newFuture.completeWith(f(box.unbox).result())
      case .Failure(let err): newFuture.completeWith(failure(err))
      }
    }
    return newFuture
  }
}

func sequence<T>(futures: [Future<T>]) -> Future<[T]> {
  return future {
    return futures.reduce(success([T]())) { acc, fu in
      switch acc {
      case .Success(let accBox):
        switch fu.result() {
        case .Success(let resultBox): return success(accBox.unbox + [resultBox.unbox])
        case .Failure(let err): return failure(err)
        }
      case .Failure(let err): return acc
      }
    }
  }
}

// FIXME: This should be combinable with the Result version.
// But not sure how to define Functor without forcing Result and Future to be subclasses
// Result is an enum, so that's hard.
func <**><T,U>(x: Future<T>, f: T -> U) -> Future<U> {
  return x.map(f)
}

func future<T>(f: () -> Result<T>) -> Future<T> {
  return Future(f)
}
{'fraction_non_alphanumeric': 0.12193319538870824, 'fraction_numerical': 0.0002955956251847473, 'mean_word_length': 3.258653241032096, 'pattern_counts': {'":': 0, '<': 46, '<?xml version=': 0, '>': 77, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '4853065', 'n_tokens_mistral': 2159, 'n_tokens_neox': 2124, 'n_words': 568}