<commit_before>/**
 * Copyright (c) 2012-2013 DataTorrent, Inc. All rights reserved.
 */
package com.datatorrent.stram.stream;

import java.net.InetSocketAddress;
import java.util.ArrayDeque;
import java.util.HashMap;
import java.util.concurrent.atomic.AtomicLong;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.datatorrent.api.Sink;
import com.datatorrent.api.StreamCodec;

import com.datatorrent.bufferserver.client.Subscriber;
import com.datatorrent.bufferserver.util.Codec;
import com.datatorrent.common.util.Slice;
import com.datatorrent.netlet.EventLoop;
import com.datatorrent.netlet.util.CircularBuffer;
import com.datatorrent.stram.codec.StatefulStreamCodec;
import com.datatorrent.stram.codec.StatefulStreamCodec.DataStatePair;
import com.datatorrent.stram.engine.ByteCounterStream;
import com.datatorrent.stram.engine.StreamContext;
import com.datatorrent.stram.engine.SweepableReservoir;
import com.datatorrent.stram.engine.WindowGenerator;
import com.datatorrent.stram.tuple.*;

/**
 * Implement tuple flow from buffer server to the node in a logical stream<p>
 * <br>
 * Extends SocketInputStream as buffer server and node communicate via a socket<br>
 * This buffer server is a read instance of a stream and takes care of connectivity with upstream buffer server<br>
 *
 * @since 0.3.2
 */
public class BufferServerSubscriber extends Subscriber implements ByteCounterStream
{
  private boolean suspended;
  private long baseSeconds;
  protected StreamCodec<Object> serde;
  protected StatefulStreamCodec<Object> statefulSerde;
  protected EventLoop eventloop;
  private final DataStatePair dsp;
  CircularBuffer<Slice> offeredFragments;
  CircularBuffer<Slice> polledFragments;
  CircularBuffer<Slice> freeFragments;
  private final ArrayDeque<CircularBuffer<Slice>> backlog;
  private int lastWindowId = WindowGenerator.MAX_WINDOW_ID;
  private final AtomicLong readByteCount;

  public BufferServerSubscriber(String id, int queueCapacity)
  {
    super(id);
    this.reservoirs = new BufferReservoir[0];
    this.reservoirMap = new HashMap<String, BufferReservoir>();
    this.readByteCount = new AtomicLong(0);
    this.dsp = new DataStatePair();
    polledFragments = offeredFragments = new CircularBuffer<Slice>(queueCapacity);
    freeFragments = new CircularBuffer<Slice>(queueCapacity);
    backlog = new ArrayDeque<CircularBuffer<Slice>>();
  }

  @Override
  public void read(int len)
  {
    super.read(len);
    readByteCount.addAndGet(len);
  }

  @Override
  public void activate(StreamContext context)
  {
    InetSocketAddress address = context.getBufferServerAddress();
    eventloop = context.get(StreamContext.EVENT_LOOP);
    eventloop.connect(address.isUnresolved() ? new InetSocketAddress(address.getHostName(), address.getPort()) : address, this);

    logger.debug("Registering subscriber: id={} upstreamId={} streamLogicalName={} windowId={} mask={} partitions={} server={}", new Object[] {context.getSinkId(), context.getSourceId(), context.getId(), Codec.getStringWindowId(context.getFinishedWindowId()), context.getPartitionMask(), context.getPartitions(), context.getBufferServerAddress()});
    activate(null, context.getId() + '/' + context.getSinkId(), context.getSourceId(), context.getPartitionMask(), context.getPartitions(), context.getFinishedWindowId(), freeFragments.capacity());
  }

