cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From alek...@apache.org
Subject [2/2] git commit: Merge branch 'cassandra-1.2' into trunk
Date Wed, 20 Mar 2013 17:09:45 GMT
Merge branch 'cassandra-1.2' into trunk

Conflicts:
	src/java/org/apache/cassandra/db/HintedHandOffManager.java


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/076a0e4e
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/076a0e4e
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/076a0e4e

Branch: refs/heads/trunk
Commit: 076a0e4e78793a11b5be78e3eb592514b7fc0d79
Parents: 08721da d224c2b
Author: Aleksey Yeschenko <aleksey@apache.org>
Authored: Wed Mar 20 20:09:31 2013 +0300
Committer: Aleksey Yeschenko <aleksey@apache.org>
Committed: Wed Mar 20 20:09:31 2013 +0300

----------------------------------------------------------------------
 CHANGES.txt                                        |    1 +
 .../apache/cassandra/db/HintedHandOffManager.java  |  178 ++++++++-------
 2 files changed, 97 insertions(+), 82 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/076a0e4e/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index 681fb44,3eb85d6..3309ab9
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -22,9 -2,9 +22,10 @@@
   * Fix mixing prepared statements between keyspaces (CASSANDRA-5352)
   * Fix consistency level during bootstrap - strike 3 (CASSANDRA-5354)
   * Fix transposed arguments in AlreadyExistsException (CASSANDRA-5362)
+  * Improve asynchronous hint delivery (CASSANDRA-5179)
  
  
 +
  1.2.3
   * add check for sstable overlap within a level on startup (CASSANDRA-5327)
   * replace ipv6 colons in jmx object names (CASSANDRA-5298, 5328)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/076a0e4e/src/java/org/apache/cassandra/db/HintedHandOffManager.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/HintedHandOffManager.java
index db85efb,53411f5..cb306ae
--- a/src/java/org/apache/cassandra/db/HintedHandOffManager.java
+++ b/src/java/org/apache/cassandra/db/HintedHandOffManager.java
@@@ -57,8 -58,7 +59,6 @@@ import org.apache.cassandra.metrics.Hin
  import org.apache.cassandra.net.MessageOut;
  import org.apache.cassandra.net.MessagingService;
  import org.apache.cassandra.service.*;
--import org.apache.cassandra.thrift.*;
  import org.apache.cassandra.utils.ByteBufferUtil;
  import org.apache.cassandra.utils.FBUtilities;
  import org.apache.cassandra.utils.UUIDGen;
@@@ -108,31 -107,11 +107,34 @@@ public class HintedHandOffManager imple
                                                                                   Integer.MAX_VALUE,
                                                                                   TimeUnit.SECONDS,
                                                                                   new LinkedBlockingQueue<Runnable>(),
-                                                                                  new NamedThreadFactory("HintedHandoff",
Thread.MIN_PRIORITY), "internal");
+                                                                                  new NamedThreadFactory("HintedHandoff",
Thread.MIN_PRIORITY),
+                                                                                  "internal");
+ 
+     private final ColumnFamilyStore hintStore = Table.open(Table.SYSTEM_KS).getColumnFamilyStore(SystemTable.HINTS_CF);
  
 +    /**
 +     * Returns a mutation representing a Hint to be sent to <code>targetId</code>
 +     * as soon as it becomes available again.
 +     */
 +    public static RowMutation hintFor(RowMutation mutation, UUID targetId) throws IOException
 +    {
 +        UUID hintId = UUIDGen.getTimeUUID();
 +
 +        // The hint TTL is set at the smallest GCGraceSeconds for any of the CFs in the
RM;
 +        // this ensures that deletes aren't "undone" by delivery of an old hint
 +        int ttl = Integer.MAX_VALUE;
 +        for (ColumnFamily cf : mutation.getColumnFamilies())
 +            ttl = Math.min(ttl, cf.metadata().getGcGraceSeconds());
 +
 +        // serialize the hint with id and version as a composite column name
 +        ByteBuffer name = comparator.decompose(hintId, MessagingService.current_version);
 +        ByteBuffer value = ByteBuffer.wrap(FBUtilities.serialize(mutation, RowMutation.serializer,
MessagingService.current_version));
 +        ColumnFamily cf = ColumnFamily.create(Schema.instance.getCFMetaData(Table.SYSTEM_KS,
SystemTable.HINTS_CF));
 +        cf.addColumn(name, value, System.currentTimeMillis(), ttl);
 +
 +        return new RowMutation(Table.SYSTEM_KS, UUIDType.instance.decompose(targetId), cf);
 +    }
 +
      public void start()
      {
          MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
@@@ -157,10 -136,10 +159,10 @@@
          StorageService.optionalTasks.scheduleWithFixedDelay(runnable, 10, 10, TimeUnit.MINUTES);
      }
  
