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

        boolean added = false;
        SettableFuture<?> notEmptyFuture;
        synchronized (lock) {
            // ignore pages after finish
            if (!finishing) {
                // buffered bytes must be updated before adding to the buffer to assure
                // the count does not go negative
                bufferedBytes.addAndGet(pageReference.getRetainedSizeInBytes());
                buffer.add(pageReference);
                added = true;
            }

            // we just added a page (or we are finishing) so we are not empty
            notEmptyFuture = this.notEmptyFuture;
            this.notEmptyFuture = NOT_EMPTY;
        }

        if (!added) {
            // dereference the page outside of lock
            pageReference.removePage();
        }

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