in_tail: Prevent IOHandler.on_notify() being called concurrently

[fujimotos]

This patch is a fix for a multi-threading bug in Fluentd core. Apparently, this bug
has been one of major causes of the long-standing testing failures on Travis CI.

Problem

Essentially, there is a concurrency bug in the implementation of in_tail.

While IOHandler (in_tail#L673) is not designed to be thread-safe, its method
on_notify can be called by multiple threads concurrently.

Here is how it occurs:

1. TailInput calls on_notify on detach/shutdown (main thread)
2. In the same time, StatWatcher triggers on_notify on filesystem events
   (eventloop thread)

In this situation, one thread can see the data retrieved (and being actively
processed) by another thread, which will result in duplication of data records.

Solution

This patch fixes the issue by introducing a mutex, which should prevent two
threads from entering into the critical session simultaneously.

[fujimotos]

Note on the relation to Travis CI failures. For example, see the following test failure:

/home/travis/build/fluent/fluentd/test/plugin/test_in_tail.rb:161:in `test_emit`
     158:       assert_equal({"message" => "test4"}, events[1][2])
     159:       assert(events[0][1].is_a?(Fluent::EventTime))
     160:       assert(events[1][1].is_a?(Fluent::EventTime))
  => 161:       assert_equal(1, d.emit_count)
     162:     end
     163: 
     164:     def test_emit_with_emit_unmatched_lines_true
<1> expected but was
<2>
Failure: test_emit[flat](TailInputTest::singleline)

(Also similar failure cases are: #4160.2, #4138.4, #4129.5 #4128.4, #4127.5)

This failure occurs because TailInput emits the same event repeatedly due to the
concurrency bug describe above ( so emit_count become "2", instead of "1")

This class of test failures will go away once this patch merged.

Also: You can reproduce this test failure in local environments simply by inserting
sleep 0.5 into in_tail#708-709 as follows:

unless @lines.empty?
   if @receive_lines.call(@lines)
      @watcher.pe.update_pos(io.pos - @fifo.bytesize)
      sleep 0.5
      @lines.clear
   else
      read_more = false
   end
end

[repeatedly]

Good catch! Thanks.