Pattern for sharing a resource across all worker threads in Beam Java SDK

[Jeff Klukas <jk...@mozilla.com>]

Beam Java users,

I've run into a few cases where I want to present a single thread-safe data
structure to all threads on a worker, and I end up writing a good bit of
custom code each time involving a synchronized method that handles creating
the resource exactly once, and then each thread has its own reference to
the singleton. I don't have extensive experience with thread safety in
Java, so it seems likely I'm going to get this wrong.

Are there any best practices for state that is shared across threads? Any
prior art I can read up on?

The most concrete case I have in mind is loading a GeoIP database for doing
city lookups from IP addresses. We're using MaxMind's API which allows
mapping a portion of memory to a file sitting on disk. We have a
synchronized method that checks if the reader has been initialized [0] ; if
not, we copy the database file from GCS to local disk, build the
DatabaseReader instance, and return it. Other threads will see the
already-initialized and just get a reference to it instead.

This all appears to work, and it saves memory compared to each thread
maintaining their own DatabaseReader. But is there a safer or more built-in
way to do this? Am I missing relevant hooks in the Beam API that would make
this cleaner?

[0]
https://github.com/mozilla/gcp-ingestion/blob/master/ingestion-beam/src/main/java/com/mozilla/telemetry/decoder/GeoCityLookup.java#L95

Re: Pattern for sharing a resource across all worker threads in Beam Java SDK

[Kenneth Knowles <ke...@apache.org>]

On Fri, May 1, 2020 at 2:01 PM Luke Cwik <lc...@google.com> wrote:

> All worker threads are going to see the same instance.
> In Dataflow there is only one JVM containing user code per VM instance.
>
> One of the reasons around the portability effort is having the worker
> nodes/JVMs be consistent across runners so users have an easier time to
> choose the runner that works best for them and not get snagged on runner
> environment differences.
>

I agree with Luke's main point - our runners have pretty wildly different
infrastructural details that make it hard to just switch between them, and
portability improves that. But I would also caution against taking this too
far, especially for this inquiry. The solution to your problem here really
shouldn't require much about the environment. Please don't make assumptions
about threads per JVM, or JVMs per VM.

Portability's predictable environments will undoubtedly make it easy to
depend on these and other things that are subject to change/breakage
without notice. Differences aren't just across runners, but across time and
codepaths.

Kenn



