Jseung21 기술 블로그

Iguazio + Nuclio를 활용한 실시간 추천시스템 구축

Wed, 02 Jan 2019 10:00:00 +0900

Iguazio

Principal pattern recognition on unsupervised learning

Wed, 02 Jan 2019 00:00:00 +0900

● Principal Pattern Recognition 이란

Log Data의 행들간의 거리값을 기반으로 군집화하여 가장 주요패턴그룹을 인지한다.

● 목적

샘플링 1000건에서 일반화된 하나의 AD Language를 생성한다.

그런데 샘플 1000건중에 오류 데이터가 있으면 그 특성까지 반영되어 버림에 따라

샘플 1000건 중에 주요 패턴 그룹을 찾아 이를 기반으로 일반화 하여 AD Language의 품질을 높인다.

● 행들간 거리 값 산출 공식

Log Data를 수치로 변환하기 위해 아래와 같은 공식으로 행들간의 거리값을 계산하여,

클러스터링 입력파라메타로 활용한다.

※ 위 결과값은 ‘0’과 ‘1’사이의 값을 가지게 되고 ‘0’에 수렴할수록 행간 유사도 높음

● 클러스터링 알고리즘

DBSCAN (Density-Based Spatial Clustering of Applications with Noise)

DBSCAN은 방문하지 않은 임의의 시작 데이터 포인트로 시작합니다. 이 점의 근방은 거리 ε (ε 거리 내에있는 모든 점은 neighborhood point)을 사용하여 추출됩니다.
이 근처에 충분한 포인트 수가 있으면 (minPoints에 따라) 클러스터링 프로세스가 시작되고 현재 데이터 포인트가 새 클러스터의 첫번째 포인트가됩니다. 그렇지 않은 경우, 포인트는 노이즈로 레이블됩니다 (나중에이 노이즈 포인트가 클러스터의 일부가 될 수 있음). 두 경우 모두 해당 지점은 “visited”로 표시됩니다.
새로운 클러스터의 이 첫번째 점에 대해 ε 거리 인접도 내의 점도 동일한 클러스터의 일부가됩니다. 클러스터 그룹에 방금 추가 된 모든 새 점에 대해 동일한 클러스터에 속한 ε 근방에있는 모든 점을 만드는 이 절차가 반복됩니다.
이 2 단계와 3 단계 프로세스는 클러스터의 모든 점이 결정될 때까지 반복됩니다. 즉 클러스터의 ε 근처에있는 모든 점을 방문하여 레이블을 지정합니다.
현재 클러스터를 완료하고 나면 새로운 미확인 지점이 검색되고 처리되어 추가 클러스터 또는 노이즈가 발견됩니다. 이 프로세스는 모든 포인트가 방문으로 표시 될 때까지 반복됩니다. 이 지점의 모든 지점을 방문 했으므로 각 지점은 클러스터에 속하거나 노이즈로 표시됩니다.

DBSCAN은 다른 클러스터링 알고리즘에 비해 큰 장점이 있습니다. 첫째로, 그것은 클러스터 집합을 전혀 필요로하지 않습니다. 또한 데이터 포인트가 매우 다르더라도 이상 값을 단순히 클러스터에 버리는 K-Means와는 달리 이상 치를 노이즈로 식별합니다. 또한 임의로 크기가 정해지고 임의로 모양이 지정된 클러스터를 매우 잘 찾을 수 있습니다.

DBSCAN의 가장 큰 단점은 클러스터의 밀도가 다양 할 때 다른 클러스터와 마찬가지로 잘 수행되지 않는다는 것입니다. 이는 밀도가 변할 때 neighborhood point를 식별하기위한 거리 임계 값 ε 및 minPoints의 설정이 클러스터마다 다양하기 때문입니다. 이 단점은 거리 임계치 ε가 다시 추정하기가 어려워지기 때문에 매우 고차원적인 데이터에서도 발생합니다.

● 처리흐름

● 예시

목적 : 입력파라메타 distance 값을 조정하면서 R3, R4를 주요패턴그룹으로 포함하고 싶다

R2 : 16열 데이터 누락으로 데이터가 앞으로 당겨짐

R3 : 9,10열 이상데이터 포함

**R4 : 뒤에 garbage 데이터 포함
**
데이터
**
**
행들간의 매트릭스

**
**

distance 0~10 변화에 따른 클러스터 값

※ distance 값이 5일때 R3, R4가 하나의 그룹으로 되었다가, distance 값이 6일때 주요패턴그룹으로 포함된다.

Nuclio is Serverless Engine

Wed, 26 Dec 2018 10:00:00 +0900

Architecture

Function processors

Function processors provide an environment for executing functions. The processor feeds functions with events, provides context and data, collects logs and statistics, and manages the function’s life cycle.

Processors can be compiled into a single binary (when using Go or C), or packaged into a container with all the required code dependencies. Processor containers can run as standalone Docker containers, or on top of a container-orchestration platform such as Kubernetes. Each function has its own function processors. Function Processors will be scaled-out automatically (by adding more container instances) to address high event frequency.

Processor architecture

Nuclio’s unique processor architecture is aimed at maximizing function performance and providing abstractions and portability across a wide set of platforms, event sources, and data services.

The processor has four main components:

Event-source listeners

Event-source listeners can listen on sockets and message queues, or periodically fetch events from external event or data sources. Received events share a common schema, which decouples the function logic from the event source implementation or specific structure, and pushed to one or more parallel runtime workers.

The event listeners also guarantee exactly-once or at-least-once event execution and handle failures — for example, by storing stream checkpoints, acknowledging or retrying message-queue events, or responding to HTTP client requests.

Runtime engine

The runtime engine (“runtime”) initializes the function environment (variables, context, log, data bindings, etc.), feeds event objects into functions workers, and returns responses to the event sources.

Runtimes can have multiple independent parallel workers (for example, Go routines, Python asyncio, Akka, or threads) to enable non-blocking operations and maximize CPU utilization.

Nuclio currently supports three types of processor runtime implementations:

Native — for real-time and inline Go or C-based routines.
SHMEM — for shared-memory languages, such as Python, Java, and Node.js. The processor communicates with the SHMEM function runtime through zero-copy shared-memory channels.
Shell — for command-line execution functions or binaries (“executables”). Upon receiving an event, the processor runs the executable in a command-line shell, using the relevant command and environment variables, and then maps the executable’s standard-output (stdout) or standard-error (stderr) logs to the function results.

Note

The shell runtime supports only file data bindings.

Data bindings

Functions can benefit from persistent data connections to external files, objects, databases, or messaging systems. The runtime initializes the data-service connection based on the type, URL, properties, and credentials specified in the function specification, and serves them to the function through the context object.

Data bindings simplify development by eliminating the need to integrate with SDKs or manage connections and credentials. They also enable function reuse and portability, because different data services of the same class are mapped to the function using the same APIs.

Data bindings can also handle aspects of data prefetching, caching, and micro-batching, to reduce execution latency and increase I/O performance. Data bindings and event sources are designed with zero-copy, zero-serialization, and non-blocking operation, and enable real-time performance without the need for any special function code.

Control framework

The control framework initializes and controls the different processor components, provides logging for the processor and the function (stored in different log streams), monitors execution statistics, and serves a mini portal for remote management.

The control framework interacts with the underlining platform through abstract interfaces, allowing for portability across different IoT devices, container orchestrators, and cloud platforms. The platform-specific processor configuration is done through a processor.yaml file in the working directory. Function developers should not modify this file. The underlying interfaces of the function processors to the required platform services are all abstracted in a way that enables porting the same function processor among different deployments types.

Event sources and mapping

Functions are event-driven. They respond to event triggers, data messages, or records that are accepted from the event source and pushed to the function runtime engine.

Event sources can be divided into classes, based on their behavior and flow management. Each class can have multiple event-source implementations. Nuclio supports the following event classes:

Synchronous Request/Response — the client issues a request and waits for an immediate response. For example, HTTP requests or Remote Procedure Calls (RPCs).
Asynchronous Message-Queue Request — messages are published to an exchange and distributed to subscribers. For example, RabbitMQ, MQTT, emails, or scheduled events.
Message or Record Streams — an ordered set of messages or record updates is processed sequentially. For example, Kafka, AWS Kinesis or Iguazio V3IO streams.
Record or Data Polling (ETL) — a filtered set of records or data objects is retrieved from an external data source or database. The retrieval can be done periodically or triggered by data changes.

New event classes and event sources can be added to the processor framework.

Event-source mapping

Event sources are mapped to a specific function version. For example, the API gateway web URL “/” may be mapped to the “production” version, and the URL “/beta” may be mapped to the “beta” version of the same function. The user needs to specify the event mapping in the function specification, or by using event mapping CRUD API calls or CLI commands. (The current CLI version does not yet support event mapping, and still requires some manual configuration.)

Multiple event sources can be associated with the same function, and the same event can trigger the invocation of multiple functions.

The event-source mapping can utilize either the exact function version or its alias (see details under versioning). The mapping also includes information such as the function name, class, type, credentials, and class-specific properties.

Event load-balancing, sharding, and dealers

Some jobs involve distributing data or work items across multiple function processor instances. For example, a Kafka stream can be divided into several partitions, and each partition should only be processed by a single processor at any given time. Or in the case of a sharded dataset, you might want to scale-out the processing of the dataset across multiple processing elements, which requires a resource-scheduling entity that will distribute data partitions to the available processors and track the execution and completion.

Nuclio includes a “dealer” entity that can dynamically distribute N resources (shards, partitions, tasks, etc.) to M processors, and can handle aspects of failures and scale-up or scale-down of resources and processors.

Event object (used by the function)

Functions are called with two elements, the context object and the event object. The event object describes the data and metadata of the incoming event. It is generalized in a way that decouples the actual event source from the function. A single function can be driven by multiple types of event sources. A function can accept a single event or an array of events (for example, when using a stream).

An event object is accessed through interfaces (methods). To enable zero-copy, eliminate serialization overhead, and add robustness, an event object can also be consumed as a JSON object, which might add some serialization and deserialization overhead. There are common event-object interfaces, such as EventID, Body, ContentType, Headers, Fields, and AsJson. There are also event-object interfaces that are class-specific.

Function build and deployment flow

You begin the development flow by implementing the function in one of the supported languages. You then build the function artifacts (a code binary, package, or container image), deploy the function on the destination cluster, and feed it with events and data, as illustrated in the following diagram.

Each function has a version-specific function specification (“spec”). The function spec defines different aspects of the function, such as its code, data binding, environment resources, credentials, and event sources.
The function spec can be written in YAML or JSON, and can also be defined or overwritten using command-line options.
The builder uses the function spec in the compilation phase, and the controller uses the spec to identify the operating requirements for the function.

The user can control whether to build and deploy the function in a single operation, or whether to perform each step separately.
The build command compiles the function and builds the artifact, so it can later be used by one or more deployments of the function.
The run command can accept sources, and can either both build and deploy the function, or skip the build and deploy an existing artifact.

The user can control the exact build flags through an optional build.yaml file that is located in the same directory as the function sources.

Function versioning

Nuclio supports function versioning. You can publish a new version of a function and tag it with aliases for referencing the function.

Nuclio provides the ability to run different versions of the same function simultaneously (for example, production and beta versions).

The event-source mapping can either specify the exact function version, or utilize a defined alias and thus eliminate the need to change the mapping when a newer version of the function is published.

Iguazio is Unified Data Platform

Wed, 26 Dec 2018 10:00:00 +0900

Serverless Platform

Iguazio combines real-time actions and AI across a variety of data sources and types, without compromising simplicity and fast time to market.

The Iguazio platform includes leading open source data science tools, managed serverless functions and data services running on Kubernetes and on a real-time unified database.

Open Platform for Intelligent Apps

Faster Dev & Ops

Accelerate time from data science exploration to production in a fully managed serverless environment with real-time functions, AI and Kubernetes

Unified Real-time Database

Ingest, enrich, analyze and serve large scale multi-model data with simultaneous access through standard APIs for NoSQL, SQL, time series and files.

Deployed Anywhere

Leverage the same data services, AI and serverless abstractions in the cloud, at the edge and on-premises, cutting endless IT tasks.

Learn more about serverless and AI

In-Memory Speed at High Density and Lower Costs

Iguazio combines fresh, historical and operational data to support smarter decisions in real-time. The platform maximizes CPU utilization and leverages non-volatile memory, 100GbE RDMA, flash and dense storage, achieving extreme performance with consistency at the lowest cost. It shifts the balance from underutilized systems and inefficient code to parallel, real-time and resource optimized implementation requiring fewer servers.

Learn more about in-memory performance at high density

Get Started Now:

TRY IGUAZIO Book a Demo

Self-Service Management

Managed Data Services

The Iguazio Continuous Data Platform changes the way you monitor, search and explore your data. The system stores rich metadata with each object, turning the platform into a smart data catalog where users can explore data and gain operational and “hidden” business insights in a matter of a few simple clicks.

Serverless

nuclio is an ultra-fast open source serverless platform which allows developers to focus on building and running auto-scaling applications without worrying about managing servers. It Supports a variety of event and data sources with common APIs and is portable across edge and multi-cloud deployments

Automated IT Operations

IT admins use a simple interface to provision and monitor multiple data clusters and their resources including CPUs, networking and storage with minimal human interaction. The platform provides a rich set of reports and views and generates alarms based on predefined filtering and escalation policies.

Fine-Grained Security

The platform accepts software-defined policies and automatically maps them to access control, multi-tenancy, monitoring, auditing, service levels and other custom operations. It integrates with identity management systems, combines API level security and visualizes effective policies per data element.

Deployment Anywhere

Cloud

On demand service running in the iguazio managed cloud or in alternative public multi-clouds

On-Premises

A complete cloud experience running on-premises on iguazio’s scalable multi-node cluster

Edge

Edge appliance handling large data volumes with small footprint, aggregated from devices in real-time

Get Started Now

Go real-time, AI and serverless with Iguazio

Self-Driving Car Engineer Advanced Lane Finding

Tue, 25 Dec 2018 10:00:00 +0900

Self-Driving Car Engineer Nanodegree Program

Advanced Lane Finding Project

The goals / steps of this project are the following:

Compute the camera calibration matrix and distortion coefficients given a set of chessboard images.
Apply a distortion correction to raw images.
Use color transforms, gradients, etc., to create a thresholded binary image.
Apply a perspective transform to rectify binary image (“birds-eye view”).
Detect lane pixels and fit to find the lane boundary.
Determine the curvature of the lane and vehicle position with respect to center.
Warp the detected lane boundaries back onto the original image.
Output visual display of the lane boundaries and numerical estimation of lane curvature and vehicle position.

Rubric Points

Here I will consider the rubric points individually and describe how I addressed each point in my implementation.

Writeup / README

1. Provide a Writeup / README that includes all the rubric points and how you addressed each one. You can submit your writeup as markdown or pdf. Here is a template writeup for this project you can use as a guide and a starting point.

You’re reading it!

Camera Calibration

1. Briefly state how you computed the camera matrix and distortion coefficients. Provide an example of a distortion corrected calibration image.

The code for this step is contained in the first two code cells of the Jupyter notebook project.ipynb.

The OpenCV functions findChessboardCorners and calibrateCamera are the backbone of the image calibration. A number of images of a chessboard, taken from different angles with the same camera, comprise the input. Arrays of object points, corresponding to the location (essentially indices) of internal corners of a chessboard, and image points, the pixel locations of the internal chessboard corners determined by findChessboardCorners, are fed to calibrateCamera which returns camera calibration and distortion coefficients. These can then be used by the OpenCV undistort function to undo the effects of distortion on any image produced by the same camera. Generally, these coefficients will not change for a given camera (and lens). The below image depicts the corners drawn onto twenty chessboard images using the OpenCV function drawChessboardCorners:

Note: Some of the chessboard images don’t appear because findChessboardCorners was unable to detect the desired number of internal corners.

The image below depicts the results of applying undistort, using the calibration and distortion coefficients, to one of the chessboard images:

Pipeline (single images)

1. Provide an example of a distortion-corrected image.

The image below depicts the results of applying undistort to one of the project dashcam images:

The effect of undistort is subtle, but can be perceived from the difference in shape of the hood of the car at the bottom corners of the image.

2. Describe how (and identify where in your code) you used color transforms, gradients or other methods to create a thresholded binary image. Provide an example of a binary image result.

I explored several combinations of sobel gradient thresholds and color channel thresholds in multiple color spaces. These are labeled clearly in the Jupyter notebook. Below is an example of the combination of sobel magnitude and direction thresholds:

The below image shows the various channels of three different color spaces for the same image:

Ultimately, I chose to use just the L channel of the HLS color space to isolate white lines and the B channel of the LAB colorspace to isolate yellow lines. I did not use any gradient thresholds in my pipeline. I did, however finely tune the threshold for each channel to be minimally tolerant to changes in lighting. As part of this, I chose to normalize the maximum values of the HLS L channel and the LAB B channel (presumably occupied by lane lines) to 255, since the values the lane lines span in these channels can vary depending on lighting conditions. Below are examples of thresholds in the HLS L channel and the LAB B channel:

Note: the B channel was not normalized if the maximum value for an image was less than 175, signifying that there was no yellow actually in the image. Otherwise the resulting thresholded binary image was very noisy.

Below are the results of applying the binary thresholding pipeline to various sample images:

