woggioni/xmemcached
1
0
Fork 0
Code Actions Packages Releases Activity

initial commit
Some checks failed
CI / build (push) Failing after 27s
Details

Browse Source
This commit is contained in:
Walter Oggioni
2024-12-28 09:02:28 +08:00
commit 82a186dc8f
374 changed files with 49435 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
src/main/java/com/google/code/yanf4j/util/ByteBufferMatcher.java Normal file
View File

@@ -0,0 +1,34 @@
/**
* Copyright [2009-2010] [dennis zhuang(killme2008@gmail.com)] Licensed 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
*/
/**
* Copyright [2009-2010] [dennis zhuang(killme2008@gmail.com)] Licensed 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
*/
package com.google.code.yanf4j.util;
import java.util.List;
import com.google.code.yanf4j.buffer.IoBuffer;
/**
* ByteBuffer matcher
*
* @author dennis
*
*/
public interface ByteBufferMatcher {
public int matchFirst(IoBuffer buffer);
public List<Integer> matchAll(final IoBuffer buffer);
}

197
src/main/java/com/google/code/yanf4j/util/ByteBufferUtils.java Normal file
View File

@@ -0,0 +1,197 @@
/**
* Copyright [2009-2010] [dennis zhuang(killme2008@gmail.com)] Licensed 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
*/
/**
* Copyright [2009-2010] [dennis zhuang(killme2008@gmail.com)] Licensed 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
*/
package com.google.code.yanf4j.util;
/*
* Copyright 2004-2005 the original author or authors.
*
* Licensed 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.
*/
/**
* 来自于cindy2.4的工具类，做了简化和新增
*/
import java.nio.ByteBuffer;
import com.google.code.yanf4j.config.Configuration;
public class ByteBufferUtils {
/**
*
* @param byteBuffer
* @return *
*/
public static final ByteBuffer increaseBufferCapatity(ByteBuffer byteBuffer) {
if (byteBuffer == null) {
throw new IllegalArgumentException("buffer is null");
}
if (Configuration.DEFAULT_INCREASE_BUFF_SIZE < 0) {
throw new IllegalArgumentException("size less than 0");
}
int capacity = byteBuffer.capacity() + Configuration.DEFAULT_INCREASE_BUFF_SIZE;
if (capacity < 0) {
throw new IllegalArgumentException("capacity can't be negative");
}
ByteBuffer result = (byteBuffer.isDirect() ? ByteBuffer.allocateDirect(capacity)
: ByteBuffer.allocate(capacity));
result.order(byteBuffer.order());
byteBuffer.flip();
result.put(byteBuffer);
return result;
}
public static final void flip(ByteBuffer[] buffers) {
if (buffers == null) {
return;
}
for (ByteBuffer buffer : buffers) {
if (buffer != null) {
buffer.flip();
}
}
}
public static final ByteBuffer gather(ByteBuffer[] buffers) {
if (buffers == null || buffers.length == 0) {
return null;
}
ByteBuffer result = ByteBuffer.allocate(remaining(buffers));
result.order(buffers[0].order());
for (int i = 0; i < buffers.length; i++) {
if (buffers[i] != null) {
result.put(buffers[i]);
}
}
result.flip();
return result;
}
public static final int remaining(ByteBuffer[] buffers) {
if (buffers == null) {
return 0;
}
int remaining = 0;
for (int i = 0; i < buffers.length; i++) {
if (buffers[i] != null) {
remaining += buffers[i].remaining();
}
}
return remaining;
}
public static final void clear(ByteBuffer[] buffers) {
if (buffers == null) {
return;
}
for (ByteBuffer buffer : buffers) {
if (buffer != null) {
buffer.clear();
}
}
}
public static final String toHex(byte b) {
return ("" + "0123456789ABCDEF".charAt(0xf & b >> 4) + "0123456789ABCDEF".charAt(b & 0xf));
}
public static final int indexOf(ByteBuffer buffer, ByteBuffer pattern) {
if (pattern == null || buffer == null) {
return -1;
}
int n = buffer.remaining();
int m = pattern.remaining();
int patternPos = pattern.position();
int bufferPos = buffer.position();
if (n < m) {
return -1;
}
for (int s = 0; s <= n - m; s++) {
boolean match = true;
for (int i = 0; i < m; i++) {
if (buffer.get(s + i + bufferPos) != pattern.get(patternPos + i)) {
match = false;
break;
}
}
if (match) {
return (bufferPos + s);
}
}
return -1;
}
public static final int indexOf(ByteBuffer buffer, ByteBuffer pattern, int offset) {
if (offset < 0) {
throw new IllegalArgumentException("offset must be greater than 0");
}
if (pattern == null || buffer == null) {
return -1;
}
int patternPos = pattern.position();
int n = buffer.remaining();
int m = pattern.remaining();
if (n < m) {
return -1;
}
if (offset < buffer.position() || offset > buffer.limit()) {
return -1;
}
for (int s = 0; s <= n - m; s++) {
boolean match = true;
for (int i = 0; i < m; i++) {
if (buffer.get(s + i + offset) != pattern.get(patternPos + i)) {
match = false;
break;
}
}
if (match) {
return (offset + s);
}
}
return -1;
}
/**
* 查看ByteBuffer数组是否还有剩余
*
* @param buffers ByteBuffers
* @return have remaining
*/
public static final boolean hasRemaining(ByteBuffer[] buffers) {
if (buffers == null) {
return false;
}
for (int i = 0; i < buffers.length; i++) {
if (buffers[i] != null && buffers[i].hasRemaining()) {
return true;
}
}
return false;
}
public static final int uByte(byte b) {
return b & 0xFF;
}
}

342
src/main/java/com/google/code/yanf4j/util/CircularQueue.java Normal file
View File

