/*!
  @file           ven84.c
  @author         JoergM
  @brief          Kernel Runtime: Requestor Thread
  @see            

\if EMIT_LICENCE

    ========== licence begin  GPL
    Copyright (c) 2001-2005 SAP AG

    This program is free software; you can redistribute it and/or
    modify it under the terms of the GNU General Public License
    as published by the Free Software Foundation; either version 2
    of the License, or (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with this program; if not, write to the Free Software
    Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
    ========== licence end



\endif
*/



#undef MOD__
#define MOD__ "ven84.c:"

/* exported functions */

#include "gen00.h"
#include "heo00.h"
#include "geo002.h"
#include "gsp01.h"
#include "geo50_0.h"
#include "gen500.h"
#include "vsp001c.h"
#include "gen003.h"
#include "gen81.h"
#include "gen84.h"

/* imported functions */

#include "hen40.h" /* sqlerrs */
#include "gen32.h"
#include "gen38.h"
#include "gen41.h"
#include "gen42.h"
#include "gen71.h"
#include "gen72.h"
#include "gen73.h"
#include "hsp77.h"
#include "RunTime/Configuration/RTEConf_ParameterAccessKernelInterface.h"


#define USE_PARAMETER_FOR_EXTERNAL_DUMP_REQUEST

/* local functions */

static	void			e84_request ( rte_header *header );
static	int			    e84_find_free_task ( connection_info *cip, 
											 struct TASK_TYPE **pTcb );
static	int			    e84_open_reply_channel (connection_info		*cip ,
                                                VF_FILENAMEC		file);
static	void			e84_write_reply_channel (connection_info		*cip ,
                                                rte_header		*header );
static	void			e84_close_reply_channel (connection_info		*cip );
static	void			e84_reject (connection_info	    *cip ,
                                    SQL_DBNAMEC          mydbname,
                                    rte_header		    *header ,
                                    int                  reason );
static	void			e84_accept ( connection_info  *cip ,
                                     SQL_DBNAMEC       mydbname,
				 					 rte_header		  *header,
									 struct TASK_TYPE *tcb );
static	void			e84_conn_send ( connection_info	*cip,
										struct TASK_TYPE *tcb );
static	int			    e84_reply_class ( connection_info  *cip );
static  void            en84_CatchAlarm ( int sig, siginfo_t *pInfo, void *arg );
static  void            en84_PreventClientStarvation(int peer_semid);
static  ten50_UKT_Control *en84_FindOptimalUkt ( int TaskTypeToFind );
static int en84_RecvConpkt (int              sd ,
                            rte_header		*header ,
                            int				*statep ,
                            char           **errm,
                            int              msecTimeoutTime );

static SAPDB_Bool  e84_dump_allowed ();

/* exported data */

int			         e84_connect_strategy = 0;
/* local data */

volatile int en84_NeverStop = 1;
static	int          en84_Timeout = 0;
static	tsp00_Bool   en84_DumpRequest = false;

/*
 * ======================================================================
 */
#undef MF__
#define MF__ MOD__"en84_requestor"
/*ARGSUSED*/
void *en84_requestor ( void *arg )
{
#undef	MF__
#define	MF__	"e84_requestor"
ten50_KT_Control         *this_kt;
    int				rc ;
    int				fifo_fd ;
    char			*errmsg ;
    rte_header			*header ;
    char			packet [ RTE_CONPKT_SIZE ];
    VF_FILENAMEC		file ;

	DBGIN;

    MSGD (( INFO_PROCESS_STARTED, "REQUESTOR" ));

    this_kt = & KGS-> requestor ;
    this_kt->start_time = KGS->current_time;

    DBG1 (( MF__,"REQUESTOR: thread %ld \n", (long)this_kt->tid ));

    sql41_get_request_fifo_name ( file , KGS-> serverdb );
    /*
     *  The fifo might exist from a previous run.
     */
    (void) unlink ( (const char *)file );
    rc = sql41_create_fifo ( file , 0622 );
    if ( rc )
    {
		MSGD (( ERR_REQUESTOR_CR_FIFO )) ;
		return NULL;
    }
    /*
     *  O_RDWR avoids blocking and does not deliver EOF when
     *  last client closes.
     */
    fifo_fd = open ( (const char *)file , O_RDWR , 0 );
    if ( fifo_fd == -1 )
    {
		MSGD (( ERR_REQUESTOR_OPEN_REQ_FIFO, sqlerrs() ));
		return NULL;
    }

    header = (rte_header*) packet ;
    while ( en84_NeverStop )
    {
		/*
		 *  Read a connect packet from the fifo.
		 *  Read error are checked after copy of debuglevel
		 *  and check of server state.
		 */
		DBG1 (( MF__,"recv_conpkt fd %d \n", fifo_fd ));

        en84_Timeout = 0;
        errmsg = "";

        rc = en84_RecvConpkt ( fifo_fd ,
                               header , 
                              &en84_Timeout , 
                              &errmsg, 
                               LOCL_CON_TMO * 1000 );

        if ( (KGS-> state == SERVER_STOP) || (KGS-> state == SERVER_KILL) )
		{
			MSGCD (( INFO_PROCESS_STOPPED, "REQUESTOR" ));
            break;
		}

		/*
		 *  Now, check return code of the receive packet operation.
		 *  Try to recover from read errors by re-opening the fifo.
		 */
		if ( rc != SP1CE_OK )
		{
			if ( en84_Timeout != CON_TIMEDOUT )
			{
                const char *errMsg;
				MSGD (( WRN_REQUESTOR_REOPEN_FIFO, rc , errmsg, file ));
				(void) close ( fifo_fd );

                /* PTS 1109407 */
				errMsg = en41CheckFifo(file, 0622);
				if ( errMsg )
				{
					MSGCD (( ERR_REQUESTOR_OPEN_REQ_FIFO, errMsg ));
					vabort(0);
                }

				fifo_fd = open ( (const char *)file , O_RDWR , 0 );
				if ( fifo_fd == -1 )
				{
					MSGCD (( ERR_REQUESTOR_OPEN_REQ_FIFO, sqlerrs() ));
					vabort(0);
				}
			}
			continue ;
		}

#ifdef DEBUG_DEBUG
        { 
            extern void e74_print_all_conns ( char * );
    		e74_print_all_conns ( "REQUESTOR" );
        }
#endif /*DEBUG_DEBUG*/

		/*
		 *  Check and process the request.
		 */
		e84_request ( header );
    }

    en81_NotifyCoordinator(0);
    while ( en84_NeverStop )
    {
        sqlyieldthread();
    }
    /*NOTREACHED*/
    return NULL;
}
/* e84_requestor */