3. Describe how (and identify where in your code) you performed a perspective transform and provide an example of a transformed image.

The code for my perspective transform is titled “Perspective Transform” in the Jupyter notebook, in the seventh and eighth code cells from the top. The unwarp() function takes as inputs an image (img), as well as source (src) and destination (dst) points. I chose to hardcode the source and destination points in the following manner:

src = np.float32([(575,464),
                  (707,464), 
                  (258,682), 
                  (1049,682)])
dst = np.float32([(450,0),
                  (w-450,0),
                  (450,h),
                  (w-450,h)])

I had considered programmatically determining source and destination points, but I felt that I would get better results carefully selecting points using one of the straight_lines test images for reference and assuming that the camera position will remain constant and that the road in the videos will remain relatively flat. The image below demonstrates the results of the perspective transform:

4. Describe how (and identify where in your code) you identified lane-line pixels and fit their positions with a polynomial?

The functions sliding_window_polyfit and polyfit_using_prev_fit, which identify lane lines and fit a second order polynomial to both right and left lane lines, are clearly labeled in the Jupyter notebook as “Sliding Window Polyfit” and “Polyfit Using Fit from Previous Frame”. The first of these computes a histogram of the bottom half of the image and finds the bottom-most x position (or “base”) of the left and right lane lines. Originally these locations were identified from the local maxima of the left and right halves of the histogram, but in my final implementation I changed these to quarters of the histogram just left and right of the midpoint. This helped to reject lines from adjacent lanes. The function then identifies ten windows from which to identify lane pixels, each one centered on the midpoint of the pixels from the window below. This effectively “follows” the lane lines up to the top of the binary image, and speeds processing by only searching for activated pixels over a small portion of the image. Pixels belonging to each lane line are identified and the Numpy polyfit() method fits a second order polynomial to each set of pixels. The image below demonstrates how this process works:

The image below depicts the histogram generated by sliding_window_polyfit; the resulting base points for the left and right lanes - the two peaks nearest the center - are clearly visible:

The polyfit_using_prev_fit function performs basically the same task, but alleviates much difficulty of the search process by leveraging a previous fit (from a previous video frame, for example) and only searching for lane pixels within a certain range of that fit. The image below demonstrates this - the green shaded area is the range from the previous fit, and the yellow lines and red and blue pixels are from the current image:

5. Describe how (and identify where in your code) you calculated the radius of curvature of the lane and the position of the vehicle with respect to center.

The radius of curvature is based upon this website and calculated in the code cell titled “Radius of Curvature and Distance from Lane Center Calculation” using this line of code (altered for clarity):

curve_radius = ((1 + (2*fit[0]*y_0*y_meters_per_pixel + fit[1])**2)**1.5) / np.absolute(2*fit[0])

In this example, fit[0] is the first coefficient (the y-squared coefficient) of the second order polynomial fit, and fit[1] is the second (y) coefficient. y_0 is the y position within the image upon which the curvature calculation is based (the bottom-most y - the position of the car in the image - was chosen). y_meters_per_pixel is the factor used for converting from pixels to meters. This conversion was also used to generate a new fit with coefficients in terms of meters.

The position of the vehicle with respect to the center of the lane is calculated with the following lines of code:

lane_center_position = (r_fit_x_int + l_fit_x_int) /2
center_dist = (car_position - lane_center_position) * x_meters_per_pix

r_fit_x_int and l_fit_x_int are the x-intercepts of the right and left fits, respectively. This requires evaluating the fit at the maximum y value (719, in this case - the bottom of the image) because the minimum y value is actually at the top (otherwise, the constant coefficient of each fit would have sufficed). The car position is the difference between these intercept points and the image midpoint (assuming that the camera is mounted at the center of the vehicle).

6. Provide an example image of your result plotted back down onto the road such that the lane area is identified clearly.

I implemented this step in the code cells titled “Draw the Detected Lane Back onto the Original Image” and “Draw Curvature Radius and Distance from Center Data onto the Original Image” in the Jupyter notebook. A polygon is generated based on plots of the left and right fits, warped back to the perspective of the original image using the inverse perspective matrix Minv and overlaid onto the original image. The image below is an example of the results of the draw_lane function:

Below is an example of the results of the draw_data function, which writes text identifying the curvature radius and vehicle position data onto the original image:

Pipeline (video)

1. Provide a link to your final video output. Your pipeline should perform reasonably well on the entire project video (wobbly lines are ok but no catastrophic failures that would cause the car to drive off the road!).

Here’s a link to my video result

Discussion

1. Briefly discuss any problems / issues you faced in your implementation of this project. Where will your pipeline likely fail? What could you do to make it more robust?

The problems I encountered were almost exclusively due to lighting conditions, shadows, discoloration, etc. It wasn’t difficult to dial in threshold parameters to get the pipeline to perform well on the original project video (particularly after discovering the B channel of the LAB colorspace, which isolates the yellow lines very well), even on the lighter-gray bridge sections that comprised the most difficult sections of the video. It was trying to extend the same pipeline to the challenge video that presented the greatest (ahem) challenge. The lane lines don’t necessarily occupy the same pixel value (speaking of the L channel of the HLS color space) range on this video that they occupy on the first video, so the normalization/scaling technique helped here quite a bit, although it also tended to create problems (large noisy areas activated in the binary image) when the white lines didn’t contrast with the rest of the image enough. This would definitely be an issue in snow or in a situation where, for example, a bright white car were driving among dull white lane lines. Producing a pipeline from which lane lines can reliably be identified was of utmost importance (garbage in, garbage out - as they say), but smoothing the video output by averaging the last n found good fits also helped. My approach also invalidates fits if the left and right base points aren’t a certain distance apart (within some tolerance) under the assumption that the lane width will remain relatively constant.

I’ve considered a few possible approaches for making my algorithm more robust. These include more dynamic thresholding (perhaps considering separate threshold parameters for different horizontal slices of the image, or dynamically selecting threshold parameters based on the resulting number of activated pixels), designating a confidence level for fits and rejecting new fits that deviate beyond a certain amount (this is already implemented in a relatively unsophisticated way) or rejecting the right fit (for example) if the confidence in the left fit is high and right fit deviates too much (enforcing roughly parallel fits). I hope to revisit some of these strategies in the future.

Addendum: Below are diagnostic versions of the output for each of the videos included in the project.

Project video - diagnostic version

Challenge video - diagnostic version

Harder challenge video - diagnostic version

kubernetes를 이용한 서비스 무중단 배포

Mon, 24 Dec 2018 10:00:00 +0900

Kubernetes는 컨테이너 오케스트레이션 영역에서 거의 표준으로 자리 잡은 오픈소스 시스템입니다. kubernetes를 사용하게 되면 여러대의 노드를 하나의 클러스터로 묶어서 사용가능하게 됩니다. 클러스터를 구성하는 노드들중에 일부에 장애가 발생하더라도 장애가 난 곳에 있던 컨테이너가 kubernetes에 의해 다른 정상상태의 노드로 옮겨가게 되어서 컨테이너로 제공하던 서비스에 지장이 없이 서비스가 지속될 수 있게 해줍니다. 그래서 실제로 서비스를 운영할 때는 컨테이너만을 단독으로 사용하기 보다는 이런 오케스트레이터와 함께 사용하는 경우가 많습니다.

kubernetes를 사용하면 배포를 보다 편리하게 할 수 있다는 장점도 있습니다. 앱을 실행할 컨테이너만 준비해서 kubernetes에 제출하면 kubernetes가 알아서 배포절차를 진행합니다. 카카오에서 컨테이너 플랫폼을 운영하면서 가장 많이 받는 질문중 하나가 “배포중에 트래픽 유실은 없나요?” 입니다. 트래픽이 큰 서비스를 운영하면서 서비스의 품질을 유지하려면 배포중에도 트래픽 유실이 없어야 합니다. 이 글에서는 kubernetes를 사용해서 배포했을때 트래픽 유실이 없게하기 위해서 어떤 점들을 유의해야 하는지 알아보도록 하겠습니다.

kubernetes pod, service, ingress 관계

먼저 kubernetes로 트래픽이 들어오는 구조를 살펴보도록 하겠습니다. kubernetes클러스터 내부에 있는 pod까지 트래픽이 도달하는 경로는 대략 다음과 같습니다.

kubernetes는 별도의 클러스터 네트워크를 구성해서 이뤄지는 경우가 많기 때문에 외부에서 클러스터 내부에서 실행중인 pod에 접근하기 위해서는 인입되는 트래픽을 클러스터 내부까지 전달해줄 LB역할을 해줄 매개체가 필요한데요. AWS, GCP, Azure같은 Public cloud를 사용할때는 거기서 제공해주는 LB를 사용하면 되지만, 내부에 구축할때는 일반적으로 앞의 그림에 있는 ingress-controller를 사용합니다. ingress-controller에도 여러가지 종류가 있지만 이 글에서는 일반적인 nginx ingress controller를 기준으로 이야기하겠습니다. Kubernetes 내부에서 pod간 통신을 위해서는 중간에 service를 두고 통신하게 되는데요. ingress를 설정할때 역시 개별 pod들을 이용하는게 아니라 pod와 연결된 service를 설정하도록 되어 있습니다. 다음 ingress 설정을 보시면 serviceName으로 ingress를 통해서 연결하려는 서비스를 지정한 걸 확인할 수 있습니다.

apiVersion: extensions/v1beta1
kind: Ingress
metadata:
  name: test-ingress
  annotations:
    nginx.ingress.kubernetes.io/rewrite-target: /
spec:
  rules:
  - http:
      paths:
      - path: /testpath
        backend:
          serviceName: test
          servicePort: 80

이렇게 설정되면 실제 nginx-ingress-controller는 이 service를 통해서 pod에 연결하는게 아니라 이 service에 연결된 pod의 정보를 가져와서 직접 nginx config로 설정하게 됩니다. 그래서 위의 그림에서는 service를 이용하긴하지만 직접 service를 거쳐서 통신하는건 아니라는 의미에서 service영역을 점선으로 표시했습니다.

kubernetes pod 배포시 기본 구조

앞에서 서비스 트래픽이 실제 어떻게 pod까지 전달되는지 알아봤습니다. 이제 실서비스에서 트래픽이 흘러가고 있는 와중에 pod를 어떻게 무중단으로 배포할 수 있는지 알아보도록 하겠습니다. 우선 pod가 배포되면 pod 교체가 어떻게 일어나는지 살펴보도록 하겠습니다. 정상적인 경우라면 아래 그림처럼 새로운 pod(v2)가 생성되고 헬스체크가 성공한 후에 트래픽이 pod(v2)쪽으로 흘러가게 됩니다. 그 후에 pod(v1) 쪽으로 가던 트래픽이 제거된 후 pod(v1)이 제거 됩니다.

이 과정에서 고려해야 할 부분들이 몇 군데 있습니다.

kubernetes 배포시 고려해야할 컨테이너 구성

Kubernetes 가 대부분의 배포 절차를 잘 수행해 주지만 컨테이너 내부에서도 종료될때 graceful shutdown구현이 필요합니다. 새로운 pod가 실행되고 이전 pod를 종료할때 kubernetes에서 노드의 컨테이너를 관리하는 프로세스인 kubelet은 먼저 pod에 SIGTERM 신호를 보내게 됩니다. 컨테이너에서 SIGTERM을 받았을때 기존에 처리중이던 요청에 대한 처리를 완료하고 새로운 요청을 받지 않도록 개발되어 있어야 합니다. 그렇지 않으면 아래 그림처럼 트래픽은 아직 Pod(v1)쪽으로 가고 있는데 Pod(v1)이 종료되어 버려서 아직 ingress-controller의 설정이 갱신되기 전에 pod(v1)으로 가는 요청들은 에러를 내게 됩니다. kubelet에서 pod에 SIGTERM을 보낸후에 일정시간동안 graceful shutdown이 되지 않는다면 강제로 SIGKILL을 보내서 pod를 종료하게 됩니다. 이 대기 기간은 terminationGracePeriodSeconds 으로 설정해 줄 수 있고 기본 대기 시간은 30초 입니다.

SIGTERM 수신 뒤 즉시 종료 : 17.44% Request 502 error

SIGTERM 시그널 처리시 : 무중단 배포 가능

kubernetes 배포시 고려해야할 kubernetes 옵션

컨테이너가 SIGTERM을 고려해서 잘 만들어져 있다면 kubernetes에서 배포 과정에 어떤 옵션들을 활용할 수 있는지 살펴보겠습니다.

먼저 pod의 롤링업데이트를 위한 maxSurge와 maxUnavailable 설정 옵션입니다. deployment를 이용해서 배포할 때 maxSurge는 deployment에 설정되어 있는 기본 pod개수보다 여분의 pod가 몇개가 더 추가될 수 있는지를 설정할 수 있습니다. maxUnavailable는 업데이트하는 동안 몇 개의 pod가 이용 불가능하게 되어도 되는지를 설정하는데 사용됩니다. 이 두개의 옵션을 운영중인 서비스의 특성에 맞게 적절히 조절해 주어야지 항상 일정 개수 이상의 pod가 이용가능하게 되기 때문에 배포중 트래픽 유실이 없게 됩니다. 둘 다 한꺼번에 0으로 설정되면 pod가 존재하지 않는 경우가 발생하기 때문에 한꺼번에 0으로 설정할 수는 없습니다.

그 다음으로는 pod의 readinessProbe 설정입니다. kubernetes에서는 pod의 헬스체크를 확인하기 위해서 2가지 상태체크 옵션을 주고 있습니다. livenessProbe와 readinessProbe입니다. livenessProbe는 컨테이너가 살아 있는지 확인하는 역할을 하고 이 헬스체크가 실패하면 kubelet이 컨테이너를 죽이게 됩니다. 그리고 컨테이너의 restart policy에 따라 컨테이너가 재시작됩니다. 무중단 배포에서 신경써서 봐야할 설정은 readinessProbe입니다. readinessProbe는 실제로 컨테이너가 서비스 요청을 처리할 준비가 되었는지를 확인하는데 사용됩니다. readinessProbe가 ok상태여야지 이 pod와 연결된 service에 pod의 ip가 추가되고 트래픽을 받을 수 있게 됩니다. 자바 프로세스 같은 경우는 프로세스가 올라와서 livenessProbe가 ok상태가 되더라도 초기화 과정이 오래 걸리기 때문에 readinessProbe를 따로 설정하지 않을때에 아직 준비되지 않은 컨테이너로 요청이 가서 응답을 제대로 하지 못하고 실패할 수 있습니다. 그런 경우를 방지하기 위해서 실제 서비스가 준비된 상태인지를 확인할 수 있는 readinessProbe를 잘 설정해 주어야 합니다.

경우에 따라서는 앱 자체가 readinessProbe를 설정해주기 어려운 상황일 수도 있습니다. 그럴때는 .spec.minReadySeconds 옵션을 이용하면 어느정도 readinessProbe와 비슷한 효과를 낼 수 있습니다. .spec.minReadySeconds은 pod의 status가 ready가 될때까지의 최소대기시간입니다. 그래서 pod가 실행되고나서 .spec.minReadySeconds에 설정된 시간동안은 트래픽을 받지 않습니다. 그렇기 때문에 readinessProbe를 설정하기 어렵고 초기화 시간이 오래 걸리는 컨테이너에 대해서 사용하면 컨테이너가 준비될때까지 일정시간동안 트래픽을 받지않고 대기할 수 있기 때문에 유용하게 사용할 수 있습니다. 하지만, readinessProbe가 완료되면 .spec.minReadySeconds에 설정된 시간이 아직 남아 있더라고도 무시되고 트래픽을 보내게 됩니다. .spec.minReadySeconds의 기본값은 0입니다.

MinReadySeconds 옵션 : pod status 가 ready 로 업데이트 될 때 까지 최소 대기 시간, 그 전까지 서비스에서 트래픽 받지 않음

Readiness Check 가 완료 되면 MinReadySeconds 설정은 무시된다.

pod가 종료될때 graceful shutdown을 구현하지 못하는 경우도 있을 수 있습니다. 앱이 사용하는 언어나 프레임워크가 지원하지 않는 경우도 있고, 오래된 레거시라서 앱을 수정하지 못한채 컨테이너로 올려야되는 경우도 있을 수 있습니다. 이런 경우에는 prestop hook을 이용할 수 있습니다. kubernetes에서는 pod 라이프사이클중에 hook을 설정할 수 있습니다. pod가 실행되고난 직후 실행하는 poststart hook과 pod가 종료되기 직전 실행되는 prestop hook입니다. prestop 훅은 pod에 SIGTERM을 보내기 전에 실행되기 때문에 prestop을 이용하면 앱과 별개로 graceful shutdown의 효과를 내게 할 수도 있습니다. Prestop 훅의 실행이 완료되기 전까지는 컨테이너에 SIGTERM을 보내지 않기 때문에 앱의 구현과는 별개로 종료되기 전에 대기시간을 주는 것도 가능해 지게 됩니다. 하지만 이렇게 prestop 훅으로 대기시간을 주더라도 terminationGracePeriodSeconds 시간을 초과한다면 프로세스 종료가 일어날 수 있으니 염두에 두고 사용해야 합니다.

