작성자: josephjang

QCon San Francisco 2015 Day 1

josephjang 1 Comment

어쩌다보니 지금까지 미국이나 유럽 등지를 한번도 여행한 적이 없어서, 올해 초 다짐한 것이 미국 여행. 페이퍼를 읽는 것이나 토크를 보는 것도 좋아하기 때문에 무겁지 않은 QCon에도 가보자라고 해서 올해 봄 무렵에 QCon SF 2015를 예약해두었다. 여러가지 위기는 있었지만, 가족여행을 겸해서 QCon SF에 참석하는 계획을 잡았고, 지난 토요일에 San Francisco에 도착했다. 이로서 목표 달성!

오늘부터 3일에 걸쳐 QCon SF 2015의 본 프로그램이 진행되고 이후 이틀동안은 Workshop이 진행된다. 오늘만 하더라도 듣고 싶었던 토크는 꽤 많았지만, 그래도 고심 끝에 골라서 들어간 토크들이니 내 입장에서 듣고 느낀 것들을 적어본다.

Avoiding the Big Crash by Bill Buxton (Slides)

Xerox PARC 출신으로 35년간 컴퓨터 사이언티스트와 디자이너로서 종사하고 현재는 MSR의 Principal Researcher. 스마트 왓치와 같은 초소형 디스플레이를 가진 제품과 벽면을 가득 채울 정도로 커다란 디스플레이를 가진 제품은 완전히 다른 설계를 가져야 함을 설명하면서, 또한 기존의 경험들을 활용하기 위해 common denominator도 찾아야 한다면서 그 예로 Bimanual In-Place Commands를 든다.

모든 새로운 제품은 물론 그 자체로도 가치와 좋은 경험을 제공해야하지만, 전체적인 복잡도를 줄이고 다른 제품들의 가치를 더 높일 수 있어야 한다는 룰을 제시하고, 여러가지 inspiration이 될만한 설명들을 제공해주었다. 단지 돈을 벌기 위한 수단, 다른 제품들과의 경쟁 우위에 있어야만 하는 것으로서만 제품을 생각하는 것이 아니라 우리 주변에 있는 다른 사람들에게 더 좋은 경험을 제공해주기 위한 고민을 우리 스스로가 해야한다는 메시지로 받아들였다. 글로만 이러한 내용을 설명하기는 어렵지만 내가 했던 일들과 하고 있는 일, 그리고 해야하는 일들을 모두 돌아보게 만드는 굉장히 인상 깊은 토크였다.

Scaling Uber by Matt Ranney

5년동안 PHP+MySQL로부터 시작해서 빠르게 개발하기 위해서 technical debt도 많이 쌓였다는 이야기를 하면서 저런 성장하는 회사의 모습은 한국의 비슷한 상황의 회사와도 크게 다르지 않구나라고 느꼈다. 한편으로는 저런 이야기라면 LINE의 이야기도 충분히 재미있는 토크감이 될 수 있다고 생각했다. (물론 영어가 문제…)

Microservices 아키텍쳐로 2개의 서비스로 시작했던 서비스의 개수가 무려 700개 이상으로 늘어났다는 것이 굉장히 놀라웠다. PHP로 시작했지만 지금은 node.js, Python, Go, Java로 언어들이 늘어났다고 한다. 이러한 환경의 어려움을 극복하기 위한 도구로서, 서비스 간의 분산 및 failover를 위해서 ringpop-node, 통신 프로토콜로서 TChannel (+ tcurl, tcap, thriftrw), 서비스 디스커버리로서 Hyperbahn 등을 소개했는데, 다른 오픈소스를 가져다 쓰기 보다는 자체적으로 개발하고 이를 오픈소스화 했다는 점이 조금 신기했다.

How Netflix Directs 1/3rd of Internet Traffic by Haley Tucker & Mohit Vora

No comment.

Beyond the Hype: 4 years of Go in Production by Travis Reeder

Go가 그리 유명하지 않던 시절 Ruby의 CPU 소비를 줄이기 위해서 Go로 이동했다고 한다. Python, Javascript, Java, Ruby등과의 성능 비교 그래프를 보여주었다. 언어를 바꾸고 나서 30대 서버 (…)에서 2대 서버로 바꿀 수 있었다고 한다. 언어의 변경에 따르는 여러가지 문제들에 대해서 논리적인 대답을 가지고 있는 것은 아니라서 좀 아쉬웠다. 토크 자체는 20분 남짓 만에 일찍 끝나고 수많은 질문들이 이어졌는데, 대답하기 쉬운 질문들은 물론 아니었지만 대부분 문제된 적이 없거나 잘 모르겠다는 답변이 돌아와서 ‘4년’의 경험에서 기대한 것보다는 많이 모자랐다. Go언어의 초보라고 할 수 있는 내가 알고 있는 수준에서 크게 다르지 않다라는 느낌.

다만 질문들은 나름대로 Go언어 생태계에서 정답을 내놓을 필요가 있는 것들이라서 나열해본다.

  • Dependency management: vendoring, submodule (Golang은 패키지에 version이라는 개념이 없다.)
  • Compile time: 의존성이 있는 C 라이브러리들이 문제가 되었지 Go가 문제된 적이 없었음. (충분히 큰 소스의 프로젝트라면?)
  • DB들을 위한 라이브러리들의 제공: 아직 불충분.
  • Testing: go test로 충분. rspec과 같은 것이 있는가란 질문에 대해 잘 모르겠지만 Go언어의 특성상 그런 것이 존재할 수 있는지는 의문이라고 코멘트.
  • IDE: sublime, intelliJ, … (Atom은 왜 언급안해…)
  • GC 문제: 지금까지 경험한 적이 없다고. (없는 것 같다고 대답했다가 concurrent GC 언급한 다른 질문자에게 따끔하게 지적받음.)
  • Debugging: 처음에 로그를 사용한다고 대답. godebug 등 오픈소스 도구들도 나타나고 있기 때문에 점차 개선되고 있다고 함. gdb등도 가능. (디버깅에 대한 IDE 지원은 절실하긴 한 듯.)
  • Monitoring: JMX 등의 퍼실러티가 있느냐란 질문이었는데, 그런 건 없고 자기네들은 statsd을 사용.
  • Java 아키텍쳐로부터의 마이그레이션 방법: 마이크로서비스별로.
  • GUI: 서버 위주이고 별로 없는 것 같다고 대답. (아닌데…)
  • Refactoring: 별로 본질적인 것은 아닌 것 같다고 대답.
  • RoR scaffolding: 잘 모르겠다고 대답.
  • Onboard engineer를 위해 특별히 하고 있는 것: 특별히 없음. Go가 variable의 type notation 등의 일부를 제외하면 다른 언어와 크게 다르지 않다고 대답.
  • 언어 개선 표준 프로세스가 있는지: 오픈소스이니 디스커션이 가능하지 않겠냐는 대답.
  • Profiling: pprof. production에 사용하기에는 어렵도 metric을 남기는 쪽이 좋을 듯 하다고 대답.
  • interface가 있는지: 있음.
  • Shared libary로 빌드할 수 있는지: 잘 모르겠다고 대답.
  • Maven 같은 의존성 관리 도구가 있는지: 모르겠다고 대답.
  • 사용하고 있는 design pattern이 있는지: 잘 모르겠다고 대답.

Personalization in the Pinterest Homefeed by Dmitry Chechik

Jet lag 때문인지 너무 졸려서 무슨 이야기를 했는지 잘 모르겠음. 🙁

Rust: unlocking systems programming by Aaron Turon

어느 한 쪽을 희생한 언어들과 달리 Rust는 Control과 Safe라는 특성을 둘다 달성하는 곳을 목표로 하는 언어이고, ownership의 이전 메커니즘에 따른 Memory safety에 대해서 설명함.

Spark: A Coding Joyride by Doug Bateman

Sponsored track인데도 Spark의 명성을 인정하듯 룸이 가득찼다.

Spark의 개요에 대해 설명. Spark, GraphX, MLlib의 코딩을 보여준다고 한 것 같은데, 실제로는 RDD와 필터링 정도만 코드를 보여주었다. 결과적으로는 내가 Spark에 대해서 아는 부분 이상을 얻지는 못한 듯. 스피커가 전문 Trainer라서 그런지 설명을 듣기에는 편했고, Databrics라는 Spark hosting 서비스는 Spark를 시험해보기에 굉장히 편리해보였다. (30일 trial 가능.)

So We Hear You Like Papers by Ines Sombra and Caitie McCaffrey Slides

Caitie McCaffrey는 Distributed system에 대해, Ines Sombra는 Verification 분야에 대해 몇몇 페이퍼들을 소개했다. 둘다 다른 컨퍼런스 등에서도 활약하는 듯 하고, Ines Sombra는 Papers we love meetup의 SF chapter에서 활동하는 듯하다.

둘다 설명하는 속도가 굉장히 빨라서 정확성을 차치하더라도 정말 대단하다고 생각. Exhibiter reception 직후 키노트라 다들 맥주 한두잔 한 후라서 그런지 조금은 덜 진지한 분위기였던 것 같기도 하고… Ines Sombra는 아예 맥주 한병을 들고 마시면서 진행했다. 어쨌든 스마트함을 마음껏 뽐내는 분위기 같아서 재미있었다. 중간중간에 관심을 가져볼만한 페이퍼들도 보였던 것 같아서 나중에 슬라이드를 확인해봐야 할 것 같다.

내일과 모레도 재미있는 토크들은 많은 듯 하니, 게으름 피지 않고 여기에 적어보려고 노력해보겠다. 🙂

Paper: TAO: Facebook’s Distributed Data Store for the Social Graph (Part 1)

josephjang Leave a comment

Bronson, Nathan, et al. “TAO: Facebook’s Distributed Data Store for the Social Graph.” USENIX Annual Technical Conference. 2013.

Facebook은 사용자들 사이의 관계, 사용자들의 포스팅, 이에 대한 코멘트 등을 MySQL에 저장하고 memcache에 캐싱하고 있었는데, 이를 개선한 TAO라는 시스템에 관한 페이퍼. geographically 분산된 단일한 인스턴스라는 점이 놀라운 점. 이후에도 설명되겠지만 graph abstraction만으로 Facebook의 주요한 데이터들을 표현한다는 것도 매우 흥미로운 점.

Before TAO, Facebook’s web servers directly accessed MySQL to read or write the social graph, aggressively using memcache [21] as a lookaside cache. TAO implements a graph abstraction directly, allowing it to avoid some of the fundamental shortcomings of a lookaside cache architecture. TAO continues to use MySQL for persistent storage, but mediates access to the database and uses its own graph-aware cache.
TAO is deployed at Facebook as a single geographically distributed instance. It has a minimal API and explicitly favors availability and per-machine efficiency over strong consistency; its novelty is its scale: TAO can sustain a billion reads per second on a changing data set of many petabytes.

Background

