Kotlin Native. Multithreading without Coroutines

In previous story we discussed how to use Coroutines for simple multithreading in Kotlin Native. This time we will talk about using native mechanism of Kotlin Native to deal with asynchronous work.

When we use Ktor library in our Kotlin Multiplatform app, we know that all asynchronous work is implemented under the hood with a help of Kotlin Native mechanism. Let’s take a look and try to apply it in our own network client.

At first, we will make a common interface for our iOS and Android Http clients. We will implement them differently, but without expect/actual mechanism:

interface IHttpClient {
    fun request(request: Request, completion: (Response)->Unit)
}

//Android with OKHttp
class HttpClientAndroid: IHttpClient {
/**...*/
}

//iOS with NSUrlSession
class HttpClientIOS: IHttpClient { 
/**...*/ 
}  

We can use OkHttp to provide all the logic in androidMain. For iOS we will use NSUrlSession in iOSMain part of our common module. All the native frameworks (such as Network) are available and accessible for us in iOSMain.

We will use standard NSUrlSession approach with some delegate to receive response:

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
      */
    }

So, in this point we need to provide some mechanism to call our task from background. For this purpose Kotlin Native has Workers. A Worker is a job queue on which you can schedule jobs to run on a different thread. Every worker performs work in its own thread:

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<*>>()

There are several details:
1. We need to use TransferMode.SAFE for thread-safe work.

2. We need to freeze a block of code to perform it in different thread. In my case, I use share() extension to incapsulate freezing.

As a result of execution we receive some Future. To deal with its completion and result value, we can use consume:

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)
}

Also we can send the completion into some another thread. For example, it could be main thread and main queue:

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

That could be done the same way we previously used to create our own MainDispatcher.
Let’s apply it to our sample. Also we need to freeze all the parameters and blocks we will use within our client:

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()
        }
    }

When we implement our own network client with a delegate approach, we can receive all the data as separated portions. So we need to use byte array to keep all the data together:

private var chunks = ByteArray(0)
 
  override fun receiveData(data: NSData) {
        updateChunks(data)
    }

    private fun updateChunks(data: NSData) {
        chunks += data.toByteArray() //!CRASH!!!!
    }

And there is a big problem. While freezing all the stuff of our client, we have frozen even the var variables. Now we cannot change our data without a crash.
To deal with it we need to use AtomicReference API. AtomicReferences hold the links to frozen variables, but allow to change their values:

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()
  }

There is a risk of potential leaks while using AtomicReferences. So we need to clear them after the work:

private fun clear() {
        clearChunks()
        completion = null
    }

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

Done. Let’s call from our iOSMain side:

actual class HttpClient : IHttpClient {
    val httpEngine = HttpEngine()

    /**
    * @param request with all parameters
    */
    actual override fun request(request: Request, completion: (Response) -> Unit) {

        httpEngine.request(request) {
            completion(it)
        }
    }