위 내용은 카카오 사내 컨테이너 오케스트레이션 서비스인 DKOS를 운영하던 중 많이 나오는 문의사항을 토대로 배포시 고려해야 할 kubernetes 구조와 옵션 등을 살펴봤습니다. DKOS는 클라우드디플로이셀의 hardy.jung, dennig.hong, scott.vim, heimer.j등이 함께 운영해오고 있습니다.

MySQL Ascending index vs Descending index

Tue, 19 Jun 2018 15:00:00 +0900

용어 정리

이 설명에서는 인덱스의 정렬 순서와 데이터 읽기 순서 등 방향에 대한 단어들이 혼재하면서, 여러 가지 혼란을 초래하기 쉬운 설명들이 있을 것으로 보인다. 그래서 우선 표준 용어는 아니지만, 나름대로 몇 개 단어들에 대해서 개념을 정립하고 그 단어를 번역 없이 영어로 그대로 표기하도록 하겠다.

Ascending index : 작은 값의 인덱스 키가 B-Tree의 왼쪽으로 정렬된 인덱스
Descending index : 큰 값의 인덱스 키가 B-Tree의 왼쪽으로 정렬된 인덱스
Forward index scan (Forward scan) : 인덱스 키의 크고 작음에 관계없이 인덱스 리프 노드의 왼쪽 페이지부터 오른쪽으로 스캔
Backward index scan (Backward scan) : 인덱스 키의 크고 작음에 관계없이 인덱스 리프 노드의 오른쪽 페이지부터 왼쪽으로 스캔

Descending index 지원

MySQL 4.x 버전부터 Feature Request로 등록되어 있던 “Descending index” 기능이 드디어 MySQL 8.0에 도입되었다. MySQL 8.0부터는 이제 아래와 같이 역순으로 정렬되는 인덱스(Descending index)를 생성할 수 있게 되었으며, 필요에 따라서 적절히 정순(ORDER BY ASC)과 역순(ORDER BY DESC)을 혼합해서 정렬하는 작업을 인덱스를 이용할 수 있게 된 것이다.

CREATE TABLE tb_wow (
  uid BIGINT PRIMARY KEY,
  age SMALLINT,
  score SMALLINT,
  INDEX ix_score_age (score DESC, age ASC)
);

SELECT * FROM tb_wow ORDER BY score ASC,  age DESC;
SELECT * FROM tb_wow ORDER BY score DESC, age ASC ;

아마도 MySQL 8.0 이전에도 Descending index가 지원되었다고 생각했을 수도 있는데, MySQL 8.0 이전에는 문법만 지원되고 실제 Descending index가 지원되는 것은 아니었다. 또한 Ascending index를 Forward scan하는 것과 Backward scan하는 것만으로 Descending index의 요건을 충분히 만족한다고 생각할 수도 있지만, 실제 그렇지 못한 경우도 많다.

MySQL 8.0에 도입된 Descending index가 필요한 가장 큰 이유는 이미 살펴본 예제와 같이 정순(ORDER BY ASC)과 역순(ORDER BY DESC) 정렬을 섞어서 여러 컬럼으로 정렬하는 경우일 것이다. 그런데 Descending index가 필요한 이유가 오직 이것뿐일까?

Descending index를 사용해야 하는 또 다른 이유

MySQL 8.0 이전 버전을 사용하면서 역순 정렬이 필요한 경우에는, 크게 성능에 대한 고려 없이 지금까지 Ascending index를 생성하고 “ORDER BY index_column DESC” 쿼리로 인덱스를 Backward scan으로 읽는 실행 계획을 사용해왔다. 이제 Ascending index를 Forward scan하는 경우와 Backward scan하는 경우의 성능 비교를 간단히 예제로 한번 살펴보자.

우선 아래와 같이 information_schema.COLUMNS 테이블의 레코드를 복사해서 대략 1천2백여만 건의 레코드를 가지는 테이블을 만들어 보자.

CREATE TABLE t1 (
  tid INT NOT NULL AUTO_INCREMENT,
  TABLE_NAME VARCHAR(64),
  COLUMN_NAME VARCHAR(64),
  ORDINAL_POSITION INT,
  PRIMARY KEY(tid)
) ENGINE=InnoDB;

INSERT INTO t1 SELECT NULL, TABLE_NAME, COLUMN_NAME, ORDINAL_POSITION FROM information_schema.COLUMNS;
INSERT INTO t1 SELECT NULL, TABLE_NAME, COLUMN_NAME, ORDINAL_POSITION FROM t1; -- // 12번 실행

mysql> SELECT COUNT(*) FROM t1;
+----------+
| COUNT(*) |
+----------+
| 12619776 |
+----------+

이제 이 테이블을 풀 스캔 하면서 정렬만 수행하는 쿼리를 아래와 같이 한번 실행해보자. 아래 두 쿼리는 테이블의 프라이머리 키를 Forward scan 또는 Backward scan으로 읽어서 마지막 레코드 1건만 반환하게 된다. 첫번째 쿼리는 tid 컬럼의 값이 가장 큰 레코드 1건을 그리고 두번째 쿼리는 tid 컬럼의 값이 가장 작은 레코드 1건을 반환하게 된다. 하지만 LIMIT .. OFFSET .. 부분의 쿼리로 인해서, 실제 MySQL 서버는 테이블의 모든 레코드를 스캔해야만 한다. (이 쿼리는 모든 레코드를 스캔하는 작업은 하지만, 화면에는 레코드 1건만 출력하려고 LIMIT .. OFFSET .. 옵션을 추가한 것임)

SELECT * FROM t1 ORDER BY tid ASC  LIMIT 12619775,1;
SELECT * FROM t1 ORDER BY tid DESC LIMIT 12619775,1;

위 두 쿼리의 실행 결과는 어떤 차이를 보여줄지 먼저 한번 예측해보자. 지금까지는 많은 사용자들이 두 쿼리가 동일한 실행 시간을 보여줄 것이라 믿어 의심치 않았을 것이다. 당연히 그렇게 작동해야 하니까 고려 대상조차 아니었다.

mysql> SELECT * FROM t1 ORDER BY tid ASC LIMIT 12619775,1;
row in set (4.14 sec)
row in set (4.15 sec)
row in set (4.15 sec)
row in set (4.14 sec)
row in set (4.15 sec)

mysql> SELECT * FROM t1 ORDER BY tid DESC LIMIT 12619775,1;
row in set (5.35 sec)
row in set (5.35 sec)
row in set (5.35 sec)
row in set (5.36 sec)
row in set (5.35 sec)

테스트 환경 (CPU Bound 테스트)

CPU : Intel(R) Xeon(R) CPU E5-2620 v3 @ 2.40GHz (x 24)

MEMORY : 64GB

DISK : NVME SSD

TABLE SIZE (Disk) : 883MB

MySQL configuration

innodb_buffer_pool_instances=2

innodb_buffer_pool_size=30G

innodb_adaptive_hash_index=OFF

1천2백여만건을 스캔하는데, “1.2초 정도의 차이쯤이야!!”라고 생각할 수도 있다. 하지만 비율로 따져보면, 역순 정렬 쿼리가 정순 정렬 쿼리보다 28.9% 더 시간이 걸리는 것을 확인할 수 있다. 하나의 인덱스를 정순으로 읽느냐 또는 역순으로 읽느냐에 따라서 이런 차이가 발생한다는 것은 쉽게 이해하기 어렵다.

Backward index scan이 느린 이유

MySQL 서버의 InnoDB 스토리지 엔진에서 (많은 사용자들이 이미 잘 알고 있듯이) Forward & Backward index scan 페이지(블록) 간의 양방향 연결 고리(Double linked list)를 통해서 전진(Forward)하느냐 후진(Backward)하느냐의 차이만 있다. 이것만 보면 Forward와 Backward index scan의 성능 차이는 이해되지 않는다.

실제 InnoDB에서 Backward index scan이 Forward index scan에 비해서 느릴 수밖에 없는 2가지 이유를 가지고 있다.

페이지 잠금이 Forward index scan에 적합한 구조
페이지 내에서 인덱스 레코드는 단방향으로만 연결된 구조 (Forwarded single linked link)

1. 페이지 잠금이 `Forward index scan`에 적합한 구조

InnoDB의 페이지 잠금 방식은 Forward index scan을 중심으로 구현되어 있는데, Forward index scan으로 인덱스 리프 페이지를 읽을 때는, 아래 코드 샘플과 같이 페이지의 잠금을 획득할 때에는 Forward scan 순서대로 잠금을 걸고 다시 잠금을 해제하게 된다.

void
btr_pcur_move_to_next_page(
/*=======================*/
    btr_pcur_t* cursor, /*!< in: persistent cursor; must be on the
                last record of the current page */
    mtr_t*      mtr)    /*!< in: mtr */
{
    // ... skip ...

    page = btr_pcur_get_page(cursor);
    next_page_no = btr_page_get_next(page, mtr);

    // ... skip ...
    
    buf_block_t*    block = btr_pcur_get_block(cursor);

    // 다음 페이지(next page)를 찾아서, 잠금 획득
    next_block = btr_block_get(
        page_id_t(block->page.id.space(), next_page_no),
        block->page.size, mode,
        btr_pcur_get_btr_cur(cursor)->index, mtr);

    next_page = buf_block_get_frame(next_block);

    // ... skip ...
    
    // 다음 페이지(next page) 잠금 획득후, 현재 페이지(이전 페이지)의 잠금을 해제
    btr_leaf_page_release(btr_pcur_get_block(cursor), mode, mtr);

    // ... skip ...
}

이제 Backward index scan시에 페이지 잠금을 획득하는 코드 샘플을 한번 살펴보자.

void
btr_pcur_move_backward_from_page(
/*=============================*/
        btr_pcur_t*     cursor, /*!< in: persistent cursor, must be on the first
                                record of the current page */
        mtr_t*          mtr)    /*!< in: mtr */
{
    // ... skip ...
    // 커서의 현재 상태 백업
    btr_pcur_store_position(cursor, mtr);

    mtr_commit(mtr);  // Mini-transaction 커밋 (페이지 잠금 해제)

    mtr_start(mtr);   // Mini-transaction 시작

    // BTR_SEARCH_PREV 모드로 커서 복구
    btr_pcur_restore_position(latch_mode2, cursor, mtr);

    page = btr_pcur_get_page(cursor);

    prev_page_no = btr_page_get_prev(page, mtr);

    /* For intrinsic table we don't do optimistic restore and so there is
       no left block that is pinned that needs to be released. */
    if (!dict_table_is_intrinsic(
         btr_cur_get_index(btr_pcur_get_btr_cur(cursor))->table)) {

        if (prev_page_no == FIL_NULL) {
        } else if (btr_pcur_is_before_first_on_page(cursor)) {

            prev_block = btr_pcur_get_btr_cur(cursor)->left_block;
            // 불필요시 현재 페이지 잠금 해제
            btr_leaf_page_release(btr_pcur_get_block(cursor),
                                        latch_mode, mtr);

            page_cur_set_after_last(prev_block,
                                        btr_pcur_get_page_cur(cursor));
         } else {
            /* The repositioned cursor did not end on an infimum
               record on a page. Cursor repositioning acquired a latch
               also on the previous page, but we do not need the latch:
               release it. */

            prev_block = btr_pcur_get_btr_cur(cursor)->left_block;
            // 불필요시 이전 페이지(Backward page) 잠금 해제
            btr_leaf_page_release(prev_block, latch_mode, mtr);
        }
    }

    cursor->latch_mode = latch_mode;
    cursor->old_stored = false;
}

대략 코드를 읽어보면, (100% 이해는 어렵더라도) 대략 Backward index scan으로 이전 페이지로 넘어가는 과정은 아래와 같은 처리가 발생하는 것을 알 수 있다.

커서의 상태를 저장하고 내부 미니 트랜잭션을 커밋해서 미니 트랜잭션 버퍼를 글로벌 리두 로그 버퍼로 복사
미니 트랜잭션을 재시작
커서의 상태를 다시 복구 (이 과정에서 현재 블록이 이동되는 것을 막기 위해서 pinning을 하고 필요에 따라서 현재 블록과 이전 블록(Backward block)의 잠금을 획득)

InnoDB의 B-Tree 리프 페이지는 Double linked list로 연결되어 있기 때문에, 사실 어느 방향이든지 이동 자체는 차이가 없다. 하지만 InnoDB 스토리지 엔진에서는 페이지 잠금 과정에서 데드락을 방지하기 위해서 B-Tree의 왼쪽에서 오른쪽 순서(Forward)로만 잠금을 획득하도록 하고 있다. 그래서 Forward index scan에서는 다음 페이지 잠금 획득이 매우 간단하지만, Backward index scan에서 이전 페이지 잠금을 획득하는 과정은 상당히 복잡한 과정을 거치게 된다.

이런 차이로 인서 많은 페이지를 스캔해야 하는 Index scan에서는 잠금 획득으로 인한 쿼리 처리 지연이 발생하게 된다.

2. 페이지 내에서 인덱스 레코드는 단방향으로만 연결된 구조

InnoDB 스토리지 엔진이 특정 레코드를 검색할 때, B-Tree를 이용해서 검색 대상 레코드(인덱스 엔트리)가 저장된 페이지(Block)까지는 검색할 수 있다. 하지만 그 페이지 내에도 수많은 레코드가 저장되어 있는데, 일반적으로 20바이트 키를 저장하는 인덱스 페이지(16K)라면, 대략 600여개 이상의 레코드가 저장될 수 있다. InnoDB 스토리지 엔진이 600여개 레코드를 하나씩 다 순차적으로 비교한다면 레코드 검색이 상당히 느릴 것이다. 그래서 InnoDB 스토리지 엔진은 하나의 페이지내에서 순차적으로 정렬된 레코드 4~8개 정도씩을 묶어서 대표 키(가장 큰 인덱스 엔트리 키 값)를 선정한다. 그리고 이 대표 키들만 모아서 별도의 리스트를 관리하는데, 이를 페이지 디렉토리(Page directory)라고 한다. 아래 그림은 “Jeremy Cole”이 그린 “InnoDB 자료 구조“중에서 “InnoDB page directory” 그림을 캡처한 것이다. 이미 충분히 이해하기 쉽도록 그려져 있어 그대로 차용하고자 한다.

InnoDB 스토리지 엔진은 하나의 페이지에서 특정 키 값을 검색할 때 Page directory를 이용해서 바이너리 서치(Binary search) 방식으로 검색 대상 키를 포함하는 대표 키를 검색한다. 그리고 대표 키를 찾으면 그때부터는 인덱스 키 값 순서대로 연결된 Linked list를 이용해서 대상 레코드를 검색하게 된다.

그런데 Double linked list로 연결된 B-Tree의 리프 페이지 구조와는 달리, 페이지 내부의 레코드(인덱스 엔트리)들은 Single linked list 구조로 구성되어 있다. Single linked list는 Ascending index에서는 키 값이 오름차순으로 정렬되어서 Linked list로 구성되는 것이다. 만약 Descending index라면 키 값이 내림차순으로 정렬되어서 Linked list 구성될 것이다. 그래서 Ascending index에서 Forward index scan은 Linked list를 그대로 따라가기만 하면 되지만, Backward index scan은 그렇게 간단하지 않다.

아래 코드 샘플은 Forward index scan을 할 때 하나의 페이지에서 Page directory를 이용해서 다음 레코드를 찾아오는 방법을 보여주고 있다.

UNIV_INLINE
const rec_t*
page_rec_get_next_low(
/*==================*/
    const rec_t*    rec,    /*!< in: pointer to record */
    ulint        comp)    /*!< in: nonzero=compact page layout */
{
    // ... skip ...
    
    page = page_align(rec);
    offs = rec_get_next_offs(rec, comp);

    // ... skip ...

    return(page + offs);
}

Forward index scan은 단순히 Linked list만 따라가면 되기 때문에 코드 역시 매우 간단하다. 이제 Backward index scan의 코드 샘플을 한번 비교해보자.

UNIV_INLINE
const rec_t*
page_rec_get_prev_const(
/*====================*/
    const rec_t*    rec)    /*!< in: pointer to record, must not be page
                infimum */
{
    // ... skip ...
    
    page = page_align(rec);
    // Page directory를 검색해서, 레코드를 저장하고 있는 슬롯(Directory)을 검색
    slot_no = page_dir_find_owner_slot(rec);
    // 현재 레코드가 저장된 슬롯의 이전 슬롯(slot_no-1)을 가져와서
    slot = page_dir_get_nth_slot(page, slot_no - 1);
    // 해당 슬롯의 대표 레코드를 가져온다
    rec2 = page_dir_slot_get_rec(slot);

    if (page_is_comp(page)) {
        // ... skip ...
    } else {
        while (rec != rec2) {
            prev_rec = rec2;
            // 다음 레코드가 자기 자신일 때까지 loop를 실행
            // 자기 자신 레코드를 찾으면, 그 이전 레코드가 이전 레코드이므로
            rec2 = page_rec_get_next_low(rec2, FALSE);
        }
    }

    // ... skip ...

    // 검색된 이전 레코드 리턴
    return(prev_rec);
}

사실 코드 자체는 많이 복잡하지 않지만, Page directory별로 4~8개 정도의 레코드(인덱스 키 엔트리)가 저장되므로 while loop을 평균적으로 2~4번 정도씩 실행해야 이전 레코드(Backwrad index scan)를 찾을 수 있는 것이다.

Backward index scan의 서비스 영향도

