Coroutine Thread Safety with Retrofit-CodePudding

I have still a little bit of trouble putting all information together about the thread-safety of using coroutines to launch network requests.

Let's say we have following use-case, there is a list of users we get and for each of those users, I will do some specific check which has to run over a network request to the API, giving me some information back about this user.

The userCheck happens inside a library, which doesn't expose suspend functions but rather still uses a callback. Inside of this library, I have seen code like this to launch each of the network requests:

internal suspend fun <T> doNetworkRequest(request: suspend () -> Response<T>): NetworkResult<T> {
    return withContext(Dispatchers.IO) {
        try {
            val response = request.invoke()
            ...

According to the documentation, Dispatchers.IO can use multiple threads for the execution of the code, also the request function is simply a function from a Retrofit API.

So what I did is to launch the request for each user, and use a single resultHandler object, which will add the results to a list and check if the length of the result list equals the length of the user list, if so, then all userChecks are done and I know that I can do something with the results, which need to be returned all together.

val userList: List<String>? = getUsers()
val userCheckResultList = mutableListOf<UserCheckResult>()
val handler = object : UserCheckResultHandler {
                  override fun onResult(
                        userCheckResult: UserCheckResult?
                  ) {
                        userCheckResult?.let {
                            userCheckResultList.add(
                                it
                            )
                        }
                        if (userCheckResultList.size == userList?.size) {
                            doSomethingWithResultList()
                            print("SUCCESS")
                        }
                    }
                }

userList?.forEach {
    checkUser(it, handler)
}

My question is: Is this implementation thread-safe? As far as I know, Kotlin objects should be thread safe, but I have gotten feedback that this is possibly not the best implementation :D

But in theory, even if the requests get launched asynchronous and multiple at the same time, only one at a time can access the lock of the thread the result handler is running on and there will be no race condition or problems with adding items to the list and comparing the sizes.

Am I wrong about this? Is there any way to handle this scenario in a better way?

CodePudding user response：

If you are executing multiple request in parallel - it's not. List is not thread safe. But it's simple fix for that. Create a Mutex object and then just wrap your operation on list in lock, like that:

val lock = Mutex()
val userList: List<String>? = getUsers()
val userCheckResultList = mutableListOf<UserCheckResult>()
val handler = object : UserCheckResultHandler {
                  override fun onResult(
                        userCheckResult: UserCheckResult?
                  ) {
                        lock.withLock {
                            userCheckResult?.let {
                                userCheckResultList.add(
                                    it
                                )
                            }
                            if (userCheckResultList.size == userList?.size) {
                                doSomethingWithResultList()
                                print("SUCCESS")
                            }
                        }
                    }
                }

userList?.forEach {
    checkUser(it, handler)
}

I have to add that this whole solution seems very hacky. I would go completely other route. Run all of your requests wrapping those in async { // network request } which will return Deferred object. Add this object to some list. After that wait for all of those deferred objects using awaitAll(). Like that:

val jobs = mutableListOf<Job>()
userList?.forEach {
   // i assume checkUser is suspendable here
   jobs  = async { checkUser(it, handler) }
}

// wait for all requests
jobs.awaitAll()

// After that you can access all results like this:
val resultOfJob0 = jobs[0].getCompleted()