//************
//  Description: Commutates a number of input files into a single output file
//
//  Author:	JGS
//
//  $Id: fanin.cc,v 1.1 2009/08/13 00:22:19 ntn Exp $
//  Modification History:
//  MIB 18-Jan-2000 DR RRCcl00652 
//           18 Jan 2000 fixed buffer allocation for file transfers
//            that could exceed 32678 for type 1000 and 2000 files.
//  KCS 10-Feb-2004 DR 501764-1 Added /NOHANG switch (and IMPLICIT NONE)
//  CJBland 25-Aug-2004 DR 505687-1 Added 64-bit port data
//  CBerster 29-Mar-2005 C++ conversion
//
//
//************

#include <vector>

#include <primitive.h>
#include <fintrinsics.h>

void mainroutine () 
{
  CPHEADER HOUT;
  int_4 syncmode, nfiles, lframe, maxbpe, n, ns, ids;
  int_4 lper, ndo, nget, ngot = 0, i, navail;

  // get the number of input files
  nfiles = m_get_switch("MF");
  if (nfiles <= 0) nfiles = Mc->number_parameters - 1;
  vector<HEADER> HIN(nfiles);

  // open and test input file #1
  m_init(HIN[0], m_apick(2), "", "", 0);
  m_inok(HIN[0]);

  // get the frame size
  lframe = m_get_switch("FS");
  if (lframe < 0) lframe = 1;
                      
  // Get the NOHANG switch
  bool nohang = m_get_switch("NOHANG") > 0;

  // setup and allocate output file
  m_propagate(HIN[0], HOUT);
  HOUT.file_name = m_apick(1);
  HOUT.size = HIN[0].size * nfiles;
  m_allocate(HOUT);
  m_force1000(HOUT);

  // open all the rest of the input files
  maxbpe = HIN[0].bpe;
  m_force1000(HIN[0]);
  for (n = 1; n < nfiles; ++n) {
    m_init(HIN[n], m_apick(n + 2), "", "", 0);
    m_inok(HIN[n]);
    if (HIN[n].format[0] != HOUT.format[0])
      m_error("Input files must have same format mode");
    m_force1000(HIN[n]);
    maxbpe = max(maxbpe, HIN[n].bpe);
    HIN[n].buf_type = HOUT.buf_type;
  }

  // real time control of commutator
  ns = m_get_switch("SELECT");
  ids = m_lwinit("Select in", ns, -1, nfiles, 1);
  
  // Support old-style /SYNC switch, but /NOBLOCK overrides
  syncmode = m_get_switch("SYNC");
  if (m_get_pswitch("NOBLOCK")) syncmode = 0;

  // get a buffer 
  lper = lframe;				//MIB
  vector<int_1> bloc(m_bytes_reqd(maxbpe, lper));
  if(HIN[0].type/1000 == 2)
    lframe = lframe*HIN[0].adjunct.t2.subsize; //MIB
  
  // signal end of preamble
  m_sync();

  // loop through buffers
  bool got_data = false;
  n = 0;
  while (m_do(lframe, ndo)) {
    
    (void)m_lwget(ids, ns);  			// new select mode
    
    while (ns == 0) {   			// block	      
      m_pause(Mc->pause);
      (void)m_lwget(ids, ns);
      if (Mc->break_) break; 
    }
    
    if (ns < 0)
      n = (n % nfiles)+1;			// commutate thru all of the input files

    while (ndo > 0) {
      nget = min(ndo, lframe);			//MIB 18-Jan-2000
      if (ns < 0) {                             // commutate
	HIN[n-1].xfer_len = nget;
        m_grabx(HIN[n-1], &bloc[0], ngot);
        m_filad(HOUT, &bloc[0], ngot);
      } else {	
        HIN[ns-1].xfer_len = nget;
        //
        //    In NOHANG mode, only call M$GRABX if enough data is in the pipe to
        //      satisfy the request.  Also look for a new select mode more often.
        //
        if (nohang) {
          m_hcbfunc(HIN[ns-1], HCBF_AVAIL, &navail);  
          got_data = false;
	  // if ((navail >= nget) || (HIN[ns-1].size - HIN[ns-1].offset < nget )) {
	  if (navail >= nget) {
            m_grabx(HIN[ns-1], &bloc[0], ngot);
            m_filad(HOUT, &bloc[0], ngot);
            got_data = true;
          }
          (void)m_lwget(ids, ns);		// new select mode
          if (Mc->break_) {
            break; 
          }
        } else {
          m_grabx(HIN[ns-1], &bloc[0], ngot);	// normal select mode sans /NOHANG
          m_filad(HOUT, &bloc[0], ngot);
        }
        //
        //  In NOBLOCK/NOHANG mode, only read as much data as is available, otherwise the
        //    pipes could wind up with the read pointer way past the write pointer
        //    and take a long time to fill up
        //
        if (syncmode == 0) {			// NOBLOCK switch set
          for (i = 1; i <= nfiles; ++i) {
            if (i != ns) {			// fake read of other inputs
              HIN[i-1].xfer_len = 0;
              if (nohang) {			// NOHANG set as well
                m_hcbfunc(HIN[i-1], HCBF_AVAIL, &navail);
                if (navail > 0) {
                  HIN[i-1].offset = HIN[i-1].offset + navail;
                }
              } else {
                HIN[i-1].offset = HIN[i-1].offset + nget;
              }
              m_grabx(HIN[i-1], &bloc[0], ngot);
            }
          }
        }					// syncmode == 0
      }						// end else
      //
      //  If we're spinning in NOHANG mode, with no data input don't 
      //     tie up the CPU.
      //
      if (got_data  ||  !nohang) {
        ndo = ndo - nget;
      } else {
        m_pause(Mc->pause);
      }
    }
  }

  // close files
  for_each(HIN.begin(), HIN.end(), m_close);
  m_close(HOUT);
}
