From 321848d09c86fd8d78331fe0ef59c5b97a3afa8e Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Wed, 11 Nov 2009 16:11:36 +0000 Subject: [PATCH] clean up transfer code from BMVH; move to Streaming.java patch by jbellis; reviewed by Jaakko Laine for CASSANDRA-435 git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@834937 13f79535-47bb-0310-9956-ffa450edef68 --- .../dht/BootstrapMetadataVerbHandler.java | 134 ++---------------- src/java/org/apache/cassandra/io/SSTable.java | 7 + .../org/apache/cassandra/io/Streaming.java | 98 +++++++++++++ 3 files changed, 120 insertions(+), 119 deletions(-) create mode 100644 src/java/org/apache/cassandra/io/Streaming.java diff --git a/src/java/org/apache/cassandra/dht/BootstrapMetadataVerbHandler.java b/src/java/org/apache/cassandra/dht/BootstrapMetadataVerbHandler.java index bc82368e7f..532990bee1 100644 --- a/src/java/org/apache/cassandra/dht/BootstrapMetadataVerbHandler.java +++ b/src/java/org/apache/cassandra/dht/BootstrapMetadataVerbHandler.java @@ -20,6 +20,7 @@ package org.apache.cassandra.dht; import java.io.File; import java.io.IOException; +import java.io.IOError; import java.util.ArrayList; import java.util.List; import java.util.Collection; @@ -28,6 +29,8 @@ import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.Table; import org.apache.cassandra.io.DataInputBuffer; import org.apache.cassandra.io.SSTableReader; +import org.apache.cassandra.io.Streaming; + import java.net.InetAddress; import org.apache.cassandra.net.IVerbHandler; import org.apache.cassandra.net.Message; @@ -35,10 +38,11 @@ import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.net.io.StreamContextManager; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.service.StreamManager; -import org.apache.cassandra.utils.LogUtil; -import org.apache.log4j.Logger; -/** +import org.apache.log4j.Logger; +import org.apache.commons.lang.StringUtils; + + /** * This verb handler handles the BootstrapMetadataMessage that is sent * by the leader to the nodes that are responsible for handing off data. */ @@ -61,125 +65,17 @@ public class BootstrapMetadataVerbHandler implements IVerbHandler { BootstrapMetadataMessage bsMetadataMessage = BootstrapMetadataMessage.serializer().deserialize(bufIn); BootstrapMetadata[] bsMetadata = bsMetadataMessage.bsMetadata_; - - /* - * This is for debugging purposes. Remove later. - */ - for ( BootstrapMetadata bsmd : bsMetadata ) - { - if (logger_.isDebugEnabled()) - logger_.debug(bsmd.toString()); - } - - for ( BootstrapMetadata bsmd : bsMetadata ) - { - long startTime = System.currentTimeMillis(); - doTransfer(bsmd.target_, bsmd.ranges_); - if (logger_.isDebugEnabled()) - logger_.debug("Time taken to boostrap " + - bsmd.target_ + - " is " + - (System.currentTimeMillis() - startTime) + - " msecs."); - } - } - catch ( IOException ex ) - { - logger_.info(LogUtil.throwableToString(ex)); - } - } - - /* - * This method needs to figure out the files on disk - * locally for each range and then stream them using - * the Bootstrap protocol to the target endpoint. - */ - private void doTransfer(InetAddress target, Collection ranges) throws IOException - { - if ( ranges.size() == 0 ) - { - if (logger_.isDebugEnabled()) - logger_.debug("No ranges to give scram ..."); - return; - } - - /* Just for debugging process - remove later */ - for ( Range range : ranges ) - { - StringBuilder sb = new StringBuilder(""); - sb.append(range.toString()); - sb.append(" "); - if (logger_.isDebugEnabled()) - logger_.debug("Beginning transfer process to " + target + " for ranges " + sb.toString()); - } - - /* - * (1) First we dump all the memtables to disk. - * (2) Run a version of compaction which will basically - * put the keys in the range specified into a directory - * named as per the endpoint it is destined for inside the - * bootstrap directory. - * (3) Handoff the data. - */ - List tables = DatabaseDescriptor.getTables(); - for ( String tName : tables ) - { - Table table = Table.open(tName); - if (logger_.isDebugEnabled()) - logger_.debug("Flushing memtables ..."); - table.flush(false); - if (logger_.isDebugEnabled()) - logger_.debug("Forcing compaction ..."); - /* Get the counting bloom filter for each endpoint and the list of files that need to be streamed */ - List fileList = new ArrayList(); - for (SSTableReader sstable : table.forceAntiCompaction(ranges, target)) - { - fileList.add(sstable.indexFilename()); - fileList.add(sstable.filterFilename()); - fileList.add(sstable.getFilename()); - } - doHandoff(target, fileList, tName); - //In Handoff, Streaming the file also deletes the file, so no cleanup needed - } - } - /** - * Stream the files in the bootstrap directory over to the - * node being bootstrapped. - */ - private void doHandoff(InetAddress target, List fileList, String table) throws IOException - { - List filesList = new ArrayList(); - for(String file : fileList) - { - filesList.add(new File(file)); + for (BootstrapMetadata bsmd : bsMetadata) + { + if (logger_.isDebugEnabled()) + logger_.debug(bsmd.toString()); + Streaming.transferRanges(bsmd.target_, bsmd.ranges_); + } } - File[] files = filesList.toArray(new File[0]); - StreamContextManager.StreamContext[] streamContexts = new StreamContextManager.StreamContext[files.length]; - int i = 0; - for ( File file : files ) + catch (IOException ex) { - streamContexts[i] = new StreamContextManager.StreamContext(file.getAbsolutePath(), file.length(), table); - if (logger_.isDebugEnabled()) - logger_.debug("Stream context metadata " + streamContexts[i]); - ++i; - } - - if ( files.length > 0 ) - { - /* Set up the stream manager with the files that need to streamed */ - StreamManager.instance(target).addFilesToStream(streamContexts); - /* Send the bootstrap initiate message */ - BootstrapInitiateMessage biMessage = new BootstrapInitiateMessage(streamContexts); - Message message = BootstrapInitiateMessage.makeBootstrapInitiateMessage(biMessage); - if (logger_.isDebugEnabled()) - logger_.debug("Sending a bootstrap initiate message to " + target + " ..."); - MessagingService.instance().sendOneWay(message, target); - if (logger_.isDebugEnabled()) - logger_.debug("Waiting for transfer to " + target + " to complete"); - StreamManager.instance(target).waitForStreamCompletion(); - if (logger_.isDebugEnabled()) - logger_.debug("Done with transfer to " + target); + throw new IOError(ex); } } } diff --git a/src/java/org/apache/cassandra/io/SSTable.java b/src/java/org/apache/cassandra/io/SSTable.java index 57dfff0983..402c779b39 100644 --- a/src/java/org/apache/cassandra/io/SSTable.java +++ b/src/java/org/apache/cassandra/io/SSTable.java @@ -24,6 +24,7 @@ package org.apache.cassandra.io; import java.io.File; import java.io.IOException; import java.util.List; +import java.util.Arrays; import org.apache.log4j.Logger; import org.apache.commons.lang.StringUtils; @@ -128,6 +129,12 @@ public abstract class SSTable return path; } + /** @return full paths to all the files associated w/ this SSTable */ + public List getAllFilenames() + { + return Arrays.asList(getFilename(), indexFilename(), filterFilename()); + } + public String getColumnFamilyName() { return columnFamilyName; diff --git a/src/java/org/apache/cassandra/io/Streaming.java b/src/java/org/apache/cassandra/io/Streaming.java new file mode 100644 index 0000000000..a42060a2f6 --- /dev/null +++ b/src/java/org/apache/cassandra/io/Streaming.java @@ -0,0 +1,98 @@ +package org.apache.cassandra.io; + +import java.net.InetAddress; +import java.util.Collection; +import java.util.List; +import java.util.ArrayList; +import java.io.IOException; +import java.io.File; + +import org.apache.log4j.Logger; +import org.apache.commons.lang.StringUtils; + +import org.apache.cassandra.dht.Range; +import org.apache.cassandra.dht.BootstrapInitiateMessage; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.Table; +import org.apache.cassandra.net.io.StreamContextManager; +import org.apache.cassandra.net.Message; +import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.service.StreamManager; + +public class Streaming +{ + private static Logger logger = Logger.getLogger(Streaming.class); + + /* + * This method needs to figure out the files on disk + * locally for each range and then stream them using + * the Bootstrap protocol to the target endpoint. + */ + public static void transferRanges(InetAddress target, Collection ranges) throws IOException + { + assert ranges.size() > 0; + + if (logger.isDebugEnabled()) + logger.debug("Beginning transfer process to " + target + " for ranges " + StringUtils.join(ranges, ", ")); + + /* + * (1) First we dump all the memtables to disk. + * (2) Run a version of compaction which will basically + * put the keys in the range specified into a directory + * named as per the endpoint it is destined for inside the + * bootstrap directory. + * (3) Handoff the data. + */ + List tables = DatabaseDescriptor.getTables(); + for (String tName : tables) + { + Table table = Table.open(tName); + if (logger.isDebugEnabled()) + logger.debug("Flushing memtables ..."); + table.flush(false); + if (logger.isDebugEnabled()) + logger.debug("Performing anticompaction ..."); + /* Get the list of files that need to be streamed */ + List fileList = new ArrayList(); + for (SSTableReader sstable : table.forceAntiCompaction(ranges, target)) + { + fileList.addAll(sstable.getAllFilenames()); + } + transferOneTable(target, fileList, tName); // also deletes the file, so no further cleanup needed + } + } + + /** + * Stream the files in the bootstrap directory over to the + * node being bootstrapped. + */ + private static void transferOneTable(InetAddress target, List fileList, String table) throws IOException + { + if (fileList.isEmpty()) + return; + + StreamContextManager.StreamContext[] streamContexts = new StreamContextManager.StreamContext[fileList.size()]; + int i = 0; + for (String filename : fileList) + { + File file = new File(filename); + streamContexts[i++] = new StreamContextManager.StreamContext(file.getAbsolutePath(), file.length(), table); + if (logger.isDebugEnabled()) + logger.debug("Stream context metadata " + streamContexts[i]); + } + + /* Set up the stream manager with the files that need to streamed */ + StreamManager.instance(target).addFilesToStream(streamContexts); + /* Send the bootstrap initiate message */ + BootstrapInitiateMessage biMessage = new BootstrapInitiateMessage(streamContexts); + Message message = BootstrapInitiateMessage.makeBootstrapInitiateMessage(biMessage); + if (logger.isDebugEnabled()) + logger.debug("Sending a bootstrap initiate message to " + target + " ..."); + MessagingService.instance().sendOneWay(message, target); + if (logger.isDebugEnabled()) + logger.debug("Waiting for transfer to " + target + " to complete"); + StreamManager.instance(target).waitForStreamCompletion(); + if (logger.isDebugEnabled()) + logger.debug("Done with transfer to " + target); + } +}