<commit_before>package org.jgroups.tests;

import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Semaphore;

import junit.framework.Test;
import junit.framework.TestCase;
import junit.framework.TestSuite;

import org.jgroups.Address;
import org.jgroups.Channel;
import org.jgroups.JChannelFactory;
import org.jgroups.Message;
import org.jgroups.util.Util;


/**
 * Tests streaming state transfer.
 * 
 * 
 * @author <NAME>
 * @version $Id$ 
 *
 */
public class StreamingStateTransferTest extends ChannelTestBase
{
   
   
   public void setUp() throws Exception
   {
      super.setUp();         
      CHANNEL_CONFIG = System.getProperty("channel.conf.flush", "flush-udp.xml");      
   }
   
   public boolean useBlocking()
   {
      return true;
   }
   
   public void testTransfer()
   {
      String channelNames [] = null;
      //mux applications on top of same channel have to have unique name
      if(isMuxChannelUsed())
      {
         channelNames = createMuxApplicationNames(1);         
      }
      else
      {
         channelNames = new String[]{"A", "B", "C", "D"};        
      }
      transferHelper(channelNames,false);
   }
   
   public void testRpcChannelTransfer()
   {
      //do this test for regular channels only
      if(!isMuxChannelUsed())
      {
         String channelNames []= new String[]{"A", "B", "C", "D"};   
         transferHelper(channelNames,true);
      }
   }
   
   public void testMultipleServiceMuxChannel()
   {
      String channelNames [] = null;
      //mux applications on top of same channel have to have unique name
      if(isMuxChannelUsed())
      {
         channelNames = createMuxApplicationNames(2);  
         transferHelper(channelNames,false);
      }           
   }
   
   public void testStateTransferWithCoordinatorCrash()
   {
      String channelNames [] = null;
      //mux applications on top of same channel have to have unique name
      if(isMuxChannelUsed())
      {
         channelNames = createMuxApplicationNames(1);         
      }
      else
      {
         channelNames = new String[]{"A", "B", "C", "D", "E", "F"};        
      }
      transferHelper(channelNames,true,false,false);      
   }
   
   public void testStateTransferWithCoordinatorCrashLargeState()
   {
      String channelNames [] = null;
      //mux applications on top of same channel have to have unique name
      if(isMuxChannelUsed())
      {
         channelNames = createMuxApplicationNames(1);         
      }
      else
      {
         channelNames = new String[]{"A", "B", "C", "D", "E"};        
      }
      transferHelper(channelNames,true,true,false);      
   }
   
   public void transferHelper(String channelNames[], boolean useDispatcher)
   {
      transferHelper(channelNames,false,false,useDispatcher);
   }