Facebook의 타임라인과 같은 데이터를 구축할 때 흔히 컨텐트가 생성될 때 타임라인 데이터를 업데이트하는 방식과 타임라인을 읽을 때 타임라인 데이터를 업데이트하는 두가지 방식이 고려되는데, 이 페이퍼에서는 Facebook에서는 모든 항목에 대해 privacy check가 동적으로 이루어져야 하기 때문에 전자의 접근을 불가능하다고 얘기한다. 즉, 실시간으로 aggregation과 filtering이 이루어져야 한다는 것.

We present each user with content tailored to them, and we filter every item with privacy checks that take into account the current viewer. This extreme customization makes it infeasible to perform most aggregation and filtering when content is created; instead we resolve data dependencies and check privacy each time the content is viewed.

MySQL과 memcache를 이용한 원래의 아키텍쳐 대신 TAO를 만들어야 했던 이유로, PHP API에서의 encapsulation 실패 – 자세히는 설명되어있지 않지만, graph abstraction이 아니어서 발생하는 데이터 모델상의 여러가지 문제들을 말하는 것이 아닐까 싶다. – 그리고, PHP가 아닌 언어로부터의 접근 – 역시 이러한 요구사항은 흔히 아키텍쳐의 변화를 이끄는 동력이 되는 듯 – 그리고 lookaside 캐시 아키텍쳐의 여러가지 문제들을 들고 있다. 캐시 아키텍쳐의 문제들로는 다음과 같은 문제들을 들고 있다.

우선 edge들의 리스트를 표현하는 데에 있어서 특정 키에 해당하는 모든 edge 리스트를 가져와야 하는 key-value 캐시는 비효율적임을 들고 있다. 캐시에서 리스트를 직접 지원한다면 이러한 문제를 해결할 수 있으나 동시적인 업데이트에 따른 문제들을 언급하고 있다.

Inefficient edge lists: A key-value cache is not a good semantic fit for lists of edges; queries must always fetch the entire edge list and changes to a single edge require the entire list to be reloaded. Basic list support in a lookaside cache would only address the first problem; something much more complicated is required to coordinate concurrent incremental updates to cached lists.

액세스의 제어를 위한 로직들이 클라이언트에 있으므로 thundering herds와 같은 문제들이 발생하는 것을 언급하고 있다. 이를 캐시에 내장함으로써 더욱 효율적으로 문제를 해결할 수 있다고 얘기하고 있다.

Distributed control logic: In a lookaside cache architecture the control logic is run on clients that don’t communicate with each other. This increases the number of failure modes, and makes it difficult to avoid thundering herds. Nishtala et al. provide an in-depth discussion of the problems and present leases, a general solution [21]. For objects and associations the fixed API allows us to move the control logic into the cache itself, where the problem can be solved more efficiently.

Facebook은 MySQL의 비동기 master/slave 리플리케이션을 사용하고 있기 때문에 슬레이브를 사용하는 데이터센터의 캐시들에는 consistency 문제가 있다. 이 페이퍼에서는 가능한 한 복제본의 캐시의 consistency를 유지하기 위한 방법들을 제시하고 있다.

Expensive read-after-write consistency: Facebook uses asynchronous master/slave replication for MySQL, which poses a problem for caches in data centers using a replica. Writes are forwarded to the master, but some time will elapse before they are reflected in the local replica. Nishtala et al.’s remote markers [21] track keys that are known to be stale, forwarding reads for those keys to the master region. By restricting the data model to objects and associations we can update the replica’s cache at write time, then use graph semantics to interpret cache maintenance messages from concurrent updates. This provides (in the absence of multiple failures) read-after-write consistency for all clients that share a cache, without requiring inter-regional communication.

TAO Data Model and API

TAO의 Object는 64-bit integer로 식별되며, object type (otype)을 가지고 있다. Association은 source object (id1), association 타입 (atype), destination object (id2)로 식별된다. 임의의 두 object들 사이에서 특정 타입의 association은 최대 1개만 존재할 수 있다는 제약이 존재한다. Object와 association은 모두 key-value attribute들을 가지고 있으며 가능한 key들과 value들의 type, 그리고 default value는 type별 schema에 따라 정해진다. per-type schema라는 이 개념은 대단히 새로운 것은 아니지만 일반적인 모델의 attribute 정의에 대해 고민하던 내게 도움이 되었다.

TAO objects are typed nodes, and TAO associations are typed directed edges between objects. Objects are identified by a 64-bit integer (id) that is unique across all objects, regardless of object type (otype). Associations are identified by the source object (id1), association type (atype) and destination object (id2). At most one association of a given type can exist between any two objects. Both objects and associations may contain data as key→value pairs. A per-type schema lists the possible keys, the value type, and a default value. Each association has a 32-bit time field, which plays a central
role in queries1.

Object: (id) → (otype, (key  value)∗)
Assoc.: (id1, atype, id2) → (time, (key  value)∗)

TAO의 Object API는 Object에 대한 기본적인 CRUD에 더해서 field들의 subset을 업데이트할 수 있는 API를 지원하고 있다.

Association API도 association의 CRUD를 위한 API를 제공하고 있다. 매우 흥미로운 점은 친구 관계와 같이 양방향을 가지는 association의 경우, 그 association type을 inverse type으로 설정을 해주면 TAO association API에서 알아서 inverse association에 대해서도 operation을 실행해준다는 점이다. Graph DB에서 당연한 기능인지는 잘 모르겠지만, 이를 application layer에서 직접 구현하고자 한다면 boilerplate 코드가 되기 쉬운 부분들이 일반적으로 해소되고 있다고 생각한다. 한편, inverse type의 association을 추가할 때 가장 걱정이 되는 점은 두 association의 write가 atomic하게 반영될 수 있는가 일텐데, 이에 대해서는 페이퍼에서 명시적으로 언급하고 있지는 않은 듯 하다.

  • assoc add(id1, atype, id2, time, (k→v)*) – Adds or overwrites the association (id1, atype,id2), and its inverse (id1, inv(atype), id2) if defined.
  • assoc delete(id1, atype, id2) – Deletes the association (id1, atype, id2) and the inverse if it exists.
  • assoc change type(id1, atype, id2, newtype) – Changes the association (id1, atype, id2) to (id1,
    newtype, id2), if (id1, atype, id2) exists.

Facebook과 같은 서비스에서는 대부분의 데이터는 오래된 것이고 자주 액세스되어야 할 데이터는 최근 생성된 일부의 데이터라는 점을 creation-time locality라는 말로 표현하고 있다. TAO의 Association Query API들은 creation-time locality에 따른 cache 가능성을 높이기 위해 time range 쿼리가 가능한 점을 엿볼 수 있다. 역시 흥미로운 점은 Association Query API로부터 리턴되는 Association List의 association type별 제약이 정해져있다는 점이다. 잘은 모르지만 이러한 점들은 일반적인 Graph DB에서는 가할 수 없는 제약이 아닐까 싶다.

  • assoc get(id1, atype, id2set, high?, low?) – returns all of the associations (id1, atype, id2) and their time and data, where id2 ∈ id2set and high ≥ time ≥ low (if specified). The optional time
    bounds are to improve cacheability for large association lists (see § 5).
  • assoc count(id1, atype) – returns the size of the association list for (id1, atype), which is the number of edges of type atype that originate at id1.
  • assoc range(id1, atype, pos, limit) – returns elements of the (id1, atype) association list with index i ∈ [pos,pos+limit).
  • assoc time range(id1, atype, high, low, limit) – returns elements from the (id1, atype) association list, starting with the first association where time ≤ high, returning only edges where time ≥ low.

Paper: Immutability Changes Everything

josephjang Leave a comment

Helland, Pat, and One Market Street. “Immutability changes everything.” (2012).

Abstract

There is an inexorable trend towards storing and sending immutable data. We need immutability to coordinate at a distance and we can afford immutability, as storage gets cheaper.
This paper is simply an amuse-bouche on the repeated patterns of computing that leverage immutability. Climbing up and down the compute stack really does yield a sense of déjà vu all over again.

Thoughts

저장 장치의 가격이 싸지면서 immutability에 기반한 컴퓨팅이 필요해지고 가능해지고 있다고 주장하며, 컴퓨팅의 여러 layer 걸쳐서 그러한 예들을 보이고 있다. immutability가 컴퓨팅에서 중요한 도구임에는 분명하지만 그렇다고 mutability가 컴퓨팅의 역사에서 사라질 것은 아니므로 선택편향이 의심된다. 차라리 로그가 중요하다는 주장이 나아보인다.

이 페이퍼에서 언급하는 Immutability를 활용하는 예들은 다음과 같다.

  • Transaction logs of DBMSs
    • The truth is the log. The database is a cache of a subset of the log.
  • Accounting: “Accountants don’t use erasers.”
  • Append-only distributed single master
  • Forms with many layers
  • Data on the outside: messages, files, documents
  • MapReduce, Dryad
    • Immutable inputs, idempotent functional computation over immutable inputs
  • Versions
    • A linear version history, DAG of version history
  • Multi-version concurrency control
  • Log structured merge trees
  • Copy-on-write
  • Log structured file systems
  • GFS, HDFS
  • Consistent hashing
  • Wear leveling of SSDs
  • Shingled disk systems

Talk: “When “Worst” is Best (in Distributed Systems)” by Peter Bailis

josephjang Leave a comment

“When “Worst” is Best (in Distributed Systems)” by Peter Bailis at Strange Loop 2015 (Video)

UC Berkeley의 Ph.D. candidate인 Peter Bailis의 Strange Loop 2015에서의 발표.

Worst case를 유리하게 만드는 최적화는 일반적으로 average case를 불리하게 만들지만, 이 Talk에서는 Worst case를 위한 디자인을 통해서 average case도 개선되는 경우들을 예를 들어 보여주고 있다. Worst case에 대한 고려를 하는 것은 매우 강력한 디자인 도구가 될 수 있다는 점을 강조하고 있다.

Worst case를 위한 디자인이 average case도 개선하는 경우들의 예로는 다음과 같은 것들을 들고 있다.

  • Coordination-free/avoding distributed system: CRDTs, I-confluence, RAMP, HAT, Bloom^L, …
  • Replication
  • Failover
  • Tail Latency
  • Universal Design
  • Accessibility

Paper: A simple totally ordered broadcast protocol

josephjang Leave a comment

Reed, Benjamin, and Flavio P. Junqueira. “A simple totally ordered broadcast protocol.” proceedings of the 2nd Workshop on Large-Scale Distributed Systems and Middleware. ACM, 2008.

이 페이퍼는 Zab라는 ZooKeeper의 내부에 구현되어있는 ordered broadcast 프로토콜을 소개하고 있다. 이 페이퍼는 informal한 형태로 Zab에 대한 요구사항과 동기, 그리고 프로토콜 자체에 대해서 설명하고 있으므로 이를 파악하기에 좋은 것 같다. Zookeeper에 대한 페이퍼와 Zab에 대한 좀 더 formal한 페이퍼도 따로 있으므로 앞으로 읽어볼 것.

The logical components of the ZooKeeper service

Zab는 Zookeeper Atomic Broadcast라는 이름대로 Zookeeper의 노드들 사이에 동일한 오퍼레이션들의 순서를 유지하고 결과적으로 노드들의 상태를 동일하게 유지되기 위해 사용된다. (Atomic Broadcast 참고.)

