<commit_before>/*
 * Copyright 2022 The Android Open Source Project
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package androidx.datastore.multiprocess

import android.os.FileObserver
import androidx.annotation.GuardedBy
import androidx.datastore.core.CorruptionException
import androidx.datastore.core.DataStore
import androidx.datastore.core.Serializer
import androidx.datastore.multiprocess.handlers.NoOpCorruptionHandler
import java.io.File
import java.io.FileInputStream
import java.io.FileNotFoundException
import java.io.FileOutputStream
import java.io.IOException
import java.util.concurrent.Semaphore
import kotlin.coroutines.CoroutineContext
import kotlin.coroutines.coroutineContext
import kotlinx.coroutines.completeWith
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.dropWhile
import kotlinx.coroutines.flow.emitAll
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import kotlinx.coroutines.withContext

/**
 * Multi process implementation of DataStore. It is multi-process safe.
 */
internal class MultiProcessDataStore<T>(
    private val produceFile: () -> File,
    private val serializer: Serializer<T>,
    /**
     * The list of initialization tasks to perform. These tasks will be completed before any data
     * is published to the data and before any read-modify-writes execute in updateData.  If
     * any of the tasks fail, the tasks will be run again the next time data is collected or
     * updateData is called. Init tasks should not wait on results from data - this will
     * result in deadlock.
     */
    initTasksList: List<suspend (api: InitializerApi<T>) -> Unit> = emptyList(),
    private val corruptionHandler: CorruptionHandler<T> = NoOpCorruptionHandler<T>(),
    private val scope: CoroutineScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
) : DataStore<T> {
    override val data: Flow<T> = flow {
        /**
         * If downstream flow is UnInitialized, no data has been read yet, we need to trigger a new
         * read then start emitting values once we have seen a new value (or exception).
         *
         * If downstream flow has a ReadException, there was an exception last time we tried to read
         * data. We need to trigger a new read then start emitting values once we have seen a new
         * value (or exception).
         *
         * If downstream flow has Data, we should start emitting from downstream flow as long as its
         * version is not stale compared to the version read from the shared counter when we enter
         * the flow.
         *
         * If Downstream flow is Final, the scope has been cancelled so the data store is no
         * longer usable. We should just propagate this exception.
         *
         * State always starts at null. null can transition to ReadException, Data or
         * Final. ReadException can transition to another ReadException, Data or Final.
         * Data can transition to another Data or Final. Final will not change.
         */
        // TODO(b/241290444): avoid coroutine switching by loading native lib during initialization
        val latestVersionAtRead = withContext(scope.coroutineContext) { sharedCounter.getValue() }
        val currentDownStreamFlowState = downstreamFlow.value

        if ((currentDownStreamFlowState !is Data) ||
            (currentDownStreamFlowState.version < latestVersionAtRead)
        ) {
            // We need to send a read request because we don't have data yet / cached data is stale.
            readActor.offer(Message.Read(currentDownStreamFlowState))
        }

        emitAll(
            downstreamFlow.dropWhile {
                if (currentDownStreamFlowState is Data<T>) {
                    // we need to drop until initTasks are completed and set to null, and data
                    // version >= the current version when entering flow
                    (it !is Data) || (it.version < latestVersionAtRead)
                } else if (currentDownStreamFlowState is Final<T>) {
                    // We don't need to drop Final values.
                    false
                } else {
                    // we need to drop the last seen state since it was either an exception or
                    // wasn't yet initialized. Since we sent a message to actor, we *will* see a
                    // new value.
                    it === currentDownStreamFlowState
                }
            }.map {
                when (it) {
                    is ReadException<T> -> throw it.readException
                    is Final<T> -> throw it.finalException
                    is Data<T> -> it.value
                    is UnInitialized -> error(
                        BUG_MESSAGE
                    )
                }
            }
        )
    }

    override suspend fun updateData(transform: suspend (t: T) -> T): T {
        val ack = CompletableDeferred<T>()
        val currentDownStreamFlowState = downstreamFlow.value

        val updateMsg =
            Message.Update(transform, ack, currentDownStreamFlowState, coroutineContext)

        writeActor.offer(updateMsg)

        return ack.await()
    }

    private val SCRATCH_SUFFIX = ".tmp"
    private val LOCK_SUFFIX = ".lock"
    private val VERSION_SUFFIX = ".version"
    private val BUG_MESSAGE = "This is a bug in DataStore. Please file a bug at: " +
        "https://issuetracker.google.com/issues/new?component=907884&template=1466542"
    private val INVALID_VERSION = -1
    private var initTasks: List<suspend (api: InitializerApi<T>) -> Unit>? =
        initTasksList.toList()
    private val sharedCounter: SharedCounter by lazy {
        SharedCounter.loadLib()
        // TODO(b/241471375): remove the enableMlock option here
        SharedCounter.create(/* enableMlock = */ false) {
            val versionFile = fileWithSuffix(VERSION_SUFFIX)
            versionFile.createIfNotExists()
            versionFile
        }
    }
    private val threadLockSemaphore = Semaphore(1)

    // file is protected rather than private to avoid requiring synthetic accessor for its usage in
    // the definition of FileObserver
    protected val file: File by lazy {
        val file = produceFile()

        file.absolutePath.let {
            synchronized(activeFilesLock) {
                check(!activeFiles.contains(it)) {
                    "There are multiple DataStores in the same process active for the same file: " +
                        "$file. You should either maintain your DataStore as a singleton or " +
                        "confirm that there is no two DataStore's active on the same file in the " +
                        "same process(by confirming that the scope is cancelled)."
                }
                activeFiles.add(it)
            }
        }

        file
    }

    private val lockFile: File by lazy {
        val lockFile = fileWithSuffix(LOCK_SUFFIX)
        lockFile.createIfNotExists()
        lockFile
    }

    private val fileObserver: FileObserver by lazy {
        @Suppress("DEPRECATION")
        object : FileObserver(file.canonicalFile.parent!!, FileObserver.MOVED_TO) {
            // It will be triggered by same-process-write as well. Shared memory version check will
            // prevent it from reading again. parameter `path` is relative to the observed directory
            override fun onEvent(event: Int, path: String?) {
                if ((downstreamFlow.value !is Final) && (path!! == file.name)) {
                    readActor.offer(Message.Read(downstreamFlow.value))
                }<commit_msg>Fix unhandled exceptions in FileObserver for MultiProcessDataStore Bug: b/253691585 Test: ./gradlew :datastore:datastore-multiprocess:connectedAndroidTest<commit_after>/*
 * Copyright 2022 The Android Open Source Project
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package androidx.datastore.multiprocess

import android.os.FileObserver
import androidx.annotation.GuardedBy
import androidx.datastore.core.CorruptionException
import androidx.datastore.core.DataStore
import androidx.datastore.core.Serializer
import androidx.datastore.multiprocess.handlers.NoOpCorruptionHandler
import java.io.File
import java.io.FileInputStream
import java.io.FileNotFoundException
import java.io.FileOutputStream
import java.io.IOException
import java.util.concurrent.Semaphore
import kotlin.coroutines.CoroutineContext
import kotlin.coroutines.coroutineContext
import kotlinx.coroutines.completeWith
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.dropWhile
import kotlinx.coroutines.flow.emitAll
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import kotlinx.coroutines.withContext

/**
 * Multi process implementation of DataStore. It is multi-process safe.
 */
internal class MultiProcessDataStore<T>(
    private val produceFile: () -> File,
    private val serializer: Serializer<T>,
    /**
     * The list of initialization tasks to perform. These tasks will be completed before any data
     * is published to the data and before any read-modify-writes execute in updateData.  If
     * any of the tasks fail, the tasks will be run again the next time data is collected or
     * updateData is called. Init tasks should not wait on results from data - this will
     * result in deadlock.
     */
    initTasksList: List<suspend (api: InitializerApi<T>) -> Unit> = emptyList(),
    private val corruptionHandler: CorruptionHandler<T> = NoOpCorruptionHandler<T>(),
    private val scope: CoroutineScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
) : DataStore<T> {
    override val data: Flow<T> = flow {
        /**
         * If downstream flow is UnInitialized, no data has been read yet, we need to trigger a new
         * read then start emitting values once we have seen a new value (or exception).
         *
         * If downstream flow has a ReadException, there was an exception last time we tried to read
         * data. We need to trigger a new read then start emitting values once we have seen a new
         * value (or exception).
         *
         * If downstream flow has Data, we should start emitting from downstream flow as long as its
         * version is not stale compared to the version read from the shared counter when we enter
         * the flow.
         *
         * If Downstream flow is Final, the scope has been cancelled so the data store is no
         * longer usable. We should just propagate this exception.
         *
         * State always starts at null. null can transition to ReadException, Data or
         * Final. ReadException can transition to another ReadException, Data or Final.
         * Data can transition to another Data or Final. Final will not change.
         */
        // TODO(b/241290444): avoid coroutine switching by loading native lib during initialization
        val latestVersionAtRead = withContext(scope.coroutineContext) { sharedCounter.getValue() }
        val currentDownStreamFlowState = downstreamFlow.value

        if ((currentDownStreamFlowState !is Data) ||
            (currentDownStreamFlowState.version < latestVersionAtRead)
        ) {
            // We need to send a read request because we don't have data yet / cached data is stale.
            readActor.offer(Message.Read(currentDownStreamFlowState))
        }

        emitAll(
            downstreamFlow.dropWhile {
                if (currentDownStreamFlowState is Data<T>) {
                    // we need to drop until initTasks are completed and set to null, and data
                    // version >= the current version when entering flow
                    (it !is Data) || (it.version < latestVersionAtRead)
                } else if (currentDownStreamFlowState is Final<T>) {
                    // We don't need to drop Final values.
                    false
                } else {
                    // we need to drop the last seen state since it was either an exception or
                    // wasn't yet initialized. Since we sent a message to actor, we *will* see a
                    // new value.
                    it === currentDownStreamFlowState
                }
            }.map {
                when (it) {
                    is ReadException<T> -> throw it.readException
                    is Final<T> -> throw it.finalException
                    is Data<T> -> it.value
                    is UnInitialized -> error(
                        BUG_MESSAGE
                    )
                }
            }
        )
    }

    override suspend fun updateData(transform: suspend (t: T) -> T): T {
        val ack = CompletableDeferred<T>()
        val currentDownStreamFlowState = downstreamFlow.value

        val updateMsg =
            Message.Update(transform, ack, currentDownStreamFlowState, coroutineContext)

        writeActor.offer(updateMsg)

        return ack.await()
    }

    private val SCRATCH_SUFFIX = ".tmp"
    private val LOCK_SUFFIX = ".lock"
    private val VERSION_SUFFIX = ".version"
    private val BUG_MESSAGE = "This is a bug in DataStore. Please file a bug at: " +
        "https://issuetracker.google.com/issues/new?component=907884&template=1466542"
    private val INVALID_VERSION = -1
    private var initTasks: List<suspend (api: InitializerApi<T>) -> Unit>? =
        initTasksList.toList()
    private val sharedCounter: SharedCounter by lazy {
        SharedCounter.loadLib()
        // TODO(b/241471375): remove the enableMlock option here
        SharedCounter.create(/* enableMlock = */ false) {
            val versionFile = fileWithSuffix(VERSION_SUFFIX)
            versionFile.createIfNotExists()
            versionFile
        }
    }
    private val threadLockSemaphore = Semaphore(1)

    // file is protected rather than private to avoid requiring synthetic accessor for its usage in
    // the definition of FileObserver
    protected val file: File by lazy {
        val file = produceFile()

        file.absolutePath.let {
            synchronized(activeFilesLock) {
                check(!activeFiles.contains(it)) {
                    "There are multiple DataStores in the same process active for the same file: " +
                        "$file. You should either maintain your DataStore as a singleton or " +
                        "confirm that there is no two DataStore's active on the same file in the " +
                        "same process(by confirming that the scope is cancelled)."
                }
                activeFiles.add(it)
            }
        }

        file
    }

    private val lockFile: File by lazy {
        val lockFile = fileWithSuffix(LOCK_SUFFIX)
        lockFile.createIfNotExists()
        lockFile
    }

    private val fileObserver: FileObserver by lazy {
        @Suppress("DEPRECATION")
        object : FileObserver(file.canonicalFile.parent!!, FileObserver.MOVED_TO) {
            // It will be triggered by same-process-write as well. Shared memory version check will
            // prevent it from reading again. parameter `path` is relative to the observed directory
            override fun onEvent(event: Int, path: String?) {
                if ((downstreamFlow.value !is Final) && (file.name == path)) {
                    readActor.offer(Message.Read(downstreamFlow.value))
                }
{'fraction_non_alphanumeric': 0.06948923943001231, 'fraction_numerical': 0.005336304462557908, 'mean_word_length': 2.4383064516129034, 'pattern_counts': {'":': 0, '<': 39, '<?xml version=': 0, '>': 53, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7156179', 'n_tokens_mistral': 4481, 'n_tokens_neox': 4323, 'n_words': 1565}