@@ -0,0 +1,342 @@
package com.google.code.yanf4j.util;
/*
* 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.
*/
import java.io.Serializable;
import java.util.AbstractList;
import java.util.Arrays;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.Queue;
/**
* A circular queue from mina
*
* @author dennis
*
* @param <E>
*/
public class CircularQueue<E> extends AbstractList<E> implements List<E>, Queue<E>, Serializable {
private static final long serialVersionUID = 3993421269224511264L;
private static final int DEFAULT_CAPACITY = 4;
private final int initialCapacity;
// XXX: This volatile keyword here is a workaround for SUN Java Compiler
// bug,
// which produces buggy byte code. I don't event know why adding a volatile
// fixes the problem. Eclipse Java Compiler seems to produce correct byte
// code.
private volatile Object[] items;
private int mask;
private int first = 0;
private int last = 0;
private boolean full;
private int shrinkThreshold;
/**
* Construct a new, empty queue.
*/
public CircularQueue() {
this(DEFAULT_CAPACITY);
}
public CircularQueue(int initialCapacity) {
int actualCapacity = normalizeCapacity(initialCapacity);
this.items = new Object[actualCapacity];
this.mask = actualCapacity - 1;
this.initialCapacity = actualCapacity;
this.shrinkThreshold = 0;
}
private static int normalizeCapacity(int initialCapacity) {
int actualCapacity = 1;
while (actualCapacity < initialCapacity) {
actualCapacity <<= 1;
if (actualCapacity < 0) {
actualCapacity = 1 << 30;
break;
}
}
return actualCapacity;
}
/**
* Returns the capacity of this queue.
*/
public int capacity() {
return this.items.length;
}
@Override
public void clear() {
if (!isEmpty()) {
Arrays.fill(this.items, null);
this.first = 0;
this.last = 0;
this.full = false;
shrinkIfNeeded();
}
}
@SuppressWarnings("unchecked")
public E poll() {
if (isEmpty()) {
return null;
}
Object ret = this.items[this.first];
this.items[this.first] = null;
decreaseSize();
if (this.first == this.last) {
this.first = this.last = 0;
}
shrinkIfNeeded();
return (E) ret;
}
public boolean offer(E item) {
if (item == null) {
throw new NullPointerException("item");
}
expandIfNeeded();
this.items[this.last] = item;
increaseSize();
return true;
}
@SuppressWarnings("unchecked")
public E peek() {
if (isEmpty()) {
return null;
}
return (E) this.items[this.first];
}
@SuppressWarnings("unchecked")
@Override
public E get(int idx) {
checkIndex(idx);
return (E) this.items[getRealIndex(idx)];
}
@Override
public boolean isEmpty() {
return (this.first == this.last) && !this.full;
}
@Override
public int size() {
if (this.full) {
return capacity();
}
if (this.last >= this.first) {
return this.last - this.first;
} else {
return this.last - this.first + capacity();
}
}
@Override
public String toString() {
return "first=" + this.first + ", last=" + this.last + ", size=" + size() + ", mask = "
+ this.mask;
}
private void checkIndex(int idx) {
if (idx < 0 || idx >= size()) {
throw new IndexOutOfBoundsException(String.valueOf(idx));
}
}
private int getRealIndex(int idx) {
return (this.first + idx) & this.mask;
}
private void increaseSize() {
this.last = (this.last + 1) & this.mask;
this.full = this.first == this.last;
}
private void decreaseSize() {
this.first = (this.first + 1) & this.mask;
this.full = false;
}
private void expandIfNeeded() {
if (this.full) {
// expand queue
final int oldLen = this.items.length;
final int newLen = oldLen << 1;
Object[] tmp = new Object[newLen];
if (this.first < this.last) {
System.arraycopy(this.items, this.first, tmp, 0, this.last - this.first);
} else {
System.arraycopy(this.items, this.first, tmp, 0, oldLen - this.first);
System.arraycopy(this.items, 0, tmp, oldLen - this.first, this.last);
}
this.first = 0;
this.last = oldLen;
this.items = tmp;
this.mask = tmp.length - 1;
if (newLen >>> 3 > this.initialCapacity) {
this.shrinkThreshold = newLen >>> 3;
}
}
}
private void shrinkIfNeeded() {
int size = size();
if (size <= this.shrinkThreshold) {
// shrink queue
final int oldLen = this.items.length;
int newLen = normalizeCapacity(size);
if (size == newLen) {
newLen <<= 1;
}
if (newLen >= oldLen) {
return;
}
if (newLen < this.initialCapacity) {
if (oldLen == this.initialCapacity) {
return;
} else {
newLen = this.initialCapacity;
}
}
Object[] tmp = new Object[newLen];
// Copy only when there's something to copy.
if (size > 0) {
if (this.first < this.last) {
System.arraycopy(this.items, this.first, tmp, 0, this.last - this.first);
} else {
System.arraycopy(this.items, this.first, tmp, 0, oldLen - this.first);
System.arraycopy(this.items, 0, tmp, oldLen - this.first, this.last);
}
}
this.first = 0;
this.last = size;
this.items = tmp;
this.mask = tmp.length - 1;
this.shrinkThreshold = 0;
}
}
@Override
public boolean add(E o) {
return offer(o);
}
@SuppressWarnings("unchecked")
@Override
public E set(int idx, E o) {
checkIndex(idx);
int realIdx = getRealIndex(idx);
Object old = this.items[realIdx];
this.items[realIdx] = o;
return (E) old;
}
@Override
public void add(int idx, E o) {
if (idx == size()) {
offer(o);
return;
}
checkIndex(idx);
expandIfNeeded();
int realIdx = getRealIndex(idx);
// Make a room for a new element.
if (this.first < this.last) {
System.arraycopy(this.items, realIdx, this.items, realIdx + 1, this.last - realIdx);
} else {
if (realIdx >= this.first) {
System.arraycopy(this.items, 0, this.items, 1, this.last);
this.items[0] = this.items[this.items.length - 1];
System.arraycopy(this.items, realIdx, this.items, realIdx + 1,
this.items.length - realIdx - 1);
} else {
System.arraycopy(this.items, realIdx, this.items, realIdx + 1, this.last - realIdx);
}
}
this.items[realIdx] = o;
increaseSize();
}
@SuppressWarnings("unchecked")
@Override
public E remove(int idx) {
if (idx == 0) {
return poll();
}
checkIndex(idx);
int realIdx = getRealIndex(idx);
Object removed = this.items[realIdx];
// Remove a room for the removed element.
if (this.first < this.last) {
System.arraycopy(this.items, this.first, this.items, this.first + 1, realIdx - this.first);
} else {
if (realIdx >= this.first) {
System.arraycopy(this.items, this.first, this.items, this.first + 1, realIdx - this.first);
} else {
System.arraycopy(this.items, 0, this.items, 1, realIdx);
this.items[0] = this.items[this.items.length - 1];
System.arraycopy(this.items, this.first, this.items, this.first + 1,
this.items.length - this.first - 1);
}
}
this.items[this.first] = null;
decreaseSize();
shrinkIfNeeded();
return (E) removed;
}
public E remove() {
if (isEmpty()) {
throw new NoSuchElementException();
}
return poll();
}
public E element() {
if (isEmpty()) {
throw new NoSuchElementException();
}
return peek();
}
}

47
src/main/java/com/google/code/yanf4j/util/ConcurrentHashSet.java Normal file
View File

@@ -0,0 +1,47 @@
/**
* Copyright [2009-2010] [dennis zhuang(killme2008@gmail.com)] Licensed 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
*/
/**
* Copyright [2009-2010] [dennis zhuang(killme2008@gmail.com)] Licensed 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
*/
package com.google.code.yanf4j.util;
import java.util.Collection;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
/**
* A {@link ConcurrentHashMap}-backed {@link Set}.
*
* @author The Apache MINA Project (dev@mina.apache.org)
* @version $Rev: 597692 $, $Date: 2007-11-23 08:56:32 -0700 (Fri, 23 Nov 2007) $
*/
public class ConcurrentHashSet<E> extends MapBackedSet<E> {
private static final long serialVersionUID = 8518578988740277828L;
public ConcurrentHashSet() {
super(new ConcurrentHashMap<E, Boolean>());
}
public ConcurrentHashSet(Collection<E> c) {
super(new ConcurrentHashMap<E, Boolean>(), c);
}
@Override
public boolean add(E o) {
Boolean answer = ((ConcurrentMap<E, Boolean>) map).putIfAbsent(o, Boolean.TRUE);
return answer == null;
}
}

22
src/main/java/com/google/code/yanf4j/util/DispatcherFactory.java Normal file
View File

@@ -0,0 +1,22 @@
package com.google.code.yanf4j.util;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.TimeUnit;
import com.google.code.yanf4j.core.impl.PoolDispatcher;
/**
* Dispatcher Factory
*
* @author dennis
*
*/
public class DispatcherFactory {
public static com.google.code.yanf4j.core.Dispatcher newDispatcher(int size,
RejectedExecutionHandler rejectedExecutionHandler, String prefix) {
if (size > 0) {
return new PoolDispatcher(size, 60, TimeUnit.SECONDS, rejectedExecutionHandler, prefix);
} else {
return null;
}
}
}

780
src/main/java/com/google/code/yanf4j/util/LinkedTransferQueue.java Normal file
View File

