카산드라에는 read repair(한국어로 굳이 번역하면 읽기 보수 정도?) 라는 것이 있다.

Eventually consistency를 제공한 것은 성능을 포기할 수 없기 때문인데..

데이터의 동기화를 하지만 완벽하지 않을 수 있다. 언제든지 데이터가 이슈가 될 수 있어서 read repair로 데이터의 이슈를 보정할 수 있다. read할 때 quorum으로 하면 데이터를 안전하게 최신 데이터의 맞춰 가져올 수 있다. 

https://docs.datastax.com/en/cassandra/2.1/cassandra/operations/opsRepairNodesReadRepair.html

https://docs.datastax.com/en/cassandra/2.1/cassandra/dml/dmlClientRequestsRead.html

http://www.datastax.com/dev/blog/common-mistakes-and-misconceptions

하지만, read repair를 한다는 것은 성능의 저하를 발생시킬 수 있는 부분이 있다. 

read_repair_chance의 기본 값이 0.1일 텐데. 0으로 바꾸면 성능 저하가 없을 수 있다. 

만약 write : 1, read: quorum으로 하면 read_repiar_chance의 값을 높일 수도 있을 것이다.

CRDT관점으로 설명된 카산드라

https://aphyr.com/posts/294-jepsen-cassandra

merge하는 부분을 잘 보면 좋을 것 같다.

* 주의할 점

DateTieredCompactionStrategy에는 read repair를 포함하기 때문에 read_repair_chance를 0으로 해야 한다.. 기본 정책은 0.1이기 때문에 cassandra.yml 파일을 수정해서 재시작을 해야 한다. 따라서 테이블에 DateTieredCompactionStrategy의 사용을 주의해서 써야 한다.

The compaction strategy DateTieredCompactionStrategy precludes using read repair, because of the way timestamps are checked for DTCS compaction. In this case, you must set read_repair_chance to zero. For other compaction strategies, read repair should be enabled with a read_repair_chancevalue of 0.2 being typical.

-----

qcon london에서 crdt 내용이 나왔다. 

https://qconlondon.com/london-2016/system/files/presentation-slides/matthiaspeter.pdf

Future에 있어서 약점은 타임아웃도 Exception이라는 점이다. 

타임아웃이 발생했을 때 Exception이 발생한다. 

import scala.concurrent.{Await, Future}
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits.global

println("Start")
val f = Future {
  println("1")
  Thread.sleep(2000) // 실제 코드가 들어갈 장소
  println("2")
}

println("End")
val result = Await.result(f, 1 second)

결과는 예외이다. 

일반 Exception과 타임아웃 Exception을 구분하기 위해 클래스를 감쌀 수 있다. 

https://github.com/PacktPublishing/Scala-High-Performance-Programming/blob/master/chapter6/src/main/scala/highperfscala/concurrency/future/SafeAwait.scala

result 함수는 Future는 Awaitable을 상속받은 클래스와 시간을 받고 Future를 감싼다. TimeoutException시는 None으로 리턴한하게 한다. 

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.{Future}
import java.util.concurrent.TimeoutException
import scala.concurrent.duration._
import scala.concurrent.{Awaitable, Await}
import scala.util.{Failure, Success, Try }

object SafeAwait {
  def result[T](awaitable: Awaitable[T],
                 atMost: Duration): Option[T] 
                 = Try(Await.result(awaitable, atMost)) match {
    case Success(t) => Some(t)
    case Failure(_: TimeoutException) => None
    case Failure(e) => throw e
  }
}

val f1 = Future {
  println("11")
  5 + 10
}

println(SafeAwait.result(f1, 1 second))

val f2 = Future {
  println("11")
  Thread.sleep(5000)
  5 + 10
}

println(SafeAwait.result(f2, 1 second))

결과는 다음과 같다. 

Future의 변환에서 Exception이 발생할 수 있는데, 이에 대한 Future에서 어떻게 대응할 수 있는 살펴본다.

문자를 숫자로 변환했으니. map 단계에서 변환할 것이다. 

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.{Future}
import java.util.concurrent.TimeoutException
import scala.concurrent.duration._
import scala.concurrent.{Awaitable, Await}
import scala.util.{Failure, Success, Try }

val f = Future("stringstring").map(_.toInt).map(i => {
  println("Multiplying")
  i * 2
})

Await.result(f, 1 second)

<결과>

Future에는 recover라는 메소드가 있다. 이를 활용하면, 다운 스트림 에러를 보정할 수 있다. 

def recover[U >: T](pf: PartialFunction[Throwable, U])(implicit executor: ExecutionContext): Future[U] = {
  val p = Promise[U]()
  onComplete { v => p complete (v recover pf) }
  p.future
}

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.{Future}
import java.util.concurrent.TimeoutException
import scala.concurrent.duration._
import scala.concurrent.{Awaitable, Await}
import scala.util.{Failure, Success, Try }

val f = Future("stringstring").map(_.toInt).recover {
  case _: NumberFormatException => 2
}.map(i => {
  println("Multiplying")
  i * 2
})
println(Await.result(f, 1 second))

결과는 다음과 같다. 

recoverWith라는 메소드도 있다. 부분 함수를 사용할 수 있으며, recover와 동일하게 처리할 수 있다. 

def recoverWith[U >: T](pf: PartialFunction[Throwable, Future[U]])
         (implicit executor: ExecutionContext): Future[U] = {

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.{Future}
import java.util.concurrent.TimeoutException
import scala.concurrent.duration._
import scala.concurrent.{Awaitable, Await}
import scala.util.{Failure, Success, Try }

val f = Future("stringstring").map(_.toInt).recoverWith {
  case _: NumberFormatException => Future(2)
}.map(i => {
  println("Multiplying")
  i * 2
})
println(Await.result(f, 1 second))

결과는 다음과 같다. 

Await.result 대신 onComplete를 사용할 수 있다. 

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.{Future}
import java.util.concurrent.TimeoutException
import scala.concurrent.duration._
import scala.concurrent.{Awaitable, Await}
import scala.util.{Failure, Success, Try }

val f = Future("stringstring").map(_.toInt).recover {
  case _: NumberFormatException => 2
}.map(i => {
  println("Multiplying")
  i * 2
}).onComplete {
  case Success(i) => println(s"$i")
  case Failure(e) => println(s"${e.getMessage}")
}

Thread.sleep(1000)

Future는 이미 계산된 값으로 만들 수 있다. 그리고 특정 값을 래핑한 Future로 만들 수 있다. (Future를 리턴할 때)

예)

val stringFuture: Future[String] = Future.successful("yes")

Future.successful이나 Future.failed를 이용할 수 도 있고, 더 추상화된 클래스인 Promise를 사용할 수 있다. 아래 예시 결과에서 본 것처럼 모두 계산된 (isCompleted=true) 상태이다. 

import scala.concurrent.{Future}
import scala.concurrent.Promise

val f1 = Future.successful(3)
println(f1.isCompleted)
println(f1)

val f2 = Future.failed(new IllegalArgumentException("no!"))
println(f2.isCompleted)
println(f2)


val promise = Promise[String]()
val future = promise.future
println(promise.success("hello").isCompleted)

결과