   public void transferHelper(String channelNames[],boolean crash, boolean largeTransfer,boolean useDispatcher)
   {                 
      int channelCount = channelNames.length;
      ArrayList channels = new ArrayList(channelCount);      

      //Create a semaphore and take all its tickets
      Semaphore semaphore = new Semaphore(channelCount);
      
      try
      {

         takeAllPermits(semaphore, channelCount);
         boolean crashed = false;
         // Create activation threads that will block on the semaphore        
         for (int i = 0; i < channelCount; i++)
         {
            StreamingStateTransferApplication channel = null;
            if(isMuxChannelUsed())
            {
               channel = new StreamingStateTransferApplication(channelNames[i],muxFactory[i%getMuxFactoryCount()],semaphore,largeTransfer);  
            }
            else
            {
               channel = new StreamingStateTransferApplication(channelNames[i],semaphore,useDispatcher,largeTransfer);
            }

            // Start threads and let them join the channel
            channels.add(channel);
            semaphore.release(1);
            channel.start();            
            sleepThread(2000);
            
            if(crash && !crashed && i>2)
            {
               StreamingStateTransferApplication coord = (StreamingStateTransferApplication) channels.remove(0);
               coord.cleanup();
               crashed = true;
            }
         }               
         
         
         if(isMuxChannelUsed())
         {
            blockUntilViewsReceived(channels,getMuxFactoryCount(), 60000);
         }
         else
         {            
            blockUntilViewsReceived(channels, 60000);            
         }
         

         //Reacquire the semaphore tickets; when we have them all
         // we know the threads are done         
         acquireSemaphore(semaphore, 60000, channelCount);          
         
         int getStateInvokedCount = 0;
         int setStateInvokedCount = 0;
         int partialGetStateInvokedCount = 0;
         int partialSetStateInvokedCount = 0;
         
         sleepThread(3000);
         for (int i = 0; i < channels.size(); i++)
         {
            StreamingStateTransferApplication current = (StreamingStateTransferApplication) channels.get(i);
            if(current.getStateInvoked)
            {
               getStateInvokedCount++;
            }
            if(current.setStateInvoked)
            {
               setStateInvokedCount++;
            }  
            if(current.partialGetStateInvoked)
            {
               partialGetStateInvokedCount++;
            } 
            if(current.partialSetStateInvoked)
            {
               partialSetStateInvokedCount++;
            }
            Map map = current.getMap();
            for (int j = 0; j < channels.size(); j++)
            {
               StreamingStateTransferApplication app = (StreamingStateTransferApplication) channels.get(j);
               List l = (List) map.get(app.getLocalAddress());
               int size = l!=null?l.size():0;
               assertEquals("Correct element count in map ",StreamingStateTransferApplication.COUNT,size);              
            }
         }
         if(isMuxChannelUsed())
         {
            int factor = channelCount/getMuxFactoryCount();
            assertEquals("Correct invocation count of getState ",1*factor, getStateInvokedCount);
            assertEquals("Correct invocation count of setState ",(channelCount/factor)-1,setStateInvokedCount/factor);
            assertEquals("Correct invocation count of partial getState ",1*factor, partialGetStateInvokedCount);
            assertEquals("Correct invocation count of partial setState ",(channelCount/factor)-1,partialSetStateInvokedCount/factor);
         }
         else
         {
            assertEquals("Correct invocation count of getState ",1, getStateInvokedCount);
            assertEquals("Correct invocation count of setState ",channelCount-1,setStateInvokedCount);
            assertEquals("Correct invocation count of partial getState ",1, partialGetStateInvokedCount);
            assertEquals("Correct invocation count of partial setState ",channelCount-1,partialSetStateInvokedCount);
         }
               
      }
      catch (Exception ex)
      {
         log.warn(ex);
      }
      finally
      {
         for (int i = 0; i < channels.size(); i++)
         {
            StreamingStateTransferApplication app = (StreamingStateTransferApplication) channels.get(i);
            sleepThread(500);
            app.cleanup();
         }
      }
   }
   
   protected class StreamingChannelTestFactory extends DefaultChannelTestFactory
   {
      public Channel createChannel(Object id) throws Exception
      {
         return createChannel(CHANNEL_CONFIG, true);
      }
   }

   protected class StreamingStateTransferApplication extends PushChannelApplicationWithSemaphore
   {            
      private final Map stateMap = new HashMap();
      
      public static final int COUNT = 25;      
      
      boolean partialSetStateInvoked = false;

      boolean partialGetStateInvoked = false;

      boolean setStateInvoked = false;

      boolean getStateInvoked = false;
      
      boolean largeTransfer = false;

      public StreamingStateTransferApplication(String name, Semaphore s,boolean useDispatcher,boolean largeTransfer) throws Exception
      {
         super(name,new StreamingChannelTestFactory(),s,useDispatcher);
         this.largeTransfer = largeTransfer;
         channel.connect("test");         
      }    
      
      public StreamingStateTransferApplication(String name, JChannelFactory factory,Semaphore s,boolean largeTransfer) throws Exception
      {
         super(name,factory,s);
         this.largeTransfer = largeTransfer;
         channel.connect("test");
      } 

      public void receive(Message msg)
      {
         Address sender = msg.getSrc();
         synchronized(stateMap)
         {
            List list = (List) stateMap.get(sender);
            if(list == null)
            {
               list = new ArrayList();
               stateMap.put(sender, list);
            }
            list.add(msg.getObject());
         }
      }
      
      public Map getMap()
      {
         return stateMap;
      }