@@ -0,0 +1,780 @@
/*
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
*
* Copyright 1997-2008 Sun Microsystems, Inc. All rights reserved.
*
* The contents of this file are subject to the terms of either the GNU General Public License
* Version 2 only ("GPL") or the Common Development and Distribution License("CDDL") (collectively,
* the "License"). You may not use this file except in compliance with the License. You can obtain a
* copy of the License at https://glassfish.dev.java.net/public/CDDL+GPL.html or
* glassfish/bootstrap/legal/LICENSE.txt. See the License for the specific language governing
* permissions and limitations under the License.
*
* When distributing the software, include this License Header Notice in each file and include the
* License file at glassfish/bootstrap/legal/LICENSE.txt. Sun designates this particular file as
* subject to the "Classpath" exception as provided by Sun in the GPL Version 2 section of the
* License file that accompanied this code. If applicable, add the following below the License
* Header, with the fields enclosed by brackets [] replaced by your own identifying information:
* "Portions Copyrighted [year] [name of copyright owner]"
*
* Contributor(s):
*
* If you wish your version of this file to be governed by only the CDDL or only the GPL Version 2,
* indicate your decision by adding "[Contributor] elects to include this software in this
* distribution under the [CDDL or GPL Version 2] license." If you don't indicate a single choice of
* license, a recipient has the option to distribute your version of this file under either the
* CDDL, the GPL Version 2 or to extend the choice of license to its licensees as provided above.
* However, if you add GPL Version 2 code and therefore, elected the GPL Version 2 license, then the
* option applies only if the new code is made subject to such option by the copyright holder.
*/
/*
* Written by Doug Lea with assistance from members of JCP JSR-166 Expert Group and released to the
* public domain, as explained at http://creativecommons.org/licenses/publicdomain
*/
package com.google.code.yanf4j.util;
import java.util.AbstractQueue;
import java.util.Collection;
import java.util.Iterator;
import java.util.NoSuchElementException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.concurrent.locks.LockSupport;
/**
* An unbounded <code>TransferQueue</code> based on linked nodes. This queue orders elements FIFO
* (first-in-first-out) with respect to any given producer. The <em>head</em> of the queue is that
* element that has been on the queue the longest time for some producer. The <em>tail</em> of the
* queue is that element that has been on the queue the shortest time for some producer.
*
* <p>
* Beware that, unlike in most collections, the <code>size</code> method is <em>NOT</em> a constant-time
* operation. Because of the asynchronous nature of these queues, determining the current number of
* elements requires a traversal of the elements.
*
* <p>
* This class and its iterator implement all of the <em>optional</em> methods of the
* {@link Collection} and {@link Iterator} interfaces.
*
* <p>
* Memory consistency effects: As with other concurrent collections, actions in a thread prior to
* placing an object into a {@code LinkedTransferQueue}
* <a href="package-summary.html#MemoryVisibility"><i>happen-before</i></a> actions subsequent to
* the access or removal of that element from the {@code LinkedTransferQueue} in another thread.
*
* @author Doug Lea
* @author The Netty Project (netty-dev@lists.jboss.org)
* @author Trustin Lee (tlee@redhat.com)
*
* @param <E> the type of elements held in this collection
*
*/
public class LinkedTransferQueue<E> extends AbstractQueue<E> implements BlockingQueue<E> {
/*
* This class extends the approach used in FIFO-mode SynchronousQueues. See the internal
* documentation, as well as the PPoPP 2006 paper "Scalable Synchronous Queues" by Scherer, Lea &
* Scott (http://www.cs.rice.edu/~wns1/papers/2006-PPoPP-SQ.pdf)
*
* The main extension is to provide different Wait modes for the main "xfer" method that puts or
* takes items. These don't impact the basic dual-queue logic, but instead control whether or how
* threads block upon insertion of request or data nodes into the dual queue. It also uses
* slightly different conventions for tracking whether nodes are off-list or cancelled.
*/
// Wait modes for xfer method
private static final int NOWAIT = 0;
private static final int TIMEOUT = 1;
private static final int WAIT = 2;
/** The number of CPUs, for spin control */
private static final int NCPUS = Runtime.getRuntime().availableProcessors();
/**
* The number of times to spin before blocking in timed waits. The value is empirically derived --
* it works well across a variety of processors and OSes. Empirically, the best value seems not to
* vary with number of CPUs (beyond 2) so is just a constant.
*/
private static final int maxTimedSpins = NCPUS < 2 ? 0 : 32;
/**
* The number of times to spin before blocking in untimed waits. This is greater than timed value
* because untimed waits spin faster since they don't need to check times on each spin.
*/
private static final int maxUntimedSpins = maxTimedSpins * 16;
/**
* The number of nanoseconds for which it is faster to spin rather than to use timed park. A rough
* estimate suffices.
*/
private static final long spinForTimeoutThreshold = 1000L;
/**
* Node class for LinkedTransferQueue. Opportunistically subclasses from AtomicReference to
* represent item. Uses Object, not E, to allow setting item to "this" after use, to avoid garbage
* retention. Similarly, setting the next field to this is used as sentinel that node is off list.
*/
private static final class QNode extends AtomicReference<Object> {
private static final long serialVersionUID = 5925596372370723938L;
transient volatile QNode next;
transient volatile Thread waiter; // to control park/unpark
final boolean isData;
public long p1, p2, p3, p4, p5, p6, p7;
QNode(Object item, boolean isData) {
super(item);
this.isData = isData;
}
private static final AtomicReferenceFieldUpdater<QNode, QNode> nextUpdater;
static {
AtomicReferenceFieldUpdater<QNode, QNode> tmp = null;
try {
tmp = AtomicReferenceFieldUpdater.newUpdater(QNode.class, QNode.class, "next");
// Test if AtomicReferenceFieldUpdater is really working.
QNode testNode = new QNode(null, false);
tmp.set(testNode, testNode);
if (testNode.next != testNode) {
// Not set as expected - fall back to the safe mode.
throw new Exception();
}
} catch (Throwable t) {
// Running in a restricted environment with a security manager.
tmp = null;
}
nextUpdater = tmp;
}
boolean casNext(QNode cmp, QNode val) {
if (nextUpdater != null) {
return nextUpdater.compareAndSet(this, cmp, val);
} else {
return alternativeCasNext(cmp, val);
}
}
private synchronized boolean alternativeCasNext(QNode cmp, QNode val) {
if (this.next == cmp) {
this.next = val;
return true;
} else {
return false;
}
}
}
/**
* Padded version of AtomicReference used for head, tail and cleanMe, to alleviate contention
* across threads CASing one vs the other.
*/
public static final class PaddedAtomicReference<T> extends AtomicReference<T> {
private static final long serialVersionUID = 4684288940772921317L;
// enough padding for 64bytes with 4byte refs
@SuppressWarnings("unused")
public Object p0, p1, p2, p3, p4, p5, p6, p7, p8, p9, pa, pb, pc, pd, pe;
public PaddedAtomicReference(T r) {
super(r);
}
}
/** head of the queue */
private final PaddedAtomicReference<QNode> head;
/** tail of the queue */
private final PaddedAtomicReference<QNode> tail;
/**
* Reference to a cancelled node that might not yet have been unlinked from queue because it was
* the last inserted node when it cancelled.
*/
private final PaddedAtomicReference<QNode> cleanMe;
/**
* Tries to cas nh as new head; if successful, unlink old head's next node to avoid garbage
* retention.
*/
private boolean advanceHead(QNode h, QNode nh) {
if (h == this.head.get() && this.head.compareAndSet(h, nh)) {
h.next = h; // forget old next
return true;
}
return false;
}
/**
* Puts or takes an item. Used for most queue operations (except poll() and tryTransfer()). See
* the similar code in SynchronousQueue for detailed explanation.
*
* @param e the item or if null, signifies that this is a take
* @param mode the wait mode: NOWAIT, TIMEOUT, WAIT
* @param nanos timeout in nanosecs, used only if mode is TIMEOUT
* @return an item, or null on failure
*/
private Object xfer(Object e, int mode, long nanos) {
boolean isData = e != null;
QNode s = null;
final PaddedAtomicReference<QNode> head = this.head;
final PaddedAtomicReference<QNode> tail = this.tail;
for (;;) {
QNode t = tail.get();
QNode h = head.get();
if (t != null && (t == h || t.isData == isData)) {
if (s == null) {
s = new QNode(e, isData);
}
QNode last = t.next;
if (last != null) {
if (t == tail.get()) {
tail.compareAndSet(t, last);
}
} else if (t.casNext(null, s)) {
tail.compareAndSet(t, s);
return awaitFulfill(t, s, e, mode, nanos);
}
}
else if (h != null) {
QNode first = h.next;
if (t == tail.get() && first != null && advanceHead(h, first)) {
Object x = first.get();
if (x != first && first.compareAndSet(x, e)) {
LockSupport.unpark(first.waiter);
return isData ? e : x;
}
}
}
}
}
/**
* Version of xfer for poll() and tryTransfer, which simplifies control paths both here and in
* xfer
*/
private Object fulfill(Object e) {
boolean isData = e != null;
final PaddedAtomicReference<QNode> head = this.head;
final PaddedAtomicReference<QNode> tail = this.tail;
for (;;) {
QNode t = tail.get();
QNode h = head.get();
if (t != null && (t == h || t.isData == isData)) {
QNode last = t.next;
if (t == tail.get()) {
if (last != null) {
tail.compareAndSet(t, last);
} else {
return null;
}
}
} else if (h != null) {
QNode first = h.next;
if (t == tail.get() && first != null && advanceHead(h, first)) {
Object x = first.get();
if (x != first && first.compareAndSet(x, e)) {
LockSupport.unpark(first.waiter);
return isData ? e : x;
}
}
}
}
}
/**
* Spins/blocks until node s is fulfilled or caller gives up, depending on wait mode.
*
* @param pred the predecessor of waiting node
* @param s the waiting node
* @param e the comparison value for checking match
* @param mode mode
* @param nanos timeout value
* @return matched item, or s if cancelled
*/
private Object awaitFulfill(QNode pred, QNode s, Object e, int mode, long nanos) {
if (mode == NOWAIT) {
return null;
}
long lastTime = mode == TIMEOUT ? System.nanoTime() : 0;
Thread w = Thread.currentThread();
int spins = -1; // set to desired spin count below
for (;;) {
if (w.isInterrupted()) {
s.compareAndSet(e, s);
}
Object x = s.get();
if (x != e) { // Node was matched or cancelled
advanceHead(pred, s); // unlink if head
if (x == s) { // was cancelled
clean(pred, s);
return null;
} else if (x != null) {
s.set(s); // avoid garbage retention
return x;
} else {
return e;
}
}
if (mode == TIMEOUT) {
long now = System.nanoTime();
nanos -= now - lastTime;
lastTime = now;
if (nanos <= 0) {
s.compareAndSet(e, s); // try to cancel
continue;
}
}
if (spins < 0) {
QNode h = this.head.get(); // only spin if at head
spins = h != null && h.next == s ? (mode == TIMEOUT ? maxTimedSpins : maxUntimedSpins) : 0;
}
if (spins > 0) {
--spins;
} else if (s.waiter == null) {
s.waiter = w;
} else if (mode != TIMEOUT) {
// LockSupport.park(this);
LockSupport.park(); // allows run on java5
s.waiter = null;
spins = -1;
} else if (nanos > spinForTimeoutThreshold) {
// LockSupport.parkNanos(this, nanos);
LockSupport.parkNanos(nanos);
s.waiter = null;
spins = -1;
}
}
}
/**
* Returns validated tail for use in cleaning methods
*/
private QNode getValidatedTail() {
for (;;) {
QNode h = this.head.get();
QNode first = h.next;
if (first != null && first.next == first) { // help advance
advanceHead(h, first);
continue;
}
QNode t = this.tail.get();
QNode last = t.next;
if (t == this.tail.get()) {
if (last != null) {
this.tail.compareAndSet(t, last); // help advance
} else {
return t;
}
}
}
}
/**
* Gets rid of cancelled node s with original predecessor pred.
*
* @param pred predecessor of cancelled node
* @param s the cancelled node
*/
void clean(QNode pred, QNode s) {
Thread w = s.waiter;
if (w != null) { // Wake up thread
s.waiter = null;
if (w != Thread.currentThread()) {
LockSupport.unpark(w);
}
}
/*
* At any given time, exactly one node on list cannot be deleted -- the last inserted node. To
* accommodate this, if we cannot delete s, we save its predecessor as "cleanMe", processing the
* previously saved version first. At least one of node s or the node previously saved can
* always be processed, so this always terminates.
*/
while (pred.next == s) {
QNode oldpred = reclean(); // First, help get rid of cleanMe
QNode t = getValidatedTail();
if (s != t) { // If not tail, try to unsplice
QNode sn = s.next; // s.next == s means s already off list
if (sn == s || pred.casNext(s, sn)) {
break;
}
} else if (oldpred == pred || // Already saved
oldpred == null && this.cleanMe.compareAndSet(null, pred)) {
break; // Postpone cleaning
}
}
}
/**
* Tries to unsplice the cancelled node held in cleanMe that was previously uncleanable because it
* was at tail.
*
* @return current cleanMe node (or null)
*/
private QNode reclean() {
/*
* cleanMe is, or at one time was, predecessor of cancelled node s that was the tail so could
* not be unspliced. If s is no longer the tail, try to unsplice if necessary and make cleanMe
* slot available. This differs from similar code in clean() because we must check that pred
* still points to a cancelled node that must be unspliced -- if not, we can (must) clear
* cleanMe without unsplicing. This can loop only due to contention on casNext or clearing
* cleanMe.
*/
QNode pred;
while ((pred = this.cleanMe.get()) != null) {
QNode t = getValidatedTail();
QNode s = pred.next;
if (s != t) {
QNode sn;
if (s == null || s == pred || s.get() != s || (sn = s.next) == s || pred.casNext(s, sn)) {
this.cleanMe.compareAndSet(pred, null);
}
} else {
break;
}
}
return pred;
}
@SuppressWarnings("unchecked")
E cast(Object e) {
return (E) e;
}
/**
* Creates an initially empty <code>LinkedTransferQueue</code>.
*/
public LinkedTransferQueue() {
QNode dummy = new QNode(null, false);
this.head = new PaddedAtomicReference<QNode>(dummy);
this.tail = new PaddedAtomicReference<QNode>(dummy);
this.cleanMe = new PaddedAtomicReference<QNode>(null);
}
/**
* Creates a <code>LinkedTransferQueue</code> initially containing the elements of the given
* collection, added in traversal order of the collection's iterator.
*
* @param c the collection of elements to initially contain
* @throws NullPointerException if the specified collection or any of its elements are null
*/
public LinkedTransferQueue(Collection<? extends E> c) {
this();
addAll(c);
}
public void put(E e) throws InterruptedException {
if (e == null) {
throw new NullPointerException();
}
if (Thread.interrupted()) {
throw new InterruptedException();
}
xfer(e, NOWAIT, 0);
}
public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException {
if (e == null) {
throw new NullPointerException();
}
if (Thread.interrupted()) {
throw new InterruptedException();
}
xfer(e, NOWAIT, 0);
return true;
}
public boolean offer(E e) {
if (e == null) {
throw new NullPointerException();
}
xfer(e, NOWAIT, 0);
return true;
}
public void transfer(E e) throws InterruptedException {
if (e == null) {
throw new NullPointerException();
}
if (xfer(e, WAIT, 0) == null) {
Thread.interrupted();
throw new InterruptedException();
}
}
public boolean tryTransfer(E e, long timeout, TimeUnit unit) throws InterruptedException {
if (e == null) {
throw new NullPointerException();
}
if (xfer(e, TIMEOUT, unit.toNanos(timeout)) != null) {
return true;
}
if (!Thread.interrupted()) {
return false;
}
throw new InterruptedException();
}
public boolean tryTransfer(E e) {
if (e == null) {
throw new NullPointerException();
}
return fulfill(e) != null;
}
public E take() throws InterruptedException {
Object e = xfer(null, WAIT, 0);
if (e != null) {
return cast(e);
}
Thread.interrupted();
throw new InterruptedException();
}
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
Object e = xfer(null, TIMEOUT, unit.toNanos(timeout));
if (e != null || !Thread.interrupted()) {
return cast(e);
}
throw new InterruptedException();
}
public E poll() {
return cast(fulfill(null));
}
public int drainTo(Collection<? super E> c) {
if (c == null) {
throw new NullPointerException();
}
if (c == this) {
throw new IllegalArgumentException();
}
int n = 0;
E e;
while ((e = poll()) != null) {
c.add(e);
++n;
}
return n;
}
public int drainTo(Collection<? super E> c, int maxElements) {
if (c == null) {
throw new NullPointerException();
}
if (c == this) {
throw new IllegalArgumentException();
}
int n = 0;
E e;
while (n < maxElements && (e = poll()) != null) {
c.add(e);
++n;
}
return n;
}
// Traversal-based methods
/**
* Return head after performing any outstanding helping steps
*/
QNode traversalHead() {
for (;;) {
QNode t = this.tail.get();
QNode h = this.head.get();
if (h != null && t != null) {
QNode last = t.next;
QNode first = h.next;
if (t == this.tail.get()) {
if (last != null) {
this.tail.compareAndSet(t, last);
} else if (first != null) {
Object x = first.get();
if (x == first) {
advanceHead(h, first);
} else {
return h;
}
} else {
return h;
}
}
}
}
}
@Override
public Iterator<E> iterator() {
return new Itr();
}
/**
* Iterators. Basic strategy is to traverse list, treating non-data (i.e., request) nodes as
* terminating list. Once a valid data node is found, the item is cached so that the next call to
* next() will return it even if subsequently removed.
*/
class Itr implements Iterator<E> {
QNode nextNode; // Next node to return next
QNode currentNode; // last returned node, for remove()
QNode prevNode; // predecessor of last returned node
E nextItem; // Cache of next item, once commited to in next
Itr() {
this.nextNode = traversalHead();
advance();
}
E advance() {
this.prevNode = this.currentNode;
this.currentNode = this.nextNode;
E x = this.nextItem;
QNode p = this.nextNode.next;
for (;;) {
if (p == null || !p.isData) {
this.nextNode = null;
this.nextItem = null;
return x;
}
Object item = p.get();
if (item != p && item != null) {
this.nextNode = p;
this.nextItem = cast(item);
return x;
}
this.prevNode = p;
p = p.next;
}
}
public boolean hasNext() {
return this.nextNode != null;
}
public E next() {
if (this.nextNode == null) {
throw new NoSuchElementException();
}
return advance();
}
public void remove() {
QNode p = this.currentNode;
QNode prev = this.prevNode;
if (prev == null || p == null) {
throw new IllegalStateException();
}
Object x = p.get();
if (x != null && x != p && p.compareAndSet(x, p)) {
clean(prev, p);
}
}
}
public E peek() {
for (;;) {
QNode h = traversalHead();
QNode p = h.next;
if (p == null) {
return null;
}
Object x = p.get();
if (p != x) {
if (!p.isData) {
return null;
}
if (x != null) {
return cast(x);
}
}
}
}
@Override
public boolean isEmpty() {
for (;;) {
QNode h = traversalHead();
QNode p = h.next;
if (p == null) {
return true;
}
Object x = p.get();
if (p != x) {
if (!p.isData) {
return true;
}
if (x != null) {
return false;
}
}
}
}
public boolean hasWaitingConsumer() {
for (;;) {
QNode h = traversalHead();
QNode p = h.next;
if (p == null) {
return false;
}
Object x = p.get();
if (p != x) {
return !p.isData;
}
}
}
/**
* Returns the number of elements in this queue. If this queue contains more than
* <code>Integer.MAX_VALUE</code> elements, returns <code>Integer.MAX_VALUE</code>.
*
* <p>
* Beware that, unlike in most collections, this method is <em>NOT</em> a constant-time operation.
* Because of the asynchronous nature of these queues, determining the current number of elements
* requires an O(n) traversal.
*
* @return the number of elements in this queue
*/
@Override
public int size() {
int count = 0;
QNode h = traversalHead();
for (QNode p = h.next; p != null && p.isData; p = p.next) {
Object x = p.get();
if (x != null && x != p) {
if (++count == Integer.MAX_VALUE) {
break;
}
}
}
return count;
}
public int getWaitingConsumerCount() {
int count = 0;
QNode h = traversalHead();
for (QNode p = h.next; p != null && !p.isData; p = p.next) {
if (p.get() == null) {
if (++count == Integer.MAX_VALUE) {
break;
}
}
}
return count;
}
public int remainingCapacity() {
return Integer.MAX_VALUE;
}
}

