// PipeWriter is similar to io.PipeWriter with wait group
type PipeWriter struct {
	*io.PipeWriter
	once sync.Once
	done func()
}

// CloseWithError close with supplied error the writer end.
func (w *PipeWriter) CloseWithError(err error) error {
	err = w.PipeWriter.CloseWithError(err)
	w.once.Do(func() {
		w.done()
	})
	return err
}

// PipeReader is similar to io.PipeReader with wait group
type PipeReader struct {

docker volume rm $(docker volume ls -q -f dangling=true) || true

# Stop two nodes, one of each pool, to check that all S3 calls work while quorum is still there
[ "${MODE}" == "pools" ] && docker-compose -f minio-${MODE}.yaml stop minio2
[ "${MODE}" == "pools" ] && docker-compose -f minio-${MODE}.yaml stop minio6

# Pause one node, to check that all S3 calls work while one node goes wrong

	if maxProcs < len(disks) {
		disks = disks[:maxProcs]
	}

	// Start one scanner per disk
	var wg sync.WaitGroup
	wg.Add(len(disks))

	for i := range disks {
		go func(i int) {
			defer wg.Done()
			disk := disks[i]

			for bucket := range bucketCh {
				select {
				case <-ctx.Done():
					return
				default:
				}

				// Load cache for bucket

  @CanIgnoreReturnValue
  boolean addEdge(EndpointPair<N> endpoints, E edge);

  /**
   * Removes {@code node} if it is present; all edges incident to {@code node} will also be removed.
   *
   * @return {@code true} if the network was modified as a result of this call
   */
  @CanIgnoreReturnValue
  boolean removeNode(N node);

  /**
   * Removes {@code edge} from this network, if it is present.
   *

  protected AbstractIterator() {}

  private enum State {
    READY,
    NOT_READY,
    DONE,
    FAILED,
  }

  @CheckForNull private T next;

  @CheckForNull
  protected abstract T computeNext();

  @CanIgnoreReturnValue
  @CheckForNull
  protected final T endOfData() {
    state = State.DONE;
    return null;
  }

  @Override
  public final boolean hasNext() {

  @CanIgnoreReturnValue
  boolean addEdge(EndpointPair<N> endpoints, E edge);

  /**
   * Removes {@code node} if it is present; all edges incident to {@code node} will also be removed.
   *
   * @return {@code true} if the network was modified as a result of this call
   */
  @CanIgnoreReturnValue
  boolean removeNode(N node);

  /**
   * Removes {@code edge} from this network, if it is present.
   *

}

var _ yaml.Unmarshaler = &BatchJobKV{}

// UnmarshalYAML - BatchJobKV extends default unmarshal to extract line, col information.
func (kv *BatchJobKV) UnmarshalYAML(val *yaml.Node) error {
	type jobKV BatchJobKV
	var tmp jobKV
	err := val.Decode(&tmp)
	if err != nil {
		return err
	}
	*kv = BatchJobKV(tmp)
	kv.line, kv.col = val.Line, val.Column
	return nil
}

}

// HasLogListeners returns true if console log listeners are registered
// for this node or peers
func (sys *HTTPConsoleLoggerSys) HasLogListeners() bool {
	return sys != nil && sys.pubsub.Subscribers() > 0
}

// Subscribe starts console logging for this node.
func (sys *HTTPConsoleLoggerSys) Subscribe(subCh chan log.Info, doneCh <-chan struct{}, node string, last int, logKind madmin.LogMask, filter func(entry log.Info) bool) error {

                ids.add(0, artifact);
                node = node.parent;
            }
            trail = ids;
        }
        return trail;
    }

    public boolean isResolved() {
        return children != null;
    }

    /**
     * Test whether the node is direct or transitive dependency.
     *
     * @return whether the node is direct or transitive dependency
     */

	wrkr := es.getWorkerCh(task.OpHash())
	if wrkr == nil {
		es.stats.missedExpiryTasks.Add(1)
		return
	}
	select {
	case <-GlobalContext.Done():
	case wrkr <- task:
	default:
		es.stats.missedExpiryTasks.Add(1)
	}
}

// globalExpiryState is the per-node instance which manages all ILM expiry tasks.
var globalExpiryState *expiryState

// newExpiryState creates an expiryState with buffered channels allocated for