Segment

Segment在 Okio 中作为数据缓冲的载体，一个 Segment 的数据缓冲大小为 8192，即 8k。每一个 Segment 都有前驱和后继结点，也就是说 Sement 是一个双向链表链表,准确的来说是一个双向循环链表。读取数据从 Segment 头结点读取，写数据从 Segment 尾结点写。

Okio 中引入池的概念也就是源码中SegmentPool的实现。SegmentPool 负责 Segment 创建和销毁，SegmentPool 最大可以缓存 8 个 Segment。

SegmentPool 是一个静态方法，因此也就是全局缓存只有 64 kb；

final class Segment {
  //Segment的容量，最大为8kb
  static final int SIZE = 8192;

  //如果Segment中字节数 > SHARE_MINIMUM时（大Segment），就可以共享，不能添加到SegmentPool
  static final int SHARE_MINIMUM = 1024;
  //存储的数据
  final byte[] data;

  //下一次读取的开始位置
  int pos;

 //写入的开始位置
  int limit;

  //当前Segment是否可以共享
  boolean shared;

  //data是否仅当前Segment独有，不share
  boolean owner;

  //后继节点
  Segment next;

  //前驱节点
  Segment prev;

  ...

  //移除当前Segment
  public final @Nullable Segment pop() {
    Segment result = next != this ? next : null;
    prev.next = next;
    next.prev = prev;
    next = null;
    prev = null;
    return result;
  }

  //在当前节点后添加一个新的节点
  public final Segment push(Segment segment) {
    segment.prev = this;
    segment.next = next;
    next.prev = segment;
    next = segment;
    return segment;
  }

  //将当前Segment分裂成2个Segment结点。前面结点pos~limit数据范围是[pos..pos+byteCount)，后面结点pos~limit数据范围是[pos+byteCount..limit)
  public final Segment split(int byteCount) {
    if (byteCount <= 0 || byteCount > limit - pos) throw new IllegalArgumentException();
    Segment prefix;

    //如果字节数大于SHARE_MINIMUM则拆分成共享节点
    if (byteCount >= SHARE_MINIMUM) {
      prefix = sharedCopy();
    } else {
      prefix = SegmentPool.take();
      System.arraycopy(data, pos, prefix.data, 0, byteCount);
    }

    prefix.limit = prefix.pos + byteCount;
    pos += byteCount;
    prev.push(prefix);
    return prefix;
  }

  //当前Segment结点和prev前驱结点合并成一个Segment，统一合并到prev，然后当前Segment结点从双向链表移除并添加到SegmentPool复用。当然合并的前提是：2个Segment的字节总和不超过8K。合并后可能会移动pos、limit
  public final void compact() {
    if (prev == this) throw new IllegalStateException();
    if (!prev.owner) return; // Cannot compact: prev isn't writable.
    int byteCount = limit - pos;
    int availableByteCount = SIZE - prev.limit + (prev.shared ? 0 : prev.pos);
    if (byteCount > availableByteCount) return; // Cannot compact: not enough writable space.
    writeTo(prev, byteCount);
    pop();
    SegmentPool.recycle(this);
  }

  //从当前节点移动byteCount个字节到sink中
  public final void writeTo(Segment sink, int byteCount) {
    if (!sink.owner) throw new IllegalArgumentException();
    if (sink.limit + byteCount > SIZE) {
      // We can't fit byteCount bytes at the sink's current position. Shift sink first.
      if (sink.shared) throw new IllegalArgumentException();
      if (sink.limit + byteCount - sink.pos > SIZE) throw new IllegalArgumentException();
      System.arraycopy(sink.data, sink.pos, sink.data, 0, sink.limit - sink.pos);
      sink.limit -= sink.pos;
      sink.pos = 0;
    }

    System.arraycopy(data, pos, sink.data, sink.limit, byteCount);
    sink.limit += byteCount;
    pos += byteCount;
  }
}

final class SegmentPool {
    //SegmentPool的最大容量
    static final long MAX_SIZE = 64 * 1024; // 64 KiB.

    //后继节点
    static Segment next;

    //当前池内的总字节数
    static long byteCount;

