You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@rocketmq.apache.org by GitBox <gi...@apache.org> on 2018/03/09 08:46:51 UTC
[GitHub] fucongchan closed pull request #206: [ROCKETMQ-334] spinlock performance improvement
fucongchan closed pull request #206: [ROCKETMQ-334] spinlock performance improvement
URL: https://github.com/apache/rocketmq/pull/206
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/store/src/main/java/org/apache/rocketmq/store/PutMessageSpinLock.java b/store/src/main/java/org/apache/rocketmq/store/PutMessageSpinLock.java
index 4243da0ad..3f2a6b513 100644
--- a/store/src/main/java/org/apache/rocketmq/store/PutMessageSpinLock.java
+++ b/store/src/main/java/org/apache/rocketmq/store/PutMessageSpinLock.java
@@ -17,25 +17,167 @@
package org.apache.rocketmq.store;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
/**
- * Spin lock Implementation to put message, suggest using this with low race conditions
+ * Spin lock Implementation to put message
*/
public class PutMessageSpinLock implements PutMessageLock {
- //true: Can lock, false : in lock.
- private AtomicBoolean putMessageSpinLock = new AtomicBoolean(true);
+
+ static final class Node {
+ /**
+ * True: Signal, false : Wait.
+ */
+ private final AtomicBoolean waitStatus = new AtomicBoolean(false);
+
+ /**
+ * The thread that enqueued this node. Initialized on
+ * construction and nulled out after use.
+ *
+ * Only for test and trace
+ */
+ volatile Thread thread;
+
+ /**
+ * Link to predecessor node that current node/thread relies on
+ * for checking waitStatus.
+ */
+ volatile Node prev;
+
+ Node(Thread thread) { this.thread = thread; }
+ }
+
+ /**
+ * Tail of the wait queue
+ */
+ private final AtomicReference<Node> tail = new AtomicReference<>();
+
+ /**
+ * Head of the wait queue
+ */
+ private final AtomicReference<Node> head = new AtomicReference<>();
+
+ /**
+ * true: can acquire, false: can not acquire
+ */
+ private final AtomicBoolean status = new AtomicBoolean(true);
+
+ /**
+ * Inserts node into queue, initializing if necessary
+ * @param node the node to insert
+ * @return node's predecessor
+ */
+ private Node enq(final Node node) {
+ for (;;) {
+ Node t = tail.get();
+ if (t == null) { // Must initialize
+ Node initNode = new Node(Thread.currentThread());
+ initNode.waitStatus.set(true);
+ if (head.compareAndSet(null, initNode)) {
+ tail.set(initNode);
+ }
+ } else {
+ node.prev = t;
+ if (tail.compareAndSet(t, node)) {
+ return t;
+ }
+ }
+ }
+ }
+
+ /**
+ * Creates and enqueues node for current thread.
+ *
+ * @return the new node
+ */
+ private Node addWaiter() {
+ Node node = new Node(Thread.currentThread());
+ // Try the fast path of enq; backup to full enq on failure
+ Node pred = tail.get();
+ if (pred != null) {
+ node.prev = pred;
+ if (tail.compareAndSet(pred, node)) {
+ return node;
+ }
+ }
+ enq(node);
+ return node;
+ }
+
+ /**
+ * Sets head of queue to be node, thus dequeuing.
+ * Also nulls out unused fields for sake of GC
+ * @param node the node
+ */
+ private void setHead(Node node) {
+ head.set(node);
+ node.prev = null;
+ node.thread = null;
+ }
+
+
+ public boolean isLocked() { return !status.get(); }
+
+ /**
+ * Queries whether any threads are waiting to acquire.
+ *
+ * @return {@code true} if there may be other threads waiting to acquire
+ */
+ public final boolean hasQueuedThreads() {
+ return head.get() != tail.get();
+ }
+
+ /**
+ * Queries whether any threads have ever contended to acquire this
+ * synchronizer
+ *
+ * @return {@code true} if there has ever been contention
+ */
+ public final boolean hasContended() {
+ return head.get() != null;
+ }
+
+ /**
+ * Returns true if the given thread is currently queued.
+ *
+ * @param thread the thread
+ * @return {@code true} if the given thread is on the queue
+ * @throws NullPointerException if the thread is null
+ */
+ public final boolean isQueued(Thread thread) {
+ if (thread == null)
+ throw new NullPointerException();
+ for (Node p = tail.get(); p != null; p = p.prev)
+ if (p.thread == thread)
+ return true;
+ return false;
+ }
@Override
public void lock() {
- boolean flag;
- do {
- flag = this.putMessageSpinLock.compareAndSet(true, false);
+ if (!status.compareAndSet(true, false)) {
+ Node node = addWaiter();
+ Node pred = node.prev;
+ AtomicBoolean preStatus = pred.waitStatus;
+ for (;;) {
+ if (preStatus.get() && status.compareAndSet(true, false)) {
+ setHead(node);
+ return;
+ }
+ }
}
- while (!flag);
}
@Override
public void unlock() {
- this.putMessageSpinLock.compareAndSet(false, true);
+ status.set(true);
+ Node h = head.get();
+ if (h != null) {
+ /**
+ * notify next waiter node
+ */
+ AtomicBoolean headStatus = h.waitStatus;
+ headStatus.set(true);
+ }
}
}
diff --git a/store/src/test/java/org/apache/rocketmq/store/PutMessageSpinLockTest.java b/store/src/test/java/org/apache/rocketmq/store/PutMessageSpinLockTest.java
new file mode 100644
index 000000000..a79b0df5b
--- /dev/null
+++ b/store/src/test/java/org/apache/rocketmq/store/PutMessageSpinLockTest.java
@@ -0,0 +1,171 @@
+/*
+ * 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.
+ */
+
+package org.apache.rocketmq.store;
+
+import org.junit.Test;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.assertFalse;
+
+public class PutMessageSpinLockTest {
+
+ class HelperRunnable implements Runnable {
+ PutMessageSpinLock lock;
+ volatile boolean canRelease = false;
+
+ HelperRunnable(PutMessageSpinLock lock) {
+ this.lock = lock;
+ }
+
+ public void run() {
+ lock.lock();
+ while (!canRelease) {
+ Thread.yield();
+ }
+ lock.unlock();
+ }
+
+ public void fireRelease() {
+ canRelease = true;
+ }
+ }
+
+ void awaitTermination(Thread t, long timeoutMillis) {
+ try {
+ t.join(timeoutMillis);
+ } catch (InterruptedException fail) {
+ }
+ }
+
+ /**
+ * Spin-waits until sync.isQueued(t) becomes true.
+ */
+ void waitForQueuedThread(PutMessageSpinLock lock, Thread t) {
+ while (!lock.isQueued(t)) {
+ Thread.yield();
+ }
+ assertTrue(t.isAlive());
+ }
+
+ /**
+ * Spin-waits until sync.isQueued(t) becomes false.
+ */
+ void waitForOutQueuedThread(PutMessageSpinLock lock, Thread t) {
+ while (lock.isQueued(t)) {
+ Thread.yield();
+ }
+ }
+
+ @Test
+ public void testIsLocked() {
+ PutMessageSpinLock lock = new PutMessageSpinLock();
+ assertFalse(lock.isLocked());
+ }
+
+ @Test
+ public void testAcquire() {
+ PutMessageSpinLock lock = new PutMessageSpinLock();
+ lock.lock();
+ assertTrue(lock.isLocked());
+ lock.unlock();
+ assertFalse(lock.isLocked());
+ }
+
+ @Test
+ public void testHasQueuedThreads() {
+ PutMessageSpinLock lock = new PutMessageSpinLock();
+ assertFalse(lock.hasQueuedThreads());
+ lock.lock();
+ HelperRunnable r1 = new HelperRunnable(lock);
+ Thread t1 = new Thread(r1);
+ t1.start();
+ waitForQueuedThread(lock, t1);
+ assertTrue(lock.hasQueuedThreads());
+ HelperRunnable r2 = new HelperRunnable(lock);
+ Thread t2 = new Thread(r2);
+ t2.start();
+ waitForQueuedThread(lock, t2);
+ assertTrue(lock.hasQueuedThreads());
+ lock.unlock();
+ waitForOutQueuedThread(lock, t1);
+ assertTrue(lock.hasQueuedThreads());
+ r1.fireRelease();
+ awaitTermination(t1, 100);
+ waitForOutQueuedThread(lock, t2);
+ r2.fireRelease();
+ awaitTermination(t2, 100);
+ assertFalse(lock.hasQueuedThreads());
+ }
+
+ @Test
+ public void testIsQueued() {
+ PutMessageSpinLock lock = new PutMessageSpinLock();
+ HelperRunnable r1 = new HelperRunnable(lock);
+ Thread t1 = new Thread(r1);
+ HelperRunnable r2 = new HelperRunnable(lock);
+ Thread t2 = new Thread(r2);
+ assertFalse(lock.isQueued(t1));
+ assertFalse(lock.isQueued(t2));
+ lock.lock();
+ t1.start();
+ waitForQueuedThread(lock, t1);
+ assertTrue(lock.isQueued(t1));
+ assertFalse(lock.isQueued(t2));
+ t2.start();
+ waitForQueuedThread(lock, t2);
+ assertTrue(lock.isQueued(t1));
+ assertTrue(lock.isQueued(t2));
+ lock.unlock();
+ waitForOutQueuedThread(lock, t1);
+ assertFalse(lock.isQueued(t1));
+ assertTrue(lock.isQueued(t2));
+ r1.fireRelease();
+ awaitTermination(t1, 100);
+ waitForOutQueuedThread(lock, t2);
+ assertFalse(lock.isQueued(t1));
+ assertFalse(lock.isQueued(t2));
+ r2.fireRelease();
+ awaitTermination(t2, 100);
+ }
+
+ @Test
+ public void testHasContended() {
+ PutMessageSpinLock lock = new PutMessageSpinLock();
+ assertFalse(lock.hasContended());
+ lock.lock();
+ assertFalse(lock.hasContended());
+ HelperRunnable r1 = new HelperRunnable(lock);
+ Thread t1 = new Thread(r1);
+ t1.start();
+ waitForQueuedThread(lock, t1);
+ assertTrue(lock.hasContended());
+ HelperRunnable r2 = new HelperRunnable(lock);
+ Thread t2 = new Thread(r2);
+ t2.start();
+ waitForQueuedThread(lock, t2);
+ assertTrue(lock.hasContended());
+ lock.unlock();
+ waitForOutQueuedThread(lock, t1);
+ r1.fireRelease();
+ awaitTermination(t1, 100);
+ assertTrue(lock.hasContended());
+ waitForOutQueuedThread(lock, t2);
+ r2.fireRelease();
+ awaitTermination(t2, 100);
+ assertTrue(lock.hasContended());
+ }
+}
\ No newline at end of file
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services