From 980643b9364f2ec7e75f9e4a391754f5db4bc24a Mon Sep 17 00:00:00 2001 From: Robert Godfrey Date: Tue, 27 May 2008 12:43:04 +0000 Subject: Refactoring updates (job queue changes, enqueue collections..) git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/broker-queue-refactor@660490 13f79535-47bb-0310-9956-ffa450edef68 --- .../org/apache/qpid/framing/AMQShortString.java | 40 ++++++++++++++++++++-- .../org/apache/qpid/pool/ReadWriteJobQueue.java | 30 ++++++++-------- .../org/apache/qpid/pool/ReadWriteRunnable.java | 27 +++++++++++++++ .../pool/ReferenceCountingExecutorService.java | 4 +-- 4 files changed, 81 insertions(+), 20 deletions(-) create mode 100644 java/common/src/main/java/org/apache/qpid/pool/ReadWriteRunnable.java (limited to 'java/common') diff --git a/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java b/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java index f1a7d4970a..c60b9ee0cb 100644 --- a/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java +++ b/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java @@ -464,15 +464,49 @@ public final class AMQShortString implements CharSequence, Comparable implements Blocki private final ReentrantLock _putLock = new ReentrantLock(); - private final ConcurrentLinkedQueue _readJobQueue = new ConcurrentLinkedQueue(); + private final ConcurrentLinkedQueue _readJobQueue = new ConcurrentLinkedQueue(); - private final ConcurrentLinkedQueue _writeJobQueue = new ConcurrentLinkedQueue(); + private final ConcurrentLinkedQueue _writeJobQueue = new ConcurrentLinkedQueue(); private class ReadWriteJobIterator implements Iterator { private boolean _onReads; - private Iterator _iter = _writeJobQueue.iterator(); + private Iterator _iter = _writeJobQueue.iterator(); public boolean hasNext() { @@ -112,12 +112,12 @@ public class ReadWriteJobQueue extends AbstractQueue implements Blocki public boolean offer(final Runnable runnable) { - final Job job = (Job) runnable; + final ReadWriteRunnable job = (ReadWriteRunnable) runnable; final ReentrantLock putLock = _putLock; putLock.lock(); try { - if(job.isReadJob()) + if(job.isRead()) { _readJobQueue.offer(job); } @@ -147,13 +147,13 @@ public class ReadWriteJobQueue extends AbstractQueue implements Blocki public void put(final Runnable runnable) throws InterruptedException { - final Job job = (Job) runnable; + final ReadWriteRunnable job = (ReadWriteRunnable) runnable; final ReentrantLock putLock = _putLock; putLock.lock(); try { - if(job.isReadJob()) + if(job.isRead()) { _readJobQueue.offer(job); } @@ -185,13 +185,13 @@ public class ReadWriteJobQueue extends AbstractQueue implements Blocki public boolean offer(final Runnable runnable, final long timeout, final TimeUnit unit) throws InterruptedException { - final Job job = (Job) runnable; + final ReadWriteRunnable job = (ReadWriteRunnable) runnable; final ReentrantLock putLock = _putLock; putLock.lock(); try { - if(job.isReadJob()) + if(job.isRead()) { _readJobQueue.offer(job); } @@ -240,7 +240,7 @@ public class ReadWriteJobQueue extends AbstractQueue implements Blocki throw ie; } - Job job = _writeJobQueue.poll(); + ReadWriteRunnable job = _writeJobQueue.poll(); if(job == null) { job = _readJobQueue.poll(); @@ -266,7 +266,7 @@ public class ReadWriteJobQueue extends AbstractQueue implements Blocki final AtomicInteger count = _count; long nanos = unit.toNanos(timeout); takeLock.lockInterruptibly(); - Job job = null; + ReadWriteRunnable job = null; try { @@ -322,7 +322,7 @@ public class ReadWriteJobQueue extends AbstractQueue implements Blocki _takeLock.lock(); try { - Job job; + ReadWriteRunnable job; while((job = _writeJobQueue.peek())!= null) { c.add(job); @@ -356,7 +356,7 @@ public class ReadWriteJobQueue extends AbstractQueue implements Blocki _takeLock.lock(); try { - Job job; + ReadWriteRunnable job; while(total<=maxElements && (job = _writeJobQueue.peek())!= null) { c.add(job); @@ -391,7 +391,7 @@ public class ReadWriteJobQueue extends AbstractQueue implements Blocki { if(_count.get() > 0) { - Job job = _writeJobQueue.poll(); + ReadWriteRunnable job = _writeJobQueue.poll(); if(job == null) { job = _readJobQueue.poll(); @@ -417,7 +417,7 @@ public class ReadWriteJobQueue extends AbstractQueue implements Blocki takeLock.lock(); try { - Job job = _writeJobQueue.peek(); + ReadWriteRunnable job = _writeJobQueue.peek(); if(job == null) { job = _readJobQueue.peek(); diff --git a/java/common/src/main/java/org/apache/qpid/pool/ReadWriteRunnable.java b/java/common/src/main/java/org/apache/qpid/pool/ReadWriteRunnable.java new file mode 100644 index 0000000000..ad04a923e1 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/pool/ReadWriteRunnable.java @@ -0,0 +1,27 @@ +package org.apache.qpid.pool; + +/* +* +* 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. +* +*/ +public interface ReadWriteRunnable extends Runnable +{ + boolean isRead(); + boolean isWrite(); +} diff --git a/java/common/src/main/java/org/apache/qpid/pool/ReferenceCountingExecutorService.java b/java/common/src/main/java/org/apache/qpid/pool/ReferenceCountingExecutorService.java index d99973cffb..ce9c6ae4cb 100644 --- a/java/common/src/main/java/org/apache/qpid/pool/ReferenceCountingExecutorService.java +++ b/java/common/src/main/java/org/apache/qpid/pool/ReferenceCountingExecutorService.java @@ -110,7 +110,7 @@ public class ReferenceCountingExecutorService * * @return An executor service. */ - ExecutorService acquireExecutorService() + public ExecutorService acquireExecutorService() { synchronized (_lock) { @@ -140,7 +140,7 @@ public class ReferenceCountingExecutorService * Releases a reference to a shared executor service, decrementing the reference count. If the refence count falls * to zero, the executor service is shut down. */ - void releaseExecutorService() + public void releaseExecutorService() { synchronized (_lock) { -- cgit v1.2.1