Kotlin: High-Performance concurrent file I/O best-practices?

Viewed 35

What's the most performant way in Kotlin to allow concurrent file I/O in multi-reader, single-writer fashion?

I have the below, but I'm unsure how much overhead is being created by the coroutine facilities:

Searching for examples of this doesn't turn up much (try searching Github for ReadWriteMutex, there are a tiny handful of Kotlin repos implementing this).

class DiskManagerImpl(file: File) : DiskManager {
    private val mutex = ReadWriteMutexImpl()
    private val channel = AsynchronousFileChannel.open(file.toPath(),
        StandardOpenOption.READ, StandardOpenOption.WRITE)

    override suspend fun readPage(pageId: PageId, buffer: MemorySegment) = withContext(Dispatchers.IO) {
        mutex.withReadLock {
            val offset = pageId * PAGE_SIZE
            val bytesRead = channel.readAsync(buffer.asByteBuffer(), offset.toLong())
            require(bytesRead == PAGE_SIZE) { "Failed to read page $pageId" }
        }
    }

    override suspend fun writePage(pageId: PageId, buffer: MemorySegment) = withContext(Dispatchers.IO) {
        mutex.withWriteLock {
            val offset = pageId * PAGE_SIZE
            val bytesWritten = channel.writeAsync(buffer.asByteBuffer(), offset.toLong())
            require(bytesWritten == PAGE_SIZE) { "Failed to write page $pageId" }
        }
    }
}
class ReadWriteMutexImpl : ReadWriteMutex {
    private val read = Mutex()
    private val write = Mutex()
    private val readers = atomic(0)

    override suspend fun lockRead() {
        if (readers.getAndIncrement() == 0) {
            read.lock()
        }
    }

    override fun unlockRead() {
        if (readers.decrementAndGet() == 0) {
            read.unlock()
        }
    }

    override suspend fun lockWrite() {
        read.lock()
        write.lock()
    }

    override fun unlockWrite() {
        write.unlock()
        read.unlock()
    }
}

suspend inline fun <T> ReadWriteMutex.withReadLock(block: () -> T): T {
    lockRead()
    return try {
        block()
    } finally {
        unlockRead()
    }
}

suspend inline fun <T> ReadWriteMutex.withWriteLock(block: () -> T): T {
    lockWrite()
    return try {
        block()
    } finally {
        unlockWrite()
    }
}
0 Answers
Related