Mailing-List: contact commits-help@nifi.incubator.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@nifi.incubator.apache.org
Date: Wed, 10 Dec 2014 16:54:12 +0000 (UTC)
From: "Mark Payne (JIRA)" <jira@apache.org>
To: commits@nifi.incubator.apache.org
Message-ID: <JIRA.12759824.1417809822000.121273.1418230452362@Atlassian.JIRA>
In-Reply-To: <JIRA.12759824.1417809822000@Atlassian.JIRA>
References: <JIRA.12759824.1417809822000@Atlassian.JIRA>
 <JIRA.12759824.1417809822419@arcas>
Subject: [jira] [Updated] (NIFI-54) Event Driven Processors can be scheduled
 to use more threads than Max Concurrent Threads allows
MIME-Version: 1.0
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: 7bit


     [ https://issues.apache.org/jira/browse/NIFI-54?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Mark Payne updated NIFI-54:
---------------------------
    Description: 
EventDrivenSchedulingAgent, when incrementing schedule state's active thread count should check the result and if it's greater than what is allowed should just decrement active thread count and return:

This must be done for trigger of Connectable and ProcessorNode, and ScheduleState must return the new value, rather than being void.

private void trigger(final Connectable worker, final ScheduleState scheduleState, final ConnectableProcessContext processContext, final ProcessSessionFactory sessionFactory) {

final int newThreadCount = scheduleState.incremenActiveThreadCount();
if (newThreadCount > worker.getMaxConcurrentTasks() && worker.getMaxConcurrentTasks() > 0) {

// its possible that the worker queue could give us a worker node that is eligible to run based
// on the number of threads but another thread has already incremented the thread count, result in
// reaching the maximum number of threads. we won't know this until we atomically increment the thread count 
// on the Schedule State, so we check it here. in this case, we cannot trigger the Processor, as doing so would
// result in using more than the maximum number of defined threads

scheduleState.decrementActiveThreadCount();
return;

}

...

  was:
EventDrivenSchedulingAgent, when incrementing schedule state's active thread count should check the result and if it's greater then what is allowed should just decrement active thread count and return:

This must be done for trigger of Connectable and ProcessorNode, and ScheduleState must return the new value, rather than being void.

private void trigger(final Connectable worker, final ScheduleState scheduleState, final ConnectableProcessContext processContext, final ProcessSessionFactory sessionFactory) {

final int newThreadCount = scheduleState.incremenActiveThreadCount();
if (newThreadCount > worker.getMaxConcurrentTasks() && worker.getMaxConcurrentTasks() > 0) {

// its possible that the worker queue could give us a worker node that is eligible to run based
// on the number of threads but another thread has already incremented the thread count, result in
// reaching the maximum number of threads. we won't know this until we atomically increment the thread count 
// on the Schedule State, so we check it here. in this case, we cannot trigger the Processor, as doing so would
// result in using more than the maximum number of defined threads

scheduleState.decrementActiveThreadCount();
return;

}

...


> Event Driven Processors can be scheduled to use more threads than Max Concurrent Threads allows
> -----------------------------------------------------------------------------------------------
>
>                 Key: NIFI-54
>                 URL: https://issues.apache.org/jira/browse/NIFI-54
>             Project: Apache NiFi
>          Issue Type: Bug
>          Components: Core Framework
>            Reporter: Matt Gilman
>             Fix For: 0.0.1
>
>
> EventDrivenSchedulingAgent, when incrementing schedule state's active thread count should check the result and if it's greater than what is allowed should just decrement active thread count and return:
> This must be done for trigger of Connectable and ProcessorNode, and ScheduleState must return the new value, rather than being void.
> private void trigger(final Connectable worker, final ScheduleState scheduleState, final ConnectableProcessContext processContext, final ProcessSessionFactory sessionFactory) {
> final int newThreadCount = scheduleState.incremenActiveThreadCount();
> if (newThreadCount > worker.getMaxConcurrentTasks() && worker.getMaxConcurrentTasks() > 0) {
> // its possible that the worker queue could give us a worker node that is eligible to run based
> // on the number of threads but another thread has already incremented the thread count, result in
> // reaching the maximum number of threads. we won't know this until we atomically increment the thread count 
> // on the Schedule State, so we check it here. in this case, we cannot trigger the Processor, as doing so would
> // result in using more than the maximum number of defined threads
> scheduleState.decrementActiveThreadCount();
> return;
> }
> ...


--
This message was sent by Atlassian JIRA
(v6.3.4#6332)