presto-main/src/main/java/io/prestosql/operator/exchange/LocalExchangeSource.java/LocalExchangeSource/close
class LocalExchangeSource:    public void close()
    {
        checkNotHoldsLock();

        List<PageReference> remainingPages = new ArrayList<>();
        SettableFuture<?> notEmptyFuture;
        synchronized (lock) {
            finishing = true;

            buffer.drainTo(remainingPages);
            bufferedBytes.addAndGet(-remainingPages.stream().mapToLong(PageReference::getRetainedSizeInBytes).sum());

            notEmptyFuture = this.notEmptyFuture;
            this.notEmptyFuture = NOT_EMPTY;
        }

        // free all the remaining pages
        remainingPages.forEach(PageReference::removePage);

        // notify readers outside of lock since this may result in a callback
        notEmptyFuture.set(null);

        // this will always fire the finished event
        checkState(isFinished(), "Expected buffer to be finished");
        checkFinished();
    }