<commit_before>/*
 * Copyright 2010-2012 <NAME> (l.garulli--at--orientechnologies.com)
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.orientechnologies.orient.enterprise.channel.binary;

import java.io.IOException;
import java.net.Socket;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

import com.orientechnologies.common.concur.OTimeoutException;
import com.orientechnologies.common.log.OLogManager;
import com.orientechnologies.orient.core.config.OContextConfiguration;
import com.orientechnologies.orient.core.config.OGlobalConfiguration;

/**
 * Implementation that supports multiple client requests.
 * 
 * @author <NAME> (l.garulli--at--orientechnologies.com)
 * 
 */
public class OChannelBinaryAsynch extends OChannelBinary {
  private final ReentrantLock lockRead    = new ReentrantLock(true);
  private final ReentrantLock lockWrite   = new ReentrantLock();
  private boolean             channelRead = false;
  private byte                currentStatus;
  private int                 currentSessionId;
  private final int           maxUnreadResponses;

  public OChannelBinaryAsynch(final Socket iSocket, final OContextConfiguration iConfig) throws IOException {
    super(iSocket, iConfig);
    maxUnreadResponses = OGlobalConfiguration.NETWORK_BINARY_READ_RESPONSE_MAX_TIMES.getValueAsInteger();
  }

  public void beginRequest() {
    lockWrite.lock();
  }

  public void endRequest() throws IOException {
    flush();
    lockWrite.unlock();
  }

  public void beginResponse(final int iRequesterId) throws IOException {
    beginResponse(iRequesterId, timeout);
  }

