API tools faq
paste
Guest User

Untitled

a guest
Jul 16th, 2012
31
0
Never
Add comment
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
Java 17.18 KB | None | 0 0
raw download clone embed print report
  1. /**
  2.  * Licensed to the Apache Software Foundation (ASF) under one
  3.  * or more contributor license agreements.  See the NOTICE file
  4.  * distributed with this work for additional information
  5.  * regarding copyright ownership.  The ASF licenses this file
  6.  * to you under the Apache License, Version 2.0 (the
  7.  * "License"); you may not use this file except in compliance
  8.  * with the License.  You may obtain a copy of the License at
  9.  *
  10.  *     http://www.apache.org/licenses/LICENSE-2.0
  11.  *
  12.  * Unless required by applicable law or agreed to in writing, software
  13.  * distributed under the License is distributed on an "AS IS" BASIS,
  14.  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15.  * See the License for the specific language governing permissions and
  16.  * limitations under the License.
  17.  */
  18. package org.apache.hama.bsp;
  19.  
  20. import java.io.IOException;
  21. import java.net.InetAddress;
  22. import java.net.InetSocketAddress;
  23. import java.util.Iterator;
  24. import java.util.List;
  25. import java.util.Map.Entry;
  26.  
  27. import junit.framework.TestCase;
  28.  
  29. import org.apache.commons.logging.Log;
  30. import org.apache.commons.logging.LogFactory;
  31. import org.apache.hadoop.conf.Configuration;
  32. import org.apache.hadoop.fs.FSDataInputStream;
  33. import org.apache.hadoop.fs.FileSystem;
  34. import org.apache.hadoop.fs.Path;
  35. import org.apache.hadoop.io.NullWritable;
  36. import org.apache.hadoop.io.Text;
  37. import org.apache.hadoop.io.Writable;
  38. import org.apache.hadoop.ipc.RPC;
  39. import org.apache.hadoop.ipc.Server;
  40. import org.apache.hama.Constants;
  41. import org.apache.hama.HamaConfiguration;
  42. import org.apache.hama.bsp.Counters.Counter;
  43. import org.apache.hama.bsp.TestBSPTaskFaults.MinimalGroomServer;
  44. import org.apache.hama.bsp.ft.CheckpointService;
  45. import org.apache.hama.bsp.ft.IFaultTolerantPeerService;
  46. import org.apache.hama.bsp.message.HadoopMessageManager;
  47. import org.apache.hama.bsp.message.MessageManager;
  48. import org.apache.hama.bsp.message.MessageManagerFactory;
  49. import org.apache.hama.bsp.message.MessageQueue;
  50. import org.apache.hama.bsp.message.type.ByteMessage;
  51. import org.apache.hama.bsp.sync.BSPPeerSyncClient;
  52. import org.apache.hama.bsp.sync.PeerSyncClient;
  53. import org.apache.hama.bsp.sync.SyncClient;
  54. import org.apache.hama.bsp.sync.SyncEvent;
  55. import org.apache.hama.bsp.sync.SyncEventListener;
  56. import org.apache.hama.bsp.sync.SyncException;
  57. import org.apache.hama.bsp.sync.SyncServiceFactory;
  58. import org.apache.hama.bsp.sync.ZooKeeperSyncClientImpl;
  59. import org.apache.hama.ipc.BSPPeerProtocol;
  60. import org.apache.hama.ipc.HamaRPCProtocolVersion;
  61. import org.apache.hama.util.BSPNetUtils;
  62. import org.apache.hama.util.KeyValuePair;
  63.  
  64. public class TestCheckpoint extends TestCase {
  65.  
  66.   public static final Log LOG = LogFactory.getLog(TestCheckpoint.class);
  67.  
  68.   static final String checkpointedDir = "checkpoint/job_201110302255_0001/0/";
  69.  
  70.   public static class TestMessageManager<Text> implements MessageManager<Writable>{
  71.  
  72.     List<Text> messageQueue;
  73.    
  74.     @Override
  75.     public void init(TaskAttemptID attemptId,
  76.         BSPPeer<?, ?, ?, ?, Writable> peer, Configuration conf,
  77.         InetSocketAddress peerAddress) {
  78.       // TODO Auto-generated method stub
  79.      
  80.     }
  81.  
  82.     @Override
  83.     public void close() {
  84.       // TODO Auto-generated method stub
  85.      
  86.     }
  87.  
  88.     @Override
  89.     public Writable getCurrentMessage() throws IOException {
  90.       // TODO Auto-generated method stub
  91.       return null;
  92.     }
  93.  
  94.     @Override
  95.     public void send(String peerName, Writable msg) throws IOException {
  96.       // TODO Auto-generated method stub
  97.      
  98.     }
  99.  
  100.     @Override
  101.     public void finishSendPhase() throws IOException {
  102.       // TODO Auto-generated method stub
  103.      
  104.     }
  105.  
  106.     @Override
  107.     public Iterator<Entry<InetSocketAddress, MessageQueue<Writable>>> getMessageIterator() {
  108.       // TODO Auto-generated method stub
  109.       return null;
  110.     }
  111.  
  112.     @Override
  113.     public void transfer(InetSocketAddress addr,
  114.         BSPMessageBundle<Writable> bundle) throws IOException {
  115.       // TODO Auto-generated method stub
  116.      
  117.     }
  118.  
  119.     @Override
  120.     public void clearOutgoingQueues() {
  121.       // TODO Auto-generated method stub
  122.      
  123.     }
  124.  
  125.     @Override
  126.     public int getNumCurrentMessages() {
  127.       // TODO Auto-generated method stub
  128.       return 0;
  129.     }
  130.  
  131.     @Override
  132.     public void loopBackMessage(BSPMessageBundle<? extends Writable> bundle) {
  133.       // TODO Auto-generated method stub
  134.      
  135.     }
  136.  
  137.     @Override
  138.     public void replayMessages() {
  139.       // TODO Auto-generated method stub
  140.      
  141.     }
  142.    
  143.   }
  144.  
  145.   public static class TestBSPPeer
  146.   implements BSPPeer<NullWritable, NullWritable, NullWritable, NullWritable, Writable> {
  147.  
  148.     Configuration conf;
  149.     long superstepCount;
  150.    
  151.     public TestBSPPeer(Configuration conf){
  152.       this.conf = conf;
  153.       superstepCount = 0;
  154.     }
  155.     @Override
  156.     public void send(String peerName, Writable msg) throws IOException {}
  157.  
  158.     @Override
  159.     public Writable getCurrentMessage() throws IOException {
  160.       return new Text("data");
  161.     }
  162.  
  163.     @Override
  164.     public int getNumCurrentMessages() {
  165.       return 1;
  166.     }
  167.  
  168.     @Override
  169.     public void sync() throws IOException, SyncException, InterruptedException {
  170.       ++superstepCount;
  171.      
  172.     }
  173.  
  174.     @Override
  175.     public long getSuperstepCount() {
  176.       // TODO Auto-generated method stub
  177.       return 0;
  178.     }
  179.  
  180.     @Override
  181.     public String getPeerName() {
  182.       // TODO Auto-generated method stub
  183.       return null;
  184.     }
  185.  
  186.     @Override
  187.     public String getPeerName(int index) {
  188.       // TODO Auto-generated method stub
  189.       return null;
  190.     }
  191.  
  192.     @Override
  193.     public int getPeerIndex() {
  194.       // TODO Auto-generated method stub
  195.       return 1;
  196.     }
  197.  
  198.     @Override
  199.     public String[] getAllPeerNames() {
  200.       // TODO Auto-generated method stub
  201.       return null;
  202.     }
  203.  
  204.     @Override
  205.     public int getNumPeers() {
  206.       // TODO Auto-generated method stub
  207.       return 0;
  208.     }
  209.  
  210.     @Override
  211.     public void clear() {
  212.       // TODO Auto-generated method stub
  213.      
  214.     }
  215.  
  216.     @Override
  217.     public void write(NullWritable key, NullWritable value) throws IOException {
  218.       // TODO Auto-generated method stub
  219.      
  220.     }
  221.  
  222.     @Override
  223.     public boolean readNext(NullWritable key, NullWritable value)
  224.         throws IOException {
  225.       // TODO Auto-generated method stub
  226.       return false;
  227.     }
  228.  
  229.     @Override
  230.     public KeyValuePair<NullWritable, NullWritable> readNext()
  231.         throws IOException {
  232.       // TODO Auto-generated method stub
  233.       return null;
  234.     }
  235.  
  236.     @Override
  237.     public void reopenInput() throws IOException {
  238.       // TODO Auto-generated method stub
  239.      
  240.     }
  241.  
  242.     @Override
  243.     public Configuration getConfiguration() {
  244.       // TODO Auto-generated method stub
  245.       return null;
  246.     }
  247.  
  248.     @Override
  249.     public Counter getCounter(Enum<?> name) {
  250.       // TODO Auto-generated method stub
  251.       return null;
  252.     }
  253.  
  254.     @Override
  255.     public Counter getCounter(String group, String name) {
  256.       // TODO Auto-generated method stub
  257.       return null;
  258.     }
  259.  
  260.     @Override
  261.     public void incrementCounter(Enum<?> key, long amount) {
  262.       // TODO Auto-generated method stub
  263.      
  264.     }
  265.  
  266.     @Override
  267.     public void incrementCounter(String group, String counter, long amount) {
  268.       // TODO Auto-generated method stub
  269.      
  270.     }
  271.    
  272.   }
  273.  
  274.   public static class TempSyncClient extends BSPPeerSyncClient {
  275.  
  276.     @Override
  277.     public String constructKey(BSPJobID jobId, String... args) {
  278.       // TODO Auto-generated method stub
  279.       return null;
  280.     }
  281.  
  282.     @Override
  283.     public boolean storeInformation(String key, Writable value,
  284.         boolean permanent, SyncEventListener listener) {
  285.       // TODO Auto-generated method stub
  286.       return false;
  287.     }
  288.  
  289.     @Override
  290.     public Writable getInformation(String key,
  291.         Class<? extends Writable> classType) {
  292.       // TODO Auto-generated method stub
  293.       return null;
  294.     }
  295.  
  296.     @Override
  297.     public boolean addKey(String key, boolean permanent,
  298.         SyncEventListener listener) {
  299.       // TODO Auto-generated method stub
  300.       return false;
  301.     }
  302.  
  303.     @Override
  304.     public boolean hasKey(String key) {
  305.       // TODO Auto-generated method stub
  306.       return false;
  307.     }
  308.  
  309.     @Override
  310.     public String[] getChildKeySet(String key, SyncEventListener listener) {
  311.       // TODO Auto-generated method stub
  312.       return null;
  313.     }
  314.  
  315.     @Override
  316.     public boolean registerListener(String key, SyncEvent event,
  317.         SyncEventListener listener) {
  318.       // TODO Auto-generated method stub
  319.       return false;
  320.     }
  321.  
  322.     @Override
  323.     public boolean remove(String key, SyncEventListener listener) {
  324.       // TODO Auto-generated method stub
  325.       return false;
  326.     }
  327.  
  328.     @Override
  329.     public void init(Configuration conf, BSPJobID jobId, TaskAttemptID taskId)
  330.         throws Exception {
  331.       // TODO Auto-generated method stub
  332.      
  333.     }
  334.  
  335.     @Override
  336.     public void enterBarrier(BSPJobID jobId, TaskAttemptID taskId,
  337.         long superstep) throws SyncException {
  338.       LOG.info("Enter barrier called - " + superstep);
  339.      
  340.     }
  341.  
  342.     @Override
  343.     public void leaveBarrier(BSPJobID jobId, TaskAttemptID taskId,
  344.         long superstep) throws SyncException {
  345.       LOG.info("Exit barrier called - " + superstep);
  346.      
  347.     }
  348.  
  349.     @Override
  350.     public void register(BSPJobID jobId, TaskAttemptID taskId,
  351.         String hostAddress, long port) {
  352.       // TODO Auto-generated method stub
  353.      
  354.     }
  355.  
  356.     @Override
  357.     public String[] getAllPeerNames(TaskAttemptID taskId) {
  358.       // TODO Auto-generated method stub
  359.       return null;
  360.     }
  361.  
  362.     @Override
  363.     public void deregisterFromBarrier(BSPJobID jobId, TaskAttemptID taskId,
  364.         String hostAddress, long port) {
  365.       // TODO Auto-generated method stub
  366.      
  367.     }
  368.  
  369.     @Override
  370.     public void stopServer() {
  371.       // TODO Auto-generated method stub
  372.      
  373.     }
  374.  
  375.     @Override
  376.     public void close() throws IOException {
  377.       // TODO Auto-generated method stub
  378.      
  379.     }
  380.    
  381.   }
  382.  
  383.  
  384.   @SuppressWarnings({ "unchecked", "rawtypes" })
  385.   public void testCheckpoint() throws Exception {
  386.     Configuration config = new Configuration();
  387.     config.set(SyncServiceFactory.SYNC_PEER_CLASS,
  388.         TempSyncClient.class.getName());
  389.     config.set(Constants.FAULT_TOLERANCE_CLASS,
  390.         CheckpointService.class.getName());
  391.     int port = BSPNetUtils.getFreePort(12502);
  392.     LOG.info("Got port = " + port);
  393.    
  394.     config.set(Constants.PEER_HOST, Constants.DEFAULT_PEER_HOST);
  395.     config.setInt(Constants.PEER_PORT, port);
  396.    
  397.     config.set("bsp.output.dir", "/tmp/hama-test_out");
  398.     FileSystem dfs = FileSystem.get(config);
  399.     BSPJob job = new BSPJob(new BSPJobID("checkpttest", 1), "/tmp");
  400.     TaskAttemptID taskId = new TaskAttemptID(new TaskID(job.getJobID(), 1), 1);
  401.     //BSPPeerImpl bspTask = new BSPPeerImpl(job, config, dfs, taskId);
  402.     BSPPeer bspTask = new TestBSPPeer(config);
  403.     TestMessageManager<Text> messenger = new TestMessageManager<Text>();
  404.        
  405.     assertNotNull("BSPPeerImpl should not be null.", bspTask);
  406.     if (dfs.mkdirs(new Path("checkpoint"))) {
  407.       if (dfs.mkdirs(new Path("checkpoint/job_201110302255_0001"))) {
  408.         if (dfs.mkdirs(new Path("checkpoint/job_201110302255_0001/0")))
  409.           ;
  410.       }
  411.     }
  412.    
  413.     LOG.info("Created bsp peer and other parameters");
  414.    
  415. //    HadoopMessageManager<? extends Writable> messageManager =
  416. //        (HadoopMessageManager<? extends Writable>) RPC.getProxy(
  417. //        HadoopMessageManager.class, HamaRPCProtocolVersion.versionID,
  418. //        new InetSocketAddress(port), config);
  419. //    InetSocketAddress peerAddress = new InetSocketAddress(port);
  420. //    MessageManager<? extends Writable> messageManager = MessageManagerFactory.getMessageManager(config);
  421. //    messageManager.init(taskId, bspTask, config, peerAddress);
  422.    
  423.     PeerSyncClient syncClient = (TempSyncClient) SyncServiceFactory
  424.         .getPeerSyncClient(config);
  425.     IFaultTolerantPeerService<Text> service = null;
  426.     boolean initialized = false;
  427.     try{
  428.       service = (new CheckpointService<Text>())
  429.         .constructPeerFaultTolerance(job, (BSPPeer)bspTask, syncClient,
  430.             new InetSocketAddress(port), taskId, -1L , config,
  431.             (MessageManager<Text>)messageManager);
  432.       initialized = true;
  433.     }
  434.     catch(Exception e){
  435.      
  436.     }
  437.     assertTrue(initialized);
  438.     LOG.info("Initialized correctly");
  439.    
  440.    
  441.     assertTrue("Make sure directory is created.",
  442.         dfs.exists(new Path(checkpointedDir)));
  443.     byte[] tmpData = "data".getBytes();
  444. //    BSPMessageBundle bundle = new BSPMessageBundle();
  445. //    bundle.addMessage(new ByteMessage("abc".getBytes(), tmpData));
  446. //    assertNotNull("Message bundle can not be null.", bundle);
  447. //    assertNotNull("Configuration should not be null.", config);
  448. //    
  449. //    messageManager.put(bundle);
  450.    
  451.     bspTask.sync();
  452.    
  453.     LOG.info("out of sync");
  454.    
  455.  
  456.    
  457.    
  458.    
  459.     //Path path = new Path()
  460.    
  461. //    bspTask.checkpointSentMessages(checkpointedDir + "/attempt_201110302255_0001_000000_0",
  462. //        bundle);
  463.    
  464.     FSDataInputStream in = dfs.open(new Path(checkpointedDir
  465.         + "/attempt_201110302255_0001_000000_0"));
  466.     BSPMessageBundle bundleRead = new BSPMessageBundle();
  467.     bundleRead.readFields(in);
  468.     in.close();
  469.     ByteMessage byteMsg = (ByteMessage) (bundleRead.getMessages()).get(0);
  470.     String content = new String(byteMsg.getData());
  471.     LOG.info("Saved checkpointed content is " + content);
  472.     assertTrue("Message content should be the same.", "data".equals(content));
  473.     dfs.delete(new Path("checkpoint"), true);
  474.   }
  475.  
  476. //  public void testCheckpointInterval() throws Exception {
  477. //
  478. //    Configuration conf = new Configuration();
  479. //    conf.set("bsp.output.dir", "/tmp/hama-test_out");
  480. //    conf.setClass(SyncServiceFactory.SYNC_PEER_CLASS,
  481. //        LocalBSPRunner.LocalSyncClient.class, SyncClient.class);
  482. //
  483. //    conf.setBoolean(Constants.CHECKPOINT_ENABLED, false);
  484. //
  485. //    int port = BSPNetUtils.getFreePort(5000);
  486. //    InetSocketAddress inetAddress = new InetSocketAddress(port);
  487. //    MinimalGroomServer groom = new MinimalGroomServer(conf);
  488. //    Server workerServer = RPC.getServer(groom, inetAddress.getHostName(),
  489. //        inetAddress.getPort(), conf);
  490. //    workerServer.start();
  491. //
  492. //    LOG.info("Started RPC server");
  493. //    conf.setInt("bsp.groom.rpc.port", inetAddress.getPort());
  494. //    conf.setInt("bsp.peers.num", 1);
  495. //
  496. //    BSPPeerProtocol umbilical = (BSPPeerProtocol) RPC.getProxy(
  497. //        BSPPeerProtocol.class, HamaRPCProtocolVersion.versionID, inetAddress,
  498. //        conf);
  499. //    LOG.info("Started the proxy connections");
  500. //
  501. //    TaskAttemptID tid = new TaskAttemptID(new TaskID(new BSPJobID(
  502. //        "job_201110102255", 1), 1), 1);
  503. //
  504. //    try {
  505. //      BSPJob job = new BSPJob(new HamaConfiguration(conf));
  506. //      job.setOutputPath(TestBSPMasterGroomServer.OUTPUT_PATH);
  507. //      job.setOutputFormat(TextOutputFormat.class);
  508. //      final BSPPeerProtocol proto = (BSPPeerProtocol) RPC.getProxy(
  509. //          BSPPeerProtocol.class, HamaRPCProtocolVersion.versionID,
  510. //          new InetSocketAddress("127.0.0.1", port), conf);
  511. //
  512. //      BSPTask task = new BSPTask();
  513. //      task.setConf(job);
  514. //
  515. //      @SuppressWarnings("rawtypes")
  516. //      BSPPeerImpl<?, ?, ?, ?, ?> bspPeer = new BSPPeerImpl(job, conf, tid,
  517. //          proto, 0, null, null, new Counters());
  518. //
  519. //      bspPeer.setCurrentTaskStatus(new TaskStatus(new BSPJobID(), tid, 1.0f,
  520. //          TaskStatus.State.RUNNING, "running", "127.0.0.1",
  521. //          TaskStatus.Phase.STARTING, new Counters()));
  522. //
  523. //      assertEquals(bspPeer.isReadyToCheckpoint(), false);
  524. //
  525. //      conf.setBoolean(Constants.CHECKPOINT_ENABLED, true);
  526. //      conf.setInt(Constants.CHECKPOINT_INTERVAL, 3);
  527. //
  528. //      bspPeer.sync();
  529. //
  530. //      LOG.info("Is Ready = " + bspPeer.isReadyToCheckpoint() + " at step "
  531. //          + bspPeer.getSuperstepCount());
  532. //      assertEquals(bspPeer.isReadyToCheckpoint(), false);
  533. //      bspPeer.sync();
  534. //      LOG.info("Is Ready = " + bspPeer.isReadyToCheckpoint() + " at step "
  535. //          + bspPeer.getSuperstepCount());
  536. //      assertEquals(bspPeer.isReadyToCheckpoint(), false);
  537. //      bspPeer.sync();
  538. //      LOG.info("Is Ready = " + bspPeer.isReadyToCheckpoint() + " at step "
  539. //          + bspPeer.getSuperstepCount());
  540. //      assertEquals(bspPeer.isReadyToCheckpoint(), true);
  541. //
  542. //      job.setCheckPointInterval(5);
  543. //      bspPeer.sync();
  544. //      LOG.info("Is Ready = " + bspPeer.isReadyToCheckpoint() + " at step "
  545. //          + bspPeer.getSuperstepCount());
  546. //      assertEquals(bspPeer.isReadyToCheckpoint(), false);
  547. //      bspPeer.sync();
  548. //      LOG.info("Is Ready = " + bspPeer.isReadyToCheckpoint() + " at step "
  549. //          + bspPeer.getSuperstepCount());
  550. //      assertEquals(bspPeer.isReadyToCheckpoint(), false);
  551. //
  552. //    } catch (Exception e) {
  553. //      LOG.error("Error testing BSPPeer.", e);
  554. //    } finally {
  555. //      umbilical.close();
  556. //      Thread.sleep(2000);
  557. //      workerServer.stop();
  558. //      Thread.sleep(2000);
  559. //    }
  560. //
  561. //  }
  562. }
Advertisement
Add Comment
Please, Sign In to add comment
Public Pastes
We use cookies for various purposes including analytics. By continuing to use Pastebin, you agree to our use of cookies as described in the Cookies Policy.  OK, I Understand
Not a member of Pastebin yet?
Sign Up, it unlocks many cool features!