Базовые примитивы многопоточности

Часть 5. Базовые примитивы многопоточности

Страничка курса: https://maxcom.github.io/scala-course-2020/

План

Домашнее задание.
Асинхронность и многопоточность.
Поток, пулы потоков.
Синхронизация, блокировки и атомики.
Future и Promise.
Реализация функций работы с Future.
Монадные трансформеры.

"Домашнее" задание: развиваем классификатор

Правки по code review
Добавляем нормализацию словоформ
Добавляем диагностику

Игнорируем окончания у слов.

Используем готовую библиотеку - Apache Lucene


libraryDependencies += 
  "org.apache.lucene" % "lucene-analyzers-common" % "7.2.1"

build.sbt

Создаем анализатор:


val analyzer = new RussianAnalyzer()

В комплекте - токенизатор и стеммер Портера

Нам нужно достать


case class Term(word: String, start: Int, end: Int)

val ts = analyzer.tokenStream("text", "тестовая строка")
ts.reset()

val out = new ArrayBuffer[Term]

while (ts.incrementToken()) {
  val word = 
    ts.getAttribute(classOf[CharTermAttribute]).toString

  val offsets = ts.getAttribute(classOf[OffsetAttribute])

  out += Term(word, offsets.startOffset(), offsets.endOffset())
}

out // Term(тестов,0,8), Term(строк,9,15)

Добавляем диагностику

При классификации для каждого класса выбираем 3 характерных слова

Для итогового класса выделяем слова в тексте символами '*'

Пример:


вот вам английский язык! Выучить от сих до сих! Приеду — проверю! 
Если *не* выучите — моргалы *выколю*, пасти *порву* и, 
как их, эти… носы пооткушу. Ясно?!

Асинхронность и многопоточность.

Потоки выполнения

Каждый поток выполняется процессором независимо.
Все потоки работают в общем адресном пространстве, но имеют свои стеки.
Процессор переключается между разными потоками.
В многоядерных системах потоки выполняются одновременно.

Потоки применяют для:

Выполнения программы более чем на одном ядре CPU.
Параллельного выполнения разных процессов программы.

В серверных проложениях часто выделяют по потоку каждому клиенту.

Программисту это удобно, но не всегда эффективно. В 7-й лекции поговорим об устройстве высоконагруженных приложений.

Асинхронность

Синхронная функция:


def func(input: String): String

Иногда работа не выполняется в нашей функции:

Запрос к БД выполняет БД
Серевой I/O выполняет ядро
Работу делает другой поток
...

Выполняем асинхронно - запускаем выполнение, которое само сигнализирует нам об его окончании.

Зачем?

Работа с большим числом сокетов
рассмотрим на 7-й лекции
"Отзывчивость"
Отказоустойчивость
рассмотрим на лекции про Akka

Асинхронная функция:


def func(input: String, onComplete: Try[String] => Unit): Unit

Иногда возвращается Cancellable

Проблема: мы не хотим работать с такими функциями напрямую.

Работа с потоками

Создание потока - часть Java Runtime:


val thread = new Thread(() => {
  println("Hello world!")
})	

thread.start()

Явно потоки (почти) никогда не нужно создавать:

создание потока - "дорогая" операция
потоков не должно быть много

Сколько должно быть потоков?

под математику: число ядер
сетевое взаимодействие и простая логика без блокирующих вызовов: N * CPU, N ~= 3
дисковый ввод-вывод: число шпинделей жесткого диска
и т. п.

Верхняя граница у современных ОС - порядка 1000 потоков на процесс. Дальше не эффективно.

Проблема: хотим управление потоками, не хотим работать с Thread напрямую.

Пул потоков создается один раз


// используем Java API
val executor: ExecutorService =
  Executors.newFixedThreadPool(10) // 10 потоков

// создаем Scala-обертку
implicit val ec: ExecutionContextExecutor =
  ExecutionContext.fromExecutor(executor)

Задачи можно передать так:


// SAM-интерфейс для Java
// def execute(runnable: Runnable): Unit

ec.execute(() => {
  println("Hello world!")
})

В пуле задачи помещаются в очередь, задачи из которой выполняются в потоках пула.

Стандартные реализации:

ForkJoinPool - по умолчанию
ThreadPoolExecutor - для редко используемых или очень больших пулов

Проблема: мы не хотим работать с execute напрямую.

Проблема: потоки - источник недетерминированности.


def run(): Unit = for (i <- 0 to 10) println(i)

(0 to 5).foreach { _ =>
  ec.execute(run _)
}

разный порядок при каждом запуске

Следующая проблема: доступ к общим ресурсам.

Например RandomAccessFile:

read(...)
seek(offset)

нельзя использовать из разных потоков

Операции над сложными объектами выполняются в много "ходов".

