# Java: concurrency questions

**URL:** <https://forums.foundationdb.org/t/java-concurrency-questions/1571>\
**Category:** Using FoundationDB\
**Tags:** bindings\
**Created:** [August 7, 2019, 7:22pm UTC](https://forums.foundationdb.org/t/java-concurrency-questions/1571 "2019-08-07T19:22:14Z")\
**Posts on this page:** 7\
**Page:** 1

<div class="post-metadata">

**Author:** ![jwr](https://sea1.discourse-cdn.com/foundationdb/user_avatar/forums.foundationdb.org/jwr/32/302_2.png) [@jwr](https://forums.foundationdb.org/u/jwr)\
**Post date:** [August 7, 2019, 7:22pm UTC](https://forums.foundationdb.org/t/java-concurrency-questions/1571/1 "2019-08-07T19:22:14Z")

</div>

I’m working on Clojure bindings for FoundationDB, which means I’m using the Java API. Some things are unclear to me even after looking at the source code and reading the documentation, so I thought I’d ask and make sure:

- Do I understand correctly that if I call `forEach` on the `AsyncIterable`, I can expect to get a callback whenever an item becomes available, without tying up the FDB Java API?
- Is there a better way to achieve the same thing (get a callback for each item becoming available)?
- What if my callback function blocks because of backpressure? Is that something that the FDB API expects?
- Calling `open` on an `FDB` instance will return a `Database` object. Is this object intended to be shared by all threads within the application? Will the underlying networking code “do the right thing”?
- Assuming I do not perform my own blocking operations in transactions, do I need to worry about thread pool size and the potential for deadlocks from thread pool exhaustion if multiple long-running operations are launched?

---

<div class="post-metadata">

**Author:** ![alloc](https://sea1.discourse-cdn.com/foundationdb/user_avatar/forums.foundationdb.org/alloc/32/9_2.png) [@alloc](https://forums.foundationdb.org/u/alloc)\
**Post date:** [August 8, 2019, 12:14am UTC](https://forums.foundationdb.org/t/java-concurrency-questions/1571/2 "2019-08-08T00:14:28Z")

</div>

> [@jwr](#):
>
> Do I understand correctly that if I call `forEach` on the `AsyncIterable` , I can expect to get a callback whenever an item becomes available, without tying up the FDB Java API?

Hm, I think it depends on what you mean. We don’t override the `forEach` method, so it’s exactly equivalent to:

```java
for (T item : asyncIterable) {
    func.apply(item);
}

```

So it will block and eat up a thread doing so, but the callback will, in fact, be invoked as results become available. That thread is the _caller_ thread, rather than, say, the FDB native thread that is actually serving requests, so it won’t tie up everything, but it will eat up resources.

There’s an asynchronous version that executes its callbacks on a user-providable executor, if you want that behavior instead: [AsyncUtil (FoundationDB Java Client API)](https://apple.github.io/foundationdb/javadoc/com/apple/foundationdb/async/AsyncUtil.html#forEach-com.apple.foundationdb.async.AsyncIterable-java.util.function.Consumer-java.util.concurrent.Executor-)

> [@jwr](#):
>
> Is there a better way to achieve the same thing (get a callback for each item becoming available)?

I think the `forEachAsync` method in `AsyncUtil` (linked above) is what you want.

> [@jwr](#):
>
> What if my callback function blocks because of backpressure? Is that something that the FDB API expects?

If your function ends up blocking, then it will block the execution thread of whatever executor is running it. The better way to handle that is to return a future to wait instead of blocking. We implemented something like that using a scheduled executor in the Record Layer: [https://github.com/FoundationDB/fdb-record-layer/blob/152d3c5b16da550edc511c8d29cff131df9d88f7/fdb-extensions/src/main/java/com/apple/foundationdb/async/MoreAsyncUtil.java#L802](https://github.com/FoundationDB/fdb-record-layer/blob/152d3c5b16da550edc511c8d29cff131df9d88f7/fdb-extensions/src/main/java/com/apple/foundationdb/async/MoreAsyncUtil.java#L802)

> [@jwr](#):
>
> Calling `open` on an `FDB` instance will return a `Database` object. Is this object intended to be shared by all threads within the application? Will the underlying networking code “do the right thing”?

Yep, the `Database` object can be shared by multiple threads (and, indeed, this is expected behavior). The underlying code executes everything on a single threaded executor (more-or-less), so all the complicated stuff won’t care about how many threads you have calling it.

> [@jwr](#):
>
> Assuming I do not perform my own blocking operations in transactions, do I need to worry about thread pool size and the potential for deadlocks from thread pool exhaustion if multiple long-running operations are launched?

Hm. If everything is, in fact, non-blocking, then in general, you shouldn’t need to worry too much about thread pool exhaustion except that (as stated above), there is one (and only one) network thread at the native level, so _that_ thread can get clogged up. I don’t know if I’d say that you “never” need to worry about backing up, say, a thread pool that executes callbacks from async work, but if you don’t have any blocking operations, I wouldn’t be too concerned about it. But be forewarned that it can be easy to accidentally introduce blocking code and accidentally tie up threads in your thread pool.

---

<div class="post-metadata">

**Author:** ![jwr](https://sea1.discourse-cdn.com/foundationdb/user_avatar/forums.foundationdb.org/jwr/32/302_2.png) [@jwr](https://forums.foundationdb.org/u/jwr)\
**Post date:** [August 8, 2019, 7:14pm UTC](https://forums.foundationdb.org/t/java-concurrency-questions/1571/3 "2019-08-08T19:14:43Z")

</div>

Hmm. I am looking at `forEach` here: [https://github.com/apple/foundationdb/blob/ff78d27575bc793f9d69d6a217b076ebd074a4df/bindings/java/src/main/com/apple/foundationdb/async/AsyncUtil.java#L95](https://github.com/apple/foundationdb/blob/ff78d27575bc793f9d69d6a217b076ebd074a4df/bindings/java/src/main/com/apple/foundationdb/async/AsyncUtil.java#L95) which would appear to call `forEachRemaining` with a `DEFAULT_EXECUTOR` [https://github.com/apple/foundationdb/blob/ff78d27575bc793f9d69d6a217b076ebd074a4df/bindings/java/src/main/com/apple/foundationdb/async/AsyncUtil.java#L131](https://github.com/apple/foundationdb/blob/ff78d27575bc793f9d69d6a217b076ebd074a4df/bindings/java/src/main/com/apple/foundationdb/async/AsyncUtil.java#L131)

I am not sure I fully understand the behavior of `forEachRemaining`, but at a first glance it appears to be what I need: it calls `consumer.accept()` for every item. I just don’t know what will happen if the consumer blocks because the downstream isn’t ready for more data.

Let me ask a more general question, which will perhaps explain where I’m coming from. Let’s imagine I’m doing a monster `getRange()` query. As the results start flowing in (I’m assuming they aren’t all buffered in memory, but are being streamed), my code is consuming them, but can’t keep up. Even if there are buffers, they are not infinite, so there needs to be a backpressure mechanism. How do I consume these results, providing backpressure if needed, but without tying up critical FDB threads? Ideally I’d just block in `consumer.accept()`, but have to know that this is expected.

I was also hoping for a mechanism where FDB controls what happens, so that I don’t need to manage a separate thread pool with threads that block waiting for data (so, promises don’t help much in this case).

> But be forewarned that it can be easy to accidentally introduce blocking code and accidentally tie up threads in your thread pool

Yes, I’ve just had this happen in a large and complex application (not using FDB) and the resulting deadlocks are incredibly difficult to debug. That’s why I’m being extra careful this time.

---

<div class="post-metadata">

**Author:** ![ajbeamon](https://sea1.discourse-cdn.com/foundationdb/user_avatar/forums.foundationdb.org/ajbeamon/32/13_2.png) [@ajbeamon](https://forums.foundationdb.org/u/ajbeamon)\
**Post date:** [August 8, 2019, 8:17pm UTC](https://forums.foundationdb.org/t/java-concurrency-questions/1571/4 "2019-08-08T20:17:05Z")

</div>

> [@jwr](#):
>
> which would appear to call `forEachRemaining` with a `DEFAULT_EXECUTOR`

There is also a variant to specify your own executor:

> <https://github.com/apple/foundationdb/blob/ff78d27575bc793f9d69d6a217b076ebd074a4df/bindings/java/src/main/com/apple/foundationdb/async/AsyncUtil.java#L114>

> [@jwr](#):
>
> I am not sure I fully understand the behavior of `forEachRemaining` , but at a first glance it appears to be what I need: it calls `consumer.accept()` for every item.

`forEachRemaining` is a non-blocking function (at least with respect to reading from the iterator) that calls your `Consumer` on every item remaining in an `AsyncIterator`. When called from `forEach`, you’ll get a fresh iterator from your `AsyncIterable` and will therefore consume every item.

> [@jwr](#):
>
> I just don’t know what will happen if the consumer blocks because the downstream isn’t ready for more data.

This will block whatever thread the executor is using to run this work. It won’t block the FDB network thread (where all of the FDB client-side work happens) unless you are using some executor that is running everything on the network thread, which I wouldn’t recommend.

> [@jwr](#):
>
> As the results start flowing in (I’m assuming they aren’t all buffered in memory, but are being streamed), my code is consuming them, but can’t keep up. Even if there are buffers, they are not infinite, so there needs to be a backpressure mechanism. How do I consume these results, providing backpressure if needed, but without tying up critical FDB threads? Ideally I’d just block in `consumer.accept()` , but have to know that this is expected.

When reading ranges, we fetch the results in batches, only starting a new batch once we’ve exhausted a prior batch. If you slow down your consumption of the iterator, then it will slow down the rate at which results are requested from the cluster. It isn’t necessary that you slow things down by blocking, though. As Alec mentioned, you could instead return a future from your function that processes each result and avoid tying up thread resources. I don’t think we have a variant of `forEach` to do this exactly, but you could write one using `whileTrue` if desired.

If you aren’t concerned about blocking threads in your executor, then blocking in the consumption of items should be safe to do from the perspective of the FDB network thread (again with the caveat that you aren’t using an executor that runs things on the FDB network thread).

In either case (either with blocking or using futures), you’ll need to avoid waiting too long if you want to be able to read a range transactionally. After the first read, a transaction will only be able to continue reading for 5 seconds, and since the results are fetched in batches you may find that later batches fail if it takes too long to get to them. In fact, even if you don’t block at all you’ll still be limited in how large a range you can read in one transaction.

If you thought that the processing of data was going to be slow enough to be a problem with respect to this time limit, you could force the reading of all the data at once using the [`asList` function](https://github.com/apple/foundationdb/blob/ff78d27575bc793f9d69d6a217b076ebd074a4df/bindings/java/src/main/com/apple/foundationdb/async/AsyncIterable.java#L50).

---

<div class="post-metadata">

**Author:** ![jwr](https://sea1.discourse-cdn.com/foundationdb/user_avatar/forums.foundationdb.org/jwr/32/302_2.png) [@jwr](https://forums.foundationdb.org/u/jwr)\
**Post date:** [August 9, 2019, 8:04pm UTC](https://forums.foundationdb.org/t/java-concurrency-questions/1571/5 "2019-08-09T20:04:57Z")

</div>

> [@ajbeamon](#):
>
> When reading ranges, we fetch the results in batches, only starting a new batch once we’ve exhausted a prior batch. If you slow down your consumption of the iterator, then it will slow down the rate at which results are requested from the cluster. It isn’t necessary that you slow things down by blocking, though. As Alec mentioned, you could instead return a future from your function that processes each result and avoid tying up thread resources. I don’t think we have a variant of `forEach` to do this exactly, but you could write one using `whileTrue` if desired.

After digging deeper into Java async programming, I think I need to do something different and access the iterator myself. I plan to write something similar to `forEachRemaining` but not in the form of a loop. The difference would be that it would call the subsequent `iterator.onHasNext()` only after the (Clojure) `put!` to my outgoing channel has succeeded (from my `put!` callback). A recursion pattern similar to the one shown on this page [Go Block Best Practices · clojure/core.async Wiki · GitHub](https://github.com/clojure/core.async/wiki/Go-Block-Best-Practices)

From what I understand, this should not block any threads, because it would use the async facilities both on the Java and on Clojure side, and it would provide backpressure. It would also not create any extra thread pools or buffers, which I’m really trying to avoid.

I do realize there is a time limit on transactions. That’s fine. I’m writing a library here, and I’d like to get things correct, regardless of what the application actually does. I’m not actually concerned about multi-second reads, but mostly about an incorrect implementation where some requests throttle others.

By the way, your (that’s a collective team “your”) answers in these forums are fantastic. I don’t know how you find the time or the energy to do this, but the resulting value is incredible. Thank you!

---

<div class="post-metadata">

**Author:** ![osamarin](https://sea1.discourse-cdn.com/foundationdb/user_avatar/forums.foundationdb.org/osamarin/32/905_2.png) [@osamarin](https://forums.foundationdb.org/u/osamarin)\
**Post date:** [June 18, 2020, 1:11pm UTC](https://forums.foundationdb.org/t/java-concurrency-questions/1571/6 "2020-06-18T13:11:13Z")

</div>

Hello, @alloc. There is no forEachAsync method in AsyncUtils class

---

<div class="post-metadata">

**Author:** ![alloc](https://sea1.discourse-cdn.com/foundationdb/user_avatar/forums.foundationdb.org/alloc/32/9_2.png) [@alloc](https://forums.foundationdb.org/u/alloc)\
**Post date:** [June 18, 2020, 3:02pm UTC](https://forums.foundationdb.org/t/java-concurrency-questions/1571/7 "2020-06-18T15:02:27Z")

</div>

Right, yeah. Sorry for the confusion. I meant the `forEach` method on `AsyncUtil`, sorry.