Question on DrainDispatcher

[canan chen <cc...@gmail.com>]

Hi folks,

I doubt that DrainDispatcher may have race condition.

Here's the sequence
1. eventQueue is empty and EventHandling Thread check the eventQueue and
set drained as true
2. new event is added to eventQueue, but has not been handled yet.
3. DrainDispatcher.await() is called, but at this time drained is true, so
await will exit at once. But actually there's still event not yet handled
in step 2.

Is the above sequence possible ? Do I miss something ? Thanks


 while (!stopped && !Thread.currentThread().isInterrupted()) {

          drained = eventQueue.isEmpty();

          // blockNewEvents is only set when dispatcher is draining to stop,

          // adding this check is to avoid the overhead of acquiring the
lock

          // and calling notify every time in the normal run of the loop.

          if (blockNewEvents) {

            synchronized (waitForDrained) {

              if (drained) {

                waitForDrained.notify();

              }

            }

          }

          Event event;

          try {

            event = eventQueue.take();

          } catch(InterruptedException ie) {

            if (!stopped) {

              LOG.warn("AsyncDispatcher thread interrupted", ie);

            }

            return;

          }

          if (event != null) {

            dispatch(event);

          }

        }

Re: Question on DrainDispatcher

[Vinod Kumar Vavilapalli <vi...@hortonworks.com>]

By definition, adding a new event == calling GenericEventHandler.handle() which sets drained to be false and adds the event to a queue. So the scenario you outlined shouldn't happen.

Thanks,
+Vinod

On Mar 12, 2015, at 1:20 AM, canan chen <cc...@gmail.com> wrote:

> Hi folks,
> 
> I doubt that DrainDispatcher may have race condition.
> 
> Here's the sequence
> 1. eventQueue is empty and EventHandling Thread check the eventQueue and
> set drained as true
> 2. new event is added to eventQueue, but has not been handled yet.
> 3. DrainDispatcher.await() is called, but at this time drained is true, so
> await will exit at once. But actually there's still event not yet handled
> in step 2.
> 
> Is the above sequence possible ? Do I miss something ? Thanks
> 
> 
> while (!stopped && !Thread.currentThread().isInterrupted()) {
> 
>          drained = eventQueue.isEmpty();
> 
>          // blockNewEvents is only set when dispatcher is draining to stop,
> 
>          // adding this check is to avoid the overhead of acquiring the
> lock
> 
>          // and calling notify every time in the normal run of the loop.
> 
>          if (blockNewEvents) {
> 
>            synchronized (waitForDrained) {
> 
>              if (drained) {
> 
>                waitForDrained.notify();
> 
>              }
> 
>            }
> 
>          }
> 
>          Event event;
> 
>          try {
> 
>            event = eventQueue.take();
> 
>          } catch(InterruptedException ie) {
> 
>            if (!stopped) {
> 
>              LOG.warn("AsyncDispatcher thread interrupted", ie);
> 
>            }
> 
>            return;
> 
>          }
> 
>          if (event != null) {
> 
>            dispatch(event);
> 
>          }
> 
>        }