/*
 * ======================================================================
 */
#undef	MF__
#define	MF__	MOD__"e84_request"
static  void    e84_request ( 
rte_header		*header )
{
    int				rc ;
    int                         retcod ;
    tsp00_Int4			pktcnt4 ;
    tsp00_Int4			semid4 ;
    tsp00_Int4			shmid4 ;
    long			lpid ;
    connection_info		*cip ;
    connection_info		connection_buffer ;
    SQL_DBNAMEC                 mydbname ;
    VF_FILENAMEC		file ;
    char			logbuf [ 256 ];
    struct TASK_TYPE		*tcb ;


    /*
     *  Extract the various information from the connect packet
     *  and store them in the connection info buffer.
     */
    cip = &connection_buffer ;
    FILL ( cip , 0 , sizeof(*cip) );
    sql42_extract_conpkt ( header , &cip->ci_mess_class ,
			    &cip->ci_my_ref , &cip->ci_peer_ref ,
			    &retcod , &cip->ci_service ,
			    &cip->ci_max_segment_size ,
			    &cip->ci_max_data_size , &cip->ci_packet_size ,
			    &cip->ci_min_reply_size , &cip->ci_peer_swap_type ,
			    mydbname , cip->ci_peer_dbname );
    sql42_get_string ( header , RSQL_OPT_NODE_EO003 , sizeof(cip->ci_peer_node) ,
							cip->ci_peer_node );
    sql42_get_string ( header , RSQL_OPT_REM_PID_EO003 , sizeof(cip->ci_remote_pid) ,
						    cip->ci_remote_pid );
    sql42_get_long ( header , RSQL_OPT_OFFSET_EO003 , &cip->ci_big_offset );
    sql42_get_int1 ( header , RSQL_OPT_ALTER_SERV_SEM_EO003 , &cip->ci_alter_server_sem );

    /*
     *  Option values are unchanged, if not present in packet.
     */
    pktcnt4 = 1 ;
    lpid = 0 ;
    semid4 = 0 ;
    shmid4 = 0 ;
    file[0] = 0 ;
    sql42_get_int4 ( header , RSQL_OPT_PKTCNT_EO003 , &pktcnt4 );
    sql42_get_long ( header , RSQL_OPT_PID_EO003    , &lpid );
    sql42_get_int4 ( header , RSQL_OPT_SEMID_EO003  , &semid4 );
    sql42_get_int4 ( header , RSQL_OPT_SHMID_EO003  , &shmid4 );
    sql42_get_string ( header , RSQL_OPT_FIFO_EO003 , sizeof(file) , file );
    cip->ci_packet_cnt = pktcnt4 ;
    cip->ci_peer_pid = (pid_t) lpid ;
    cip->ci_peer_semid = semid4 ;
    cip->ci_shmid = shmid4 ;
    DBG1 (( MF__,"REQUESTOR: mess_class %d service %d pktcnt %d \n",
	    cip->ci_mess_class , cip->ci_service , cip->ci_packet_cnt ));
    DBG1 (( MF__,"REQUESTOR: peerpid %ld peerref %d semid %d shmid %d \n",
	    (long) cip->ci_peer_pid , cip->ci_peer_ref ,
	    cip->ci_peer_semid , cip->ci_shmid ));
    DBG1 (( MF__,"REQUESTOR: node '%s' db '%s' rempid '%s' \n",
	    cip->ci_peer_node , cip->ci_peer_dbname , cip->ci_remote_pid ));
    DBG1 (( MF__,"REQUESTOR: fifo '%s' \n", file ));

    rc = e84_open_reply_channel ( cip , file );
    if ( rc != SP1CE_OK )
    {
		return ;
    }

    if ( (KGS-> state != SERVER_COLD)         &&
         (KGS-> state != SERVER_STANDBY_EO00) &&
		 (KGS-> state != SERVER_WARM)    )
    {
		MSGD (( INFO_REQUESTOR_REJECT, KGS-> state ));
		switch ( KGS-> state )
		{
		case SERVER_STARTING :
			e84_reject ( cip, mydbname, header , SP1CE_START_REQUIRED );
			break ;
		default :
			e84_reject ( cip , mydbname, header , SP1CE_SHUTDOWN );
			break ;
		}
		e84_write_reply_channel ( cip , header );
		e84_close_reply_channel ( cip );
		return ;
    }

    /*
     *  When handling a CONNECT REQUEST or an INFO REQUEST
     *  we insert the current time into 'tcb->connection->ci_connect_time'
     *  to give 'e84_test_application' a chance to clear tasks
     *  which have (tcb->connection->ci_connect_time != 0) and
     *  (tcb->state == TSK_CONNECTWAIT) for too long a time.
     */
    switch ( cip->ci_mess_class )
	{
	case RSQL_INFO_REQUEST_EO003 :
    case RSQL_INFO_REQUEST_KEEP_ALIVE_EO003:
        /*
		 *  Reply to the client.
		 */
		e84_accept ( cip, mydbname, header, tcb );
		break ;

    case RSQL_USER_CONN_REQUEST_EO003 :
	case RSQL_KERN_CONN_REQUEST_EO003 :

		rc = e84_find_free_task ( cip, &tcb );
		if ( rc != SP1CE_OK )
		{
			MSGD (( WRN_REQUESTOR_REJ_REQ )) 
			e84_reject ( cip , mydbname, header , rc );
			break ;
		}
		if ( ! tcb )
		{
			MSGD (( IERR_REQUESTOR_CONN_NULL_TSK ));
			e84_reject ( cip , mydbname, header , SP1CE_TASKLIMIT );
			break ;
		}
		if ( ((tcb->type == TT_US) 
          ||  (tcb->type == TT_EV)) &&
			(tcb->state != TSK_CONNECTWAIT) &&
			(tcb->state != TSK_INACTIVE) )
		{
			MSGD (( IERR_REQUESTOR_CONN_WRONG_ST, 
					tcb->state ));
			e84_reject ( cip, mydbname, header , SP1CE_TASKLIMIT );
			break ;
		}

		tcb->rte_comm.to_cancel = 0 ; /* reset cancellation flag */

		/*
		 *  Log the connect request.
		 */
		(void) sp77sprintf ( logbuf , sizeof(logbuf), "Connecting T%d" , tcb->index );
		rc = strlen ( logbuf );
		if ( cip->ci_peer_node[0] )
			(void) sp77sprintf ( logbuf + rc , sizeof(logbuf)-rc, " %s" , cip->ci_peer_node );
		else
			(void) sp77sprintf ( logbuf + rc , sizeof(logbuf)-rc, " local" );
		rc = strlen ( logbuf );
		if ( cip->ci_peer_dbname[0] )
		{
			(void) sp77sprintf ( logbuf + rc , sizeof(logbuf)-rc, ":%s" , cip->ci_peer_dbname );
			rc = strlen ( logbuf );
		}
		if ( cip->ci_remote_pid[0] )
			(void) sp77sprintf ( logbuf + rc , sizeof(logbuf)-rc, " %s" , cip->ci_remote_pid );
		else
			(void) sp77sprintf ( logbuf + rc ,  sizeof(logbuf)-rc," %ld" , (long) cip->ci_peer_pid );

        MSGD (( INFO_CONNECTING_TASK, logbuf ));

        /*
		 *  Reply to the client.
		 */
		e84_accept ( cip, mydbname, header, tcb );

		if ( cip->ci_mess_class != RSQL_INFO_REQUEST_EO003 )
		{
			/*
			 *  Inform the UKP of the connect request.
			 */
			e84_conn_send ( cip, tcb );
		}
		break ;

	case RSQL_USER_CANCEL_REQUEST_EO003 :
		MSGD (( INFO_REQUESTOR_CANCEL_REQ, cip->ci_my_ref ));
		if (    (cip->ci_my_ref < 1)
			 || (cip->ci_my_ref > KGS-> ulNumOfTasks) )
		{
			MSGD (( IERR_REQUESTOR_CANCEL_ILL_T, cip->ci_my_ref ));
			break ;
		}
		tcb = &KGS-> pFirstTaskCtrl[cip->ci_my_ref-1] ;
		tcb->rte_comm.to_cancel = 1 ;
        /* PTS 1105149 */
        if ( tcb->type == TT_EV )
        {
          if ( tcb->connection )
          {
            tcb->connection->ci_state = CON_ABORTED ;
            e72_wake ( tcb->ukt ); /* wakeup if blocked in dispatcher */
          }
  		  tcb->rte_comm.to_cancel = 1 ;  /* if in kernel code */
          vcontinue(tcb->index); /* wakeup if block in vstop() */
        }
        else
        {
          /* If client was not able to attach to shared memory, it must abort connection */
          /* Its only way to do that, is by sending a cancel request. This leads to ABORT of connection */
          /* If the client was already connected, the behaviour of a cancel is as before. Only commands */
          /* are canceled, but the connection is kept alive */
          if ( tcb->connection 
            && (CON_CONNECTING == tcb->connection->ci_state
             || TSK_CONNECTWAIT == tcb->state
             || TSK_INACTIVE     == tcb->state
             || (TSK_VRECEIVE     == tcb->state
              && 0 == tcb->connection->BytesReceived) ) )
          {
             MSGD (( WRN_REQUESTOR_CNCL_ABORT_0, cip->ci_my_ref ));
             MSGD (( WRN_REQUESTOR_CNCL_ABORT_1 ));
             tcb->connection->ci_state = CON_ABORTED ;
             e72_wake ( tcb->ukt ); /* wakeup if blocked in dispatcher */
          }
        }
		break ;

	case RSQL_DUMP_REQUEST_EO003 :

        if ( e84_dump_allowed () )
        {
            MSGCD (( INFO_REQUESTOR_DUMP_REQ )) 
                if ( en84_DumpRequest )
                {
                    MSGCD (( ERR_REQUESTOR_NO_MORE_DUMP )) 
                }
                else
                {
                    en84_DumpRequest = true ;
                    KGS-> mode = MODE_DUMP ;
                    (void) en81_kill_database ();
                }
        }
		break ;

	default :
		MSGD (( WRN_REQUESTOR_ILL_MSG_TYPE , cip->ci_mess_class ));
		e84_reject ( cip , mydbname, header , SP1CE_NOTOK );
		break;
	}

    e84_write_reply_channel ( cip , header );
    e84_close_reply_channel ( cip );
}
/* e84_request */