이로써 Backward index scan이 Forward index scan보다 느린 이유를 알게 되었다. 그렇다면 실제 Backward index scan을 사용하면 서비스가 엄청나게 느려지는 것일까? Forward index scan과 Backward index scan의 실제 서비스 영향도는 일반적으로 그렇게 크지 않았다. 아주 랜덤한 키 값으로 검색해서 Index range scan을 실행하는 경우 대략 아래 그래프와 같이 10% 정도의 쿼리 스루풋 차이를 보였으며, CPU 사용량의 차이는 미미했다 (Test thread를 16개 정도로 안정적인 쿼리 처리가 가능한 상황에서의 테스트 결과).

Forward index scan

set @random_tid=floor(rand()*12994454);

select tid from t1 where tid>=@random_tid order by tid ASC limit 50;

Backward index scan

set @random_tid=floor(rand()*12994454);

select tid from t1 where tid<=@random_tid order by tid DESC limit 50;

하지만 정렬된 Index의 특정 부분(인덱스의 앞부분 또는 끝부분)을 집중적으로 읽는 경우, 44% 정도의 스루풋 차이를 보이며 CPU 사용량도 큰 차이를 보였다.

Forward index scan

select tid from t1 order by tid ASC limit 1000;

Backward index scan

select tid from t1 order by tid DESC limit 1000;

첫번째 테스트에서는 매번 쿼리가 실행될 때마다 인덱스 스캔 위치를 랜덤하게 선택하도록 해서, 실제 각 쓰레드간의 페이지 잠금 경합이 그다지 심하지 않았던 것이다. 그리고 인덱스를 Backward scan하는 데 추가로 더 걸린 시간은 쿼리의 처리 시간에 그다지 크게 영향을 미치지 않았던 것이다. 그런데 이번 테스트 케이스(두번째 테스트 케이스)에서는 유입되는 모든 쿼리가 동일 페이지에 집중되는 상황인데, 이때에는 쿼리를 실행중인 쓰레드들끼리 경합을 하면서 아주 짧은 페이지 잠금 시간이 더 길어지는 효과를 내게 된 것이다. 그래서 쿼리의 Ascending index scan보다 Descending index scan이 훨씬 더 많은 쿼리 시간이 걸리게 된 것이다.

두번째 테스트 케이스에서는 1000건의 레코드를 조회(LIMIT 1000)하도록 했는데, 이 건수가 하나의 페이지에 저장된 레코드 건수보다 크면 Ascending과 Descending index scan의 성능 차이가 커졌으며, 조회 건수가 페이지에 저장된 레코드 건수보다 적으면 Ascending과 Descending index scan의 성능 차이는 줄어들었다. 즉 성능 차이에 영향을 미치는 2가지 구조적 이유 중에서, 페이지 잠금이 Forward index scan에 적합한 구조가 더 크게 영향을 미치는 것으로 보인다.

사용자에게 일반적으로 노출되는 잠금(Table & Row lock, …) 이외에도 InnoDB 스토리지 엔진에서는 내부적으로 페이지의 레코드를 접근할 때마다, 페이지에 대해서 잠금을 걸어야 한다. 이때 InnoDB 스토리지 엔진은 RW-lock(Semaphore)이 아닌 Mutex를 사용하기 때문에 읽기 쿼리와 쓰기 쿼리뿐만 아니라 읽기 쿼리들끼리도 페이지 잠금을 점유하기 위해서 경쟁해야 하기 때문에, 동시 쓰레드가 많아지면 많아질수록 성능 영향도는 더 커지게 되는 결과를 만들게 될 것으로 보인다.

참고로, 테이블의 데이터 파일을 구성하는 B-Tree의 각 페이지에 저장된 레코드 건수가 성능을 영향을 미치게 되는데, 이 테스트를 위한 테이블의 각 페이지에 저장된 레코드 건수는 아래와 같았다.

+-------+-------+-------+-------+-------+---------+--------------+
| page  | index | level | data  | free  | records | min_key      |
+-------+-------+-------+-------+-------+---------+--------------+
|     4 |    52 |     0 |  7443 |  8737 |     146 | tid=1        |
|     5 |    52 |     0 | 15035 |  1079 |     282 | tid=147      |
|     6 |    52 |     0 | 15084 |  1036 |     271 | tid=429      |
...
| 55354 |    52 |     0 | 15083 |  1057 |     229 | tid=12993805 |
| 55355 |    52 |     0 | 15095 |  1049 |     222 | tid=12994034 |
| 55356 |    52 |     0 | 13049 |  3107 |     199 | tid=12994256 |
+-------+-------+-------+-------+-------+---------+--------------+

Ascending vs Descending index의 선택 기준

일반적으로 인덱스를 ORDER BY ... DESC하는 쿼리가 소량의 레코드를 드물게 실행되는 경우라면, Descending index를 굳이 고려할 필요는 없어 보인다.

SELECT * FROM tab WHERE userid=? ORDER BY score DESC LIMIT 10;

예를 들어서 아래와 같은 쿼리인 경우, 아래 2가지 인덱스 모두 적절한 선택이 될 수 있다.

  - Ascending index  : INDEX (userid ASC, score ASC)
  - Descending index : INDEX (userid DESC, score DESC)

하지만 위 쿼리가 조금 더 많은 레코드를 빈번하게 실행된다면, Ascending index보다는 Descending index가 더 효율적이라고 볼 수 있다. 또한 많은 쿼리가 인덱스의 앞쪽만 또는 뒤쪽만 집중적으로 읽어서 인덱스의 특정 페이지 잠금이 병목 지점이 될 것으로 예상된다면, 적절히 Descending index를 생성하는 것이 경합 감소에 도움이 될 것으로 보인다.

물론 ASC와 DESC 정렬을 혼합해서 동시에 사용하는 쿼리라면, 당연히 ASC와 DESC를 섞어서 인덱스를 생성해야 하므로, 고민할 필요 없이 쿼리의 정렬 조건에 맞게 인덱스를 생성하면 될 것으로 보인다.

그리고 Ascending index와 Descending index의 선택은 MySQL 서버가 CPU Bound로 쿼리를 처리할 때의 이야기이다. 만약 MySQL 서버가 데이터를 읽기 위해서 매번 Disk를 읽어야 한다면, Ascending index나 Descending index의 구조적 장단점은 Disk 반응 속도(Latency)에 이미 상쇄되어 버리기 때문에 그다지 쿼리 처리상 성능 영향 요소가 아니라고 볼 수 있다.

사용하면서 알게 된 Reactor, 예제 코드로 살펴보기

Tue, 29 May 2018 17:00:00 +0900

Reactor는 Pivotal의 오픈소스 프로젝트로, JVM 위에서 동작하는 논블럭킹 애플리케이션을 만들기 위한 리액티브 라이브러리입니다. Reactor는 RxJava 2와 함께 Reactive Stream의 구현체이기도 하고, Spring Framework 5부터 리액티브 프로그래밍을 위해 지원되는 라이브러리입니다. RxJava에 익숙한 필자가 Reactor를 사용하면서 느낀 것은 RxJava와 많은 공통점이 있으며 큰 차이점이 있다면 Reactor는 최소 Java8에서 동작하며 Java8의 피쳐를 잘 지원한다는 점입니다.
본문에서는 필자가 스프링 프레임워크 WebFlux환경에서 개발을 할 때 Reactor에 대해서 알게 된 점을 공유하고자 합니다. 쉬운 이해를 위해 간단한 자바 애플리케이션 예제 코드를 통해 소개하도록 하겠습니다. 본 예제에서 사용된 자바 버전은 Java8이며, Reactor 버전은 3.1.7 임을 알려드립니다.

Mono와 Flux

우선 예제에 들어가기 앞서 Mono와 Flux의 차이점을 알 필요가 있습니다. Mono는 0-1개의 결과만을 처리하기 위한 Reactor의 객체이고, Flux는 0-N개인 여러 개의 결과를 처리하는 객체입니다. Reactor를 사용해 일련의 스트림을 코드로 작성하다 보면 보통 여러 스트림을 하나의 결과를 모아줄 때 Mono를 쓰고, 각각의 Mono를 합쳐서 여러 개의 값을 여러 개의 값을 처리하는 Flux로 표현할 수도 있습니다. 자세한 설명은 Reactor의 Mono reference와 Flux reference를 읽어 보시면 됩니다.
Mono와 Flux모두 Reactive Stream의 Publisher 인터페이스를 구현하고 있으며, Reactor에서 제공하는 풍부한 연산자들(operators)의 조합을 통해 스트림을 표현할 수 있습니다. 예를 들어 Flux에서 하나의 결과로 값을 모아주는 reduce연산자는 Mono를 리턴하고, Mono에서 flatMapMany라는 연산자를 사용하면 하나의 값으로부터 여러 개의 값을 취급하는 Flux를 리턴할 수 있습니다. 그리고 Publisher인터페이스에 정의된 subscribe메서드를 호출함으로써 Mono나 Flux가 동작하도록 할 수 있습니다. 자세한 내용은 예제 코드를 통해 다루도록 하겠습니다.

하나의 스트림에서 여러 스트림으로

여기서는 하나의 스트림에서 여러 개의 스트림으로 갈라질 때 Flux와 Mono를 어떻게 적절히 섞어서 사용했는지 예제를 통해 보도록 하겠습니다.

과일바구니 예제

여기서 사용할 예제는 과일바구니 예제입니다. basket1부터 basket3까지 3개의 과일바구니가 있으며, 과일바구니 안에는 과일을 중복해서 넣을 수 있습니다. 그리고 이 바구니를 List로 가지는 baskets가 있습니다. Flux.fromIterable에 Iterable type의 인자를 넘기면 이 Iterable을 Flux로 변환해줍니다. 이러한 예제를 만든 이유는 스프링에서 WebClient를 이용하여 특정 HTTP API를 호출하고 받은 JSON 응답에 여러 배열이 중첩되어 있고, 여기서 또 다른 API를 호출하거나 데이터를 조작하는 경우가 있었습니다.
처음엔 API 호출을 몇 번하고 원하는 데이터로 조작하고자 하는 필요에서 시작하였지만, 받은 데이터를 막상 적절한 연산자의 조합으로 하려고 접근하다 보면 어려워지고, 어떻게 하면 쉽게 풀 수 있을까 생각해보았습니다. 그리고 간단한 예제를 만들어 연습해보는 것이 좋겠다는 생각이 들었습니다. 따라서 이러한 연습을 위해 여기서는 과일바구니를 표현하는 리스트를 만들어 시작을 해보도록 하겠습니다.

public static void main(String [] args) {
    final List<String> basket1 = Arrays.asList(new String[]{"kiwi", "orange", "lemon", "orange", "lemon", "kiwi"});
    final List<String> basket2 = Arrays.asList(new String[]{"banana", "lemon", "lemon", "kiwi"});
    final List<String> basket3 = Arrays.asList(new String[]{"strawberry", "orange", "lemon", "grape", "strawberry"});
    final List<List<String>> baskets = Arrays.asList(basket1, basket2, basket3);
    final Flux<List<String>> basketFlux = Flux.fromIterable(baskets);
}

바구니 속 과일 종류(중복 없이) 및 각 종류별 개수 나누기

basketFlux로부터 중복 없이 각 과일 종류를 나열하고, 각 과일이 몇 개씩 들어있는지 각 바구니마다 출력하는 코드를 작성해보도록 하겠습니다. 단순히 Reactor사용 없이 Java에 있는 방법만으로 함수형 프로그래밍도 아닌 절차 지향적으로 생각해보면 어렵지 않습니다. baskets를 for each loop로 돌면서 Set에 담을 수도 있고, 각 과일의 개수는 Set에 처음으로 들어가는 경우 Map에 key값으로 1 값을 갖게 만들고, 그 외의 경우는 1씩 증가시키는 방법으로 과일의 개수를 셀 수 있습니다.
그렇다면 basketFlux에서 연산자들의 조합으로 어떻게 이 2가지 과제를 할 수 있을까요? 우선 이 기능을 추상화한 연산자를 찾아볼 수 있습니다. 중복 없이 값을 처리하는 연산자로 distinct가 있고, 각 key별로 스트림을 관리하기 원한다면 key를 기준으로 Flux로 그룹을 묶을 수 있는 groupBy가 있습니다. 그리고 스트림에서 내려주는 값의 개수를 셀 수 있는 count라는 연산자도 있습니다. 그런데 이것들을 하려면 basketFlux로부터 각각의 바구니들을 꺼내야 합니다. 이렇게 값을 꺼내서 새로운 Publisher로 바꿔줄 수 있는 연산자로는 대표적으로 flatMap, flatMapSequential, concatMap이 있습니다. flatMap은 리턴하는 Publisher가 비동기로 동작할 때 순서를 보장하지 않으므로, 순서 보장을 하려면 flatMapSequential 또는 concatMap을 사용해야 하는데, 여기서는 concatMap을 사용하도록 하겠습니다. flatMapSequential과 concatMap의 차이는 concatMap은 인자로 지정된 함수에서 리턴하는 Publisher의 스트림이 다 끝난 후에 그다음 넘어오는 값의 Publisher스트림을 처리하지만, flatMapSequential은 일단 오는 대로 구독하고 결과는 순서에 맞게 리턴하는 역할을 해서, 비동기 환경에서 동시성을 지원하면서도 순서를 보장할 때 쓰이는 것이 차이점입니다.

basketFlux.concatMap(basket -> {
    final Mono<List<String>> distinctFruits = Flux.fromIterable(basket).distinct().collectList();
    final Mono<Map<String, Long>> countFruitsMono = Flux.fromIterable(basket)
            .groupBy(fruit -> fruit) // 바구니로 부터 넘어온 과일 기준으로 group을 묶는다.
            .concatMap(groupedFlux -> groupedFlux.count()
                .map(count -> {
                    final Map<String, Long> fruitCount = new LinkedHashMap<>();
                    fruitCount.put(groupedFlux.key(), count);
                    return fruitCount;
                }) // 각 과일별로 개수를 Map으로 리턴
            ) // concatMap으로 순서보장
            .reduce((accumulatedMap, currentMap) -> new LinkedHashMap<String, Long>() { {
                putAll(accumulatedMap);
                putAll(currentMap);
            }}); // 그동안 누적된 accumulatedMap에 현재 넘어오는 currentMap을 합쳐서 새로운 Map을 만든다. // map끼리 putAll하여 하나의 Map으로 만든다.
    // return ???
});

distinctFruits와 countFruitsMono라는 이름으로 앞서 소개한 연산자들로 조합을 했습니다.
과일의 중복이 없도록 모으는 distinctFruits에서는 List로 변환하기 위해 Flux에서 제공하는 collectList를 이용했습니다. 이렇게 하면 Flux에서 넘어오는 각각의 항목들을 하나의 리스트로 모아주는 Mono로 변환할 수 있습니다. countFruitsMono에서는 각 과일의 개수를 key, value를 갖는 Map (자료구조 interface Map)으로 모으기 위해서 reduce를 이용해서 합쳤습니다. 순서를 보장하기 위해 concatMap과 LinkedHashMap을 사용했습니다. 이렇게 하면 순서대로 넘어온 각 과일의 개수를 순서대로 Map에 순서에 따라 모을 수 있습니다.
이렇게 만들어 놓으니 이제 이 둘을 각각 합쳐서 하나의 스트림으로 리턴을 해줘야 합니다. 리턴 부분은 아직 물음표로 주석처리를 해두었습니다. 이렇게 2개의 스트림을 하나의 객체를 리턴하는 Publisher로 합쳐주는 연산자로는 Flux.zip이 있습니다. 이것을 합쳐줄 객체를 만들기 위해 FruitInfo라는 클래스를 정의하고 zip연산자를 사용해보도록 하겠습니다.

FruitInfo 클래스

public class FruitInfo {
    private final List<String> distinctFruits;
    private final Map<String, Long> countFruits;

    public FruitInfo(List<String> distinctFruits, Map<String, Long> countFruits) {
        this.distinctFruits = distinctFruits;
        this.countFruits = countFruits;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;

        FruitInfo fruitInfo = (FruitInfo) o;

        if (distinctFruits != null ? !distinctFruits.equals(fruitInfo.distinctFruits) : fruitInfo.distinctFruits != null)
            return false;
        return countFruits != null ? countFruits.equals(fruitInfo.countFruits) : fruitInfo.countFruits == null;
    }

    @Override
    public int hashCode() {
        int result = distinctFruits != null ? distinctFruits.hashCode() : 0;
        result = 31 * result + (countFruits != null ? countFruits.hashCode() : 0);
        return result;
    }

    @Override
    public String toString() {
        return "FruitInfo{" +
                "distinctFruits=" + distinctFruits +
                ", countFruits=" + countFruits +
                '}';
    }
}

zip연산자로 합친 예제

