1818 */
1919package org .elasticsearch .grok ;
2020
21+ import org .joni .Matcher ;
22+
2123import java .util .Map ;
2224import java .util .concurrent .ConcurrentHashMap ;
2325import java .util .concurrent .atomic .AtomicBoolean ;
2729
2830/**
2931 * Protects against long running operations that happen between the register and unregister invocations.
30- * Threads that invoke {@link #register()}, but take too long to invoke the {@link #unregister()} method
32+ * Threads that invoke {@link #register(Matcher )}, but take too long to invoke the {@link #unregister(Matcher )} method
3133 * will be interrupted.
3234 *
3335 * This is needed for Joni's {@link org.joni.Matcher#search(int, int, int)} method, because
3436 * it can end up spinning endlessly if the regular expression is too complex. Joni has checks
3537 * that for every 30k iterations it checks if the current thread is interrupted and if so
3638 * returns {@link org.joni.Matcher#INTERRUPTED}.
3739 */
38- public interface ThreadWatchdog {
39-
40+ public interface MatcherWatchdog {
41+
4042 /**
41- * Registers the current thread and interrupts the current thread
42- * if the takes too long for this thread to invoke {@link #unregister()}.
43+ * Registers the current matcher and interrupts the this matcher
44+ * if the takes too long for this thread to invoke {@link #unregister(Matcher)}.
45+ *
46+ * @param matcher The matcher to register
4347 */
44- void register ();
45-
48+ void register (Matcher matcher );
49+
4650 /**
47- * @return The maximum allowed time in milliseconds for a thread to invoke {@link #unregister()}
48- * after {@link #register()} has been invoked before this ThreadWatchDog starts to interrupting that thread.
51+ * @return The maximum allowed time in milliseconds for a thread to invoke {@link #unregister(Matcher )}
52+ * after {@link #register(Matcher )} has been invoked before this ThreadWatchDog starts to interrupting that thread.
4953 */
5054 long maxExecutionTimeInMillis ();
51-
55+
5256 /**
53- * Unregisters the current thread and prevents it from being interrupted.
57+ * Unregisters the current matcher and prevents it from being interrupted.
58+ *
59+ * @param matcher The matcher to unregister
5460 */
55- void unregister ();
56-
61+ void unregister (Matcher matcher );
62+
5763 /**
58- * Returns an implementation that checks for each fixed interval if there are threads that have invoked {@link #register()}
59- * and not {@link #unregister()} and have been in this state for longer than the specified max execution interval and
64+ * Returns an implementation that checks for each fixed interval if there are threads that have invoked {@link #register(Matcher )}
65+ * and not {@link #unregister(Matcher )} and have been in this state for longer than the specified max execution interval and
6066 * then interrupts these threads.
6167 *
6268 * @param interval The fixed interval to check if there are threads to interrupt
6369 * @param maxExecutionTime The time a thread has the execute an operation.
6470 * @param relativeTimeSupplier A supplier that returns relative time
6571 * @param scheduler A scheduler that is able to execute a command for each fixed interval
6672 */
67- static ThreadWatchdog newInstance (long interval ,
73+ static MatcherWatchdog newInstance (long interval ,
6874 long maxExecutionTime ,
6975 LongSupplier relativeTimeSupplier ,
7076 BiConsumer <Long , Runnable > scheduler ) {
7177 return new Default (interval , maxExecutionTime , relativeTimeSupplier , scheduler );
7278 }
73-
79+
7480 /**
7581 * @return A noop implementation that does not interrupt threads and is useful for testing and pre-defined grok expressions.
7682 */
77- static ThreadWatchdog noop () {
83+ static MatcherWatchdog noop () {
7884 return Noop .INSTANCE ;
7985 }
80-
81- class Noop implements ThreadWatchdog {
82-
86+
87+ class Noop implements MatcherWatchdog {
88+
8389 private static final Noop INSTANCE = new Noop ();
84-
90+
8591 private Noop () {
8692 }
87-
93+
8894 @ Override
89- public void register () {
95+ public void register (Matcher matcher ) {
9096 }
91-
97+
9298 @ Override
9399 public long maxExecutionTimeInMillis () {
94100 return Long .MAX_VALUE ;
95101 }
96-
102+
97103 @ Override
98- public void unregister () {
104+ public void unregister (Matcher matcher ) {
99105 }
100106 }
101-
102- class Default implements ThreadWatchdog {
103-
107+
108+ class Default implements MatcherWatchdog {
109+
104110 private final long interval ;
105111 private final long maxExecutionTime ;
106112 private final LongSupplier relativeTimeSupplier ;
107113 private final BiConsumer <Long , Runnable > scheduler ;
108114 private final AtomicInteger registered = new AtomicInteger (0 );
109115 private final AtomicBoolean running = new AtomicBoolean (false );
110- final ConcurrentHashMap <Thread , Long > registry = new ConcurrentHashMap <>();
111-
116+ final ConcurrentHashMap <Matcher , Long > registry = new ConcurrentHashMap <>();
117+
112118 private Default (long interval ,
113119 long maxExecutionTime ,
114120 LongSupplier relativeTimeSupplier ,
@@ -118,30 +124,30 @@ private Default(long interval,
118124 this .relativeTimeSupplier = relativeTimeSupplier ;
119125 this .scheduler = scheduler ;
120126 }
121-
122- public void register () {
127+
128+ public void register (Matcher matcher ) {
123129 registered .getAndIncrement ();
124- Long previousValue = registry .put (Thread . currentThread () , relativeTimeSupplier .getAsLong ());
130+ Long previousValue = registry .put (matcher , relativeTimeSupplier .getAsLong ());
125131 if (running .compareAndSet (false , true ) == true ) {
126132 scheduler .accept (interval , this ::interruptLongRunningExecutions );
127133 }
128134 assert previousValue == null ;
129135 }
130-
136+
131137 @ Override
132138 public long maxExecutionTimeInMillis () {
133139 return maxExecutionTime ;
134140 }
135-
136- public void unregister () {
137- Long previousValue = registry .remove (Thread . currentThread () );
141+
142+ public void unregister (Matcher matcher ) {
143+ Long previousValue = registry .remove (matcher );
138144 registered .decrementAndGet ();
139145 assert previousValue != null ;
140146 }
141-
147+
142148 private void interruptLongRunningExecutions () {
143149 final long currentRelativeTime = relativeTimeSupplier .getAsLong ();
144- for (Map .Entry <Thread , Long > entry : registry .entrySet ()) {
150+ for (Map .Entry <Matcher , Long > entry : registry .entrySet ()) {
145151 if ((currentRelativeTime - entry .getValue ()) > maxExecutionTime ) {
146152 entry .getKey ().interrupt ();
147153 // not removing the entry here, this happens in the unregister() method.
@@ -153,7 +159,7 @@ private void interruptLongRunningExecutions() {
153159 running .set (false );
154160 }
155161 }
156-
162+
157163 }
158-
164+
159165}
0 commit comments