Общие ресурсы требуют последовательного доступа - пока один поток выполняется другие его ждут.

Синхронизация из Java


val lock = new Object()

var counter: Int = 0

lock.synchronized {
  counter += 1
}

Синхронизация на this - плохо

В Java SDK есть много других вариантов: с поддержкой таймаута, latch, семафоры и т.п.


val lock = new ReentrantReadWriteLock()
var data = Vector(42)

def readData(): Vector[Int] = {
  try {
    lock.readLock().tryLock(1, TimeUnit.MINUTES)

    data
  } finally {
    lock.readLock().unlock()
  }
}

Deadlock

Если брать lock'и в разном порядке в разных потоках можно получить взаимную блокировку, из которой не выйти.

Пример: передача денег между счетами.

Dealock'ы бывают хитрые, но сводятся к той же схеме.

Пример: извлечение объектов парами из ограниченного пула.

Следующая проблема: оптимизации в JVM и процессоре.

Модель исполнения и памяти сложнее чем кажется:

Процессор, компилятор и JVM меняют реальный порядок выполения
Кеши процессоров сами не синхронизируются на некоторых платформах
Запись long/double не атомарна на некоторых платформах

Модель памяти Java сложна - подробности смотрите Java Memory Model прагматика модели

volatile

volatile переменная - создает барьер, без блокировок.


@volatile 
var vcounter: Int = 0

операции над ней "упорядочены", но защиты от "гонки потоков" нет

Atomic

Специальная инструкция процессора - CompareAndSet (CAS)

Меняет значение на новое если старое равно заданному

Потокобезопасно

Atomic счетчик работает без блокировок:


// реализация из исходников JDK; Java
public final int incrementAndGet() {
    for (;;) {
        int current = get();
        int next = current + 1;
        if (compareAndSet(current, next))
            return next;
    }
}

Такой подход может использоваться для списков, деревьев и хеш-таблиц.

Разделяемое изменяемое состояние + параллелизм = проблемы.

Нет хорошего способа расставить блокировки в сложной системе.

Разделяемое состояние требует "протокола" доступа, который нужно разработать и потом ему следовать.

Декомпозиция не работает - проблемы возникают в момент интеграции.

Качественно расставить блокировки не всегда удается:

Например с GUI можно работать только из выделеного потока
Некоторые среды используют один лок на все вызовы

Решение: разделяемое изменяемое состояние + ~~параллелизм~~.

Делаем всю работу в одном потоке.

Пример: СУБД Redis, веб-сервер nginx, сервер приложений node.js

Решение: разделяемое ~~изменяемое~~ состояние + параллелизм.

Используем иммутабельные структуры и цепочки обработки.

Сегодня об этом поговорим, продолжим на 10-м занятии.

Решение: ~~разделяемое~~ изменяемое состояние + параллелизм.

Изменяемое состояние приватное, работаем с ним из одного потока.

Это подход "Акторов", о которых поговорим на 7-м занятии.

Перерыв 5 минут

Future[T]

Future - результат отложенного вычисления.


def func(input: String, onComplete: Try[String] => Unit): Unit

переходим к


val f = def func(input: String): Future[String]
f.onComplete(...) // добавляем Callback

Добавляем callback


import scala.concurrent.{ExecutionContext, Future}
import scala.util.{Failure, Success}

val f: Future[Int] = ???
implicit val ec: ExecutionContext = ???

f.onComplete { // выполняется в потоке ec
  case Success(value) ⇒
    println(value)
  case Failure(ex) ⇒
    println(s"Failed: ${ex.toString}")
}

Еще есть более два механизма

Ожидание выполнения


import scala.concurrent.{Await, Future}
import scala.concurrent.duration._

val f: Future[Int] = ???

val result: Int = Await.result(f, 5 minutes)

блокирует текущий поток до получения результата

Опрос состояния


import scala.concurrent.Future
import scala.util.{Failure, Success}

val f: Future[Int] = ???

f.value match {
  case Some(Success(value)) ⇒
    println(value)
  case Some(Failure(ex)) ⇒
    throw ex
  case None ⇒
    println("Not completed :-(")
}

Promise[T]

Promise - контейнер для результата отложенного вычисления.

Значение можно сохранить один раз
Значение завершает связанную с ним Future


def func(input: String, onComplete: Try[String] => Unit): Unit

def futureFunc(input: String): Future[String] = {
  val p: Promise[String] = Promise()

  func(input, p.complete)

  p.future
}

Запускаем синхронную функцию в отдельном потоке:

def func(input: String): String = ???

def futureFunc(input: String): Future[String] = Future {
  func(input)
}

Реализация Future.apply:


def run[T](f :=> T)
          (implicit ec: ExecutionContext): Future[T] = {
  val p = Promise[T]()

  ec.execute(() ⇒ {
    p.complete(Try(f))
  })

  p.future
}