Zookeeper는 클라이언트로부터의 request들을 그대로 Zab로 보내는 것이 아니라 conditional operation이나 version increment를 포함하고 있어서 본질적으로 idempotent하지 않은 request를 idempotent한 transaction으로 변환하는 과정을 가지고 있다. 이 변환은 leader 노드에서만 발생할 수 있으며, 그 이유는 leader 노드만이 오퍼레이션으로 인해 변할 미래의 상태에 대한 완벽한 정보를 가지고, 새로운 값을 정할 수 있기 때문이다.

Requirements

Zookeeper는 다음과 같은 요구사항을 가지고 있다.

  • Reliable delivery
    • If a message, m, is delivered by one server, then it will be eventually delivered by all correct servers.
  • Total order
    • If message a is delivered before message b by one server, then every server that delivers a and b delivers a before b.
  • Causal order
    • If message a causally precedes message b and both messages delivered, then a must be ordered before b.
  • Prefix property
    • If m is the last message delivered for a leader L, any message proposed before m by L must also be delivered.

우리가 알고 있는 Zookeeper의 특성 상 다른 것들은 쉽게 추론할 수 있으나, 클라이언트 입장에서의 correctness를 표현한 Causal order requirement는 이러한 말로 표현하는 것을 아직 자주 본 것은 아니라서 눈여겨보게 되었다.

Zookeeper의 어떤 노드가 실패했을 때 snapshot을 읽어들인 다음 snapshot 이후에 배달된 transaction들을 리플레이하게 되는데, 위에서 설명했던 idempotent한 transaction 덕에 이러한 복구 과정에서 atomic broadcast는 at most once delivery를 보장하지 않아도 된다. 이는 또한 복구 과정에서는 total order requirement가 완화되는 것을 의미하기도 한다.

Motivation

이 부분에서는 왜 이미 존재하는 프로토콜을 사용하지 않고 Zab를 고안해서 사용하게 되었는가를 각 프로토콜을 들어 설명하고 있다. 굉장히 많은 페이퍼들을 인용하며 설명을 하고 있는데, 가장 중요한 것은 역시 Paxos와의 비교라고 할 수 있다. 이 페이퍼는 현실의 조건에 따른 가정을 토대로 Paxos를 단순화시킬 수 있다고 이야기 하고 있다.

Paxos는 메시지의 손실과 순서가 바뀌는 것 조차도 허용하고 있지만, TCP를 사용한다면 메시지가 FIFO 순서로 전달되는 것을 가정할 수 있고, Zab는 이에 따라 per-proposer causality를 보장할 수 있다. Paxos는 FIFO channel을 가정하지 않기 때문에 이러한 보장을 할 수 없다.

두번째로 Paxos에서는 복수의 leader로부터 하나의 인스턴스에 대한 proposal이 가능하므로, proposal이 서로 충돌하거나 어떤 값이 commit 되었는지 알아내기 힘든 문제가 있다. Paxos에서는 어떤 노드라도 높은 투표만 얻으면 리더가 될 수 있는 반면, Zab에서는 다수의 노드가 기존의 리더를 버리지 않는 이상 새로운 리더가 될 수 없다. 이로 인해 Zab에서는 주어진 인스턴스에 대해서 단 하나의 proposal을 보장할 수 있고, 이는 프로토콜을 여러 면에서 단순화 시킨다.

Protocol

Zab protocol은 recovery와 broadcast 두 개의 모드로 이루어져있다. Zookeeper를 처음으로 실행했을 때나 leader 실패가 발생했을 때는 recovery 모드가 되고, leader가 성립되고 quorum이 leader와 동기화를 완료하게 되면 broadcast 모드가 된다.

Zookeeper는 broadcast의 leader를 (request를 transaction으로 변환하기 위한) write request의 leader로도 사용하기 때문에, 만약 분리되었을 경우 발생할 둘 사이의 latency를 제거한다.

Broadcast

단순화된 two-phase commit과 유사하다고 얘기하고 있다. two-phase commit과는 달리 Zab에는 abort가 없기 때문에 follower는 leader의 proposal을 수용하거나 leader를 포기하는 두가지의 선택지 밖에는 없으며, 또한 leader는 quorum을 이룬 follower들로부터 ack을 받는다면 commit을 진행할 수 있다.

broadcast 프로토콜은 모든 커뮤니케이션에 대해 FIFO 채널을 이용하므로 순서를 보증하는 것이 매우 쉽다.

leader가 proposal을 보낼 때는 zxid라고 불리는 단조증가하는 ID를 부여한다. Zab는 causal ordering을 보존하기 때문에 전달된 메시지는 zxid들에 의해서도 정렬된다. 이 후에 proposal은 follower별로 존재하는 queue에 넣어지고, 이는 FIFO 채널을 통해 follower로 전달된다. follower는 받은 proposal을 디스크에 쓰고 leader에게 ack을 보낸다. leader가 다수의 follower로부터 ack을 받으면 commit을 broadcast하고 leader 자신에게 메시지를 배달(deliver)한다. follower들은 commit을 받았을 때 메시지를 배달한다.

Recovery

Recovery에서 중요한 것은 commit된 모든 메시지를 제대로 보존하는 것과 commit되지 않은 – skip된 메시지를 보존하지 않는 것이다.

commit된 모든 메시지를 보존하는 문제는 leader election 과정에서 quorum 서버 중에서 가장 높은 proposal number를 가지고 있는 노드를 leader로 선택하는 것이다. 새로 선출된 leader는 새로운 proposal을 처리하기 전에 트랜잭션 로그에 있는 모든 메시지들이 propose되고 quorum에 의해 commit되도록 한다.

새로운 follower 노드가 추가되었을 때는 leader는 우선 follower가 보지 못한 proposal을 모두 queueing하고 그 다음 마지막으로 commit된 proposal까지 commit들을 queueing한 후에 새로운 follower노드를 broadcast 리스트에 등록한다.

skip된 메시지를 보존하지 않는 것은 epoch을 이용한다. Zookeeper의 zxid는 64bit 숫자인데, lower 32bit은새로운 proposal을 생성할 때마다 증가하는 counter이고, high order 32bit은 epoch에 해당한다. 새로운 leader가 선출될 때마다 로그 상에 가지고 있는 가장 높은 zxid의 epoch에 1을 더하고, counter를 0으로 리셋한 zxid를 이후의 proposal을 위해 사용한다.

epoch을 통해서 여러 리더들이 동일한 zxid를 사용하는 것을 방지할 수 있을 뿐만 아니라, 더욱 중요하게는, 한번 실패했던 리더가 되살아나더라도 zxid가 낮기 때문에 리더가 될 수 없고, 이 노드가 follower가 되었을 때 leader는 기존 epoch의 proposal을 truncate하도록 해주기 때문에, recovery 과정이 단순해지고 빨라지는 장점을 가지고 있다.

Closing

Raft 등에서도 채용된 여러가지 개념들을 볼 수 있었다는 점이 재미있었다. two-phase commit-like 프로토콜이나 quorum이 유지되는 한 leadership의 유지, epoch (term)의 개념 등. 또한 다른 consensus 프로토콜들을 통해 어렴풋이 추측하고 있던 Zookeeper의 프로토콜에 대해 조금 더 정확하게 이해할 수 있게된 것 같다. 사실 Zab 이외에도 여러가지 알고리즘이나 최적화 등이 Zookeeper에 활용되었으리라 추측하는데, 위에서 언급한 Zookeeper에 관련한 다른 페이퍼들이나 Zookeeper의 코드를 읽어보아도 재미있을 것 같다.

Paper: Conflict-free Replicated Data Types

josephjang Leave a comment

Shapiro, Marc, et al. “Conflict-free replicated data types.” Stabilization, Safety, and Security of Distributed Systems. Springer Berlin Heidelberg, 2011. 386-400.

Distributed system에서 replica 사이의 consistency를 유지하는 것은 매우 어려운 문제지만, CRDT라고 불리는 특별한 데이터구조들은 semilattice나 commutativity와 같은 그 자체의 성질을 이용하여 consistency의 문제를 훨씬 단순한 문제로 만들어준다. CRDT에 해당하는 데이터구조나 수학적인 성질들은 오래전부터 알려져있었고 활용되어왔지만, 이 페이퍼의 저자들이 2007년에 처음으로 CRDT라는 이름을 붙였다. 2009년에는 Treedoc이라는 온라인 협업 에디팅을 위한 데이터구조를 제안하면서 역시 CRDT의 한가지 예로서 제시한다. 하지만 이 때까지도 CRDT의 정의는 commutativity를 이용한다 정도로 매우 모호한 상태였지만, 2011년에 출판된 이 페이퍼는 CRDT의 성질을 수학적으로 정의하고 증명한 것에 큰 의미가 있는 것으로 보인다. 이 페이퍼와 같은 해에 출판된 “A comprehensive study of convergent and commutative replicated data types.”란 페이퍼에서는 이 페이퍼의 개념들을 더욱 자세하게 설명할 뿐만 아니라 알려진 CRDT들의 카탈로그를 제시하고 자세하게 설명하고 있다. 이 페이퍼는 오히려 그 페이퍼의 정수만을 간추려놓은 논문으로 보인다.

Strong Eventual Consistency

Eventual consistent 시스템의 경우 필연적으로 conflict resolution이 필요하게 되는데, 이러한 conflict resolution의 부담을 덜어주는 메커니즘으로서 Version vector나 Dotted version vector 등을 사용한다. CRDT는 근본적으로 conflict가 발생하지 않는 데이터 구조이며, 따라서 causality tracking을 필요로 하지 않을 뿐더러 conflict resolution을 위한 coordination 등을 필요로 하지 않는다. 이러한 특성을 Strong Eventual Consistency라고 이야기 하고 있다.

We propose a simple, theoretically-sound approach to eventual consistency. Our system model, Strong Eventual Consistency or SEC, avoids the complexity of conflict resolution and of roll-back. Conflict-freedom ensures safety and liveness despite any number of failures. It leverages simple mathematical properties that ensure absence of conflict, i.e., monotonicity in a semi-lattice and/or commutativity. […] In our conflict-free replicated data types (CRDTs), an update does not require synchronisation, and CRDT replicas provably converge to a correct common state. CRDTs remain responsive, available and scalable despite high network latency, faults, or disconnection.

우선, Eventual Consistency를 어떤 replica의 causal history에 포함된 method는 결국에는 (eventually) 다른 replica의 causal history에 포함되는 것으로 정의하고 있다. 그리고 Convergence의 특성을 추가적으로 실행되는 method가 없어서 서로 다른 replica의 causal history가 동일하게 유지되는 상황에서는 결국에는 두 replica의 상태가 동등해지는 것으로 정의하고 있다. □(Globally)나 ◊(Finally)와 같은 notation은 처음으로 본 것인데 Temporal Logic에서 사용되는 notation인 것 같다.

Definition 2.2 (Eventual Consistency (EC)). Eventual delivery: An update delivered at
some correct replica is eventually delivered to all correct replicas: ∀i, j : f ∈ ci ⇒ ◊f ∈
cj .
Convergence: Correct replicas that have delivered the same updates eventually reach equivalent
state: ∀i, j : □ci = cj ⇒ ◊□si ≡ sj .
Termination: All method executions terminate.