/*
 * ======================================================================
 */

#undef	MF__
#define	MF__	MOD__"e84_dump_allowed"
static SAPDB_Bool e84_dump_allowed()
{
  SAPDB_Bool                     dumpEnabled = false;

#if defined (USE_PARAMETER_FOR_EXTERNAL_DUMP_REQUEST)
  SAPDB_Char*                  errText;
  SAPDB_Char*                  valueString;

  if( RTEConf_ParameterAccessGetString( KERNELPARAM_EXTERNAL_DUMP_REQUEST, &valueString, &errText ))
  {
      if ( 0 == strcmp("YES", valueString ))
          dumpEnabled = true;
      else
          dumpEnabled = false;
  }
  else
      dumpEnabled = false;
#else
    dumpEnabled = true;
#endif

    return dumpEnabled;
}

/*
 * ======================================================================
 */

#undef	MF__
#define	MF__	MOD__"e84_find_free_task"
static  int  e84_find_free_task ( 
connection_info		*cip,
struct TASK_TYPE **pTcb )
{
    int				    idx ;
    ten50_UKT_Control	*Ukt ;
    ten50_UKT_Control	*OptimalUkt ;
    struct TASK_TYPE	*tcb ;
    connection_info		*tsk_cip ;
    tsp00_Int4           TaskType;

    DBGIN;

    if ( cip->ci_packet_cnt != 1 )
    {
		MSGD (( WRN_REQUESTOR_PACK_CNT_NE_1  ));
		return ( SP1CE_NOTOK );
    }

    /*
     *  Is the CONNECT REQUEST for the Utility Task ?
     */
    switch ( cip->ci_service )
    {
    case srvUtility_esp01:
		DBG1 (( MF__,"e84_find: trying to connect utility \n" ));
		/*
		 *  Test whether a possibly existing utility connection is still alive.
		 */
	    e84_test_application ( KGS-> ut );

		if  (( KGS-> ut->state == TSK_CONNECTWAIT ) ||
             ( KGS-> ut->state == TSK_INACTIVE ))
		{
			DBG1 (( MF__,"utility is waiting for connect \n" ));
			if ( ! KGS-> ut->connection->ci_connect_time )
			{
				DBG1 (( MF__,"connecting UTILITY \n" ));
				tcb = KGS-> ut ;
				WAIT_UNTIL_ALONE ( tcb->ukt->exclusive );
				tcb->connection->ci_connect_time = KGS->current_time;
				CLEARLOCK( tcb->ukt->exclusive );
				*pTcb = tcb;
				DBG1 (( MF__,"returning SP1CE_OK \n" ));
				return ( SP1CE_OK );
			}
		}
		MSGD (( WRN_REQUESTOR_UT_IS_BUSY, 
			KGS-> ut->state , (long) KGS-> ut->connection->ci_connect_time ));
		*pTcb = NULL;
		return ( SP1CE_TASKLIMIT );
    case srvDistribution_esp01: /* no longer supported */
	    return ( SP1CE_NOTOK );
    case srvEvent_esp01:
        if ( XPARAM(ulMaxEventTasks) == 0 )
        {
    		*pTcb = NULL;
	    	return ( SP1CE_TASKLIMIT );
        }
	    break;
    default:
	    break;
    }