60
src/main/java/com/google/code/yanf4j/util/MapBackedSet.java Normal file
View File

@@ -0,0 +1,60 @@
package com.google.code.yanf4j.util;
import java.io.Serializable;
import java.util.AbstractSet;
import java.util.Collection;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
/**
* A {@link Map}-backed {@link Set}.
*
* @author The Apache MINA Project (dev@mina.apache.org)
* @version $Rev: 597692 $, $Date: 2007-11-23 08:56:32 -0700 (Fri, 23 Nov 2007) $
*/
public class MapBackedSet<E> extends AbstractSet<E> implements Serializable {
private static final long serialVersionUID = -8347878570391674042L;
protected final Map<E, Boolean> map;
public MapBackedSet(Map<E, Boolean> map) {
this.map = map;
}
public MapBackedSet(Map<E, Boolean> map, Collection<E> c) {
this.map = map;
addAll(c);
}
@Override
public int size() {
return map.size();
}
@Override
public boolean contains(Object o) {
return map.containsKey(o);
}
@Override
public Iterator<E> iterator() {
return map.keySet().iterator();
}
@Override
public boolean add(E o) {
return map.put(o, Boolean.TRUE) == null;
}
@Override
public boolean remove(Object o) {
return map.remove(o) != null;
}
@Override
public void clear() {
map.clear();
}
}

