Performance Engineering Learnings @arvindmurthy-perf - Tumblr Blog

Open-sourcing Pulsar, Pub-sub Messaging at Scale

By Joe Francis and Matteo Merli, Yahoo Platforms

Pub-sub messaging is a very common design pattern that is increasingly found in distributed systems powering Internet applications. These applications provide real-time services, and need publish-latencies of 5ms on average and no more than 15ms at the 99th percentile. At Internet scale, these applications require a messaging system with ordering, strong durability, and delivery guarantees. In order to handle the “five 9’s” durability requirements of a production environment, the messages have to be committed on multiple disks or nodes.

At the time we started, we could not find any existing open-source messaging solution that could provide the scale, performance, and features Yahoo required to provide messaging as a hosted service, supporting a million topics. So we set out to build Pulsar as a general messaging solution, that also addresses these specific requirements.

Pulsar is a highly scalable, low latency pub-sub messaging system running on commodity hardware. It provides simple pub-sub messaging semantics over topics, guaranteed at-least-once delivery of messages, automatic cursor management for subscribers, and cross-datacenter replication.

Using Pulsar, one can set up a centrally-managed cluster to provide pub-sub messaging as a service; applications can be onboarded as tenants. Pulsar is horizontally scalable; the number of topics, messages processed, throughput, and storage capacity can be expanded by adding servers to the pool.

Pulsar has a robust set of APIs to manage the service, namely, account management activities like provisioning users, allocating capacity, accounting usage, and monitoring the service. Tenants can administer, manage, and monitor their own domains via APIs. Pulsar also provides security via a pluggable authentication scheme, and access control features that let tenants manage access to their data.

Application development using Pulsar is easy due to the simple messaging model and API. Pulsar includes a client library that encapsulates the messaging protocol; complex functions like service discovery, as well as connection establishment and recovery, are handled internally by the library.

Architecture

At a high level, a Pulsar instance is composed of multiple clusters, typically residing in different geographical regions. A Pulsar cluster is composed of a set of Brokers and BookKeepers (bookies), plus ZooKeeper ensembles for coordination and configuration management.

A Pulsar broker serves topics. Each topic is assigned to a broker, and a broker serves thousands of topics. The broker accepts messages from writers, commits them to a durable store, and dispatches them to readers. The broker also serves admin requests. It has no durable state. The broker has built-in optimizations; for example, it caches the data in order to avoid additional disk reads when dispatching messages to clients as well as replication clusters. Pulsar brokers also manage the replicators, which asynchronously push messages published in the local cluster to remote clusters.

Apache BookKeeper is the building block for Pulsar’s durable storage. BookKeeper is a distributed write-ahead log system, a top-level Apache project that was originally developed at and open-sourced by Yahoo in 2011. BookKeeper has an active developer community with contributors across the industry. Using the BookKeeper built-in semantics, Pulsar creates multiple independent logs, called ledgers, and uses them for durable message storage. Bookkeeper hosts, called bookies, are designed to handle thousands of ledgers with concurrent reads and writes. BookKeeper is horizontally scalable in capacity and throughput; from an operational perspective we can elastically add more bookies to a Pulsar cluster to increase capacity.

By using separate physical disks (one for journal and another for general storage), bookies are able to isolate the effects of read operations from impacting the latency of ongoing write operations, and vice-versa. Since read and write paths are decoupled, spikes in reads – which commonly occur when readers drain backlog to catch up – do not impact publish latencies in Pulsar. This sets Pulsar apart from other commonly-used messaging systems.

Managed Ledger represents the storage layer for a single topic. It is the abstraction of a stream of messages, with a single writer, and multiple readers, each with its own associated cursor position, the offset of the reader in the message stream. A single managed ledger uses multiple BookKeeper ledgers to store the data. Cursor positions are maintained in per-cursor ledgers.

A Pulsar cluster runs a ZooKeeper (another top-level Apache project open-sourced by Yahoo in 2008) ensemble used for coordinating assignment of topics among brokers, and storing BookKeeper metadata. In addition, Pulsar runs a Global ZooKeeper ensemble to store the provisioning and configuration data. At Yahoo, we have presence in multiple regions and our users create global topics that are replicated between these regions. The Global Zookeeper ensemble keeps provisioning and configuration data consistent globally. We can tolerate higher write latencies on these writes (e.g.: ~150ms latency for configuration writes).

The load balancer is a distributed service that runs on the brokers, to make sure the traffic is equally spread across all available brokers. Since Pulsar brokers have no durable state, topics can be redistributed within seconds.

Messaging Model

The Pulsar topic is the core of the system; applications and components communicate by publishing to and consuming from the same topic. Topics are created dynamically as needed when a producer (writer) starts publishing on it; and topics are removed when not in use.

Subscriptions are created automatically when a consumer (reader) subscribes to the topic. A subscription persists until it is deleted, and receives all messages published during its lifetime. Common messaging semantics (like JMS Topic or Queue) are available as subscription modes; an exclusive subscription is equivalent to a “topic,” and a shared subscription is equivalent to a “queue.”