basketFlux.concatMap(basket -> {
    final Mono<List<String>> distinctFruits = Flux.fromIterable(basket).distinct().collectList();
    final Mono<Map<String, Long>> countFruitsMono = Flux.fromIterable(basket)
            .groupBy(fruit -> fruit) // 바구니로 부터 넘어온 과일 기준으로 group을 묶는다.
            .concatMap(groupedFlux -> groupedFlux.count()
                .map(count -> {
                    final Map<String, Long> fruitCount = new LinkedHashMap<>();
                    fruitCount.put(groupedFlux.key(), count);
                    return fruitCount;
                }) // 각 과일별로 개수를 Map으로 리턴
            ) // concatMap으로 순서보장
            .reduce((accumulatedMap, currentMap) -> new LinkedHashMap<String, Long>() { {
                putAll(accumulatedMap);
                putAll(currentMap);
            }}) // 그동안 누적된 accumulatedMap에 현재 넘어오는 currentMap을 합쳐서 새로운 Map을 만든다. // map끼리 putAll하여 하나의 Map으로 만든다.
    return Flux.zip(distinctFruits, countFruitsMono, (distinct, count) -> new FruitInfo(distinct, count));
}).subscribe(System.out::println);

결과

FruitInfo{distinctFruits=[kiwi, orange, lemon], countFruits={kiwi=2, orange=2, lemon=2}}
FruitInfo{distinctFruits=[banana, lemon, kiwi], countFruits={banana=1, lemon=2, kiwi=1}}
FruitInfo{distinctFruits=[strawberry, orange, lemon, grape], countFruits={strawberry=2, orange=1, lemon=1, grape=1}}

distinct 된 리스트와 각 과일의 개수를 묶은 리스트를 하나의 리스트로 묶어서 각각 출력하는 것을 확인할 수 있습니다. 그런데 이런 식의 방식은 너무 비효율적입니다. distinctFruits와 countFruitsMono모두 Flux.fromIterable(basket)로부터 시작해서 각각 basket을 독립적으로 순회합니다. 절차 지향으로 생각하면 하나의 for each loop 안에서 2가지를 한 번에 해결할 수 있는데 여기서는 총 2번 basket을 순회하고, 특별히 스레드를 지정하지 않았기 때문에 동기, 블록킹 방식으로 동작합니다. 논 블록킹 라이브러리의 장점을 전혀 살릴 수 없고, 효율성도 떨어집니다. 단순히 Reactor에서 제공하는 연산자들의 조합의 코드일 뿐입니다.
그래서 비동기 논 블록킹으로 동시성도 살리면서, 순차적으로 basket을 두 번 순회하지 않는 방법을 다음 코드에서 다뤄보고자 합니다.

병렬로 두 스트림 합쳐보기

Reactor나 RxJava 모두 동시성 지원을 위해 Scheduler를 제공합니다. 적절한 Scheduler를 적절한 위치에 지정함으로써 동시성과 실행 순서를 적절히 관리할 수 있습니다. 기본적으로 스케줄러를 지정하지 않고 사용하는 연산자가 특정한 스케줄러에서 동작하지 않는다면, Reactor의 Flux나 Mono는 구독할 때 현재 스레드에서 동작합니다. 따라서 위의 코드도 로그를 찍어봄으로써 스레드를 확인해보면 모든 코드가 main thread에서 동기로 실행되는 것을 확인할 수 있습니다. 스프링 프레임워크 5부터 제공하는 WebClient를 사용하면 다른 스레드로 이미 바뀌어서 비동기로 동작하겠지만, 해당 스레드 안에서도 몇몇 스트림은 병렬로 실행시키고자 할 필요가 있을 수 있습니다. 그렇다면 어떻게 스케줄러를 지정하여 1) 과일의 종류를 뽑아내는 것과 2) 과일의 개수를 뽑아내는 이 2가지 작업을 병렬로 실행할 수 있을지 알아보도록 하겠습니다.

병렬 스케줄러 (Schedulers.parallel())

Reactor의 스케줄러 관련 문서를 보면 기본적으로 제공하는 몇 가지 스케줄러에 대한 설명이 있습니다. 여기서는 병렬로 여러 개를 실행시키기 위해 Schedulers.parallel()을 사용하도록 했는데, 이 스케줄러는 병렬 실행을 위해 CPU 코어 개수만큼 워커를 만들어 병렬로 실행을 지원하는 스케줄러 입니다. RxJava에서는 Schedulers.computation()이 해당 스케줄러에 해당되는 것으로 알고 있습니다. 필자의 머신에서는 쿼드코어에 하이퍼스레딩의 영향으로 최대 8개의 스레드로 해당 스케줄러가 동작하는 것을 확인할 수 있었습니다.

subscribeOn으로 스케줄러 전환하기

subscribeOn연산자는 해당 스트림을 구독할 때 동작하는 스케줄러를 지정할 수 있습니다. 여기서 distinctFruits와 countFruitsMono가 각각 병렬로 동작하길 원하므로 subscribeOn(Schedulers.parallel())을 각각 추가하여 실행시켜 보았습니다. 그런데 이 자바 애플리케이션은 아무 결과도 확인 못하고 끝나 버립니다.

basketFlux.concatMap(basket -> {
    final Mono<List<String>> distinctFruits = Flux.fromIterable(basket).distinct().collectList().subscribeOn(Schedulers.parallel());
    final Mono<Map<String, Long>> countFruitsMono = Flux.fromIterable(basket)
            .groupBy(fruit -> fruit) // 바구니로 부터 넘어온 과일 기준으로 group을 묶는다.
            .concatMap(groupedFlux -> groupedFlux.count()
                .map(count -> {
                    final Map<String, Long> fruitCount = new LinkedHashMap<>();
                    fruitCount.put(groupedFlux.key(), count);
                    return fruitCount;
                }) // 각 과일별로 개수를 Map으로 리턴
            ) // concatMap으로 순서보장
            .reduce((accumulatedMap, currentMap) -> new LinkedHashMap<String, Long>() { {
                putAll(accumulatedMap);
                putAll(currentMap);
            }}) // 그동안 누적된 accumulatedMap에 현재 넘어오는 currentMap을 합쳐서 새로운 Map을 만든다. // map끼리 putAll하여 하나의 Map으로 만든다.
            .subscribeOn(Schedulers.parallel());
    return Flux.zip(distinctFruits, countFruitsMono, (distinct, count) -> new FruitInfo(distinct, count));
}).subscribe(System.out::println);

왜냐하면 parallel스케줄러는 데몬스레드(Deamon Thread)인데 이 애플리케이션이 처음 동작하는 main thread는 데몬스레드가 아닌 비-데몬스레드(Non-Deamon Thread)이기 때문에 main메서드가 끝나버리며 아무런 비-데몬스레드가 남지 않아 종료되기 때문입니다. 종료되지 않고 계속 동작해야 하는 서버 환경의 애플리케이션에서는 괜찮겠지만, 본 애플리케이션은 main메서드가 끝난 후 비-데몬스레드가 하나도 남아있지 않아 종료됩니다. 이를 방지하기 위해 CountDownLatch를 이용하여 await()으로 애플리케이션이 종료되지 않게 main스레드가 동작이 끝날 때까지 기다리도록 하겠습니다. 위의 스트림이 정상적으로 또는 에러가 나서 종료한 경우에 countDown메서드를 호출하여 더 이상 기다리지 않고 종료되도록 하겠습니다. 혹시 모르게 오랫동안 기다리는 경우를 막기 위해 await(2, TimeUnit.SECONDS)으로 2초 정도의 timeout을 두도록 했습니다.

basketFlux.concatMap(basket -> {
    // ... 생략
    return Flux.zip(distinctFruits, countFruitsMono, (distinct, count) -> new FruitInfo(distinct, count));
}).subscribe(
        System.out::println,  // 값이 넘어올 때 호출 됨, onNext(T)
        error -> {
            System.err.println(error);
            countDownLatch.countDown();
        }, // 에러 발생시 출력하고 countDown, onError(Throwable)
        () -> {
            System.out.println("complete");
            countDownLatch.countDown();
        } // 정상적 종료시 countDown, onComplete()
);
countDownLatch.await(2, TimeUnit.SECONDS);

결과

FruitInfo{distinctFruits=[kiwi, orange, lemon], countFruits={kiwi=2, orange=2, lemon=2}}
FruitInfo{distinctFruits=[banana, lemon, kiwi], countFruits={banana=1, lemon=2, kiwi=1}}
FruitInfo{distinctFruits=[strawberry, orange, lemon, grape], countFruits={strawberry=2, orange=1, lemon=1, grape=1}}
complete

결과가 잘 나오는 것을 확인할 수 있습니다. 병렬로 실행되는지 확인하기 위해 distinctFruits와 countFruitsMono의 각각의 시작점 Flux.fromIterable(basket)에 log()라는 메서드를 추가하도록 하겠습니다. 그러면 log()를 호출한 지점 위에서 넘어오는 값을 로그로 확인하며 디버깅하기 좋습니다. Reactor에서 유용한 메서드 중 하나입니다.

basketFlux.concatMap(basket -> {
    final Mono<List<String>> distinctFruits = Flux.fromIterable(basket).log().distinct().collectList().subscribeOn(Schedulers.parallel());
    final Mono<Map<String, Long>> countFruitsMono = Flux.fromIterable(basket).log()
            .groupBy(fruit -> fruit) // 바구니로 부터 넘어온 과일 기준으로 group을 묶는다.
            .concatMap(groupedFlux -> groupedFlux.count()
                .map(count -> {
                    final Map<String, Long> fruitCount = new LinkedHashMap<>();
                    fruitCount.put(groupedFlux.key(), count);
                    return fruitCount;
                }) // 각 과일별로 개수를 Map으로 리턴
            ) // concatMap으로 순서보장
            .reduce((accumulatedMap, currentMap) -> new LinkedHashMap<String, Long>() { {
                putAll(accumulatedMap);
                putAll(currentMap);
            }}) // 그동안 누적된 accumulatedMap에 현재 넘어오는 currentMap을 합쳐서 새로운 Map을 만든다. // map끼리 putAll하여 하나의 Map으로 만든다.
            .subscribeOn(Schedulers.parallel());
    return Flux.zip(distinctFruits, countFruitsMono, (distinct, count) -> new FruitInfo(distinct, count));
}).subscribe(
    System.out::println,  // 값이 넘어올 때 호출 됨, onNext(T)
    error -> {
        System.err.println(error);
        countDownLatch.countDown();
    }, // 에러 발생시 출력하고 countDown, onError(Throwable)
    () -> {
        System.out.println("complete");
        countDownLatch.countDown();
    } // 정상적 종료시 countDown, onComplete()
);

[ INFO] (parallel-1) | onSubscribe([Synchronous Fuseable] FluxIterable.IterableSubscriptionConditional)
[ INFO] (parallel-2) | onSubscribe([Synchronous Fuseable] FluxIterable.IterableSubscription)
[ INFO] (parallel-1) | request(unbounded)
[ INFO] (parallel-2) | request(256)
[ INFO] (parallel-1) | onNext(kiwi)
[ INFO] (parallel-2) | onNext(kiwi)
[ INFO] (parallel-1) | onNext(orange)
[ INFO] (parallel-1) | onNext(lemon)
[ INFO] (parallel-1) | onNext(orange)
[ INFO] (parallel-1) | onNext(lemon)
[ INFO] (parallel-1) | onNext(kiwi)
[ INFO] (parallel-1) | onComplete()
[ INFO] (parallel-2) | request(1)
[ INFO] (parallel-2) | onNext(orange)
[ INFO] (parallel-2) | onNext(lemon)
[ INFO] (parallel-2) | onNext(orange)
[ INFO] (parallel-2) | onNext(lemon)
[ INFO] (parallel-2) | onNext(kiwi)
[ INFO] (parallel-2) | request(1)
[ INFO] (parallel-2) | onComplete()
[ INFO] (parallel-2) | request(2)
FruitInfo{distinctFruits=[kiwi, orange, lemon], countFruits={kiwi=2, orange=2, lemon=2}}
[ INFO] (parallel-2) | cancel()
[ INFO] (parallel-2) | cancel()
[ INFO] (parallel-3) | onSubscribe([Synchronous Fuseable] FluxIterable.IterableSubscriptionConditional)
[ INFO] (parallel-4) | onSubscribe([Synchronous Fuseable] FluxIterable.IterableSubscription)
[ INFO] (parallel-3) | request(unbounded)
[ INFO] (parallel-4) | request(256)
[ INFO] (parallel-3) | onNext(banana)
[ INFO] (parallel-4) | onNext(banana)
[ INFO] (parallel-3) | onNext(lemon)
[ INFO] (parallel-4) | request(1)
[ INFO] (parallel-3) | onNext(lemon)
[ INFO] (parallel-4) | onNext(lemon)
[ INFO] (parallel-3) | onNext(kiwi)
[ INFO] (parallel-4) | onNext(lemon)
[ INFO] (parallel-3) | onComplete()
[ INFO] (parallel-4) | onNext(kiwi)
[ INFO] (parallel-4) | onComplete()
[ INFO] (parallel-4) | request(2)
[ INFO] (parallel-4) | request(1)
FruitInfo{distinctFruits=[banana, lemon, kiwi], countFruits={banana=1, lemon=2, kiwi=1}}
... 이하 생략 ...

처음엔 parallel-1과 parallel-2이 같이 동작하고, 그다음 순서로 parallel-3과 parallel-4이 같이 동작하는 것을 로그를 통해 확인할 수 있습니다.

basket당 하나의 스트림만 공유하며 과일종류와 개수 뽑아내기

지금까지 진행해온 과정에서 아쉬운 점을 좀 더 개선해보고자 합니다. distinctFruits와 countFruitsMono 모두 Flux.fromIterable(basket)에서 출발하기 때문에 병렬로 동작해도 baskets를 각각 순회하여 중복되는 동작이라는 점은 처음의 예제와 다를 게 없습니다. 또한 몇 개 안 되는 데이터를 각각 병렬로 처리하는 것은 스레드를 생성하고 컨텍스트 스위칭을 하는 비용을 생각하면 배보다 배꼽이 클 수도 있습니다. baskets을 순회하는 공통 작업을 하나의 스트림에서 하는 방법은 없을까요? 먼저 관련된 배경지식을 다루고 그 방법을 함께 살펴보도록 하겠습니다.

Hot, Cold 개념

Hot과 Cold 개념은 RxJava에도 있는 개념으로 Reactor의 Hot, Cold 개념은 공식문서 Hot vs Cold를 통해서 확인하실 수 있습니다.
간단히 설명하면 Cold는 각 Flux나 Mono를 subscribe 할 때마다 매번 독립적으로 새로 데이터를 생성해서 동작합니다. 즉, subscribe호출 전까지 아무런 동작도 하지 않고, subscribe를 호출하면 새로운 데이터를 생성합니다. 기본적으로 특별하게 Hot을 취급하는 연산자가 아닌 이상 Flux나 Mono는 Cold로 동작합니다. 따라서 지금까지 예제는 basket으로부터 값을 꺼내어 각각 따로 새로운 데이터를 생성하기 때문에 각각 중복된 작업을 새로 시작하게 동작합니다.
그러나 Hot은 구독하기 전부터 데이터의 스트림이 동작할 수 있습니다. 예를 들어서 마우스 클릭이나 키보드 입력 같은 이벤트 성은 구독여부와 상관없이 발생하고 있다가 이 이벤트를 구독하는 여러 구독자가 붙으면 해당 이벤트가 발생할 때 모두 동일한 값을 전달받을 수 있습니다. 즉, Hot에 해당하는 스트림을 여러 곳에서 구독을 하면 현재 스트림에서 나오는 값을 구독하는 구독자들은 동일하게 받을 수 있습니다.
여기서 주목할 점은 Cold를 Hot으로 바꿀 수 있는 연산자가 있다는 것입니다. 처음 Hot에 대해서 예제를 든 마우스클릭이나 키보드입력같은 이벤트를 떠올리면 쉽게 상상이 안되는데요. Cold는 구독을 하면 값을 생성하기 시작합니다. 그러나 Cold를 Hot으로 바꾸면 구독여부와 상관없이 값을 생성 안 하다가 특정 시점에 값을 생성하도록 제어하여 구독하는 구독자(Subscriber)들이 동일한 값을 받을 수 있도록 할 수 있습니다. 이에 대해 알아보겠습니다.

Connectable Flux

Connectable Flux는 Cold에서 Hot으로 바꾸기 위해서는 Connectable Flux로 변환하는 과정이 필요합니다. 공식문서에 설명되어 있듯 기본적으로 publish라는 연산자를 호출하면 바꿀 수 있습니다. 이렇게 변환된 Flux에서 connect()라는 메서드를 호출할 수 있는데, 이 메서드가 여러 구독자들이 Connectable Flux를 구독한 후 값을 생성하여 각 구독자에게 보내기 시작하게 하는 메서드입니다. 즉, 우리의 예제에서는 distinctFruits와 countFruitsMono가 구독을 모두 완료한 후에 connect()를 호출할 수 있게 해주면 됩니다. 어떻게 할 수 있을까요? 다행히 Reactor에서는 autoConnect나 refCount에 인자 값으로 최소 구독하는 구독자의 개수를 지정해서 이 개수가 충족되면 자동으로 값을 생성할 수 있게 연산자를 제공합니다. 2개의 차이점이 있다면 autoConnect는 이름 그대로 최소 구독 개수를 만족하면 자동으로 connect()를 호출하는 역할만 하고, refCount는 autoConnect가 하는 일에 더해서 구독하고 있는 구독자의 개수를 세다가 하나도 구독하는 곳이 없으면 기존 소스의 스트림도 구독을 해제하는 역할을 합니다. interval처럼 무한히 일정 간격으로 값이 나오는 스트림을 Hot으로 바꾼다면 refCount를 고려해 볼 수 있을 것입니다. 호출되는 소스가 무한으로 값을 생성한다면 더 이상 구독을 안 할 때 해제하게 refCount를 고려해볼 수 있지만, 우리는 제한된 개수의 데이터만 다 다루면 알아서 완료가 되기 때문에 autoConnect로 충분하다고 생각합니다. 참고로 여기서 소개한 2개의 연산자는 인자 값을 없을 경우 최초 구독이 발생할 때 connect()를 호출하도록 동작하고, publish().refCount()를 하나의 연산자로 추상화한 연산자를 share()라고 합니다. 그 밖에 RxJava의 ConnectableFlowable도 동일한 기능을 제공하는 것을 문서를 통해 확인할 수 있었습니다.