      public void useChannel() throws Exception
      {        
         for(int i = 0;i < COUNT;i++)
         {
            channel.send(null,null,new Integer(i));
         }
         channel.getState(null, 25000);                         
         channel.getState(null, name, 25000);
      }

      public void getState(OutputStream ostream)
      {
         if(largeTransfer)
            sleepThread(4000);
         
         super.getState(ostream);
         ObjectOutputStream oos = null;
         try
         {
            oos = new ObjectOutputStream(ostream);
            HashMap copy = null;
            synchronized (stateMap)
            {
               copy = new HashMap(stateMap);
            }              
            oos.writeObject(copy);                         
            oos.flush();
         }
         catch (IOException e)
         {
            e.printStackTrace();
         }
         finally
         {
            getStateInvoked = true;
            Util.close(oos);
         }
      }
      
      public byte[] getState()
      {
         if(largeTransfer)
            sleepThread(4000);
         
         byte[] result = null;
         try
         {
            synchronized (stateMap)
            {
               result = Util.objectToByteBuffer(stateMap);
            }
         }
         catch (Exception e)
         {            
            e.printStackTrace();
         }
         finally
         {
            getStateInvoked = true;           
         }
         return result;
      }
      
      public void setState(byte [] state)
      {
         if(largeTransfer)
            sleepThread(4000);
         
         Map result = null;
         try
         {
            result = (Map) Util.objectFromByteBuffer(state);
         }
         catch (Exception e)
         {            
            e.printStackTrace();
         }
         finally
         {
            setStateInvoked = true;           
         }
         synchronized(stateMap)
         {
            stateMap.clear();
            stateMap.putAll(result);
         }
      }

      public void setState(InputStream istream)
      {
         if(largeTransfer)
            sleepThread(4000);
         
         super.setState(istream);
         ObjectInputStream ois = null;
         try
         {
            ois = new ObjectInputStream(istream);
            Map map = (Map) ois.readObject();
            synchronized (stateMap)
            {
               stateMap.clear();
               stateMap.putAll(map);
            }           
         }
         catch (Exception e)
         {
            e.printStackTrace();
         }
         finally
         {
            setStateInvoked = true;
            Util.close(ois);
         }
      }
      
      public void setState(String stateId,byte [] state)
      {
         if(largeTransfer)
            sleepThread(4000);
         
         Object nameTransfer = null;         
         try
         {
            nameTransfer = Util.objectFromByteBuffer(state);
            TestCase.assertEquals("Got partial state requested ", nameTransfer, name);
         }
         catch (Exception e)
         {            
            e.printStackTrace();
         }
         finally
         {
            partialSetStateInvoked = true;            
         }
      }
      
      public byte[] getState(String stateId)
      {
         if(largeTransfer)
            sleepThread(4000);
         
         byte[] result = null;
         try
         {
            result = Util.objectToByteBuffer(stateId);
         }
         catch (Exception e)
         {
            e.printStackTrace();
         }
         finally
         {
            partialGetStateInvoked = true;            
         }
         return result;
      }

      public void setState(String state_id, InputStream istream)
      {
         if(largeTransfer)
            sleepThread(4000);
         
         super.setState(state_id, istream);
         ObjectInputStream ois = null;
         try
         {
            ois = new ObjectInputStream(istream);            
            TestCase.assertEquals("Got partial state requested ", ois.readObject(), name);                       
         }
         catch (Exception e)
         {
            e.printStackTrace();
         }
         finally
         {
            partialSetStateInvoked = true;
            Util.close(ois);
         }
      }

      public void getState(String state_id, OutputStream ostream)
      {
         if(largeTransfer)
            sleepThread(4000);
         
         super.getState(state_id, ostream);
         ObjectOutputStream oos = null;
         try
         {
            oos = new ObjectOutputStream(ostream);
            oos.writeObject(state_id);            
            oos.flush();
         }
         catch (IOException e)
         {
            e.printStackTrace();
         }
         finally
         {
            partialGetStateInvoked  = true;
            Util.close(oos);
         }
      }        
   }

