CVS update: JGroups/src/org/jgroups/protocols DAISYCHAIN.java
"Bela Ban" <[email protected]>
| Newsgroups | gmane.comp.java.javagroups.cvs |
|---|---|
| Message-ID | <[email protected]> |
User: belaban
Date: 10/08/24 10:07:53
Modified: src/org/jgroups/protocols DAISYCHAIN.java
Log:
made non-blocking
Revision Changes Path
1.9 +37 -47 JGroups/src/org/jgroups/protocols/DAISYCHAIN.java
Index: DAISYCHAIN.java
===================================================================
RCS file: /cvsroot/javagroups/JGroups/src/org/jgroups/protocols/DAISYCHAIN.java,v
retrieving revision 1.8
retrieving revision 1.9
diff -u -r1.8 -r1.9
--- DAISYCHAIN.java 23 Aug 2010 16:54:38 -0000 1.8
+++ DAISYCHAIN.java 24 Aug 2010 10:07:53 -0000 1.9
@@ -3,16 +3,14 @@
import org.jgroups.*;
import org.jgroups.annotations.*;
import org.jgroups.stack.Protocol;
-import org.jgroups.util.ConcurrentLinkedBlockingQueue;
import org.jgroups.util.Util;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
-import java.util.concurrent.BlockingQueue;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
/**
* Implementation of daisy chaining. Multicast messages are sent to our neighbor, which sends them to its neighbor etc.
@@ -22,7 +20,7 @@
* send another message. This leads to much better throughput, see the ref in the JIRA.<p/>
* JIRA: https://jira.jboss.org/browse/JGRP-1021
* @author Bela Ban
- * @version $Id: DAISYCHAIN.java,v 1.8 2010/08/23 16:54:38 belaban Exp $
+ * @version $Id: DAISYCHAIN.java,v 1.9 2010/08/24 10:07:53 belaban Exp $
*/
@Experimental @Unsupported
@MBean(description="Protocol just above the transport which disseminates multicasts via daisy chaining")
@@ -40,15 +38,13 @@
int forward_queue_max_size=1000;
/* --------------------------------------------- Fields ------------------------------------------------------ */
- protected Address local_addr, next;
- protected int view_size=0;
-
- protected final BlockingQueue<Message> send_queue=new ConcurrentLinkedBlockingQueue<Message>(send_queue_max_size);
- protected final BlockingQueue<Message> forward_queue=new ConcurrentLinkedBlockingQueue<Message>(forward_queue_max_size);
- protected boolean forward=false; // flipped between true and false, to ensure fairness
- protected final Lock lock=new ReentrantLock();
- protected Executor default_pool=null;
- protected Executor oob_pool=null;
+ protected Address local_addr, next;
+ protected int view_size=0;
+ protected final Queue<Message> send_queue=new ConcurrentLinkedQueue<Message>();
+ protected final Queue<Message> forward_queue=new ConcurrentLinkedQueue<Message>();
+ protected volatile boolean forward=false; // flipped between true and false, to ensure fairness
+ protected Executor default_pool=null;
+ protected Executor oob_pool=null;
@ManagedAttribute
@@ -87,14 +83,7 @@
DaisyHeader hdr=new DaisyHeader(hdr_ttl);
copy.setDest(next);
copy.putHeader(getId(), hdr);
-
- try {
- send_queue.put(copy);
- }
- catch(InterruptedException e) {
- Thread.currentThread().interrupt();
- return null;
- }
+ send_queue.offer(copy);
if(loopback) {
if(log.isTraceEnabled()) log.trace(new StringBuilder("looping back message ").append(msg));
@@ -140,16 +129,12 @@
if(log.isTraceEnabled())
log.trace(local_addr + ": received message from " + msg.getSrc() + " with ttl=" + ttl);
if(--ttl > 0) {
- Message copy=msg.copy(true);
- copy.setDest(next);
+ Message copy=msg.copy(true);
+ copy.setDest(next);
copy.putHeader(getId(), new DaisyHeader(ttl));
- try {
- forward_queue.put(copy);
- }
- catch(InterruptedException e) {
- }
- forward();
- }
+ forward_queue.offer(copy);
+ forward();
+ }
// 2. Pass up
msg.setDest(null);
@@ -160,30 +145,35 @@
protected Object forward() {
- Message msg=null;
-
- lock.lock();
- try {
- String tmp=forward? " forwarding" : " sending";
- msg=forward? forward_queue.poll() : send_queue.poll();
- if(msg == null) {
- msg=forward? send_queue.poll() : forward_queue.poll();
+ Message msg=null;
+ Queue<Message> queue=null;
+ String tmp=null;
+
+ while(!(send_queue.isEmpty() && forward_queue.isEmpty())) {
+ tmp=forward? " forwarding" : " sending";
+ queue=forward? forward_queue : send_queue;
+ if(queue.isEmpty()) {
+ queue=forward? send_queue : forward_queue;
msgs_sent++;
}
else {
msgs_forwarded++;
}
- if(log.isTraceEnabled()) {
- DaisyHeader hdr=(DaisyHeader)msg.getHeader(getId());
- log.trace(local_addr + ": " + tmp + " message with ttl=" + hdr.getTTL() + " to " + next);
- }
- }
- finally {
+
+ msg=queue.poll();
+ if(msg != null)
+ break;
forward=!forward;
- lock.unlock();
}
- return msg != null? down_prot.down(new Event(Event.MSG, msg)) : null;
+ if(msg == null)
+ return null;
+
+ if(log.isTraceEnabled()) {
+ DaisyHeader hdr=(DaisyHeader)msg.getHeader(getId());
+ log.trace(local_addr + ": " + tmp + " message with ttl=" + hdr.getTTL() + " to " + next);
+ }
+ return down_prot.down(new Event(Event.MSG, msg));
}
------------------------------------------------------------------------------
Sell apps to millions through the Intel(R) Atom(Tm) Developer Program
Be part of this innovative community and reach millions of netbook users
worldwide. Take advantage of special opportunities to increase revenue and
speed time-to-market. Join now, and jumpstart your future.
http://p.sf.net/sfu/intel-atom-d2d