-     private static void deleteHint(ByteBuffer tokenBytes, ByteBuffer columnName, long timestamp)
throws IOException
+     private static void deleteHint(ByteBuffer tokenBytes, ByteBuffer columnName, long timestamp)
      {
          RowMutation rm = new RowMutation(Table.SYSTEM_KS, tokenBytes);
 -        rm.delete(new QueryPath(SystemTable.HINTS_CF, null, columnName), timestamp);
 +        rm.delete(SystemTable.HINTS_CF, columnName, timestamp);
          rm.applyUnsafe(); // don't bother with commitlog since we're going to flush as soon
as we're done with delivery
      }
  
@@@ -339,37 -302,40 +325,40 @@@
          int throttleInKB = DatabaseDescriptor.getHintedHandoffThrottleInKB();
          RateLimiter rateLimiter = RateLimiter.create(throttleInKB == 0 ? Double.MAX_VALUE
: throttleInKB * 1024);
  
+         delivery:
          while (true)
          {
-             // check if hints delivery has been paused during the process
-             if (hintedHandOffPaused)
-             {
-                 logger.debug("Hints delivery process is paused, aborting");
-                 break;
-             }
+             QueryFilter filter = QueryFilter.getSliceFilter(epkey,
 -                                                            new QueryPath(SystemTable.HINTS_CF),
++                                                            SystemTable.HINTS_CF,
+                                                             startColumn,
+                                                             ByteBufferUtil.EMPTY_BYTE_BUFFER,
+                                                             false,
+                                                             pageSize);
  
-             QueryFilter filter = QueryFilter.getSliceFilter(epkey, SystemTable.HINTS_CF,
startColumn, ByteBufferUtil.EMPTY_BYTE_BUFFER, false, pageSize);
+             ColumnFamily hintsPage = ColumnFamilyStore.removeDeleted(hintStore.getColumnFamily(filter),
+                                                                      (int) (System.currentTimeMillis()
/ 1000));
  
-             ColumnFamily hintsPage = ColumnFamilyStore.removeDeleted(hintStore.getColumnFamily(filter),
(int)(System.currentTimeMillis() / 1000));
              if (pagingFinished(hintsPage, startColumn))
-             {
-                 if (ByteBufferUtil.EMPTY_BYTE_BUFFER.equals(startColumn))
-                 {
-                     // we've started from the beginning and could not find anything (only
maybe tombstones)
-                     break;
-                 }
-                 else
-                 {
-                     // restart query from the first column until we read an empty row;
-                     // that will tell us everything was delivered successfully with no timeouts
-                     startColumn = ByteBufferUtil.EMPTY_BYTE_BUFFER;
-                     continue;
-                 }
+                 break;
  
+             // check if node is still alive and we should continue delivery process
+             if (!FailureDetector.instance.isAlive(endpoint))
+             {
+                 logger.info("Endpoint {} died during hint delivery; aborting ({} delivered)",
endpoint, rowsReplayed);
+                 return;
              }
  
+             List<WriteResponseHandler> responseHandlers = Lists.newArrayList();
+ 
 -            for (final IColumn hint : hintsPage.getSortedColumns())
 +            for (final Column hint : hintsPage.getSortedColumns())
              {
+                 // check if hints delivery has been paused during the process
+                 if (hintedHandOffPaused)
+                 {
+                     logger.debug("Hints delivery process is paused, aborting");
+                     break delivery;
+                 }
+ 
                  // Skip tombstones:
                  // if we iterate quickly enough, it's possible that we could request a new
page in the same millisecond
                  // in which the local deletion timestamp was generated on the last column
in the old page, in which


Mime
View raw message