47

According to this document, using wait and notify is discouraged in Kotlin: https://kotlinlang.org/docs/reference/java-interop.html

wait()/notify()

Effective Java Item 69 kindly suggests to prefer concurrency utilities to wait() and notify(). Thus, these methods are not available on references of type Any.

However the document does not propose any correct way of doing it.

Basically, I would like to implement a service, which would read the input data and process them. If there were no input data, it would suspend itself until someone notifies that there are new input data. Something like

while (true) {
    val data = fetchData()
    processData(data)
    if (data.isEmpty()) {
        wait()
    }
}

EDIT:

I don't want to use these not recommended methods (antipatterns), I really want to find out how to do this properly.

In my case fetchData reads data from the database, so queues in my case cannot be used.

4
  • 1
    Did you check Effective Java Item 69?
    – Alberto S.
    Jun 16, 2017 at 12:51
  • 1
    You could use actors from Kotlin coroutines to implement your service. It waits for items to be sent to a channel. More info here: github.com/Kotlin/kotlinx.coroutines/blob/master/…
    – marstran
    Jun 16, 2017 at 12:56
  • You can cast any object to java.lang.Object and implement such antipatterns easily.
    – Miha_x64
    Jun 16, 2017 at 12:59
  • 1
    Currently actor from coroutines is deprecated, so this is also not good option. Sep 25, 2019 at 20:43

4 Answers 4

71

In general you should use higher-level concurrency utilities when possible.

However, if none of the higher-level constructs work in your case, the direct replacement is to use a ReentrantLock and a single Condition on that lock.

For example, if your Java code was something like:

private Object lock = new Object();

...

synchronized(lock) {
    ...
    lock.wait();
    ...
    lock.notify();
    ...
    lock.notifyAll();
    ...
}

You can change it to the following Kotlin:

private val lock = ReentrantLock()
private val condition = lock.newCondition()

lock.withLock {           // like synchronized(lock)
    ...
    condition.await()     // like wait()
    ...
    condition.signal()    // like notify()
    ...
    condition.signalAll() // like notifyAll()
    ...
}

While this is slightly more verbose, conditions do provide some extra flexibility, as you can have multiple conditions on a single lock, and there are also other kinds of locks (notably ReentrantReadWriteLock.ReadLock and ReentrantReadWriteLock.WriteLock).

Note that withLock is a Kotlin-provided extension function that takes care of calling Lock.lock()/Lock.unlock() before/after invoking the supplied lambda.

10
  • 2
    Kotlin also adds the withLock higher level function which makes this even nicer!
    – Jon Tirsen
    Oct 15, 2018 at 15:42
  • 2
    @JonTirsen Thanks! Somehow I'd missed the existence of withLock in the standard library (I swear I'd looked for something like this). I've updated the answer to use it. Oct 15, 2018 at 17:05
  • 1
    you still have to use variable boolean to check if thread is waiting or not, so for me it's still better to use synchronized + Object
    – user924
    Jun 18, 2019 at 8:19
  • 2
    Using wait()/notify() and ReeantrantLock/Condition together with Coroutines does not make sense. Both approaches block thread which is not acceptable in Couroutines. May 15, 2020 at 10:28
  • 2
    Not coroutine friendly. Nov 16, 2021 at 5:05
27

A BlockingQueue can be a suitable high-level concurrency utility for your use case, but applying it requires knowing and modifying your code structure.

The idea is, fetchData() should .take() an item from the queue, and if the queue is empty, that will block the execution until an item appears, which eliminates the .wait() in your code. The producer of the data should .put(t) the data into the queue.


If you really need to use wait and notify, e.g. for implementing a concurrency utility at low-level, you can cast a Kotlin object to java.lang.Object and call these functions afterwards, as said in the language reference. Or, written as extension functions:

@Suppress("PLATFORM_CLASS_MAPPED_TO_KOTLIN")
private fun Any.wait() = (this as java.lang.Object).wait()
3
  • Ad BlockingQueue, in my case this wouldn't work, because when notified, I need to check the data in the database. Would there be something useful for it? Ad notify, I don't really need to use it, I just need to to find something suitable :-)
    – Vojtěch
    Jun 16, 2017 at 14:23
  • @Vojtěch Maybe a Semaphore? Jun 16, 2017 at 15:23
  • 2
    Instead of the cast and @Suppress, create dedicated val lock=Object(), and use any synchronized(lock){ lock.wait(); lock.notify()} on that. This is without compiler warning. Sep 25, 2019 at 20:49