52
src/main/java/com/google/code/yanf4j/util/PropertyUtils.java Normal file
View File

@@ -0,0 +1,52 @@
/**
* Copyright [2009-2010] [dennis zhuang(killme2008@gmail.com)] Licensed 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
*/
/**
* Copyright [2009-2010] [dennis zhuang(killme2008@gmail.com)] Licensed 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
*/
package com.google.code.yanf4j.util;
import java.util.Properties;
/**
* java.util.Property utils
*
* @author dennis
*
*/
public class PropertyUtils {
public static int getPropertyAsInteger(Properties props, String propName) {
return Integer.parseInt(PropertyUtils.getProperty(props, propName));
}
public static String getProperty(Properties props, String name) {
return props.getProperty(name).trim();
}
public static boolean getPropertyAsBoolean(Properties props, String name) {
return Boolean.valueOf(getProperty(props, name));
}
public static long getPropertyAsLong(Properties props, String name) {
return Long.parseLong(getProperty(props, name));
}
public static short getPropertyAsShort(Properties props, String name) {
return Short.parseShort(getProperty(props, name));
}
public static byte getPropertyAsByte(Properties props, String name) {
return Byte.parseByte(getProperty(props, name));
}
}

208
src/main/java/com/google/code/yanf4j/util/ResourcesUtils.java Normal file
View File

