Kotlin Native. Работаем с потоками с корутинами и без. Ч1

Всем доброго дня! С вами Анна Жаркова, ведущий мобильный разработчик компании Usetech. Совсем недавно компания JetBrains выпустила preview новой модели управления памятью. Это отличный повод сделать подробный ее разбор на практике, а также сравнить с моделью, используемой в KMM в текущих версиях. Но для начала неплохо было бы поговорить о тех возможностях работы в Kotlin Native, которые мы на практике не используем.

Если вы уже знакомы с тем, как работать с корутинами в Kotlin/Native и/или читали предыдущие статьи 1 и 2 автора, то пролистайте немного вниз. Материал может немного повторяться.

Когда мы работаем с Kotlin и Kotlin Multiplatform, то самым и простым удобным способом для настройки работы с многопоточностью в приложении являются Kotlin Coroutines. Наша задача сводится к настройке скоупов CoroutineScope для запуска корутин и suspend функций в основном потоке и фоновом. Т.к. в разных платформенных версиях языка Kotlin этот механизм реализуется по-разному, то необходимо кастомизировать получение контекста корутин с помощью expect/actual:

expect val defaultDispatcher: CoroutineContext  expect val uiDispatcher: CoroutineContext

В случае Kotlin JVM и Android у нас проблем нет (у Kotlin JVM вообще проблем нет, поэтому мы его и не рассматриваем) :

actual val uiDispatcher: CoroutineContext     get() = Dispatchers.Main  actual val defaultDispatcher: CoroutineContext     get() = Dispatchers.Default

Мы просто используем доступные диспетчеры корутин Main и Default (можно взять IO).

Но в случае iOS и Kotlin Native не все так просто. Мы, конечно, можем использовать те же самые диспетчеры Main и Default, но при переключении контекста запроса мы получим исключение:

Дело в том, что для того, чтобы мы могли передавать изменяемые блоки кода и объекты между потоками, нам нужно их замораживать перед передачей с помощью команды freeze() . Делать это, конечно, удобнее при скрытом переключении контекста. Поэтому мы создадим свои диспетчеры, которые будут обращаться к нужным DispatchQueue, и будем использовать их как контексты корутин:

actual val defaultDispatcher: CoroutineContext get() = IODispatcher  actual val uiDispatcher: CoroutineContext get() = MainDispatcher  private object MainDispatcher: CoroutineDispatcher(){     override fun dispatch(context: CoroutineContext, block: Runnable) {         dispatch_async(dispatch_get_main_queue()) {             try {                 block.run().freeze()             }catch (err: Throwable) {                 throw err             }         }     } } //Background private object IODispatcher: CoroutineDispatcher(){     override fun dispatch(context: CoroutineContext, block: Runnable) {         dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT.toLong(), 0.toULong())) {             try {                 block.run().freeze()             }catch (err: Throwable) {                 throw err             }         }     }

Но если для основного потока использование такого MainDispatcher корректно, для для фонового потока мы не сможем работать с dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT.toLong(), 0.toULong()), потому что эта очередь не привязана ни к одному потоку. Поэтому нам стоит поменять вызов на MainDispatcher для обоих контестов.

