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

  User: belaban 
  Date: 10/08/23 16:28:57

  Modified:    src/org/jgroups/protocols DAISYCHAIN.java
  Log:
  ns
  
  Revision  Changes    Path
  1.7       +21 -15    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.6
  retrieving revision 1.7
  diff -u -r1.6 -r1.7
  --- DAISYCHAIN.java	13 Aug 2010 15:21:13 -0000	1.6
  +++ DAISYCHAIN.java	23 Aug 2010 16:28:57 -0000	1.7
  @@ -22,7 +22,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.6 2010/08/13 15:21:13 belaban Exp $
  + * @version $Id: DAISYCHAIN.java,v 1.7 2010/08/23 16:28:57 belaban Exp $
    */
   @Experimental @Unsupported
   @MBean(description="Protocol just above the transport which disseminates multicasts via daisy chaining")
  @@ -40,13 +40,15 @@
       int forward_queue_max_size=1000;
   
       /* --------------------------------------------- Fields ------------------------------------------------------ */
  -    protected Address local_addr, next;
  -    protected int     view_size=0;
  +    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 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;
   
   
       @ManagedAttribute
  @@ -62,6 +64,11 @@
       public int getSendQueueSize() {return send_queue.size();}
   
   
  +    public void init() throws Exception {
  +        default_pool=getTransport().getDefaultThreadPool();
  +        oob_pool=getTransport().getOOBThreadPool();
  +    }
  +
       public Object down(final Event evt) {
           switch(evt.getType()) {
               case Event.MSG:
  @@ -93,8 +100,7 @@
                       if(log.isTraceEnabled()) log.trace(new StringBuilder("looping back message ").append(msg));
                       msg.setSrc(local_addr);
   
  -                    Executor pool=msg.isFlagSet(Message.OOB)? getTransport().getOOBThreadPool()
  -                            : getTransport().getDefaultThreadPool();
  +                    Executor pool=msg.isFlagSet(Message.OOB)? oob_pool : default_pool;
                       pool.execute(new Runnable() {
                           public void run() {
                               up_prot.up(evt);
  @@ -134,16 +140,16 @@
                   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) {
  +                        try {
  +                            forward_queue.put(copy);
  +                        }
  +                        catch(InterruptedException e) {
  +                        }
  +                        forward();
                       }
  -                    forward();
  -                }
   
                   // 2. Pass up
                   msg.setDest(null);
  @@ -186,7 +192,7 @@
   
   
       public static class DaisyHeader extends Header {
  -        private short ttl;
  +        private short   ttl;
   
           public DaisyHeader() {
           }
  
  
  

------------------------------------------------------------------------------
This SF.net email is sponsored by 

Make an app they can't live without
Enter the BlackBerry Developer Challenge
http://p.sf.net/sfu/RIM-dev2dev