Strong Eventual Consistency는 여기에 더해서 서로 다른 replica의 causal history가 동일하다면 동등한 상태를 가지는 특성으로 정의하고 있다. 이 paper에서는 한 replica의 causal history를 실행된 method들의 집합으로 정의하고 있으므로 – 그 method들에 대한 순서나 인과성 등의 정보는 포함하지 않고 있으므로 causal history라고 부르는 것은 조금 이상한 것 같다 – 정확히는 실행된 method들의 집합이 동일할 경우라고 보면 될 것 같다.

Definition 2.3 (Strong eventual consistency (SEC)). An object is Strongly Eventually Consistent if it is Eventually Consistent and:
Strong Convergence: Correct replicas that have delivered the same updates have equivalent state: ∀i, j : ci = cj ⇒ si ≡ sj .

State-based CRDT (CvRDT) and Operation-based CRDT (CmRDT)

CRDT를 2가지로 분류하고 있다. 첫번째는 상태들을 merge하는 method를 기반으로 정의하는, Replica들의 상태가 converge할 수 있는 형태의 CRDT인 Convergent Replicated Data Type (CvRDT)이고, 두번째는 operation들이 commutative한 성질을 기반으로 정의하는 CRDT인 Commutative Replicated Data Type (CmRDT)이다. Paper에서는 두 가지의 CRDT는 서로를 emulation할 수 있으므로 equivalent하다고 이야기 하고 있다.

우선 CvRDT의 정의를 이해하기 위해서는 lattice와 semilattice라는 수학 개념에 대해서 이해를 해야한다. Lattice란 기본적으로 어떤 ordering에 대해 정의되는 부분순서집합이며, 임의의 두 원소에 대해서 유일한 join과 meet가 그 집합 내에 존재해야한다. 여기서, join과 meet는 least upper bound와 greatest lower bound라고도 불리며, 우리에게 익숙한 개념으로 보자면, 최소공배수와 최소공약수에 해당하는 개념이다. 이 때, 최소공배수와 최소공약수가 각각 join과 meet가 되는 lattice는 배수를 ordering으로하는 정수집합에 해당한다. 정수집합의 모든 두 원소가 서로 배수관계가 있는 것은 아니지만 – 즉 부분순서집합 – 모든 두원소는 최소공배수와 최소공약수를 가진다. Semilattice란 join 또는 meet 중 하나를 가지는 부분순서집합이고, 어느 쪽이냐에 따라서 join-semilattice와 meet-semilattice로 불린다.

CvRDT에서 정의되는 객체는 객체들의 상태 집합에 대한 join-semilattice이고, merge method는 join 즉 least upper bound로 정의된다. 다시 말하면, 서로 다른 두 replica의 어떤 두 상태가 주어지더라도 이 상태를 merge한 상태가 정의되어있다고 할 수 있다. 이러한 객체가 SEC를 만족하는 것에 대한 증명은 어느 한쪽의…

Definition 2.4 (Monotonic semilattice object). A state-based object, equipped with partial order ≤, noted (S,≤, s0, q, u,m), that has the following properties, is called a monotonic semi-lattice: (i) Set S of payload values forms a semilattice ordered by ≤. (ii) Merging states with remote state s′ computes the LUB of the two states, i.e., s •m(s′) = s⊔s′. (iii) State is monotonically non-decreasing across updates, i.e., s ≤ s • u.
Theorem 2.1 (Convergent Replicated Data Type (CvRDT)). Assuming eventual delivery and termination, any state-based object that satisfies the monotonic semilattice property is SEC.

CmRDT는 말그대로 update method의 commutativity에 기초해서 상대적으로 단순하게 정의되고 있는데, 중요한 것은 concurrent update에 대해서만 commutativity 특성을 요구하고 다른 모든 update는 causal하게 전달되는 것을 가정하고 있다. 현실적으로는 어떤 경우에 어떤 update가 commutative하게 일어나고 어떤 경우에는 그렇지 않은 것은 쉽지 않기 때문에 이러한 제약을 주는 이유에 대해서 조금 의문이 들고, 특히 엄밀하게 정의되어있지 않은 CvRDT와 CmRDT의 equivalence에 대해서도 의문이 들게 만드는 부분이다.

Definition 2.6 (Commutativity). Updates (t, u) and (t′, u′) commute, iff for any reachable replica state s where both u and u′ are enabled, u (resp. u′) remains enabled in state s • u′ (resp. s • u), and s • u • u′ ≡ s • u′ • u.
Theorem 2.2 (Commutative Replicated Data Type (CmRDT)). Assuming causal delivery of updates and method termination, any op-based object that satisfies the commutativity property for all concurrent updates, and whose delivery precondition is satisfied by causal delivery, is SEC.

Example CRDTs

먼저 increment-only integer counter (G-Counter)를 위한 CRDT를 제시하고 있는데, 여러 replica는 integer vector의 각 replica에 해당하는 component를 increment하고, 두 vector의 merge operation은 각 component의 max를 취한 vector를 돌려주는 것으로 정의하고 있다. value method는 모든 component의 합에 해당하는 값을 돌려주게 된다.

increment와 decrement 둘다 가능한 integer counter (PN-Counter)를 위해서는 위에서 정의한 increment-only counter 2개를 결합하여, 하나는 increment를 기록하고 다른 하나는 decrement를 기록해서 실제 value method는 둘 사이의 차에 해당하는 값을 돌려주는 CRDT를 제시하고 있다.

여기서 설명한 G-Counter, PN-Counter를 포함해 G-Set, 2P-Set 등으로 이름지어진 여러 CRDT, 그리고 Directed Graph의 operation-based CRDT 구현 등에 대해 간략하게 설명하고 있는데, 위에서 언급한 “A comprehensive study of convergent and commutative replicated data types.”에서 훨씬 자세한 설명을 하고 있으므로 이 글에서는 생략하도록 한다. 이에 대해서 알고자 한다면 다음 페이지를 참고하면 좋을 것이라고 생각한다.

  • https://en.wikipedia.org/wiki/Conflict-free_replicated_data_type
  • https://vaughnvernon.co/?p=1012
  • http://blog.plasmaconduit.com/crdts-distributed-semilattices/
  • https://github.com/aphyr/meangirls

Closing

  • CRDT가 distributed system의 여러가지 도구들 – vector clock, log, … – 과 commutativity라는 관점에서 유사성을 공유하고 있는 것을 고려하면, CRDT의 정식화나 실제 CRDT들에 대해서 생각해보는 것은 좋은 도구가 되며 distributed system을 다루기 위한 사고 훈련이 되는 것 같다.
  • CRDT가 모든 문제를 해결해줄 수는 없으나 이미 알려진 CRDT들을 이용해서도 애플리케이션 관점에서 의미있는 이익을 얻을 수 있는 것으로 보인다. Garbage collection과 같은 실제적인 이슈들이 있으나 해결이 불가능한 것은 아니고, Riak 2.0 등에서 CRDT를 도입한 것을 볼 때 실제적인 응용을 시도해 볼 가치가 있다고 생각한다.
  • 이 논문의 저자들의 2007년, 2009년에 출판한 논문들은 CRDT라는 이름만 붙어있을 뿐이지 수학적인 정식화 등은 전혀들어있지 않다. 하지만, 이름을 붙이고 수년동안 연구를 하면서 완벽하지는 않지만 어느 정도의 엄밀함을 가진 모델을 성립시키고 하나의 분야로서 다루어질 수 있게 된 것 같다. 물론 이것만 하더라도 대단한 것이지만, A급 컨퍼런스의 A급 논문들만 보다보면 처음부터 거추장스러운 것은 하나도 없고, 논리적으로 완벽하며 또한 해당 분야에서도 중요한 의미를 지니고 있는 세상에 없을만한 것으로 보이곤 한다. 이 둘을 보면서, 평범한 과학자나 엔지니어들의 현실은 오히려 이러한 A급 논문보다 이들 논문들과 더 가깝지 않을까라는 생각이 조금 들었다.

Paper: Facebook Wormhole

josephjang Leave a comment

Sharma, Yogeshwer, et al. “Wormhole: Reliable pub-sub to support geo-replicated internet services.” NSDI, May. 2015.

Challenges to Adopting Stronger Consistency at Scale 논문을 소개할 때도 언급한 바가 있는 이 논문은 Wormhole이라는 페이스북의 스토리지 업데이트를 여러 애플리케이션에 전달하기 위한 pub-sub 시스템을 소개하고 있다.

페이스북에서 사용자가 포스트를 올리면 데이터베이스의 쓰기가 발생하게 되는데, 이러한 쓰기 오퍼레이션은 다른 사용자의 뉴스 피드를 갱신한다거나, 캐시 무효화, 인덱스 등의 내부 시스템에 사용할 필요가 있다. 이를 위한 한가지 접근은 데이터베이스의 쓰기 오퍼레이션을 전부 메시지 큐로 전달하고 이를 여러 consumer에게 전달하는 방식인데, 페이스북 웜홀은 페이스북의 기존의 스토리지 (MySQL, RocksDB, HDFS)의 데이터가 엄청나기 때문에, 이를 별도의 메시지큐에 복제하는 것을 피하고 접근을 선택해야만 했다.

wormhole_architecture

Wormhole의 전체적인 구조는 Producer, Datastore, Publisher, Subscriber로 이루어지며, Facebook의 서비스에 해당하는 Producer가 역시 서비스를 위한 Datastore (MySQL, RocksDB, HDFS)를 업데이트하면, 일반적으로 Datastore와 동일한 서버에서 동작하는 Publisher가 Datastore의 WAL을 읽어서 이를 ‘Wormhole update’라는 공통적인 포맷 (키-밸류로 이루어진 해시 + 메타데이터)으로 만들고, 이를 Subscriber들에 전달하는 방식이다.

Datastore가 sharding 되어있기 때문에, Wormhole update에는 shard에 대한 정보가 추가되어 전달된다. 아마도 이를 이용해서 Replication 등에 활용할 수 있으리라 생각된다. 하나의 Subscriber 애플리케이션도 당연히 여러 Subscriber로 sharding될 수 있는데, 하나의 Datastore shard로부터의 Flow는 하나의 Subscriber로만 전달된다. (Replication 등을 위해) 하나의 shard로부터의 순서를 보장하기 위해서는 이 요건은 필수적인 것으로 보인다. 이러한 방식 때문에 Subscriber들 사이의 조율에 대한 필요성이 줄어들기도 할텐데, 이러한 부분에서는 Kafka가 떠오르기도 한다.

Publisher는 ZooKeeper를 이용해서 Subscriber들을 관리하며, Subscriber가 마지막으로 읽고나서 acknowledgement를 보낸 로그상의 위치를 기록한다. Subscriber는 읽고 있는 위치를 특별히 상태로서 관리하지 않아도 된다. 이러한 면에서, Kafka의 경우 Consumer가 읽어야 하는 위치를 상태로서 유지하고 broker로부터 pull하는 방식 임에 반해, Wormhole은 Publisher가 Subscriber로 push하는 정반대의 방식이라고 볼 수 있다.