2개 구독자가 구독하면 자동으로 동작하게 Hot Flux 만들기

이제 코드로 공통적으로 사용하는 Flux.fromIterable(basket)을 Hot으로 만들어보도록 하겠습니다.
설명에서 소개한 바와 같이 publish().autoConnect(2)를 사용하도록 하겠습니다. 이렇게 하면 Hot으로 변환된 ConnectableFlux를 최소 2개의 구독자가 구독을 하면 자동으로 구독하는 Flux를 리턴합니다. 이를 소스코드에서는 source라는 변수에 지정하고, distinctFruits와 countFruitsMono모두 source를 공통으로 쓰도록 했습니다. Flux.fromIterable(basket) 뒤에 log()는 그대로 두었습니다.
한 번만 값이 나오나 확인하기 위함입니다. 추가로 distinctFruits와 countFruitsMono에서 subscribeOn은 제거했습니다. 굳이 몇 개 안 되는 데이터를 병렬로 돌릴 필요가 없어 보였고, 하나의 스레드에서 Flux.fromIterable(basket)가 한 번만 동작하면서 2가지 스트림으로 동작할 수 있음을 보여주기 위함입니다.

basketFlux.concatMap(basket -> {
    final Flux<String> source = Flux.fromIterable(basket).log().publish().autoConnect(2);
    final Mono<List<String>> distinctFruits = source.distinct().collectList();
    final Mono<Map<String, Long>> countFruitsMono = source
            .groupBy(fruit -> fruit) // 바구니로 부터 넘어온 과일 기준으로 group을 묶는다.
            .concatMap(groupedFlux -> groupedFlux.count()
                    .map(count -> {
                        final Map<String, Long> fruitCount = new LinkedHashMap<>();
                        fruitCount.put(groupedFlux.key(), count);
                        return fruitCount;
                    }) // 각 과일별로 개수를 Map으로 리턴
            ) // concatMap으로 순서보장
            .reduce((accumulatedMap, currentMap) -> new LinkedHashMap<String, Long>() { {
                putAll(accumulatedMap);
                putAll(currentMap);
            }}); // 그동안 누적된 accumulatedMap에 현재 넘어오는 currentMap을 합쳐서 새로운 Map을 만든다. // map끼리 putAll하여 하나의 Map으로 만든다.
    return Flux.zip(distinctFruits, countFruitsMono, (distinct, count) -> new FruitInfo(distinct, count));
}).subscribe(
    System.out::println,  // 값이 넘어올 때 호출 됨, onNext(T)
    error -> {
        System.err.println(error);
        countDownLatch.countDown();
    }, // 에러 발생시 출력하고 countDown, onError(Throwable)
    () -> {
        System.out.println("complete");
        countDownLatch.countDown();
    } // 정상적 종료시 countDown, onComplete()
);

결과

[ INFO] (main) | onSubscribe([Synchronous Fuseable] FluxIterable.IterableSubscription)
[ INFO] (main) | onNext(kiwi)
[ INFO] (main) | onNext(orange)
[ INFO] (main) | onNext(lemon)
[ INFO] (main) | onNext(orange)
[ INFO] (main) | onNext(lemon)
[ INFO] (main) | onNext(kiwi)
[ INFO] (main) | onComplete()
FruitInfo{distinctFruits=[kiwi, orange, lemon], countFruits={kiwi=2, orange=2, lemon=2}}
...이하생략

이하 결과를 생략했지만 모두 main스레드에서 값이 잘 나오는 것을 확인할 수 있습니다. 여기서는 동일한 스레드이기 때문에 countDownLatch도 필요가 없습니다. 두 군데서 구독할 때만 값이 나오는지 확인하기 위해 zip에서 쓰이는 distinctFruits의 source를 Flux.fromIterable(basket)로 바꿔서 source가 한 번만 구독되도록 해보겠습니다.

source를 한 곳 countFruitsMono에서만 구독할 때 예제

basketFlux.concatMap(basket -> {
    final Flux<String> source = Flux.fromIterable(basket).log().publish().autoConnect(2);
    final Mono<List<String>> distinctFruits = Flux.fromIterable(basket).distinct().collectList();
    final Mono<Map<String, Long>> countFruitsMono = source.
    //생략...
    return Flux.zip(distinctFruits, countFruitsMono, (distinct, count) -> new FruitInfo(distinct, count));
}).subscribe(/* 생략.. */);

구독한 후 아무 값도 나오지 않습니다. complete도 불리지 않는 것을 확인할 수 있습니다.

Hot 이후에 각 스트림만 비동기로 처리하기

위의 예제에서는 몇 개 안 되는 데이터라 같은 스레드에서 동기로 실행시켰지만, 데이터가 많아지거나 현재 스레드가 다른 일을 하게 하기 위해서 distinct나 groupBy count 등의 연산을 하는 지점은 각각 비동기로 처리하고자 하는 필요가 생길 수도 있습니다. 그러나 앞서 소개한 subscribeOn은 이런 경우 적절하지 않습니다.
subscribeOn을 호출한 객체를 구독할 때는 해당 스트림 전체가 해당 스케줄러로 다 바뀌기 때문에, Hot인 source도 2개의 구독자가 구독을 하면 subscribeOn이 지정한 스레드에서 실행되게 되며, 그러면 distinct와 count로 갈라져 나오는 부분도 같은 스레드에서 실행되기 때문입니다. 이를 확인하기 위해 기존의 예제에서 distinctFruits와 countFruitsMono에 각각 subscribeOn(Schedulers.parallel())을 붙여서 실행해보겠습니다. 이때 각각 어느 스레드에서 동작하는지 확인하기 위해 log()를 붙였습니다.

basketFlux.concatMap(basket -> {
   final Flux<String> source = Flux.fromIterable(basket).log().publish().autoConnect(2);
   final Mono<List<String>> distinctFruits = source.distinct().collectList().log().subscribeOn(Schedulers.parallel());
   final Mono<Map<String, Long>> countFruitsMono = source
           .groupBy(fruit -> fruit) // 바구니로 부터 넘어온 과일 기준으로 group을 묶는다.
           .concatMap(groupedFlux -> groupedFlux.count()
                   .map(count -> {
                       final Map<String, Long> fruitCount = new LinkedHashMap<>();
                       fruitCount.put(groupedFlux.key(), count);
                       return fruitCount;
                   }) // 각 과일별로 개수를 Map으로 리턴
           ) // concatMap으로 순서보장
           .reduce((accumulatedMap, currentMap) -> new LinkedHashMap<String, Long>() { {
               putAll(accumulatedMap);
               putAll(currentMap);
           }}) // 그동안 누적된 accumulatedMap에 현재 넘어오는 currentMap을 합쳐서 새로운 Map을 만든다. // map끼리 putAll하여 하나의 Map으로 만든다.
           .log()
           .subscribeOn(Schedulers.parallel());
   return Flux.zip(distinctFruits, countFruitsMono, (distinct, count) -> new FruitInfo(distinct, count));
}).subscribe(/* 생략.. */);

결과

[ INFO] (parallel-1) | onSubscribe([Fuseable] MonoCollectList.MonoBufferAllSubscriber)
[ INFO] (parallel-2) | onSubscribe([Fuseable] MonoReduce.ReduceSubscriber)
[ INFO] (parallel-1) | request(32)
[ INFO] (parallel-2) | request(32)
[ INFO] (parallel-1) | onSubscribe([Synchronous Fuseable] FluxIterable.IterableSubscription)
[ INFO] (parallel-1) | onNext(kiwi)
[ INFO] (parallel-1) | onNext(orange)
[ INFO] (parallel-1) | onNext(lemon)
[ INFO] (parallel-1) | onNext(orange)
[ INFO] (parallel-1) | onNext(lemon)
[ INFO] (parallel-1) | onNext(kiwi)
[ INFO] (parallel-1) | onComplete()
[ INFO] (parallel-1) | onNext([kiwi, orange, lemon])
[ INFO] (parallel-1) | onComplete()
[ INFO] (parallel-1) | onNext({kiwi=2, orange=2, lemon=2})
FruitInfo{distinctFruits=[kiwi, orange, lemon], countFruits={kiwi=2, orange=2, lemon=2}}

처음엔 parallel-1과 parallel-2로 동작하는 듯했다가 결국 source는 parallel-1에서 실행됩니다. 2개의 구독자가 구독을 한 후 source가 parallel-1에서 시작되니 그 이후 동작도 다 parallel-1에서 실행되는 것입니다. subscribeOn으로 스케줄러를 지정하여 스위칭이 되면 이후 스트림은 계속 그 스케줄러에 의해 동작되기 때문입니다.

이때 필요한 것이 publishOn연산자입니다. 이 연산자가 호출된 위치 이후에 실행되는 연산자들은 publishOn에서 지정된 스케줄러에서 실행되도록 할 수 있습니다. subscribeOn과의 차이점은 마블 다이어그램을 통해 확인하실 수 있습니다. 그렇다면 publishOn을 각각 source 바로 뒤에 호출되도록 붙여 보겠습니다.

basketFlux.concatMap(basket -> {
    final Flux<String> source = Flux.fromIterable(basket).log().publish().autoConnect(2);
    final Mono<List<String>> distinctFruits = source.publishOn(Schedulers.parallel()).distinct().collectList().log();
    final Mono<Map<String, Long>> countFruitsMono = source.publishOn(Schedulers.parallel())
            .groupBy(fruit -> fruit) // 바구니로 부터 넘어온 과일 기준으로 group을 묶는다.
            .concatMap(groupedFlux -> groupedFlux.count()
                    .map(count -> {
                        final Map<String, Long> fruitCount = new LinkedHashMap<>();
                        fruitCount.put(groupedFlux.key(), count);
                        return fruitCount;
                    }) // 각 과일별로 개수를 Map으로 리턴
            ) // concatMap으로 순서보장
            .reduce((accumulatedMap, currentMap) -> new LinkedHashMap<String, Long>() { {
                putAll(accumulatedMap);
                putAll(currentMap);
            }}) // 그동안 누적된 accumulatedMap에 현재 넘어오는 currentMap을 합쳐서 새로운 Map을 만든다. // map끼리 putAll하여 하나의 Map으로 만든다.
            .log();
    return Flux.zip(distinctFruits, countFruitsMono, (distinct, count) -> new FruitInfo(distinct, count));
}).subscribe(/* 생략.. */);

결과

...생략...
[ INFO] (main) | onNext(kiwi)
[ INFO] (main) | onNext(orange)
[ INFO] (main) | onNext(lemon)
[ INFO] (main) | onNext(orange)
[ INFO] (main) | onNext(lemon)
[ INFO] (main) | onNext(kiwi)
[ INFO] (main) | onComplete()
[ INFO] (parallel-1) | onNext([kiwi, orange, lemon])
[ INFO] (parallel-1) | onComplete()
[ INFO] (parallel-2) | onNext({kiwi=2, orange=2, lemon=2})
FruitInfo{distinctFruits=[kiwi, orange, lemon], countFruits={kiwi=2, orange=2, lemon=2}}
..중략..
[ INFO] (parallel-2) | onNext(banana)
[ INFO] (parallel-2) | onNext(lemon)
[ INFO] (parallel-2) | onNext(lemon)
[ INFO] (parallel-2) | onNext(kiwi)
[ INFO] (parallel-2) | onComplete()
[ INFO] (parallel-3) | onNext([banana, lemon, kiwi])
[ INFO] (parallel-3) | onComplete()
[ INFO] (parallel-4) | onNext({banana=1, lemon=2, kiwi=1})
FruitInfo{distinctFruits=[banana, lemon, kiwi], countFruits={banana=1, lemon=2, kiwi=1}}
...생략...

의도한 대로 각각 parallel-1과 parallel-2에서 실행되는 것을 확인할 수 있습니다. 그런데 처음에 source에서 시작할 때는 main스레드에서 시작하다가 publishOn으로 전체 스트림의 스케줄러가 바뀌면서 그다음 source는 parallel-2에서 실행되는 것을 확인할 수 있습니다. 만약 source가 하나의 지정한 스레드에서만 실행되도록 하고 싶다면 source에 subscribeOn을 추가할 수 있습니다. 이렇게 되면 source는 해당 스케줄러에 의해 동작하고 그 이후는 publishOn에 의해 바뀔 수 있습니다. 어떤 스케줄러를 지정할지는 필요에 따라 다를 듯 하지만, Reactor 공식문서 스케줄러를 읽어보고 저의 경우는 Schedulers.single()을 선택했습니다. 호출할 때마다 같은 스레드를 유지할 수 있기 때문에 매번 source를 구독할 때마다 같은 스레드에서 동작할 수 있기 때문입니다.

basketFlux.concatMap(basket -> {
    final Flux<String> source = Flux.fromIterable(basket).log().publish().autoConnect(2).subscribeOn(Schedulers.single());
    final Mono<List<String>> distinctFruits = source.publishOn(Schedulers.parallel()).distinct().collectList().log();
    final Mono<Map<String, Long>> countFruitsMono = source.publishOn(Schedulers.parallel())
            .groupBy(fruit -> fruit) // 바구니로 부터 넘어온 과일 기준으로 group을 묶는다.
            .concatMap(groupedFlux -> groupedFlux.count()
                    .map(count -> {
                        final Map<String, Long> fruitCount = new LinkedHashMap<>();
                        fruitCount.put(groupedFlux.key(), count);
                        return fruitCount;
                    }) // 각 과일별로 개수를 Map으로 리턴
            ) // concatMap으로 순서보장
            .reduce((accumulatedMap, currentMap) -> new LinkedHashMap<String, Long>() { {
                putAll(accumulatedMap);
                putAll(currentMap);
            }}) // 그동안 누적된 accumulatedMap에 현재 넘어오는 currentMap을 합쳐서 새로운 Map을 만든다. // map끼리 putAll하여 하나의 Map으로 만든다.
            .log();
    return Flux.zip(distinctFruits, countFruitsMono, (distinct, count) -> new FruitInfo(distinct, count));
}).subscribe(/* 생략.. */);

결과 (설명상 필요한 부분만)

[ INFO] (single-1) | onNext(kiwi)
[ INFO] (single-1) | onNext(orange)
[ INFO] (single-1) | onNext(lemon)
[ INFO] (single-1) | onNext(orange)
[ INFO] (single-1) | onNext(lemon)
[ INFO] (single-1) | onNext(kiwi)
[ INFO] (single-1) | onComplete()
[ INFO] (parallel-1) | onNext([kiwi, orange, lemon])
[ INFO] (parallel-1) | onComplete()
[ INFO] (parallel-2) | onNext({kiwi=2, orange=2, lemon=2})
FruitInfo{distinctFruits=[kiwi, orange, lemon], countFruits={kiwi=2, orange=2, lemon=2}}
[ INFO] (single-1) | onNext(banana)
[ INFO] (single-1) | onNext(lemon)
[ INFO] (single-1) | onNext(lemon)
[ INFO] (single-1) | onNext(kiwi)
[ INFO] (single-1) | onComplete()
[ INFO] (parallel-3) | onNext([banana, lemon, kiwi])
[ INFO] (parallel-4) | onNext({banana=1, lemon=2, kiwi=1})
[ INFO] (parallel-3) | onComplete()
FruitInfo{distinctFruits=[banana, lemon, kiwi], countFruits={banana=1, lemon=2, kiwi=1}}
[ INFO] (single-1) | onSubscribe([Synchronous Fuseable] FluxIterable.IterableSubscription)
[ INFO] (single-1) | onNext(strawberry)
[ INFO] (single-1) | onNext(orange)
[ INFO] (single-1) | onNext(lemon)
[ INFO] (single-1) | onNext(grape)
[ INFO] (single-1) | onNext(strawberry)
[ INFO] (single-1) | onComplete()
[ INFO] (parallel-5) | onNext([strawberry, orange, lemon, grape])
[ INFO] (parallel-6) | onNext({strawberry=2, orange=1, lemon=1, grape=1})
[ INFO] (parallel-5) | onComplete()

결과는 필요한 부분만 뽑아 보았습니다. source는 single-1이라는 스레드에서 항상 동작하고, 그 이후에는 각각 parallel-1과parallel-2, parallel-3과parallel-4, parallel-5과parallel-6으로 동작하는 것을 확인할 수 있습니다.

디버깅과 테스팅

디버깅

