<commit_before>var net = require('net'),
  debug = require('util').debug,
  inspect = require('util').inspect,
  EventEmitter = require("events").EventEmitter,
  BinaryParser = require('./bson/binary_parser').BinaryParser,
  inherits = require('util').inherits,
  Server = require('./connections/server').Server;

var Connection = exports.Connection = function(host, port, autoReconnect, options) {
  this.options = options == null ? {} : options;
  this.host = host;
  this.port = port;
  this.autoReconnect = autoReconnect;
  this.drained = true;
  // Fetch the poolsize
  this.poolSize = this.options["poolSize"] == null ? 1 : this.options["poolSize"];    
  // Reconnect buffer for messages
  this.messages = [];

  // Status messages
  this.sizeOfMessage = 0;
  this.bytesRead = 0;
  this.buffer = '';
  this.stubBuffer = '';
  this.connected = false;
  
  // Connection pool variables
  this.pool = [];
  this.poolByReference = {};
  this.poolIndex = 0;
};

inherits(Connection, EventEmitter);

var getConnection = function(self) {
  return self.pool[self.poolIndex++ % self.pool.length];
}

// Setup the connection pool
var setupConnectionPool = function(self, poolSize, reconnect) {
  // Pool off connections and status variables
  var connectionPool = [];  
  var connectedTo = 0;
  var errors = 0;
  var connectionError = null;

  //
  // Listener that handles callbacks for the connection
  // Uses the internal object states to keep individual tcp connections seperate
  var receiveListener = function(result, fd) {    
    fd = fd == null ? this.fd : fd;
    
    // Fetch the pool reference
    var conObj = self.poolByReference[fd];
    
    // if(conObj == null) {
    //   debug("================================================================ failed to find connection :: " + this.fd)
    //   debug(inspect(self.poolByReference))
    // }
    
    // Check if we have an unfinished message
    if(conObj != null && conObj.bytesRead > 0 && conObj.sizeOfMessage > 0) {
      // Calculate remaing bytes to fetch
      var remainingBytes = conObj.sizeOfMessage - conObj.bytesRead;
      // Check if we have multiple packet messages and save the pieces otherwise emit the message
      if(remainingBytes > result.length) {
        conObj.buffer = conObj.buffer + result; conObj.bytesRead = conObj.bytesRead + result.length;
      } else {
        // Cut off the remaining message
        conObj.buffer = conObj.buffer + result.substr(0, remainingBytes);
        // Emit the message
        self.emit("data", conObj.buffer);
        // Reset the variables
        conObj.buffer = ''; conObj.bytesRead = 0; conObj.sizeOfMessage = 0;
        // If message is longer than the current one, keep parsing
        if(remainingBytes < result.length) {
          // debug("--------------------------------------- remainingBytes < result.length :: " + this.fd)
          // receiveListener.call(this, result.substr(remainingBytes, (result.length - remainingBytes)));
          receiveListener(result.substr(remainingBytes, (result.length - remainingBytes)), fd);
        }
      }
    } else if(conObj != null){
      if(conObj.stubBuffer.length > 0) {
        result = conObj.stubBuffer + result;
        conObj.stubBuffer = '';
      }

      if(result.length > 4) {
        var sizeOfMessage = BinaryParser.toInt(result.substr(0, 4));
        // We got a partial message, store the result and wait for more
        if(sizeOfMessage > result.length) {
          conObj.buffer = conObj.buffer + result; conObj.bytesRead = result.length; conObj.sizeOfMessage = sizeOfMessage;
        } else if(sizeOfMessage == result.length) {
          self.emit("data", result);
        } else if(sizeOfMessage < result.length) {
          self.emit("data", result.substr(0, sizeOfMessage));
          // debug("--------------------------------------- sizeOfMessage < result.length :: " + this.fd)
          // receiveListener.call(this, result.substr(sizeOfMessage, (result.length - sizeOfMessage)));
          receiveListener(result.substr(sizeOfMessage, (result.length - sizeOfMessage)), fd);
        }
      } else {
        conObj.stubBuffer = result;
      }
    }
  };
  
  // Fill the pool
  for(var i = 0; i < poolSize; i++) {
    // Create the associated connection
    var connection = net.createConnection(self.port, self.host);    
    // Set up the net client
    connection.setEncoding("binary");
    // Add connnect listener
    connection.addListener("connect", function() {
      this.setEncoding("binary");
      this.setTimeout(0);
      this.setNoDelay();
      // Update number of connected to server
      connectedTo = connectedTo + 1;
    });
    
    connection.addListener("error", function(err) {
      // Update number of errors
      errors = errors + 1;
      connectionError = err;
    });
    
    connection.addListener("timeout", function(err) {
      // Update number of errors
      errors = errors + 1;
      connectionError = err;
    });
    
    // Add a close listener
    connection.addListener("close", function() {
      self.emit("close");
    });
    
    // Add connection to the pool array
    connectionPool.push({"connection": connection,
      "sizeOfMessage": 0,
      "bytesRead": 0,
      "buffer": '',
      "stubBuffer": ''});      
    // Add the listener to the connection
    connection.addListener("data", receiveListener);
  }
  
  // Function that wait for connection to finish up
  var waitForConnections = function() {
    // Emit a connect message once all connections are up
    if(connectedTo == connectionPool.length) {
      if(reconnect == null || !reconnect) {
        self.connected = true;
        self.emit("connect");
      } else {
        self.connected = false;
        self.emit("reconnect");
      }
    } else if(errors + connectedTo == connectionPool.length) {
      if(reconnect == null || !reconnect) {
        self.connected = false;
        self.emit("error", connectionError);
      } else {
        self.connected = false;
        self.emit("reconnect");
      }              
    } else {
      process.nextTick(waitForConnections);
    }
  }
  
  // Wait until we are done connected to all pool entries before emitting connect signal
  process.nextTick(waitForConnections);
  
  // Return the pool
  return connectionPool;
}