Kafka와 같은 시스템에서는 일정 기간 동안의 버퍼를 가지고 있기 때문에 하나의 Consumer가 지연되더라도 단순히 이전의 offset으로 pull하는 방식으로 해결하는데, Wormhole은 내부적으로 이러한 버퍼에 해당하는 스토리지를 따로 가지고 있는 것이 아니기 때문에, Publisher는 각 Subscriber의 Flow를 별도로 관리해서 서로 다른 WAL의 위치를 읽어들일 수 있는 기능을 제공한다. 이러한 기능 덕분에 하나의 Subscriber가 느려진다고 해서 다른 Subscriber가 반드시 느려지지 않을 수 있게 된다.
여기서 문제는 서로 다른 n개의 Subscriber가 Datastore의 서로 다른 위치를 읽어들이고 있다면 Datastore의 랜덤 디스크 I/O가 증가해서 Producer 즉, 서비스의 성능에도 영향을 주게 된다. Wormhole은 이러한 문제를 피하기 위해 서로 다른 속도를 가진 Subscriber들에 대해 항상 별도의 Flow가 할당되는 것이 아니라, Subscriber가 읽고 있는 위치나 속도에 따라서 Flow들을 적당히 묶어서 Caravan이라는 개념을 도입하고 있다. 이러한 수준의 최적화까지 갖추고 있는 것은 물론 페이스북의 실제적인 필요에 의한 것이겠지만, Wormhole이 상당히 성숙된 시스템임을 느끼게 한다.

Wormhole은 간단한 필터링 메커니즘을 가지고 있어서, 각각의 Subscriber 애플리케이션이 제공한 필터에 따라서 Publisher는 필터링된 업데이트만을 Subscriber에 전달한다. 필터링은 키-밸류들의 집합에 해당하는 Wormhole update에 대한 오퍼레이션들에 대한 AND, OR로 표현된다. 이 오퍼레이션들에는 키가 존재하는지, 어떤 키의 값이 특정 값인지, 어떤 키의 값이 특정 집합에 포함되는지, 이러한 오퍼레이션들의 역들이 있다. 이러한 간단한 필터링이 존재할 수 있는 것은 불투명한 데이터를 전달하는 것이 아니라 공통적인 데이터 포맷이 정의되어 때문이다. 또한, Subscriber 애플리케이션에 따라서 모든 데이터가 필요하지 않은 경우는 매우 쉽게 상상할 수 있으므로, 이러한 기능 역시 페이스북의 매우 실제적인 요구사항에 근거한 것으로 보인다.

wormhole_mcrd_architecture

쉽게 상상할 수 있듯이 Datastore는 보통 복제본을 가지고 있고, 어떤 Datastore 샤드가 실패했을 때, 해당 샤드에 대한 Wormhole의 설정을 직접 바꾸거나 마스터가 복구될 때까지 기다리는 것이 아니라, 자동적으로 복제본을 이용해 Subscriber로 데이터를 계속 전달할 수 있도록 하기 위한 메커니즘을 가지고 있는데, Wormhole 논문에서는 이를 Multiple-Copy Reliable Delivery (MCRD)라고 부르고 있다. 각 Publisher는 Zookeeper의 ephemeral node를 이용해서 fail-over에 대한 coordination을 한다.

Wormhole에서 MCRD가 매우 독특한 점은 복제본들 사이에서 WAL내의 동일한 레코드의 물리적인 위치는 서로 다르기 때문에, MCRD가 아닌 경우와 달리 물리적인 위치 정보를 기록해두어도 fail-over 시에는 아무런 의미가 없다는 점이다. 따라서, MCRD 모드에서는 논리적인 위치 – 단조증가하는 sequence number나 timestamp를 ZooKeeper에 저장해둔다. 여기서 어려운 이슈는 실제로 DataStore의 종류에 따라 이러한 논리적인 위치를 정의하고, 이러한 논리적인 위치가 주어졌을 때, 이를 물리적인 위치로 변환하는 일인데, 이를 위해 DataStore에 따라 logical positions mapper가 제공된다.

Wormhole의 운영적인 측면에서 재미있는 사실 하나를 언급하고 있는데, 1%의 Datastore에 장애가 발생해서 Wormhole publisher들이 제대로 동작하지 않게 되면, 1%의 데이터가 제대로 전달되지 않아 1%의 stale한 캐시가 발생하게 될텐데, 좀 더 자세히 들여다보면 100% 사용자의 1% 데이터가 stale하게 되는 것이 아니고, 1% 사용자의 100% 데이터가 stale하게 되는 것이기 때문에, Publisher의 신뢰성이 중요하다라고 강조를 하고 있다. 이러한 설명으로 미루어볼 때 Facebook의 대부분의 Datastore는 사용자별로 sharding이 되어있는 것을 알 수 있다. 🙂

또 하나 재미있는 점은 분산된 deployment 시스템을 이용한다는 점이다. Datastore는 장비가 추가되거나 빠지는 경우가 흔하기 때문에, Wormhole Publisher도 이에 따른 관리를 해주기 위해서 초기에는 중앙 집중적인 관리 시스템을 사용했다고 한다. 하지만, 이러한 시스템 하에서는 아주 많은 장비들에 대해 관리를 할 때 실수가 발생할 가능성이 높았기 때문에, Wormhole monitor라는 가벼운 프로그램을 동작시키고, 이 프로그램이 주기적으로 설정을 검사해서 Publisher를 실행할지 말지, 그리고 어떤 설정으로 Publisher를 실행할지 등을 정하는 분산 관리 시스템으로 바꾸었고, 훨씬 더 신뢰성있고 사용하기 쉬워졌다고 한다.

Wormhole은 페이스북과 같은 대규모 인터넷 서비스를 위한 아키텍쳐에서 스토리지에 대한 요구사항을 특별히 늘리지 않으면서도 스토리지에 저장되는 데이터를 소비해야하는 다양한 애플리케이션들에게 매우 효율적이고 신뢰성 있게 데이터를 전달해주는 시스템으로 보인다. 보다 다양한 Datastore에 대한 지원 등이 쉽지는 않을 수도 있다고 생각하지만, 페이스북 내에서는 충분한 수준이고, Wormhole의 디자인에서 엿볼 수 있는 여러가지 실용적인 선택들은 다른 시스템에도 여러가지로 적용해볼만 하다고 생각한다.

Paper: Paxos Made Live – An Engineering Perspective

josephjang Leave a comment

Chandra, Tushar D., Robert Griesemer, and Joshua Redstone. “Paxos made live: an engineering perspective.” Proceedings of the twenty-sixth annual ACM symposium on Principles of distributed computing. ACM, 2007.

이 논문은 Paxos를 실제로 구현하고자 할 때 고려해야할 현실적인 문제들과 해결방식을 설명하고 있다. Google은 Chubby에서 필요로하는 분산 로그 스토리지를 위해 기존의 상용 솔루션을 대체하는 Paxos를 구현하게 되었다고 한다.

single_chubby_system

Paxos를 이용한 분산 로그 스토리지의 기본적인 아이디어는 데이터베이스 오퍼레이션에 대해서 Paxos 알고리즘을 반복적으로 적용하면, 복제본들 사이에 데이터베이스 오퍼레이션들의 로그를 동일하게 쌓아올릴 수 있고, 결과적으로 분산 로그 스토리지를 구현할 수 있게 된다. Chubby가 기반하고 있는 분산 DB는 Paxos로 구현된 분산 로그 스토리지에 저장된 데이터오퍼레이션들의 리플레이 로그와, 스냅샷을 관리한다. 흥미로운 점은 분산 로그 스토리지는 다른 분산 시스템을 만들 때에도 강력한 기초가 될 수 있기 때문에, 분산 로그와 분산 DB 레이어 사이를 깔끔한 인터페이스로 정의해서 분산 로그를 재사용할 수 있도록 모듈화를 꾀한 점이다.

이 논문에서는 Paxos 알고리즘에 대한 괜찮은 설명이 나오는데, Multi-Paxos에 대한 언급도 나온다. 그동안 Paxos 알고리즘의 최적화에 해당하는 Multi-Paxos라는 단어를 도입한 것이 이 논문이라고 생각하고 있었는데, 이번에 블로그 글을 정리하다가 다시 찾아보니, 애초에 Paxos의 오리지널 논문인 Lamport의 “Part-time Parliament”에서 Multi-decree parliament라는 알고리즘으로 소개되고, 이를 Prisco, Lampson, Lynch의 “Revisiting the Paxos algorithm”에서 처음으로 “Multi-Paxos”라는 이름을 사용한 것으로 보인다.

Handling disk corruption

디스크 손상에 의한 두가지 상황, 즉 파일의 내용이 바뀌었거나 파일 자체가 없어지는 경우에 대한 탐지 방법을 제시하고 있다. 전자는 각 파일에 대한 checksum을 유지하는 방법으로 해결하고, 후자는 리플리카였음을 나타내는 marker를 GFS에 보관하는 방법으로 디스크 손상을 탐지한다. (이러한 방법과 관련해 역시 유명한 Viewstamped Replication 논문을 인용하고 있으므로 나중에 꼭 읽어보도록 하자.)

디스크 손상이 탐지되었을 경우에는 리빌드를 진행하게 되는데, non-voting member 즉, promise나 acknowledgement 메시지에는 응답하지 않는 노드로서 Paxos에 참가하고 catch-up mechanism을 사용한다. catch-up mechanism 자체에 대해서는 이 논문에는 그리 자세한 설명은 없다.

이렇게 디스크 손상에 대비하는 메커니즘이 있다면 모든 쓰기를 디스크에 바로 flush하지 않아도 되는 최적화가 가능해진다.

Master leases

stale data를 읽지 않도록 보장하기 위해서는, 읽기 오퍼레이션 조차도 Paxos로 실행해서 다른 업데이트들과 직렬화되도록 하는 것이 필요하다. 일반적으로 읽기 오퍼레이션이 대부분의 오퍼레이션을 차지하므로 이는 매우 비효율적이다.

이에 대한 해결책으로 제시하는 것이 마스터 리스이고, 이는 마스터가 리스를 가지고 있는 동안에는 다른 복제본이 Paxos로 업데이트를 진행할 수 없는 것을 보장하기 때문에, 읽기 오퍼레이션을 로컬에서 제공할 수 있게 된다. 그리고, 리스가 만료되기 전에 보통 갱신되기 때문에 마스터가 대체로 항상 리스를 가지고 있게 된다.

Epoch numbers

어떤 복제본으로 업데이트 요청이 왔을 때 마스터의 지위를 잃어버리거나 다시 얻었다면, 그 요청은 중단되어야 하는데, 이를 구분하기 위해 어떤 마스터가 연속적으로 재임하는 동안에는 같은 값으로 유지되는 Epoch number라는 개념을 도입하고 있다. 아마도 Raft 알고리즘의 term과 동일한 개념이 아닐까 싶다.

Group membership