Reactor나 RxJava와 같이 스트림을 여러 연산자의 조합으로 체이닝 동작하는 방식의 라이브러리는 디버깅이나 테스트가 쉽지 않은 점이 있습니다. 예를 들어 에러가 났을 때 어떤 연산자에서 오류가 났는지 알기 어려워서 중간에 doOnNext와 같은 연산자로 로그를 찍어야 할 수도 있습니다. 물론 Reactor에는 앞서 소개한 log()라는 연산자도 있어서 편리하게 로그를 찍어볼 수 있습니다. 그러나 이 log()도 직전의 값만 출력해주기 때문에 사이사이에 log()를 넣어주는 것도 번거로워 보였습니다. 그래서 Reactor의 디버깅 관련된 문서를 찾아보았습니다. Flux나 Mono를 구독하기 전 애플리케이션 시작 단계에서 Hooks.onOperatorDebug();를 호출하면 디버깅 모드를 활성화할 수 있으며, 이럴 경우 에러가 발생했을 때 출력되는 스택트레이스에 시작부터 에러가 났을 때까지 연산자의 목록을 모두 볼 수 있습니다. 공식문서에서는 다음과 같이 나온다고 소개되어 있는데, 해보니 실제로 에러가 났을 때 연산자들이 나오는 것을 확인할 수 있었습니다.

Error has been observed by the following operator(s):
        |_        Flux.map ⇢ reactor.guide.FakeRepository.findAllUserByName(FakeRepository.java:27)
        |_        Flux.map ⇢ reactor.guide.FakeRepository.findAllUserByName(FakeRepository.java:28)
        |_        Flux.filter ⇢ reactor.guide.FakeUtils1.lambda$static$1(FakeUtils1.java:29)
        |_        Flux.transform ⇢ reactor.guide.GuideDebuggingExtraTests.debuggingActivatedWithDeepTraceback(GuideDebuggingExtraTests.java:40)
        |_        Flux.elapsed ⇢ reactor.guide.FakeUtils2.lambda$static$0(FakeUtils2.java:30)
        |_        Flux.transform ⇢ reactor.guide.GuideDebuggingExtraTests.debuggingActivatedWithDeepTraceback(GuideDebuggingExtraTests.java:41)

테스팅

테스트 코드 작성을 위해 테스트 관련 문서를 참고해 보았습니다. io.projectreactor:reactor-test를 의존성으로 추가하고 StepVerifier를 통해 테스트 코드를 작성할 수 있었습니다. StepVerifier.create로 테스트할 객체를 만들 때 인자로 테스트 대상이 되는 Flux나 Mono를 넘깁니다. 그리고 테스트에 필요한 메서드들을 연달아서 호출해서 기대한 값이 나왔는지 확인할 수 있습니다. 여기서는 expectNext와 verifyComplete를 이용해서 next로 넘어온 값이 기대한 값인지 그리고 complete이 호출되었는지 검증해보도록 하겠습니다. 여기서 getFruitsFlux()는 지금까지 만든 예제와 관련된 Flux를 리턴하는 메서드이며, JUnit4를 이용하여 테스트해보았습니다.

@Test
public void testFruitBaskets() {
    final FruitInfo expected1 = new FruitInfo(
        Arrays.asList("kiwi", "orange", "lemon"),
        new LinkedHashMap<String, Long>() { {
            put("kiwi", 2L);
            put("orange", 2L);
            put("lemon", 2L);
        }}
    );
    final FruitInfo expected2 = new FruitInfo(
        Arrays.asList("banana", "lemon", "kiwi"),
        new LinkedHashMap<String, Long>() { {
            put("banana", 1L);
            put("lemon", 2L);
            put("kiwi", 1L);
        }}
    );
    final FruitInfo expected3 = new FruitInfo(
        Arrays.asList("strawberry", "orange", "lemon", "grape"),
        new LinkedHashMap<String, Long>() { {
            put("strawberry", 2L);
            put("orange", 1L);
            put("lemon", 1L);
            put("grape", 1L);
        }}
    );
    StepVerifier.create(getFruitsFlux())
        .expectNext(expected1)
        .expectNext(expected2)
        .expectNext(expected3)
        .verifyComplete();
}

테스트는 잘 통과되는 것을 확인할 수 있고, 값을 값을 하나 바꿔서 실패하게 만들면 어떤 값이 스트림에서 넘어왔는데 기대하는 값은 무엇인지도 잘 출력됩니다. 그리고 verifyComplete이 리턴하는 타입은 Duration이란 타입인데, 여기에는 테스트하는 동안 걸린 시간 정보가 들어가게 됩니다.

마치며

지금까지 간단한 예제를 이용해 Reactor의 연산자들을 어떻게 조합해서 데이터를 변환하는지, Cold에서 Hot으로 변환하는 과정을 이해하고 하나의 스트림에서 여러 스트림으로 나갈 때 어떻게 하는지, 스케줄러를 지정해 어떻게 실행 컨텍스트를 전환시켰는지 살펴보았습니다. 그밖에 log()나 디버깅 모드를 활용해서 에러난 위치나 스트림의 흐름을 쉽게 볼 수 있다는 것도 유용해 보였습니다.
스프링 프레임워크 5부터는 리액티브 프로그래밍을 할 수 있게 Reactor와 RxJava를 사용하도록 지원해주고 있지만, 현재는 실무에서 보편적으로 사용되는 것 같아 보이진 않습니다. RxJava에 익숙하다면 Reactor도 공식문서를 보고 적응하기 어렵지 않아 보였습니다.
사용하면서 어려운 점은 연산자만으로 해결이 가능한 부분이 있음에도 아직 익숙하지 않고 높은 학습곡선과 수많은 연산자들을 다 모르기에, 중간에 Flux나 Mono에서 값을 꺼내서 동기방식과 명령형 프로그래밍으로 조작한 후 다시 넣어야 하나 유혹을 받는 경우가 온다는 것입니다. 이렇게 하면 함수형 프로그래밍과 명령형 프로그래밍이 섞여 있어 코드도 복잡해지고 중간에 block 같은 연산자를 써서 논 블록킹 라이브러리의 장점을 살릴 수 없는 상황이 오게 된다는 것입니다. 따라서 Reactor를 사용하여 리액티브 프로그래밍을 해야 하는 상황을 고려한다면, 1) 필요한 이유, 2) 추후 코드 변경 시 유지보수성, 3)함께 협업하는 사람들의 숙련도나 관심 등을 고려하여 선택해야 한다고 생각합니다. 필자의 경험으로는 여러 스레드를 전환해야 해서 순서가 복잡해져서 그 사이에 순서를 잘 관리하거나, 복잡하게 데이터를 변환해야 하는 경우, 그리고 여러 스트림을 블록킹으로 처리하기보다 논 블록킹으로 처리하기 좋은 경우에 Reactor나 RxJava같은 라이브러리가 유용해 보였습니다.
이번에 Reactor를 사용해보면서 정리했으면 좋겠다는 패턴이 보여서 간단한 예제로 정리를 했습니다. 본문에서 Reactor의 모든 것을 다룰 순 없었어도, 필요한 부분이 참고가 돼서 Reactor 같은 라이브러리로 리액티브 프로그래밍을 하는데 도움이 되었으면 좋겠습니다.

NumPy와 C++ Extensions의 성능 비교

Tue, 15 May 2018 10:00:00 +0900

파이썬은 놀라운 생산성을 발휘하는 언어입니다. 하지만 성능 문제는 늘 발목을 잡게 합니다. 이 문제를 극복하는 방법으로 일반적으로 C Extension을 작성하는 방법이 권장되며, 여기서는 표준 편차를 구하는 함수를 작성하여 순수 파이썬의 성능과 NumPy, 각종 C++ Extensions의 성능을 비교해 보도록 합니다.

Python

표준 편차를 구하는 파이썬 코드는 아래와 같이 작성할 수 있습니다.

def mean(lst):
    return sum(lst) / len(lst)

def standard_deviation(lst):
    m = mean(lst)
    variance = sum([(value - m) ** 2 for value in lst])
    return math.sqrt(variance / len(lst))

NumPy

NumPy로는 매우 간단하게 한 줄로 처리 가능합니다.

np.std(lst)

NumPy를 사용하면 코드가 간단해지고, 일반적으로 NumPy는 C로 최적화한 매우 효율적인 라이브러리로 알려져 있으나 NumPy는 싱글 코어와 대형 배열에 최적화된 라이브러리라는 한계가 존재합니다. 실제로 배열의 크기가 100개 이내인 경우 NumPy는 순수 파이썬 구현 보다도 오히려 낮은 성능을 보입니다.

C++ Extension

여기서는 C++로 Extension을 작성하여 성능을 최적화 해보도록 합니다. C++로 표준 편차를 구하는 코드는 아래와 같이 작성했습니다.

double standardDeviation(std::vector<double> v) {
    double sum = std::accumulate(v.begin(), v.end(), 0.0);
    double mean = sum / v.size();

    double squareSum = std::inner_product(v.begin(), v.end(), v.begin(), 0.0);
    return sqrt(squareSum / v.size() - mean * mean);
}

C++을 파이썬과 연동 하려면 계산 코드 외에도 wrapper 함수를 작성해야 합니다.

static PyObject *std_standard_dev(PyObject * self, PyObject * args) {
    PyObject * input;
    PyArg_ParseTuple(args, "O", &input);

    int size = PyList_Size(input);

    std::vector<double> list;
    list.resize(size);

    for (int i = 0; i < size; i++) {
        list[i] = PyFloat_AS_DOUBLE(PyList_GET_ITEM(input, i));
    }

    return PyFloat_FromDouble(standardDeviation(list));
}

이 wrapper 함수는 파이썬 리스트를 받아와 값을 하나씩 끄집어 낸 다음 std::vector에 담아 C++ 함수에 전달하는 역할을 합니다. wrapper에는 이외에도 여러가지 처리를 위한 boilerplate 코드가 들어갑니다. 매우 번거로운 작업이고 저 역시도 wrapper 코드를 작성하며 사소한 실수로 무수한 컴파일 오류를 맞이 해야만 했습니다.

번거로운 작업입니다.

Cython

Cython은 CPython과 이름이 비슷하여 혼동될 수 있으나 전혀 다릅니다. 원래의 목적은 Pyrex 기반의 파이썬 코드를 작성하면 이를 C로 변환해 성능을 최적화 해주는 컴파일러입니다. 그러나 외부 C 라이브러리를 랩핑 하는데도 매우 유용합니다. 아울러 C 뿐만 아니라 C++도 네이티브로 지원합니다. 여기서는 C++ 함수를 랩핑하는 용도로 사용했으며, 아래와 같이 코드를 작성했습니다.

cdef extern from "std.h":
    double standardDeviation(vector[double])

def standard_dev(lst):
    # This pre-conversion has some performance improvements.
    cdef vector[double] v = lst

    return standardDeviation(v)

헤더를 읽어 함수를 정의한 다음 파이썬 스타일의 함수에서 C++ 함수의 리턴값을 전달합니다. C++과 파이썬이 오묘한 형태로 결합되어 있습니다. 이 부분은 장점이자 단점이 될 수 있는데 기존 파이썬의 함수를 그대로 사용할 수 있는 장점이 있는 반면 기존 파이썬 함수의 낮은 성능 또한 그대로 반영됩니다. 상식적으로 파이썬 코드가 단순히 C 코드로 변환되었다고 성능이 개선되진 않습니다. (물론 약간의 효과는 있습니다.) cdef를 이용해 변수를 C/C++ 네이티브로 선언하고 주요 계산 알고리즘은 외부 C/C++ 함수로 따로 작성해서 랩핑해야 진정한 성능 개선 효과를 기대할 수 있습니다.

원래 Cython은 파이썬의 리스트를 C++ std::vector로 자동으로 컨버전 하지만 여기서는 조금이나마 성능을 개선하고자 상단에 컨버전을 직접 정의했습니다. 컨버전을 직접 구현할때는 14줄이 필요했으나 여기서는 단 한 줄로 가능했습니다.

pybind11

pybind11는 “Seamless operability between C++11 and Python”라는 모토로 최근에 등장한 C++ 전용 헤더 라이브러리 입니다. ctypes를 사용할 수 있는 C의 잇점이 있다면 pybind11은 C++에서만 사용이 가능합니다. 파이썬 연동을 마치 C++ 코드의 연장선 처럼 부드럽게^seamless 할 수 있습니다. cmake도 잘 지원하여 빌드나 IDE 연동도 편리합니다. 앞서 Cython이 파이썬 중심의 라이브러리 였다면 pybind11는 C++ 중심의 라이브러리라 할 수 있습니다.

랩핑 코드 또한 파이썬으로 작성했던 Cython과 달리 아래와 같이 C++로 작성합니다.

PYBIND11_MODULE(stdpy, m) {
    m.def("standard_dev", &standardDeviation);
}

Cython과 마찬가지로 오토 컨버전을 지원하며 헤더를 include 하면 나머지는 자동으로 처리됩니다.

#include <pybind11/stl.h>

C++에 최적화 되어 있으므로 성능이 매우 좋을 것 같지만 아쉽게도 Cython 보다 못한 성능을 보여줍니다. 특히 오토 컨버전은 편리하지만 별도로 제어할 수 없으며, 이로 인한 성능 저하가 뚜렷합니다.

실험 결과

최대 1만개까지 배열의 표준 편차를 구하는 성능 테스트 결과는 아래와 같습니다.

순수 파이썬 구현의 성능은 따로 언급할 필요가 없을듯 하며, 앞서 잠깐 언급했지만 기대를 모았던 NumPy의 성능이 그다지 높지 않습니다. 이는 대형 배열에 최적화 되어 있기 때문이며 이후에 이어지는 대형 배열의 성능 테스트에선 좋은 성능을 확인할 수 있습니다.

C++ 구현은 직접 구현하든 Cython, pybind11을 사용하든 어느쪽이든 좋은 성능을 보여줍니다. 그러나 일반적으로 직접 wrapper를 작성하는 쪽이 가장 성능이 좋으며 오토 컨버전을 지원하는 Cython과 pybind11는 그 만큼의 성능 저하가 있습니다. 특히 pybind11 쪽의 성능 저하가 뚜렷합니다.

Cython w/ class

Cython w/ class는 Cython 구현에 type conversion을 없애기 위해 별도로 구현한 방식입니다. 사실 이 테스트는 NumPy에게 지나치게 유리한데, 왜냐면 파이썬 리스트를 np.array()로 컨버전 하는 것을 벤치마크 바깥에서 별도로 진행했기 때문입니다. 밀리 세컨드 단위로 수행되는 벤치마크에서 NumPy 컨버전은 매우 무거운 편이기도 하고(만약 따로 컨버전 하지 않으면 가장 나쁜 성능이 나옵니다.) 과학 계산에서 NumPy는 사실상 표준 라이브러리의 위치에 있기 때문에 이미 원본 데이터가 NumPy 타입임을 감안하여 어드밴티지를 부여했습니다.

그러나, NumPy를 제외한 다른 모든 구현은 type conversion이 포함되며 C++은 직접 컨버전 코드를 작성했고, 나머지는 오토 컨버전이 되도록 처리했습니다. 테스트는 각 100번씩 수행되므로 컨버전이 필요 없는 NumPy에 비해 다른 구현은 모두 100번씩 별도로 컨버전되는 오버헤드가 발생해 공정한 비교가 될 수 없습니다.

이를 개선하고자 C++ 클래스를 Cython으로 구현했고 미리 컨버전한 값을 private 변수로 갖고 있다가 벤치마크시 계산만 하는 방식으로 최적화 했습니다. 따라서 계산 코드가 다른 구현과 다릅니다.

cyc_rands = stdcyc.pystd(rands)
print('stdcyc: ', cyc_rands.standard_dev())

그 결과 가장 좋은 성능을 확보할 수 있었습니다.

5만개까지 대형 배열로 성능 테스트한 결과는 아래와 같습니다.

먼저 순수 파이썬으로 계산한 결과는 다른 것과 비교가 불가능 할 정도로 느려 아예 비교에서 제외했습니다. 여기서는 NumPy, C++, Cython, Cython w/ class, pybind11만 비교했는데 앞서 다소 실망스런 모습을 보였던 NumPy가 좋은 모습을 보여줍니다. 일정 갯수를 넘어서면서 부터 가장 좋은 성능을 보이며, 배열의 크기가 늘어나도 전체 계산 속도는 크게 증가하지 않습니다. 대형 배열에 최적화된 C 라이브러리의 진가가 드러나는 순간입니다. 컨버전을 배제한 Cython w/ class 조차도 NumPy에 비해 성능이 낮습니다.

참고로 벤치마크는 각 100번씩 수행해 결과를 측정했는데, 실제 프로덕션에서는 같은 계산을 반복하지 않기 때문에 각 1번만 수행하여 NumPy와 Cython w/ class의 비교를 10만개까지 측정해봤습니다.

약 3만개 정도 부터 근소한 차이로 NumPy의 성능이 앞서는 것을 확인할 수 있습니다.

Cython w/ threads

일반적으로 NumPy만 잘 사용해도 충분한 성능을 얻을 수 있습니다. 그러나 NumPy에는 결정적인 한계가 있는데 싱글 코어에 최적화 되어 있다는 점입니다. 애초에 GIL로 인해 멀티 쓰레드가 유명무실한 파이썬과 마찬가지로 NumPy 또한 싱글 코어에 최적화 되어 있다는 한계가 있으며, 멀티 코어를 제대로 활용하기 위해선 C/C++로 쓰레드 프로그래밍을 해야 합니다. 이 때문에 텐서플로를 포함한 대부분의 딥러닝 프레임워크 또한 C++에서 멀티 코어를 활용하며(이 글에선 언급하지 않았지만 SWIG로 파이썬과 연동하여) 계산을 수행하는 방식으로 구현되어 있습니다.