    OptimalUkt = en84_FindOptimalUkt( (cip->ci_service == srvEvent_esp01 ? TT_EV : TT_US) );

    /* if no optimal UKT was found, try to release connections no longer in use first */
    if ( OptimalUkt == NULL )
    {
        /*
         *  The CONNECT REQUEST is for a user or event task.
         *
         *  Test, whether the application processes for all user tasks
         *  are still alive.
         */
        DBG1 (( MF__,"testing applications \n" ));
        for ( tcb = KGS-> first_event_task ; tcb < KGS-> first_backup_task ; tcb ++ )
        {
			e84_test_application ( tcb );
            sqlyieldthread();
        }
        for ( tcb = KGS-> first_user ; tcb <= KGS-> pLastTaskCtrl ; tcb ++ )
        {
			e84_test_application ( tcb );
            sqlyieldthread();
        }
        DBG1 (( MF__,"tested  applications \n" ));
        /*
         * Search again (hoping to find a free slot now...
         */
        OptimalUkt = en84_FindOptimalUkt( (cip->ci_service == srvEvent_esp01 ? TT_EV : TT_US) );
    }

    /*
     * Dump State of all connection in all UKTS for debugging purpose
     */
    if ( ! OptimalUkt )
    {
		MSGD (( WRN_REQUESTOR_ALL_UKPS_BUSY ));

        for ( tcb = KGS->pFirstTaskCtrl ; tcb <= KGS->pLastTaskCtrl ; tcb ++ )
        { 
            if ( tcb->connection )
            {
                MSGD (( INFO_REQUESTOR_FIND_ALL_TSK, 
                    THREAD_INDEX(tcb->ukt), tcb->index ,
                    (long) tcb->connection->ci_connect_time ,
                    e38_tskstate_name(tcb->state) ));
            }
		}

		*pTcb = NULL;
		return ( SP1CE_TASKLIMIT );
    }

    /*
     *  Locate the free user task within the optimal UKT.
     */
    DBG1 (( MF__,"searching free task of KT%d \n", THREAD_INDEX(OptimalUkt) ));
    for ( tcb = KGS->pFirstTaskCtrl ; tcb <= KGS->pLastTaskCtrl ; tcb ++ )
    { 
        if ( tcb->ukt != OptimalUkt ) 
            continue; /* we only searching for the optimal UKTs tasks */

		DBG1 (( MF__,"task type  %d \n", tcb->type ));
        if ( (cip->ci_service != srvEvent_esp01 && tcb->type != TT_US)
          || (cip->ci_service == srvEvent_esp01 && tcb->type != TT_EV) )
            continue ;

		DBG1 (( MF__,"task state %d, connected %ld \n",
			tcb->state , (long) tcb->connection->ci_connect_time ));
		if ( tcb->connection->ci_connect_time ||
			 ( (tcb->state != TSK_CONNECTWAIT) &&
			   (tcb->state != TSK_INACTIVE)         ) ) continue ;

		DBG1 (( MF__,"found T%d \n", tcb->index ));

		WAIT_UNTIL_ALONE ( tcb->ukt->exclusive );
		tcb->connection->ci_connect_time = KGS->current_time;
		tsk_cip = tcb->connection ;
		tsk_cip->ci_service = cip->ci_service ;
		CLEARLOCK ( tcb->ukt->exclusive );

        *pTcb = tcb ;
		DBG1 (( MF__,"returning SP1CE_OK \n" ));
		return ( SP1CE_OK );
    }

    /*
     * Dump State of all connection in optimal UKT for debugging purpose
     */
    MSGD (( WRN_REQUESTOR_ALL_IN_UKP_BUS , THREAD_INDEX(OptimalUkt) ));

    for ( tcb = KGS->pFirstTaskCtrl ; tcb <= KGS->pLastTaskCtrl ; tcb ++ )
    { 
        MSGD (( INFO_REQUESTOR_FIND_UKP_TSK, 
            THREAD_INDEX(tcb->ukt), ((tcb->ukt == OptimalUkt)?'*':' ') ,
            tcb->index , (long) tcb->connection->ci_connect_time ,
            e38_tskstate_name(tcb->state) ));
	}

    *pTcb = NULL;
    DBG1 (( MF__,"returning SP1CE_TASKLIMIT \n" ));
    return ( SP1CE_TASKLIMIT );
}

/*
 * ======================================================================
 */