> On Fri, May 1, 2020 at 7:01 AM Jeff Klukas <jk...@mozilla.com> wrote:
>
>> Cameron's point about protecting return values is quite reasonable. I
>> agree it would be good to make sure we're returning an returning an
>> unmodifiable set in my getOrCreateSingletonAllowedCities case.
>>
>> Luke's suggestion of Supplier.memoize sounds like a more elegant way of
>> doing essentially the same thing we're currently implementing, so
>> definitely worth looking into.
>>
>> The PipelineOptions-based solution is also intriguing. When we pass
>> values into transforms from PipelineOptions, are all worker threads going
>> to see the same instance? Or does PipelineOptions end up getting serialized
>> and reconstructed by each thread? I see that your example uses JsonIgnore
>> to prevent a computed value being serialized, but it's unclear whether the
>> default value will be generated once per thread or once per worker.
>>
>> I suppose the Beam model doesn't make any statement about how runners
>> handle multiple threads per worker, so there's no guarantee about whether
>> multiple threads are executing in the same JVM or not. Am I correct in
>> assuming at least in the Dataflow case that there is one JVM per worker
>> node?
>>
>> On Thu, Apr 30, 2020 at 7:05 PM Luke Cwik <lc...@google.com> wrote:
>>
>>> I looked at your example and the custom logic for the singleton is
>>> basically:
>>> static transient T value;
>>> public static synchronized getOrCreate(...) {
>>>   if (value == null) {
>>>      ... instantiate value ...
>>>   }
>>>   return value;
>>> }
>>>
>>> Which is only a few lines. You could use one of Java's injection
>>> frameworks like Spring or Guice or ... as they commonly provide you a
>>> singleton pattern but the complexity overhead a lot of the time isn't worth
>>> it.
>>>
>>> If all your logic can be provided by PipelineOptions you could combine a
>>> DefaultValueFactory and a PipelineOption that returns T. First time an
>>> unset value is accessed the PipelineOption will be created and the
>>> synchronization is provided for you by PipelineOptions. It would look like:
>>>
>>> public interface MyPipelineOptions extends PipelineOptions {
>>>   @Default.InstanceFactory(TFactory.class);
>>>   @JsonIgnore  // prevents the value from being saved and makes it
>>> "local" to the process
>>>   T getT();
>>>   void setT(T t);
>>>
>>>   public static class TFactory implements DefaultValueFactory<T> {
>>>     public T create(PipelineOptions options) {
>>>       ... instantiate value ...
>>>     }
>>>   }
>>> }
>>> and then in your DoFn you access your PipelineOption as normal. This is
>>> also convenient because you can set the value during testing of your DoFn.
>>>
>>> Another alternative would be to look for something like Guavas
>>> memoize[1] functions and use them as they are very lightweight.
>>>
>>> 1:
>>> https://guava.dev/releases/19.0/api/docs/com/google/common/base/Suppliers.html#memoize(com.google.common.base.Supplier)
>>>
>>> On Thu, Apr 30, 2020 at 12:45 PM Jeff Klukas <jk...@mozilla.com>
>>> wrote:
>>>
>>>> Beam Java users,
>>>>
>>>> I've run into a few cases where I want to present a single thread-safe
>>>> data structure to all threads on a worker, and I end up writing a good bit
>>>> of custom code each time involving a synchronized method that handles
>>>> creating the resource exactly once, and then each thread has its own
>>>> reference to the singleton. I don't have extensive experience with thread
>>>> safety in Java, so it seems likely I'm going to get this wrong.
>>>>
>>>> Are there any best practices for state that is shared across threads?
>>>> Any prior art I can read up on?
>>>>
>>>> The most concrete case I have in mind is loading a GeoIP database for
>>>> doing city lookups from IP addresses. We're using MaxMind's API which
>>>> allows mapping a portion of memory to a file sitting on disk. We have a
>>>> synchronized method that checks if the reader has been initialized [0] ; if
>>>> not, we copy the database file from GCS to local disk, build the
>>>> DatabaseReader instance, and return it. Other threads will see the
>>>> already-initialized and just get a reference to it instead.
>>>>
>>>> This all appears to work, and it saves memory compared to each thread
>>>> maintaining their own DatabaseReader. But is there a safer or more built-in
>>>> way to do this? Am I missing relevant hooks in the Beam API that would make
>>>> this cleaner?
>>>>
>>>> [0]
>>>> https://github.com/mozilla/gcp-ingestion/blob/master/ingestion-beam/src/main/java/com/mozilla/telemetry/decoder/GeoCityLookup.java#L95
>>>>
>>>

Re: Pattern for sharing a resource across all worker threads in Beam Java SDK

[Luke Cwik <lc...@google.com>]

All worker threads are going to see the same instance.
In Dataflow there is only one JVM containing user code per VM instance.

One of the reasons around the portability effort is having the worker
nodes/JVMs be consistent across runners so users have an easier time to
choose the runner that works best for them and not get snagged on runner
environment differences.

On Fri, May 1, 2020 at 7:01 AM Jeff Klukas <jk...@mozilla.com> wrote:

