<commit_before>/*
  <NAME>, <EMAIL>, 2015-05-19
*/

var net = require('net');
require('buffer').INSPECT_MAX_BYTES = 500;

var VXI11_READ_TIMEOUT = 2000;  /* ms */
var VXI11_IO_TIMEOUT = 10000;   /* ms */
var VXI11_LOCK_TIMEOUT = 10000; /* ms */

var REQUEST_SIZE = 100000;
var END_FLAG = 8;

var CALL = 0;
var REPLY = 1;

var RPC_VERSION = 2;

// Device async
var DEVICE_ASYNC_PROG = 0x0607b0;
var DEVICE_ASYNC_VERS = 1;
var DEVICE_ABORT = 1;

// Device intr
var DEVICE_INTR = 0x0607b1;

// Device core
var DEVICE_CORE_PROG = 0x0607af;
var DEVICE_CORE_VERS = 1;
var CREATE_LINK = 10;
var DEVICE_WRITE = 11;
var DEVICE_READ = 12;
var DEVICE_READSTB = 13;
var DEVICE_TRIGGER = 14;
var DEVICE_CLEAR = 15;
var DEVICE_REMOTE = 16;
var DEVICE_LOCAL = 17;
var DEVICE_LOCK = 18;
var DEVICE_UNLOCK = 19;
var DEVICE_ENABLE_SRQ = 20;
var DEVICE_DOCMD = 22;
var DESTROY_LINK = 23;
var CREATE_INTR_CHAN = 25;
var DESTROY_INTR_CHAN = 26;
var DEVICE_INTR_SRQ = 30;

// Program number, version and port number
var PMAP_PROG = 100000;
var PMAP_VERS = 2;
var PMAP_PORT = 111;

// Procedure numbers
var PMAPPROC_NULL = 0;
var PMAPPROC_SET = 1;
var PMAPPROC_UNSET = 2;
var PMAPPROC_GETPORT = 3;
var PMAPPROC_DUMP = 4;
var PMAPPROC_CALLIT = 5;

var IPPROTO_TCP = 6;
var IPPROTO_UDP = 17;

var LAST_RECORD = 0x80000000;
var XID_OFFS = 4;
var PROC_OFFS = 24;
var LINK_ID_OFFS = 32;
var PORT_OFFS = 28;

var ERROR_OFFS = 28;
var LEN_OFFS = 36;
var DATA_OFFS = LEN_OFFS + 4;

var pnames = {};
pnames[PMAPPROC_GETPORT] = 'GETPORT';
pnames[CREATE_LINK] = 'CREATE_LINK';
pnames[DEVICE_WRITE] = 'DEVICE_WRITE';
pnames[DEVICE_READ] = 'DEVICE_READ';
pnames[DESTROY_LINK] = 'DESTROY_LINK';

var errors = {};
errors[0] = 'no error';
errors[1] = 'syntax error';
errors[3] = 'device not accessible';
errors[4] = 'invalid link identifier';
errors[5] = 'parameter error';
errors[6] = 'channel not established';
errors[8] = 'operation not supported';
errors[9] = 'out of resources';
errors[11] = 'device locked by another link';
errors[12] = 'no lock held by this link';
errors[15] = 'I/O timeout';
errors[17] = 'I/O error';
errors[21] = 'invalid address';
errors[23] = 'abort';
errors[29] = 'channel already established';

var IO_TIMEOUT = 15;

function inspectBuffer(clink, str, buf) {
  clink.logger.log(str + ' [%d]: %s', buf.length, buf.inspect());
}

function getAlignedBuffer(str) {
  var l =  str.length;
  var len = l % 4 ? l + 4 - l % 4 : l;// multiple of 4 Bytes
  var buf = new Buffer(len);
  buf.fill(0);
  new Buffer(str, 'ascii').copy(buf);
  return buf;
}

function getNewXID() {// 32bit unsigned
  return (Date.now() & 0x0ffffffff) >>> 0;
}

function getPort(buf) {
  return buf.length >= PORT_OFFS + 4 ? buf.readUInt32BE(PORT_OFFS) : -1;
}

function getErrorMsg(proc, err) {
  var error = errors[err] ? errors[err] : err;
  return '[' + pnames[proc] + '] ' + error;
}