#undef	MF__
#define	MF__	MOD__"e84_open_reply_channel"
static	int	e84_open_reply_channel ( 
connection_info		*cip ,
VF_FILENAMEC		file )
{
    int				rc ;


    /*
     *  Open the reply channel.
     */
    cip->ci_pipe_fd = -1 ;
    if ( file[0] )
    {
	/*
	 *  O_RDWR avoids blocking.
	 *  The data are written into the fifo, wether the client has
	 *  the fifo open or not. If it opens later it gets the data then.
	 */
	cip->ci_pipe_fd = open ( (const char *)file , O_RDWR , 0 );
	if ( cip->ci_pipe_fd != -1 )
	{
	    DBG1 (( MF__,"opened '%s' fd %d \n", file , cip->ci_pipe_fd ));
	    return ( SP1CE_OK );
	}
	MSGD (( ERR_REQUESTOR_OPEN_REPL_FIFO, file , sqlerrs() ));
    }
    else
    {
	if ( (cip->ci_shmid != 0) && (cip->ci_peer_semid != 0) )
	{
#ifdef FIXED_COMSEG
	    cip->ci_loccomseg = KGS-> first_kp->connection->ci_loccomseg ;
#endif
	    rc = sql32_attach_comseg ( cip );
	    if ( rc == SP1CE_OK )
	    {
		/* initialize as 'read by server' */
		cip->ci_comseg->cs_client_flag = 2 ;
		cip->ci_comseg->cs_server_flag = 0 ;
		return ( SP1CE_OK );
	    }
	    else
	    {
		MSGD (( ERR_REQUESTOR_ATT_SHM, cip->ci_shmid , sqlerrs() ));
	    }
	}
	else
	{
	    DBG1 (( MF__,"no reply channel available" ));
	    DBG1 (( MF__,"fifo '%s' shmid %d semid %d",
			    file , cip->ci_shmid , cip->ci_peer_semid ));
	    /*
	     *  Cancel and Dump don't need a reply.
	     */
	    return ( SP1CE_OK );
	}
    }

    if (cip->ci_shmid > 0)
    {
	DBG1 (( MF__,"removing  comseg %d \n", cip->ci_shmid ));
	(void) sql41_remove_shm ( &cip->ci_shmid , "us" , KGS-> serverdb );
    }

    /*
     *  If there is a semid available,
     *  try to save the client from hanging on its semaphore.
     */
    if ( cip->ci_peer_semid )
    {
	rc = semop ( cip->ci_peer_semid , &semsnd , 1 );
	if ( rc == -1 )
	    MSGD (( ERR_REQUESTOR_REL_SEMID, 
					cip->ci_peer_semid , sqlerrs() ));
	DBG1 (( MF__,"semop semid %d rc %d \n", cip->ci_peer_semid , rc ));

    DBG1 (( MF__,"removing semid %d \n", cip->ci_peer_semid ));
    (void) sql41_remove_sem ( &cip->ci_peer_semid ,
					"us" , KGS-> serverdb );
    }

    return ( SP1CE_NOTOK );
}

/*
 * ======================================================================
 */
#undef	MF__
#define	MF__	MOD__"e84_write_reply_channel"
static  void    e84_write_reply_channel ( 
connection_info		*cip ,
rte_header		*header )
{
    tsp00_ErrTextc pErrText;

    if ( cip->ci_pipe_fd != -1 )
    {
	DBG1 (( MF__,"writing conpkt to pipe fd %d lgt %d \n",
		cip->ci_pipe_fd , header->rh_max_send_len ));
	(void) sql42_send_conpkt ( cip->ci_pipe_fd , header , pErrText );
    }
    else
    {
	if ( cip->ci_request )
	{
	    DBG1 (( MF__,"copying conpkt to shm \n" ));
	    COPY ( cip->ci_request , header , header->rh_max_send_len );
	}
	if ( cip->ci_comseg )
	{
	    DBG1 (( MF__,"setting comseg flag \n" ));
	    cip->ci_comseg->cs_server_state = header->rh_rte_return_code ;
	    cip->ci_comseg->cs_server_flag = 1 ;
	    /* setting the semaphore is done by e84_close_reply_channel */
	}
    }
}

/*
 * ======================================================================
 */

#undef	MF__
#define	MF__	MOD__"e84_close_reply_channel"
static	void	e84_close_reply_channel ( 
connection_info		*cip )
{
    int				rc ;
    char			*ptr ;


    /*
     *  Close the reply channel.
     */
    if ( cip->ci_pipe_fd != -1 )
    {
	/*
	 *  O_RDWR avoids blocking.
	 *  The data are written into the fifo, wether the client has
	 *  the fifo open or not. If it opens later it gets the data then.
	 */
	DBG1 (( MF__,"closing pipe \n" ));
	(void) close ( cip->ci_pipe_fd );
	return ;
    }

    if ( cip->ci_big_comseg || cip->ci_comseg )
    {
	ptr = (char*) cip->ci_big_comseg ;
	if ( ! ptr ) ptr = (char*) cip->ci_comseg ;
	DBG1 (( MF__,"detaching comseg 0x%08lx \n", (long) ptr ));
	(void) sql41_detach_shm ( &ptr );
	cip->ci_big_comseg = 0 ;
	cip->ci_comseg = 0 ;
	cip->ci_request = 0 ;
	cip->ci_reply = 0 ;
    }

    if ( cip->ci_shmid > 0)
    {
	DBG1 (( MF__,"removing  comseg %d \n", cip->ci_shmid ));
	(void) sql41_remove_shm ( &cip->ci_shmid , "us" , KGS-> serverdb );
    }

    /*
     *  Free the client's semaphore.
     */
    if ( cip->ci_peer_semid )
    {
	rc = semop ( cip->ci_peer_semid , &semsnd , 1 );
	if ( rc == -1 )
	    MSGD ((ERR_REQUESTOR_FREE_SEMID, 
				    cip->ci_peer_semid , sqlerrs() ));
	DBG1 (( MF__,"semop semid %d rc %d \n", cip->ci_peer_semid , rc ));

    DBG1 (( MF__,"removing semid %d \n", cip->ci_peer_semid ));
    (void) sql41_remove_sem ( &cip->ci_peer_semid ,	"us" , KGS-> serverdb );
    }
}

/*
 * ======================================================================
 */
#undef	MF__
#define	MF__	MOD__"e84_reject"
static  void    e84_reject (connection_info		*cip,
                            SQL_DBNAMEC          mydbname,
                            rte_header		    *header,
                            int                  reason)
{
    int				messclass ;


    DBG1 (( MF__,"entered, reason %d \n", reason ));

    messclass = e84_reply_class ( cip );
    sql42_create_conpkt ( header , messclass ,
			    (tsp00_Int4)0 , cip->ci_peer_ref , reason ,
			    cip->ci_service , cip->ci_max_segment_size ,
			    cip->ci_max_data_size , cip->ci_packet_size ,
			    cip->ci_min_reply_size ,
			    mydbname, cip->ci_peer_dbname );

    DBGOUT;
}

/*
 * ======================================================================
 */
#undef	MF__
#define	MF__	MOD__"e84_accept"
static	void	e84_accept (connection_info		*cip ,
                            SQL_DBNAMEC          mydbname,
                            rte_header		    *header,
                            struct TASK_TYPE    *tcb )
{
    int				messclass ;
    long			maxdatasize ;
    connection_info		*tsk_cip ;

    DBGIN;

    if ( cip->ci_comseg && ! cip->ci_big_offset )
		cip->ci_protocol = PROT_SHM_EO003 ;
    else
		cip->ci_protocol = PROT_BIGSHM_EO003 ;

    if (    (cip->ci_packet_size <= 0)
	 || (cip->ci_packet_size > XPARAM(ulPacketSize)) )
	cip->ci_packet_size = XPARAM(ulPacketSize) ;