> Cameron's point about protecting return values is quite reasonable. I
> agree it would be good to make sure we're returning an returning an
> unmodifiable set in my getOrCreateSingletonAllowedCities case.
>
> Luke's suggestion of Supplier.memoize sounds like a more elegant way of
> doing essentially the same thing we're currently implementing, so
> definitely worth looking into.
>
> The PipelineOptions-based solution is also intriguing. When we pass values
> into transforms from PipelineOptions, are all worker threads going to see
> the same instance? Or does PipelineOptions end up getting serialized and
> reconstructed by each thread? I see that your example uses JsonIgnore to
> prevent a computed value being serialized, but it's unclear whether the
> default value will be generated once per thread or once per worker.
>
> I suppose the Beam model doesn't make any statement about how runners
> handle multiple threads per worker, so there's no guarantee about whether
> multiple threads are executing in the same JVM or not. Am I correct in
> assuming at least in the Dataflow case that there is one JVM per worker
> node?
>
> On Thu, Apr 30, 2020 at 7:05 PM Luke Cwik <lc...@google.com> wrote:
>
>> I looked at your example and the custom logic for the singleton is
>> basically:
>> static transient T value;
>> public static synchronized getOrCreate(...) {
>>   if (value == null) {
>>      ... instantiate value ...
>>   }
>>   return value;
>> }
>>
>> Which is only a few lines. You could use one of Java's injection
>> frameworks like Spring or Guice or ... as they commonly provide you a
>> singleton pattern but the complexity overhead a lot of the time isn't worth
>> it.
>>
>> If all your logic can be provided by PipelineOptions you could combine a
>> DefaultValueFactory and a PipelineOption that returns T. First time an
>> unset value is accessed the PipelineOption will be created and the
>> synchronization is provided for you by PipelineOptions. It would look like:
>>
>> public interface MyPipelineOptions extends PipelineOptions {
>>   @Default.InstanceFactory(TFactory.class);
>>   @JsonIgnore  // prevents the value from being saved and makes it
>> "local" to the process
>>   T getT();
>>   void setT(T t);
>>
>>   public static class TFactory implements DefaultValueFactory<T> {
>>     public T create(PipelineOptions options) {
>>       ... instantiate value ...
>>     }
>>   }
>> }
>> and then in your DoFn you access your PipelineOption as normal. This is
>> also convenient because you can set the value during testing of your DoFn.
>>
>> Another alternative would be to look for something like Guavas memoize[1]
>> functions and use them as they are very lightweight.
>>
>> 1:
>> https://guava.dev/releases/19.0/api/docs/com/google/common/base/Suppliers.html#memoize(com.google.common.base.Supplier)
>>
>> On Thu, Apr 30, 2020 at 12:45 PM Jeff Klukas <jk...@mozilla.com> wrote:
>>
>>> Beam Java users,
>>>
>>> I've run into a few cases where I want to present a single thread-safe
>>> data structure to all threads on a worker, and I end up writing a good bit
>>> of custom code each time involving a synchronized method that handles
>>> creating the resource exactly once, and then each thread has its own
>>> reference to the singleton. I don't have extensive experience with thread
>>> safety in Java, so it seems likely I'm going to get this wrong.
>>>
>>> Are there any best practices for state that is shared across threads?
>>> Any prior art I can read up on?
>>>
>>> The most concrete case I have in mind is loading a GeoIP database for
>>> doing city lookups from IP addresses. We're using MaxMind's API which
>>> allows mapping a portion of memory to a file sitting on disk. We have a
>>> synchronized method that checks if the reader has been initialized [0] ; if
>>> not, we copy the database file from GCS to local disk, build the
>>> DatabaseReader instance, and return it. Other threads will see the
>>> already-initialized and just get a reference to it instead.
>>>
>>> This all appears to work, and it saves memory compared to each thread
>>> maintaining their own DatabaseReader. But is there a safer or more built-in
>>> way to do this? Am I missing relevant hooks in the Beam API that would make
>>> this cleaner?
>>>
>>> [0]
>>> https://github.com/mozilla/gcp-ingestion/blob/master/ingestion-beam/src/main/java/com/mozilla/telemetry/decoder/GeoCityLookup.java#L95
>>>
>>

Re: Pattern for sharing a resource across all worker threads in Beam Java SDK

[Jeff Klukas <jk...@mozilla.com>]

Cameron's point about protecting return values is quite reasonable. I agree
it would be good to make sure we're returning an returning an unmodifiable
set in my getOrCreateSingletonAllowedCities case.

Luke's suggestion of Supplier.memoize sounds like a more elegant way of
doing essentially the same thing we're currently implementing, so
definitely worth looking into.

The PipelineOptions-based solution is also intriguing. When we pass values
into transforms from PipelineOptions, are all worker threads going to see
the same instance? Or does PipelineOptions end up getting serialized and
reconstructed by each thread? I see that your example uses JsonIgnore to
prevent a computed value being serialized, but it's unclear whether the
default value will be generated once per thread or once per worker.