그룹 멤버쉽 문제란 복제본이 추가되거나 제거되는 등 복제본의 집합에 일어난 변화를 어떻게 다루는지에 대한 문제다. 그룹 멤버쉽 문제에도 Paxos를 활용해 해결할 수가 있다는 아이디어를 인용하고 있는데, 아이디어 자체는 단순하나 – 복제본 집합의 변화의 로그를 Paxos로 유지하면 될 것이다 – 디스크 손상 등의 문제들을 고려해서 구현하는 것은 쉽지 않다고 얘기하고 있다.

Snapshots

로그를 무한정 쌓을 수 없으므로 스냅샷을 생성해서 기존의 로그가 필요하지 않게 만드는 메커니즘을 필요로 한다. Paxos 자체는 복제된 로그의 일관성에만 관심이 있지, 복제되고 있는 데이터 구조 자체에 대해서는 인지하지 못하므로, 스냅샷을 생성하는 것은 애플리케이션의 책임이 된다. 따라서, 애플리케이션은 자유롭게 스냅샷을 생성하고, 스냅샷이 생성된 것에 대해서 Paxos 프레임워크 쪽에 알려주어, Paxos 프레임워크가 스냅샷 이전의 로그를 삭제할 수 있도록 한다. 스냅샷은 복제본 사이에 동기화되지 않으며, 각각의 복제본은 독립적으로 스냅샷을 생성할 시점을 결정한다.

이러한 메커니즘은 간단해보이지만, 로그와 스냅샷이 서로 일관성있게 유지되도록 하기 위한 복잡성들이 존재한다.

  • 각각의 스냅샷에 관련된 모든 Paxos 관련 정보 – 스냅샷을 생성하는 시점의 Paxos instance number와 그 시점의 그룹 멤버쉽 – 를 담고 있는 스냅샷 핸들이라는 개념이 존재한다. 스냅샷을 복구할 때는, 애플리케이션은 스냅샷 핸들을 Paxos 프레임워크 쪽에 전달하고, Paxos 프레임워크 이를 이용해 복구를 진행한다.
  • 클라이언트가 스냅샷을 생성하려고 할 때는 먼저 스냅샷 핸들을 요청한다. 클라이언트는 스냅샷을 생성하는데, 이 스냅샷은 스냅샷 핸들의 정보와 일관되어야 하므로 Paxos가 진행됨에 따라서 업데이트라 발생하는 것에 대해서 주의를 기울여야 한다. 마지막으로 스냅샷이 생성되었다면 클라이언트는 Paxos 프레임워크에 스냅샷 핸들을 전달하면서 스냅샷이 생성되었음을 알린다. 스냅샷의 생성에 실패했다면 단순히 Paxos 프레임워크에 이를 알리지 않는 것으로 충분하다.
  • catch-up을 위해서는 다른 복제본으로부터 최근의 스냅샷을 얻어오고, 나머지 로그 레코드들도 다른 복제본으로부터 얻어온다.

Database transactions

CAS (compare and swap) 오퍼레이션을 atomic하게 유지하기 위해서 CAS에 관련된 데이터를 Paxos 상에서의 하나의 값으로 취급한다. 실제로 데이터베이스 트랜잭션을 구현하지 않더라도, 이러한 방식을 확장함으로써 트랜잭션 스타일의 지원이 가능하다.
MultiOp은 이러한 트랜잭션 지원을 위해 atomic하게 적용되고 guard, top, fop의 세가지 부분으로 이루어져 있다. guard라고 불리는 테스트의 리스트가 한가지 부분이고, 이 모든 테스트가 참이라면 데이터베이스 오퍼레이션의 리스트인 top을 실행하고, 그렇지 않다면 역시 데이터베이스 오퍼레이션의 리스트인 fop을 실행한다.

Software Engineering

알고리즘의 표현, 실행시간 일관성 검증, 테스팅, 동시성 등에 관련한 이슈들을 설명하고 있는데, 특별히 흥미로운 테스팅 항목에 대해 살펴보자.

시스템에 랜덤한 실패들을 집어넣더라도 Safety와 Liveness를 만족하는지를 테스트하는데, 테스트가 실패했을 때 이를 재현할 수 있어야 하므로, 테스트가 시작할 때 설정된 시드를 확보해서 동일한 테스트를 재현할 수 있도록 하고 있다.

이외에도 Chubby 시스템이 하위 시스템의 실패에도 잘 동작하는지를 테스트하기 위해서, 분산 로그 시스템에 여러 hook을 구현하고 역시 랜덤하게 실패하도록 했을 때 Chubby 시스템이 그러한 실패들을 잘 극복하는지를 테스트 하기도 한다.

이러한 테스트 방식은 체계적인 방식은 아니나, 최근 분산시스템을 테스트하는 것에 대해서는 일반적인 방법으로 자리잡아 나가고 있는 것 같다.

Summary

이 논문의 마지막 부분은 컴파일러 개발 분야가 매우 복잡하지만, 이론이 널리 알려져있고 yacc, ANTLR 등의 도구와 같이 현실에도 잘 적용되어있는 점을 지적하면서, 분산시스템 분야는 그렇지 않음을 지적하고 있다.

  • There are significant gaps between the description of the Paxos algorithm and the needs of a real-world system. In order to build a real-world system, an expert needs to use numerous ideas scattered in the literature and make several relatively small protocol extensions. The cumulative effort will be substantial and the final system will be based on an unproven protocol.
  • The fault-tolerance computing community has not developed the tools to make it easy to implement their algorithms.
  • The fault-tolerance computing community has not paid enough attention to testing, a key ingredient for building fault-tolerant systems.

References

Paxos는 Paxos made simple을 통해서 공부했었는데, 시간이 날 때 오리지널 페이퍼인 Lamport, Leslie. "The part-time parliament." ACM Transactions on Computer Systems (TOCS) 16.2 (1998): 133-169.를 읽어보는 것이 좋을 듯 하다.

리스에 관해서는 Gray, Cary, and David Cheriton. Leases: An efficient fault-tolerant mechanism for distributed file cache consistency. Vol. 23. No. 5. ACM, 1989.을 참고하자.

그룹 멤버쉽 문제에 관해서 Cristian, Flaviu. "Reaching agreement on processor-group membership in synchronous distributed systems." Distributed Computing 4.4 (1991): 175-187.

Paper: Dotted version vectors: Logical clocks for optimistic replication (Part 2)

josephjang Leave a comment

Paper: Dotted version vectors: Logical clocks for optimistic replication (Part 1)

A Kernel for Eventual Consistency

인과성을 이용하는 분산 스토리지의 동작에 있어서 논리적 시계 집합에 대한 sync 와 update 2개의 오퍼레이션이 핵심을 이루고 있다고 주장하고 있다.

먼저 sync 오퍼레이션의 경우에는 두 개의 시계 집합을 취해서 두 집합의 원소들인 논리적 시계들 사이에 인과성의 관계가 있다면 이전에 해당하는 시계를 모두 버리고, 남아있는 원소들로 구성된 집합을 반환하는 오퍼레이션이다. 결과적으로 반환되는 집합은 동시적 (concurrent)인 관계에 있는 시계들로만 이루어지게 된다.

sync는 클라이언트와 서버 도는 서버의 노드들 사이의 동기화가 필요한 시점에 논리적 시계에 기반해서 과거의 값들을 버리기 위한 오퍼레이션이라고 볼 수 있다. 여기서 재미있는 것은 sync는 논리적 시계가 실제로 어떻게 구현되어있는지에 상관없이 시계들 사이의 부분순서 (partial order)만을 이용해서 일반적으로 정의할 수 있다는 것이다.

sync_defined_by_partial_order

 

update 오퍼레이션은 어떤 시계 집합 (통상적으로 클라이언트)과 서버의 어떤 노드의 시계 집합, 서버의 식별자를 취하고 하나의 시계를 반환하는 오퍼레이션이다. 이 시계는 클라이언트 시계 집합 내의 모든 시계들을 dominate하고, 시스템 내의 시계들의 어떤 join에 의해서도 dominate되지 않아야 한다. (즉, dominate하거나 concurrent 해야한다.)

인과적인 이력 (causal histories)의 경우, update 오퍼레이션은 다음과 같이 정의할 수 있다. 시스템 전체에서 고유한 사건 식별자를 얻어서 클라이언트 시계 집합의 각 시계에 추가하는 방식이다.

update_operation_of_causal_histories

이어서 분산 스토리지의 get/put 오퍼레이션에서 위에서 정의한 sync/update 오퍼레이션을 이용해서 논리적 시계를 어떻게 다루는지에 대해서 설명하고 있으나 여기서는 생략하기로 하자.

Dotted Version Vectors

버전 벡터가 (id, m)의 형태로 표기된다면 dotted version vector는 (id, m, n)과 같이 표기할 수 있다. 버전 벡터가 연속적인 인과적인 이력을 표현하고 있다면, dotted version vector는 그 연속적인 이력에 n에 해당하는 독립적인 사건을 추가한 것을 표현할 수 있다.

dotted_version_vector_definition

 

예를 들어, {(a,2),(b,1),(c,3,7)}이라는 dotted version vector는 {a1, a2, b1, c1, c2, c3, c7}와 같은 인과적인 이력을 표현하고 있다.

dotted version vector를 인과적인 이력으로 정의했다면, 부분순서가 어떻게 정의되는 지를 살펴볼 때다.

dotted_version_vector_partial_order

 

 

dotted version vector를 인과적인 이력으로 변환해서 생각하면 당연한 결과라고 할 수 있다.

부분순서가 정의되어있으므로 sync 오퍼레이션은 추가적으로 정의할 필요가 없으나, update 오퍼레이션은 dotted version vector에 대해 정의할 필요가 있다.

dotted_version_vector_update_function

 

이 간결한 식에 매우 많은 의미를 담고 있는데, 합집합(union)의 좌항의 경우, 클라이언트의 시계집합(S)에 속하는 노드의 식별자들 중 파라미터에서 제공한 노드의 식별자 (r)가 아닌 것들에 대해서,  각각의 식별자와 클라이언트 시계집합에서 해당 식별자에 대한 가장 큰 sequence를 pair로 하는 버전 벡터들을 나타낸다. 우항의 경우에는 파라미터에서 제공한 노드의 식별자(r)에 대한 dotted version vector를 구성하고 있는데, 좌항의 경우와 유사하게 첫번째 정수는 클라이언트의 시계집합에서 해당 식별자에 대한 가장 큰 sequence가 된다. 양쪽 모두 클라이언트의 문맥을 표현하는 것이라고 볼 수 있다. 두번째 정수는 조금 특이한데, 파라미터에서 제공한 노드의 시계 집합에서 역시 해당 노드의 식별자에 대한 가장 큰 sequence를 얻은 후 1만큼 증가시켜준 값으로 설정된다. 이는 노드 상의 문맥에서 업데이트로 인한 새로운 사건을 기록한 것이라고 볼 수 있다.

이러한 update 오퍼레이션을 그대로 적용하면 다음과 같이 아름답게 움직이는 시스템이 된다.

dotted_version_vector_operations

 

논문에서는 dotted version vector에 대한 correctness에 대해서 설명하고 있는 듯 하나 생략하도록 한다. 이 논문을 여러번 다시 읽어보았지만 dotted version vector가 어떻게 문제를 해결하는지에 대한 직관적인 설명을 하는 것이 아직도 어려운 것 같다.