Performance

Pulsar is designed for low-publish latencies at scale. Our typical publish latencies on average are well below 5ms. With SSD as the bookie journal device, Pulsar can achieve 99 percentile latencies of 5ms with two guaranteed copies and total ordering.

The latency remains within the acceptable range until the throughput reaches the limit of the disk IO capacity.

Pulsar supports partitioned topics, which can further increase the per-topic throughput.

Pulsar at Yahoo

Pulsar backs major Yahoo applications like Mail, Finance, Sports, Gemini Ads, and Sherpa, Yahoo’s distributed key-value service.

We deployed our first Pulsar instance in Q2 2015. Pulsar use has rapidly grown since then, and as of today, Yahoo runs Pulsar at scale.

Deployed globally, in 10+ data-centers, with full mesh replication capability

Greater than 100 billion messages/day published

More than 1.4 million topics

Average publish latency across the service of less than 5 ms

As Pulsar use grows at Yahoo, we have been scaling the service horizontally. Most of the challenges we faced were with JVM GC impacting publish latencies, and reducing failover times when the number of topics on a broker went up to tens of 1000s (now 40,000). This led to significant changes to the Pulsar broker and to BookKeeper.

Looking to the Future

We are actively engaged in pushing the scale and reliability boundaries of Pulsar further. Current improvements being worked on include:

Migrate topic between brokers in under 1 sec, from 10 sec

Improve 99.9%ile publish latencies to 5ms

Provide additional language bindings for Pulsar

Conclusion

Pulsar is a highly scalable pub-sub messaging system, production-ready and battled tested at Yahoo. We are glad to make Pulsar available as open source under Apache License Version 2.0. Detailed instructions and documentation are available at Yahoo’s Github repository. Our goal is to make Pulsar widely used and well integrated with other large-scale open source software, and we welcome contributions from the community to make that happen.

Additional Links

Pulsar Github project page: https://github.com/yahoo/pulsar

Pulsar BookKeeper enhancements: https://github.com/yahoo/bookkeeper/tree/yahoo-4.3

arvindmurthy-perf

Good one

Publication on distributed tracing

Source code

Efficient way to get timestamps

#include<sys/time.h> #include<time.h> #include<stdio.h> typedef unsigned long long ticks; static __inline__ ticks getticks(void) { unsigned a, d; asm("cpuid"); asm volatile("rdtsc" : "=a" (a), "=d" (d)); return (((ticks)a) | (((ticks)d) << 32)); }

Python script to parse pcap files

#!/usr/bin/env python import dpkt import socket f = open('test-gso-off.pcap', 'rb') pcap = dpkt.pcap.Reader(f) for ts, buf in pcap: eth = dpkt.ethernet.Ethernet(buf) if eth.type != dpkt.ethernet.ETH_TYPE_IP: continue ip = eth.data if type(ip.data) != dpkt.tcp.TCP: continue tcp = ip.data dst_ip_addr_str = socket.inet_ntop(socket.AF_INET, ip.dst) print "Local : %d >> %s : %d. SEQ : %d, WIN : %d " %(tcp.sport, dst_ip_addr_str ,tcp.dport, tcp.seq, tcp.win) f.close()

source : https://lh5.googleusercontent.com/-jqnY6HX0mqk/Ux3k5V76G7I/AAAAAAAAB7Q/wpEmNch26zQ/w1194-h1320-no/sorting.jpg

Sampling is used to obtain estimates of properties of the population under study, and the sampling system will be evaluated by how good the estimates obtained are and its ability to represent the population. Individual estimates could represent the population with varying error margins.

How far the sample is representative of the population needs to be measured.

Sampling methods :

Random : The basic concept in all sampling is the random sample. A sample of objects from a population is random if all the members of the population have an equal chance of appearing in the sample.

Startified : While sampling a heterogeneous population, the precision could e increased whilst reducing the risk of bias. By classifying the population into classes, where each are relatively homogeneous, further sampling each class (stratum) separately. Each stratum is further sampled independently. The combined classes give an estimate for the whole population.

Multi stage : If the population being sampled is complex, there are practical problems in drawing a simple random sample. Time to draw a small sample could be more. The time required for sampling could be reduced by carrying out the sampling in two stages, further making this a divide and conquer strategy. In step-1 the complete population could be divided into a number of sub-populations, and from these fixed number of samples are collected.

The problem of sampling while the entire population is not available. There could be instances where sampling needs to be done at source. For example, collecting user behavior. It may not be feasible at all to collect behavior of all the users. There are various techniques that could be applied to sample.

Temporal-Fixed sampling : Samples are drawn at uniformly over time, using a fixed %. Temporal-Variable sampling : Samples are drawn using a predefined frequency vector. Different number of samples are drawn at different points in time. Prior knowledge of population behavior or heuristics is used to build the frequency vector.

