Mailing-List: contact dev-help@apex.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@apex.apache.org
MIME-Version: 1.0
In-Reply-To: <0420E79F-3673-409E-AA67-FB3B67DFC8AB@datatorrent.com>
References: <71c00c32-c6b9-ed22-853f-ac47f6f9ae3c@datatorrent.com>
 <CAG2CJAdQvvpzLfk9+WMobfWeNypu8vCMNTAb+CdUtw=4wvCj5w@mail.gmail.com>
 <CAJ484BisxV49mEoo9VUUr7ZVQDgOt1p4gMyR0Ap8nGS7JMV7cw@mail.gmail.com>
 <CA+QE8bej3dpVbhEL5+Er4cSu_kq-RSL4wBDWksC=E3QSN7Cv1A@mail.gmail.com>
 <CA+QE8bdg-BJKz2okLv5bN9oQ58epo4k+P58ZuzJp_PzYAK92aA@mail.gmail.com>
 <5BADFD13-FEB9-45C8-BF17-06EB35C30D94@datatorrent.com> <CAMaZnT4tObj-3y0qYUrjpM5mbydJ60gAP+cPw6LUoszWhnuWFA@mail.gmail.com>
 <74dfdbbf-191e-63b0-39c7-0ce004ca6d56@datatorrent.com> <CAMaZnT5JLHPmZT9nA=1JAN0biOyqEWoJbmpymWVotNAbr5W7AQ@mail.gmail.com>
 <06D81B67-240D-47B6-BBC3-CEBF61AE9A23@datatorrent.com> <1b807a39-e175-73d4-17c8-c42fa04f6702@datatorrent.com>
 <641367C2-B275-463A-889A-9C075A1A8916@datatorrent.com> <CAJ484BjpQ5Uzp=QhMFaXx1TbmhajdX0Ts4WoFWwwFGGrFWscEQ@mail.gmail.com>
 <0420E79F-3673-409E-AA67-FB3B67DFC8AB@datatorrent.com>
From: Sanjay Pujare <sanjay@datatorrent.com>
Date: Wed, 12 Oct 2016 11:07:13 -0700
Message-ID: <CALPtS-+-ZXiz6baTTc+K7ps3kyXaYt6eXNV6+8RaUhj-ZoUtKA@mail.gmail.com>
Subject: Re: can operators emit on a different from the operator itself thread?
To: dev@apex.apache.org
Content-Type: multipart/alternative; boundary=001a113d020ed6dcc8053eaedc6e
archived-at: Wed, 12 Oct 2016 18:07:28 -0000

--001a113d020ed6dcc8053eaedc6e
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: quoted-printable

A JIRA has been created for adding this thread affinity check
https://issues.apache.org/jira/browse/APEXCORE-510 . I have made this
enhancement in a branch
https://github.com/sanjaypujare/apex-core/tree/malhar-510.thread_affinity
and I have been benchmarking the performance with this change. I will be
publishing the results in the above JIRA where we can discuss them and
hopefully agree on merging this change.

On Thu, Aug 11, 2016 at 1:41 PM, Sanjay Pujare <sanjay@datatorrent.com>
wrote:

> You are right, I was subconsciously thinking about the THREAD_LOCAL case
> with a single container and a simple DAG and in that case Vlad=E2=80=99s =
assumption
> might not be valid but may be it is.
>
> On 8/11/16, 11:47 AM, "Munagala Ramanath" <ram@datatorrent.com> wrote:
>
>     If I understand Vlad correctly, what he is saying is that each operat=
or
>     saves currentThread in
>     its own setup() and checks it in its own output methods. The threads =
in
>     different operators are
>     running potentially on different nodes and/or processes and there wil=
l
> be
>     no connection between them.
>
>     Ram
>
>     On Thu, Aug 11, 2016 at 11:41 AM, Sanjay Pujare <
> sanjay@datatorrent.com>
>     wrote:
>
>     > Name check is expensive, agreed, but there isn=E2=80=99t anything e=
lse
> currently.
>     > Ideally the stram engine (considering that it is an engine providin=
g
>     > resources like threads etc) should use a ThreadFactory or a
> ThreadGroup to
>     > create operator threads so identification and adding functionality =
is
>     > easier.
>     >
>     > The idea of checking for the same thread between setup() and emit()
> won=E2=80=99t
>     > work because the emit() check will have to be in the Sink hierarchy
> and
>     > AFAIK a Sink object doesn=E2=80=99t have access to the correspondin=
g
> operator,
>     > right? Another more fundamental problem probably is that these
> threads
>     > don=E2=80=99t have to match. The emit() for any operator (or rather=
 a Sink
> related
>     > to an operator) is ultimately triggered by an emitTuple() on the
> topmost
>     > input operator in that path which happens in that input operator=E2=
=80=99s
> thread
>     > which doesn=E2=80=99t have to match the thread calling setup() in t=
he
> downstream
>     > operators, right?
>     >
>     >
>     > On 8/11/16, 10:59 AM, "Vlad Rozov" <v.rozov@datatorrent.com> wrote:
>     >
>     >     Name verification is too expensive, it will be sufficient to
> store
>     >     currentThread during setup() and verify that it is the same
> during
>     > emit.
>     >     Checks should be supported not only for DefaultOutputPort, so w=
e
> may
>     >     have it implemented in various Sinks.
>     >
>     >     Vlad
>     >
>     >     On 8/11/16 10:21, Sanjay Pujare wrote:
>     >     > Thinking more about this =E2=80=93 all of the =E2=80=9Coperat=
or=E2=80=9D threads are
> created
>     > by the Stram engine with appropriate names. So we can put checks in
> the
>     > DefaultOutputPort.emit() or in the various implementations of
> Sink.put()
>     > that the current-thread is one created by the Stram engine (by
> verifying
>     > the name).
>     >     >
>     >     > We can even use a special Thread object for operator threads
> so the
>     > above detection is easier.
>     >     >
>     >     >
>     >     >
>     >     > On 8/10/16, 6:11 PM, "Amol Kekre" <amol@datatorrent.com>
> wrote:
>     >     >
>     >     >      +1 on debug proposal. Even if tuples lands up within the
>     > window, it breaks
>     >     >      all guarantees. A rerun (after restart from a checkpoint=
)
> can
>     > have tuples
>     >     >      in different windows from this thread. A separate thread
> simply
>     > exposes
>     >     >      users to unwarranted risks.
>     >     >
>     >     >      Thks
>     >     >      Amol
>     >     >
>     >     >
>     >     >      On Wed, Aug 10, 2016 at 6:05 PM, Vlad Rozov <
>     > v.rozov@datatorrent.com> wrote:
>     >     >
>     >     >      > Tuples emitted between end and begin windows is only
> one of
>     > possible
>     >     >      > behaviors that emitting tuples on a separate from the
>     > operator thread may
>     >     >      > introduce. It will be good to have both checks in plac=
e
> at
>     > run-time and if
>     >     >      > checking for the operator thread for every emitted
> tuple is
>     > too expensive,
>     >     >      > we may have it enabled only in DEBUG or mode with more
> checks
>     > in place.
>     >     >      >
>     >     >      > Vlad
>     >     >      >
>     >     >      >
>     >     >      > Sanjay just reminded me of my typo -> I meant between
>     > end_window and
>     >     >      >> start_window :)
>     >     >      >>
>     >     >      >> Thks
>     >     >      >> Amol
>     >     >      >>
>     >     >      >> On Wed, Aug 10, 2016 at 2:36 PM, Sanjay Pujare <
>     > sanjay@datatorrent.com>
>     >     >      >> wrote:
>     >     >      >>
>     >     >      >> If the goal is to do this validation through static
> analysis
>     > of operator
>     >     >      >>> code, I guess it is possible but is going to be
>     > non-trivial. And there
>     >     >      >>> could be false positives and false negatives.
>     >     >      >>>
>     >     >      >>> Also I suppose this discussion applies to processor
>     > operators (those
>     >     >      >>> having both in and out ports) so Ram=E2=80=99s examp=
le of
>     > JdbcPollInputOperator
>     >     >      >>> may
>     >     >      >>> not be applicable here?
>     >     >      >>>
>     >     >      >>> On 8/10/16, 2:04 PM, "Ashwin Chandra Putta" <
>     > ashwinchandrap@gmail.com>
>     >     >      >>> wrote:
>     >     >      >>>
>     >     >      >>>      In a separate thread I mean.
>     >     >      >>>
>     >     >      >>>      Regards,
>     >     >      >>>      Ashwin.
>     >     >      >>>
>     >     >      >>>      On Wed, Aug 10, 2016 at 2:01 PM, Ashwin Chandra
> Putta <
>     >     >      >>>      ashwinchandrap@gmail.com> wrote:
>     >     >      >>>
>     >     >      >>>      > + dev@apex.apache.org
>     >     >      >>>      > - users@apex.apache.org
>     >     >      >>>      >
>     >     >      >>>      > This is one of those best practices that we
> learn by
>     > experience
>     >     >      >>> during
>     >     >      >>>      > operator development. It will save a lot of
> time
>     > during operator
>     >     >      >>>      > development if we can catch and throw
> validation
>     > error when
>     >     >      >>> someone
>     >     >      >>> emits
>     >     >      >>>      > tuples in a non separate thread.
>     >     >      >>>      >
>     >     >      >>>      > Regards,
>     >     >      >>>      > Ashwin
>     >     >      >>>      >
>     >     >      >>>      > On Wed, Aug 10, 2016 at 1:57 PM, Munagala
> Ramanath <
>     >     >      >>> ram@datatorrent.com>
>     >     >      >>>      > wrote:
>     >     >      >>>      >
>     >     >      >>>      >> For cases where use of a different thread is
>     > needed, it can write
>     >     >      >>> tuples
>     >     >      >>>      >> to a queue from where the operator thread
> pulls
>     > them --
>     >     >      >>>      >> JdbcPollInputOperator in Malhar has an
> example.
>     >     >      >>>      >>
>     >     >      >>>      >> Ram
>     >     >      >>>      >>
>     >     >      >>>      >> On Wed, Aug 10, 2016 at 1:50 PM,
> hsy541@gmail.com <
>     >     >      >>> hsy541@gmail.com
>     >     >      >>>      >> wrote:
>     >     >      >>>      >>
>     >     >      >>>      >>> Hey Vlad,
>     >     >      >>>      >>>
>     >     >      >>>      >>> Thanks for bringing this up. Is there an
> easy way
>     > to detect
>     >     >      >>> unexpected
>     >     >      >>>      >>> use of emit method without hurt the
> performance.
>     > Or at least if
>     >     >      >>> we
>     >     >      >>> can
>     >     >      >>>      >>> detect this in debug mode.
>     >     >      >>>      >>>
>     >     >      >>>      >>> Regards,
>     >     >      >>>      >>> Siyuan
>     >     >      >>>      >>>
>     >     >      >>>      >>> On Wed, Aug 10, 2016 at 11:27 AM, Vlad Rozo=
v
> <
>     >     >      >>> v.rozov@datatorrent.com>
>     >     >      >>>      >>> wrote:
>     >     >      >>>      >>>
>     >     >      >>>      >>>> The short answer is no, creating worker
> thread to
>     > emit tuples
>     >     >      >>> is
>     >     >      >>> not
>     >     >      >>>      >>>> supported by Apex and will lead to an
> undefined
>     > behavior.
>     >     >      >>> Operators in Apex
>     >     >      >>>      >>>> have strong thread affinity and all
> interaction
>     > with the
>     >     >      >>> platform
>     >     >      >>> must
>     >     >      >>>      >>>> happen on the operator thread.
>     >     >      >>>      >>>>
>     >     >      >>>      >>>> Vlad
>     >     >      >>>      >>>>
>     >     >      >>>      >>>
>     >     >      >>>      >>>
>     >     >      >>>      >>
>     >     >      >>>      >
>     >     >      >>>      >
>     >     >      >>>      > --
>     >     >      >>>      >
>     >     >      >>>      > Regards,
>     >     >      >>>      > Ashwin.
>     >     >      >>>      >
>     >     >      >>>
>     >     >      >>>
>     >     >      >>>
>     >     >      >>>      --
>     >     >      >>>
>     >     >      >>>      Regards,
>     >     >      >>>      Ashwin.
>     >     >      >>>
>     >     >      >>>
>     >     >      >>>
>     >     >      >>>
>     >     >      >>>
>     >     >      >
>     >     >
>     >     >
>     >     >
>     >
>     >
>     >
>     >
>     >
>
>
>

--001a113d020ed6dcc8053eaedc6e--