mirror of https://github.com/apache/cassandra
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
This commit is contained in:
Jonathan Ellis
parent
93c5afa5c9
commit
321848d09c
|
|
@ -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<Range> 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<String> 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<String> fileList = new ArrayList<String>();
|
||||
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<String> fileList, String table) throws IOException
|
||||
{
|
||||
List<File> filesList = new ArrayList<File>();
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<String> getAllFilenames()
|
||||
{
|
||||
return Arrays.asList(getFilename(), indexFilename(), filterFilename());
|
||||
}
|
||||
|
||||
public String getColumnFamilyName()
|
||||
{
|
||||
return columnFamilyName;
|
||||
|
|
|
|||
|
|
@ -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<Range> 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<String> 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<String> fileList = new ArrayList<String>();
|
||||
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<String> 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);
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue