src/java/org/apache/cassandra/utils/concurrent/Ref.java - cassandra - Git at Google

 /*
  *
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  *
  */
 package org.apache.cassandra.utils.concurrent;

 import java.lang.ref.PhantomReference;
 import java.lang.ref.Reference;
 import java.lang.ref.ReferenceQueue;
 import java.lang.ref.WeakReference;
 import java.lang.reflect.Field;
 import java.lang.reflect.Modifier;
 import java.util.*;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;

 import org.apache.cassandra.concurrent.InfiniteLoopExecutor;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;

 import org.apache.cassandra.concurrent.NamedThreadFactory;
 import org.apache.cassandra.db.ColumnFamilyStore;
 import org.apache.cassandra.db.Keyspace;
 import org.apache.cassandra.db.lifecycle.View;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.io.util.Memory;
 import org.apache.cassandra.io.util.SafeMemory;
 import org.apache.cassandra.utils.ExecutorUtils;
 import org.apache.cassandra.utils.NoSpamLogger;
 import org.apache.cassandra.utils.Pair;
 import org.cliffc.high_scale_lib.NonBlockingHashMap;

 import static java.util.Collections.emptyList;
 import org.apache.cassandra.concurrent.InfiniteLoopExecutor.InterruptibleRunnable;

 import static org.apache.cassandra.utils.ExecutorUtils.awaitTermination;
 import static org.apache.cassandra.utils.ExecutorUtils.shutdownNow;
 import static org.apache.cassandra.utils.Throwables.maybeFail;
 import static org.apache.cassandra.utils.Throwables.merge;

 /**
  * An object that needs ref counting does the two following:
  *   - defines a Tidy object that will cleanup once it's gone,
  *     (this must retain no references to the object we're tracking (only its resources and how to clean up))
  * Then, one of two options:
  * 1) Construct a Ref directly pointing to it, and always use this Ref; or
  * 2)
  *   - implements RefCounted
  *   - encapsulates a Ref, we'll call selfRef, to which it proxies all calls to RefCounted behaviours
  *   - users must ensure no references to the selfRef leak, or are retained outside of a method scope.
  *     (to ensure the selfRef is collected with the object, so that leaks may be detected and corrected)
  *
  * This class' functionality is achieved by what may look at first glance like a complex web of references,
  * but boils down to:
  *
  * {@code
  * Target --> selfRef --> [Ref.State] <--> Ref.GlobalState --> Tidy
  *                                             ^
  *                                             |
  * Ref ----------------------------------------
  *                                             |
  * Global -------------------------------------
  * }
  * So that, if Target is collected, Impl is collected and, hence, so is selfRef.
  *
  * Once ref or selfRef are collected, the paired Ref.State's release method is called, which if it had
  * not already been called will update Ref.GlobalState and log an error.
  *
  * Once the Ref.GlobalState has been completely released, the Tidy method is called and it removes the global reference
  * to itself so it may also be collected.
  */
 public final class Ref<T> implements RefCounted<T>
 {
     static final Logger logger = LoggerFactory.getLogger(Ref.class);
     public static final boolean DEBUG_ENABLED = System.getProperty("cassandra.debugrefcount", "false").equalsIgnoreCase("true");

     final State state;
     final T referent;

     public Ref(T referent, Tidy tidy)
     {
         this.state = new State(new GlobalState(tidy), this, referenceQueue);
         this.referent = referent;
     }

     Ref(T referent, GlobalState state)
     {
         this.state = new State(state, this, referenceQueue);
         this.referent = referent;
     }

     /**
      * Must be called exactly once, when the logical operation for which this Ref was created has terminated.
      * Failure to abide by this contract will result in an error (eventually) being reported, assuming a
      * hard reference to the resource it managed is not leaked.
      */
     public void release()
     {
         state.release(false);
     }

     public Throwable ensureReleased(Throwable accumulate)
     {
         return state.ensureReleased(accumulate);
     }

     public void ensureReleased()
     {
         maybeFail(state.ensureReleased(null));
     }

     public void close()
     {
         ensureReleased();
     }

     public T get()
     {
         state.assertNotReleased();
         return referent;
     }

     public Ref<T> tryRef()
     {
         return state.globalState.ref() ? new Ref<>(referent, state.globalState) : null;
     }

     public Ref<T> ref()
     {
         Ref<T> ref = tryRef();
         // TODO: print the last release as well as the release here
         if (ref == null)
             state.assertNotReleased();
         return ref;
     }

     public String printDebugInfo()
     {
         if (DEBUG_ENABLED)
         {
             state.debug.log(state.toString());
             return "Memory was freed by " + state.debug.deallocateThread;
         }
         return "Memory was freed";
     }

     /**
      * A convenience method for reporting:
      * @return the number of currently extant references globally, including the shared reference
      */
     public int globalCount()
     {
         return state.globalState.count();
     }

     // similar to Ref.GlobalState, but tracks only the management of each unique ref created to the managed object
     // ensures it is only released once, and that it is always released
     static final class State extends PhantomReference<Ref>
     {
         final Debug debug = DEBUG_ENABLED ? new Debug() : null;
         final GlobalState globalState;
         private volatile int released;

         private static final AtomicIntegerFieldUpdater<State> releasedUpdater = AtomicIntegerFieldUpdater.newUpdater(State.class, "released");

         public State(final GlobalState globalState, Ref reference, ReferenceQueue<? super Ref> q)
         {
             super(reference, q);
             this.globalState = globalState;
             globalState.register(this);
         }

         void assertNotReleased()
         {
             if (DEBUG_ENABLED && released == 1)
                 debug.log(toString());
             assert released == 0;
         }

         Throwable ensureReleased(Throwable accumulate)
         {
             if (releasedUpdater.getAndSet(this, 1) == 0)
             {
                 accumulate = globalState.release(this, accumulate);
                 if (DEBUG_ENABLED)
                     debug.deallocate();
             }
             return accumulate;
         }

         void release(boolean leak)
         {
             if (!releasedUpdater.compareAndSet(this, 0, 1))
             {
                 if (!leak)
                 {
                     String id = this.toString();
                     logger.error("BAD RELEASE: attempted to release a reference ({}) that has already been released", id);
                     if (DEBUG_ENABLED)
                         debug.log(id);
                     throw new IllegalStateException("Attempted to release a reference that has already been released");
                 }
                 return;
             }
             Throwable fail = globalState.release(this, null);
             if (leak)
             {
                 String id = this.toString();
                 logger.error("LEAK DETECTED: a reference ({}) to {} was not released before the reference was garbage collected", id, globalState);
                 if (DEBUG_ENABLED)
                     debug.log(id);
             }
             else if (DEBUG_ENABLED)
             {
                 debug.deallocate();
             }
             if (fail != null)
                 logger.error("Error when closing {}", globalState, fail);
         }
     }

     static final class Debug
     {
         String allocateThread, deallocateThread;
         StackTraceElement[] allocateTrace, deallocateTrace;
         Debug()
         {
             Thread thread = Thread.currentThread();
             allocateThread = thread.toString();
             allocateTrace = thread.getStackTrace();
         }
         synchronized void deallocate()
         {
             Thread thread = Thread.currentThread();
             deallocateThread = thread.toString();
             deallocateTrace = thread.getStackTrace();
         }
         synchronized void log(String id)
         {
             logger.error("Allocate trace {}:\n{}", id, print(allocateThread, allocateTrace));
             if (deallocateThread != null)
                 logger.error("Deallocate trace {}:\n{}", id, print(deallocateThread, deallocateTrace));
         }
         String print(String thread, StackTraceElement[] trace)
         {
             StringBuilder sb = new StringBuilder();
             sb.append(thread);
             sb.append("\n");
             for (StackTraceElement element : trace)
             {
                 sb.append("\tat ");
                 sb.append(element );
                 sb.append("\n");
             }
             return sb.toString();
         }
     }

     // the object that manages the actual cleaning up; this does not reference the target object
     // so that we can detect when references are lost to the resource itself, and still cleanup afterwards
     // the Tidy object MUST NOT contain any references to the object we are managing
     static final class GlobalState
     {
         // we need to retain a reference to each of the PhantomReference instances
         // we are using to track individual refs
         private final Collection<State> locallyExtant = new ConcurrentLinkedDeque<>();
         // the number of live refs
         private final AtomicInteger counts = new AtomicInteger();
         // the object to call to cleanup when our refs are all finished with
         private final Tidy tidy;

         GlobalState(Tidy tidy)
         {
             this.tidy = tidy;
             globallyExtant.add(this);
         }

         void register(Ref.State ref)
         {
             locallyExtant.add(ref);
         }

         // increment ref count if not already tidied, and return success/failure
         boolean ref()
         {
             while (true)
             {
                 int cur = counts.get();
                 if (cur < 0)
                     return false;
                 if (counts.compareAndSet(cur, cur + 1))
                     return true;
             }
         }

         // release a single reference, and cleanup if no more are extant
         Throwable release(Ref.State ref, Throwable accumulate)
         {
             locallyExtant.remove(ref);
             if (-1 == counts.decrementAndGet())
             {
                 globallyExtant.remove(this);
                 try
                 {
                     if (tidy != null)
                         tidy.tidy();
                 }
                 catch (Throwable t)
                 {
                     accumulate = merge(accumulate, t);
                 }
             }
             return accumulate;
         }

         int count()
         {
             return 1 + counts.get();
         }

         public String toString()
         {
             if (tidy != null)
                 return tidy.getClass() + "@" + System.identityHashCode(tidy) + ":" + tidy.name();
             return "@" + System.identityHashCode(this);
         }
     }

     private static final Class<?>[] concurrentIterableClasses = new Class<?>[]
     {
         ConcurrentLinkedQueue.class,
         ConcurrentLinkedDeque.class,
         ConcurrentSkipListSet.class,
         CopyOnWriteArrayList.class,
         CopyOnWriteArraySet.class,
         DelayQueue.class,
         NonBlockingHashMap.class,
     };
     static final Set<Class<?>> concurrentIterables = Collections.newSetFromMap(new IdentityHashMap<>());
     private static final Set<GlobalState> globallyExtant = Collections.newSetFromMap(new ConcurrentHashMap<>());
     static final ReferenceQueue<Object> referenceQueue = new ReferenceQueue<>();
     private static final InfiniteLoopExecutor EXEC = new InfiniteLoopExecutor("Reference-Reaper", Ref::reapOneReference).start();
     static final ScheduledExecutorService STRONG_LEAK_DETECTOR = !DEBUG_ENABLED ? null : Executors.newScheduledThreadPool(1, new NamedThreadFactory("Strong-Reference-Leak-Detector"));
     static
     {
         if (DEBUG_ENABLED)
         {
             STRONG_LEAK_DETECTOR.scheduleAtFixedRate(new Visitor(), 1, 15, TimeUnit.MINUTES);
             STRONG_LEAK_DETECTOR.scheduleAtFixedRate(new StrongLeakDetector(), 2, 15, TimeUnit.MINUTES);
         }
         concurrentIterables.addAll(Arrays.asList(concurrentIterableClasses));
     }

     private static void reapOneReference() throws InterruptedException
     {
         Object obj = referenceQueue.remove(100);
         if (obj instanceof Ref.State)
         {
             ((Ref.State) obj).release(true);
         }
     }

     static final Deque<InProgressVisit> inProgressVisitPool = new ArrayDeque<InProgressVisit>();

     @SuppressWarnings({ "rawtypes", "unchecked" })
     static InProgressVisit newInProgressVisit(Object o, List<Field> fields, Field field, String name)
     {
         Preconditions.checkNotNull(o);
         InProgressVisit ipv = inProgressVisitPool.pollLast();
         if (ipv == null)
             ipv = new InProgressVisit();

         ipv.o = o;
         if (o instanceof Object[])
             ipv.collectionIterator = Arrays.asList((Object[])o).iterator();
         else if (o instanceof ConcurrentMap)
         {
             ipv.isMapIterator = true;
             ipv.collectionIterator = ((Map)o).entrySet().iterator();
         }
         else if (concurrentIterables.contains(o.getClass()) | o instanceof BlockingQueue)
             ipv.collectionIterator = ((Iterable)o).iterator();

         ipv.fields = fields;
         ipv.field = field;
         ipv.name = name;
         return ipv;
     }

     static void returnInProgressVisit(InProgressVisit ipv)
     {
         if (inProgressVisitPool.size() > 1024)
             return;
         ipv.name = null;
         ipv.fields = null;
         ipv.o = null;
         ipv.fieldIndex = 0;
         ipv.field = null;
         ipv.collectionIterator = null;
         ipv.mapEntryValue = null;
         ipv.isMapIterator = false;
         inProgressVisitPool.offer(ipv);
     }

     /*
      * Stack state for walking an object graph.
      * Field index is the index of the current field being fetched.
      */
     @SuppressWarnings({ "rawtypes"})
     static class InProgressVisit
     {
         String name;
         List<Field> fields;
         Object o;
         int fieldIndex = 0;
         Field field;

         //Need to know if Map.Entry should be returned or traversed as an object
         boolean isMapIterator;
         //If o is a ConcurrentMap, BlockingQueue, or Object[], this is populated with an iterator over the contents
         Iterator<Object> collectionIterator;
         //If o is a ConcurrentMap the entry set contains keys and values. The key is returned as the first child
         //And the associated value is stashed here and returned next
         Object mapEntryValue;

         private Field nextField()
         {
             if (fields.isEmpty())
                 return null;

             if (fieldIndex >= fields.size())
                 return null;

             Field retval = fields.get(fieldIndex);
             fieldIndex++;
             return retval;
         }

         Pair<Object, Field> nextChild() throws IllegalAccessException
         {
             //If the last child returned was a key from a map, the value from that entry is stashed
             //so it can be returned next
             if (mapEntryValue != null)
             {
                 Pair<Object, Field> retval = Pair.create(mapEntryValue, field);
                 mapEntryValue = null;
                 return retval;
             }

             //If o is a ConcurrentMap, BlockingQueue, or Object[], then an iterator will be stored to return the elements
             if (collectionIterator != null)
             {
                 if (!collectionIterator.hasNext())
                     return null;
                 Object nextItem = null;
                 //Find the next non-null element to traverse since returning null will cause the visitor to stop
                 while (collectionIterator.hasNext() && (nextItem = collectionIterator.next()) == null){}
                 if (nextItem != null)
                 {
                     if (isMapIterator & nextItem instanceof Map.Entry)
                     {
                         Map.Entry entry = (Map.Entry)nextItem;
                         mapEntryValue = entry.getValue();
                         return Pair.create(entry.getKey(), field);
                     }
                     return Pair.create(nextItem, field);
                 }
                 else
                 {
                     return null;
                 }
             }

             //Basic traversal of an object by its member fields
             //Don't return null values as that indicates no more objects
             while (true)
             {
                 Field nextField = nextField();
                 if (nextField == null)
                     return null;

                 //A weak reference isn't strongly reachable
                 //subclasses of WeakReference contain strong references in their fields, so those need to be traversed
                 //The weak reference fields are in the common Reference class base so filter those out
                 if (o instanceof WeakReference & nextField.getDeclaringClass() == Reference.class)
                     continue;

                 Object nextObject = nextField.get(o);
                 if (nextObject != null)
                     return Pair.create(nextField.get(o), nextField);
             }
         }

         @Override
         public String toString()
         {
             return field == null ? name : field.toString() + "-" + o.getClass().getName();
         }
     }

     static class Visitor implements Runnable
     {
         final Deque<InProgressVisit> path = new ArrayDeque<>();
         final Set<Object> visited = Collections.newSetFromMap(new IdentityHashMap<>());
         @VisibleForTesting
         int lastVisitedCount;
         @VisibleForTesting
         long iterations = 0;
         GlobalState visiting;
         Set<GlobalState> haveLoops;

         public void run()
         {
             try
             {
                 for (GlobalState globalState : globallyExtant)
                 {
                     if (globalState.tidy == null)
                         continue;

                     // do a graph exploration of the GlobalState, since it should be shallow; if it references itself, we have a problem
                     path.clear();
                     visited.clear();
                     lastVisitedCount = 0;
                     iterations = 0;
                     visited.add(globalState);
                     visiting = globalState;
                     traverse(globalState.tidy);
                 }
             }
             catch (Throwable t)
             {
                 t.printStackTrace();
             }
             finally
             {
                 lastVisitedCount = visited.size();
                 path.clear();
                 visited.clear();
             }
         }

         /*
          * Searches for an indirect strong reference between rootObject and visiting.
          */
         void traverse(final RefCounted.Tidy rootObject)
         {
             path.offer(newInProgressVisit(rootObject, getFields(rootObject.getClass()), null, rootObject.name()));

             InProgressVisit inProgress = null;
             while (inProgress != null || !path.isEmpty())
             {
                 //If necessary fetch the next object to start tracing
                 if (inProgress == null)
                     inProgress = path.pollLast();

                 try
                 {
                     Pair<Object, Field> p = inProgress.nextChild();
                     Object child = null;
                     Field field = null;

                     if (p != null)
                     {
                         iterations++;
                         child = p.left;
                         field = p.right;
                     }

                     if (child != null && visited.add(child))
                     {
                         path.offer(inProgress);
                         inProgress = newInProgressVisit(child, getFields(child.getClass()), field, null);
                         continue;
                     }
                     else if (visiting == child)
                     {
                         if (haveLoops != null)
                             haveLoops.add(visiting);
                         NoSpamLogger.log(logger,
                                 NoSpamLogger.Level.ERROR,
                                 rootObject.getClass().getName(),
                                 1,
                                 TimeUnit.SECONDS,
                                 "Strong self-ref loop detected {}",
                                 path);
                     }
                     else if (child == null)
                     {
                         returnInProgressVisit(inProgress);
                         inProgress = null;
                         continue;
                     }
                 }
                 catch (IllegalAccessException e)
                 {
                     NoSpamLogger.log(logger, NoSpamLogger.Level.ERROR, 5, TimeUnit.MINUTES, "Could not fully check for self-referential leaks", e);
                 }
             }
         }
     }

     static final Map<Class<?>, List<Field>> fieldMap = new HashMap<>();
     static List<Field> getFields(Class<?> clazz)
     {
         if (clazz == null || clazz == PhantomReference.class || clazz == Class.class || java.lang.reflect.Member.class.isAssignableFrom(clazz))
             return emptyList();
         List<Field> fields = fieldMap.get(clazz);
         if (fields != null)
             return fields;
         fieldMap.put(clazz, fields = new ArrayList<>());
         for (Field field : clazz.getDeclaredFields())
         {
             if (field.getType().isPrimitive() || Modifier.isStatic(field.getModifiers()))
                 continue;
             field.setAccessible(true);
             fields.add(field);
         }
         fields.addAll(getFields(clazz.getSuperclass()));
         return fields;
     }

     public static class IdentityCollection
     {
         final Set<Tidy> candidates;
         public IdentityCollection(Set<Tidy> candidates)
         {
             this.candidates = candidates;
         }

         public void add(Ref<?> ref)
         {
             candidates.remove(ref.state.globalState.tidy);
         }
         public void add(SelfRefCounted<?> ref)
         {
             add(ref.selfRef());
         }
         public void add(SharedCloseable ref)
         {
             if (ref instanceof SharedCloseableImpl)
                 add((SharedCloseableImpl)ref);
         }
         public void add(SharedCloseableImpl ref)
         {
             add(ref.ref);
         }
         public void add(Memory memory)
         {
             if (memory instanceof SafeMemory)
                 ((SafeMemory) memory).addTo(this);
         }
     }

     private static class StrongLeakDetector implements Runnable
     {
         Set<Tidy> candidates = new HashSet<>();

         public void run()
         {
             final Set<Tidy> candidates = Collections.newSetFromMap(new IdentityHashMap<>());
             for (GlobalState state : globallyExtant)
                 candidates.add(state.tidy);
             removeExpected(candidates);
             this.candidates.retainAll(candidates);
             if (!this.candidates.isEmpty())
             {
                 List<String> names = new ArrayList<>();
                 for (Tidy tidy : this.candidates)
                     names.add(tidy.name());
                 logger.warn("Strong reference leak candidates detected: {}", names);
             }
             this.candidates = candidates;
         }

         private void removeExpected(Set<Tidy> candidates)
         {
             final Ref.IdentityCollection expected = new Ref.IdentityCollection(candidates);
             for (Keyspace ks : Keyspace.all())
             {
                 for (ColumnFamilyStore cfs : ks.getColumnFamilyStores())
                 {
                     View view = cfs.getTracker().getView();
                     for (SSTableReader reader : view.allKnownSSTables())
                         reader.addTo(expected);
                 }
             }
         }
     }

     @VisibleForTesting
     public static void shutdownReferenceReaper(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException
     {
         ExecutorUtils.shutdownNowAndWait(timeout, unit, EXEC, STRONG_LEAK_DETECTOR);
     }
 }
	/*
	*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*
	*/
	package org.apache.cassandra.utils.concurrent;

	import java.lang.ref.PhantomReference;
	import java.lang.ref.Reference;
	import java.lang.ref.ReferenceQueue;
	import java.lang.ref.WeakReference;
	import java.lang.reflect.Field;
	import java.lang.reflect.Modifier;
	import java.util.*;
	import java.util.concurrent.*;
	import java.util.concurrent.atomic.AtomicInteger;
	import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;

	import org.apache.cassandra.concurrent.InfiniteLoopExecutor;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import com.google.common.annotations.VisibleForTesting;
	import com.google.common.base.Preconditions;

	import org.apache.cassandra.concurrent.NamedThreadFactory;
	import org.apache.cassandra.db.ColumnFamilyStore;
	import org.apache.cassandra.db.Keyspace;
	import org.apache.cassandra.db.lifecycle.View;
	import org.apache.cassandra.io.sstable.format.SSTableReader;
	import org.apache.cassandra.io.util.Memory;
	import org.apache.cassandra.io.util.SafeMemory;
	import org.apache.cassandra.utils.ExecutorUtils;
	import org.apache.cassandra.utils.NoSpamLogger;
	import org.apache.cassandra.utils.Pair;
	import org.cliffc.high_scale_lib.NonBlockingHashMap;

	import static java.util.Collections.emptyList;
	import org.apache.cassandra.concurrent.InfiniteLoopExecutor.InterruptibleRunnable;

	import static org.apache.cassandra.utils.ExecutorUtils.awaitTermination;
	import static org.apache.cassandra.utils.ExecutorUtils.shutdownNow;
	import static org.apache.cassandra.utils.Throwables.maybeFail;
	import static org.apache.cassandra.utils.Throwables.merge;

	/**
	* An object that needs ref counting does the two following:
	* - defines a Tidy object that will cleanup once it's gone,
	* (this must retain no references to the object we're tracking (only its resources and how to clean up))
	* Then, one of two options:
	* 1) Construct a Ref directly pointing to it, and always use this Ref; or
	* 2)
	* - implements RefCounted
	* - encapsulates a Ref, we'll call selfRef, to which it proxies all calls to RefCounted behaviours
	* - users must ensure no references to the selfRef leak, or are retained outside of a method scope.
	* (to ensure the selfRef is collected with the object, so that leaks may be detected and corrected)
	*
	* This class' functionality is achieved by what may look at first glance like a complex web of references,
	* but boils down to:
	*
	* {@code
	* Target --> selfRef --> [Ref.State] <--> Ref.GlobalState --> Tidy
	* ^
	* \|
	* Ref ----------------------------------------
	* \|
	* Global -------------------------------------
	* }
	* So that, if Target is collected, Impl is collected and, hence, so is selfRef.
	*
	* Once ref or selfRef are collected, the paired Ref.State's release method is called, which if it had
	* not already been called will update Ref.GlobalState and log an error.
	*
	* Once the Ref.GlobalState has been completely released, the Tidy method is called and it removes the global reference
	* to itself so it may also be collected.
	*/
	public final class Ref<T> implements RefCounted<T>
	{
	static final Logger logger = LoggerFactory.getLogger(Ref.class);
	public static final boolean DEBUG_ENABLED = System.getProperty("cassandra.debugrefcount", "false").equalsIgnoreCase("true");

	final State state;
	final T referent;

	public Ref(T referent, Tidy tidy)
	{
	this.state = new State(new GlobalState(tidy), this, referenceQueue);
	this.referent = referent;
	}

	Ref(T referent, GlobalState state)
	{
	this.state = new State(state, this, referenceQueue);
	this.referent = referent;
	}

	/**
	* Must be called exactly once, when the logical operation for which this Ref was created has terminated.
	* Failure to abide by this contract will result in an error (eventually) being reported, assuming a
	* hard reference to the resource it managed is not leaked.
	*/
	public void release()
	{
	state.release(false);
	}

	public Throwable ensureReleased(Throwable accumulate)
	{
	return state.ensureReleased(accumulate);
	}

	public void ensureReleased()
	{
	maybeFail(state.ensureReleased(null));
	}

	public void close()
	{
	ensureReleased();
	}

	public T get()
	{
	state.assertNotReleased();
	return referent;
	}

	public Ref<T> tryRef()
	{
	return state.globalState.ref() ? new Ref<>(referent, state.globalState) : null;
	}

	public Ref<T> ref()
	{
	Ref<T> ref = tryRef();
	// TODO: print the last release as well as the release here
	if (ref == null)
	state.assertNotReleased();
	return ref;
	}

	public String printDebugInfo()
	{
	if (DEBUG_ENABLED)
	{
	state.debug.log(state.toString());
	return "Memory was freed by " + state.debug.deallocateThread;
	}
	return "Memory was freed";
	}

	/**
	* A convenience method for reporting:
	* @return the number of currently extant references globally, including the shared reference
	*/
	public int globalCount()
	{
	return state.globalState.count();
	}

	// similar to Ref.GlobalState, but tracks only the management of each unique ref created to the managed object
	// ensures it is only released once, and that it is always released
	static final class State extends PhantomReference<Ref>
	{
	final Debug debug = DEBUG_ENABLED ? new Debug() : null;
	final GlobalState globalState;
	private volatile int released;

	private static final AtomicIntegerFieldUpdater<State> releasedUpdater = AtomicIntegerFieldUpdater.newUpdater(State.class, "released");

	public State(final GlobalState globalState, Ref reference, ReferenceQueue<? super Ref> q)
	{
	super(reference, q);
	this.globalState = globalState;
	globalState.register(this);
	}

	void assertNotReleased()
	{
	if (DEBUG_ENABLED && released == 1)
	debug.log(toString());
	assert released == 0;
	}

	Throwable ensureReleased(Throwable accumulate)
	{
	if (releasedUpdater.getAndSet(this, 1) == 0)
	{
	accumulate = globalState.release(this, accumulate);
	if (DEBUG_ENABLED)
	debug.deallocate();
	}
	return accumulate;
	}

	void release(boolean leak)
	{
	if (!releasedUpdater.compareAndSet(this, 0, 1))
	{
	if (!leak)
	{
	String id = this.toString();
	logger.error("BAD RELEASE: attempted to release a reference ({}) that has already been released", id);
	if (DEBUG_ENABLED)
	debug.log(id);
	throw new IllegalStateException("Attempted to release a reference that has already been released");
	}
	return;
	}
	Throwable fail = globalState.release(this, null);
	if (leak)
	{
	String id = this.toString();
	logger.error("LEAK DETECTED: a reference ({}) to {} was not released before the reference was garbage collected", id, globalState);
	if (DEBUG_ENABLED)
	debug.log(id);
	}
	else if (DEBUG_ENABLED)
	{
	debug.deallocate();
	}
	if (fail != null)
	logger.error("Error when closing {}", globalState, fail);
	}
	}

	static final class Debug
	{
	String allocateThread, deallocateThread;
	StackTraceElement[] allocateTrace, deallocateTrace;
	Debug()
	{
	Thread thread = Thread.currentThread();
	allocateThread = thread.toString();
	allocateTrace = thread.getStackTrace();
	}
	synchronized void deallocate()
	{
	Thread thread = Thread.currentThread();
	deallocateThread = thread.toString();
	deallocateTrace = thread.getStackTrace();
	}
	synchronized void log(String id)
	{
	logger.error("Allocate trace {}:\n{}", id, print(allocateThread, allocateTrace));
	if (deallocateThread != null)
	logger.error("Deallocate trace {}:\n{}", id, print(deallocateThread, deallocateTrace));
	}
	String print(String thread, StackTraceElement[] trace)
	{
	StringBuilder sb = new StringBuilder();
	sb.append(thread);
	sb.append("\n");
	for (StackTraceElement element : trace)
	{
	sb.append("\tat ");
	sb.append(element );
	sb.append("\n");
	}
	return sb.toString();
	}
	}

	// the object that manages the actual cleaning up; this does not reference the target object
	// so that we can detect when references are lost to the resource itself, and still cleanup afterwards
	// the Tidy object MUST NOT contain any references to the object we are managing
	static final class GlobalState
	{
	// we need to retain a reference to each of the PhantomReference instances
	// we are using to track individual refs
	private final Collection<State> locallyExtant = new ConcurrentLinkedDeque<>();
	// the number of live refs
	private final AtomicInteger counts = new AtomicInteger();
	// the object to call to cleanup when our refs are all finished with
	private final Tidy tidy;

	GlobalState(Tidy tidy)
	{
	this.tidy = tidy;
	globallyExtant.add(this);
	}

	void register(Ref.State ref)
	{
	locallyExtant.add(ref);
	}

	// increment ref count if not already tidied, and return success/failure
	boolean ref()
	{
	while (true)
	{
	int cur = counts.get();
	if (cur < 0)
	return false;
	if (counts.compareAndSet(cur, cur + 1))
	return true;
	}
	}

	// release a single reference, and cleanup if no more are extant
	Throwable release(Ref.State ref, Throwable accumulate)
	{
	locallyExtant.remove(ref);
	if (-1 == counts.decrementAndGet())
	{
	globallyExtant.remove(this);
	try
	{
	if (tidy != null)
	tidy.tidy();
	}
	catch (Throwable t)
	{
	accumulate = merge(accumulate, t);
	}
	}
	return accumulate;
	}

	int count()
	{
	return 1 + counts.get();
	}

	public String toString()
	{
	if (tidy != null)
	return tidy.getClass() + "@" + System.identityHashCode(tidy) + ":" + tidy.name();
	return "@" + System.identityHashCode(this);
	}
	}

	private static final Class<?>[] concurrentIterableClasses = new Class<?>[]
	{
	ConcurrentLinkedQueue.class,
	ConcurrentLinkedDeque.class,
	ConcurrentSkipListSet.class,
	CopyOnWriteArrayList.class,
	CopyOnWriteArraySet.class,
	DelayQueue.class,
	NonBlockingHashMap.class,
	};
	static final Set<Class<?>> concurrentIterables = Collections.newSetFromMap(new IdentityHashMap<>());
	private static final Set<GlobalState> globallyExtant = Collections.newSetFromMap(new ConcurrentHashMap<>());
	static final ReferenceQueue<Object> referenceQueue = new ReferenceQueue<>();
	private static final InfiniteLoopExecutor EXEC = new InfiniteLoopExecutor("Reference-Reaper", Ref::reapOneReference).start();
	static final ScheduledExecutorService STRONG_LEAK_DETECTOR = !DEBUG_ENABLED ? null : Executors.newScheduledThreadPool(1, new NamedThreadFactory("Strong-Reference-Leak-Detector"));
	static
	{
	if (DEBUG_ENABLED)
	{
	STRONG_LEAK_DETECTOR.scheduleAtFixedRate(new Visitor(), 1, 15, TimeUnit.MINUTES);
	STRONG_LEAK_DETECTOR.scheduleAtFixedRate(new StrongLeakDetector(), 2, 15, TimeUnit.MINUTES);
	}
	concurrentIterables.addAll(Arrays.asList(concurrentIterableClasses));
	}

	private static void reapOneReference() throws InterruptedException
	{
	Object obj = referenceQueue.remove(100);
	if (obj instanceof Ref.State)
	{
	((Ref.State) obj).release(true);
	}
	}

	static final Deque<InProgressVisit> inProgressVisitPool = new ArrayDeque<InProgressVisit>();

	@SuppressWarnings({ "rawtypes", "unchecked" })
	static InProgressVisit newInProgressVisit(Object o, List<Field> fields, Field field, String name)
	{
	Preconditions.checkNotNull(o);
	InProgressVisit ipv = inProgressVisitPool.pollLast();
	if (ipv == null)
	ipv = new InProgressVisit();

	ipv.o = o;
	if (o instanceof Object[])
	ipv.collectionIterator = Arrays.asList((Object[])o).iterator();
	else if (o instanceof ConcurrentMap)
	{
	ipv.isMapIterator = true;
	ipv.collectionIterator = ((Map)o).entrySet().iterator();
	}
	else if (concurrentIterables.contains(o.getClass()) \| o instanceof BlockingQueue)
	ipv.collectionIterator = ((Iterable)o).iterator();

	ipv.fields = fields;
	ipv.field = field;
	ipv.name = name;
	return ipv;
	}

	static void returnInProgressVisit(InProgressVisit ipv)
	{
	if (inProgressVisitPool.size() > 1024)
	return;
	ipv.name = null;
	ipv.fields = null;
	ipv.o = null;
	ipv.fieldIndex = 0;
	ipv.field = null;
	ipv.collectionIterator = null;
	ipv.mapEntryValue = null;
	ipv.isMapIterator = false;
	inProgressVisitPool.offer(ipv);
	}

	/*
	* Stack state for walking an object graph.
	* Field index is the index of the current field being fetched.
	*/
	@SuppressWarnings({ "rawtypes"})
	static class InProgressVisit
	{
	String name;
	List<Field> fields;
	Object o;
	int fieldIndex = 0;
	Field field;

	//Need to know if Map.Entry should be returned or traversed as an object
	boolean isMapIterator;
	//If o is a ConcurrentMap, BlockingQueue, or Object[], this is populated with an iterator over the contents
	Iterator<Object> collectionIterator;
	//If o is a ConcurrentMap the entry set contains keys and values. The key is returned as the first child
	//And the associated value is stashed here and returned next
	Object mapEntryValue;

	private Field nextField()
	{
	if (fields.isEmpty())
	return null;

	if (fieldIndex >= fields.size())
	return null;

	Field retval = fields.get(fieldIndex);
	fieldIndex++;
	return retval;
	}

	Pair<Object, Field> nextChild() throws IllegalAccessException
	{
	//If the last child returned was a key from a map, the value from that entry is stashed
	//so it can be returned next
	if (mapEntryValue != null)
	{
	Pair<Object, Field> retval = Pair.create(mapEntryValue, field);
	mapEntryValue = null;
	return retval;
	}

	//If o is a ConcurrentMap, BlockingQueue, or Object[], then an iterator will be stored to return the elements
	if (collectionIterator != null)
	{
	if (!collectionIterator.hasNext())
	return null;
	Object nextItem = null;
	//Find the next non-null element to traverse since returning null will cause the visitor to stop
	while (collectionIterator.hasNext() && (nextItem = collectionIterator.next()) == null){}
	if (nextItem != null)
	{
	if (isMapIterator & nextItem instanceof Map.Entry)
	{
	Map.Entry entry = (Map.Entry)nextItem;
	mapEntryValue = entry.getValue();
	return Pair.create(entry.getKey(), field);
	}
	return Pair.create(nextItem, field);
	}
	else
	{
	return null;
	}
	}

	//Basic traversal of an object by its member fields
	//Don't return null values as that indicates no more objects
	while (true)
	{
	Field nextField = nextField();
	if (nextField == null)
	return null;

	//A weak reference isn't strongly reachable
	//subclasses of WeakReference contain strong references in their fields, so those need to be traversed
	//The weak reference fields are in the common Reference class base so filter those out
	if (o instanceof WeakReference & nextField.getDeclaringClass() == Reference.class)
	continue;

	Object nextObject = nextField.get(o);
	if (nextObject != null)
	return Pair.create(nextField.get(o), nextField);
	}
	}

	@Override
	public String toString()
	{
	return field == null ? name : field.toString() + "-" + o.getClass().getName();
	}
	}

	static class Visitor implements Runnable
	{
	final Deque<InProgressVisit> path = new ArrayDeque<>();
	final Set<Object> visited = Collections.newSetFromMap(new IdentityHashMap<>());
	@VisibleForTesting
	int lastVisitedCount;
	@VisibleForTesting
	long iterations = 0;
	GlobalState visiting;
	Set<GlobalState> haveLoops;

	public void run()
	{
	try
	{
	for (GlobalState globalState : globallyExtant)
	{
	if (globalState.tidy == null)
	continue;

	// do a graph exploration of the GlobalState, since it should be shallow; if it references itself, we have a problem
	path.clear();
	visited.clear();
	lastVisitedCount = 0;
	iterations = 0;
	visited.add(globalState);
	visiting = globalState;
	traverse(globalState.tidy);
	}
	}
	catch (Throwable t)
	{
	t.printStackTrace();
	}
	finally
	{
	lastVisitedCount = visited.size();
	path.clear();
	visited.clear();
	}
	}

	/*
	* Searches for an indirect strong reference between rootObject and visiting.
	*/
	void traverse(final RefCounted.Tidy rootObject)
	{
	path.offer(newInProgressVisit(rootObject, getFields(rootObject.getClass()), null, rootObject.name()));

	InProgressVisit inProgress = null;
	while (inProgress != null \|\| !path.isEmpty())
	{
	//If necessary fetch the next object to start tracing
	if (inProgress == null)
	inProgress = path.pollLast();

	try
	{
	Pair<Object, Field> p = inProgress.nextChild();
	Object child = null;
	Field field = null;

	if (p != null)
	{
	iterations++;
	child = p.left;
	field = p.right;
	}

	if (child != null && visited.add(child))
	{
	path.offer(inProgress);
	inProgress = newInProgressVisit(child, getFields(child.getClass()), field, null);
	continue;
	}
	else if (visiting == child)
	{
	if (haveLoops != null)
	haveLoops.add(visiting);
	NoSpamLogger.log(logger,
	NoSpamLogger.Level.ERROR,
	rootObject.getClass().getName(),
	1,
	TimeUnit.SECONDS,
	"Strong self-ref loop detected {}",
	path);
	}
	else if (child == null)
	{
	returnInProgressVisit(inProgress);
	inProgress = null;
	continue;
	}
	}
	catch (IllegalAccessException e)
	{
	NoSpamLogger.log(logger, NoSpamLogger.Level.ERROR, 5, TimeUnit.MINUTES, "Could not fully check for self-referential leaks", e);
	}
	}
	}
	}

	static final Map<Class<?>, List<Field>> fieldMap = new HashMap<>();
	static List<Field> getFields(Class<?> clazz)
	{
	if (clazz == null \|\| clazz == PhantomReference.class \|\| clazz == Class.class \|\| java.lang.reflect.Member.class.isAssignableFrom(clazz))
	return emptyList();
	List<Field> fields = fieldMap.get(clazz);
	if (fields != null)
	return fields;
	fieldMap.put(clazz, fields = new ArrayList<>());
	for (Field field : clazz.getDeclaredFields())
	{
	if (field.getType().isPrimitive() \|\| Modifier.isStatic(field.getModifiers()))
	continue;
	field.setAccessible(true);
	fields.add(field);
	}
	fields.addAll(getFields(clazz.getSuperclass()));
	return fields;
	}

	public static class IdentityCollection
	{
	final Set<Tidy> candidates;
	public IdentityCollection(Set<Tidy> candidates)
	{
	this.candidates = candidates;
	}

	public void add(Ref<?> ref)
	{
	candidates.remove(ref.state.globalState.tidy);
	}
	public void add(SelfRefCounted<?> ref)
	{
	add(ref.selfRef());
	}
	public void add(SharedCloseable ref)
	{
	if (ref instanceof SharedCloseableImpl)
	add((SharedCloseableImpl)ref);
	}
	public void add(SharedCloseableImpl ref)
	{
	add(ref.ref);
	}
	public void add(Memory memory)
	{
	if (memory instanceof SafeMemory)
	((SafeMemory) memory).addTo(this);
	}
	}

	private static class StrongLeakDetector implements Runnable
	{
	Set<Tidy> candidates = new HashSet<>();

	public void run()
	{
	final Set<Tidy> candidates = Collections.newSetFromMap(new IdentityHashMap<>());
	for (GlobalState state : globallyExtant)
	candidates.add(state.tidy);
	removeExpected(candidates);
	this.candidates.retainAll(candidates);
	if (!this.candidates.isEmpty())
	{
	List<String> names = new ArrayList<>();
	for (Tidy tidy : this.candidates)
	names.add(tidy.name());
	logger.warn("Strong reference leak candidates detected: {}", names);
	}
	this.candidates = candidates;
	}

	private void removeExpected(Set<Tidy> candidates)
	{
	final Ref.IdentityCollection expected = new Ref.IdentityCollection(candidates);
	for (Keyspace ks : Keyspace.all())
	{
	for (ColumnFamilyStore cfs : ks.getColumnFamilyStores())
	{
	View view = cfs.getTracker().getView();
	for (SSTableReader reader : view.allKnownSSTables())
	reader.addTo(expected);
	}
	}
	}
	}

	@VisibleForTesting
	public static void shutdownReferenceReaper(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException
	{
	ExecutorUtils.shutdownNowAndWait(timeout, unit, EXEC, STRONG_LEAK_DETECTOR);
	}
	}