    maxdatasize = XPARAM(ulPacketSize) - sizeof(comseg_header)
			- 8 /*alignment*/ - 2 * sizeof(rte_header);
    if (    (cip->ci_max_data_size <= 0)
	 || (cip->ci_max_data_size > maxdatasize) )
	cip->ci_max_data_size = maxdatasize ;

    if (    (cip->ci_min_reply_size <= 0)
	 || (cip->ci_min_reply_size > XPARAM(ulMinReplySize)) )
	cip->ci_min_reply_size = XPARAM(ulMinReplySize) ;

    cip->ci_max_request_size = cip->ci_max_data_size - cip->ci_min_reply_size ;

    if (   cip->ci_mess_class != RSQL_INFO_REQUEST_EO003
        && cip->ci_mess_class != RSQL_INFO_REQUEST_KEEP_ALIVE_EO003)
    {
	    /*
	     *  Copy the connect information to the task's structure.
	     */
	    tsk_cip = tcb->connection;
	    tsk_cip->ci_state = CON_CONNECTING ;
        tsk_cip->BytesReceived = tsk_cip->BytesSend = 0;
	    tsk_cip->ci_use_count ++ ;
	    tsk_cip->ci_service = cip->ci_service ;
	    tsk_cip->ci_protocol = cip->ci_protocol ;
	    tsk_cip->ci_packet_cnt = cip->ci_packet_cnt ;
	    tsk_cip->ci_packet_size = cip->ci_packet_size ;
	    tsk_cip->ci_min_reply_size = cip->ci_min_reply_size ;
	    tsk_cip->ci_max_data_size = cip->ci_max_data_size ;
	    tsk_cip->ci_max_request_size = cip->ci_max_request_size ;
	    tsk_cip->ci_peer_pid = cip->ci_peer_pid ;
	    tsk_cip->ci_peer_ref = cip->ci_peer_ref ;
	    tsk_cip->ci_peer_semid = cip->ci_peer_semid ;
      tsk_cip->ci_alter_server_sem = cip->ci_alter_server_sem;

	    COPY ( tsk_cip->ci_peer_node , cip->ci_peer_node ,
				        sizeof(cip->ci_peer_node) );
	    COPY ( tsk_cip->ci_peer_dbname , cip->ci_peer_dbname ,
				        sizeof(cip->ci_peer_dbname) );
	    COPY ( tsk_cip->ci_remote_pid , cip->ci_remote_pid ,
				        sizeof(cip->ci_remote_pid) );

	    /*
	     *  If the client does not supply a comseg, use own big comseg.
	     *  This is true for both: a local big comseg request
	     *  and a remote request.
	     */
	    if ( cip->ci_shmid <= 0)
	    {
	        tsk_cip->ci_shmid = 0 ;
	        tsk_cip->ci_big_comseg = tcb->bigComSegBelongsToUKT->connection->ci_big_comseg ;
	        tsk_cip->ci_big_offset = tsk_cip->ci_locbigoffset ;
	        tsk_cip->ci_big_size   = tcb->bigComSegBelongsToUKT->connection->ci_big_size ;
	    }
	    else
	    {
	        tsk_cip->ci_shmid = cip->ci_shmid ;
	    }

        tsk_cip->ci_my_ref = tcb->index ;
	    DBG1 (( MF__,"T%d my_ref %d peer_ref %d \n",
		    tcb->index,
		    tsk_cip->ci_my_ref , tsk_cip->ci_peer_ref ));

    }
    /* preliminary */

    messclass = e84_reply_class ( cip );
    if (    (cip->ci_mess_class != RSQL_INFO_REQUEST_EO003
             && cip->ci_mess_class != RSQL_INFO_REQUEST_KEEP_ALIVE_EO003)
	     && (cip->ci_shmid <= 0))
	{

        sql42_create_conpkt ( header , messclass ,
			    tcb->index , cip->ci_peer_ref , SP1CE_OK ,
			    cip->ci_service , cip->ci_max_segment_size ,
			    cip->ci_max_data_size , cip->ci_packet_size ,
			    cip->ci_min_reply_size ,
			    mydbname, cip->ci_peer_dbname);

        /* local connect */
	    sql42_put_int4 ( header , RSQL_OPT_PKTCNT_EO003 ,(tsp00_Int4) cip->ci_packet_cnt );
	    sql42_put_int4 ( header , RSQL_OPT_SEMID_EO003 , (tsp00_Int4) tcb->ukt->semid );
	    sql42_put_long ( header , RSQL_OPT_PID_EO003   , (long)tcb->ukt->tid );
	    sql42_put_int4 ( header , RSQL_OPT_SHMID_EO003 , (tsp00_Int4) tcb->bigComSegBelongsToUKT->connection->ci_shmid );
	    sql42_put_long ( header , RSQL_OPT_OFFSET_EO003 , tsk_cip->ci_locbigoffset );
	}
    else
    {
        sql42_create_conpkt ( header , messclass ,
			    cip->ci_my_ref , cip->ci_peer_ref , SP1CE_OK ,
			    cip->ci_service , cip->ci_max_segment_size ,
			    cip->ci_max_data_size , cip->ci_packet_size ,
			    cip->ci_min_reply_size ,
			    mydbname, cip->ci_peer_dbname);
    }

    DBGOUT;
}

/*
 * =======================================================================
 */
#undef	MF__
#define	MF__	MOD__"e84_conn_send"
static  void    e84_conn_send ( 
connection_info		*cip,
struct TASK_TYPE *tcb )
{
    struct DOUBLY_LINKED	*lquu ;

    DBGIN;

    /*  start task if inactive! */
    if ((( tcb->type == TT_US ||
           tcb->type == TT_EV )||
         ( tcb->type == TT_UT ))  &&
         ( tcb->state == TSK_INACTIVE ))
    {
        lquu = e73_dl_dequ ( & KGS-> freelist );
        lquu->req_type = REQ_INITOK;
        en71_iocEnquAndWake ( tcb->ukt, tcb, lquu );
    }

    /*
     *  After an accepted connect: send request to proper UKP
     *  This procedure is called from the REQUESTOR process, and
     *  enters the queue element in the IOC queue, the only
     *  queue for external processes
     */
    lquu = e73_dl_dequ ( & KGS-> freelist );
    lquu->req_type = REQ_CONNECT ;
    lquu->args.conn_auftrag.task_index = tcb->index;
    DBG1 (( MF__,"put IOC request %d T%d \n", lquu->req_type , tcb->index ));
    en71_iocEnquAndWake ( tcb->ukt, tcb, lquu );