function rpc(clink, clbk) {
  var a = [], b, buf, d;
  var socket = clink.socket;
  for(var i=2; i<arguments.length; i++) {// variable arguments (buffer content)
    d = arguments[i];
    if (typeof d == 'number') {
      b = new Buffer(4);
      b.writeUInt32BE(d, 0, true);
      a.push(b);
    } else if (typeof d == 'string') a.push(getAlignedBuffer(d));
  }
  buf = Buffer.concat(a);
  if (!buf.length) {
    var e = 'no data';
    clink.logger.error(e);
    if (typeof clbk == 'function') clbk(clink, null, e); return;
  }
  buf.writeUInt32BE(LAST_RECORD + buf.length - 4, 0, true);
  var procedure = buf.readUInt32BE(PROC_OFFS);
  inspectBuffer(clink, pnames[procedure] + ' call', buf);

  var data = new Buffer(0), first = true;

  function getLastFragment(buf) {
    var flen = 0, len = 0, ret = false, idx = buf.length - 4, mask = 1 << 31;
    while (idx > -1) {
      flen = buf.readUInt32BE(idx);
      if ((flen & mask) != 0) {// is the highest bit set?
        flen &= ~mask;// clear the highest bit
        if (flen == len) {
          ret = { idx:idx };
          break;
        }
      }
      idx -= 4;
      len += 4;
    }
    return ret;
  }

  function onData(chunk) {
    var lastFragment = false, buf;
    if (first) {
      first = false;
      data = Buffer.concat([data,chunk]);
      lastFragment = data.readUInt8(0) >= 128;
      clink.logger.log('FIRST CALL' + (lastFragment ? ' (last fragment)' : ''));
    } else {
      lastFragment = getLastFragment(chunk);
      if (lastFragment) {
        buf = chunk.slice(0, lastFragment.idx);
        data = Buffer.concat([data,buf]);
        buf = chunk.slice(lastFragment.idx+4);
        data = Buffer.concat([data,buf]);
      } else {
        data = Buffer.concat([data,chunk]);
      }
      inspectBuffer(clink, 'chunk', chunk.slice(48));
      clink.logger.log('NEXT CALL' + (lastFragment ? ' (last fragment)' : ''));
    }

    if (lastFragment) {
      var xid = data.readUInt32BE(XID_OFFS);
      var error = data.readUInt32BE(ERROR_OFFS);
      clink.logger.log('XID[call]: %d      XID[reply]: %d', clink.xid, xid);
      inspectBuffer(clink, pnames[procedure] + ' reply', data);
      socket.removeListener('data', onData);
      if (typeof clbk == 'function') clbk(clink, data, error); return;
    }
  }

  socket.on('data', onData);

  socket.write(buf);
}

