同期フラッシュモードの起動判定

メッセージキューサーバーにおいて、永続化信頼性とスループットのバランスを取る手法として同期フラッシュ（SYNC_FLUSH）が提供されています。このモードが有効な場合、Producerからの応答待ち状態が維持され、オペレーティングシステムレベルでの物理メモリ→ストレージの転送完了まで処理がブロックされます。

エントリポイントとなるメソッドでは、構成設定に基づきフラッシュ方針を判定し、要求をバックグラウンドのスレッドプールへ委譲します。

public void dispatchFlushAction(AppendMessageResult commitMeta, OperationResult ctx, MessageExt messageEnvelope) {
    final boolean isSyncEnabled = configuration.getFlushPolicy() == FlushMode.SYNC;
    if (!isSyncEnabled) {
        handleAsyncPath(ctx);
        return;
    }

    BatchFlushCoordinator service = (BatchFlushCoordinator) this.diskFlushEngine;
    if (messageEnvelope.isAckRequired()) {
        long targetOffset = commitMeta.getWrittenEnd();
        FlushToken task = new FlushToken(targetOffset);
        service.register(task);
        
        boolean completionFlag = task.awaitResult(configuration.getSyncTimeoutMs());
        if (!completionFlag) {
            log.error("Sync flush timed out | topic={} | addr={}", messageEnvelope.getTopic(), messageEnvelope.getClientAddress());
            ctx.setStatus(OperationStatus.FLUSH_TIMEOUT);
        }
    } else {
        service.triggerExecution();
    }
}

タスクキューリングと通知機構

複数のメッセージが短時間内に到着した場合、バラバラにディスクI/Oを発行するのではなく、リクエストをバッチ化して処理します。ここで使用されるのはCAS（Compare-And-Swap）による通知フラグと、セマフォを活用した効率的な待機パターンです。

public synchronized void register(FlushToken workItem) {
    synchronized (pendingQueue) {
        pendingQueue.offer(workItem);
    }
    if (hasWorkNotification.compareAndSet(false, true)) {
        executionLatch.countDown();
    }
}

protected void awaitNextRound(long timeoutMs) {
    if (hasWorkNotification.compareAndSet(true, false)) {
        return;
    }
    executionLatch.reset();
    try {
        executionLatch.await(timeoutMs, TimeUnit.MILLISECONDS);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    } finally {
        hasWorkNotification.set(false);
    }
}

バッチコミット実行フロー

待機から解除されたワーカースレッドは、キューに蓄積されたすべての要求を対象にフラッシュ処理を走査します。旧版の設計思想では、ファイル境界を跨ぐケースに対応するため最大2回の呼び出しを試行する実装が見られますが、現代のデフォルト構成では単一ファイル内での完結が前提となっています。

private void executeBatchCommit() {
    List<FlushToken> activeTasks = swapWorkingBuffers();
    
    for (FlushToken req : activeTasks) {
        boolean synced = false;
        for (int attempt = 0; attempt < 2 && !synced; attempt++) {
            long currentSyncedPointer = queueManager.getLatestSyncedOffset();
            if (currentSyncedPointer >= req.getTargetLocation()) {
                synced = true;
                break;
            }
            queueManager.forcePersistentWrite();
        }
        req.notifyClient(synced);
    }
    
    recordPersistTimestamp(queueManager.getStoreChrono());
    activeTasks.clear();
}

private List<FlushToken> swapWorkingBuffers() {
    List<FlushToken> temp;
    synchronized (this) {
        temp = writerQueue;
        writerQueue = readerQueue;
        readerQueue = temp;
    }
    return readerQueue;
}

低レベルI/OとOS同期呼び出し

最終的な永続化はJava NIOの `MappedByteBuffer.force()` を経由して実施されます。この処理はJVM内部でC言語ベースのネイティブ呼び出しに変換され、Linux環境では `msync` システムコールが実行されます。

public int forcePages(int minimumThreshold) {
    if (isEligibleForFlush(minimumThreshold)) {
        resourceLock.lock();
        int originPosition = getReadHead();
        
        try {
            if (ioBufferPool != null || channel.position() != 0) {
                channel.force(false);
            } else {
                virtualMemoryBuffer.force();
            }
        } catch (Throwable ioException) {
            log.error("Failed to invoke OS sync layer", ioException);
        }
        
        flushedCursor.set(originPosition);
        resourceLock.unlock();
    }
    return getPositionRecord();
}

JVM側のネイティブ実装（`jdk/src/solaris/native/java/io/MappedByteBuffer.c` 等）では以下のようになります。`MS_SYNC` フラグを渡すことで、カーネルがキャッシュ内のデータを確実にディスクコントローラへライトバックするまで呼び出し元のスレッドをブロックします。

JNIEXPORT void JNICALL
Java_java_nio_MappedByteBuffer_force0(JNIEnv* env, jobject obj, jobject fdo,
                                      jlong address, jlong len) {
    void* memoryRegion = (void*)jlong_to_ptr(address);
    int sysCallResult = msync(memoryRegion, (size_t)len, MS_SYNC);
    if (sysCallResult == -1) {
        throwIOException(env, "msync failed");
    }
}

主な障害要因と特性

• マップされた仮想アドレス範囲が無効または破損している場合
• プロセスに該当ファイルパスへの書き込み権限がない場合
• OSのメモリ管理サブシステムやページアウト機構のリソース枯渇
• 同時実行される複数のプロセス間でのキャッシュ競合・デッドロック状態

例外時オフセット更新の注意点

上記の `forcePages` 実装で確認できる通り、`flushedCursor.set(originPosition)` の実行は `try-catch` ブロックの外側に配置されています。これはパフォーマンス最適化の意図であり、仮に `msync` がタイムアウトやハードウェアエラーで失敗した場合でも、アプリケーション層には「成功済み」という事実が返却される設計です。結果として、クラッシュ直前にACKが発行されたメッセージが実際のストレージに残らないという理論上のデータ消失リスクが存在しますが、分散キューアーキテクチャ全体としての整合性担保は WAL（Write-Ahead Log）やレプリケーション機能によって補完される構造となっています。