    private SegmentPool() {
    }
    //从池中获取一个Segment对象
    static Segment take() {
        synchronized (SegmentPool.class) {
            if (next != null) {
                Segment result = next;
                next = result.next;
                result.next = null;
                byteCount -= Segment.SIZE;
                return result;
            }
        }
        return new Segment(); // Pool is empty. Don't zero-fill while holding a lock.
    }
    //将Segment状态初始化并放入池中
    static void recycle(Segment segment) {
        if (segment.next != null || segment.prev != null) throw new IllegalArgumentException();
        if (segment.shared) return; // This segment cannot be recycled.
        synchronized (SegmentPool.class) {
            if (byteCount + Segment.SIZE > MAX_SIZE) return; // Pool is full.
            byteCount += Segment.SIZE;
            segment.next = next;
            segment.pos = segment.limit = 0;
            next = segment;
        }
    }
} 

    @Override
    public Buffer write(byte[] source, int offset, int byteCount) {
        if (source == null) throw new IllegalArgumentException("source == null");
        // 检测参数的合法性
        checkOffsetAndCount(source.length, offset, byteCount);

        // 计算 source 要写入的最后一个字节的 index 值
        int limit = offset + byteCount;
        while (offset < limit) {
            // 获取循环链表尾部的一个 Segment
            Segment tail = writableSegment(1);
            // 计算最多可写入的字节
            int toCopy = Math.min(limit - offset, Segment.SIZE - tail.limit);
            // 把 source 复制到 data 中
            System.arraycopy(source, offset, tail.data, tail.limit, toCopy);
            // 调整写入的起始位置
            offset += toCopy;
            // 调整尾部Segment 的 limit 位置
            tail.limit += toCopy;
        }
        // 调整 Buffer 的 size 大小
        size += byteCount;
        return this;
    }

  /**
   * Returns a tail segment that we can write at least {@code minimumCapacity}
   * bytes to, creating it if necessary.
   */
  Segment writableSegment(int minimumCapacity) {
    if (minimumCapacity < 1 || minimumCapacity > Segment.SIZE) throw new IllegalArgumentException();

    // 如果链表的头指针为null，就会SegmentPool中取出一个
    if (head == null) {
      head = SegmentPool.take(); // Acquire a first segment.
      return head.next = head.prev = head;
    }

    // 获取前驱结点，也就是尾部结点
    Segment tail = head.prev;
    // 如果能写的字节数限制超过了8192，或者不是拥有者
    if (tail.limit + minimumCapacity > Segment.SIZE || !tail.owner) {
        // 从SegmentPool中获取一个Segment，插入到循环双链表当前结点的后面
      tail = tail.push(SegmentPool.take()); // Append a new empty segment to fill up.
    }
    return tail;
  }

    if (head == null) {
      head = SegmentPool.take(); // Acquire a first segment.
      return head.next = head.prev = head;
    }

  @Override 
  public int read(byte[] sink, int offset, int byteCount) {
    checkOffsetAndCount(sink.length, offset, byteCount);
    //取到Segment循环链表的表头
    Segment s = head;
    if (s == null) return -1;
    // 计算最多可写入的字节
    int toCopy = Math.min(byteCount, s.limit - s.pos);
    //将数据拷贝到链头的data字节数组当中
    System.arraycopy(s.data, s.pos, sink, offset, toCopy);
    
    //调整链头的data数组的起始postion和Buffer的size
    s.pos += toCopy;
    size -= toCopy;
    //pos等于limit的时候，从循环链表中移除该Segment并从SegmentPool中回收复用
    if (s.pos == s.limit) {
      head = s.pop();//移除的同时返回下一个Segment作为表头
      SegmentPool.recycle(s);
    }

    return toCopy;
  }