function _rpc(clink, procedure, clbk) {
  var args = [clink, clbk];
  clink.xid = getNewXID();
  switch (procedure) {<commit_msg>multi-fragment data; cleaner structure<commit_after>/*
  <NAME>, <EMAIL>, 2015-05-21
*/

var net = require('net');
require('buffer').INSPECT_MAX_BYTES = 500;

var VXI11_READ_TIMEOUT = 2000;  /* ms */
var VXI11_IO_TIMEOUT = 10000;   /* ms */
var VXI11_LOCK_TIMEOUT = 10000; /* ms */

var REQUEST_SIZE = 1000000;
var END_FLAG = 8;

var CALL = 0;
var REPLY = 1;

var RPC_VERSION = 2;

// Device async
var DEVICE_ASYNC_PROG = 0x0607b0;
var DEVICE_ASYNC_VERS = 1;
var DEVICE_ABORT = 1;

// Device intr
var DEVICE_INTR = 0x0607b1;

// Device core
var DEVICE_CORE_PROG = 0x0607af;
var DEVICE_CORE_VERS = 1;
var CREATE_LINK = 10;
var DEVICE_WRITE = 11;
var DEVICE_READ = 12;
var DEVICE_READSTB = 13;
var DEVICE_TRIGGER = 14;
var DEVICE_CLEAR = 15;
var DEVICE_REMOTE = 16;
var DEVICE_LOCAL = 17;
var DEVICE_LOCK = 18;
var DEVICE_UNLOCK = 19;
var DEVICE_ENABLE_SRQ = 20;
var DEVICE_DOCMD = 22;
var DESTROY_LINK = 23;
var CREATE_INTR_CHAN = 25;
var DESTROY_INTR_CHAN = 26;
var DEVICE_INTR_SRQ = 30;

// Program number, version and port number
var PMAP_PROG = 100000;
var PMAP_VERS = 2;
var PMAP_PORT = 111;

// Procedure numbers
var PMAPPROC_NULL = 0;
var PMAPPROC_SET = 1;
var PMAPPROC_UNSET = 2;
var PMAPPROC_GETPORT = 3;
var PMAPPROC_DUMP = 4;
var PMAPPROC_CALLIT = 5;

var IPPROTO_TCP = 6;
var IPPROTO_UDP = 17;

var LAST_RECORD = 0x80000000;
var XID_OFFS = 4;
var PROC_OFFS = 24;
var LINK_ID_OFFS = 32;
var PORT_OFFS = 28;

var ERROR_OFFS = 28;
var LEN_OFFS = 36;
var DATA_OFFS = LEN_OFFS + 4;

var pnames = {};
pnames[PMAPPROC_GETPORT] = 'GETPORT';
pnames[CREATE_LINK] = 'CREATE_LINK';
pnames[DEVICE_WRITE] = 'DEVICE_WRITE';
pnames[DEVICE_READ] = 'DEVICE_READ';
pnames[DESTROY_LINK] = 'DESTROY_LINK';

var errors = {};
errors[0] = 'no error';
errors[1] = 'syntax error';
errors[3] = 'device not accessible';
errors[4] = 'invalid link identifier';
errors[5] = 'parameter error';
errors[6] = 'channel not established';
errors[8] = 'operation not supported';
errors[9] = 'out of resources';
errors[11] = 'device locked by another link';
errors[12] = 'no lock held by this link';
errors[15] = 'I/O timeout';
errors[17] = 'I/O error';
errors[21] = 'invalid address';
errors[23] = 'abort';
errors[29] = 'channel already established';

var IO_TIMEOUT = 15;

function inspectBuffer(clink, str, buf) {
  clink.logger.log(str + ' [%d]: %s', buf.length, buf.inspect());
}

function getAlignedBuffer(str) {
  var l =  str.length;
  var len = l % 4 ? l + 4 - l % 4 : l;// multiple of 4 Bytes
  var buf = new Buffer(len);
  buf.fill(0);
  new Buffer(str, 'ascii').copy(buf);
  return buf;
}

function getNewXID() {// 32bit unsigned
  return (Date.now() & 0x0ffffffff) >>> 0;
}

function getPort(buf) {
  return buf.length >= PORT_OFFS + 4 ? buf.readUInt32BE(PORT_OFFS) : -1;
}

function getErrorMsg(proc, err) {
  var error = errors[err] ? errors[err] : err;
  return '[' + pnames[proc] + '] ' + error;
}

function rpc(clink, clbk) {
  var a = [], b, buf, d;
  var socket = clink.socket;
  for(var i=2; i<arguments.length; i++) {// variable arguments (buffer content)
    d = arguments[i];
    if (typeof d == 'number') {
      b = new Buffer(4);
      b.writeUInt32BE(d, 0, true);
      a.push(b);
    } else if (typeof d == 'string') a.push(getAlignedBuffer(d));
  }
  buf = Buffer.concat(a);
  if (!buf.length) {
    var e = 'no data';
    clink.logger.error(e);
    if (typeof clbk == 'function') clbk(clink, null, e); return;
  }
  buf.writeUInt32BE(LAST_RECORD + buf.length - 4, 0, true);
  var procedure = buf.readUInt32BE(PROC_OFFS);
  inspectBuffer(clink, pnames[procedure] + ' call', buf);

  var data = new Buffer(0), chunkCnt = 0;

  function reassemble(buf) {
    // remove possible rpc headers
    var a = [], flen = 0, len = 0, idx = buf.length - 4, mask = 1 << 31,
      prevIdx = buf.length - 1;
    clink.logger.log('[reassemble] buf.length: ' + buf.length);
    while (idx > -1) {
      flen = buf.readUInt32BE(idx) & ~mask;// clear the highest bit
      if (flen == len) {// most likely a size marker of a rpc fragment
        clink.logger.log('[reassemble] flen: %d  idx: %d', flen, idx);
        // prepend the fragment data to the array
        if (idx > 0) {
          // only the real data
          a.unshift(buf.slice(idx+4, idx+4+flen));
        } else {
          // the vxi header and the data
          a.unshift(buf.slice(0, prevIdx));
        }
        len = -4;
        prevIdx = idx;
      }
      idx -= 4;
      len += 4;
    }
    for (var i=0; i<a.length; i++) {
       clink.logger.log('a[%d].length=%d', i, a[i].length);
    }
    return Buffer.concat(a);
  }

  function onData(chunk) {
    var lastFragment = false, expected = 0;

    data = Buffer.concat([data,chunk]);
    chunkCnt++;

    if (procedure == DEVICE_READ) {
      expected = data.readUInt32BE(LEN_OFFS);
      lastFragment = data.length >= expected;
      clink.logger.log('expected: %d  actual: %d', expected, data.length);
    } else {
      lastFragment = data.readUInt8(0) >= 128;
    }

    inspectBuffer(clink, 'chunk', chunk);

    if (lastFragment) {
      clink.logger.log('*** last fragment ***');
      if (procedure == DEVICE_READ && chunkCnt > 1) data = reassemble(data);
      var xid = data.readUInt32BE(XID_OFFS);
      var error = data.readUInt32BE(ERROR_OFFS);
      clink.logger.log('XID[call]: %d      XID[reply]: %d', clink.xid, xid);
      inspectBuffer(clink, pnames[procedure] + ' reply', data);
      socket.removeListener('data', onData);
      if (typeof clbk == 'function') clbk(clink, data, error); return;
    }
  }

  socket.on('data', onData);

  socket.write(buf);
}

function _rpc(clink, procedure, clbk) {
  var args = [clink, clbk];
  clink.xid = getNewXID();
  switch (procedure) {
{'fraction_non_alphanumeric': 0.15139153295258817, 'fraction_numerical': 0.03843713171969903, 'mean_word_length': 3.3433070866141734, 'pattern_counts': {'":': 0, '<': 14, '<?xml version=': 0, '>': 22, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 4, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3456587', 'n_tokens_mistral': 4547, 'n_tokens_neox': 4154, 'n_words': 1088}