  public void beginResponse(final int iRequesterId, final long iTimeout) throws IOException {
    try {
      int unreadResponse = 0;
      final long startClock = iTimeout > 0 ? System.currentTimeMillis() : 0;

      // WAIT FOR THE RESPONSE
      do {
        if (iTimeout <= 0)
          lockRead.lock();
        else if (!lockRead.tryLock(iTimeout, TimeUnit.MILLISECONDS))
          throw new OTimeoutException("Cannot acquire read lock against channel: " + this);

        if (!channelRead) {
          channelRead = true;

          try {
            currentStatus = readByte();
            currentSessionId = readInt();

            if (debug)
              OLogManager.instance().debug(this, "%s - Read response: %d-%d", socket.getLocalAddress(), (int) currentStatus,
                  currentSessionId);

          } catch (IOException e) {
            // UNLOCK THE RESOURCE AND PROPAGATES THE EXCEPTION
            lockRead.unlock();
            channelRead = false;
            throw e;
          }
        }

        if (currentSessionId == iRequesterId)
          // IT'S FOR ME
          break;

        if (debug)
          OLogManager.instance().debug(this, "%s - Session %d skip response, it is for %d", socket.getLocalAddress(), iRequesterId,
              currentSessionId);

        if (iTimeout > 0 && (System.currentTimeMillis() - startClock) > iTimeout)
          throw new OTimeoutException("Timeout on reading response from the server for the request " + iRequesterId);

        if (unreadResponse > maxUnreadResponses) {
          if (debug)
            OLogManager.instance().info(this, "Unread responses %d > %d, consider the buffer as dirty: clean it", unreadResponse,
                maxUnreadResponses);

          // CALL THE SUPER-METHOD TO AVOID LOCKING AGAIN
          // super.clearInput();
          close();
          throw new IOException("Timeout on reading response");
        }

        final long start = System.currentTimeMillis();

        // WAIT 1 SECOND AND RETRY
        synchronized (this) {
          lockRead.unlock();
          notifyAll();

          try {
            if (debug)
              OLogManager.instance().debug(this, "Session %d is going to sleep...", iRequesterId);

            wait(1000);
            final long now = System.currentTimeMillis();<commit_msg>Fixed nasty race condition again st socket read that caused 100% of CPU with hundreds of threads and few network connections<commit_after>/*
 * Copyright 2010-2012 <NAME> (l.garulli--at--orientechnologies.com)
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.orientechnologies.orient.enterprise.channel.binary;

import java.io.IOException;
import java.net.Socket;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

import com.orientechnologies.common.concur.OTimeoutException;
import com.orientechnologies.common.log.OLogManager;
import com.orientechnologies.orient.core.config.OContextConfiguration;
import com.orientechnologies.orient.core.config.OGlobalConfiguration;

/**
 * Implementation that supports multiple client requests.
 * 
 * @author <NAME> (l.garulli--at--orientechnologies.com)
 * 
 */
public class OChannelBinaryAsynch extends OChannelBinary {
  private final ReentrantLock lockRead    = new ReentrantLock(true);
  private final ReentrantLock lockWrite   = new ReentrantLock();
  private boolean             channelRead = false;
  private byte                currentStatus;
  private int                 currentSessionId;
  private final int           maxUnreadResponses;

  public OChannelBinaryAsynch(final Socket iSocket, final OContextConfiguration iConfig) throws IOException {
    super(iSocket, iConfig);
    maxUnreadResponses = OGlobalConfiguration.NETWORK_BINARY_READ_RESPONSE_MAX_TIMES.getValueAsInteger();
  }

  public void beginRequest() {
    lockWrite.lock();
  }

  public void endRequest() throws IOException {
    flush();
    lockWrite.unlock();
  }

  public void beginResponse(final int iRequesterId) throws IOException {
    beginResponse(iRequesterId, timeout);
  }

  public void beginResponse(final int iRequesterId, final long iTimeout) throws IOException {
    try {
      int unreadResponse = 0;
      final long startClock = iTimeout > 0 ? System.currentTimeMillis() : 0;

      // WAIT FOR THE RESPONSE
      do {
        if (iTimeout <= 0)
          lockRead.lock();
        else if (!lockRead.tryLock(iTimeout, TimeUnit.MILLISECONDS))
          throw new OTimeoutException("Cannot acquire read lock against channel: " + this);

        if (!channelRead) {
          channelRead = true;

          try {
            currentStatus = readByte();
            currentSessionId = readInt();

            if (debug)
              OLogManager.instance().debug(this, "%s - Read response: %d-%d", socket.getLocalAddress(), (int) currentStatus,
                  currentSessionId);

          } catch (IOException e) {
            // UNLOCK THE RESOURCE AND PROPAGATES THE EXCEPTION
            lockRead.unlock();
            channelRead = false;
            throw e;
          }
        }

        if (currentSessionId == iRequesterId)
          // IT'S FOR ME
          break;

        if (debug)
          OLogManager.instance().debug(this, "%s - Session %d skip response, it is for %d", socket.getLocalAddress(), iRequesterId,
              currentSessionId);

        if (iTimeout > 0 && (System.currentTimeMillis() - startClock) > iTimeout)
          throw new OTimeoutException("Timeout on reading response from the server for the request " + iRequesterId);

        if (unreadResponse > maxUnreadResponses) {
          if (debug)
            OLogManager.instance().info(this, "Unread responses %d > %d, consider the buffer as dirty: clean it", unreadResponse,
                maxUnreadResponses);

          // CALL THE SUPER-METHOD TO AVOID LOCKING AGAIN
          // super.clearInput();
          close();
          throw new IOException("Timeout on reading response");
        }

        // WAIT 1 SECOND AND RETRY
        synchronized (this) {
          lockRead.unlock();
          notifyAll();

          try {
            if (debug)
              OLogManager.instance().debug(this, "Session %d is going to sleep...", iRequesterId);

            final long start = System.currentTimeMillis();

            wait(1000);
            final long now = System.currentTimeMillis();
{'fraction_non_alphanumeric': 0.0894649751792609, 'fraction_numerical': 0.005184776613348042, 'mean_word_length': 2.704944830404577, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '6749366', 'n_tokens_mistral': 2477, 'n_tokens_neox': 2401, 'n_words': 731}