presto-main/src/main/java/io/prestosql/execution/buffer/OutputBufferMemoryManager.java/OutputBufferMemoryManager/updateMemoryUsage
class OutputBufferMemoryManager:    public synchronized void updateMemoryUsage(long bytesAdded)
    {
        Optional<LocalMemoryContext> systemMemoryContext = getSystemMemoryContext();

        // If closed is true, that means the task is completed. In that state,
        // the output buffers already ignore the newly added pages, and therefore
        // we can also safely ignore any calls after OutputBufferMemoryManager is closed.
        // If the systemMemoryContext doesn't exist, the task is probably already
        // aborted, so we can just return (see the comment in getSystemMemoryContext()).
        if (closed || !systemMemoryContext.isPresent()) {
            return;
        }

        long currentBufferedBytes = bufferedBytes.addAndGet(bytesAdded);
        peakMemoryUsage.accumulateAndGet(currentBufferedBytes, Math::max);
        this.blockedOnMemory = systemMemoryContext.get().setBytes(currentBufferedBytes);
        if (!isBufferFull() && !isBlockedOnMemory() && !bufferBlockedFuture.isDone()) {
            // Complete future in a new thread to avoid making a callback on the caller thread.
            // This make is easier for callers to use this class since they can update the memory
            // usage while holding locks.
            SettableFuture<?> future = this.bufferBlockedFuture;
            notificationExecutor.execute(() -> future.set(null));
            return;
        }
        this.blockedOnMemory.addListener(this::onMemoryAvailable, notificationExecutor);
    }