CVS update: JGroups/src/org/jgroups/protocols DAISYCHAIN.java

  User: belaban 
  Date: 10/09/14 12:19:21

  Modified:    src/org/jgroups/protocols DAISYCHAIN.java
  Log:
  ns
  
  Revision  Changes    Path
  1.13      +18 -10    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.12
  retrieving revision 1.13
  diff -u -r1.12 -r1.13
  --- DAISYCHAIN.java	14 Sep 2010 11:55:36 -0000	1.12
  +++ DAISYCHAIN.java	14 Sep 2010 12:19:21 -0000	1.13
  @@ -20,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.12 2010/09/14 11:55:36 belaban Exp $
  + * @version $Id: DAISYCHAIN.java,v 1.13 2010/09/14 12:19:21 belaban Exp $
    */
   @Experimental @Unsupported
   @MBean(description="Protocol just above the transport which disseminates multicasts via daisy chaining")
  @@ -44,7 +44,8 @@
       protected Executor               oob_pool=null;
       protected BlockingQueue<Message> send_queue;
       protected BlockingQueue<Message> forward_queue;
  -    protected boolean                forward=false; // flipped between true and false, to ensure fairness
  +    protected volatile boolean       forward=false; // flipped between true and false, to ensure fairness
  +    protected volatile boolean       running=true;
   
       @ManagedAttribute
       public int msgs_forwarded=0;
  @@ -65,6 +66,16 @@
           forward_queue=new ConcurrentLinkedBlockingQueue<Message>(forward_queue_size);
       }
   
  +    public void start() throws Exception {
  +        super.start();
  +        running=true;
  +    }
  +
  +    public void stop() {
  +        super.stop();
  +        running=false;
  +    }
  +
       public Object down(final Event evt) {
           switch(evt.getType()) {
               case Event.MSG:
  @@ -140,13 +151,9 @@
                       Message copy=msg.copy(true);
                       copy.setDest(next);
                       copy.putHeader(getId(), new DaisyHeader(ttl));
  -                    try {
  -                        msgs_forwarded++;
  -                        forward_queue.put(copy);
  -                    }
  -                    catch(InterruptedException e) {
  -                    }
  -                    processQueues();
  +                    msgs_forwarded++;
  +                    if(forward_queue.offer(copy)) // we don't want incoming threads to block
  +                        processQueues();
                   }
   
                   // 2. Pass up
  @@ -158,7 +165,8 @@
   
   
       protected Object processQueues() {
  -        for(int i=0; i < 10; i++) {
  +        int cnt=0;
  +        while(running && cnt++ < 10000) { // cnt is a second line of defense against loops and should never be used !
               try {
                   Message msg=forward? forward_queue.poll() : send_queue.poll();
                   if(msg == null) {
  
  
  

------------------------------------------------------------------------------
Start uncovering the many advantages of virtual appliances
and start using them to simplify application deployment and
accelerate your shift to cloud computing.
http://p.sf.net/sfu/novell-sfdev2dev