dotted_version_vector_reason_of_correctness

Related Work

  • Lamport clock
    • L. Lamport, “Time, clocks and the ordering of events in a distributed system,” Communications of the ACM, vol. 21, no. 7, pp. 558–565, Jul. 1978.
      • http://blog.lastmind.net/archives/720
  • Version vector
    • D. S. Parker, et al., “Detection of mutual inconsistency in distributed systems,” Transactions on Software Engineering, vol. 9, no. 3, pp. 240–246, 1983.
  • Vector clock
    • C. Fidge, “Timestamps in message-passing systems that preserve the partial ordering,” in 11th Australian Computer Science Conference, 1989, pp. 55–66.
      • http://blog.lastmind.net/archives/736
    • F. Mattern, “Virtual time and global clocks in distributed systems,” in Workshop on Parallel and Distributed Algorithms, 1989, pp. 215–226.
  • Dynamic creation and retirement of vector entries
    • R. A. Golding, “A weak-consistency architecture for distributed information services,” Computing Systems, vol. 5, pp. 5–4, 1992.
    • K. Petersen, et al., “Flexible update propagation for weakly consistent replication,” in Sixteen ACM Symposium on Operating Systems Principles, Saint Malo, France, Oct. 1997.
    • P. S. Almeida, et al., “Interval tree clocks,” in Proceedings of the 12th International
      Conference on Principles of Distributed Systems, ser. OPODIS ’08. Berlin, Heidelberg: Springer-Verlag, 2008, pp. 259–274.
  • Scalability problems
    • B. Charron-Bost, “Concerning the size of logical clocks in distributed systems,” Information Processing Letters, vol. 39, pp. 11–16, 1991.
    • D. H. Ratner, “Roam: A scalable replication system for mobile and distributed computing,” Ph.D. dissertation, 1998, uCLA-CSD-970044.
    • R. Prakash and M. Singhal, “Dependency sequences and hierarchical clocks: Efficient alternatives to vector clocks for mobile computing systems,” Wireless Networks, pp. 349–360, 1997, also presented in Mobicom96.
    • P. Mahajan, S. Setty, S. Lee, A. Clement, L. Alvisi, M. Dahlin, and M. Walfish, “Depot: Cloud storage with minimal trust,” in OSDI 2010, Oct. 2010.
    • D. Malkhi and D. B. Terry, “Concise version vectors in winfs,” in DISC, ser. Lecture Notes in Computer Science, P. Fraigniaud, Ed., vol. 3724. Springer, 2005, pp. 339–353.
    • V. Ramasubramanian, et al., “Cimbiosys: a platform for contentbased partial replication,” in Proceedings of the 6th USENIX symposium on Networked systems design and implementation. Berkeley, CA, USA: USENIX Association, 2009, pp. 261–276.
    • F. J. Torres-Rojas and M. Ahamad, “Plausible clocks: constant size logical clocks for distributed systems,” Distributed Computing, vol. 12, no. 4, pp. 179–196,
      1999.
  • Trade-off
    • B. B. Kang, R. Wilensky, and J. Kubiatowicz, “The hash history approach for reconciling mutual inconsistency,” in Proceedings of the 23nd International Conference
      on Distributed Computing Systems (ICDCS). IEEE Computer Society, 2003, pp. 670–677.

 

Paper: Challenges to Adopting Stronger Consistency at Scale

josephjang 1 Comment

AJOUX, Phillipe, et al. Challenges to Adopting Stronger Consistency at Scale. In: 15th Workshop on Hot Topics in Operating Systems (HotOS XV). USENIX Association. (pdf, slides)

최근 geo-replicated data store에서 높은 consistency를 제공하기 위한 연구들이 많이 있다고 한다. – 이 논문에서 나열하고 있는 그런 논문들의 목록도 흥미로운데 이는 마지막에 한번 정리해보자. – Facebook에서도 사용자들에게 – 특히, 읽기에 대한 – end-to-end consistency를 제공하기 위해서 이러한 연구 결과들을 도입하고 싶으나, 여러가지 이유로 이것이 쉽지가 않은데, 그 이유가 무엇인지 일단 알아보자…라는 것이 이 논문의 취지.

There have been many recent advances in distributed systems that provide stronger semantics for geo-replicated data stores like those underlying Facebook. These research systems provide a range of consistency models and transactional abilities while demonstrating good performance and scalability on experimental workloads. At Facebook we are excited by these lines of research, but fundamental and operational challenges currently make it infeasible to incorporate these advances into deployed systems. This paper describes some of these challenges with the hope that future advances will address them.