   public static Test suite()
   {
      return new TestSuite(StreamingStateTransferTest.class);
   }

   public static void main(String[] args)
   {
      String[] testCaseName = {StreamingStateTransferTest.class.getName()};
      junit.textui.TestRunner.main(testCaseName);
   }
}
<commit_msg>remove testing of state transfer during node crash (for now)<commit_after>package org.jgroups.tests;

import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Semaphore;

import junit.framework.Test;
import junit.framework.TestCase;
import junit.framework.TestSuite;

import org.jgroups.Address;
import org.jgroups.Channel;
import org.jgroups.JChannelFactory;
import org.jgroups.Message;
import org.jgroups.util.Util;


/**
 * Tests streaming state transfer.
 * 
 * 
 * @author <NAME>
 * @version $Id$ 
 *
 */
public class StreamingStateTransferTest extends ChannelTestBase
{
   
   
   public void setUp() throws Exception
   {
      super.setUp();         
      CHANNEL_CONFIG = System.getProperty("channel.conf.flush", "flush-udp.xml");      
   }
   
   public boolean useBlocking()
   {
      return true;
   }
   
   public void testTransfer()
   {
      String channelNames [] = null;
      //mux applications on top of same channel have to have unique name
      if(isMuxChannelUsed())
      {
         channelNames = createMuxApplicationNames(1);         
      }
      else
      {
         channelNames = new String[]{"A", "B", "C", "D"};        
      }
      transferHelper(channelNames,false);
   }
   
   public void testRpcChannelTransfer()
   {
      //do this test for regular channels only
      if(!isMuxChannelUsed())
      {
         String channelNames []= new String[]{"A", "B", "C", "D"};   
         transferHelper(channelNames,true);
      }
   }
   
   public void testMultipleServiceMuxChannel()
   {
      String channelNames [] = null;
      //mux applications on top of same channel have to have unique name
      if(isMuxChannelUsed())
      {
         channelNames = createMuxApplicationNames(2);  
         transferHelper(channelNames,false);
      }           
   }
   
   public void transferHelper(String channelNames[], boolean useDispatcher)
   {
      transferHelper(channelNames,false,false,useDispatcher);
   }

