Concurrent System에서의 Non-Heap Memory를 이용한 IPC 성능 향상

JDK 1.5를 기준으로 작성하는 글

Introduce

일반적인 Java 프로그램처럼 데이터가 JVM Heap 상에서 생성되고 JVM Garbage Collection에 의해 Lifecycle이 관리되는 프로그램이 아닌, JVM Heap을 사용하지 않는(Non-Heap) 데이터를 생성하고 관리(Management)할 수 있다.

"JVM Heap을 사용하지 않아 Garbage Collecting 대상이 되지 않는 데이터"를 다룰 수 있다는 것이 무슨 의미냐 하면... GC로 인한 Page Fault가 발생되지 않기에 IPC(Inter-Process Communication) 퍼포먼스 향상을 꾀할 수 있게 된다는 말.

또한 일반적으로 사용되는 방법인 네트워크를 이용한 데이터 교환보다 월등히 빠른데, 무의미한 헤더를 페이로드에 붙이지 않으며, 파일을 매개로 하기에 이러한 퍼포먼스 향상이 가능케 되는 것이다.

이는 java.nio 패키지의 ByteBuffer 클래스를 이용한다.

`ByteBuffer` IPC Implementation

가장 먼저 (1) I/O 작업을 수행할 파일을 생성한 후, (2) 이를 이용해 ByteBuffer 클래스의 인스턴스를 생성한다. (3) 마지막으로 I/O 작업을 진행하며 데이터를 교환하는 방식으로 ByteBuffer 클래스를 이용한 IPC 메커니즘의 구현이 가능한 것.

말로 풀어 쓰니 어려워 보이는 것이지 코드로 보면 정말 간단하다.


String basePath = System.getProperty("java.io.tmpdir");
String filename = "data";
String path = basePath + filename; // Linux에서는 basePath 뒤에 '/'가 붙지 않음을 유의

ByteBuffer buf = new RandomAccessFile(path, "rw") // (1)
	.getChannel()
	.map(FileChannel.MapMode.READ_WRITE, 0, 100L); // (2)

// (3)
buf.put(0, (byte)10);
int value = buf.getInt(0); // value := 2 (= 0x10)

위 코드로 data 라는 파일의 0 번재 인덱스에 0x10 이라는 값이 Write 되었으며, 값을 가져오는 프로세스 역시 동일한 방법을 이용해 Read하는 방식으로 동작한다.

이렇게 파일에 대한 I/O 작업을 통해 두 프로세스가 통신(Communication)을 할 수 있게 되는 것.

How to maintaining of the Thread-Safety

단, 위와 같은 방법은 Thread-Safety하지 않다. 아니, 애초에 ByteBuffer 클래스 자체가 Non Thread-Safety이다. 즉, Multi-Thread 환경에서는 다른 스레드에 의해 지정된 인덱스가 아닌 다른 곳에 값이 Write 될 수 있다는 말.

따라서 Concurrency System에서 ByteBuffer 클래스를 안전하게 사용하기 위해서는 다음과 같이 java.lang 패키지 내 ThreadLocal 클래스를 이용해 구현하자.


public class SafetyByteBuffer extends ThreadLocal<ByteBuffer> {
	private ByteBuffer _src;
	private int curPosition = 0;
	private int LIMIT = 10; // byte buffer limit

	public SafetyByteBuffer() {
		String basePath = System.getProperty("java.io.tmpdir");
		String filename = "data";
		String path = basePath + filename;

		try {
				_src = new RandomAccessFile(path, "rw")
									.getChannel()
									.map(FileChannel.MapMode.READ_WRITE, 0, 10000L);
		} catch (Exception ex) {
			// nothing
		}
	}

	@Override
	protected synchronized ByteBuffer initialValue() {
		ByteBuffer buf = _src.duplicate();

		buf.limit(curPosition + LIMIT);
		buf.position(curPosition);
		buf.mark();

		curPosition += LIMIT;

		return buf;
	}
}

재정의된 initialValue 메서드는 내부에서 각각의 버퍼를 LIMIT 크기만큼 자른 뒤 이를 반환하는 작업을 진행한다.

이 때 synchronized 키워드와 함께 재정의가 되었기에 각 스레드에 대해 동기적으로 버퍼를 복사할 수 있게 되며, 이 결과 중복 없이 스레드마다 사용할 수 있는 제한된 크기(LIMIT)의 버퍼를 제공할 수 있게 되는 것.

위 SafeByteBuffer 클래스를 사용한 예제는 다음과 같다.


public class UseSafetyByteBuffer extends Thread {
	private static SafetyByteBuffer bufPool = new SafetyByteBuffer();
	
	public static void main(String[] args) {
		for (int i = 0; i < 1000; i++) {
			new UseSafetyByteBuffer().start();
		}
	}
	
	@Override
	public void run() {
		ByteBuffer buf = bufPool.initialValue();
		
		buf.putInt((byte)10);
		int value = buf.getInt();
	}
}

각 스레드마다 동기적으로 버퍼를 할당받기에 Thread-Safety하게 파일에 대한 I/O 작업을 진행할 수 있게 된다.

Pitfalls

다만 이와 같이 구현하여 I/O를 진행하는 data 파일 내 데이터는 Read, Write, Update 메커니즘 뿐만 아니라, 필요하다면 Garbage Collecting Lifecycle 또한 프로그래머가 직접 구현해야 한다는 것을 유의.

실제로 해당 데이터를 Binary로 직접 다룬다는 것을 잘 생각하자.

Concurrent System에서의 Non-Heap Memory를 이용한 IPC 성능 향상

Introduce

ByteBuffer IPC Implementation

How to maintaining of the Thread-Safety

Pitfalls

`ByteBuffer` IPC Implementation