Thanks Bozzo, that sorts those failures (so long as the issue from your other change is worked around, to let the build get that far).
Robbie On 9 July 2015 at 11:22, Bozo Dragojevic <bo...@digiverse.si> wrote: > Hi Robbie, > > Yeah, my bad. I was sitting on some local changes so I missed this. > I'll test on a clean checkout next time. Sorry for all the mess. > I've commited the missing methods to proton-j > > Bozzo > > On 9. 07. 15 12.14, Robbie Gemmell wrote: >> Hi Bozzo, >> >> This change also seems to be causing test failures when using the >> maven build (if you update things to get past the earlier failures, >> caused by the commit mentioned in the other thread on proton@): >> >> proton_tests.reactor.ExceptionTest.test_schedule_cancel ................. >> fail >> Error during test: Traceback (most recent call last): >> File "/home/gemmellr/workspace/proton/tests/python/proton-test", >> line 360, in run >> phase() >> File >> "/home/gemmellr/workspace/proton/tests/python/proton_tests/reactor.py", >> line 181, in test_schedule_cancel >> now = self.reactor.mark() >> File >> "/home/gemmellr/workspace/proton/tests/../proton-c/bindings/python/proton/reactor.py", >> line 118, in mark >> return pn_reactor_mark(self._impl) >> NameError: global name 'pn_reactor_mark' is not defined >> >> >> Robbie >> >> On 9 July 2015 at 10:55, Bozo Dragojevic <bo...@digiverse.si> wrote: >>> Hi Ken, >>> >>> I've installed python3.4 and tox and friends and tried to reproduce it here >>> and I can confirm that some completely unrelated test fail mysteriously >>> with python3.4 and that reverting my change makes that failure go away :) >>> >>> I've added more task cancellation tests to force the error while still >>> in the implicated code, as I suspected it could be some refcounting >>> problem but the new tests do not show anything unusual. >>> >>> What is even weirder, with the new tests even the python3.4 suite passes >>> without segfault! >>> >>> So, I consider this a false positive and have left the change in, >>> including the new tests at ca47d72. >>> >>> Does such solution work for you? >>> >>> Bozzo >>> >>> On 8. 07. 15 16.32, Ken Giusti wrote: >>>> Hi Bozzo, >>>> >>>> Can you please revert this change? >>>> >>>> It is causing a segfault in the python unit tests when they are run under >>>> python3.4. >>>> >>>> I haven't hit the segfault on python2.7, only on python3.4 >>>> >>>> thanks, >>>> >>>> -K >>>> >>>> ----- Original Message ----- >>>>> From: bo...@apache.org >>>>> To: comm...@qpid.apache.org >>>>> Sent: Tuesday, July 7, 2015 3:50:16 PM >>>>> Subject: [2/2] qpid-proton git commit: PROTON-928: cancellable tasks >>>>> >>>>> PROTON-928: cancellable tasks >>>>> >>>>> A scheduled task can be cancelled. >>>>> A cancelled task does not prevent reactor from stopping running >>>>> >>>>> >>>>> Project: http://git-wip-us.apache.org/repos/asf/qpid-proton/repo >>>>> Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton/commit/d4d22ee3 >>>>> Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton/tree/d4d22ee3 >>>>> Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton/diff/d4d22ee3 >>>>> >>>>> Branch: refs/heads/master >>>>> Commit: d4d22ee396163babcac19c48845b1f10ca3b5a48 >>>>> Parents: 09af375 >>>>> Author: Bozo Dragojevic <bo...@digiverse.si> >>>>> Authored: Tue Jul 7 10:17:40 2015 +0200 >>>>> Committer: Bozo Dragojevic <bo...@digiverse.si> >>>>> Committed: Tue Jul 7 21:49:44 2015 +0200 >>>>> >>>>> ---------------------------------------------------------------------- >>>>> proton-c/bindings/python/proton/reactor.py | 5 +++- >>>>> proton-c/include/proton/reactor.h | 1 + >>>>> proton-c/src/reactor/timer.c | 25 +++++++++++++++++++- >>>>> proton-c/src/tests/reactor.c | 15 ++++++++++++ >>>>> .../org/apache/qpid/proton/reactor/Task.java | 4 ++++ >>>>> .../qpid/proton/reactor/impl/TaskImpl.java | 10 ++++++++ >>>>> .../apache/qpid/proton/reactor/impl/Timer.java | 19 ++++++++++++--- >>>>> proton-j/src/main/resources/creactor.py | 3 +++ >>>>> tests/python/proton_tests/reactor.py | 14 +++++++++++ >>>>> 9 files changed, 91 insertions(+), 5 deletions(-) >>>>> ---------------------------------------------------------------------- >>>>> >>>>> >>>>> http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/d4d22ee3/proton-c/bindings/python/proton/reactor.py >>>>> ---------------------------------------------------------------------- >>>>> diff --git a/proton-c/bindings/python/proton/reactor.py >>>>> b/proton-c/bindings/python/proton/reactor.py >>>>> index c66334b..d019554 100644 >>>>> --- a/proton-c/bindings/python/proton/reactor.py >>>>> +++ b/proton-c/bindings/python/proton/reactor.py >>>>> @@ -53,6 +53,9 @@ class Task(Wrapper): >>>>> def _init(self): >>>>> pass >>>>> >>>>> + def cancel(self): >>>>> + pn_task_cancel(self._impl) >>>>> + >>>>> class Acceptor(Wrapper): >>>>> >>>>> def __init__(self, impl): >>>>> @@ -112,7 +115,7 @@ class Reactor(Wrapper): >>>>> pn_reactor_yield(self._impl) >>>>> >>>>> def mark(self): >>>>> - pn_reactor_mark(self._impl) >>>>> + return pn_reactor_mark(self._impl) >>>>> >>>>> def _get_handler(self): >>>>> return WrappedHandler.wrap(pn_reactor_get_handler(self._impl), >>>>> self.on_error) >>>>> >>>>> http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/d4d22ee3/proton-c/include/proton/reactor.h >>>>> ---------------------------------------------------------------------- >>>>> diff --git a/proton-c/include/proton/reactor.h >>>>> b/proton-c/include/proton/reactor.h >>>>> index 59b2282..6f52d22 100644 >>>>> --- a/proton-c/include/proton/reactor.h >>>>> +++ b/proton-c/include/proton/reactor.h >>>>> @@ -96,6 +96,7 @@ PN_EXTERN pn_task_t *pn_timer_schedule(pn_timer_t >>>>> *timer, >>>>> pn_timestamp_t deadlin >>>>> PN_EXTERN int pn_timer_tasks(pn_timer_t *timer); >>>>> >>>>> PN_EXTERN pn_record_t *pn_task_attachments(pn_task_t *task); >>>>> +PN_EXTERN void pn_task_cancel(pn_task_t *task); >>>>> >>>>> PN_EXTERN pn_reactor_t *pn_class_reactor(const pn_class_t *clazz, void >>>>> *object); >>>>> PN_EXTERN pn_reactor_t *pn_object_reactor(void *object); >>>>> >>>>> http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/d4d22ee3/proton-c/src/reactor/timer.c >>>>> ---------------------------------------------------------------------- >>>>> diff --git a/proton-c/src/reactor/timer.c b/proton-c/src/reactor/timer.c >>>>> index 1ad0821..61efd31 100644 >>>>> --- a/proton-c/src/reactor/timer.c >>>>> +++ b/proton-c/src/reactor/timer.c >>>>> @@ -27,12 +27,14 @@ struct pn_task_t { >>>>> pn_list_t *pool; >>>>> pn_record_t *attachments; >>>>> pn_timestamp_t deadline; >>>>> + bool cancelled; >>>>> }; >>>>> >>>>> void pn_task_initialize(pn_task_t *task) { >>>>> task->pool = NULL; >>>>> task->attachments = pn_record(); >>>>> task->deadline = 0; >>>>> + task->cancelled = false; >>>>> } >>>>> >>>>> void pn_task_finalize(pn_task_t *task) { >>>>> @@ -68,6 +70,11 @@ pn_record_t *pn_task_attachments(pn_task_t *task) { >>>>> return task->attachments; >>>>> } >>>>> >>>>> +void pn_task_cancel(pn_task_t *task) { >>>>> + assert(task); >>>>> + task->cancelled = true; >>>>> +} >>>>> + >>>>> // >>>>> // timer >>>>> // >>>>> @@ -113,8 +120,22 @@ pn_task_t *pn_timer_schedule(pn_timer_t *timer, >>>>> pn_timestamp_t deadline) { >>>>> return task; >>>>> } >>>>> >>>>> +void pni_timer_flush_cancelled(pn_timer_t *timer) { >>>>> + while (pn_list_size(timer->tasks)) { >>>>> + pn_task_t *task = (pn_task_t *) pn_list_get(timer->tasks, 0); >>>>> + if (task->cancelled) { >>>>> + pn_task_t *min = (pn_task_t *) pn_list_minpop(timer->tasks); >>>>> + assert(min == task); >>>>> + pn_decref(min); >>>>> + } else { >>>>> + break; >>>>> + } >>>>> + } >>>>> +} >>>>> + >>>>> pn_timestamp_t pn_timer_deadline(pn_timer_t *timer) { >>>>> assert(timer); >>>>> + pni_timer_flush_cancelled(timer); >>>>> if (pn_list_size(timer->tasks)) { >>>>> pn_task_t *task = (pn_task_t *) pn_list_get(timer->tasks, 0); >>>>> return task->deadline; >>>>> @@ -130,7 +151,8 @@ void pn_timer_tick(pn_timer_t *timer, pn_timestamp_t >>>>> now) >>>>> { >>>>> if (now >= task->deadline) { >>>>> pn_task_t *min = (pn_task_t *) pn_list_minpop(timer->tasks); >>>>> assert(min == task); >>>>> - pn_collector_put(timer->collector, PN_OBJECT, min, PN_TIMER_TASK); >>>>> + if (!min->cancelled) >>>>> + pn_collector_put(timer->collector, PN_OBJECT, min, >>>>> PN_TIMER_TASK); >>>>> pn_decref(min); >>>>> } else { >>>>> break; >>>>> @@ -140,5 +162,6 @@ void pn_timer_tick(pn_timer_t *timer, pn_timestamp_t >>>>> now) >>>>> { >>>>> >>>>> int pn_timer_tasks(pn_timer_t *timer) { >>>>> assert(timer); >>>>> + pni_timer_flush_cancelled(timer); >>>>> return pn_list_size(timer->tasks); >>>>> } >>>>> >>>>> http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/d4d22ee3/proton-c/src/tests/reactor.c >>>>> ---------------------------------------------------------------------- >>>>> diff --git a/proton-c/src/tests/reactor.c b/proton-c/src/tests/reactor.c >>>>> index fe6c769..059d099 100644 >>>>> --- a/proton-c/src/tests/reactor.c >>>>> +++ b/proton-c/src/tests/reactor.c >>>>> @@ -440,6 +440,20 @@ static void test_reactor_schedule_handler(void) { >>>>> pn_free(tevents); >>>>> } >>>>> >>>>> +static void test_reactor_schedule_cancel(void) { >>>>> + pn_reactor_t *reactor = pn_reactor(); >>>>> + pn_handler_t *root = pn_reactor_get_handler(reactor); >>>>> + pn_list_t *events = pn_list(PN_VOID, 0); >>>>> + pn_handler_add(root, test_handler(reactor, events)); >>>>> + pn_task_t *task = pn_reactor_schedule(reactor, 0, NULL); >>>>> + pn_task_cancel(task); >>>>> + pn_reactor_run(reactor); >>>>> + pn_reactor_free(reactor); >>>>> + expect(events, PN_REACTOR_INIT, PN_SELECTABLE_INIT, >>>>> PN_SELECTABLE_UPDATED, >>>>> + PN_SELECTABLE_FINAL, PN_REACTOR_FINAL, END); >>>>> + pn_free(events); >>>>> +} >>>>> + >>>>> int main(int argc, char **argv) >>>>> { >>>>> test_reactor(); >>>>> @@ -461,5 +475,6 @@ int main(int argc, char **argv) >>>>> test_reactor_transfer(4*1024, 1024); >>>>> test_reactor_schedule(); >>>>> test_reactor_schedule_handler(); >>>>> + test_reactor_schedule_cancel(); >>>>> return 0; >>>>> } >>>>> >>>>> http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/d4d22ee3/proton-j/src/main/java/org/apache/qpid/proton/reactor/Task.java >>>>> ---------------------------------------------------------------------- >>>>> diff --git >>>>> a/proton-j/src/main/java/org/apache/qpid/proton/reactor/Task.java >>>>> b/proton-j/src/main/java/org/apache/qpid/proton/reactor/Task.java >>>>> index 69701ab..7fb5964 100644 >>>>> --- a/proton-j/src/main/java/org/apache/qpid/proton/reactor/Task.java >>>>> +++ b/proton-j/src/main/java/org/apache/qpid/proton/reactor/Task.java >>>>> @@ -43,4 +43,8 @@ public interface Task extends Extendable { >>>>> /** @return the reactor that created this task. */ >>>>> Reactor getReactor(); >>>>> >>>>> + /** >>>>> + * Cancel the execution of this task. No-op if invoked after the task >>>>> was already executed. >>>>> + */ >>>>> + void cancel(); >>>>> } >>>>> >>>>> http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/d4d22ee3/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/TaskImpl.java >>>>> ---------------------------------------------------------------------- >>>>> diff --git >>>>> a/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/TaskImpl.java >>>>> b/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/TaskImpl.java >>>>> index 00c9a84..11bb6b8 100644 >>>>> --- >>>>> a/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/TaskImpl.java >>>>> +++ >>>>> b/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/TaskImpl.java >>>>> @@ -31,6 +31,7 @@ import org.apache.qpid.proton.reactor.Task; >>>>> public class TaskImpl implements Task, Comparable<TaskImpl> { >>>>> private final long deadline; >>>>> private final int counter; >>>>> + private boolean cancelled = false; >>>>> private final AtomicInteger count = new AtomicInteger(); >>>>> private Record attachments = new RecordImpl(); >>>>> private Reactor reactor; >>>>> @@ -58,6 +59,15 @@ public class TaskImpl implements Task, >>>>> Comparable<TaskImpl> { >>>>> return deadline; >>>>> } >>>>> >>>>> + public boolean isCancelled() { >>>>> + return cancelled; >>>>> + } >>>>> + >>>>> + @Override >>>>> + public void cancel() { >>>>> + cancelled = true; >>>>> + } >>>>> + >>>>> public void setReactor(Reactor reactor) { >>>>> this.reactor = reactor; >>>>> } >>>>> >>>>> http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/d4d22ee3/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/Timer.java >>>>> ---------------------------------------------------------------------- >>>>> diff --git >>>>> a/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/Timer.java >>>>> b/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/Timer.java >>>>> index 32bb4f6..b8df19d 100644 >>>>> --- >>>>> a/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/Timer.java >>>>> +++ >>>>> b/proton-j/src/main/java/org/apache/qpid/proton/reactor/impl/Timer.java >>>>> @@ -31,7 +31,7 @@ import org.apache.qpid.proton.reactor.Task; >>>>> public class Timer { >>>>> >>>>> private CollectorImpl collector; >>>>> - private PriorityQueue<Task> tasks = new PriorityQueue<Task>(); >>>>> + private PriorityQueue<TaskImpl> tasks = new >>>>> PriorityQueue<TaskImpl>(); >>>>> >>>>> public Timer(Collector collector) { >>>>> this.collector = (CollectorImpl)collector; >>>>> @@ -44,6 +44,7 @@ public class Timer { >>>>> } >>>>> >>>>> long deadline() { >>>>> + flushCancelled(); >>>>> if (tasks.size() > 0) { >>>>> Task task = tasks.peek(); >>>>> return task.deadline(); >>>>> @@ -52,12 +53,23 @@ public class Timer { >>>>> } >>>>> } >>>>> >>>>> + private void flushCancelled() { >>>>> + while (!tasks.isEmpty()) { >>>>> + TaskImpl task = tasks.peek(); >>>>> + if (task.isCancelled()) >>>>> + tasks.poll(); >>>>> + else >>>>> + break; >>>>> + } >>>>> + } >>>>> + >>>>> void tick(long now) { >>>>> while(!tasks.isEmpty()) { >>>>> - Task task = tasks.peek(); >>>>> + TaskImpl task = tasks.peek(); >>>>> if (now >= task.deadline()) { >>>>> tasks.poll(); >>>>> - collector.put(Type.TIMER_TASK, task); >>>>> + if (!task.isCancelled()) >>>>> + collector.put(Type.TIMER_TASK, task); >>>>> } else { >>>>> break; >>>>> } >>>>> @@ -65,6 +77,7 @@ public class Timer { >>>>> } >>>>> >>>>> int tasks() { >>>>> + flushCancelled(); >>>>> return tasks.size(); >>>>> } >>>>> } >>>>> >>>>> http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/d4d22ee3/proton-j/src/main/resources/creactor.py >>>>> ---------------------------------------------------------------------- >>>>> diff --git a/proton-j/src/main/resources/creactor.py >>>>> b/proton-j/src/main/resources/creactor.py >>>>> index e179b23..1f8514e 100644 >>>>> --- a/proton-j/src/main/resources/creactor.py >>>>> +++ b/proton-j/src/main/resources/creactor.py >>>>> @@ -78,6 +78,9 @@ def pn_selectable_set_fd(s, fd): >>>>> def pn_acceptor_close(a): >>>>> a.close() >>>>> >>>>> +def pn_task_cancel(t): >>>>> + t.cancel() >>>>> + >>>>> def pn_object_reactor(o): >>>>> if hasattr(o, "impl"): >>>>> if hasattr(o.impl, "getSession"): >>>>> >>>>> http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/d4d22ee3/tests/python/proton_tests/reactor.py >>>>> ---------------------------------------------------------------------- >>>>> diff --git a/tests/python/proton_tests/reactor.py >>>>> b/tests/python/proton_tests/reactor.py >>>>> index 6afee30..067c5c0 100644 >>>>> --- a/tests/python/proton_tests/reactor.py >>>>> +++ b/tests/python/proton_tests/reactor.py >>>>> @@ -171,3 +171,17 @@ class ExceptionTest(Test): >>>>> assert False, "expected to barf" >>>>> except Barf: >>>>> pass >>>>> + >>>>> + def test_schedule_cancel(self): >>>>> + barf = self.reactor.schedule(10, BarfOnTask()) >>>>> + class CancelBarf: >>>>> + def on_timer_task(self, event): >>>>> + barf.cancel() >>>>> + self.reactor.schedule(0, CancelBarf()) >>>>> + now = self.reactor.mark() >>>>> + try: >>>>> + self.reactor.run() >>>>> + elapsed = self.reactor.mark() - now >>>>> + assert elapsed < 10, "expected cancelled task to not delay >>>>> the >>>>> reactor by " + elapsed >>>>> + except Barf: >>>>> + assert False, "expected barf to be cancelled" >>>>> >>>>> >>>>> --------------------------------------------------------------------- >>>>> To unsubscribe, e-mail: commits-unsubscr...@qpid.apache.org >>>>> For additional commands, e-mail: commits-h...@qpid.apache.org >>>>> >>>>> >
