code: @Override
      public void close() {
          // waiting phase
          int count = 0, last = 0;
          do {
              last = count;
              try { Thread.sleep(500); } catch (Exception x) {}
              count = 0;
              for (Pair<T, Future<?>> pair: _pairs) {
                  if (pair.second.isDone()) ++count;
              }
          } while ((_waiting == null || _waiting.invoke(count, count - last)) && count < _pairs.size());
          if (_waiting != null) _waiting.invoke(count, -1);
  
          // closing phase
          for (Pair<T, Future<?>> pair: _pairs) {
              if (!pair.second.isDone()) {
                  try {
                      if (_closing != null) _closing.invoke(pair.first, null);
                      pair.second.cancel(true);
                      pair.first.close();
                  } catch (Exception x) {
                      if (_closing != null) _closing.invoke(pair.first, x);
                  }
              }
          }
      }
docs: Closes all Futures.
  Waits for futures to complete, then closes those still running after the "waiting phase" has ended.