Mailing-List: contact user-help@flink.apache.org; run by ezmlm
Precedence: bulk
From: Nico Kruber <nico@data-artisans.com>
To: Chao Wang <chaowang@wustl.edu>
Cc: user@flink.apache.org
Subject: Re: Flink multithreading, CoFlatMapFunction, CoProcessFunction, internal state
Date: Mon, 21 Aug 2017 16:35:38 +0200
Message-ID: <1771535.rGALQnymKj@nico-work>
Organization: data Artisans
In-Reply-To: <38baca6e-5ba9-1e29-0ae5-07dd04c85332@wustl.edu>
References: <ca8d0d4a-ada7-7fdb-efbd-335b75b4fb2f@wustl.edu> <16518078.snpQ2NUW06@nico-work> <38baca6e-5ba9-1e29-0ae5-07dd04c85332@wustl.edu>
MIME-Version: 1.0
Content-Type: multipart/signed; boundary="nextPart5509606.eGDR5tQT6E"; micalg="pgp-sha1"; protocol="application/pgp-signature"
archived-at: Mon, 21 Aug 2017 14:35:51 -0000

--nextPart5509606.eGDR5tQT6E
Content-Transfer-Encoding: 7Bit
Content-Type: text/plain; charset="us-ascii"

Hi Chao,
what I meant by "per-record base" was actually supposed to be "per-event base" 
(event = one entity of whatever the stream contains). As from the API: 
processing is supposed to be one event at a time and this is what is performed 
internally, too.

Nico

On Thursday, 17 August 2017 05:06:07 CEST Chao Wang wrote:
> Thank you! Nico. That helps me a lot!
> 
> 2a) That really clarifies my understanding about Flink. Yes, I think I
> have used static references, since I invoked a native function
> (implemented through JNI) which I believe only has one instance per
> process. And I guess the reason why those Java synchronization
> mechanisms were in vain is because of separate function objects at
> runtime, which results in separate lock objects. Now I use c++ mutex
> within the native function and it resolves my case.
> 
> BTW, could you elaborate a bit more about what do you mean by
> "per-record base"? what do you mean by a record?
> 
> 3) I do not intend to store the CoProcessFunction.Context. I was just
> wondering that since the document said it is only valid during the
> invocation, for maintaining custom states of my program logic I guess I
> cannot use it.
> 
> 
> Thank you,
> Chao
> 
> On 08/16/2017 03:31 AM, Nico Kruber wrote:
> > Hi Chao,
> > 
> > 1) regarding the difference of CoFlatMapFunction/CoProcessFunction, let me
> > quote the javadoc of the CoProcessFunction:
> > 
> > "Contrary to the {@link CoFlatMapFunction}, this function can also query
> > the time (both event and processing) and set timers, through the provided
> > {@link Context}. When reacting to the firing of set timers the function
> > can emit yet more elements."
> > 
> > So, imho, both deliver a different level of abstraction and control (high-
> > vs. low-level). Also note the different methods available for you to
> > implement.
> > 
> > 2a) In general, Flink calls functions on a per-record base in a serialized
> > fashion per task. For each task at a TaskManager, in case of it having
> > multiple slots, separate function objects are used where you should only
> > get in trouble if you share static references. Otherwise you do not need
> > to worry about thread-safety.
> > 
> > 2b) From what I see in the code (StreamTwoInputProcessor), the same should
> > apply to CoFlatMapFunction and CoProcessFunction so that calls to
> > flatMap1/2 and processElement1/2 are not called in parallel!
> > 
> > 3) why would you want to store the CoProcessFunction.Context?
> > 
> > 
> > Nico
> > 
> > On Monday, 14 August 2017 18:36:38 CEST Chao Wang wrote:
> >> Hi,
> >> 
> >> I'd like to know if CoFlatMapFunction/CoProcessFunction is thread-safe,
> >> and to what extent? What's the difference between the two Functions? and
> >> in general, how does Flink prevent race conditions? Here's my case:
> >> 
> >> I tried to condition on two input streams and produce the third stream
> >> if the condition is met. I implemented CoFlatMapFunction and tried to
> >> monitor a state using a field in the implemented class (I want to
> >> isolate my application from the checkpointing feature, and therefore I
> >> do not use the states as documented here
> >> https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/stream/st
> >> ate .html). The field served as a flag indicating whether there are some
> >> pending data from either input stream, and if yes, processing it along
> >> with the arriving data from the other input stream (the processing
> >> invokes a native function).
> >> 
> >> But then I got double free error and segmentation fault, which I believe
> >> was due to unintentional concurrent access to the native function. Then
> >> I tried to wrap the access into a synchronized method, as well as
> >> explicitly lock/unlock the flatMap1/flatMap2 methods, but all in vain
> >> and the error remained.
> >> 
> >> I considered using CoProcessFunction in my case, but seems to me that it
> >> does not handle customary internal states, stating in the javadoc "The
> >> context [CoProcessFunction.Context] is only valid during the invocation
> >> of this method, do not store it."
> >> 
> >> 
> >> 
> >> Thanks,
> >> Chao


--nextPart5509606.eGDR5tQT6E
Content-Type: application/pgp-signature; name="signature.asc"
Content-Description: This is a digitally signed message part.
Content-Transfer-Encoding: 7Bit

-----BEGIN PGP SIGNATURE-----

iF0EABECAB0WIQTIh4KsbsNd3l7wd+cg8nJL2uqeWQUCWZrvugAKCRAg8nJL2uqe
WfvLAJ0YoJWNx4a6CyI3bd/cVp0VUv7L2ACfWxqvk0QjyF33qtiBWepamPILCXQ=
=nFCD
-----END PGP SIGNATURE-----

--nextPart5509606.eGDR5tQT6E--