I suppose the Beam model doesn't make any statement about how runners
handle multiple threads per worker, so there's no guarantee about whether
multiple threads are executing in the same JVM or not. Am I correct in
assuming at least in the Dataflow case that there is one JVM per worker
node?

On Thu, Apr 30, 2020 at 7:05 PM Luke Cwik <lc...@google.com> wrote:

> I looked at your example and the custom logic for the singleton is
> basically:
> static transient T value;
> public static synchronized getOrCreate(...) {
>   if (value == null) {
>      ... instantiate value ...
>   }
>   return value;
> }
>
> Which is only a few lines. You could use one of Java's injection
> frameworks like Spring or Guice or ... as they commonly provide you a
> singleton pattern but the complexity overhead a lot of the time isn't worth
> it.
>
> If all your logic can be provided by PipelineOptions you could combine a
> DefaultValueFactory and a PipelineOption that returns T. First time an
> unset value is accessed the PipelineOption will be created and the
> synchronization is provided for you by PipelineOptions. It would look like:
>
> public interface MyPipelineOptions extends PipelineOptions {
>   @Default.InstanceFactory(TFactory.class);
>   @JsonIgnore  // prevents the value from being saved and makes it "local"
> to the process
>   T getT();
>   void setT(T t);
>
>   public static class TFactory implements DefaultValueFactory<T> {
>     public T create(PipelineOptions options) {
>       ... instantiate value ...
>     }
>   }
> }
> and then in your DoFn you access your PipelineOption as normal. This is
> also convenient because you can set the value during testing of your DoFn.
>
> Another alternative would be to look for something like Guavas memoize[1]
> functions and use them as they are very lightweight.
>
> 1:
> https://guava.dev/releases/19.0/api/docs/com/google/common/base/Suppliers.html#memoize(com.google.common.base.Supplier)
>
> On Thu, Apr 30, 2020 at 12:45 PM Jeff Klukas <jk...@mozilla.com> wrote:
>
>> Beam Java users,
>>
>> I've run into a few cases where I want to present a single thread-safe
>> data structure to all threads on a worker, and I end up writing a good bit
>> of custom code each time involving a synchronized method that handles
>> creating the resource exactly once, and then each thread has its own
>> reference to the singleton. I don't have extensive experience with thread
>> safety in Java, so it seems likely I'm going to get this wrong.
>>
>> Are there any best practices for state that is shared across threads? Any
>> prior art I can read up on?
>>
>> The most concrete case I have in mind is loading a GeoIP database for
>> doing city lookups from IP addresses. We're using MaxMind's API which
>> allows mapping a portion of memory to a file sitting on disk. We have a
>> synchronized method that checks if the reader has been initialized [0] ; if
>> not, we copy the database file from GCS to local disk, build the
>> DatabaseReader instance, and return it. Other threads will see the
>> already-initialized and just get a reference to it instead.
>>
>> This all appears to work, and it saves memory compared to each thread
>> maintaining their own DatabaseReader. But is there a safer or more built-in
>> way to do this? Am I missing relevant hooks in the Beam API that would make
>> this cleaner?
>>
>> [0]
>> https://github.com/mozilla/gcp-ingestion/blob/master/ingestion-beam/src/main/java/com/mozilla/telemetry/decoder/GeoCityLookup.java#L95
>>
>

Re: Pattern for sharing a resource across all worker threads in Beam Java SDK

[Luke Cwik <lc...@google.com>]

I looked at your example and the custom logic for the singleton is
basically:
static transient T value;
public static synchronized getOrCreate(...) {
  if (value == null) {
     ... instantiate value ...
  }
  return value;
}

Which is only a few lines. You could use one of Java's injection frameworks
like Spring or Guice or ... as they commonly provide you a singleton
pattern but the complexity overhead a lot of the time isn't worth it.

If all your logic can be provided by PipelineOptions you could combine a
DefaultValueFactory and a PipelineOption that returns T. First time an
unset value is accessed the PipelineOption will be created and the
synchronization is provided for you by PipelineOptions. It would look like:

public interface MyPipelineOptions extends PipelineOptions {
  @Default.InstanceFactory(TFactory.class);
  @JsonIgnore  // prevents the value from being saved and makes it "local"
to the process
  T getT();
  void setT(T t);