@@ -0,0 +1,208 @@
/**
* Copyright [2008] [dennis zhuang] Licensed 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
*/
package com.google.code.yanf4j.util;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.Reader;
import java.net.URL;
import java.util.Properties;
/**
* Resource utils
*
* @author dennis
*
*/
public class ResourcesUtils extends Object {
/** */
/**
* Returns the URL of the resource on the classpath
*
* @param resource The resource to find
* @throws IOException If the resource cannot be found or read
* @return The resource
*/
public static URL getResourceURL(String resource) throws IOException {
URL url = null;
ClassLoader loader = ResourcesUtils.class.getClassLoader();
if (loader != null) {
url = loader.getResource(resource);
}
if (url == null) {
url = ClassLoader.getSystemResource(resource);
}
if (url == null) {
throw new IOException("Could not find resource " + resource);
}
return url;
}
/** */
/**
* Returns the URL of the resource on the classpath
*
* @param loader The classloader used to load the resource
* @param resource The resource to find
* @throws IOException If the resource cannot be found or read
* @return The resource
*/
public static URL getResourceURL(ClassLoader loader, String resource) throws IOException {
URL url = null;
if (loader != null) {
url = loader.getResource(resource);
}
if (url == null) {
url = ClassLoader.getSystemResource(resource);
}
if (url == null) {
throw new IOException("Could not find resource " + resource);
}
return url;
}
/** */
/**
* Returns a resource on the classpath as a Stream object
*
* @param resource The resource to find
* @throws IOException If the resource cannot be found or read
* @return The resource
*/
public static InputStream getResourceAsStream(String resource) throws IOException {
InputStream in = null;
ClassLoader loader = ResourcesUtils.class.getClassLoader();
if (loader != null) {
in = loader.getResourceAsStream(resource);
}
if (in == null) {
in = ClassLoader.getSystemResourceAsStream(resource);
}
if (in == null) {
throw new IOException("Could not find resource " + resource);
}
return in;
}
/** */
/**
* Returns a resource on the classpath as a Stream object
*
* @param loader The classloader used to load the resource
* @param resource The resource to find
* @throws IOException If the resource cannot be found or read
* @return The resource
*/
public static InputStream getResourceAsStream(ClassLoader loader, String resource)
throws IOException {
InputStream in = null;
if (loader != null) {
in = loader.getResourceAsStream(resource);
}
if (in == null) {
in = ClassLoader.getSystemResourceAsStream(resource);
}
if (in == null) {
throw new IOException("Could not find resource " + resource);
}
return in;
}
/** */
/**
* Returns a resource on the classpath as a Properties object
*
* @param resource The resource to find
* @throws IOException If the resource cannot be found or read
* @return The resource
*/
public static Properties getResourceAsProperties(String resource) throws IOException {
Properties props = new Properties();
InputStream in = null;
String propfile = resource;
in = getResourceAsStream(propfile);
props.load(in);
in.close();
return props;
}
/** */
/**
* Returns a resource on the classpath as a Properties object
*
* @param loader The classloader used to load the resource
* @param resource The resource to find
* @throws IOException If the resource cannot be found or read
* @return The resource
*/
public static Properties getResourceAsProperties(ClassLoader loader, String resource)
throws IOException {
Properties props = new Properties();
InputStream in = null;
String propfile = resource;
in = getResourceAsStream(loader, propfile);
props.load(in);
in.close();
return props;
}
/** */
/**
* Returns a resource on the classpath as a Reader object
*
* @param resource The resource to find
* @throws IOException If the resource cannot be found or read
* @return The resource
*/
public static InputStreamReader getResourceAsReader(String resource) throws IOException {
return new InputStreamReader(getResourceAsStream(resource));
}
/** */
/**
* Returns a resource on the classpath as a Reader object
*
* @param loader The classloader used to load the resource
* @param resource The resource to find
* @throws IOException If the resource cannot be found or read
* @return The resource
*/
public static Reader getResourceAsReader(ClassLoader loader, String resource) throws IOException {
return new InputStreamReader(getResourceAsStream(loader, resource));
}
/** */
/**
* Returns a resource on the classpath as a File object
*
* @param resource The resource to find
* @throws IOException If the resource cannot be found or read
* @return The resource
*/
public static File getResourceAsFile(String resource) throws IOException {
return new File(getResourceURL(resource).getFile());
}
/** */
/**
* Returns a resource on the classpath as a File object
*
* @param loader The classloader used to load the resource
* @param resource The resource to find
* @throws IOException If the resource cannot be found or read
* @return The resource
*/
public static File getResourceAsFile(ClassLoader loader, String resource) throws IOException {
return new File(getResourceURL(loader, resource).getFile());
}
}

158
src/main/java/com/google/code/yanf4j/util/SelectorFactory.java Normal file
View File

@@ -0,0 +1,158 @@
/*
* The contents of this file are subject to the terms of the Common Development and Distribution
* License (the License). You may not use this file except in compliance with the License.
*
* You can obtain a copy of the license at https://glassfish.dev.java.net/public/CDDLv1.0.html or
* glassfish/bootstrap/legal/CDDLv1.0.txt. See the License for the specific language governing
* permissions and limitations under the License.
*
* When distributing Covered Code, include this CDDL Header Notice in each file and include the
* License file at glassfish/bootstrap/legal/CDDLv1.0.txt. If applicable, add the following below
* the CDDL Header, with the fields enclosed by brackets [] replaced by you own identifying
* information: "Portions Copyrighted [year] [name of copyright owner]"
*
* Copyright 2007 Sun Microsystems, Inc. All rights reserved.
*/
package com.google.code.yanf4j.util;
import java.io.IOException;
import java.nio.channels.Selector;
import java.util.EmptyStackException;
import java.util.Stack;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Temp selector factory,come from grizzly
*
* @author dennis zhuang
*/
public class SelectorFactory {
public static final int DEFAULT_MAX_SELECTORS = 20;
private static final Logger logger = LoggerFactory.getLogger(SelectorFactory.class);
/**
* The timeout before we exit.
*/
public final static long timeout = 5000;
/**
* The number of <code>Selector</code> to create.
*/
private static int maxSelectors;
/**
* Cache of <code>Selector</code>
*/
private final static Stack<Selector> selectors = new Stack<Selector>();
/**
* Creates the <code>Selector</code>
*/
static {
try {
setMaxSelectors(DEFAULT_MAX_SELECTORS);
} catch (IOException ex) {
logger.warn("SelectorFactory initializing Selector pool", ex);
}
}
/**
* Set max selector pool size.
*
* @param size max pool size
*/
public final static void setMaxSelectors(int size) throws IOException {
synchronized (selectors) {
if (size < maxSelectors) {
reduce(size);
} else if (size > maxSelectors) {
grow(size);
}
maxSelectors = size;
}
}
/**
* Returns max selector pool size
*
* @return max pool size
*/
public final static int getMaxSelectors() {
return maxSelectors;
}
/**
* Get a exclusive <code>Selector</code>
*
* @return <code>Selector</code>
*/
public final static Selector getSelector() {
synchronized (selectors) {
Selector s = null;
try {
if (selectors.size() != 0) {
s = selectors.pop();
}
} catch (EmptyStackException ex) {
}
int attempts = 0;
try {
while (s == null && attempts < 2) {
selectors.wait(timeout);
try {
if (selectors.size() != 0) {
s = selectors.pop();
}
} catch (EmptyStackException ex) {
break;
}
attempts++;
}
} catch (InterruptedException ex) {
}
return s;
}
}
/**
* Return the <code>Selector</code> to the cache
*
* @param s <code>Selector</code>
*/
public final static void returnSelector(Selector s) {
synchronized (selectors) {
selectors.push(s);
if (selectors.size() == 1) {
selectors.notify();
}
}
}
/**
* Increase <code>Selector</code> pool size
*/
private static void grow(int size) throws IOException {
for (int i = 0; i < size - maxSelectors; i++) {
selectors.add(Selector.open());
}
}
/**
* Decrease <code>Selector</code> pool size
*/
private static void reduce(int size) {
for (int i = 0; i < maxSelectors - size; i++) {
try {
Selector selector = selectors.pop();
selector.close();
} catch (IOException e) {
logger.error("SelectorFactory.reduce", e);
}
}
}
}