   public void transferHelper(String channelNames[],boolean crash, boolean largeTransfer,boolean useDispatcher)
   {                 
      int channelCount = channelNames.length;
      ArrayList channels = new ArrayList(channelCount);      

      //Create a semaphore and take all its tickets
      Semaphore semaphore = new Semaphore(channelCount);
      
      try
      {

         takeAllPermits(semaphore, channelCount);
         boolean crashed = false;
         // Create activation threads that will block on the semaphore        
         for (int i = 0; i < channelCount; i++)
         {
            StreamingStateTransferApplication channel = null;
            if(isMuxChannelUsed())
            {
               channel = new StreamingStateTransferApplication(channelNames[i],muxFactory[i%getMuxFactoryCount()],semaphore,largeTransfer);  
            }
            else
            {
               channel = new StreamingStateTransferApplication(channelNames[i],semaphore,useDispatcher,largeTransfer);
            }

            // Start threads and let them join the channel
            channels.add(channel);
            semaphore.release(1);
            channel.start();            
            sleepThread(2000);
            
            if(crash && !crashed && i>2)
            {
               StreamingStateTransferApplication coord = (StreamingStateTransferApplication) channels.remove(0);
               coord.cleanup();
               crashed = true;
            }
         }               
         
         
         if(isMuxChannelUsed())
         {
            blockUntilViewsReceived(channels,getMuxFactoryCount(), 60000);
         }
         else
         {            
            blockUntilViewsReceived(channels, 60000);            
         }
         

         //Reacquire the semaphore tickets; when we have them all
         // we know the threads are done         
         acquireSemaphore(semaphore, 60000, channelCount);          
         
         int getStateInvokedCount = 0;
         int setStateInvokedCount = 0;
         int partialGetStateInvokedCount = 0;
         int partialSetStateInvokedCount = 0;
         
         sleepThread(3000);
         for (int i = 0; i < channels.size(); i++)
         {
            StreamingStateTransferApplication current = (StreamingStateTransferApplication) channels.get(i);
            if(current.getStateInvoked)
            {
               getStateInvokedCount++;
            }
            if(current.setStateInvoked)
            {
               setStateInvokedCount++;
            }  
            if(current.partialGetStateInvoked)
            {
               partialGetStateInvokedCount++;
            } 
            if(current.partialSetStateInvoked)
            {
               partialSetStateInvokedCount++;
            }
            Map map = current.getMap();
            for (int j = 0; j < channels.size(); j++)
            {
               StreamingStateTransferApplication app = (StreamingStateTransferApplication) channels.get(j);
               List l = (List) map.get(app.getLocalAddress());
               int size = l!=null?l.size():0;
               assertEquals("Correct element count in map ",StreamingStateTransferApplication.COUNT,size);              
            }
         }
         if(isMuxChannelUsed())
         {
            int factor = channelCount/getMuxFactoryCount();
            assertEquals("Correct invocation count of getState ",1*factor, getStateInvokedCount);
            assertEquals("Correct invocation count of setState ",(channelCount/factor)-1,setStateInvokedCount/factor);
            assertEquals("Correct invocation count of partial getState ",1*factor, partialGetStateInvokedCount);
            assertEquals("Correct invocation count of partial setState ",(channelCount/factor)-1,partialSetStateInvokedCount/factor);
         }
         else
         {
            assertEquals("Correct invocation count of getState ",1, getStateInvokedCount);
            assertEquals("Correct invocation count of setState ",channelCount-1,setStateInvokedCount);
            assertEquals("Correct invocation count of partial getState ",1, partialGetStateInvokedCount);
            assertEquals("Correct invocation count of partial setState ",channelCount-1,partialSetStateInvokedCount);
         }
               
      }
      catch (Exception ex)
      {
         log.warn(ex);
      }
      finally
      {
         for (int i = 0; i < channels.size(); i++)
         {
            StreamingStateTransferApplication app = (StreamingStateTransferApplication) channels.get(i);
            sleepThread(500);
            app.cleanup();
         }
      }
   }
   
   protected class StreamingChannelTestFactory extends DefaultChannelTestFactory
   {
      public Channel createChannel(Object id) throws Exception
      {
         return createChannel(CHANNEL_CONFIG, true);
      }
   }

   protected class StreamingStateTransferApplication extends PushChannelApplicationWithSemaphore
   {            
      private final Map stateMap = new HashMap();
      
      public static final int COUNT = 25;      
      
      boolean partialSetStateInvoked = false;

      boolean partialGetStateInvoked = false;

      boolean setStateInvoked = false;

      boolean getStateInvoked = false;
      
      boolean largeTransfer = false;

      public StreamingStateTransferApplication(String name, Semaphore s,boolean useDispatcher,boolean largeTransfer) throws Exception
      {
         super(name,new StreamingChannelTestFactory(),s,useDispatcher);
         this.largeTransfer = largeTransfer;
         channel.connect("test");         
      }    
      
      public StreamingStateTransferApplication(String name, JChannelFactory factory,Semaphore s,boolean largeTransfer) throws Exception
      {
         super(name,factory,s);
         this.largeTransfer = largeTransfer;
         channel.connect("test");
      } 

      public void receive(Message msg)
      {
         Address sender = msg.getSrc();
         synchronized(stateMap)
         {
            List list = (List) stateMap.get(sender);
            if(list == null)
            {
               list = new ArrayList();
               stateMap.put(sender, list);
            }
            list.add(msg.getObject());
         }
      }
      
      public Map getMap()
      {
         return stateMap;
      }

      public void useChannel() throws Exception
      {        
         for(int i = 0;i < COUNT;i++)
         {
            channel.send(null,null,new Integer(i));
         }
         channel.getState(null, 25000);                         
         channel.getState(null, name, 25000);
      }