2

I'll post the unit tests to GitHub when I can. Java style wait, wait(timeout), notify using coroutines.


package com.spillikinaerospace.android.core

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel

/**
 * A Kotlin coroutines friendly method of implementing Java wait/notify.
 *
 * Exist safely on mis-matched calls. More than one wait is ignored as
 * is more than one notify if no wait was called. Only works with 2 threads. No notifyall.  
 *
 * See Locks
 * https://proandroiddev.com/synchronization-and-thread-safety-techniques-in-java-and-kotlin-f63506370e6d
 */
class SAWaitNotify {

    private val channel: Channel<Unit> = Channel<Unit>(0)
    // re-enrty block.
    private var waitingForNotify = false

    /**
     * Wait until Notify is called.
     */
    public fun saWait() {
        synchronized(this) {
            if (waitingForNotify) {
                return
            }
            waitingForNotify = true
        }

        // block forever until notify
        runBlocking {
            channel.receive()
        }

        synchronized(this) {
            waitingForNotify = false
        }
    }

    /**
     * Wait until Notify is called or the given number of seconds have passed.
     *
     * @param seconds - Max delay before unlock.
     */
    public fun saWait(seconds: Long) {
        synchronized(this) {
            if (waitingForNotify) {
                return
            }
            waitingForNotify = true
        }

        // block until timeout or notify
        runBlocking {
            // Setup our delty before calling offer.
            val job = launch(Dispatchers.IO)
            {
                delay((seconds * 1000))
                channel.offer(Unit)
            }
            // This will continue when someone calls offer.
            channel.receive()
            // Somebody called offer. Maybe the timer, maybe saNotify().
            // Channel has released but the timer job may still be running.
            // It's ok to call this even if the job has finished.
            job.cancel()
        }

        synchronized(this) {
            waitingForNotify = false
        }
    }

    /**
     * Release wait()
     */
    public fun saNotify() {
        synchronized(this) {
            if (waitingForNotify == false) {
                return
            }
            waitingForNotify = false
        }
        channel.offer(Unit)
    }
}
1
  • wait() should not tolerate a call from a second thread. This would be a coding mistake that one should know about. Modified the re-entry locks to read... if (waitingForNotify) { throw IllegalStateException("Wait() called by a second thread before notify().") } Apr 23, 2022 at 19:17
-1

You could use the kotlin Semaphore to suspend the coroutine instead of blocking a thread:

val semaphore = Semaphore(1, 0)

suspend fun runFetchLoop() = withContext(Dispatchers.Default){
    while (isActive) {
        val data = fetchData()
        processData(data)
        semaphore.acquire()
        if (data.isEmpty()) {
            semaphore.acquire()
        }
        semaphore.release()
    }
}

fun notifyDataAvailable() = semaphore.release()

Here is a running example: https://pl.kotl.in/bnKeOspfs .

However I would prefer a cold Flow or a hot Channel to solve this problem. Here is a good article about cold flows and hot channels of the great Roman Elizarov

3
  • I think the example in this answer is wrong. There is a race condition. I assume, that notify() is called after fetchData() returns something different than before. If notify() is called after data.isEmpty() is true, but before mutex.lock(), then coroutine waiting on mutex.lock() will wait forever. May 15, 2020 at 10:25
  • @PavelČernocký you are right. To fix it, I replaced the mutex by a semaphore and two acquire calls. This chains condition and suspension together. If the notify call comes in between the isEmpty condition and the acquire call, then it won't result in an unwanted unlimited suspension anymore. May 18, 2020 at 8:46
  • This fix has another problems. acquire()/release() has to be in pair what is not guaranteed here. Especially when notifyDataAvailable() is called multiple times, it will throw an exception, because there will be no Semaphore permits to release. In general i don't see this approach is right, it certainly can not be generalized. The whole while is constructed badly. data is local variable which is filled by fetchData, so fetchData should be coordinated with notifyDataAvailable. May 20, 2020 at 7:26

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.