94
src/main/java/com/google/code/yanf4j/util/ShiftAndByteBufferMatcher.java Normal file
View File

@@ -0,0 +1,94 @@
/**
* Copyright [2009-2010] [dennis zhuang(killme2008@gmail.com)] Licensed 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
*/
/**
* Copyright [2009-2010] [dennis zhuang(killme2008@gmail.com)] Licensed 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
*/
package com.google.code.yanf4j.util;
import java.util.ArrayList;
import java.util.List;
import com.google.code.yanf4j.buffer.IoBuffer;
/**
* ByteBuffer matcher based on shift-and algorithm
*
* @author dennis
*
*/
public class ShiftAndByteBufferMatcher implements ByteBufferMatcher {
private int[] b;
private int mask;
private int patternLimit;
private int patternPos;
private int patternLen;
public ShiftAndByteBufferMatcher(IoBuffer pat) {
if (pat == null || pat.remaining() == 0) {
throw new IllegalArgumentException("blank buffer");
}
this.patternLimit = pat.limit();
this.patternPos = pat.position();
this.patternLen = pat.remaining();
preprocess(pat);
this.mask = 1 << (this.patternLen - 1);
}
/**
* Ԥ<><D4A4><EFBFBD><EFBFBD>
*
* @param pat
*/
private void preprocess(IoBuffer pat) {
this.b = new int[256];
for (int i = this.patternPos; i < this.patternLimit; i++) {
int p = ByteBufferUtils.uByte(pat.get(i));
this.b[p] = this.b[p] | (1 << i);
}
}
public final List<Integer> matchAll(IoBuffer buffer) {
List<Integer> matches = new ArrayList<Integer>();
int bufferLimit = buffer.limit();
int d = 0;
for (int pos = buffer.position(); pos < bufferLimit; pos++) {
d <<= 1;
d |= 1;
d &= this.b[ByteBufferUtils.uByte(buffer.get(pos))];
if ((d & this.mask) != 0) {
matches.add(pos - this.patternLen + 1);
}
}
return matches;
}
public final int matchFirst(IoBuffer buffer) {
if (buffer == null) {
return -1;
}
int bufferLimit = buffer.limit();
int d = 0;
for (int pos = buffer.position(); pos < bufferLimit; pos++) {
d <<= 1;
d |= 1;
d &= this.b[buffer.get(pos) & 0XFF];
if ((d & this.mask) != 0) {
return pos - this.patternLen + 1;
}
}
return -1;
}
}

93
src/main/java/com/google/code/yanf4j/util/ShiftOrByteBufferMatcher.java Normal file
View File

@@ -0,0 +1,93 @@
/**
* Copyright [2009-2010] [dennis zhuang(killme2008@gmail.com)] Licensed 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
*/
/**
* Copyright [2009-2010] [dennis zhuang(killme2008@gmail.com)] Licensed 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
*/
package com.google.code.yanf4j.util;
import java.util.ArrayList;
import java.util.List;
import com.google.code.yanf4j.buffer.IoBuffer;
/**
* ByteBuffer matcher based on shift-or algorithm
*
* @author dennis
*
*/
public class ShiftOrByteBufferMatcher implements ByteBufferMatcher {
private int[] b;
private int lim;
private int patternLen;
public ShiftOrByteBufferMatcher(IoBuffer pat) {
if (pat == null || pat.remaining() == 0) {
throw new IllegalArgumentException("blank buffer");
}
this.patternLen = pat.remaining();
preprocess(pat);
}
/**
* Ԥ<><D4A4><EFBFBD><EFBFBD>
*
* @param pat
*/
private void preprocess(IoBuffer pat) {
this.b = new int[256];
this.lim = 0;
for (int i = 0; i < 256; i++) {
this.b[i] = ~0;
}
for (int i = 0, j = 1; i < this.patternLen; i++, j <<= 1) {
this.b[ByteBufferUtils.uByte(pat.get(i))] &= ~j;
this.lim |= j;
}
this.lim = ~(this.lim >> 1);
}
public final List<Integer> matchAll(IoBuffer buffer) {
List<Integer> matches = new ArrayList<Integer>();
int bufferLimit = buffer.limit();
int state = ~0;
for (int pos = buffer.position(); pos < bufferLimit; pos++) {
state <<= 1;
state |= this.b[ByteBufferUtils.uByte(buffer.get(pos))];
if (state < this.lim) {
matches.add(pos - this.patternLen + 1);
}
}
return matches;
}
public final int matchFirst(IoBuffer buffer) {
if (buffer == null) {
return -1;
}
int bufferLimit = buffer.limit();
int state = ~0;
for (int pos = buffer.position(); pos < bufferLimit; pos++) {
state = (state <<= 1) | this.b[ByteBufferUtils.uByte(buffer.get(pos))];
if (state < this.lim) {
return pos - this.patternLen + 1;
}
}
return -1;
}
}

45
src/main/java/com/google/code/yanf4j/util/SimpleQueue.java Normal file
View File

@@ -0,0 +1,45 @@
package com.google.code.yanf4j.util;
import java.util.Iterator;
import java.util.LinkedList;
/**
* Simple queue. All methods are thread-safe.
*
* @author dennis zhuang
*/
public class SimpleQueue<T> extends java.util.AbstractQueue<T> {
protected final LinkedList<T> list;
public SimpleQueue(int initializeCapacity) {
this.list = new LinkedList<T>();
}
public SimpleQueue() {
this(100);
}
public synchronized boolean offer(T e) {
return this.list.add(e);
}
public synchronized T peek() {
return this.list.peek();
}
public synchronized T poll() {
return this.list.poll();
}
@Override
public Iterator<T> iterator() {
return this.list.iterator();
}
@Override
public int size() {
return this.list.size();
}
}

149
src/main/java/com/google/code/yanf4j/util/SystemUtils.java Normal file
View File

@@ -0,0 +1,149 @@
/**
* Copyright [2009-2010] [dennis zhuang(killme2008@gmail.com)] Licensed 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
*/
/**
* Copyright [2009-2010] [dennis zhuang(killme2008@gmail.com)] Licensed 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
*/
package com.google.code.yanf4j.util;
import java.io.IOException;
import java.lang.reflect.Method;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.channels.Selector;
import java.nio.channels.spi.SelectorProvider;
import java.util.Queue;
import java.util.Random;
/**
* System utils
*
* @author dennis
*
*/
public final class SystemUtils {
private SystemUtils() {
}
public static final String OS_NAME = System.getProperty("os.name");
private static boolean isLinuxPlatform = false;
static {
if (OS_NAME != null && OS_NAME.toLowerCase().indexOf("linux") >= 0) {
isLinuxPlatform = true;
}
}
public static final String JAVA_VERSION = System.getProperty("java.version");
private static boolean isAfterJava6u4Version = false;
static {
if (JAVA_VERSION != null) {
// java4 or java5
if (JAVA_VERSION.indexOf("1.4.") >= 0 || JAVA_VERSION.indexOf("1.5.") >= 0) {
isAfterJava6u4Version = false;
} else if (JAVA_VERSION.indexOf("1.6.") >= 0) {
int index = JAVA_VERSION.indexOf("_");
if (index > 0) {
String subVersionStr = JAVA_VERSION.substring(index + 1);
if (subVersionStr != null && subVersionStr.length() > 0) {
try {
int subVersion = Integer.parseInt(subVersionStr);
if (subVersion >= 4) {
isAfterJava6u4Version = true;
}
} catch (NumberFormatException e) {
}
}
}
// after java6
} else {
isAfterJava6u4Version = true;
}
}
}
public static final boolean isLinuxPlatform() {
return isLinuxPlatform;
}
public static final boolean isAfterJava6u4Version() {
return isAfterJava6u4Version;
}
public static void main(String[] args) {
System.out.println(isAfterJava6u4Version());
}
public static final int getSystemThreadCount() {
int cpus = getCpuProcessorCount();
if (cpus <= 8) {
return cpus;
} else {
return 8 + (cpus - 8) * 5 / 8;
}
}
public static final int getCpuProcessorCount() {
return Runtime.getRuntime().availableProcessors();
}
public static final Selector openSelector() throws IOException {
Selector result = null;
// check if it is linux os
if (SystemUtils.isLinuxPlatform()) {
try {
Class<?> providerClazz = Class.forName("sun.nio.ch.EPollSelectorProvider");
if (providerClazz != null) {
try {
Method method = providerClazz.getMethod("provider");
if (method != null) {
SelectorProvider selectorProvider = (SelectorProvider) method.invoke(null);
if (selectorProvider != null) {
result = selectorProvider.openSelector();
}
}
} catch (Exception e) {
// ignore
}
}
} catch (Exception e) {
// ignore
}
}
if (result == null) {
result = Selector.open();
}
return result;
}
public static final String getRawAddress(InetSocketAddress inetSocketAddress) {
InetAddress address = inetSocketAddress.getAddress();
return address != null ? address.getHostAddress() : inetSocketAddress.getHostName();
}
public static final Queue<?> createTransferQueue() {
try {
return (Queue<?>) Class.forName("java.util.concurrent.LinkedTransferQueue").newInstance();
} catch (Exception e) {
return new LinkedTransferQueue<Object>();
}
}
public static Random createRandom() {
return new Random();
}
}