// Functions to open the connection
Connection.prototype.open = function() {
  var self = this;
  // Create the pool with connections
  this.pool = setupConnectionPool(this, this.poolSize);
  this.poolByReference = {};
  
  // Save the connections by the fd reference
  this.pool.forEach(function(con) {
    self.poolByReference[con.connection.fd] = con;
  })
}

Connection.prototype.close = function() {
  this.connected = false;
  // Close all entries in the pool
  for(var i = 0; i < this.pool.length; i++) {<commit_msg>fix race condition in setting up connection pool<commit_after>var net = require('net'),
  debug = require('util').debug,
  inspect = require('util').inspect,
  EventEmitter = require("events").EventEmitter,
  BinaryParser = require('./bson/binary_parser').BinaryParser,
  inherits = require('util').inherits,
  Server = require('./connections/server').Server;

var Connection = exports.Connection = function(host, port, autoReconnect, options) {
  this.options = options == null ? {} : options;
  this.host = host;
  this.port = port;
  this.autoReconnect = autoReconnect;
  this.drained = true;
  // Fetch the poolsize
  this.poolSize = this.options["poolSize"] == null ? 1 : this.options["poolSize"];    
  // Reconnect buffer for messages
  this.messages = [];

  // Status messages
  this.sizeOfMessage = 0;
  this.bytesRead = 0;
  this.buffer = '';
  this.stubBuffer = '';
  this.connected = false;
  
  // Connection pool variables
  this.pool = [];
  this.poolByReference = {};
  this.poolIndex = 0;
};

inherits(Connection, EventEmitter);

var getConnection = function(self) {
  return self.pool[self.poolIndex++ % self.pool.length];
}

// Setup the connection pool
var setupConnectionPool = function(self, poolSize, reconnect) {
  // Pool off connections and status variables
  var connectionPool = [];  
  var connectedTo = 0;
  var errors = 0;
  var connectionError = null;

  //
  // Listener that handles callbacks for the connection
  // Uses the internal object states to keep individual tcp connections seperate
  var receiveListener = function(result, fd) {    
    fd = fd == null ? this.fd : fd;
    
    // Fetch the pool reference
    var conObj = self.poolByReference[fd];
    
    // if(conObj == null) {
    //   debug("================================================================ failed to find connection :: " + this.fd)
    //   debug(inspect(self.poolByReference))
    // }
    
    // Check if we have an unfinished message
    if(conObj != null && conObj.bytesRead > 0 && conObj.sizeOfMessage > 0) {
      // Calculate remaing bytes to fetch
      var remainingBytes = conObj.sizeOfMessage - conObj.bytesRead;
      // Check if we have multiple packet messages and save the pieces otherwise emit the message
      if(remainingBytes > result.length) {
        conObj.buffer = conObj.buffer + result; conObj.bytesRead = conObj.bytesRead + result.length;
      } else {
        // Cut off the remaining message
        conObj.buffer = conObj.buffer + result.substr(0, remainingBytes);
        // Emit the message
        self.emit("data", conObj.buffer);
        // Reset the variables
        conObj.buffer = ''; conObj.bytesRead = 0; conObj.sizeOfMessage = 0;
        // If message is longer than the current one, keep parsing
        if(remainingBytes < result.length) {
          // debug("--------------------------------------- remainingBytes < result.length :: " + this.fd)
          // receiveListener.call(this, result.substr(remainingBytes, (result.length - remainingBytes)));
          receiveListener(result.substr(remainingBytes, (result.length - remainingBytes)), fd);
        }
      }
    } else if(conObj != null){
      if(conObj.stubBuffer.length > 0) {
        result = conObj.stubBuffer + result;
        conObj.stubBuffer = '';
      }

      if(result.length > 4) {
        var sizeOfMessage = BinaryParser.toInt(result.substr(0, 4));
        // We got a partial message, store the result and wait for more
        if(sizeOfMessage > result.length) {
          conObj.buffer = conObj.buffer + result; conObj.bytesRead = result.length; conObj.sizeOfMessage = sizeOfMessage;
        } else if(sizeOfMessage == result.length) {
          self.emit("data", result);
        } else if(sizeOfMessage < result.length) {
          self.emit("data", result.substr(0, sizeOfMessage));
          // debug("--------------------------------------- sizeOfMessage < result.length :: " + this.fd)
          // receiveListener.call(this, result.substr(sizeOfMessage, (result.length - sizeOfMessage)));
          receiveListener(result.substr(sizeOfMessage, (result.length - sizeOfMessage)), fd);
        }
      } else {
        conObj.stubBuffer = result;
      }
    }
  };
  
  // Fill the pool
  for(var i = 0; i < poolSize; i++) {
    // Create the associated connection
    var connection = net.createConnection(self.port, self.host);    
    // Set up the net client
    connection.setEncoding("binary");
    // Add connnect listener
    connection.addListener("connect", function() {
      this.setEncoding("binary");
      this.setTimeout(0);
      this.setNoDelay();
      // Update number of connected to server
      connectedTo = connectedTo + 1;
    });
    
    connection.addListener("error", function(err) {
      // Update number of errors
      errors = errors + 1;
      connectionError = err;
    });
    
    connection.addListener("timeout", function(err) {
      // Update number of errors
      errors = errors + 1;
      connectionError = err;
    });
    
    // Add a close listener
    connection.addListener("close", function() {
      self.emit("close");
    });
    
    // Add connection to the pool array
    connectionPool.push({"connection": connection,
      "sizeOfMessage": 0,
      "bytesRead": 0,
      "buffer": '',
      "stubBuffer": ''});      
    // Add the listener to the connection
    connection.addListener("data", receiveListener);
  }
  
  // Function that wait for connection to finish up
  var waitForConnections = function() {
    // Emit a connect message once all connections are up
    if(connectedTo == connectionPool.length) {
      if(reconnect == null || !reconnect) {
        self.connected = true;
        self.emit("connect");
      } else {
        self.connected = false;
        self.emit("reconnect");
      }
    } else if(errors + connectedTo == connectionPool.length) {
      if(reconnect == null || !reconnect) {
        self.connected = false;
        self.emit("error", connectionError);
      } else {
        self.connected = false;
        self.emit("reconnect");
      }              
    } else {
      process.nextTick(waitForConnections);
    }
  }
  
  // Wait until we are done connected to all pool entries before emitting connect signal
  process.nextTick(waitForConnections);

  self.addListener("connect", function() {
    self.poolByReference = {};
 
     // Save the connections by the fd reference
    this.pool.forEach(function(con) {
      self.poolByReference[con.connection.fd] = con;
    });
  });  
  
  // Return the pool
  return connectionPool;
}

// Functions to open the connection
Connection.prototype.open = function() {
  var self = this;
  // Create the pool with connections
  this.pool = setupConnectionPool(this, this.poolSize);
}

Connection.prototype.close = function() {
  this.connected = false;
  // Close all entries in the pool
  for(var i = 0; i < this.pool.length; i++) {
{'fraction_non_alphanumeric': 0.14115423753452536, 'fraction_numerical': 0.0034888791975577847, 'mean_word_length': 2.5812077043206663, 'pattern_counts': {'":': 10, '<': 15, '<?xml version=': 0, '>': 15, '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': '4521243', 'n_tokens_mistral': 3859, 'n_tokens_neox': 3809, 'n_words': 1009}