      public void getState(OutputStream ostream)
      {
         if(largeTransfer)
            sleepThread(4000);
         
         super.getState(ostream);
         ObjectOutputStream oos = null;
         try
         {
            oos = new ObjectOutputStream(ostream);
            HashMap copy = null;
            synchronized (stateMap)
            {
               copy = new HashMap(stateMap);
            }              
            oos.writeObject(copy);                         
            oos.flush();
         }
         catch (IOException e)
         {
            e.printStackTrace();
         }
         finally
         {
            getStateInvoked = true;
            Util.close(oos);
         }
      }
      
      public byte[] getState()
      {
         if(largeTransfer)
            sleepThread(4000);
         
         byte[] result = null;
         try
         {
            synchronized (stateMap)
            {
               result = Util.objectToByteBuffer(stateMap);
            }
         }
         catch (Exception e)
         {            
            e.printStackTrace();
         }
         finally
         {
            getStateInvoked = true;           
         }
         return result;
      }
      
      public void setState(byte [] state)
      {
         if(largeTransfer)
            sleepThread(4000);
         
         Map result = null;
         try
         {
            result = (Map) Util.objectFromByteBuffer(state);
         }
         catch (Exception e)
         {            
            e.printStackTrace();
         }
         finally
         {
            setStateInvoked = true;           
         }
         synchronized(stateMap)
         {
            stateMap.clear();
            stateMap.putAll(result);
         }
      }

      public void setState(InputStream istream)
      {
         if(largeTransfer)
            sleepThread(4000);
         
         super.setState(istream);
         ObjectInputStream ois = null;
         try
         {
            ois = new ObjectInputStream(istream);
            Map map = (Map) ois.readObject();
            synchronized (stateMap)
            {
               stateMap.clear();
               stateMap.putAll(map);
            }           
         }
         catch (Exception e)
         {
            e.printStackTrace();
         }
         finally
         {
            setStateInvoked = true;
            Util.close(ois);
         }
      }
      
      public void setState(String stateId,byte [] state)
      {
         if(largeTransfer)
            sleepThread(4000);
         
         Object nameTransfer = null;         
         try
         {
            nameTransfer = Util.objectFromByteBuffer(state);
            TestCase.assertEquals("Got partial state requested ", nameTransfer, name);
         }
         catch (Exception e)
         {            
            e.printStackTrace();
         }
         finally
         {
            partialSetStateInvoked = true;            
         }
      }
      
      public byte[] getState(String stateId)
      {
         if(largeTransfer)
            sleepThread(4000);
         
         byte[] result = null;
         try
         {
            result = Util.objectToByteBuffer(stateId);
         }
         catch (Exception e)
         {
            e.printStackTrace();
         }
         finally
         {
            partialGetStateInvoked = true;            
         }
         return result;
      }

      public void setState(String state_id, InputStream istream)
      {
         if(largeTransfer)
            sleepThread(4000);
         
         super.setState(state_id, istream);
         ObjectInputStream ois = null;
         try
         {
            ois = new ObjectInputStream(istream);            
            TestCase.assertEquals("Got partial state requested ", ois.readObject(), name);                       
         }
         catch (Exception e)
         {
            e.printStackTrace();
         }
         finally
         {
            partialSetStateInvoked = true;
            Util.close(ois);
         }
      }

      public void getState(String state_id, OutputStream ostream)
      {
         if(largeTransfer)
            sleepThread(4000);
         
         super.getState(state_id, ostream);
         ObjectOutputStream oos = null;
         try
         {
            oos = new ObjectOutputStream(ostream);
            oos.writeObject(state_id);            
            oos.flush();
         }
         catch (IOException e)
         {
            e.printStackTrace();
         }
         finally
         {
            partialGetStateInvoked  = true;
            Util.close(oos);
         }
      }        
   }

   public static Test suite()
   {
      return new TestSuite(StreamingStateTransferTest.class);
   }

   public static void main(String[] args)
   {
      String[] testCaseName = {StreamingStateTransferTest.class.getName()};
      junit.textui.TestRunner.main(testCaseName);
   }
}

{'fraction_non_alphanumeric': 0.08418518647734012, 'fraction_numerical': 0.00655897847399086, 'mean_word_length': 1.5756132626471382, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5661955', 'n_tokens_mistral': 7301, 'n_tokens_neox': 7105, 'n_words': 1526}