C++11에서 지원하는 std::thread를 이용해 sum, squaredSum을 멀티 쓰레드로 계산해 최적화 해보도록 합니다.

int start = i * round(v.size() / NUM_THREADS);
int end = (i == NUM_THREADS - 1) ? v.size() : (i + 1) * round(v.size() / NUM_THREADS);

ths.push_back(std::thread(&stddev::calc, this, &v, start, end, std::ref(sum), std::ref(squaredSum)));

각 쓰레드별로 분할하여 시작(start)과 끝(end)으로 지정할 인덱스를 계산하고 전체 배열이 포함된 매우 큰 std::vector를 포인터로 넘깁니다. 만약 포인터가 아닌 밸류로 넘긴다면 성능 저하가 심각해 질 것입니다. 계산 함수에서는 포인터의 밸류를 조회하여 start에서 end까지 값을 끄집어내 연산을 수행합니다.

for (int i = start; i < end; i++) {
    double it = (*v).at(i);

    th_sum += it;
    th_squaredSum += it * it;
}

sum += th_sum;
squaredSum += th_squaredSum;

결과를 담을 변수는 C++ 레퍼런스로 넘겼으며, 가장 마지막에 전체 값을 한 번만 업데이트 합니다. 사실 처음 코딩할때는 실수로 레퍼런스 변수를 for loop 사이에 넣고 매 번 업데이트 했는데, 그렇게 할 경우 쓰레드를 4개만 생성해도 변수를 업데이트 하기 위한 atomic lock 경쟁이 생겨 심각한 성능 저하를 가져옵니다. 위 코드 처럼 loop 바깥에서 수행하여 성능을 개선했으며, 컴파일러가 따로 경고하지 않아 실수하기 쉽기 때문에 주의해야 합니다.

성능 최적화를 위해 포인터와 레퍼런스를 번갈아 사용해봤는데 이에 따른 성능 이슈는 없었습니다. 둘 중 선호하는 쪽을 사용하면 되며, 일반적으로 C++에서는 안전하고 편리한 레퍼런스를 사용하는 편이 권장됩니다.

성능 비교

기존에 NumPy와 근사할 정도로 성능이 좋았던 Cython w/ class와 쓰레드로 구현한 Cython w/ threads의 성능을 비교해보도록 합니다. 성능 향상을 극대화 하기 위해 대형 배열로 정했으며 2억개의 엘리먼트로 비교를 진행했습니다.

200,000,000 elements
--
generate rands:  28.931 seconds.
type conversion(stdcyc):  14.959 seconds.

stdcyc elapsed:  2.494 seconds.

랜덤을 생성하는데만 해도 약 30초가 걸리는 대형 배열이고, Cython w/ class로 2.49초 정도 소요됩니다.

200,000,000 elements
--
generate rands:  29.194 seconds.
type conversion(stdcyt):  14.139 seconds.

[C++] std::vector loading: 1600 ms
[C++] NUM_THREADS: 8
[C++] thread execution elapsed: 76 ms
stdcyt elapsed:  1.843 seconds.

Cython w/ threads로는 1.84초에 완료되어 25% 정도 성능 개선 효과가 있습니다. 맥북 프로에서 진행하여 맥북의 코어 갯수인 8개만큼 쓰레드를 돌렸으며, 쓰레드를 8개나 풀 가동 했음에도 불구하고 성능 개선 효과가 그리 크진 않습니다. 이유는 벡터 로딩에 1.6초나 걸렸기 때문이며, 이처럼 정작 연산이 아닌 엉뚱한 곳에 병목이 있을 수 있으므로 주의깊게 디버깅하는 것이 매우 중요합니다.

8개의 쓰레드가 데이터를 나눠 각 쓰레드별 최대 76ms 이내에 모든 연산을 완료 했습니다. 만약 쓰레드를 하나만 사용했다면 600ms가 걸렸을 텐데 이처럼 멀티 쓰레드 프로그래밍으로 524ms를 단축시킬 수 있었습니다.

참고로 두 테스트를 동시에 실행하면 대형 변수의 메모리 복사로 인해 서로간의 성능에 영향을 끼치므로 각각 따로 테스트하여 비교했습니다.

결론

앞서 성능 비교에서 벡터 로딩에만 1.6초가 걸리는걸 확인할 수 있었습니다. 또한 type conversion에만 14초가 걸렸습니다. 이 처럼 대형 배열에서 파이썬과 C/C++ 네이티브 타입(여기서는 C++의 std::vector)의 type conversion, 변수의 메모리 복사는 상당한 오버헤드를 발생시키며, 이를 최소화 하는 것이 무엇보다 중요합니다. 포인터나 레퍼런스를 잘 활용하여 줄일 수 있는 부분은 가능한 줄이는 편이 좋으며, 여기서도 처음부터 벡터를 포인터로 가져오면 시간을 획기적으로 줄일 수 있으나 NumPy와 비교를 위해 일부러 복사하는 방식을 택했습니다. 프로덕션에는 당연히 포인터를 사용해야 합니다.

아울러 직접 모든 wrapper를 작성하는 일은 많은 고난이 뒤따르기 때문에 성능을 약간 타협하여 Cython 또는 pybind11를 택하는 편이 최적의 선택입니다. 그 중에서도 Cython의 성능이 돋보이며, C++ 연동 방식은 pybind11가 좀 더 우아한 편이지만 오토 컨버전의 성능 이슈는 하루빨리 해결되어야 할 과제입니다.

참고

이 문서에서 사용한 표준 편차를 구하는 함수와 최초 C++ 구현은 Speeding up Python and NumPy: C++ing the Way를 참조했으며 이를 fork 하여 C++ wrapper를 개선하고 Cython, pybind11 바인딩을 추가했습니다. 전체 코드는 아래 깃헙에 올려 두었습니다.

likejazz/PythonCExtensions - GitHub

분산 웹 캐시 (Wcache)의 개선과정 - Part 2

Mon, 23 Oct 2017 12:00:00 +0900

Overview

Part 1: 분산 웹 캐시에서는 카카오의 트래픽을 처리하고 있는 Wcache에 대한 간략한 소개를 하였습니다. 이전 버전의 Wcache는 기본적으로 준수한 응답속도를 보이고 있었지만, metadata를 집중된 DB에 저장하는 방식 및 기타 구조상의 문제로 인한 성능 안정성 문제, 그리고 기능상의 문제들이 잠재되어 있었습니다. 본 포스트에서는 이전 버전의 Wcache에서 어떠한 구조적 문제가 있었는지, 또한 이를 어떻게 해결하였는지를 다룹니다.

Part 2: Wcache 저장 구조 변경

저장 구조 변경 - 성능 안정화

기존 BigFile 저장방식의 문제점

이전 포스트에 나와있듯이 Wcache는 metadata 정보를 컨텐츠와는 별도로 DB에 저장해 두고 있었습니다. 메모리에 올라와 있는 LRU hashtable에 원하는 컨텐츠가 캐싱되어 있지 않다면, SQLite DB에 컨텐츠 키를 가지고 어떤 BigFile에 있는지, 헤더의 길이는 얼마나 되는지 등의 정보들을 조회를 해야 하는 것이죠.

위 과정은 평상시 읽는 과정에서는 문제가 없었지만, block 단위로 컨텐츠를 디스크에 저장할 때와 BigFile이 가득 차 오래된 BigFile block들을 replace 할 때 문제가 발생합니다. n개의 컨텐츠가 하나의 block에 있을 경우 BigFile에서의 디스크 write operation은 한 번만 발생하지만 SQLite DB에서의 metadata 관련 정보 업데이트는 n번 발생하게 됩니다. 특히, 주기적으로 용량 확보를 위해 오래된 블록 또는 BigFile 전체를 비워야 하는 경우, DB 전체를 스캔해 관련 항목을 삭제하는 작업(ex: “DELETE FROM table WHERE BigFileNo = 1”)으로 인한 부하가 증가하게 되고, 결국 이러한 작업이 있을 때마다 Wcache의 성능은 일시적으로 평상시에 비해 최대 80%까지 하락하는 모습을 보이게 됩니다. DB 부하를 줄이기 위해 block 크기를 줄이는 방법을 고려해 보았으나 그만큼 디스크 write operation 횟수가 증가합니다. 게다가 구조상 block 크기보다 큰 컨텐츠를 저장할 수 없기 때문에 그만큼 캐시 효율이 감소하게 됩니다.

이러한 중앙 집중적인 SQLite구조에는 한계가 있다고 파악하였고, 근본적으로 저장 구조를 개선해야 된다고 결론 내렸습니다.

Journaling BigFile (JBF)

중앙 집중적인 SQLite대신 각 BigFile에 metadata를 함께 저장하는 방식을 취하기로 하였습니다. 이를 위해 SQLite처럼 journaling이 가능하면서 안정적으로 metadata와 컨텐츠를 저장하기 위해 자체적으로 Journaling BigFile, 일명 JBF를 개발해 적용하였습니다.

JBF는 위 그림과 같이 BigFile 내부에 block-level journal(write-ahead log), space map, B-Tree 등을 가지고 있어 BigFile 개별로 복구, 조회, 포맷이 가능합니다. 또한, Metadata 정보를 JBF 내부 B-Tree에 적재하고, 컨텐츠 body는 JBF의 일반 영역에 저장합니다. JBF는 BigFile별 포맷이 매우 빠르게 이루어 지기 때문에 기존 버전과는 다르게 가득 찬 BigFile을 비울 때 발생하는 성능 하락 이슈가 없습니다. 컨텐츠를 여러 block에 걸쳐 쓰는것이 가능하기 때문에 저장 가능한 컨텐츠의 최대 크기 제한 역시 BigFile 전체 크기로 늘어나게 됩니다.

JBF를 이용한 저장구조

기존 SQLite에 의존했던 탐색 로직을 개편하기 위해서는 어떠한 JBF에 컨텐츠가 존재하는지를 캐시 key 만으로 알아내야 합니다. 이를 위해 JBF들을 몇 개의 그룹으로 나누고, 캐시 key를 그룹 개수로 modular 연산하여 특정 그룹을 정하고, 해당 그룹 내부의 있는 JBF들에서 순차적으로 탐색하도록 구성하였습니다.

전체 JBF들을 동시에 사용하지 않고 그룹으로 묶어 사용하는 이유는 효율적인 디스크 용량 활용이 가능해지기 때문입니다. JBF는 BigFile 단위로 포맷을 하게 되는데, 그룹을 만들지 않고 모든 JBF에 골고루 저장하게 되면 JBF들이 가득 차게 되는 시점이 비슷해지고, 이후 오래된 캐시를 비우는 작업 시 전체 JBF들을 비슷한 시기에 비워야 하는 이슈가 생깁니다. 그렇게 되면 순간적으로 hit율이 낮아짐은 물론 새롭게 컨텐츠를 캐싱하는 양도 많아져 소스 서버의 부하까지 발생하게 됩니다. 반면 그룹을 묶어서 사용하게 되면 그룹 별로 하나씩만 비울 수 있게 되어 순간적인 캐시 hit율 하락을 완화시킬 수 있습니다.

또한 각 그룹별로 write가 이루어지는 JBF들의 순번이 정해져 있고 컨텐츠가 쌓이는 속도도 일정하기 때문에 이점을 활용해 디스크 locality도 증가시킬 수 있습니다. Wcache 초기 세팅 시 각 그룹에서 같은 순번에 있는 JBF들을 물리적 디스크상에서 인접하게 생성시키면(group 1의 1번 JBF, group 2의 1번 JBF, … group 1의 2번 JBF, …) 컨텐츠를 디스크에 캐싱할 때 인접한 구역을 쓰기 때문에 HDD를 사용하는 경우 디스크의 seek time을 줄일 수 있는 중요한 요인이 됩니다.

Bloom filter를 통한 JBF 성능 최적화

JBF들의 그룹을 만드는 구조는 한 가지 치명적인 단점이 있습니다. 바로 중복된 JBF 탐색입니다. 위 그림에서 볼 수 있듯이 캐싱된지 오래된 컨텐츠를 읽기 위해서는 쓸모없는 JBF 탐색이 들어가야 하는데, 이는 개편 후 초기 테스트 시 Wcache에 쌓이는 컨텐츠가 많아질수록 성능을 저하시키는 주요 원인이 되었습니다.

이를 해결하기 위해 각 JBF 앞단에 bloom filter¹를 적용하게 됩니다. Bloom filter는 어떤 원소가 집합에 속하였는지 확률적으로 알 수 있게 하는 함수입니다. 약간의 false positive만 가지고 있고 false negative는 없기 때문에 해당 필터를 통과하는 경우에만 JBF 조회를 하도록 변경을 함으로써 기존 대비 2 ~ 5배의 응답속도를 향상시켰습니다.

Metadata 저장구조 변경 - Vary Object 관리 개선

HTTP 헤더 중에는 ‘Vary’라는 항목이 있습니다. 동일한 URL에 대해 요청을 하더라도 요청한 사용자의 특징(User Agent, Accept Encoding, Origin 등등)에 따라 서로 다른 응답을 해 주기 위해서 존재하는 헤더입니다. 따라서 웹 캐시에서는 vary 헤더를 확인하고 해당 헤더에서 명시하는 조건에 따라 동일 URL이라 하더라도 다른 종류의 컨텐츠를 캐싱하고, 제공해야 합니다.

기존 Wcache에서는 이런 vary object에 대한 고려가 충분하지 않았습니다. 아래 그림처럼 vary object 저장은 가능했지만, 리스트 형식으로 원하는 vary object가 나올 때까지 탐색해야 했고 그마저도 vary 헤더가 없는 normal object와 동시에 캐싱을 할 수 없는 이슈가 있었습니다.

이를 해결하기 위해 저장 구조 설계 시 처음부터 vary object의 존재를 고려하였습니다. Metadata를 두 개의 타입 (Meta / Info)으로 나누어, Meta에는 현재 Wcache에 캐싱되어있는 vary object의 종류와 info의 포인터를, Info에는 실제 컨텐츠의 정보 (offset, header 등)를 담도록 설계하였습니다. 이 구조를 통해 여러 개의 vary object가 있어도, normal object와 공존해도 문제없이 사용자가 원하는 컨텐츠를 제공할 수 있게 됩니다.

Lock 구조 개선 - 동일 컨텐츠 접근 시 생기는 병목현상 제거

Wcache는 느슨한 구조의 actor model²을 가지고 있습니다. 사용자의 요청이 들어오면 다수의 actor(HTTP 요청 수신 / 응답 / 컨텐츠 처리 / 소스 서버 통신)를 돌아다니며 처리됩니다. 각 actor는 I/O multiplexing 형식으로 동시에 여러 개의 요청을 다룰 수 있게 하여 최대한의 성능을 낼 수 있도록 설계되었습니다. 그러나 디스크 I/O를 하는 actor는 내부적으로 blocking read / write를 수행하게 되어 단일 thread로는 원활한 성능을 얻을 수 없기에 내부적으로 multi thread 모델을 채택하였습니다.

그러다보니 해당 구조에서는 캐싱된 컨텐츠를 처리하는 actor에서 컨텐츠를 조회하고 검증하는 동안 불가피하게 캐시 key에 대해 광범위한 lock 처리를 해야 했습니다. 이 lock은 컨텐츠가 새로 갱신되거나 삭제(Purge)될 때를 위해 필요하지만, 이전 구조에서는 read / write lock 구분이 없었습니다. 이는 동일 컨텐츠를 집중적으로 요청할 때 해당 컨텐츠의 초당 처리수가 제한되는 문제를 야기합니다. 일반적으로 특정 이벤트³로 인해 트래픽이 증가하는 경우, 소수의 인기 있는 컨텐츠들이 집중적으로 요청되는 경우가 많으므로 이러한 성능 제한은 급작스런 트래픽 대응 시 잠재적인 문제가 될 수 있습니다.

이를 해결하기 위해 먼저 read / write lock을 분리하여 평시에 컨텐츠가 집중적으로 요청되는 경우 동시에 접근이 가능할 수 있도록 하였습니다. 더 나아가 lock의 범위를 줄이고 추후 유지보수를 쉽게 하기 위해 순수하게 디스크 I/O 부분과 캐시 컨텐츠를 검증하는 부분을 별도의 actor로 분리해 조금 더 효율적인 처리를 할 수 있도록 변경하였습니다. 이 조치는 동일 컨텐츠 요청에 대해 기존 대비 약 3배의 성능 향상을 이루어냅니다.

결과

이번 개편은 개발부터 성능 안정화까지 약 1년에 걸쳐 진행되었습니다. 기존 Wcache 구조의 핵심 부분들을 전부 뜯어고치는 대 작업이었습니다. 이번 개편을 통해 더 안정적인 성능을 이끌어 낼 수 있었고, 기능적인 측면들 - vary object 저장 유연화, 컨텐츠 크기 제한 해제, regular expression 기반 대량 캐시 퍼지 등 - 에서 큰 발전을 이룰 수 있게 되었습니다.

Bloom Filter ↩
Actor Model ↩
다음 앱에서의 뉴스속보 및 날씨알림 등의 푸시, 카카오톡 채널탭 뱃지 등.. ↩