val f: Future[Int] = run { 2 * 2 }

Иногда хотим поместить готовое значение в Future:

Future.successful - конструирование завершенной Future
Future.failed - конструирование завершенной с ошибкой Future

Callback - не удобно.

Future - функтор

Функция map

источник: Functors and Applicatives

Функция map преобразует значение в Future, например:


val f3: Future[Int] = f.map(_ * 10)

функция будет выполнена тогда, когда
значение f будет вычислено

это может работать вот так:


implicit class MyFuture[T](val f: Future[T]) extends AnyVal {
  def myMap[R](func: T ⇒ R)
              (implicit ec: ExecutionContext): Future[R] = {
    val p = Promise[R]()

    f.onComplete {
      case Success(v)  ⇒ p.complete(Try(func(v)))
      case Failure(ex) ⇒ p.failure(ex)
    }

    p.future
  }
}

Future - монада

flatMap[S](f: T => Future[S]): Future[S]

Пример цепочки трансформаций:


def userByEmail(email: String): Future[Int] = ???

def ticketsByUser(user: Int): Future[Seq[Int]] = ???

val count: Future[Int] = 
  userByEmail("user1@test")
    .flatMap(ticketsByUser)
    .map(_.length)

Можно использовать for:


val count: Future[Int] = for {
  user <- userByEmail("user1@test")
  tickets <- ticketsByUser(user)
} yield {
  tickets.length
}

Еще полезные методы

transform[S](f: Try[T] => Try[S]): Future[S]
transformWith[S](f: Try[T] => Future[S]): Future[S]

Совет: избегайте в своем коде таких типов как Future[Future[...]], Future[Try[...]] и Try[Future[...]].
Это источник проблем.

Сборка независимых Future


val res: Future[Result] = for {
  info  <- getUserInfo(user)
  stats <- getUserStat(user)
} yield Result(info, stats)

проблема - задержка вызова getUserStat


val infoF = getUserInfo(user)
val statsF = getUserStat(user) 

val res: Future[Result] = for {
  info  <- infoF
  stats <- statsF
} yield Result(info, stats)

Монада - абстракция цепочки вычислений;
тут не подходит

Вспомним аппликативный функтор

product - комбинирует два функтора
в функтор от пары


def product[A, B](fa: F[A], fb: F[B]): F[(A, B)]

Future - аппликативный функтор
(в библиотеке Cats)


import cats.instances.future._
import cats.syntax.all._

(getUserInfo(user), getUserInfo(user)).tupled

(getUserInfo(user), getUserInfo(user)).mapN(Result.apply)

Нет Cats?
Используем метод zip

Future.sequence для списков


def process(i: Int): Future[Int] = ???

val processed: Seq[Future[Int]] = 
  Seq(1, 2, 3).map(process)

val completed: Future[Seq[Int]] = 
  Future.sequence(processed)

Future.sequence может быть опасен - загружает очередь пула потоков и может "перегрузить" process()

Монадные трансформеры^*

дополнительная часть, если успеем

Что делать с Future[Option[T]]?

Пример map:


val maybeF = Future.successful(Some("value"))

maybeF.map(_.map(_.length))

Функторы можно собрать в один функтор:


import cats.data._
import cats.implicits._

import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global

val maybeF: Future[Option[String]] = 
  Future.successful(Some("value"))

val nested = Nested(maybeF)

nested.map(_.length).value // Future[Option[Int]]

Универсального способа объединить две монады нет.

Что мы хотим?

Асинхронное выполнение, как у Future
Выполнять операции пока не возникнет None,
как у Option

Но можно сделать такой flatMap:


class FutureO[+A](val fut: Future[Option[A]]) extends AnyVal {
  def flatMap[B](f: A ⇒ FutureO[B])
                (implicit ec: ExecutionContext): FutureO[B] = {
    val newFuture = fut.flatMap {
      case Some(a) ⇒ f(a).fut
      case None    ⇒ Future.successful(None)
    }
    new FutureO(newFuture)
  }
}

Что в этом коде от Option?

Логика работы.

Что в этом коде от Future?

flatMap и функция создания

На месте Future может быть любая другая монада.

В библиотеке cats есть трансформер OptionT


val greetingFO: Future[Option[String]] = ???
val firstnameF: Future[String] = ???
val lastnameO: Option[String] = ???

val ot = for { // OptionT[Future, String]
  g <- OptionT(greetingFO)
  f <- OptionT.liftF(firstnameF)
  l <- OptionT.fromOption[Future](lastnameO)
} yield s"$g $f $l"

val result: Future[Option[String]] = ot.value

Трансформеры есть и для некоторых других монад.

Тема семинара

Практика по использованию Future/Promise.

Напоминаю:

Страничка курса: https://maxcom.github.io/scala-course-2020/