[20240817] XTDB Internals 2 - LocalBufferPool

• requiring-resolve : https://clojuredocs.org/clojure.core/requiring-resolve

clojure 함수 requiring-resolve 가 하는 일은 그럼 무엇인가?

1. 심볼 해결: 주어진 네임스페이스-한정 심볼(namespace-qualified symbol)을 해결하려고 시도합니다.
2. 자동 require: 만약 초기 해결 시도가 실패하면, 해당 심볼의 네임스페이스를 자동으로 require(로드)하고 다시 해결을 시도합니다.
3. 지연 로딩: 필요한 시점에 네임스페이스를 로드할 수 있게 해줍니다.
4. 에러 처리: 심볼 해결에 실패하면 예외를 발생시킵니다.

(defn requiring-resolve
  "Resolves namespace-qualified sym per 'resolve'. If initial resolve
fails, attempts to require sym's namespace and retries."
  {:added "1.10"}
  [sym]
  (if (qualified-symbol? sym)
    (or (resolve sym)
	(do (-> sym namespace symbol serialized-require)
	    (resolve sym)))
    (throw (IllegalArgumentException. (str "Not a qualified symbol: " sym)))))

사용예시:

(defn debug? [] (= (env :debug?) "true"))

;; If you need to do conditional requires such as the one below:
(if (debug?)
  (do
    (require '[nrepl.cmdline :refer [-main] :rename {-main -nrepl-main}])
    (resolve 'nrepl.cmdline))
  (declare -nrepl-main))

(defn -main [& args]
  (when (debug?)
    (future (apply -nrepl-main args))))

;; This is strictly better!
(defn -main [& args]
  (when (debug?)
    ;; Swap to require and resolve in one step!
    (future (apply (requiring-resolve 'nrepl.cmdline/-main) args))))

기본적으로 resolve 함수는 주어진 심볼에 해당하는 var 를 찾아 반환하는 것이다.

여기서 require 는 네임스페이스의 코드를 읽어 로드하는 것이다. requiring-resolve 는 코드를 로드하는 과정까지 함께 한다.

XTDB와 같은 시스템에서는 이 두 개념이 동적으로 버퍼풀을 초기화할 때 아래 두 기능이 필요한 것이다.

• require는 필요한 모듈이나 라이브러리를 로드하기
• resolve는 런타임에 필요한 특정 기능을 동적으로 찾아내기

requiring-resolve는 이 둘을 결합하여, 필요한 네임스페이스를 자동으로 로드하면서 동시에 특정 심볼을 해결할 수 있게 해준다.

(.storage config ,,,) 함수는 딱봐도 storage 를 이곳에 저장하는 기능 같지만 혹시 모르니 한번 확인해보자.

package xtdb.api

object Xtdb {

    @Serializable
    data class Config(
	var txLog: Log.Factory = inMemoryLog(),
	var storage: Storage.Factory = inMemoryStorage(),
	var metrics: Metrics.Factory? = null,
	var defaultTz: ZoneId = ZoneOffset.UTC,
	@JvmField val indexer: IndexerConfig = IndexerConfig()
    ) {
	private val modules: MutableList<XtdbModule.Factory> = mutableListOf()

	fun storage(storage: Storage.Factory) = apply { this.storage = storage }

	fun getModules(): List<XtdbModule.Factory> = modules
	fun module(module: XtdbModule.Factory) = apply { this.modules += module }
	fun modules(vararg modules: XtdbModule.Factory) = apply { this.modules += modules }
	fun modules(modules: List<XtdbModule.Factory>) = apply { this.modules += modules }

	@JvmSynthetic
	fun indexer(configure: IndexerConfig.() -> Unit) = apply { indexer.configure() }

	fun open() = requiringResolve("xtdb.node.impl/open-node").invoke(this) as IXtdb
	fun txLog(txLog: Log.Factory) = apply { this.txLog = txLog }
	fun defaultTz(defaultTz: ZoneId) = apply { this.defaultTz = defaultTz }

	@JvmSynthetic
	fun metrics(metrics: Metrics.Factory) = apply { this.metrics = metrics }
    }

    @JvmStatic
    fun configure() = Config()

    @JvmStatic
    @JvmOverloads
    fun openNode(config: Config = Config()) = config.open()

    /**
     * Opens a node using a YAML configuration file - will throw an exception if the specified path does not exist
     * or is not a valid `.yaml` extension file.
     */
    @JvmStatic
    fun openNode(path: Path): IXtdb {
	if (path.extension != "yaml") {
	    throw IllegalArgumentException("Invalid config file type - must be '.yaml'")
	} else if (!path.toFile().exists()) {
	    throw IllegalArgumentException("Provided config file does not exist")
	}

	val yamlString = Files.readString(path)
	val config = nodeConfig(yamlString)

	return config.open()
    }

    @JvmSynthetic
    fun openNode(configure: Config.() -> Unit) = openNode(Config().also(configure))
}

그리고 아래 localStorage 함수를 이용하여 LocalStorageFactory를 만드는 함수가 있긴하지만 실제로 사용하는 코드인지는 모르겠다. 테스트용으로 만든 코드 같기도 하다.

(defn dir->buffer-pool
  "Creates a local storage buffer pool from the given directory."
  ^xtdb.IBufferPool [^BufferAllocator allocator, ^Path dir]
  (let [bp-path (util/tmp-dir "tmp-buffer-pool")
	storage-root (.resolve bp-path storage-root)]
    (util/copy-dir dir storage-root)
    (.openStorage (Storage/localStorage bp-path) allocator)))

->mmap-path 함수는 Java NIO(New I/O)의 메모리 매핑 기능을 사용하여 파일을 메모리에 매핑하는 유틸리티 함수.

(defn ->mmap-path
  (^java.nio.MappedByteBuffer [^Path path]
   (->mmap-path path FileChannel$MapMode/READ_ONLY))

  (^java.nio.MappedByteBuffer [^Path path ^FileChannel$MapMode map-mode]
   (with-open [in (->file-channel path (if (= FileChannel$MapMode/READ_ONLY map-mode)
					 #{:read}
					 #{:read :write}))]
     (.map in map-mode 0 (.size in))))

  (^java.nio.MappedByteBuffer [^Path path ^FileChannel$MapMode map-mode ^long start ^long len]
   (with-open [in (->file-channel path (if (= FileChannel$MapMode/READ_ONLY map-mode)
					 #{:read}
					 #{:read :write}))]
     (.map in map-mode start len))))

Direct Memory와 Memory-Mapped 파일의 차이점 정리함.

Direct Memory:

• JVM 힙 외부 메모리임
• 네이티브 I/O 작업용
• ByteBuffer.allocateDirect()로 할당
• GC 영향 없음, 네이티브 I/O에 효율적
• 명시적 메모리 관리 필요

Memory-Mapped 파일 (->mmap-path 생성):

• 파일을 가상 메모리에 매핑
• 대용량 파일 고속 읽기/쓰기용
• FileChannel.map()으로 생성
• 대용량 파일 처리 효율적
• 가상 메모리 사용량 주의 필요

주요 차이:

1. 용도: Direct Memory는 네트워크 I/O, Memory-Mapped는 파일 처리
2. 데이터 소스: Direct Memory는 순수 메모리, Memory-Mapped는 파일 기반
3. 지속성: Direct Memory는 휘발성, Memory-Mapped는 지속적
4. 크기 제한: Direct Memory는 시스템 메모리, Memory-Mapped는 파일 시스템 용량 기준

XTDB 등 데이터베이스 시스템에서 활용:

• Direct Memory: 캐시, 임시 데이터 구조에 적합
• Memory-Mapped 파일: 데이터 파일, 로그 파일 등 영구 저장소에 적합

결론: 둘 다 효율적 I/O 처리 메커니즘 제공. 사용 사례와 요구사항에 따라 선택 필요.

테스트 코드를 구경해보자.

이렇게 getBuffer 는 가져올 때, 문자열 값을 Path 로 변환하는 로직이 있다. 아마도 키 값을 우리는 keyword로 요청하면 그것을 문자열로 바꾸고 또 Path 로 변경해서 사용하는 것인가? 싶다.

(util/with-open [buf (.getBuffer bp (util/->path "aw"))]
  (let [{:keys [root]} (util/read-arrow-buf buf)]
    (util/close root)))

실제로 사용하는 코드는 어디있을까?

목적: 파일의 원자적 이동

인자:

• from-path: 원본 경로 (Path)
• to-path: 목적지 경로 (Path)

동작:

• Java NIO의 Files.move 메소드 사용
• ATOMIC_MOVE 옵션 적용

특징:

• 원자성 보장: 이동 작업이 완전히 성공하거나 실패
• 중간 상태 없음: 부분적 이동 방지
• 파일 시스템 레벨에서 지원 (가능한 경우)

주의: 파일 시스템이 원자적 이동을 지원하지 않으면 예외 발생 가능

(defn atomic-move [^Path from-path ^Path to-path]
  (Files/move from-path to-path (into-array CopyOption [StandardCopyOption/ATOMIC_MOVE]))
  to-path)

(make-array FileVisitOption 0) 의미가 궁금해서 공식문서를 찾아봄

varargs 로 넘기기 위해 FileVisitOption 타입의 빈배열(길이 0) 으로 넣어버림.

public static Stream<Path> walk(Path start,
				FileVisitOption... options)
			 throws IOException
Return a Stream that is lazily populated with Path by walking the file tree rooted at a given starting file.
The file tree is traversed depth-first, the elements in the stream are Path objects that are obtained as if by resolving the relative path against start.
This method works as if invoking it were equivalent to evaluating the expression:

 walk(start, Integer.MAX_VALUE, options)

In other words, it visits all levels of the file tree.
The returned stream encapsulates one or more DirectoryStreams. If timely disposal of file system resources is required, the try-with-resources construct should be used to ensure that the stream's close method is invoked after the stream operations are completed. Operating on a closed stream will result in an IllegalStateException.

Parameters:
start - the starting file
options - options to configure the traversal

Returns:
the Stream of Path

Throws:
SecurityException - If the security manager denies access to the starting file. In the case of the default provider, the checkRead method is invoked to check read access to the directory.
IOException - if an I/O error is thrown when accessing the starting file.

Since:
1.8
See Also:
walk(Path, int, FileVisitOption...)

[java nio File Visitor]​

이는 읽는 파일의 path를 루트 패스(disk-store) 기준으로 상대경로로 만드는 것이다. 아마 이것을 키 값으로 사용하는 듯?

XTDB의 listAllObjects에서 relativize 사용 예시:

가정:

• disk-store 경로: "/data/xtdb"
• 실제 파일 경로: "/data/xtdb/chunks/001.data"

코드:

(let [path (Paths/get "/data/xtdb/chunks/001.data")
      relativized (.relativize disk-store path)]
  relativized)

리턴 : relativized 값: "chunks/001.data"

설명:

1. disk-store ("/data/xtdb")를 기준으로
2. 전체 경로 ("/data/xtdb/chunks/001.data")의 상대 경로 계산
3. 공통 부분 제거 후 나머지 경로 반환

목적:

• 저장소 루트 기준 파일 위치 표현
• 이동 가능한 상대 경로 유지
• 저장소 구조 독립적인 파일 참조

설명으로는 불투명한 파일이 있는 일반파일인지 확인한다?

아마도 불투하다는 것은 데이터가 저장되어 있는 데이터만 읽겠다는 것으로 보임.

public static boolean isRegularFile(Path path,
				    LinkOption... options)
Tests whether a file is a regular file with opaque content.

The options array may be used to indicate how symbolic links are handled for the case that the file is a symbolic link.
By default, symbolic links are followed and the file attribute of the final target of the link is read.
If the option NOFOLLOW_LINKS is present then symbolic links are not followed.

Where it is required to distinguish an I/O exception from the case that the file is not a regular file
then the file attributes can be read with the readAttributes method and the file type tested with the BasicFileAttributes.isRegularFile() method.

Parameters:
path - the path to the file
options - options indicating how symbolic links are handled

Returns:
true if the file is a regular file; false if the file does not exist, is not a regular file, or it cannot be determined if the file is a regular file or not.

Throws:
SecurityException - In the case of the default provider, and a security manager is installed, its checkRead method denies read access to the file.

[isRegularFile DOC]​

일단 느낌상 그렇다지만 스택오버플로우에 좀 더 정리된 내용이 있어서 기록함. (맞는 것인지는 모르겠음)

1. 응답 1

   For example in UNIX, a regular file is one that is not special in some way.
   Special files include symbolic links and directories.
   A regular file is a sequence of bytes stored permanently in a file system.
   
   ---
   
   I figure rm -i is an alias, possibly rm -i. The "regular" part doesn't mean anything in particular,
   it only means that it's not a pipe, device, socket or anything other "special".
   
   it means the file is not a symlink, pipe, rand, null, cpu, etc. Perhaps you have heard the linux philosophy everything is a text.
   This isn't literally true, but it suggests a dominant operational context where string processing tools can be applied to filesystem elements directly.
   In this case, it means that in a more literal fashion. To see the detection step in isolation, try the command file, as in file /etc/passwd or file /dev/null.
   
   

2. 응답 2

   From Files Reference - AIX IBM
   
   A file is a collection of data that can be read from or written to. A file can be a program you create, text you write, data you acquire, or a device you use. Commands, printers, terminals, and application programs are all stored in files. This allows users to access diverse elements of the system in a uniform way and gives the operating system great flexibility. No format is implied when a file is created.
   
   There are three types of files
   
   Regular - Stores data (text, binary, and executable).
   Directory - Contains information used to access other files.
   Special - Defines a FIFO (first-in, first-out) file or a physical device.
   Regular files are the most common. When a word processing program is used to create a document, both the program and the document are contained in regular files.
   
   Regular files contain either text or binary information. Text files are readable by the user.
   Binary files are readable by the computer. Binary files can be executable files that instruct the system to accomplish a job.
   Commands, shell scripts, and other programs are stored in executable files.
   
   Directories contain information the system needs to access all types of files, but they do not contain the actual file data.
   As a result, directories occupy less space than a regular file and give the file-system structure flexibility and depth.
   Each directory entry represents either a file or subdirectory and contains the name of a file and the file's i-node (index node reference) number.
   The i-node number represents the unique i-node that describes the location of the data associated with the file. Directories are created and controlled by a separate set of commands. See "Directories" in Operating system and device management for more information.
   
   Special files define devices for the system or temporary files created by processes.
   There are three basic types of special files: FIFO (first-in, first-out), block, and character.
   FIFO files are also called pipes. Pipes are created by one process to temporarily allow communication with another process.
   These files cease to exist when the first process finishes. Block and character files define devices.
   
   All this above is from the first link. I've checked in many other sources regarding Operational Systems differences and it seems this one is the most common definition on all sources i've found.
   

뭐 java nio 와 완벽하게 맞춰진 데이터는 아니지만 regular 파일이란, 일반적으로 데이터를 저장하는 용도의 파일을 말하는 듯?

[stack over flow link]​

writeBatch 는 이전 xtdb-internals-1 에 이미 설명이 있다. 코드만 옮겨놓자. 채널을 받아서 그곳에 vectors 를 저장하는 듯.

이렇게 구성된 데이터는 Arrow 포맷으로 쉽게 직렬화되어 파일에 저장되거나 네트워크로 전송될 수 있습니다.

// Apache Arrow 파일 포맷에서 사용하는 매직 바이트 인듯?
private val MAGIC = "ARROW1".toByteArray()

inner class RelationUnloader(private val ch: WriteChannel) : AutoCloseable {

  private val vectors = this@Relation.vectors.values
  private val schema = Schema(vectors.map { it.field })
  private val recordBlocks = mutableListOf<ArrowBlock>()

  init {
    // 파일 시작에 매직 바이트 쓰기.
    ch.write(MAGIC)
    ch.align()
    MessageSerializer.serialize(ch, schema)
  }

  fun writeBatch() {
    // 각 필드(컬럼)의 메타데이터를 저장
    // 주로 각 필드의 유효한 데이터 개수(null이 아닌 값)를 포함합니다.
    // 데이터의 논리적 구조
    val nodes = mutableListOf<ArrowFieldNode>()
    // 실제 데이터 값을 저장합니다.
    // Arrow의 컬럼나 포맷에 따라 여러 버퍼가 사용될 수 있습니다.
    // 예를 들어, 가변 길이 데이터는 오프셋과 값 버퍼가 필요합니다.
    val buffers = mutableListOf<ArrowBuf>()

    vectors.forEach { it.unloadBatch(nodes, buffers) }

    ArrowRecordBatch(rowCount, nodes, buffers).use { recordBatch ->
	MessageSerializer.serialize(ch, recordBatch)  // 직렬화
	    .also { recordBlocks.add(it) }
    }
  }

  fun endStream() {
    ch.writeIntLittleEndian(MessageSerializer.IPC_CONTINUATION_TOKEN)
    ch.writeIntLittleEndian(0)
  }

  fun endFile() {
    endStream()

    val footerStart = ch.currentPosition
    ch.write(ArrowFooter(schema, emptyList(), recordBlocks), false)

    val footerLength = ch.currentPosition - footerStart
    check(footerLength > 0) { "Footer length must be positive" }
    ch.writeIntLittleEndian(footerLength.toInt())
    // 파일 마지막에 다시 매직 바이트 쓰기 
    ch.write(MAGIC)
  }

  override fun close() {
      ch.close()
  }
}