    DBGOUT;
}

/*
 * ======================================================================
 */
#undef	MF__
#define	MF__	MOD__"e84_test_application"
void    e84_test_application ( 
struct TASK_TYPE                *tcb )
{
    pid_t			  pid ;
    int				  task_state ;
    time_t			  connect_time ;

    /*
     *  Test, whether the application process is still alive.
     *  If died, cancel its task session.
     */
    if ( tcb->type != TT_UT 
      && tcb->type != TT_EV 
      && tcb->type != TT_US )
    {
		MSGD (( IERR_REQUESTOR_TEST_ILL_TASK ));
		return ;
    }

    /*
     *  Copy the states to avoid inconsistent values
     *  due to process scheduling during task release.
     */
    task_state   = tcb->state ;
    connect_time = tcb->connection->ci_connect_time ;

    if ( ! connect_time ) return ;

    DBG1 (( MF__,"check T%d \n", tcb->index ));

    /*
     *  If a task had been selected for a connect,
     *  'tcb->connection->ci_connect_time' got the current time value.
     *  If the parcitipating processes failed to complete the
     *  connection establishment, the task hangs in an intermediate state.
     *  If this lasts too long, free the task.
     */
    if ( ((task_state == TSK_CONNECTWAIT) || (task_state == TSK_INACTIVE) )
	     && (    tcb->connection->ci_state == CON_CONNECTING ))
	{
		time_t time_now = KGS->current_time;

        if ( (time_now - connect_time) > (2 * SOCK_CON_TMO) )
		{
			MSGD (( WRN_REQUESTOR_CNCL_FROM_CONN , tcb->index , task_state ));
			MSGD (( WRN_REQUESTOR_CNCL_CONN_TIME,  tcb->index , (long) connect_time , (long) time_now ));
            /* Previous implementation destroyed connect_time information 
               This was more than deadly, since almost all tests used connect_time
               to find active communication segments... Changing communication state
               from CON_CONNECTING to CON_ABORTED is possibly colliding with 
               e75_connect_request changing from CON_CONNECTING to CON_ESTABLISHED.
               But if interleaved either 'CON_ABORTED' will overwrite CON_CON_ESTABLISHED
               or 'CON_CON_ESTABLISHED' will overwrite 'CON_ABORTED'. The latter case is
               unproblematik, since it simply erases the effect of a timeout condition.
               The first one leads to abort of communication soon afterwards, which is ok
               since this is what was intended by the timeout anyhow... But in most cases
               the CON_ABORT will be noticed during connect request handling and will
               simply release the connection as intended. */
            tcb->connection->ci_state = CON_ABORTED ; /* PTS 1113931 */
            e72_wake ( tcb->ukt );
		}
	}

    /*
     *  Test applications of locally connected tasks only.
     */
    if ( (  (tcb->connection->ci_state != CON_TIMEDOUT)
         && (tcb->connection->ci_state != CON_ABORTED)
         && (tcb->connection->ci_state != CON_CONNECTING)
         && (tcb->connection->ci_state != CON_UNUSED)
         && (tcb->connection->ci_peer_semid > 0) )
      || (  (tcb->connection->ci_state == CON_UNUSED)
         && (tcb->state == TSK_VRELEASE)
         && tcb->connection->ci_connect_time ) )
    {
		pid = tcb->connection->ci_peer_pid ;
		DBG1 (( MF__,"test pid %ld \n", (long) pid ));
		if ( pid > 1 )
		{
			if ( kill ( pid , 0 ) != 0 )
			{
				if ( errno == ESRCH )
				{
					MSGD (( WRN_KILL_TSK_DIED_APPL, tcb->index , (long) pid ));
					tcb->connection->ci_state = CON_ABORTED ;
					tcb->rte_comm.to_cancel = 1 ;  /* if in kernel code */
                    e72_wake ( tcb->ukt );
				}
			}
		}
    }

    /*
     *  This should never be true here!
     */
    if (    (tcb->connection->ci_state == CON_UNUSED)
         && (tcb->state != TSK_VRELEASE)
         && tcb->connection->ci_connect_time )
    {
		MSGD (( WRN_POSSIBLE_UNUSED_CONN, tcb->index ));
    }
}


/*
 * ======================================================================
 */
#undef	MF__
#define	MF__	MOD__"e84_reply_class"
static  int    e84_reply_class ( 
connection_info		*cip )
{
    int				messclass ;


    switch ( cip->ci_mess_class )
    {
    case RSQL_INFO_REQUEST_EO003 :
    case RSQL_INFO_REQUEST_KEEP_ALIVE_EO003 :
	messclass = RSQL_INFO_REPLY_EO003 ;
	break ;
    case RSQL_USER_CONN_REQUEST_EO003 :
	messclass = RSQL_USER_CONN_REPLY_EO003 ;
	break ;
    case RSQL_KERN_CONN_REQUEST_EO003 :
	messclass = RSQL_KERN_CONN_REPLY_EO003 ;
	break ;
    default :
	messclass = cip->ci_mess_class ;
	break ;
    }

    return ( messclass );
}

/*==========================================================================*/
 
