killnine commented on a change in pull request #9: [AMQNET-565] - .net standard conversion, project reorganization URL: https://github.com/apache/activemq-nms-openwire/pull/9#discussion_r304428170
########## File path: src/Threads/PooledTaskRunner.cs ########## @@ -1,193 +1,193 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -using System; -using System.Threading; - -namespace Apache.NMS.ActiveMQ.Threads -{ - class PooledTaskRunner : TaskRunner - { - private readonly int maxIterationsPerRun; - private readonly Task task; - private readonly Object runable = new Object(); - private bool queued; - private bool _shutdown; - private bool iterating; - private volatile System.Threading.Thread runningThread; - - public void Run(Object o) - { - PooledTaskRunner p = o as PooledTaskRunner; - p.runningThread = System.Threading.Thread.CurrentThread; - try - { - p.RunTask(); - } - finally - { - p.runningThread = null; - } - } - - public PooledTaskRunner(Task task, int maxIterationsPerRun) - { - this.maxIterationsPerRun = maxIterationsPerRun; - this.task = task; - this._shutdown = false; - this.iterating = false; - this.queued = true; - ThreadPool.QueueUserWorkItem(new WaitCallback(Run), this); - } - - /// <summary> - /// We Expect MANY wakeup calls on the same TaskRunner. - /// </summary> - public void Wakeup() - { - lock(runable) - { - // When we get in here, we make some assumptions of state: - // queued=false, iterating=false: wakeup() has not be called and - // therefore task is not executing. - // queued=true, iterating=false: wakeup() was called but, task - // execution has not started yet - // queued=false, iterating=true : wakeup() was called, which caused - // task execution to start. - // queued=true, iterating=true : wakeup() called after task - // execution was started. - - if(queued || _shutdown) - { - return; - } - - queued = true; - - // The runTask() method will do this for me once we are done - // iterating. - if(!iterating) - { - ThreadPool.QueueUserWorkItem(new WaitCallback(Run), this); - } - } - } - - /// <summary> - /// shut down the task - /// </summary> - /// <param name="timeout"></param> - public void Shutdown(TimeSpan timeout) - { - lock(runable) - { - _shutdown = true; - // the check on the thread is done - // because a call to iterate can result in - // shutDown() being called, which would wait forever - // waiting for iterating to finish - if(runningThread != System.Threading.Thread.CurrentThread) - { - if(iterating) - { - System.Threading.Thread.Sleep(timeout); - } - } - } - } - - public void ShutdownWithAbort(TimeSpan timeout) - { - lock(runable) - { - _shutdown = true; - - if(runningThread != System.Threading.Thread.CurrentThread) - { - if(iterating) - { - System.Threading.Thread.Sleep(timeout); - } - - if(iterating) - { - runningThread.Abort(); - } - } - } - } - - public void Shutdown() - { - Shutdown(new TimeSpan(Timeout.Infinite)); - } - - internal void RunTask() - { - lock(runable) - { - queued = false; - if(_shutdown) - { - iterating = false; - return; - } - iterating = true; - } - - // Don't synchronize while we are iterating so that - // multiple wakeup() calls can be executed concurrently. - bool done = false; - try - { - for(int i = 0; i < maxIterationsPerRun; i++) - { - if(!task.Iterate()) - { - done = true; - break; - } - } - } - finally - { - lock(runable) - { - iterating = false; - if(_shutdown) - { - queued = false; - } - else - { - // If we could not iterate all the items - // then we need to re-queue. - if(!done) - { - queued = true; - } - - if(queued) - { - ThreadPool.QueueUserWorkItem(new WaitCallback(Run), this); - } - } - } - } - } - } -} +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +using System; +using System.Threading; + +namespace Apache.NMS.ActiveMQ.Threads +{ + class PooledTaskRunner : TaskRunner + { + private readonly int maxIterationsPerRun; + private readonly Task task; + private readonly Object runable = new Object(); + private bool queued; + private bool _shutdown; + private bool iterating; + private volatile System.Threading.Thread runningThread; + + public void Run(Object o) + { + PooledTaskRunner p = o as PooledTaskRunner; + p.runningThread = System.Threading.Thread.CurrentThread; + try + { + p.RunTask(); + } + finally + { + p.runningThread = null; + } + } + + public PooledTaskRunner(Task task, int maxIterationsPerRun) + { + this.maxIterationsPerRun = maxIterationsPerRun; + this.task = task; + this._shutdown = false; + this.iterating = false; + this.queued = true; + ThreadPool.QueueUserWorkItem(new WaitCallback(Run), this); + } + + /// <summary> + /// We Expect MANY wakeup calls on the same TaskRunner. + /// </summary> + public void Wakeup() + { + lock(runable) + { + // When we get in here, we make some assumptions of state: + // queued=false, iterating=false: wakeup() has not be called and + // therefore task is not executing. + // queued=true, iterating=false: wakeup() was called but, task + // execution has not started yet + // queued=false, iterating=true : wakeup() was called, which caused + // task execution to start. + // queued=true, iterating=true : wakeup() called after task + // execution was started. + + if(queued || _shutdown) + { + return; + } + + queued = true; + + // The runTask() method will do this for me once we are done + // iterating. + if(!iterating) + { + ThreadPool.QueueUserWorkItem(new WaitCallback(Run), this); + } + } + } + + /// <summary> + /// shut down the task + /// </summary> + /// <param name="timeout"></param> + public void Shutdown(TimeSpan timeout) + { + lock(runable) + { + _shutdown = true; + // the check on the thread is done + // because a call to iterate can result in + // shutDown() being called, which would wait forever + // waiting for iterating to finish + if(runningThread != System.Threading.Thread.CurrentThread) + { + if(iterating) + { + System.Threading.Thread.Sleep(timeout); + } + } + } + } + + public void ShutdownWithAbort(TimeSpan timeout) + { + lock(runable) + { + _shutdown = true; + + if(runningThread != System.Threading.Thread.CurrentThread) + { + if(iterating) + { + System.Threading.Thread.Sleep(timeout); + } + + if(iterating) + { + runningThread.Abort(); Review comment: Good catch. Perhaps this is why we're getting more handles than we're expecting in the unit tests? ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected] With regards, Apache Git Services