101
src/main/java/com/google/code/yanf4j/util/TransferQueue.java Normal file
View File

@@ -0,0 +1,101 @@
/*
* Written by Doug Lea with assistance from members of JCP JSR-166 Expert Group and released to the
* public domain, as explained at http://creativecommons.org/licenses/publicdomain
*/
package com.google.code.yanf4j.util;
import java.util.concurrent.*;
/**
* A {@link BlockingQueue} in which producers may wait for consumers to receive elements. A
* {@code TransferQueue} may be useful for example in message passing applications in which
* producers sometimes (using method {@code transfer}) await receipt of elements by consumers
* invoking {@code take} or {@code poll}, while at other times enqueue elements (via method
* {@code put}) without waiting for receipt. Non-blocking and time-out versions of
* {@code tryTransfer} are also available. A TransferQueue may also be queried via
* {@code hasWaitingConsumer} whether there are any threads waiting for items, which is a converse
* analogy to a {@code peek} operation.
*
* <p>
* Like any {@code BlockingQueue}, a {@code TransferQueue} may be capacity bounded. If so, an
* attempted {@code transfer} operation may initially block waiting for available space, and/or
* subsequently block waiting for reception by a consumer. Note that in a queue with zero capacity,
* such as {@link SynchronousQueue}, {@code put} and {@code transfer} are effectively synonymous.
*
* <p>
* This interface is a member of the
* <a href="{@docRoot} /../technotes/guides/collections/index.html"> Java Collections Framework</a>.
*
* @since 1.7
* @author Doug Lea
* @param <E> the type of elements held in this collection
*/
public interface TransferQueue<E> extends BlockingQueue<E> {
/**
* Transfers the specified element if there exists a consumer already waiting to receive it,
* otherwise returning {@code false} without enqueuing the element.
*
* @param e the element to transfer
* @return {@code true} if the element was transferred, else {@code false}
* @throws ClassCastException if the class of the specified element prevents it from being added
* to this queue
* @throws NullPointerException if the specified element is null
* @throws IllegalArgumentException if some property of the specified element prevents it from
* being added to this queue
*/
boolean tryTransfer(E e);
/**
* Inserts the specified element into this queue, waiting if necessary for space to become
* available and the element to be dequeued by a consumer invoking {@code take} or {@code poll}.
*
* @param e the element to transfer
* @throws InterruptedException if interrupted while waiting, in which case the element is not
* enqueued.
* @throws ClassCastException if the class of the specified element prevents it from being added
* to this queue
* @throws NullPointerException if the specified element is null
* @throws IllegalArgumentException if some property of the specified element prevents it from
* being added to this queue
*/
void transfer(E e) throws InterruptedException;
/**
* Inserts the specified element into this queue, waiting up to the specified wait time if
* necessary for space to become available and the element to be dequeued by a consumer invoking
* {@code take} or {@code poll}.
*
* @param e the element to transfer
* @param timeout how long to wait before giving up, in units of {@code unit}
* @param unit a {@code TimeUnit} determining how to interpret the {@code timeout} parameter
* @return {@code true} if successful, or {@code false} if the specified waiting time elapses
* before completion, in which case the element is not enqueued.
* @throws InterruptedException if interrupted while waiting, in which case the element is not
* enqueued.
* @throws ClassCastException if the class of the specified element prevents it from being added
* to this queue
* @throws NullPointerException if the specified element is null
* @throws IllegalArgumentException if some property of the specified element prevents it from
* being added to this queue
*/
boolean tryTransfer(E e, long timeout, TimeUnit unit) throws InterruptedException;
/**
* Returns {@code true} if there is at least one consumer waiting to dequeue an element via
* {@code take} or {@code poll}. The return value represents a momentary state of affairs.
*
* @return {@code true} if there is at least one waiting consumer
*/
boolean hasWaitingConsumer();
/**
* Returns an estimate of the number of consumers waiting to dequeue elements via {@code take} or
* {@code poll}. The return value is an approximation of a momentary state of affairs, that may be
* inaccurate if consumers have completed or given up waiting. The value may be useful for
* monitoring and heuristics, but not for synchronization control. Implementations of this method
* are likely to be noticeably slower than those for {@link #hasWaitingConsumer}.
*
* @return the number of consumers waiting to dequeue elements
*/
int getWaitingConsumerCount();
}

65
src/main/java/com/google/code/yanf4j/util/WorkerThreadFactory.java Normal file
View File

@@ -0,0 +1,65 @@
/**
* Copyright [2009-2010] [dennis zhuang(killme2008@gmail.com)] Licensed 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
*/
/**
* Copyright [2009-2010] [dennis zhuang(killme2008@gmail.com)] Licensed 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
*/
package com.google.code.yanf4j.util;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
/**
* Thread factory for worker thread
*
* @author dennis
*
*/
public class WorkerThreadFactory implements ThreadFactory {
private static final AtomicInteger poolNumber = new AtomicInteger(1);
private final ThreadGroup group;
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final String namePrefix;
public WorkerThreadFactory(ThreadGroup group, String prefix) {
if (group == null) {
SecurityManager s = System.getSecurityManager();
this.group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
} else {
this.group = group;
}
if (prefix == null) {
this.namePrefix = "pool-" + poolNumber.getAndIncrement() + "-thread-";
} else {
this.namePrefix = prefix + "-" + poolNumber.getAndIncrement() + "-thread-";
}
}
public WorkerThreadFactory(String prefix) {
this(null, prefix);
}
public WorkerThreadFactory() {
this(null, null);
}
public Thread newThread(Runnable r) {
Thread t = new Thread(this.group, r, this.namePrefix + this.threadNumber.getAndIncrement(), 0);
t.setDaemon(true);
if (t.getPriority() != Thread.NORM_PRIORITY) {
t.setPriority(Thread.NORM_PRIORITY);
}
return t;
}
}

10
src/main/java/com/google/code/yanf4j/util/package.html Normal file
View File

@@ -0,0 +1,10 @@
<!--?xml version="1.0" encoding="UTF-8"?--><!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Strict//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1-strict.dtd">
<!-- Copyright (c) 2007 Dustin Sallings <dustin+html@spy.net> -->
<html>
<head>
<title>Yanf4j utilities</title>
</head>
<body>
<h1>Yanf4j utilities</h1>
</body>
</html>