  public static class TFactory implements DefaultValueFactory<T> {
    public T create(PipelineOptions options) {
      ... instantiate value ...
    }
  }
}
and then in your DoFn you access your PipelineOption as normal. This is
also convenient because you can set the value during testing of your DoFn.

Another alternative would be to look for something like Guavas memoize[1]
functions and use them as they are very lightweight.

1:
https://guava.dev/releases/19.0/api/docs/com/google/common/base/Suppliers.html#memoize(com.google.common.base.Supplier)

On Thu, Apr 30, 2020 at 12:45 PM Jeff Klukas <jk...@mozilla.com> wrote:

> Beam Java users,
>
> I've run into a few cases where I want to present a single thread-safe
> data structure to all threads on a worker, and I end up writing a good bit
> of custom code each time involving a synchronized method that handles
> creating the resource exactly once, and then each thread has its own
> reference to the singleton. I don't have extensive experience with thread
> safety in Java, so it seems likely I'm going to get this wrong.
>
> Are there any best practices for state that is shared across threads? Any
> prior art I can read up on?
>
> The most concrete case I have in mind is loading a GeoIP database for
> doing city lookups from IP addresses. We're using MaxMind's API which
> allows mapping a portion of memory to a file sitting on disk. We have a
> synchronized method that checks if the reader has been initialized [0] ; if
> not, we copy the database file from GCS to local disk, build the
> DatabaseReader instance, and return it. Other threads will see the
> already-initialized and just get a reference to it instead.
>
> This all appears to work, and it saves memory compared to each thread
> maintaining their own DatabaseReader. But is there a safer or more built-in
> way to do this? Am I missing relevant hooks in the Beam API that would make
> this cleaner?
>
> [0]
> https://github.com/mozilla/gcp-ingestion/blob/master/ingestion-beam/src/main/java/com/mozilla/telemetry/decoder/GeoCityLookup.java#L95
>

Re: Pattern for sharing a resource across all worker threads in Beam Java SDK

[Cameron Bateman <cb...@gmail.com>]

I'm interested in this area too.  One limitation I guess is that this
assumes your runner is going to be single JVM if you need your singletons
to be globally unique.  I'm mostly using DirectRunner (I'm still new to all
this) for which this holds.  I suppose for more distributed runners this
would be a more challenging problem.

One tip I would give for your code is protect your singleton return
values.  For example, the Set<Integer> that you return from
getOrCreateSingletonAllowedCities.
If you want that to be modifiable, you should wrap it using something like
Collections.synchronizedSet().  If you want it to be immutable, use
unmodifiableSet().  Note that even if the general problem of making these
singletons globally available is solved by the framework, you will still
need to make your singletons thread-safe.

--Cam

On Thu, Apr 30, 2020 at 12:45 PM Jeff Klukas <jk...@mozilla.com> wrote:

> Beam Java users,
>
> I've run into a few cases where I want to present a single thread-safe
> data structure to all threads on a worker, and I end up writing a good bit
> of custom code each time involving a synchronized method that handles
> creating the resource exactly once, and then each thread has its own
> reference to the singleton. I don't have extensive experience with thread
> safety in Java, so it seems likely I'm going to get this wrong.
>
> Are there any best practices for state that is shared across threads? Any
> prior art I can read up on?
>
> The most concrete case I have in mind is loading a GeoIP database for
> doing city lookups from IP addresses. We're using MaxMind's API which
> allows mapping a portion of memory to a file sitting on disk. We have a
> synchronized method that checks if the reader has been initialized [0] ; if
> not, we copy the database file from GCS to local disk, build the
> DatabaseReader instance, and return it. Other threads will see the
> already-initialized and just get a reference to it instead.
>
> This all appears to work, and it saves memory compared to each thread
> maintaining their own DatabaseReader. But is there a safer or more built-in
> way to do this? Am I missing relevant hooks in the Beam API that would make
> this cleaner?
>
> [0]
> https://github.com/mozilla/gcp-ingestion/blob/master/ingestion-beam/src/main/java/com/mozilla/telemetry/decoder/GeoCityLookup.java#L95
>