actual val defaultDispatcher: CoroutineContext get() = MainDispatcher  actual val uiDispatcher: CoroutineContext get() = MainDispatcher   @ThreadLocal private object MainDispatcher: CoroutineDispatcher(){     override fun dispatch(context: CoroutineContext, block: Runnable) {         dispatch_async(dispatch_get_main_queue()) {             try {                 block.run().freeze()             }catch (err: Throwable) {                 throw err             }         }     }

Аннотация ThreadLocal используется для корректного шаринга неизменяемого объекта (синглтона) между разными потоками.

Кого-то может смутить корректность получившегося решения. Но так советуют разработчики технологии:

В качестве примера корректности такого поведения разбирается асинхронная работа с сетевыми запросами через библиотеку Ktor. Действительно большинство случаев, когда нам в мобильном приложении нужно работать с разными потоками, сводятся к сетевым запросам. И при разработке с помощью Kotlin Multiplatform в основном все мы используем готовую библиотеку Ktor, которая уже реализует всю асинхронную работу под капотом.

А что же нам делать, если наша асинхронная работа не связана с сетевыми запросами, и/или мы не хотим использовать Ktor? Да, многопоточные корутины под Kotlin/Native не поддерживаются (вернее, поддерживаются, но не в основных ветках, а, например, реализациях «1.5.0-native-mt»), но нам же надо что-то делать.

Можем, конечно, попробовать осуществлять маршрутизацию в GlobalScope, ведь он позволяет выполнять корутины именно в фоне. Но это не рекомендованное решение, потому что невозможно отследить выполнение такой корутины и управлять ею. Также это создает потенциальные утечки памяти. Поэтому рекомендуется собирать ссылки на job, которые мы запускаем в GlobalScope, и ожидать их окончания.

//Не рекомендовано for i in 0..10 { GlobalScope.launch {    work(i) } }  //Workaround  val jobs = mutableListOf<Job>() for i in 0..10 {   jobs += GlobalScope.launch {     work(i)   } } jobs.forEach{ it.join() }

Что же нам делать в таком случае? Обратимся к опыту предшественников, т.е Ktor, а также документации по многопоточности в Kotlin Native.

Все, чего нам не хватает при работе с корутинами и скоупами корутин в Kotlin Native, делается некоторыми другими средствами, доступными в самом Kotlin Native из-под коробки. Давайте же разберем их, а заодно и посмотрим, как написать свой сетевой клиент под iOS именно средствами Kotlin Native.

В данном случае будем делать разные реализации классов сетевых клиентов под iOS и Android не с помощью expect/actual, а путем создания классов с общим интерфейсом, через который мы и будем к нашим клиентам обращаться:

interface IHttpClient {     fun request(request: Request, completion: (Response)->Unit) }  //Android with OKHttp class HttpClientAndroid: IHttpClient { /**...*/ }  //iOS with NSUrlSession class HttpClientIOS: IHttpClient {  /**...*/  }  

Для iOS будем использовать тот же механизм NSURLSession, что и при нативной разработке, т.к весь Network фреймворк нам доступен в нативном модуле IOS общей части. Единственное, что из-за смены контекстов мы можем работать с сессией только через делегат:

class HttpEngine : ResponseListener {     private var completion: ((Response) -> Unit)? = null      fun request(request: Request, completion: (Response) -> Unit) {         this.completion = completion          val urlSession =             NSURLSession.sessionWithConfiguration(...)          val urlRequest =             NSMutableURLRequest(NSURL.URLWithString(request.url)!!)         // background             val task = urlSession.share().dataTaskWithRequest(               urlRequest)             task?.resume()     }     override fun receiveData(data: NSData) {       /**       Main block       */     }

В качестве решения для выполнения кода в фоновом потоке мы рассмотрим Worker. Worker является механизмом Kotlin Native для работы с многопоточностью из-под коробки. Каждый worker представляет собой некую очередь работы, с помощью которой можно выполнять некоторые задачи в отдельных потоках. И на каждый worker создается свой отдельный поток:

internal fun background(block: () -> (Any?)) {     val future = worker.execute(TransferMode.SAFE, { block.share() }) {         it()     }     collectFutures.add(future) } private val worker = Worker.start() private val collectFutures = mutableListOf<Future<*>>()

Future создается при запуске некоторого функционального блока, который мы передаем в worker.execute для выполнения в producer. Мы также можем отследить выполнение future с помощью сохранения в коллекцию ссылки на него (главное, не забыть потом очистить). Чтобы наш блок выполнялся потокобезопасно, используем параметр TransferMode.SAFE и используем расширение для заморозки нашего блока:

internal fun <T> T.share(): T {     return this.freeze() }

Также мы можем получить из нашего future результат нашей работы с помощью consume и отправить его в некоторый callback-блок:

internal fun background(block: () -> (Any?), callback: (Any?)->Unit) {     val future = worker.execute(TransferMode.SAFE, { block.share() }) {         it()     }     future.consume {         main {             callback(it)         }     }     collectFutures.add(future) }

Мы можем вызвать выполнение нашего callback и в главном потоке. Для этого создаем обертку, где мы DispatchQueue.Main будем вызывать наш замороженный блок:

internal fun main(block:()->Unit) {     block.share().apply {         val freezedBlock = this         dispatch_async(dispatch_get_main_queue()) {             freezedBlock()         }     } }

Еще один момент, прежде, чем мы применим наш worker к коду. Чтобы мы не столкнулись с InvalidMutabilityException: mutation attempt of frozen, нам потребуется заморозить практически все параметры, которые мы используем для нашего сетевого запроса:

fun request(request: Request, completion: (Response) -> Unit) {   /**....*/    val urlSession =             NSURLSession.sessionWithConfiguration(                 NSURLSessionConfiguration.defaultSessionConfiguration,                responseReader.share(),                 delegateQueue = NSOperationQueue.currentQueue()             )  /**....*/         background {             val task = urlSession.share().dataTaskWithRequest(urlRequest)             task?.resume()         }     }

И теперь нам надо как-то собирать ответ, который может прийти не весь сразу, а порциями, в массив байтов:

 private var chunks = ByteArray(0)     override fun receiveData(data: NSData) {         updateChunks(data)     }      private fun updateChunks(data: NSData) {         chunks += data.toByteArray() //!CRASH!!!!     }   

Но тут мы получим краш с исключением заморозки. Просто потому, что при работе с замороженными блоками мы заморозили и сетевой клиент, и все его свойства и переменные. И изменить наш массив данных мы просто так не можем.

На помощь нам приходит API AtomicReference. Это не костыль, а полноценное, рекомендуемое и рабочее решение. AtomicReference держит ссылку на замороженный объект, который можно изменять. Для этого добавим расширение, превращающее наши объекты в атомарные, и изменим код:

internal fun <T> T.atomic(): AtomicReference<T>{     return AtomicReference(this.share()) }  //iOS Client private val chunks = ByteArray(0).atomic()  private fun updateChunks(data: NSData) {         var newValue = ByteArray(0)         newValue += chunks.value         newValue += data.toByteArray()         chunks.value = newValue.share()   }

Также добавим очистку AtomicReference после окончания работы, чтобы предотвратить захламление памяти и утечки:

 private fun clear() {         clearChunks()         completion = null     }      private fun clearChunks() {        chunks.value = ByteArray(0).share()     }

В принципе готово. Мы сможем обращаться к нашему клиенту таким образом:

actual class HttpClient : IHttpClient {     val httpEngine = HttpEngine()      /**     * @param request наш запрос с внутренними параметрами     */     actual override fun request(request: Request, completion: (Response) -> Unit) {          httpEngine.request(request) {             completion(it)         }     }

В итоге мы обошлись без скоупов корутин при запросе. И реализовали собственный асинхронный клиент iOS. Но это еще не все, что мы можем сделать в следующей части мы рассмотрим, как работать с DetachedObjectGraph и использовать Kotlin Flow для удобства.

https://github.com/anioutkazharkova/kotlin_native_network_client