  @Override
  public void onMessage(byte[] buffer, int offset, int length)
  {
    Slice f;
    if (freeFragments.isEmpty()) {
      f = new Slice(buffer, offset, length);
    }
    else {
      f = freeFragments.pollUnsafe();
      f.buffer = buffer;
      f.offset = offset;
      f.length = length;
    }

    if (!offeredFragments.offer(f)) {
      synchronized (backlog) {
        if (!suspended) {
          suspendRead();
          suspended = true;
        }
        int newsize = offeredFragments.capacity() == 32 * 1024 ? offeredFragments.capacity() : offeredFragments.capacity() << 1;
        backlog.add(offeredFragments = new CircularBuffer<Slice>(newsize));<commit_msg>SPOI-4383 #comment support NETLET.MAX_SENDBUFFER_SIZE system property for controlling scaling of networking buffers<commit_after>/**
 * Copyright (c) 2012-2013 DataTorrent, Inc. All rights reserved.
 */
package com.datatorrent.stram.stream;

import java.net.InetSocketAddress;
import java.util.ArrayDeque;
import java.util.HashMap;
import java.util.concurrent.atomic.AtomicLong;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.datatorrent.api.Sink;
import com.datatorrent.api.StreamCodec;

import com.datatorrent.bufferserver.client.Subscriber;
import com.datatorrent.bufferserver.util.Codec;
import com.datatorrent.common.util.Slice;
import com.datatorrent.netlet.EventLoop;
import com.datatorrent.netlet.util.CircularBuffer;
import com.datatorrent.stram.codec.StatefulStreamCodec;
import com.datatorrent.stram.codec.StatefulStreamCodec.DataStatePair;
import com.datatorrent.stram.engine.ByteCounterStream;
import com.datatorrent.stram.engine.StreamContext;
import com.datatorrent.stram.engine.SweepableReservoir;
import com.datatorrent.stram.engine.WindowGenerator;
import com.datatorrent.stram.tuple.*;

/**
 * Implement tuple flow from buffer server to the node in a logical stream<p>
 * <br>
 * Extends SocketInputStream as buffer server and node communicate via a socket<br>
 * This buffer server is a read instance of a stream and takes care of connectivity with upstream buffer server<br>
 *
 * @since 0.3.2
 */
public class BufferServerSubscriber extends Subscriber implements ByteCounterStream
{
  private boolean suspended;
  private long baseSeconds;
  protected StreamCodec<Object> serde;
  protected StatefulStreamCodec<Object> statefulSerde;
  protected EventLoop eventloop;
  private final DataStatePair dsp;
  CircularBuffer<Slice> offeredFragments;
  CircularBuffer<Slice> polledFragments;
  CircularBuffer<Slice> freeFragments;
  private final ArrayDeque<CircularBuffer<Slice>> backlog;
  private int lastWindowId = WindowGenerator.MAX_WINDOW_ID;
  private final AtomicLong readByteCount;

  public BufferServerSubscriber(String id, int queueCapacity)
  {
    super(id);
    this.reservoirs = new BufferReservoir[0];
    this.reservoirMap = new HashMap<String, BufferReservoir>();
    this.readByteCount = new AtomicLong(0);
    this.dsp = new DataStatePair();
    polledFragments = offeredFragments = new CircularBuffer<Slice>(queueCapacity);
    freeFragments = new CircularBuffer<Slice>(queueCapacity);
    backlog = new ArrayDeque<CircularBuffer<Slice>>();
  }

  @Override
  public void read(int len)
  {
    super.read(len);
    readByteCount.addAndGet(len);
  }

  @Override
  public void activate(StreamContext context)
  {
    InetSocketAddress address = context.getBufferServerAddress();
    eventloop = context.get(StreamContext.EVENT_LOOP);
    eventloop.connect(address.isUnresolved() ? new InetSocketAddress(address.getHostName(), address.getPort()) : address, this);

    logger.debug("Registering subscriber: id={} upstreamId={} streamLogicalName={} windowId={} mask={} partitions={} server={}", new Object[] {context.getSinkId(), context.getSourceId(), context.getId(), Codec.getStringWindowId(context.getFinishedWindowId()), context.getPartitionMask(), context.getPartitions(), context.getBufferServerAddress()});
    activate(null, context.getId() + '/' + context.getSinkId(), context.getSourceId(), context.getPartitionMask(), context.getPartitions(), context.getFinishedWindowId(), freeFragments.capacity());
  }

  @Override
  public void onMessage(byte[] buffer, int offset, int length)
  {
    Slice f;
    if (freeFragments.isEmpty()) {
      f = new Slice(buffer, offset, length);
    }
    else {
      f = freeFragments.pollUnsafe();
      f.buffer = buffer;
      f.offset = offset;
      f.length = length;
    }

    if (!offeredFragments.offer(f)) {
      synchronized (backlog) {
        if (!suspended) {
          suspendRead();
          suspended = true;
        }
        int newsize = offeredFragments.capacity() == MAX_SENDBUFFER_SIZE ? offeredFragments.capacity() : offeredFragments.capacity() << 1;
        backlog.add(offeredFragments = new CircularBuffer<Slice>(newsize));
{'fraction_non_alphanumeric': 0.10802243355279201, 'fraction_numerical': 0.005120702267739576, 'mean_word_length': 4.935600578871201, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 37, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3439232', 'n_tokens_mistral': 2499, 'n_tokens_neox': 2468, 'n_words': 541}