multithreading - 在 Kotlin Native 中，如何将对象保存在单独的线程中，并在不使用 C 指针的情况下从任何其他线程中改变其状态？

Question

我正在探索 Kotlin Native 并且有一个包含一堆 Workers 的程序做并发的事情
(在 Windows 上运行，但这是一个普遍问题)。

现在，我想添加简单的日志记录。一个组件，它通过将字符串作为新行附加到文件 来简单地记录字符串。保持开放 在“附加”模式下。

(理想情况下，我只有一个“全局”功能......

fun log(text:String) {...} ] 

...我可以从任何地方打电话，包括从“内部”其他 worker 那里打电话，这会起作用。这里的含义是，由于 Kotlin Native 关于在线程之间传递对象的规则(TLDR:你不应该传递可变对象。参见: https://github.com/JetBrains/kotlin-native/blob/master/CONCURRENCY.md#object-transfer-and-freezing)，这样做并非易事。
此外，我的日志函数理想情况下会接受任何 frozen目的。 )

我想出的是使用 DetachedObjectGraph 的解决方案:

首先，我创建一个分离的记录器对象

val loggerGraph = DetachedObjectGraph { FileLogger("/foo/mylogfile.txt")}

然后使用 loggerGraph.asCPointer() ( asCPointer() ) 获得 COpaquePointer到分离图:

val myPointer = loggerGraph.asCPointer()

现在我可以将此指针传递给工作人员(通过工作人员的 producer lambda 的 execute function )，并在那里使用它。或者我可以将指针存储在 @ThreadLocal 中全局变量

对于写入文件的代码，每当我想记录一行时，我必须创建一个 DetachedObjectGraph对象再次从指针，
和 attach()它是为了获得对我的 fileLogger 对象的引用:

val fileLogger = DetachedObjectGraph(myPointer).attach()

现在我可以在记录器上调用日志函数:

fileLogger.log("My log message")

这就是我在查看 Kotlin Native 中可用(从 Kotlin 1.3.61 开始)并发的 API 时想到的，
但是 我想知道更好的方法是什么(使用 Kotlin，而不是诉诸 C)。显然，为写入的每一行都创建一个 DetachedObjectGraph 对象是不好的。

可以用更一般的方式提出这个问题:如何在单独的线程(或 worker )中保持可变资源打开，并向其发送消息。

旁注:有 Coroutines真正使用线程会解决这个问题，但问题是如何使用当前可用的 API(Kotlin 1.3.61)解决这个任务。

Answer 1

你绝对不应该使用 DetachedObjectGraph以问题中提出的方式。没有什么可以阻止您尝试在多个线程上附加，或者如果您传递相同的指针，则尝试附加到一个无效的一个又一个附加到它的线程上。

正如多米尼克所说，您可以保留 DetachedObjectGraph在 AtomicReference .但是，如果您要保留 DetachedObjectGraph在 AtomicReference , 确保类型为 AtomicRef<DetachedObjectGraph?>和忙循环而 DetachedObjectGraph一片空白。这将防止相同的 DetachedObjectGraph被多个线程使用。确保将其设置为 null，并以原子方式重新填充它。

但是，是否FileLogger需要是可变的吗？如果您正在写入文件，则似乎并非如此。即使是这样，我也会将可变对象隔离到一个单独的工作人员并向其发送日志消息，而不是执行 DetachedObjectGraph在 AtomicRef 中。

根据我的经验，DetachedObjectGraph在生产代码中非常罕见。我们目前不在任何地方使用它。

将可变状态隔离到 Worker ，像这样:

class MutableThing<T:Any>(private val worker:Worker = Worker.start(), producer:()->T){
    private val arStable = AtomicReference<StableRef<T>?>(null)
    init {
        worker.execute(TransferMode.SAFE, {Pair(arStable, producer).freeze()}){
            it.first.value = StableRef.create(it.second()).freeze()
        }
    }
    fun <R> access(block:(T)->R):R{
        return worker.execute(TransferMode.SAFE, {Pair(arStable, block).freeze()}){
            it.second(it.first.value!!.get())
        }.result
    }
}

object Log{
    private val fileLogger = MutableThing { FileLogger() }

    fun log(s:String){
        fileLogger.access { fl -> fl.log(s) }
    }
}

class FileLogger{
    fun log(s:String){}
}

MutableThing用途 StableRef内部。 producer使您想要隔离的可变状态。要记录某些内容，请调用 Log.log ，这将最终调用可变 FileLogger .

查看 MutableThing 的基本示例，运行以下测试:

@Test
fun goIso(){
    val mt = MutableThing { mutableListOf("a", "b")}
    val workers = Array(4){Worker.start()}
    val futures = mutableListOf<Future<*>>()
    repeat(1000) { rcount ->
        val future = workers[rcount % workers.size].execute(
            TransferMode.SAFE,
            { Pair(mt, rcount).freeze() }
        ) { pair ->
            pair.first.access {
                val element = "ttt ${pair.second}"
                println(element)
                it.add(element)
            }
        }
        futures.add(future)
    }

    futures.forEach { it.result }

    workers.forEach { it.requestTermination() }

    mt.access {
        println("size: ${it.size}")
    }
}