9. Memory Management Internals

// if we have a 1GB JVM, then the memory used for execution and storage
// will be (1024 - 300) * 0.6 = 434MB by default.
private val RESERVED_SYSTEM_MEMORY_BYTES = 300 * 1024 * 1024

private def getMaxMemory(conf: SparkConf): Long = {
  ...
  val minSystemMemory = (reservedMemory * 1.5).ceil.toLong   // 450MB 미만이면 시작 실패
  ...
  val usableMemory = systemMemory - reservedMemory
  (usableMemory * conf.get(config.MEMORY_FRACTION)).toLong
}

def maybeGrowExecutionPool(extraMemoryNeeded: Long): Unit = {
  if (extraMemoryNeeded > 0) {
    // Storage의 빈 공간, 또는 R을 넘어 빌려 간 부분만 회수 가능
    val memoryReclaimableFromStorage = math.max(
      storagePool.memoryFree,
      storagePool.poolSize - storageRegionSize)
    if (memoryReclaimableFromStorage > 0) {
      val spaceToReclaim = storagePool.freeSpaceToShrinkPool(
        math.min(extraMemoryNeeded, memoryReclaimableFromStorage))
      storagePool.decrementPoolSize(spaceToReclaim)
      executionPool.incrementPoolSize(spaceToReclaim)
    }
  }
}

override def acquireStorageMemory(blockId: BlockId, numBytes: Long, memoryMode: MemoryMode)
    : Boolean = synchronized {
  ...
  if (numBytes > storagePool.memoryFree) {
    // Execution의 "비어 있는" 메모리만 빌림
    val memoryBorrowedFromExecution = Math.min(executionPool.memoryFree,
      numBytes - storagePool.memoryFree)
    executionPool.decrementPoolSize(memoryBorrowedFromExecution)
    storagePool.incrementPoolSize(memoryBorrowedFromExecution)
  }
  storagePool.acquireMemory(blockId, numBytes)
}

while (true) {
  val numActiveTasks = memoryForTask.keys.size
  val curMem = memoryForTask(taskAttemptId)
  maybeGrowPool(numBytes - memoryFree)          // 먼저 Storage에서 회수 시도
  val maxPoolSize = computeMaxPoolSize()
  val maxMemoryPerTask = maxPoolSize / numActiveTasks       // 상한 1/N
  val minMemoryPerTask = poolSize / (2 * numActiveTasks)    // 하한 1/2N
  val maxToGrant = math.min(numBytes, math.max(0, maxMemoryPerTask - curMem))
  val toGrant = math.min(maxToGrant, memoryFree)
  if (toGrant < numBytes && curMem + toGrant < minMemoryPerTask) {
    lock.wait()          // 1/2N도 못 채우면 다른 Task가 반납할 때까지 대기
  } else {
    memoryForTask(taskAttemptId) += toGrant
    return toGrant
  }
}

// on-heap 모드: 상위 13비트 = 페이지 번호, 하위 51비트 = 페이지 안 offset
private static final int PAGE_NUMBER_BITS = 13;
static final int OFFSET_BITS = 64 - PAGE_NUMBER_BITS;  // 51
private final MemoryBlock[] pageTable = new MemoryBlock[PAGE_TABLE_SIZE];

public static long encodePageNumberAndOffset(int pageNumber, long offsetInPage) {
  return (((long) pageNumber) << OFFSET_BITS) | (offsetInPage & MASK_LONG_LOWER_51_BITS);
}

public long acquireExecutionMemory(long required, MemoryConsumer requestingConsumer) {
  long got = memoryManager.acquireExecutionMemory(required, taskAttemptId, mode);
  if (got < required) {
    // 필요한 양 이상을 가진 소비자 중 가장 작은 것부터 Spill
    TreeMap<Long, List<MemoryConsumer>> sortedConsumers = new TreeMap<>();
    for (MemoryConsumer c: consumers) {
      if (c.getUsed() > 0 && c.getMode() == mode) {
        long key = c == requestingConsumer ? 0 : c.getUsed();  // 요청자 자신은 최후순위
        sortedConsumers.computeIfAbsent(key, k -> new ArrayList<>(1)).add(c);
      }
    }
    while (got < required && !sortedConsumers.isEmpty()) {
      Map.Entry<Long, List<MemoryConsumer>> currentEntry =
        sortedConsumers.ceilingEntry(required - got);
      if (currentEntry == null) currentEntry = sortedConsumers.lastEntry();  // 없으면 가장 큰 것
      got += trySpillAndAcquire(...);
    }
  }
  ...
}

/** Spill some data to disk to release memory, which will be called by TaskMemoryManager
 *  when there is not enough memory for the task.
 *  Note: In order to avoid possible deadlock, should not call acquireMemory() from spill(). */
public abstract long spill(long size, MemoryConsumer trigger) throws IOException;

private def getOrElseUpdate[T](...): Either[BlockResult, Iterator[T]] = {
  if (isCacheVisible) {
    get[T](blockId)(classTag) match {          // 로컬 → 원격 순서로 조회
      case Some(block) => return Left(block)
      case _ =>
    }
  }
  doPutIterator(blockId, iterator, level, classTag, keepReadLock = true) match {
    case None =>       // 저장 성공 → 읽기 락을 쥔 채로 값 반환
      ...
    case Some(iter) => // 저장 실패 → 계산된 Iterator를 그대로 돌려줌
      Right(iter)
  }
}

private val entries = new LinkedHashMap[BlockId, MemoryEntry[_]](32, 0.75f, true)

def blockIsEvictable(blockId: BlockId, entry: MemoryEntry[_]): Boolean = {
  entry.memoryMode == memoryMode && (rddToAdd.isEmpty || rddToAdd != getRddId(blockId))
}
entries.synchronized {
  val iterator = entries.entrySet().iterator()   // LRU 순서
  while (freedMemory < space && iterator.hasNext) {
    ...
    if (blockIsEvictable(blockId, entry)) {
      // 기다리지 않는 쓰기 락 → 누가 읽고 있는 블록은 건너뜀
      if (blockInfoManager.lockForWriting(blockId, blocking = false).isDefined) {
        selectedBlocks += blockId
        freedMemory += entry.size
      }
    }
  }
}

keepUnrolling = reserveUnrollMemoryForThisTask(blockId, initialMemoryThreshold, memoryMode) // 1MB
while (values.hasNext && keepUnrolling) {
  valuesHolder.storeValue(values.next())
  if (elementsUnrolled % memoryCheckPeriod == 0) {         // 16개마다
    val currentSize = valuesHolder.estimatedSize()
    if (currentSize >= memoryThreshold) {
      val amountToRequest = (currentSize * memoryGrowthFactor - memoryThreshold).toLong  // ×1.5
      keepUnrolling = reserveUnrollMemoryForThisTask(blockId, amountToRequest, memoryMode)
    }
  }
  elementsUnrolled += 1
}

if (level.useMemory) {
  // Put it in memory first, even if it also has useDisk set to true;
  // We will drop it to disk later if the memory store can't hold it.
  memoryStore.putIteratorAsValues(...) match {
    case Right(s) => size = s
    case Left(iter) =>
      if (level.useDisk) {
        diskStore.put(blockId) { channel => ... serialize stream ... }
      } else {
        iteratorFromFailedMemoryStorePut = Some(iter)  // MEMORY_ONLY → 저장 안 함
      }
  }
}