Facebook은 확장성과 효율을 위해서 샤딩, 복제, 캐싱 등을 아주 많이 활용하고 있는 것으로 보인다. Sodcial graph 시스템을 예로 들고 있는데, 하나의 node나 edge가 MySQL(!)에 쓰여지면, 이것이 비동기적으로 다른 데이터 센터로 복제되고, 여러 계층의 캐시들을 업데이트하고, pub-sub 시스템에 의해 검색이나 뉴스피드를 위한 다른 시스템으로 전달된다. 당연한 이야기지만, Facebook 내의 여러 기능에 해당하는 시스템들은 효율을 위해 각자 특별한 샤딩 방식이나 캐시, 인덱스들을 유지해야한다. Twitter (http://blog.lastmind.net/archives/664)나 LinkedIn (http://blog.lastmind.net/archives/657)의 시스템과 같이 Primary Data Store (DB)로부터 비동기적인 복제를 통해 Secondary Index를 생성하거나 다른 서비스로 이벤트를 전달하는 방식은 현재의 대규모 인터넷 서비스에서는 매우 공통적인 부분으로 보인다.

Facebook relies on sharding, data replication, and caching to efficiently operate at scale. When a node or edge is added to the social graph, it is first written to a single MySQL instance. From there it is asynchronously replicated to all of our data centers, updated or invalidated in multiple layers of cache, and delivered through a scalable publisher-subscriber system to other services such as Search and News Feed. No single data placement strategy can efficiently serve all workloads in a heavily sharded system, so many of these services choose a specialized sharding function and maintain their own data store, caches, and indexes.

facebook_data_layout

Fundamental Challenges

Integrating Across Stateful Services

Facebook의 Social graph는 여러 서비스들을 위한 시스템에 각 서비스를 위한 형태로 복제되고 있다고 한다. 문제는, 대부분의 연구 결과들에서 나타나는 설계들은 단일한 서비스를 상정하고 있다는 것이다. Facebook는 그러한 여러한 서비스가 따로 사용자에게 제공되기 보다는 하나의 서비스처럼 제공되므로 여러가지 문제들이 발생하게 된다.

Facebook’s architecture of cooperating heterogeneous services is different from the monolithic service that
most research designs assume. This difference alone is not fundamental—Facebook can be externally viewed as a single service even though internally it is comprised of many services.

Storing Data Consistently Across Services

Facebook의 각 서비스들은 Social graph의 일부에 대한 캐시, 인덱스, 복제본 등을 가지고 있다. Wormhole이라는 Facebook의 pub-sub 시스템은 업데이트를 비동기적으로 여러서비스들에 전달해주고 있다.

In a sharded and scaled environment like Facebook’s, services may maintain their own optimized cache, index, or copy of a portion of the social graph. Wormhole [40] is our scalable pub-sub system that asynchronously delivers update notifications to all of the services that manage state.

이 때, 만약 높은 consistency를 제공하기 위해서는 동기적으로 업데이트 통지를 전달할텐데, availability와 latency를 떨어뜨리는 요인이 된다. 인과성을 위한 시스템을 사용하더라도 여러 서비스들 사이의 의존관계를 표현하는 것은 굉장한 복잡도를 가져오게 된다.

To provide strong consistency across our services, we would need to propagate update notifications synchronously impacting availability and latency. A causal system could avoid atomic updates, but would still need to be integrated into every service and threaded through every communication channel greatly increasing the complexity of the dependency graph.

Decentralized Access to Services

클라이언트는 한 화면에서 표시되는 여러 컨텐츠를 별도의 리퀘스트를 통해 가져와서 클라이언트에서 통합한다. 이 때문에 클라이언트는 일관되지 않은 결과들을 가지고 session-based guarantee를 보장함으로써 사용자에게 일관된 상태를 보여주어야 한다.

The Facebook web site and its mobile applications use parallelism and pipelining to reduce latency. The notification bar and the central content are fetched by separate requests, for example, and then merged at the client. Mobile applications also aggressively cache results using local storage, and then merge query results with previously seen values. […] but it leaves the user’s device as the only safe location to define session-based guarantees or demarcate isolation boundaries.

단일한 사용자는 보통 동일한 지역의 동일한 클러스터로 라우팅 되므로 이러한 성질은 사용자가 inconsistency를 경험할 확률을 낮추어준다. 하지만, 그러한 성질이 보장되는 것은 아니므로 역시 문제는 있다.

Requests for a single user are usually routed to the same cluster in the same region. This routing stability assists in reducing the number of user-visible inconsistencies, because most of the variance in asynchronous update propagation is inter-region.

Composing Results

조금 설명이 애매하기는 한데, 여러 서비스로부터의 결과를 조합해야할 경우, 그 사이의 의존관계를 다루어야 하는데, 각 서비스는 각자의 서비스를 위한 일부의 결과를 가지고 있기 때문에, 이를 일반적으로 처리하기가 쉽지 않다는 의미인 듯 하다.

Each service can be thought of as implementing its own limited query language, so it is not clear that the dependencies can be captured in a way that is both sufficiently generic and sufficiently precise.

Query Amplification

사용자 입장에서는 하나의 쿼리지만, 내부적으로는 여러 서비스들에 대한 수천개의 쿼리로 실행될 수 있다는 이야기.

For example, a user’s home page queries News Feed to get a list of stories, which depends on a separate service to ensure story order of previously seen content. These stories are then fetched from TAO to get the content, list of comments, likes, etc. In practice, this behavior produces fork/join query patterns that have fan-out degrees in the hundreds and critical paths dozens deep.

Slowdown Cascades

동일한 데이터가 각각의 서비스에서는 서로 다른 샤딩 방식과 인덱싱을 통해 저장되므로, 서로 다른 서비스들의 샤드 사이에 all-to-all ordering 의존 관계가 발생하고, 높은 consistency를 요구하는 시스템에서는 하나의 서비스에서 하나의 샤드만 실패 또는 지연되더라도 이에 의존하는 모든 다른 서비스의 모든 샤드에 영향을 줄 수 있다는 이야기.

Different services use different sharding and indexing schemes for the same data. For instance, TAO shards data based on a 64-bit integer id, News Feed shards based on a user id, and the secondary index system shards based on values. The use of different sharding schemes creates all-to-all ordering dependencies between shards in different services. This connectedness accelerates slowdown propagation in strongly consistent systems. A failure in one shard of one service would propagate to all shards of a downstream service.

Latency Outliers

여러 쿼리들을 조합해야하는 경우, 하나의 쿼리만 느리더라도 최종적인 응답에 영향을 미칠 수 있음. 이것이 strong consistency와 직접적인 연관성이 있는지는 명확하게 이해하기 어려우나, strongly consistent system의 응답속도나 outlier들이 그렇지 않은 시스템들에 비해 더 높기 때문에 언급되는 것 같음.

Parallel subqueries amplify the impact of latency outliers. A user request does not return until it has joined the results of all of its queries, and thus must wait for the slowest of its internal requests.

Linchpin Objects

데이터의 개수도 많고 매우 자주 읽히는데다 쓰기도 많은 무언가를 Linchpin Object라고 부르고 있다. 흔히 Facebook의 연예인 페이지나 유명 대기업의 페이지를 상상하면 될 것 같다. 역시 좋은 성능을 제공하는 것이 strongly consistent system에서는 중요하다고 얘기하고 있다.

The most frequently read objects are often also frequently written to and highly connected. Examples of these linchpin objects include popular users like celebrities, popular pages like major brands, and popular locations like tourist attractions. […] A system that provides stronger consistency must provide good performance for these linchpin objects.

Net Benefit to Users

stronger consistency는 프로그래머가 프로그램의 정확성에 대해 생각하기 편리하고, 사용자에게 일관성을 제공한다는 측면에서 이점을 가지고 있지만, 이를 정량적으로 측정하기는 어려울 뿐만 아니라, 그러한 이점들이 단점들 – 커뮤니케이션 오버헤드, 무거운 상태관리, 응답속도의 증가 – 을 넘어서지 않을지도 모름.

Systems with strong consistency are easier for programmers to reason about and build on top of [6, 12] and they provide a better experience for users because they avoid some anomalous application behavior. However, these benefits are hard to quantify precisely, and they do not appear in isolation. When all aspects of user experience are considered it might even be the case that the benefit does not outweigh the detriment. […] Stronger properties are provided through added communication and heavier-weight state management mechanisms, increasing latency. Although optimizations may minimize these overheads during failure-free execution, failures are frequent at scale. Higher latency may lead to a worse user experience, potentially resulting in a net detriment.

Operational Challenges

Fully Characterizing Worst-Case Throughput

여러 서비스가 얽혀있는 복잡한 시스템에서 실패나 복구 방법에 대해 미리 예측하거나 관찰하기 쉽지 않을 뿐더러 이를 정형화하기도 쉽지 않다는 얘기.

While there is no theoretical impediment to preserving worst-case throughput despite strengthening Facebook’s consistency guarantees, our experience is that failure and recovery behavior of a complex system is very difficult to characterize fully. Emergent behavior in cross-service interactions is difficult to find ahead of time, and may even be difficult to identify when it is occurring

Polyglot Environment

(아마도 백엔드) 서비스 내에서 C++이 가장 보편적으로 사용되는 언어인 것은 매우 신기함.

While C++ is the predominant language for standalone services at Facebook, Python, Java, Haskell, D, Ruby and Go are also supported by Thrift, our inter-service serialization format. The application code coordinating service invocations is generally written in Hack and PHP, and final composition of query
results may be performed by JavaScript in the browser, Objective C/C++ on IOS, or Java on Android.

여러 언어를 사용하고 있기 때문에 인터페이스나 쓰레딩 모델 등에 대해 가정하는 것이 어렵다는 이야기.

Ports or foreign function interfaces must be provided to give all of these codebases access to an end-to-end
consistency platform. Perhaps the trickiest part of this multi-language support is the wide variety of threading models that are idiomatic for different languages and runtimes, because it is likely the underlying consistency system will need to initiate or delay communication.

Varying Deployment Schedules

낮은 레벨의 서비스들은 매우 보수적인 deployment 프로세스를 가지고 있으므로, 이에 관련한 시스템을 빠르게 개발하는 것은 쉽지 않을거라는 이야기.

Facebook has an extremely aggressive deployment process for the stateless application logic, but we are necessarily more conservative with stateful and mature low level services. An inter-service coordination mechanism that is used by or interposes on these low-level services will have a deployment velocity that is constrained by the release engineering requirements of the most conservative component.

Reduced Incremental Benefit in a Mature System

consistency에 관해 이미 적용된 workaround들이 있으므로, 이를 일반적으로 개선하는 시스템의 이득이 떨어진다는 이야기.

While these workarounds would not exist in a newly built system, their presence in Facebook reduces the incremental benefits of deploying a generic system for stronger consistency guarantees.

Related Work

Prior Facebook Publications

  • memcache cache
    • B. Atikoglu, et al. Workload analysis of a large-scale keyvalue store. In SIGMETRICS, 2012.
    • R. Nishtala, et al. Scaling memcache at facebook. In NSDI, 2013.
  • TAO graph store
    • N. Bronson, et al. Tao: Facebook’s distributed data store for the social graph. In USENIX ATC, 2013.
  • Wormhole pub-sub system
    • Y. Sharma, et al. Wormhole: Reliable pub-sub to support geo-replicated internet services. In NSDI, May 2015
  • a characterization of load imbalance in the caches
    • Q. Huang, et al. Characterizing load imbalance in real-world networked caches. In HotNets, 2014.
  • an analysis of the messages use case
    • T. Harter, et al. Analysis of hdfs under hbase: A facebook messages case study. In FAST, 2014.
  • understanding and improving photo/BLOB storage and delivery
    • D. Beaver, et al. Finding a needle in haystack: Facebook’s photo storage. In OSDI, 2010.
      Q. Huang, et al. An Analysis of Facebook Photo Caching. In SOSP. ACM, 2013.
    • S. Muralidhar, et al. f4: Facebook’s warm blob storage system. In OSDI, 2014.
    • L. Tang, et al. RIPQ: Advanced Photo Caching on Flash For Facebook. In FAST, Feb. 2015.

Systems with Stronger Semantics

  • replicated state machines
    • L. Lamport. Time, clocks, and the ordering of events in a distributed system. Comm. ACM, 21(7), 1978.
    • F. B. Schneider. Implementing fault-tolerant services using the state machine approach: a tutorial. ACM Computer Surveys, 22(4), Dec. 1990.
  • causal and atomic broadcasts
    • K. P. Birman and R. V. Renesse. Reliable Distributed Computing with the ISIS Toolkit. IEEE Comp. Soc. Press, 1994.
  • systems designed for high availability and stronger consistency
    • K. Petersen, et al. Flexible update propagation for weakly consistent replication. In SOSP, Oct. 1997.
  • scalable causal consistency for datacenter-scale and/or geo-replicated services
    • S. Almeida, et al. Chainreaction: a causal+ consistent datastore based on chain replication. In EuroSys, 2013.
    • P. Bailis, et al. Bolton causal consistency. In SIGMOD, 2013.
    • J. Du, et al. Orbe: Scalable causal consistency using dependency matrices and physical clocks. In SOCC, 2013.
    • J. Du, et al. Gentlerain: Cheap and scalable causal consistency with physical clocks. In SOCC, 2014.
    • W. Lloyd, et al. Don’t settle for eventual: Scalable causal consistency for wide-area storage with COPS. In SOSP, Oct. 2011.
    • W. Lloyd,  et al. Stronger semantics for low-latency geo-replicated storage. In NSDI, Apr. 2013.
  • strong consistency for datacenter-scale and/or geo-replicated services
    • J. C. Corbett, et al. Spanner: Google’s globally distributed database. In OSDI, Oct. 2012.
    • R. Escriva, et al. HyperKV: A distributed, searchable key-value store for cloud computing. In SIGCOMM, 2012.
    • L. Glendenning, et al. Scalable consistency in Scatter. In SOSP, Oct. 2011.
    • C. Li, D. Porto, A. Clement, J. Gehrke, N. Preguic¸a, and R. Rodrigues. Making geo-replicated systems fast as possible, consistent when necessary. In OSDI, Oct. 2012.
    • J. Ousterhout, P. Agrawal, D. Erickson, C. Kozyrakis, J. Leverich, D. Mazi`eres, S. Mitra, A. Narayanan,
      M. Rosenblum, S. M. Rumble, E. Stratmann, and R. Stutsman. The case for ramcloud. ACM SIGOPS Operating Systems Review, 43(4), 2010.
    • D. B. Terry, V. Prabhakaran, R. Kotla, M. Balakrishnan, M. K. Aguilera, and H. Abu-Libdeh. Consistency-based service level agreements for cloud storage. In SOSP, 2013.
  • transactions for datacenter-scale and/or geo-replicated services
    • P. Bailis, et al. Scalable atomic visibility with ramp transactions. In SIGMOD, 2014.
    • J. Baker, et al. Megastore: Providing scalable, highly available storage for interactive services. In CIDR, Jan. 2011.
    • J. C. Corbett, et al. Spanner: Google’s globally distributed database. In OSDI, Oct. 2012.
    • T. Kraska, et al. Mdcc: Multi-data center consistency. In EuroSys, 2013.
    • W. Lloyd, et al. Stronger semantics for low-latency geo-replicated storage. In NSDI, Apr. 2013.
    • S. Mu, et al. Extracting more concurrency from distributed transactions. In OSDI, 2014.
    • Y. Sovran, et al. Transactional storage for geo-replicated systems. In SOSP, Oct. 2011.
    • A. Thomson, et al. Calvin: fast distributed transactions for partitioned database systems. In SIGMOD, May 2012.
    • C. Xie, et al. Salt: combining acid and base in a distributed database. In OSDI, 2014.
    • Y. Zhang, et al. Transaction chains: achieving serializability with low latency in geo-distributed storage systems. In SOSP, pages 276–291. ACM, 2013.

Discussions of Challenges to Stronger Consistency

  • criticism of enforcing an order in a communication substrate instead of end to end
    • D. R. Cheriton, et al. Understanding the limitations of causally and totally ordered communication. In
      SOSP, Dec. 1993.
  • slowdown cascades to motivate enforcing only an explicit subset of causal consistency
    • P. Bailis, et al. The potential dangers of causal consistency and an explicit solution. In SOCC, 2012.
  • detail the importance and challenges for tail latency in high-scale services at Google
    • J. Dean, et al. The tail at scale. Comm. ACM, 56(2), 2013.

My Opinion

Strongly consistent system의 연구자들에게 방향을 제시하기 위해 Facebook의 시스템이라는 도메인 내에서 발생하는 숙제들을 정의하고 있는 논문이라고 볼 수 있는데, 마치 Facebook에서 strongly consistent system을 일반적으로 도입해서는 안되는 이유들을 정리해놓은 듯한 느낌이 들었다. 오히려 나의 관심은 Facebook에서 데이터의 복제에 관한 시스템과 복제본 사이의 consistency 문제를 어떻게 다루고 있는가 였는데, 우리가 고심하고 있는 문제들이나 해결 방향이 상당히 닮아 있다는 인상을 받았다. 여러 서비스에 걸친 업데이트 통지를 위해 인프라로서 pub-sub을 적극적으로 활용하고 있는 점이나, 서비스별로 소셜 그래프의 별도 인덱스나 캐시 등을 잘 구축해서 활용하고 있는 점은 잘하고 있다고 생각했다. Facebook의 graph storage인 TAO, pub-sub 시스템인 Wormhole 등에 대해 관심이 생겼다.