#undef MF__
#define MF__ MOD__"en84_FindOptimalUkt"
static ten50_UKT_Control *en84_FindOptimalUkt ( int TaskTypeToFind )
{
    tsp00_Int4			ActiveTasks ;
    tsp00_Int4			BusyTasks ;
    tsp00_Int4			OptUktBusyTasks = 0 ;
    tsp00_Int4			ConnectableTasks ;
    ten50_UKT_Control	*Ukt ;
    ten50_UKT_Control   *OptimalUkt = NULL;
    struct TASK_TYPE	*tcb ;

    /*
     *  update the active tasks counts
     *  and find optimal process for the new connection.
     *  The following strategies apply:
     *  1.  STRAT_EQUALIZE
     *      all processes should be equally filled with connections.
     *
     *  2.  STRAT_COMPRESS
     *      processes should be filled up with connections.
     *      as few processes as possible will be used.
     */
    switch ( e84_connect_strategy )
    {
        case STRAT_EQUALIZE :
                 DBG1 (( MF__,"Strategy: EQUALIZE" ));
                 break;
        case STRAT_COMPRESS :
                 DBG1 (( MF__,"Strategy: COMPRESS" ));
                 break;
        default :
                 MSGD (( IERR_REQUESTOR_ILL_CON_STRAT, e84_connect_strategy ));
                 return ( NULL );
    }

    for ( Ukt = KGS-> first_kp ; Ukt <= KGS-> last_kp ; Ukt ++ )
    {
		DBG1 (( MF__,"scanning tasks in UKT%d \n", THREAD_INDEX(Ukt) ));

	    WAIT_UNTIL_ALONE ( Ukt->exclusive );

        BusyTasks = ActiveTasks = ConnectableTasks = 0 ;

        for ( tcb = KGS->pFirstTaskCtrl ; tcb <= KGS->pLastTaskCtrl ; tcb ++ )
        { 
            if ( tcb->ukt != Ukt ) 
                continue; 

		    if ( tcb->type == TaskTypeToFind )
			{
                DBG1 (( MF__,"T%d (%s) connect_time %d state %s", 
                      tcb->index , e38_type_name(tcb->type),
                      tcb->connection->ci_connect_time,
                      e38_tskstate_name(tcb->state) ));
				if ( tcb->connection->ci_connect_time ||
					 ( (tcb->state != TSK_CONNECTWAIT) &&
					   (tcb->state != TSK_INACTIVE)         ) )
				{
					ActiveTasks ++ ;
                    BusyTasks++ ;
				}
				else
				{
    				ConnectableTasks ++ ;
				}
			}
			else 
			{
                DBG1 (( MF__,"T%d (%s) state %s", 
                      tcb->index , e38_type_name(tcb->type),
                      e38_tskstate_name(tcb->state) ));
				if ( tcb->state != TSK_INACTIVE )
					ActiveTasks ++ ;
			}
		}
		/* for tasks in process */


		CLEARLOCK ( Ukt->exclusive );
		DBG1 (( MF__,"active tasks %d \n", ActiveTasks ));

		switch ( e84_connect_strategy )
		{
		case STRAT_EQUALIZE :
			if (ConnectableTasks > 0 && (!OptimalUkt || (BusyTasks < OptUktBusyTasks)))
			{
                OptimalUkt = Ukt ;
                OptUktBusyTasks = BusyTasks ;
                DBG1 (( MF__,"optimal Ukt now UKT%d \n", THREAD_INDEX(OptimalUkt) ));
			}
			break ;

		case STRAT_COMPRESS :
			if (ConnectableTasks > 0 && (!OptimalUkt || (ActiveTasks > OptimalUkt->activeTasks)))
            {
    			OptimalUkt = Ukt ;
	    		DBG1 (( MF__,"optimal Ukt now UKT%d \n", THREAD_INDEX(OptimalUkt) ));
			}
			break ;

		default :
    	    OptimalUkt = NULL ;
			break ;
		}
    }
    /* for UKPs */
    return OptimalUkt;
}

#include <poll.h>

static int en84_RecvConpkt (
int              sd ,
rte_header		*header ,
int				*statep ,
char           **errm,
int              msecTimeoutTime )
{
#undef	MF__
#define	MF__	"en84_RecvConpkt"
    int				    rc ;
    int				    lgt ;
    int				    len ;
    int				    swap_type ;
    tsp00_Int4			len4 ;
    rte_connect_packet *cpack ;
    char			   *ptr ;
    struct pollfd       fifoFd[1];    

    DBGIN;
    cpack = (rte_connect_packet*) (header + 1);
    ptr = (char*) header ;
    /*
     *  In the fist place, try to receive the minimal packet only.
     *  This is to avoid reading parts of packets that follow
     *  immediately after the connect packet.
     *  After the minimal part of the connect packet is received,
     *  we can determine the exact packet length.
     */
    lgt = RTE_CONPKT_MINSIZ ;
    len = 0 ;
    len4 = 0 ;

    fifoFd[0].fd = sd;
    fifoFd[0].events = POLLIN;
    /*
     *  Receive connect packet.
     */
    if ( msecTimeoutTime <= 0 ) msecTimeoutTime = -1;

    for ( ; lgt > 0 ; )
    {
        fifoFd[0].revents = 0;
        rc = poll(fifoFd, 1, msecTimeoutTime);
        if ( rc < 0 )
        {
            if ( errno == EINTR )
            {
		        DBG1 (( MF__,"poll interrupted, continuing \n" ))
                continue;
            }
            else
            {
        	    DBG1 (( MF__,"poll error: %s\n" , sqlerrs() ))
        		*errm = "requestor fifo poll error" ;
        		return ( SP1CE_NOTOK );
            }
        }
        else if ( rc == 0 )
        {
    		if ( statep )
            {
                *statep = CON_TIMEDOUT;
            }
		    *errm = "connect timed out" ;
		    DBG1 (( MF__,"%s \n", *errm ));
		    return ( SP1CE_NOTOK );
        }


        DBG1 (( MF__,"read sd %d size %d \n", sd , lgt ));
    	rc = read ( sd , ptr , lgt );
	    DBG1 (( MF__,"read sd %d rc %d \n", sd , rc ));
        if ( rc == -1 )
        {
            if ( errno == EINTR )
            {
                DBG1 (( MF__,"fifo read interrupted, continuing \n" ));
                rc = 0 ;
                continue ;
            }

    	    if ( (errno == ECONNRESET) || (errno == EPIPE) )
	        {
		        *errm = "connection closed by counterpart" ;
        		DBG1 (( MF__,"%s \n", *errm ));
		        rc = SP1CE_RELEASED ;
    	    }
	        else
	        {
        		*errm = "fifo read error" ;
		        DBG1 (( MF__,": %s: %s\n", *errm , sqlerrs() ));
        		rc = SP1CE_NOTOK ;
	        }
            return ( rc );
        }

        if ( rc == 0 )
    	{
	        *errm = "connection closed by counterpart" ;
	        DBG1 (( MF__,"%s \n", *errm ));
    	    return ( SP1CE_RELEASED );
	    }

        DBG1 (( MF__,"received %d bytes \n", rc ));
    	ptr += rc ;
	    lgt -= rc ;
	    len += rc ;
    	if ( ! len4 && (len >= RTE_CONPKT_MINSIZ) )
	    {
    	    swap_type = cpack->cp_mess_code[1];
            sql42_unpack_int4 ( swap_type , header->rh_act_send_len , &len4 );
	        if ( (len4 < RTE_CONPKT_MINSIZ) || (len4 > RTE_CONPKT_SIZE) )
	        {
        		*errm = "received a garbled packet" ;
		        DBG1 (( MF__,": %s, length %d \n", *errm , len4 ));
        		return ( SP1CE_NOTOK );
	        }
    	    lgt = len4 - len ;
	    }
    }

    DBG1 (( MF__,"returning %d \n", SP1CE_OK ));
    return ( SP1CE_OK );
}