Selective sampling : The population is sliced by dimensions and samples are drawn from the slice of interest. The population is sliced upon dimensions for example, geo-location, age, etc.

Fixed vs variable population members : The sampling could be based on a fixed set of members in the population. Here observations over time are made using the same members. In variable case, the members change over time giving coverage of the entire population to be sampled over time. The degree of new introductions and evictions are dependent on the application and accuracy required.

For every SSL connection four messages are exchanged between client and server. Moreover, to share a common secret, client and the server perform cryptographic operations that are expensive computation-wise. Abbreviated handshake is smaller, complete expensive round-trip is avoided. There are two ways to achieve abbreviated handshake. Session identifiers as stated in RFC 5246 and Session tickets as detailed in RFC 5077. RFC 5077 TLS Session tickets https://issues.apache.org/jira/browse/TS-1146 RFC 5077 TLS Tools https://github.com/vincentbernat/rfc5077

#ssl

The computation cost of an handshake is more important on the server side than on the client side. The advisory explains that a server will require 15 times the processing power of a client. This means a single average workstation could challenge a multi-core high-end server.

The use of SSL renegotiation allows you to trigger hundreds of handshakes in the same TCP connection. Even a client behind a DSL connection can therefore bomb a server with a lot of renegotiation requests.

There is no definitive solution to this attack but there exists some workarounds. Since the DoS tool from THC relies heavily on renegotiation, the most obvious one is to disable this mechanism on the server side but we will explore other possibilities.

#ssl #DoS

Little’s Law

Little’s Law: John Dutton Conant Little, is an Institute Professor at the Massachusetts Institute of Technology, best known for his result in Operations Research, Little’s Law. Little’s law is truly amazing in its simplicity and can be used to describe the most complex of systems including resources within those systems. N – Number of users present within the system C – Number of completions X – Throughput or Rate of Departure A – Number of Arrivals λ – Rate of Arrival Rt – Time spent by Customers within the system Little’s Law basically states that the long-term average number of customers in a stable system N is equal to the long-term average effective arrival rate, λ, multiplied by the average time a customer spends in the system, W or Rt, or expressed algebraically: N = λ * Rt ……………….. [ N = Number of Users in the System, Rt = Response Time, λ = Arrival Rate ] Little’s Law can also be stated as: N = Rt * X ……………….. [ N = Number of Users in the System, Rt = Response Time, X = Throughput ] For a system where Zt (Think Time is Non Zero) Little’s Law can be stated as: N = [Rt + Zt] * X ……………….. [ N = Number of Users in the System, Rt = Response Time, Zt = Think Time, X = Throughput ]

Little’s Law can be applied to the following scenarios:

Model Performance Testing Workload

Validate Performance Testing Results

Validate the Sizing Models that your Infrastructure guys have come up with

Validate the Non Functional Requirements that your Architects have recommended

Example – Determine Concurrent Users Using Little’s Law: A system at peak processes 8000 Transactions/Hour with an average response time of ~5s per transaction. The average Think Time per user is ~30s. What is the number of concurrent users on the system? Throughput or X = 8000/3600 = 2.22 TPS We know: N = [Rt + Zt] * X ……….. [ N = Number of Users in the System, Rt = Response Time, Zt = Think Time, X = Throughput ] Applying the equation: N = [ Rt + Z ] * X = [ 5 + 30 ] * 2.22 = 77.7 The above system has approximately 77 users at peak driving 8000 Transactions/Hour. Example – Validate System Performance Testing Results: The outcome of a performance test is as follows: Peak User Concurrency = 1000 Average Transactional Response Time = 5s Average User Think Time = 40s Peak Transactional Throughput ( as recorded by the tool) = 500 TPS We know: N = [Rt + Zt] * X ……….. [ N = Number of Users in the System, Rt = Response Time, Zt = Think Time, X = Throughput ] Applying the equation: X = N / [ Rt + Z ] = 1000 / [ 5 + 40 ] = 22.22 TPSYou know from the above results, something is really wrong. Little’s Law suggests that the actual TPS for the above system should be ~22 TPS but the Performance Testing results show transactional throughput as 500 TPS. Could this be errors in the Performance Testing results due to poorly written scripts? If your scripts are not smart enough to catch all the relevant errors, the Performance Testing tool will include the errors as part of the true transactional throughput. Source : http://www1.practicalperformanceanalyst.com/resources/important-formulae/what-is-littles-law/

#Little’s Law

Here is some little known data which may be of interest to computer hackers. The items and examples are so sketchy that to decipher them may require more sincerity and curiosity than a non-hacker can muster.

An Optimal Decomposition Algorithm for Tree Edit Distance

Source : http://www.cs.haifa.ac.il/~oren/Publications/TEDinTALG.pdf

WebP is a new image format which provides lossy and lossless compression of images, with significant byte savings: 30-80% smaller image files when compared to jpeg and png!

Trending Blogs

Recently Viewed Blogs

Performance Engineering Learnings