@Override 
public void write(Buffer source, long byteCount) {
    // Move bytes from the head of the source buffer to the tail of this buffer
    // while balancing two conflicting goals: don't waste CPU and don't waste
    // memory.
    //
    //
    // Don't waste CPU (ie. don't copy data around).
    //
    // Copying large amounts of data is expensive. Instead, we prefer to
    // reassign entire segments from one buffer to the other.
    //
    //
    // Don't waste memory.
    //
    // As an invariant, adjacent pairs of segments in a buffer should be at
    // least 50% full, except for the head segment and the tail segment.
    //
    // The head segment cannot maintain the invariant because the application is
    // consuming bytes from this segment, decreasing its level.
    //
    // The tail segment cannot maintain the invariant because the application is
    // producing bytes, which may require new nearly-empty tail segments to be
    // appended.
    //
    //
    // Moving segments between buffers
    //
    // When writing one buffer to another, we prefer to reassign entire segments
    // over copying bytes into their most compact form. Suppose we have a buffer
    // with these segment levels [91%, 61%]. If we append a buffer with a
    // single [72%] segment, that yields [91%, 61%, 72%]. No bytes are copied.
    //
    // Or suppose we have a buffer with these segment levels: [100%, 2%], and we
    // want to append it to a buffer with these segment levels [99%, 3%]. This
    // operation will yield the following segments: [100%, 2%, 99%, 3%]. That
    // is, we do not spend time copying bytes around to achieve more efficient
    // memory use like [100%, 100%, 4%].
    //
    // When combining buffers, we will compact adjacent buffers when their
    // combined level doesn't exceed 100%. For example, when we start with
    // [100%, 40%] and append [30%, 80%], the result is [100%, 70%, 80%].
    //
    //
    // Splitting segments
    //
    // Occasionally we write only part of a source buffer to a sink buffer. For
    // example, given a sink [51%, 91%], we may want to write the first 30% of
    // a source [92%, 82%] to it. To simplify, we first transform the source to
    // an equivalent buffer [30%, 62%, 82%] and then move the head segment,
    // yielding sink [51%, 91%, 30%] and source [62%, 82%].

    if (source == null) throw new IllegalArgumentException("source == null");
    if (source == this) throw new IllegalArgumentException("source == this");
    checkOffsetAndCount(source.size, 0, byteCount);

    while (byteCount > 0) {
      // Is a prefix of the source's head segment all that we need to move?
     // 如果 Source Buffer 的头结点可用字节数大于要写出的字节数
      if (byteCount < (source.head.limit - source.head.pos)) {
        //取到当前buffer的尾节点
        Segment tail = head != null ? head.prev : null;
        // 如果尾部结点有足够空间可以写数据，并且这个结点是底层数组的拥有者
        if (tail != null && tail.owner
            && (byteCount + tail.limit - (tail.shared ? 0 : tail.pos) <= Segment.SIZE)) {
          // Our existing segments are sufficient. Move bytes from source's head to our tail.
          //source头结点的数据写入到当前尾节点中，然后就直接结束返回了
          source.head.writeTo(tail, (int) byteCount);
          source.size -= byteCount;
          size += byteCount;
          return;
        } else {
          // We're going to need another segment. Split the source's head
          // segment in two, then move the first of those two to this buffer.
          //如果尾节点空间不足或者不是持有者，这时就需要把 Source Buffer 的头结点分割为两个 Segment，
          //然后将source的头指针更新为分割后的第一个Segment, 如[92%, 82%]变成[30%, 62%, 82%]这样
          source.head = source.head.split((int) byteCount);
        }
      }

      // Remove the source's head segment and append it to our tail.
      //从 Source Buffer 的链表中移除头结点, 并加入到当前Buffer的链尾
      Segment segmentToMove = source.head;
      long movedByteCount = segmentToMove.limit - segmentToMove.pos;
      //移除操作，并移动更新source中的head
      source.head = segmentToMove.pop();
       // 如果当前buffer的头结点为 null，则头结点直接指向source的头结点，初始化双向链表
      if (head == null) {
        head = segmentToMove;
        head.next = head.prev = head;
      } else {
        //否则就把Source Buffer的 head 加入到当前Buffer的链尾
        Segment tail = head.prev;
        tail = tail.push(segmentToMove);//压入链尾，并更新尾节点
        tail.compact();//尾节点尝试合并，如果合并成功，则尾节点会被SegmentPool回收掉
      }
      source.size -= movedByteCount;
      size += movedByteCount;
      byteCount -= movedByteCount;
    }
  }

public class AsyncTimeout extends Timeout {
    //头结点，占位使用
    static
    AsyncTimeout head;

    //是否在链表中
    private boolean inQueue;

    //后继节点
    private
    AsyncTimeout next;

    //超时时间
    private long timeoutAt;
    //把当前AsyncTimeout对象加入节点
    public final void enter() {
        ...
        scheduleTimeout(this, timeoutNanos, hasDeadline);
    }

    private static synchronized void scheduleTimeout(
            AsyncTimeout node, long timeoutNanos, boolean hasDeadline) {
        //创建占位头结点并开启子线程
        if (head == null) {
            head = new AsyncTimeout();
            new Watchdog().start();
        }

        ...

        //插入到链表中，按照时间长短进行排序，等待事件越长越靠后
        for (AsyncTimeout prev = head; true; prev = prev.next) {
            if (prev.next == null || remainingNanos < prev.next.remainingNanos(now)) {
                node.next = prev.next;
                prev.next = node;
                if (prev == head) {
                    AsyncTimeout.class.notify(); // Wake up the watchdog when inserting at the front.
                }
                break;
            }
        }
    }

    //从链表中移除节点
    public final boolean exit() {
        if (!inQueue) return false;
        inQueue = false;
        return cancelScheduledTimeout(this);
    }

    //执行真正的移除操作
    private static synchronized boolean cancelScheduledTimeout(AsyncTimeout node) {
        // Remove the node from the linked list.
        for (AsyncTimeout prev = head; prev != null; prev = prev.next) {
            if (prev.next == node) {
                prev.next = node.next;
                node.next = null;
                return false;
            }
        }

        // The node wasn't found in the linked list: it must have timed out!
        return true;
    }

    //在子类中重写了该方法，主要是进行socket的关闭
    protected void timedOut() {
    }

    //监听节点是否超时的子线程
    private static final class Watchdog extends Thread {
        Watchdog() {
            super("Okio Watchdog");
            setDaemon(true);
        }

