1 /*
   2  * Copyright (c) 2016, Oracle and/or its affiliates. All rights reserved.
   3  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   4  *
   5  * This code is free software; you can redistribute it and/or modify it
   6  * under the terms of the GNU General  License version 2 only, as
   7  * published by the Free Software Foundation.  Oracle designates this
   8  * particular file as subject to the "Classpath" exception as provided
   9  * by Oracle in the LICENSE file that accompanied this code.
  10  *
  11  * This code is distributed in the hope that it will be useful, but WITHOUT
  12  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  13  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General  License
  14  * version 2 for more details (a copy is included in the LICENSE file that
  15  * accompanied this code).
  16  *
  17  * You should have received a copy of the GNU General  License version
  18  * 2 along with this work; if not, write to the Free Software Foundation,
  19  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
  20  *
  21  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
  22  * or visit www.oracle.com if you need additional information or have any
  23  * questions.
  24  */
  25 package java.net.http;
  26 
  27 import java.nio.Buffer;
  28 import java.util.concurrent.BlockingQueue;
  29 import java.util.concurrent.LinkedBlockingQueue;
  30 import java.util.concurrent.atomic.AtomicInteger;
  31 import java.util.function.Supplier;
  32 
  33 import static java.lang.System.Logger.Level.TRACE;
  34 import static java.net.http.Shared.duplicate;
  35 import static java.net.http.Utils.logger;
  36 import static java.util.Objects.requireNonNull;
  37 
  38 final class SharedPool<T extends Buffer> implements Supplier<Shared<T>> {
  39 
  40     private final Supplier<T> factory;
  41     private final BlockingQueue<T> queue;
  42 
  43     SharedPool(Supplier<T> factory, int maxPoolSize) {
  44         this.factory = requireNonNull(factory);
  45         this.queue = new LinkedBlockingQueue<>(maxPoolSize);
  46     }
  47 
  48     @Override
  49     public Pooled get() {
  50         T b = queue.poll();
  51         if (b == null) {
  52             logger.log(TRACE, "Pool {0} contains no free buffers", this);
  53             b = requireNonNull(factory.get());
  54         }
  55         Pooled buf = new Pooled(new AtomicInteger(1), b, duplicate(b));
  56         logger.log(TRACE, "Pool {0} created new buffer {1}", this, buf);
  57         return buf;
  58     }
  59 
  60     private void put(Pooled b) {
  61         assert b.disposed.get() && b.refCount.get() == 0
  62                 : Utils.dump(b.disposed, b.refCount, b);
  63         b.shared.clear();
  64         boolean accepted = queue.offer(b.getShared());
  65         if (logger.isLoggable(TRACE)) {
  66             if (accepted) {
  67                 logger.log(TRACE, "Pool {0} accepted {1}", this, b);
  68             } else {
  69                 logger.log(TRACE, "Pool {0} discarded {1}", this, b);
  70             }
  71         }
  72     }
  73 
  74     @Override
  75     public String toString() {
  76         return super.toString() + "[queue.size=" + queue.size() + "]";
  77     }
  78 
  79     private final class Pooled extends Shared<T> {
  80 
  81         private final AtomicInteger refCount;
  82         private final T shared;
  83 
  84         private Pooled(AtomicInteger refCount, T shared, T region) {
  85             super(region);
  86             this.refCount = refCount;
  87             this.shared = shared;
  88         }
  89 
  90         private T getShared() {
  91             return shared;
  92         }
  93 
  94         @Override
  95         @SuppressWarnings("unchecked")
  96         public Pooled share(final int pos, final int limit) {
  97             synchronized (this) {
  98                 T buffer = buffer();
  99                 checkRegion(pos, limit, buffer);
 100                 final int oldPos = buffer.position();
 101                 final int oldLimit = buffer.limit();
 102                 select(pos, limit, buffer);
 103                 T slice = Shared.slice(buffer);
 104                 select(oldPos, oldLimit, buffer);
 105                 referenceAndGetCount();
 106                 Pooled buf = new Pooled(refCount, shared, slice);
 107                 logger.log(TRACE, "Shared {0} from {1}", buf, this);
 108                 return buf;
 109             }
 110         }
 111 
 112         @Override
 113         public void dispose() {
 114             logger.log(TRACE, "Disposed {0}", this);
 115             super.dispose();
 116             if (dereferenceAndGetCount() == 0) {
 117                 SharedPool.this.put(this);
 118             }
 119         }
 120 
 121         private int referenceAndGetCount() {
 122             return refCount.updateAndGet(n -> {
 123                 if (n != Integer.MAX_VALUE) {
 124                     return n + 1;
 125                 } else {
 126                     throw new IllegalArgumentException
 127                             ("Too many references: " + this);
 128                 }
 129             });
 130         }
 131 
 132         private int dereferenceAndGetCount() {
 133             return refCount.updateAndGet(n -> {
 134                 if (n > 0) {
 135                     return n - 1;
 136                 } else {
 137                     throw new InternalError();
 138                 }
 139             });
 140         }
 141 
 142         @Override
 143         public String toString() {
 144             return Utils.toStringSimple(this) + "[" + Utils.toString(buffer)
 145                     + "[refCount=" + refCount + ", disposed=" + disposed + "]]";
 146         }
 147     }
 148 }