        public void run() {
            while (true) {
                try {
                    AsyncTimeout timedOut;
                    synchronized (AsyncTimeout.class) {
                        timedOut = awaitTimeout();
                        //代表头结点的后继节点已超时，
                        if (timedOut == null) continue;
                        //除头结点外没有任何其他节点
                        if (timedOut == head) {
                            head = null;
                            return;
                        }
                    }

                    //关闭socket
                    timedOut.timedOut();
                } catch (InterruptedException ignored) {
                }
            }
        }
    }

    
    static AsyncTimeout awaitTimeout() throws InterruptedException {
        AsyncTimeout node = head.next;
        //除了头结点外没有任何其他节点
        if (node == null) {
            long startNanos = System.nanoTime();
            AsyncTimeout.class.wait(IDLE_TIMEOUT_MILLIS);
            return head.next == null && (System.nanoTime() - startNanos) >= IDLE_TIMEOUT_NANOS
                    ? head  // The idle timeout elapsed.
                    : null; // The situation has changed.
        }

        long waitNanos = node.remainingNanos(System.nanoTime());

        //进行等待
        if (waitNanos > 0) {
            //等待
            long waitMillis = waitNanos / 1000000L;
            waitNanos -= (waitMillis * 1000000L);
            AsyncTimeout.class.wait(waitMillis, (int) waitNanos);
            return null;
        }

        //代表node节点已超时
        head.next = node.next;
        node.next = null;
        return node;
    }
}

 默认都是未设置超时时间的，需要我们自己来设置，同步及异步的超时时间设置方式是一样的，通过下面代码即可。

    sink.timeout().deadline(1, TimeUnit.SECONDS);
    source.timeout().deadline(1,TimeUnit.MILLISECONDS);

public final class Pipe {
  //Pipe的最大容量
  final long maxBufferSize;
  //Pipe对应的Buffer
  final Buffer buffer = new Buffer();
  boolean sinkClosed;
  boolean sourceClosed;
  //写入流，对应着生产者
  private final Sink sink = new PipeSink();
  //读取流，对应着消费者
  private final Source source = new PipeSource();

  public Pipe(long maxBufferSize) {
    //最大容量不能小于1
    if (maxBufferSize < 1L) {
      throw new IllegalArgumentException("maxBufferSize < 1: " + maxBufferSize);
    }
    this.maxBufferSize = maxBufferSize;
  }
  ...
  //写入数据到Pipe中
  final class PipeSink implements Sink {
    final Timeout timeout = new Timeout();

    @Override public void write(Buffer source, long byteCount) throws IOException {
      synchronized (buffer) {
        ...

        while (byteCount > 0) {
          ...

          long bufferSpaceAvailable = maxBufferSize - buffer.size();
          if (bufferSpaceAvailable == 0) {
            //buffer中，没有剩余空间，等待消费者消费
            timeout.waitUntilNotified(buffer); // Wait until the source drains the buffer.
            continue;
          }

          long bytesToWrite = Math.min(bufferSpaceAvailable, byteCount);
　　　　　　// 将数据写入到 buffer的 data 中
          buffer.write(source, bytesToWrite);
          // 还需要写多少数据
          byteCount -= bytesToWrite;
          //通知buffer，有新的数据了，
          buffer.notifyAll(); // Notify the source that it can resume reading.
        }
      }
    }

    ...
  }
  //从 Pipe 中读取数据
  final class PipeSource implements Source {
    final Timeout timeout = new Timeout();

    @Override public long read(Buffer sink, long byteCount) throws IOException {
      synchronized (buffer) {
        ...

        while (buffer.size() == 0) {
          if (sinkClosed) return -1L;
          //Pipe中没有数据，等待生产者写入
          timeout.waitUntilNotified(buffer); // Wait until the sink fills the buffer.
        }

        long result = buffer.read(sink, byteCount);
        buffer.notifyAll(); // Notify the sink that it can resume writing.
        return result;
      }
    }
    ...
  }
}

    public void pipe() throws IOException {
        //设置 Pipe 的容量为1024字节，即 1kb
        Pipe pipe = new Pipe(1024);
        new Thread(new Runnable() {
            @Override
            public void run() {
　　　　　　　　　　//1 pipe.source 会从 pipe 的 buffer 中读取数据
                try (BufferedSource bufferedSource = Okio.buffer(pipe.source())) {
                    //2 将Pipe中数据写入env4.txt这个文件中,；
                    bufferedSource.readAll(Okio.sink(new File("file/env4.txt")));
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }).start();

        new Thread(new Runnable() {
            @Override
            public void run() {
　　　　　　　　　　//3 pipe.sink 会将数据写入到 pipe 的 buffer 中
                try (BufferedSink bufferedSink = Okio.buffer(pipe.sink())) {
                    //4 将env3.txt中数据写入到Pipe中;
                    bufferedSink.writeAll(Okio.source(new File("file/env3.txt")));
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }